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()