fix: evict stale CounterVec entries to prevent TCP metrics series explosion

containers/container.go

@@ -1495,6 +1495,32 @@ type activeConnAgg struct {
 	count int
 }
 
+// activeCounterLabelKeys returns the set of labelKey() values for destinations
+// that have at least one open connection, and the set of destination strings
+// for destinations still in lastConnectionAttempts (for failed-connect eviction).
+// Called from the event handler goroutine.
+func (c *Container) activeCounterLabelKeys() (activeKeys map[string]struct{}, activeFailedDsts map[string]struct{}) {
+	c.lock.RLock()
+	defer c.lock.RUnlock()
+
+	activeKeys = make(map[string]struct{})
+	for _, conn := range c.activeConnections {
+		if !conn.Closed.IsZero() {
+			continue
+		}
+		enrichedKey := c.enrichDestinationKey(conn.DestinationKey)
+		labels := tcpLabels(enrichedKey, conn.srcWorkload)
+		activeKeys[labelKey(labels)] = struct{}{}
+	}
+
+	activeFailedDsts = make(map[string]struct{}, len(c.lastConnectionAttempts))
+	for dst := range c.lastConnectionAttempts {
+		activeFailedDsts[dst.String()] = struct{}{}
+	}
+
+	return activeKeys, activeFailedDsts
+}
+
 func (c *Container) onRetransmission(src netaddr.IPPort, dst netaddr.IPPort) bool {
 	c.lock.RLock()
 	conn, ok := c.activeConnections[ConnectionKey{src: src, dst: dst}]

containers/registry.go

@@ -222,10 +222,14 @@ func (r *Registry) handleEvents(ch <-chan ebpftracer.Event) {
 	defer gcTicker.Stop()
 	ebpfStatsTicker := time.NewTicker(MinTrafficStatsUpdateInterval)
 	defer ebpfStatsTicker.Stop()
+	counterEvictTicker := time.NewTicker(gcInterval)
+	defer counterEvictTicker.Stop()
 	for {
 		select {
 		case <-ebpfStatsTicker.C:
 			r.updateEbpfStatsAndActiveConns()
+		case <-counterEvictTicker.C:
+			r.evictStaleCounterLabels()
 		case now := <-gcTicker.C:
 			for pid, c := range r.containersByPid {
 				cg, err := proc.ReadCgroup(pid)
@@ -757,6 +761,22 @@ func (r *Registry) updateEbpfStatsAndActiveConns() {
 	}
 }
 
+// evictStaleCounterLabels removes CounterVec entries for destinations that no
+// longer have active connections. Runs at gcInterval from the event handler goroutine.
+func (r *Registry) evictStaleCounterLabels() {
+	r.containerLock.RLock()
+	containers := make([]*Container, 0, len(r.containersById))
+	for _, c := range r.containersById {
+		containers = append(containers, c)
+	}
+	r.containerLock.RUnlock()
+
+	for _, c := range containers {
+		activeKeys, activeFailedDsts := c.activeCounterLabelKeys()
+		c.tcpMetrics.EvictStaleLabels(activeKeys, activeFailedDsts)
+	}
+}
+
 func (r *Registry) getDomain(ip netaddr.IP) *common.Domain {
 	r.ip2fqdnLock.RLock()
 	defer r.ip2fqdnLock.RUnlock()

containers/tcp_metrics.go

@@ -1,6 +1,7 @@
 package containers
 
 import (
+	"strings"
 	"sync"
 
 	"github.com/coroot/coroot-node-agent/common"
@@ -11,7 +12,7 @@ import (
 // Event handlers update these directly (lock-free for scrapes).
 // Collect() just forwards pre-built metrics without holding c.lock.
 type TCPMetrics struct {
-	mu sync.RWMutex // protects lazy initialization only
+	mu sync.RWMutex // protects initialization and eviction vs collect
 
 	successful  *prometheus.CounterVec
 	totalTime   *prometheus.CounterVec
@@ -23,6 +24,14 @@ type TCPMetrics struct {
 	restarts    prometheus.Counter
 	oomKills    prometheus.Counter
 
+	// knownLabels tracks label combinations pushed to the 11-label CounterVecs.
+	// Key is null-byte-joined label values; value is the original slice for DeleteLabelValues.
+	// Written only from the event handler goroutine (no extra lock needed for writes).
+	knownLabels map[string][]string
+
+	// knownFailedLabels tracks label combinations pushed to the 7-label failed CounterVec.
+	knownFailedLabels map[string][]string
+
 	constLabels prometheus.Labels
 	initialized bool
 }
@@ -74,9 +83,33 @@ func (t *TCPMetrics) ensureInitialized() {
 	t.restarts = newCounter("container_restarts_total", "Number of times the container was restarted", cl)
 	t.oomKills = newCounter("container_oom_kills_total", "Total number of times the container was terminated by the OOM killer", cl)
 
+	t.knownLabels = make(map[string][]string)
+	t.knownFailedLabels = make(map[string][]string)
 	t.initialized = true
 }
 
+// labelKey produces a map key from a label slice using null-byte separator.
+func labelKey(labels []string) string {
+	return strings.Join(labels, "\x00")
+}
+
+// trackLabels records a label combination for the 11-label CounterVecs.
+// Only called from the event handler goroutine — no extra lock needed.
+func (t *TCPMetrics) trackLabels(labels []string) {
+	k := labelKey(labels)
+	if _, exists := t.knownLabels[k]; !exists {
+		t.knownLabels[k] = labels
+	}
+}
+
+// trackFailedLabels records a label combination for the 7-label failed CounterVec.
+func (t *TCPMetrics) trackFailedLabels(labels []string) {
+	k := labelKey(labels)
+	if _, exists := t.knownFailedLabels[k]; !exists {
+		t.knownFailedLabels[k] = labels
+	}
+}
+
 // tcpLabels builds the 11-label value slice for TCP metrics.
 func tcpLabels(key common.DestinationKey, src common.Workload) []string {
 	dest := key.GetDestinationWorkload()
@@ -93,30 +126,36 @@ func tcpLabels(key common.DestinationKey, src common.Workload) []string {
 func (t *TCPMetrics) ObserveConnectionOpen(key common.DestinationKey, src common.Workload, durationSeconds float64) {
 	t.ensureInitialized()
 	labels := tcpLabels(key, src)
+	t.trackLabels(labels)
 	t.successful.WithLabelValues(labels...).Inc()
 	t.totalTime.WithLabelValues(labels...).Add(durationSeconds)
 }
 
 // ObserveConnectionFailed records a failed connection attempt.
 func (t *TCPMetrics) ObserveConnectionFailed(dst common.HostPort, workload common.Workload) {
 	t.ensureInitialized()
-	t.failed.WithLabelValues(
+	labels := []string{
 		dst.String(),
 		workload.Name, workload.Namespace, workload.Kind,
 		workload.Name, workload.Namespace, workload.Kind,
-	).Inc()
+	}
+	t.trackFailedLabels(labels)
+	t.failed.WithLabelValues(labels...).Inc()
 }
 
 // ObserveRetransmission records a TCP retransmission.
 func (t *TCPMetrics) ObserveRetransmission(key common.DestinationKey, src common.Workload) {
 	t.ensureInitialized()
-	t.retransmits.WithLabelValues(tcpLabels(key, src)...).Inc()
+	labels := tcpLabels(key, src)
+	t.trackLabels(labels)
+	t.retransmits.WithLabelValues(labels...).Inc()
 }
 
 // ObserveTraffic records byte count deltas for a connection.
 func (t *TCPMetrics) ObserveTraffic(key common.DestinationKey, src common.Workload, sentDelta, recvDelta uint64) {
 	t.ensureInitialized()
 	labels := tcpLabels(key, src)
+	t.trackLabels(labels)
 	if sentDelta > 0 {
 		t.bytesSent.WithLabelValues(labels...).Add(float64(sentDelta))
 	}
@@ -153,6 +192,39 @@ func (t *TCPMetrics) ObserveOOMKill() {
 	t.oomKills.Inc()
 }
 
+// EvictStaleLabels removes CounterVec entries for label combinations that are
+// no longer active. activeKeys contains labelKey() values for currently open
+// connections. activeFailedDsts contains destination strings (labels[0]) for
+// destinations still in lastConnectionAttempts.
+// Must be called from the event handler goroutine. Takes mu.Lock to exclude
+// concurrent collect() calls during eviction.
+func (t *TCPMetrics) EvictStaleLabels(activeKeys map[string]struct{}, activeFailedDsts map[string]struct{}) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	if !t.initialized {
+		return
+	}
+
+	for k, labels := range t.knownLabels {
+		if _, active := activeKeys[k]; !active {
+			t.successful.DeleteLabelValues(labels...)
+			t.totalTime.DeleteLabelValues(labels...)
+			t.retransmits.DeleteLabelValues(labels...)
+			t.bytesSent.DeleteLabelValues(labels...)
+			t.bytesRecv.DeleteLabelValues(labels...)
+			delete(t.knownLabels, k)
+		}
+	}
+
+	for k, labels := range t.knownFailedLabels {
+		// labels[0] is the destination string
+		if _, active := activeFailedDsts[labels[0]]; !active {
+			t.failed.DeleteLabelValues(labels...)
+			delete(t.knownFailedLabels, k)
+		}
+	}
+}
+
 // collect forwards all pre-built metrics to the channel. No c.lock needed.
 func (t *TCPMetrics) collect(ch chan<- prometheus.Metric) {
 	t.mu.RLock()