From 112e911020bb6fa92690faabdee5dca799dc01af Mon Sep 17 00:00:00 2001 From: mayankpande88 Date: Thu, 26 Mar 2026 11:09:54 +0530 Subject: [PATCH] fix: evict stale CounterVec label combinations to prevent series explosion CounterVec entries for closed TCP connections were never removed, causing series count to grow linearly with pod uptime. On a 40h pod, 89% of emitted counter series were stale (85/95 destinations in a single container had no active connection). At customer scale this accumulated to 989K series (54% of all Prometheus series). Track label combinations pushed to each CounterVec and periodically delete entries whose destinations are no longer in activeConnections. Eviction runs every gcInterval (5min) from the event handler goroutine, taking mu.Lock to exclude concurrent collect() calls. --- containers/container.go | 26 +++++++++++++ containers/registry.go | 20 ++++++++++ containers/tcp_metrics.go | 80 +++++++++++++++++++++++++++++++++++++-- 3 files changed, 122 insertions(+), 4 deletions(-) diff --git a/containers/container.go b/containers/container.go index 7f6b361..af4ea03 100644 --- a/containers/container.go +++ b/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}] diff --git a/containers/registry.go b/containers/registry.go index a17a039..6980794 100644 --- a/containers/registry.go +++ b/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() diff --git a/containers/tcp_metrics.go b/containers/tcp_metrics.go index d3e20fd..fb955fd 100644 --- a/containers/tcp_metrics.go +++ b/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,6 +126,7 @@ 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) } @@ -100,23 +134,28 @@ func (t *TCPMetrics) ObserveConnectionOpen(key common.DestinationKey, src common // 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()