From 841874d18a2b05c90478a1a55009876b266622c6 Mon Sep 17 00:00:00 2001 From: mayankpande88 Date: Sun, 8 Mar 2026 01:10:05 +0530 Subject: [PATCH 1/5] fix: eliminate WrapRegistererWith allocation storm on Prometheus scrapes Removes the 3-layer WrapRegistererWith wrapping chain for container metrics that caused ~200MB of transient allocations per scrape on busy nodes. Container metrics now embed const labels directly, and L7 CounterVec/HistogramVec use ConstLabels on their opts for zero per-scrape overhead. ip2fqdn and LLM metrics retain minimal wrapping. Co-Authored-By: Claude Opus 4.6 --- containers/container.go | 124 ++++++++++++++++++++++++---------------- containers/jvm.go | 14 ++--- containers/l7.go | 22 +++---- containers/metrics.go | 15 ++++- containers/registry.go | 19 ++++-- main.go | 11 +++- 6 files changed, 128 insertions(+), 77 deletions(-) diff --git a/containers/container.go b/containers/container.go index 0748236c..f79da7b9 100644 --- a/containers/container.go +++ b/containers/container.go @@ -172,6 +172,7 @@ type Container struct { done chan struct{} ip_resolver IPResolver srcWorkload common.Workload + constLabels []string // [container_id, app_id, machine_id, system_uuid, az, region] // Atomic throttling fields for lock-free access collectCallCount int64 @@ -215,6 +216,22 @@ func NewContainer(id ContainerID, cg *cgroup.Cgroup, md *ContainerMetadata, pid if appId == cid { appId = "" } + + // Build const labels for direct embedding in metrics (avoids WrapRegistererWith overhead) + nodeLabels := registry.nodeConstLabels + constLabels := make([]string, 0, 2+len(nodeLabels)) + constLabels = append(constLabels, cid, appId) + constLabels = append(constLabels, nodeLabels...) + + promConstLabels := prometheus.Labels{ + "container_id": cid, + "app_id": appId, + "machine_id": nodeLabels[0], + "system_uuid": nodeLabels[1], + "az": nodeLabels[2], + "region": nodeLabels[3], + } + c := &Container{ id: id, appId: appId, @@ -232,7 +249,7 @@ func NewContainer(id ContainerID, cg *cgroup.Cgroup, md *ContainerMetadata, pid lastConnectionAttempts: map[common.HostPort]time.Time{}, activeConnections: map[ConnectionKey]*ActiveConnection{}, connectionsByPidFd: map[PidFd]*ActiveConnection{}, - l7Stats: NewL7Stats(), + l7Stats: NewL7Stats(promConstLabels), gpuStats: map[string]*GpuUsage{}, @@ -248,6 +265,7 @@ func NewContainer(id ContainerID, cg *cgroup.Cgroup, md *ContainerMetadata, pid ip_resolver: registry.ip_resolver, registry: registry, srcWorkload: src_workload, + constLabels: constLabels, } // Initialize the LRU cache for connection stats @@ -296,8 +314,10 @@ func (c *Container) Dead(now time.Time) bool { } func (c *Container) Describe(ch chan<- *prometheus.Desc) { - // some fixed metric description is required here to register/unregister the collector correctly - ch <- prometheus.NewDesc("container", "", nil, nil) + // Unchecked collector: each container emits varying metrics over its lifecycle + // (different L7 protocols, connections, etc.), and we manage label uniqueness + // via constLabels embedded directly in each metric. + // Sending no descriptors lets prometheus handle registration/unregistration by pointer. } func (c *Container) Collect(ch chan<- prometheus.Metric) { @@ -313,7 +333,7 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) { atomic.StoreInt64(&c.lastCollectTime, nowNanos) if c.metadata.image != "" || !c.metadata.systemd.IsEmpty() { - ch <- gauge(metrics.ContainerInfo, 1, c.metadata.image, c.metadata.systemd.TriggeredBy, c.metadata.systemd.Type) + ch <- c.gauge(metrics.ContainerInfo, 1, c.metadata.image, c.metadata.systemd.TriggeredBy, c.metadata.systemd.Type) } c.lock.RLock() @@ -321,41 +341,41 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) { oomKills := c.oomKills c.lock.RUnlock() - ch <- counter(metrics.Restarts, float64(restarts)) + ch <- c.counter(metrics.Restarts, float64(restarts)) if cpu := c.cgroup.CpuStat(); cpu != nil { if cpu.LimitCores > 0 { - ch <- gauge(metrics.CPULimit, cpu.LimitCores) + ch <- c.gauge(metrics.CPULimit, cpu.LimitCores) } - ch <- counter(metrics.CPUUsage, cpu.UsageSeconds) - ch <- counter(metrics.ThrottledTime, cpu.ThrottledTimeSeconds) + ch <- c.counter(metrics.CPUUsage, cpu.UsageSeconds) + ch <- c.counter(metrics.ThrottledTime, cpu.ThrottledTimeSeconds) } if taskstatsClient != nil { c.updateDelays() - ch <- counter(metrics.CPUDelay, float64(c.delays.cpu)/float64(time.Second)) - ch <- counter(metrics.DiskDelay, float64(c.delays.disk)/float64(time.Second)) + ch <- c.counter(metrics.CPUDelay, float64(c.delays.cpu)/float64(time.Second)) + ch <- c.counter(metrics.DiskDelay, float64(c.delays.disk)/float64(time.Second)) } if s := c.cgroup.MemoryStat(); s != nil { - ch <- gauge(metrics.MemoryRss, float64(s.RSS)) - ch <- gauge(metrics.MemoryCache, float64(s.Cache)) + ch <- c.gauge(metrics.MemoryRss, float64(s.RSS)) + ch <- c.gauge(metrics.MemoryCache, float64(s.Cache)) if s.Limit > 0 { - ch <- gauge(metrics.MemoryLimit, float64(s.Limit)) + ch <- c.gauge(metrics.MemoryLimit, float64(s.Limit)) } } if psi := c.cgroup.PSI(); psi != nil { - ch <- counter(metrics.PsiCPU, psi.CPUSecondsSome, "some") - ch <- counter(metrics.PsiCPU, psi.CPUSecondsFull, "full") - ch <- counter(metrics.PsiMemory, psi.MemorySecondsSome, "some") - ch <- counter(metrics.PsiMemory, psi.MemorySecondsFull, "full") - ch <- counter(metrics.PsiIO, psi.IOSecondsSome, "some") - ch <- counter(metrics.PsiIO, psi.IOSecondsFull, "full") + ch <- c.counter(metrics.PsiCPU, psi.CPUSecondsSome, "some") + ch <- c.counter(metrics.PsiCPU, psi.CPUSecondsFull, "full") + ch <- c.counter(metrics.PsiMemory, psi.MemorySecondsSome, "some") + ch <- c.counter(metrics.PsiMemory, psi.MemorySecondsFull, "full") + ch <- c.counter(metrics.PsiIO, psi.IOSecondsSome, "some") + ch <- c.counter(metrics.PsiIO, psi.IOSecondsFull, "full") } if oomKills > 0 { - ch <- counter(metrics.OOMKills, float64(oomKills)) + ch <- c.counter(metrics.OOMKills, float64(oomKills)) } if disks, err := node.GetDisks(); err == nil { @@ -367,15 +387,15 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) { } for mountPoint, fsStat := range mounts { dls := []string{mountPoint, device, c.metadata.volumes[mountPoint]} - ch <- gauge(metrics.DiskSize, float64(fsStat.CapacityBytes), dls...) - ch <- gauge(metrics.DiskUsed, float64(fsStat.UsedBytes), dls...) - ch <- gauge(metrics.DiskReserved, float64(fsStat.ReservedBytes), dls...) + ch <- c.gauge(metrics.DiskSize, float64(fsStat.CapacityBytes), dls...) + ch <- c.gauge(metrics.DiskUsed, float64(fsStat.UsedBytes), dls...) + ch <- c.gauge(metrics.DiskReserved, float64(fsStat.ReservedBytes), dls...) if ioStat != nil { if io, ok := ioStat[majorMinor]; ok { - ch <- counter(metrics.DiskReadOps, float64(io.ReadOps), dls...) - ch <- counter(metrics.DiskReadBytes, float64(io.ReadBytes), dls...) - ch <- counter(metrics.DiskWriteOps, float64(io.WriteOps), dls...) - ch <- counter(metrics.DiskWriteBytes, float64(io.WrittenBytes), dls...) + ch <- c.counter(metrics.DiskReadOps, float64(io.ReadOps), dls...) + ch <- c.counter(metrics.DiskReadBytes, float64(io.ReadBytes), dls...) + ch <- c.counter(metrics.DiskWriteOps, float64(io.WriteOps), dls...) + ch <- c.counter(metrics.DiskWriteBytes, float64(io.WrittenBytes), dls...) } } } @@ -389,11 +409,11 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) { c.lock.RUnlock() for addr, open := range listens { - ch <- gauge(metrics.NetListenInfo, float64(open), addr.String(), "") + ch <- c.gauge(metrics.NetListenInfo, float64(open), addr.String(), "") } for proxy, addrs := range proxiedListens { for addr := range addrs { - ch <- gauge(metrics.NetListenInfo, 1, addr.String(), proxy) + ch <- c.gauge(metrics.NetListenInfo, 1, addr.String(), proxy) } } @@ -437,13 +457,13 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) { workload_dest := enrichedKey.GetDestinationWorkload() actualDestWorkload := enrichedKey.GetActualDestinationWorkload() - ch <- counter(metrics.NetConnectionsSuccessful, float64(stats.Count), enrichedKey.DestinationLabelValue(), enrichedKey.ActualDestinationLabelValue(), workload_src.Name, workload_src.Namespace, workload_src.Kind, workload_dest.Name, workload_dest.Namespace, workload_dest.Kind, actualDestWorkload.Name, actualDestWorkload.Namespace, actualDestWorkload.Kind) - ch <- counter(metrics.NetConnectionsTotalTime, stats.TotalTime.Seconds(), enrichedKey.DestinationLabelValue(), enrichedKey.ActualDestinationLabelValue(), workload_src.Name, workload_src.Namespace, workload_src.Kind, workload_dest.Name, workload_dest.Namespace, workload_dest.Kind, actualDestWorkload.Name, actualDestWorkload.Namespace, actualDestWorkload.Kind) + ch <- c.counter(metrics.NetConnectionsSuccessful, float64(stats.Count), enrichedKey.DestinationLabelValue(), enrichedKey.ActualDestinationLabelValue(), workload_src.Name, workload_src.Namespace, workload_src.Kind, workload_dest.Name, workload_dest.Namespace, workload_dest.Kind, actualDestWorkload.Name, actualDestWorkload.Namespace, actualDestWorkload.Kind) + ch <- c.counter(metrics.NetConnectionsTotalTime, stats.TotalTime.Seconds(), enrichedKey.DestinationLabelValue(), enrichedKey.ActualDestinationLabelValue(), workload_src.Name, workload_src.Namespace, workload_src.Kind, workload_dest.Name, workload_dest.Namespace, workload_dest.Kind, actualDestWorkload.Name, actualDestWorkload.Namespace, actualDestWorkload.Kind) if stats.Retransmissions > 0 { - ch <- counter(metrics.NetRetransmits, float64(stats.Retransmissions), enrichedKey.DestinationLabelValue(), enrichedKey.ActualDestinationLabelValue(), workload_src.Name, workload_src.Namespace, workload_src.Kind, workload_dest.Name, workload_dest.Namespace, workload_dest.Kind, actualDestWorkload.Name, actualDestWorkload.Namespace, actualDestWorkload.Kind) + ch <- c.counter(metrics.NetRetransmits, float64(stats.Retransmissions), enrichedKey.DestinationLabelValue(), enrichedKey.ActualDestinationLabelValue(), workload_src.Name, workload_src.Namespace, workload_src.Kind, workload_dest.Name, workload_dest.Namespace, workload_dest.Kind, actualDestWorkload.Name, actualDestWorkload.Namespace, actualDestWorkload.Kind) } - ch <- counter(metrics.NetBytesSent, float64(stats.BytesSent), enrichedKey.DestinationLabelValue(), enrichedKey.ActualDestinationLabelValue(), workload_src.Name, workload_src.Namespace, workload_src.Kind, workload_dest.Name, workload_dest.Namespace, workload_dest.Kind, actualDestWorkload.Name, actualDestWorkload.Namespace, actualDestWorkload.Kind) - ch <- counter(metrics.NetBytesReceived, float64(stats.BytesReceived), enrichedKey.DestinationLabelValue(), enrichedKey.ActualDestinationLabelValue(), workload_src.Name, workload_src.Namespace, workload_src.Kind, workload_dest.Name, workload_dest.Namespace, workload_dest.Kind, actualDestWorkload.Name, actualDestWorkload.Namespace, actualDestWorkload.Kind) + ch <- c.counter(metrics.NetBytesSent, float64(stats.BytesSent), enrichedKey.DestinationLabelValue(), enrichedKey.ActualDestinationLabelValue(), workload_src.Name, workload_src.Namespace, workload_src.Kind, workload_dest.Name, workload_dest.Namespace, workload_dest.Kind, actualDestWorkload.Name, actualDestWorkload.Namespace, actualDestWorkload.Kind) + ch <- c.counter(metrics.NetBytesReceived, float64(stats.BytesReceived), enrichedKey.DestinationLabelValue(), enrichedKey.ActualDestinationLabelValue(), workload_src.Name, workload_src.Namespace, workload_src.Kind, workload_dest.Name, workload_dest.Namespace, workload_dest.Kind, actualDestWorkload.Name, actualDestWorkload.Namespace, actualDestWorkload.Kind) } // Copy failedConnectionAttempts under read lock to avoid concurrent map access c.lock.RLock() @@ -455,7 +475,7 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) { for dst, count := range failedConnCopy { workload := c.ip_resolver.ResolveIP(dst.IP().String()) - ch <- counter(metrics.NetConnectionsFailed, float64(count), dst.String(), workload.Name, workload.Namespace, workload.Kind, workload.Name, workload.Namespace, workload.Kind) + ch <- c.counter(metrics.NetConnectionsFailed, float64(count), dst.String(), workload.Name, workload.Namespace, workload.Kind, workload.Name, workload.Namespace, workload.Kind) } connections := map[common.DestinationKey]int{} @@ -470,7 +490,7 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) { for enrichedKey, count := range connections { actualDestWorkload := enrichedKey.GetActualDestinationWorkload() destWorkload := enrichedKey.GetDestinationWorkload() - ch <- gauge(metrics.NetConnectionsActive, float64(count), enrichedKey.DestinationLabelValue(), enrichedKey.ActualDestinationLabelValue(), c.srcWorkload.Name, c.srcWorkload.Namespace, c.srcWorkload.Kind, destWorkload.Name, destWorkload.Namespace, destWorkload.Kind, actualDestWorkload.Name, actualDestWorkload.Namespace, actualDestWorkload.Kind) + ch <- c.gauge(metrics.NetConnectionsActive, float64(count), enrichedKey.DestinationLabelValue(), enrichedKey.ActualDestinationLabelValue(), c.srcWorkload.Name, c.srcWorkload.Namespace, c.srcWorkload.Kind, destWorkload.Name, destWorkload.Namespace, destWorkload.Kind, actualDestWorkload.Name, actualDestWorkload.Namespace, actualDestWorkload.Kind) } // Copy logParsers under read lock to avoid concurrent map access @@ -496,11 +516,11 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) { } c.lock.Unlock() } - ch <- counter(metrics.LogMessages, float64(ctr.Messages), source, ctr.Level.String(), ctr.Hash, sample) + ch <- c.counter(metrics.LogMessages, float64(ctr.Messages), source, ctr.Level.String(), ctr.Hash, sample) } } for _, c := range p.parser.GetSensitiveCounters() { - ch <- counter(metrics.SensitiveLogMessages, float64(c.Messages), source, c.Pattern, common.TruncateUtf8(c.Sample, *flags.MaxLabelLength), c.Regex, c.Name, c.Hash) + ch <- c.counter(metrics.SensitiveLogMessages, float64(c.Messages), source, c.Pattern, common.TruncateUtf8(c.Sample, *flags.MaxLabelLength), c.Regex, c.Name, c.Hash) } } @@ -541,7 +561,7 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) { } switch { case proc.IsJvm(cmdline): - jvm, jMetrics := jvmMetrics(pid) + jvm, jMetrics := c.jvmMetrics(pid) if len(jMetrics) > 0 && !seenJvms[jvm] { seenJvms[jvm] = true for _, m := range jMetrics { @@ -573,18 +593,18 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) { } } for uuid, usage := range c.gpuStats { - ch <- gauge(metrics.GpuUsagePercent, usage.GPU, uuid) - ch <- gauge(metrics.GpuMemoryUsagePercent, usage.Memory, uuid) + ch <- c.gauge(metrics.GpuUsagePercent, usage.GPU, uuid) + ch <- c.gauge(metrics.GpuMemoryUsagePercent, usage.Memory, uuid) } for appType := range appTypes { - ch <- gauge(metrics.ApplicationType, 1, appType) + ch <- c.gauge(metrics.ApplicationType, 1, appType) } if c.pythonStats != nil { - ch <- counter(metrics.PythonThreadLockWaitTime, c.pythonStats.ThreadLockWaitTime.Seconds()) + ch <- c.counter(metrics.PythonThreadLockWaitTime, c.pythonStats.ThreadLockWaitTime.Seconds()) } if c.nodejsStats != nil { - ch <- counter(metrics.NodejsEventLoopBlockedTime, c.nodejsStats.EventLoopBlockedTime.Seconds()) + ch <- c.counter(metrics.NodejsEventLoopBlockedTime, c.nodejsStats.EventLoopBlockedTime.Seconds()) } c.l7Stats.collect(ch) @@ -592,7 +612,7 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) { if !*flags.DisablePinger { for ip, rtt := range c.ping() { destination_workload := c.ip_resolver.ResolveIP(ip.String()) - ch <- gauge(metrics.NetLatency, rtt, ip.String(), destination_workload.Name, destination_workload.Namespace, destination_workload.Kind) + ch <- c.gauge(metrics.NetLatency, rtt, ip.String(), destination_workload.Name, destination_workload.Namespace, destination_workload.Kind) } } @@ -2202,10 +2222,16 @@ func detectProviderFromRequestStructure(requestData []byte) LLMProvider { return ProviderUnknown } -func counter(desc *prometheus.Desc, value float64, labelValues ...string) prometheus.Metric { - return prometheus.MustNewConstMetric(desc, prometheus.CounterValue, value, labelValues...) +func (c *Container) counter(desc *prometheus.Desc, value float64, labelValues ...string) prometheus.Metric { + allLabels := make([]string, 0, len(c.constLabels)+len(labelValues)) + allLabels = append(allLabels, c.constLabels...) + allLabels = append(allLabels, labelValues...) + return prometheus.MustNewConstMetric(desc, prometheus.CounterValue, value, allLabels...) } -func gauge(desc *prometheus.Desc, value float64, labelValues ...string) prometheus.Metric { - return prometheus.MustNewConstMetric(desc, prometheus.GaugeValue, value, labelValues...) +func (c *Container) gauge(desc *prometheus.Desc, value float64, labelValues ...string) prometheus.Metric { + allLabels := make([]string, 0, len(c.constLabels)+len(labelValues)) + allLabels = append(allLabels, c.constLabels...) + allLabels = append(allLabels, labelValues...) + return prometheus.MustNewConstMetric(desc, prometheus.GaugeValue, value, allLabels...) } diff --git a/containers/jvm.go b/containers/jvm.go index c919ab24..d62d40c8 100644 --- a/containers/jvm.go +++ b/containers/jvm.go @@ -13,7 +13,7 @@ import ( "k8s.io/klog/v2" ) -func jvmMetrics(pid uint32) (string, []prometheus.Metric) { +func (c *Container) jvmMetrics(pid uint32) (string, []prometheus.Metric) { nsPid, err := proc.GetNsPid(pid) if err != nil { if !common.IsNotExist(err) { @@ -37,7 +37,7 @@ func jvmMetrics(pid uint32) (string, []prometheus.Metric) { jvm := pd.getString("sun.rt.javaCommand") var res []prometheus.Metric - res = append(res, gauge(metrics.JvmInfo, 1, jvm, pd.getString("java.property.java.version"))) + res = append(res, c.gauge(metrics.JvmInfo, 1, jvm, pd.getString("java.property.java.version"))) func() { size := float64(0) @@ -49,8 +49,8 @@ func jvmMetrics(pid uint32) (string, []prometheus.Metric) { used += float64(pd.getInt64("sun.gc.generation.%d.space.%d.used", gen, s)) } } - res = append(res, gauge(metrics.JvmHeapSize, size, jvm)) - res = append(res, gauge(metrics.JvmHeapUsed, used, jvm)) + res = append(res, c.gauge(metrics.JvmHeapSize, size, jvm)) + res = append(res, c.gauge(metrics.JvmHeapUsed, used, jvm)) }() gc := func(prefix string) { @@ -58,14 +58,14 @@ func jvmMetrics(pid uint32) (string, []prometheus.Metric) { if name == "" { return } - res = append(res, counter(metrics.JvmGCTime, time.Duration(pd.getInt64(prefix+"time")).Seconds(), jvm, name)) + res = append(res, c.counter(metrics.JvmGCTime, time.Duration(pd.getInt64(prefix+"time")).Seconds(), jvm, name)) } gc("sun.gc.collector.0.") gc("sun.gc.collector.1.") gc("sun.gc.collector.2.") - res = append(res, counter(metrics.JvmSafepointTime, time.Duration(pd.getInt64("sun.rt.safepointTime")).Seconds(), jvm)) - res = append(res, counter(metrics.JvmSafepointSyncTime, time.Duration(pd.getInt64("sun.rt.safepointSyncTime")).Seconds(), jvm)) + res = append(res, c.counter(metrics.JvmSafepointTime, time.Duration(pd.getInt64("sun.rt.safepointTime")).Seconds(), jvm)) + res = append(res, c.counter(metrics.JvmSafepointSyncTime, time.Duration(pd.getInt64("sun.rt.safepointSyncTime")).Seconds(), jvm)) return jvm, res } diff --git a/containers/l7.go b/containers/l7.go index 24c9009c..bd60344e 100644 --- a/containers/l7.go +++ b/containers/l7.go @@ -101,17 +101,19 @@ func normalizeHttpPath(path string) string { } type L7Stats struct { - mu sync.RWMutex - requests map[l7.Protocol]*prometheus.CounterVec - latency map[l7.Protocol]*prometheus.HistogramVec - initialized map[l7.Protocol]bool + mu sync.RWMutex + requests map[l7.Protocol]*prometheus.CounterVec + latency map[l7.Protocol]*prometheus.HistogramVec + initialized map[l7.Protocol]bool + promConstLabels prometheus.Labels // container_id, app_id, machine_id, system_uuid, az, region } -func NewL7Stats() L7Stats { +func NewL7Stats(constLabels prometheus.Labels) L7Stats { return L7Stats{ - requests: make(map[l7.Protocol]*prometheus.CounterVec), - latency: make(map[l7.Protocol]*prometheus.HistogramVec), - initialized: make(map[l7.Protocol]bool), + requests: make(map[l7.Protocol]*prometheus.CounterVec), + latency: make(map[l7.Protocol]*prometheus.HistogramVec), + initialized: make(map[l7.Protocol]bool), + promConstLabels: constLabels, } } @@ -253,7 +255,7 @@ func (s *L7Stats) ensureInitialized(protocol l7.Protocol) { if cOpts, exists := L7Requests[metricsProtocol]; exists { s.requests[protocol] = prometheus.NewCounterVec( - prometheus.CounterOpts{Name: cOpts.Name, Help: cOpts.Help}, + prometheus.CounterOpts{Name: cOpts.Name, Help: cOpts.Help, ConstLabels: s.promConstLabels}, requestLabels, ) } @@ -270,7 +272,7 @@ func (s *L7Stats) ensureInitialized(protocol l7.Protocol) { if hOpts, exists := L7Latency[metricsProtocol]; exists { s.latency[protocol] = prometheus.NewHistogramVec( - prometheus.HistogramOpts{Name: hOpts.Name, Help: hOpts.Help}, + prometheus.HistogramOpts{Name: hOpts.Name, Help: hOpts.Help, ConstLabels: s.promConstLabels}, histogramLabels, ) } diff --git a/containers/metrics.go b/containers/metrics.go index 9cd0c877..5739bd2a 100644 --- a/containers/metrics.go +++ b/containers/metrics.go @@ -5,6 +5,11 @@ import ( "github.com/prometheus/client_golang/prometheus" ) +// constLabelNames are prepended to every container metric descriptor. +// Values are set per-container and prepended by Container.counter()/gauge(). +// Must be declared before `metrics` so it's initialized when metric() is called. +var constLabelNames = []string{"container_id", "app_id", "machine_id", "system_uuid", "az", "region"} + var metrics = struct { ContainerInfo *prometheus.Desc Restarts *prometheus.Desc @@ -111,7 +116,10 @@ var metrics = struct { JvmSafepointTime: metric("container_jvm_safepoint_time_seconds", "Time the application has been stopped for safepoint operations in seconds", "jvm"), JvmSafepointSyncTime: metric("container_jvm_safepoint_sync_time_seconds", "Time spent getting to safepoints in seconds", "jvm"), - Ip2Fqdn: metric("ip_to_fqdn", "Mapping IP addresses to FQDNs based on DNS requests initiated by containers", "ip", "fqdn"), + // Ip2Fqdn is emitted by Registry.Collect, not Container.Collect. + // It gets machine_id/system_uuid/az/region from the wrapped registerer, + // so it does NOT need constLabelNames. + Ip2Fqdn: prometheus.NewDesc("ip_to_fqdn", "Mapping IP addresses to FQDNs based on DNS requests initiated by containers", []string{"ip", "fqdn"}, nil), PythonThreadLockWaitTime: metric("container_python_thread_lock_wait_time_seconds", "Time spent waiting acquiring GIL in seconds"), NodejsEventLoopBlockedTime: metric("container_nodejs_event_loop_blocked_time_seconds_total", "Total time the Node.js event loop spent blocked"), @@ -156,7 +164,10 @@ var ( ) func metric(name, help string, labels ...string) *prometheus.Desc { - return prometheus.NewDesc(name, help, labels, nil) + allLabels := make([]string, 0, len(constLabelNames)+len(labels)) + allLabels = append(allLabels, constLabelNames...) + allLabels = append(allLabels, labels...) + return prometheus.NewDesc(name, help, allLabels, nil) } func newCounter(name, help string, constLabels prometheus.Labels) prometheus.Counter { diff --git a/containers/registry.go b/containers/registry.go index 0b15bf67..69093551 100644 --- a/containers/registry.go +++ b/containers/registry.go @@ -54,7 +54,8 @@ type IPResolver interface { } type Registry struct { - reg prometheus.Registerer + reg prometheus.Registerer // wrapped registerer for ip2fqdn and LLM metrics + rawReg prometheus.Registerer // raw registry for containers (no wrapping overhead) tracer *ebpftracer.Tracer events chan ebpftracer.Event @@ -78,6 +79,10 @@ type Registry struct { gpuProcessUsageSampleChan chan gpu.ProcessUsageSample + // nodeConstLabels holds [machine_id, system_uuid, az, region] values + // for embedding directly in metrics (avoids WrapRegistererWith overhead). + nodeConstLabels []string + // pendingL7Events stores L7 events that arrived before their connection was established // This handles the race condition between ring buffer (L7 events) and perf buffer (TCP events) pendingL7Events []pendingL7Event @@ -91,7 +96,7 @@ type pendingL7Event struct { retryCount int } -func NewRegistry(reg prometheus.Registerer, processInfoCh chan<- ProcessInfo, ip_resolver *common.K8sIPResolver, gpuProcessUsageSampleChan chan gpu.ProcessUsageSample) (*Registry, error) { +func NewRegistry(reg prometheus.Registerer, rawReg prometheus.Registerer, processInfoCh chan<- ProcessInfo, ip_resolver *common.K8sIPResolver, gpuProcessUsageSampleChan chan gpu.ProcessUsageSample, machineId, systemUuid, az, region string) (*Registry, error) { ns, err := proc.GetSelfNetNs() if err != nil { return nil, err @@ -131,6 +136,7 @@ func NewRegistry(reg prometheus.Registerer, processInfoCh chan<- ProcessInfo, ip r := &Registry{ reg: reg, + rawReg: rawReg, events: make(chan ebpftracer.Event, 2000), containersById: map[ContainerID]*Container{}, containersByCgroupId: map[string]*Container{}, @@ -146,6 +152,7 @@ func NewRegistry(reg prometheus.Registerer, processInfoCh chan<- ProcessInfo, ip pythonStatsUpdateCh: make(chan *PythonStatsUpdate), gpuProcessUsageSampleChan: gpuProcessUsageSampleChan, + nodeConstLabels: []string{machineId, systemUuid, az, region}, } // Register LLM metrics with the same registerer used for other container metrics RegisterLLMMetrics(reg) @@ -170,7 +177,7 @@ func (r *Registry) Collect(ch chan<- prometheus.Metric) { r.ip2fqdnLock.RLock() defer r.ip2fqdnLock.RUnlock() for ip, domain := range r.ip2fqdn { - ch <- gauge(metrics.Ip2Fqdn, 1, ip.String(), domain.FQDN) + ch <- prometheus.MustNewConstMetric(metrics.Ip2Fqdn, prometheus.GaugeValue, 1, ip.String(), domain.FQDN) } } @@ -239,7 +246,7 @@ func (r *Registry) handleEvents(ch <-chan ebpftracer.Event) { delete(r.containersById, c.id) // Unregister from Prometheus (do this after removing from maps) - if ok := prometheus.WrapRegistererWith(prometheus.Labels{"container_id": string(c.id), "app_id": c.appId}, r.reg).Unregister(c); !ok { + if ok := r.rawReg.Unregister(c); !ok { klog.Warningf("failed to unregister container: %s", c.id) } c.Close() @@ -633,7 +640,7 @@ func (r *Registry) getOrCreateContainer(pid uint32) *Container { } if c := r.containersById[id]; c != nil { klog.Warningln("id conflict, replacing container:", id) - prometheus.WrapRegistererWith(prometheus.Labels{"container_id": string(c.id), "app_id": c.appId}, r.reg).Unregister(c) + r.rawReg.Unregister(c) delete(r.containersById, c.id) for cgid, container := range r.containersByCgroupId { if container == c { @@ -655,7 +662,7 @@ func (r *Registry) getOrCreateContainer(pid uint32) *Container { return nil } klog.InfoS("detected a new container", "pid", pid, "cg", cg.Id, "id", id, "app", c.appId) - if err := prometheus.WrapRegistererWith(prometheus.Labels{"container_id": string(id), "app_id": c.appId}, r.reg).Register(c); err != nil { + if err := r.rawReg.Register(c); err != nil { klog.Warningf("failed to register container %s: %v", id, err) return nil } diff --git a/main.go b/main.go index a6fda55f..b334db70 100644 --- a/main.go +++ b/main.go @@ -219,15 +219,20 @@ func main() { } registerer.MustRegister(info("node_agent_info", version)) + az := "" + region := "" if md := nodeCollector.Metadata(); md != nil { - region := md.Region - az := md.AvailabilityZone + region = md.Region + az = md.AvailabilityZone if region != "" && az != "" { registerer = prometheus.WrapRegistererWith(prometheus.Labels{"az": az, "region": region}, registerer) } } processInfoCh := profiling.Init(machineId, hostname) - cr, err := containers.NewRegistry(registerer, processInfoCh, resolver, gpuCollector.ProcessUsageSampleCh) + // Pass both the wrapped registerer (for ip2fqdn and LLM metrics) and the raw + // registry (for containers) to avoid WrapRegistererWith allocation overhead on + // the hot path. Container metrics embed const labels directly. + cr, err := containers.NewRegistry(registerer, registry, processInfoCh, resolver, gpuCollector.ProcessUsageSampleCh, machineId, systemUuid, az, region) if err != nil { klog.Exitln(err) } From 7bb8b438ceebdf979225d221cae2d4b6aab6c94f Mon Sep 17 00:00:00 2001 From: mayankpande88 Date: Sun, 8 Mar 2026 01:16:15 +0530 Subject: [PATCH 2/5] fix: use NodeConstLabels struct instead of magic indices, fix variable shadowing Addresses PR review: replaces []string with named struct fields for node-level const labels. Also fixes variable shadowing where loop variable `c` in GetSensitiveCounters range shadowed the Container receiver, causing a compile error. Co-Authored-By: Claude Opus 4.6 --- containers/container.go | 19 ++++++++++--------- containers/registry.go | 20 ++++++++++++++++---- 2 files changed, 26 insertions(+), 13 deletions(-) diff --git a/containers/container.go b/containers/container.go index f79da7b9..722905bd 100644 --- a/containers/container.go +++ b/containers/container.go @@ -218,18 +218,19 @@ func NewContainer(id ContainerID, cg *cgroup.Cgroup, md *ContainerMetadata, pid } // Build const labels for direct embedding in metrics (avoids WrapRegistererWith overhead) - nodeLabels := registry.nodeConstLabels - constLabels := make([]string, 0, 2+len(nodeLabels)) + nl := registry.nodeConstLabels + nodeValues := nl.Values() + constLabels := make([]string, 0, 2+len(nodeValues)) constLabels = append(constLabels, cid, appId) - constLabels = append(constLabels, nodeLabels...) + constLabels = append(constLabels, nodeValues...) promConstLabels := prometheus.Labels{ "container_id": cid, "app_id": appId, - "machine_id": nodeLabels[0], - "system_uuid": nodeLabels[1], - "az": nodeLabels[2], - "region": nodeLabels[3], + "machine_id": nl.MachineID, + "system_uuid": nl.SystemUUID, + "az": nl.AZ, + "region": nl.Region, } c := &Container{ @@ -519,8 +520,8 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) { ch <- c.counter(metrics.LogMessages, float64(ctr.Messages), source, ctr.Level.String(), ctr.Hash, sample) } } - for _, c := range p.parser.GetSensitiveCounters() { - ch <- c.counter(metrics.SensitiveLogMessages, float64(c.Messages), source, c.Pattern, common.TruncateUtf8(c.Sample, *flags.MaxLabelLength), c.Regex, c.Name, c.Hash) + for _, sc := range p.parser.GetSensitiveCounters() { + ch <- c.counter(metrics.SensitiveLogMessages, float64(sc.Messages), source, sc.Pattern, common.TruncateUtf8(sc.Sample, *flags.MaxLabelLength), sc.Regex, sc.Name, sc.Hash) } } diff --git a/containers/registry.go b/containers/registry.go index 69093551..ca076abc 100644 --- a/containers/registry.go +++ b/containers/registry.go @@ -53,6 +53,18 @@ type IPResolver interface { ResolvePodOwner(string, string) common.Workload } +type NodeConstLabels struct { + MachineID string + SystemUUID string + AZ string + Region string +} + +// Values returns the label values in the order matching constLabelNames. +func (n NodeConstLabels) Values() []string { + return []string{n.MachineID, n.SystemUUID, n.AZ, n.Region} +} + type Registry struct { reg prometheus.Registerer // wrapped registerer for ip2fqdn and LLM metrics rawReg prometheus.Registerer // raw registry for containers (no wrapping overhead) @@ -79,9 +91,9 @@ type Registry struct { gpuProcessUsageSampleChan chan gpu.ProcessUsageSample - // nodeConstLabels holds [machine_id, system_uuid, az, region] values - // for embedding directly in metrics (avoids WrapRegistererWith overhead). - nodeConstLabels []string + // nodeConstLabels holds node-level label values for embedding directly + // in container metrics (avoids WrapRegistererWith overhead). + nodeConstLabels NodeConstLabels // pendingL7Events stores L7 events that arrived before their connection was established // This handles the race condition between ring buffer (L7 events) and perf buffer (TCP events) @@ -152,7 +164,7 @@ func NewRegistry(reg prometheus.Registerer, rawReg prometheus.Registerer, proces pythonStatsUpdateCh: make(chan *PythonStatsUpdate), gpuProcessUsageSampleChan: gpuProcessUsageSampleChan, - nodeConstLabels: []string{machineId, systemUuid, az, region}, + nodeConstLabels: NodeConstLabels{MachineID: machineId, SystemUUID: systemUuid, AZ: az, Region: region}, } // Register LLM metrics with the same registerer used for other container metrics RegisterLLMMetrics(reg) From 19606a30087ac74fe53c096cf5bc6c9aa3f7fe18 Mon Sep 17 00:00:00 2001 From: mayankpande88 Date: Sun, 8 Mar 2026 02:14:05 +0530 Subject: [PATCH 3/5] fix: reduce K8s informer memory by stripping cached objects and removing redundant API calls - Add SetTransform functions to all 9 informers to strip unneeded fields (ManagedFields, Annotations, Spec.Containers, Status.Conditions, etc.) before objects enter the informer cache, reducing memory by ~80%. - Remove getFullClusterSnapshot() which did redundant List() API calls for all resource types at startup (informers already do their own initial List). Replace with factory.WaitForCacheSync() + updateIpMapping() to eliminate the double-fetch memory spike. Co-Authored-By: Claude Opus 4.6 --- common/ip_resolver.go | 267 +++++++++++++++++++----------------------- 1 file changed, 123 insertions(+), 144 deletions(-) diff --git a/common/ip_resolver.go b/common/ip_resolver.go index d56d0b11..35680019 100644 --- a/common/ip_resolver.go +++ b/common/ip_resolver.go @@ -2,7 +2,6 @@ package common import ( "context" - "errors" "fmt" "log" "sync" @@ -191,6 +190,100 @@ func (resolver *K8sIPResolver) StopWatching() { close(resolver.stopSignal) } +// stripPod removes all fields from a Pod that the resolver doesn't need, +// reducing the informer cache footprint by ~90% per pod object. +func stripPod(obj interface{}) (interface{}, error) { + pod, ok := obj.(*v1.Pod) + if !ok { + return obj, nil + } + pod.ManagedFields = nil + pod.Annotations = nil + pod.Finalizers = nil + pod.Spec.Containers = nil + pod.Spec.InitContainers = nil + pod.Spec.Volumes = nil + pod.Spec.EphemeralContainers = nil + pod.Spec.ImagePullSecrets = nil + pod.Spec.Tolerations = nil + pod.Spec.Affinity = nil + pod.Spec.TopologySpreadConstraints = nil + pod.Status.Conditions = nil + pod.Status.ContainerStatuses = nil + pod.Status.InitContainerStatuses = nil + pod.Status.EphemeralContainerStatuses = nil + return pod, nil +} + +// stripNode removes unneeded fields from a Node object. +func stripNode(obj interface{}) (interface{}, error) { + node, ok := obj.(*v1.Node) + if !ok { + return obj, nil + } + node.ManagedFields = nil + node.Annotations = nil + node.Spec = v1.NodeSpec{} + node.Status.Conditions = nil + node.Status.Images = nil + node.Status.VolumesAttached = nil + node.Status.VolumesInUse = nil + node.Status.Capacity = nil + node.Status.Allocatable = nil + return node, nil +} + +// stripService removes unneeded fields from a Service object. +func stripService(obj interface{}) (interface{}, error) { + svc, ok := obj.(*v1.Service) + if !ok { + return obj, nil + } + svc.ManagedFields = nil + svc.Annotations = nil + // Keep: Name, Namespace, Spec.Selector, Spec.ClusterIPs + svc.Spec.Ports = nil + svc.Status = v1.ServiceStatus{} + return svc, nil +} + +// stripOwnerOnly removes everything except ObjectMeta (for UID and OwnerReferences) +// from resources where we only need the ownership chain (ReplicaSet, Deployment, etc.). +func stripOwnerOnly(obj interface{}) (interface{}, error) { + type objectMetaAccessor interface { + GetObjectMeta() metav1.Object + } + if accessor, ok := obj.(objectMetaAccessor); ok { + meta := accessor.GetObjectMeta() + meta.SetManagedFields(nil) + meta.SetAnnotations(nil) + meta.SetLabels(nil) + } + // Type-specific spec/status clearing + switch o := obj.(type) { + case *appsv1.ReplicaSet: + o.Spec = appsv1.ReplicaSetSpec{} + o.Status = appsv1.ReplicaSetStatus{} + // Restore OwnerReferences (cleared spec removes nothing we need, they're in ObjectMeta) + case *appsv1.Deployment: + o.Spec = appsv1.DeploymentSpec{} + o.Status = appsv1.DeploymentStatus{} + case *appsv1.DaemonSet: + o.Spec = appsv1.DaemonSetSpec{} + o.Status = appsv1.DaemonSetStatus{} + case *appsv1.StatefulSet: + o.Spec = appsv1.StatefulSetSpec{} + o.Status = appsv1.StatefulSetStatus{} + case *batchv1.Job: + o.Spec = batchv1.JobSpec{} + o.Status = batchv1.JobStatus{} + case *batchv1.CronJob: + o.Spec = batchv1.CronJobSpec{} + o.Status = batchv1.CronJobStatus{} + } + return obj, nil +} + func (resolver *K8sIPResolver) StartWatching() error { factory := informers.NewSharedInformerFactory(resolver.clientset, time.Minute*10) // Define individual informers @@ -204,6 +297,20 @@ func (resolver *K8sIPResolver) StartWatching() error { serviceInformer := factory.Core().V1().Services().Informer() deploymentInformer := factory.Apps().V1().Deployments().Informer() + // Strip unneeded fields from objects before they enter the informer cache. + // The informer stores full K8s objects internally, but we only need minimal + // fields. This reduces memory by ~80% for the informer cache (the biggest + // contributor on clusters with hundreds of pods/replicasets). + podInformer.SetTransform(stripPod) + nodeInformer.SetTransform(stripNode) + replicaSetInformer.SetTransform(stripOwnerOnly) + daemonSetInformer.SetTransform(stripOwnerOnly) + statefulSetInformer.SetTransform(stripOwnerOnly) + jobInformer.SetTransform(stripOwnerOnly) + cronJobInformer.SetTransform(stripOwnerOnly) + serviceInformer.SetTransform(stripService) + deploymentInformer.SetTransform(stripOwnerOnly) + resolver.addPodHandlers(podInformer) resolver.addNodeHandlers(nodeInformer) resolver.addReplicaSetHandlers(replicaSetInformer) @@ -214,12 +321,22 @@ func (resolver *K8sIPResolver) StartWatching() error { resolver.addServiceHandlers(serviceInformer) resolver.addDeploymentHandlers(deploymentInformer) - // get initial state - err := resolver.getResolvedClusterSnapshot() - if err != nil { - return fmt.Errorf("error retrieving cluster's initial state: %v", err) + // Start informers and wait for initial list+watch sync. + // The event handlers populate resolver.snapshot.* as objects arrive. + factory.Start(resolver.stopSignal) + synced := factory.WaitForCacheSync(resolver.stopSignal) + for informerType, ok := range synced { + if !ok { + return fmt.Errorf("informer sync failed for %v", informerType) + } } - factory.Start(resolver.stopSignal) // runs in background + + // Re-process IP mappings in the correct priority order + // (services → nodes → pods) to handle IP collisions deterministically. + log.Printf("Informer cache synced, building IP mappings") + resolver.updateIpMapping() + log.Printf("IP mappings built") + return nil } @@ -530,144 +647,6 @@ func (resolver *K8sIPResolver) handleNodeEvent(node *v1.Node) bool { return false } -func (resolver *K8sIPResolver) getResolvedClusterSnapshot() error { - log.Printf("Generating full cluster snapshot") - err := resolver.getFullClusterSnapshot() - if err != nil { - return err - } - resolver.updateIpMapping() - log.Printf("Generated full cluster snapshot") - return nil -} - -// iterate the API for initial coverage of the cluster's state -func (resolver *K8sIPResolver) getFullClusterSnapshot() error { - pods, err := resolver.clientset.CoreV1().Pods("").List(context.Background(), metav1.ListOptions{}) - if err != nil { - return errors.New("error getting pods, aborting snapshot update") - } - log.Printf("loaded pods data %d", pods.Size()) - - for _, pod := range pods.Items { - minPod := MinimalPod{ - UID: pod.UID, - Name: pod.Name, - Namespace: pod.Namespace, - Labels: pod.Labels, - NodeName: pod.Spec.NodeName, - PodIPs: pod.Status.PodIPs, - OwnerReferences: pod.ObjectMeta.OwnerReferences, - } - resolver.snapshot.Pods.Store(pod.UID, minPod) - resolver.snapshot.PodNameIndex.Store(pod.Namespace+"/"+pod.Name, pod.UID) - } - - nodes, err := resolver.clientset.CoreV1().Nodes().List(context.Background(), metav1.ListOptions{}) - if err != nil { - return errors.New("error getting nodes, aborting snapshot update") - } - for _, node := range nodes.Items { - // extract region and zone from attributes - region := node.Labels["topology.kubernetes.io/region"] - zone := node.Labels["topology.kubernetes.io/zone"] - meta := InstanceMeta{ - Region: region, - Zone: zone, - Instance: node.Name, - } - resolver.instanceMetaMap.Store(node.Name, meta) - resolver.snapshot.Nodes.Store(node.UID, MinimalNode{ - Name: node.Name, - Labels: node.Labels, - Addresses: node.Status.Addresses, - }) - } - - replicasets, err := resolver.clientset.AppsV1().ReplicaSets("").List(context.Background(), metav1.ListOptions{}) - if err != nil { - return errors.New("error getting replicasets, aborting snapshot update") - } - for _, rs := range replicasets.Items { - resolver.snapshot.ReplicaSets.Store(rs.ObjectMeta.UID, MinimalOwnerInfo{ - OwnerReferences: rs.OwnerReferences, - }) - } - - daemonsets, err := resolver.clientset.AppsV1().DaemonSets("").List(context.Background(), metav1.ListOptions{}) - if err != nil { - return errors.New("error getting daemonsets, aborting snapshot update") - } - for _, ds := range daemonsets.Items { - resolver.snapshot.DaemonSets.Store(ds.ObjectMeta.UID, MinimalOwnerInfo{ - OwnerReferences: ds.OwnerReferences, - }) - } - - statefulsets, err := resolver.clientset.AppsV1().StatefulSets("").List(context.Background(), metav1.ListOptions{}) - if err != nil { - return errors.New("error getting statefulsets, aborting snapshot update") - } - for _, ss := range statefulsets.Items { - resolver.snapshot.StatefulSets.Store(ss.ObjectMeta.UID, MinimalOwnerInfo{ - OwnerReferences: ss.OwnerReferences, - }) - } - - jobs, err := resolver.clientset.BatchV1().Jobs("").List(context.Background(), metav1.ListOptions{}) - if err != nil { - return errors.New("error getting jobs, aborting snapshot update") - } - for _, job := range jobs.Items { - resolver.snapshot.Jobs.Store(job.ObjectMeta.UID, MinimalOwnerInfo{ - OwnerReferences: job.OwnerReferences, - }) - } - - services, err := resolver.clientset.CoreV1().Services("").List(context.Background(), metav1.ListOptions{}) - if err != nil { - return errors.New("error getting services, aborting snapshot update") - } - for _, service := range services.Items { - resolver.snapshot.Services.Store(service.UID, MinimalService{ - Name: service.Name, - Namespace: service.Namespace, - Selector: service.Spec.Selector, - ClusterIPs: service.Spec.ClusterIPs, - }) - } - - deployments, err := resolver.clientset.AppsV1().Deployments("").List(context.Background(), metav1.ListOptions{}) - if err != nil { - return errors.New("error getting deployments, aborting snapshot update") - } - for _, deployment := range deployments.Items { - resolver.snapshot.Deployments.Store(deployment.UID, MinimalOwnerInfo{ - OwnerReferences: deployment.OwnerReferences, - }) - } - - cronJobs, err := resolver.clientset.BatchV1().CronJobs("").List(context.Background(), metav1.ListOptions{}) - if err != nil { - cronJobs, err := resolver.clientset.BatchV1beta1().CronJobs("").List(context.Background(), metav1.ListOptions{}) - if err != nil { - return errors.New("error getting cronjobs, aborting snapshot update") - } - for _, cronJob := range cronJobs.Items { - resolver.snapshot.CronJobs.Store(cronJob.UID, MinimalOwnerInfo{ - OwnerReferences: cronJob.OwnerReferences, - }) - } - } - for _, cronJob := range cronJobs.Items { - resolver.snapshot.CronJobs.Store(cronJob.UID, MinimalOwnerInfo{ - OwnerReferences: cronJob.OwnerReferences, - }) - } - - return nil -} - // add mapping from ip to resolved host to an existing map, // based on the given cluster snapshot func (resolver *K8sIPResolver) updateIpMapping() { From ec67408ff040255baf1b72d26f0df1bc43db3d8a Mon Sep 17 00:00:00 2001 From: mayankpande88 Date: Sun, 8 Mar 2026 09:47:42 +0530 Subject: [PATCH 4/5] fix: push-model metrics to eliminate c.lock deadlock on scrapes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Root cause: Collect() held c.lock.RLock() for extended periods while reading connectionStats, failedConnectionAttempts, activeConnections, and logSamples. Event handlers (onConnectionOpen, onL7Request, etc.) need c.lock.Lock() — creating a deadlock when a pending writer blocks recursive readers. Additionally, Registry.Collect() sent to trafficStatsUpdateCh (consumed by the event handler goroutine), so if the event handler was blocked on c.lock.Lock(), Registry.Collect() blocked too → Gather() never returns → zombie goroutines accumulate. Changes: - New TCPMetrics struct (CounterVec/GaugeVec) updated by event handlers at event time, not scrape time. Collect() calls tcpMetrics.collect(ch) without any c.lock — same pattern as existing L7Stats. - Remove connectionStats LRU cache (no longer needed for accumulation) - Remove failedConnectionAttempts map (push directly to CounterVec) - onConnectionOpen/onRetransmission push metrics before acquiring c.lock - updateConnectionTrafficStats pushes byte deltas to TCPMetrics - Active connections gauge refreshed periodically by event handler - Move updateStatsFromEbpfMapsIfNecessary() from Registry.Collect() into event handler ticker — eliminates trafficStatsUpdateCh deadlock - Remove trafficStatsUpdateCh/nodejsStatsUpdateCh/pythonStatsUpdateCh channels — event handler calls container methods directly - logSamples changed from map[string]string to sync.Map for lock-free LoadOrStore (eliminates c.lock.Lock() in Collect for log samples) - Restarts/OOMKills pushed to TCPMetrics counter at event time Co-Authored-By: Claude Opus 4.6 --- containers/container.go | 293 +++++++++++--------------------------- containers/registry.go | 144 +++++++------------ containers/tcp_metrics.go | 172 ++++++++++++++++++++++ 3 files changed, 310 insertions(+), 299 deletions(-) create mode 100644 containers/tcp_metrics.go diff --git a/containers/container.go b/containers/container.go index 722905bd..445a6f08 100644 --- a/containers/container.go +++ b/containers/container.go @@ -4,7 +4,6 @@ import ( "bytes" "encoding/base64" "errors" - "fmt" "os" "path/filepath" "sort" @@ -23,7 +22,6 @@ import ( "github.com/coroot/coroot-node-agent/pinger" "github.com/coroot/coroot-node-agent/proc" "github.com/coroot/coroot-node-agent/tracing" - lru "github.com/hashicorp/golang-lru/v2" "github.com/nudgebee/logparser" "github.com/prometheus/client_golang/prometheus" "github.com/vishvananda/netns" @@ -40,10 +38,6 @@ var ( gpuStatsWindow = 15 * time.Second ) -const ( - connectionStatsCacheSize = 2048 // LRU cache size for connection stats -) - type ContainerID string type ContainerNetwork struct { @@ -114,14 +108,6 @@ type PidFd struct { Fd uint64 } -type ConnectionStats struct { - Count uint64 - TotalTime time.Duration - Retransmissions uint64 - BytesSent uint64 - BytesReceived uint64 -} - type Container struct { id ContainerID appId string @@ -132,28 +118,25 @@ type Container struct { startedAt time.Time zombieAt time.Time - restarts int delays Delays delaysByPid map[uint32]Delays listens map[netaddr.IPPort]map[uint32]*ListenDetails - connectionStats *lru.Cache[common.DestinationKey, *ConnectionStats] - failedConnectionAttempts map[common.HostPort]int64 - lastConnectionAttempts map[common.HostPort]time.Time - activeConnections map[ConnectionKey]*ActiveConnection - connectionsByPidFd map[PidFd]*ActiveConnection - googleHTTP2Parsers map[PidFd]*l7.Http2Parser // Per-connection HTTP/2 parsers (keyed by pid:fd for correct HPACK state) + lastConnectionAttempts map[common.HostPort]time.Time + activeConnections map[ConnectionKey]*ActiveConnection + connectionsByPidFd map[PidFd]*ActiveConnection + googleHTTP2Parsers map[PidFd]*l7.Http2Parser // Per-connection HTTP/2 parsers (keyed by pid:fd for correct HPACK state) - l7Stats L7Stats + l7Stats L7Stats + tcpMetrics *TCPMetrics // LLM stream tracker for SSE-based completion detection llmStreamTracker *LLMStreamTracker gpuStats map[string]*GpuUsage - oomKills int nodejsStats *ebpftracer.NodejsStats pythonStats *ebpftracer.PythonStats @@ -161,7 +144,7 @@ type Container struct { seenMounts map[uint64]struct{} logParsers map[string]*LogParser - logSamples map[string]string + logSamples sync.Map // map[string]string — hash -> truncated sample (write-once) tracer *tracing.Tracer @@ -245,12 +228,11 @@ func NewContainer(id ContainerID, cg *cgroup.Cgroup, md *ContainerMetadata, pid listens: map[netaddr.IPPort]map[uint32]*ListenDetails{}, - connectionStats: nil, // Will be initialized below - failedConnectionAttempts: map[common.HostPort]int64{}, - lastConnectionAttempts: map[common.HostPort]time.Time{}, - activeConnections: map[ConnectionKey]*ActiveConnection{}, - connectionsByPidFd: map[PidFd]*ActiveConnection{}, - l7Stats: NewL7Stats(promConstLabels), + lastConnectionAttempts: map[common.HostPort]time.Time{}, + activeConnections: map[ConnectionKey]*ActiveConnection{}, + connectionsByPidFd: map[PidFd]*ActiveConnection{}, + l7Stats: NewL7Stats(promConstLabels), + tcpMetrics: NewTCPMetrics(promConstLabels), gpuStats: map[string]*GpuUsage{}, @@ -258,7 +240,6 @@ func NewContainer(id ContainerID, cg *cgroup.Cgroup, md *ContainerMetadata, pid seenMounts: map[uint64]struct{}{}, logParsers: map[string]*LogParser{}, - logSamples: map[string]string{}, tracer: tracing.GetContainerTracer(string(id)), @@ -269,13 +250,6 @@ func NewContainer(id ContainerID, cg *cgroup.Cgroup, md *ContainerMetadata, pid constLabels: constLabels, } - // Initialize the LRU cache for connection stats - connStatsCache, err := lru.New[common.DestinationKey, *ConnectionStats](connectionStatsCacheSize) - if err != nil { - return nil, fmt.Errorf("failed to create connection stats cache: %w", err) - } - c.connectionStats = connStatsCache - // Initialize LLM stream tracker with completion callback c.llmStreamTracker = NewLLMStreamTracker(func(stream *LLMStream) { c.onLLMStreamComplete(stream) @@ -337,12 +311,7 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) { ch <- c.gauge(metrics.ContainerInfo, 1, c.metadata.image, c.metadata.systemd.TriggeredBy, c.metadata.systemd.Type) } - c.lock.RLock() - restarts := c.restarts - oomKills := c.oomKills - c.lock.RUnlock() - - ch <- c.counter(metrics.Restarts, float64(restarts)) + // --- Cgroup/procfs metrics (no c.lock needed) --- if cpu := c.cgroup.CpuStat(); cpu != nil { if cpu.LimitCores > 0 { @@ -375,10 +344,6 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) { ch <- c.counter(metrics.PsiIO, psi.IOSecondsFull, "full") } - if oomKills > 0 { - ch <- c.counter(metrics.OOMKills, float64(oomKills)) - } - if disks, err := node.GetDisks(); err == nil { ioStat := c.cgroup.IOStat() for majorMinor, mounts := range c.getMounts() { @@ -403,7 +368,7 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) { } } - // Snapshot listens under lock since getListens/getProxiedListens access c.listens and c.processes + // --- Listens: brief snapshot under c.lock --- c.lock.RLock() listens := c.getListens() proxiedListens := c.getProxiedListens() @@ -418,83 +383,10 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) { } } - // Snapshot connectionStats under lock since the LRU cache is not concurrent-safe - type connStatSnapshot struct { - key common.DestinationKey - stats ConnectionStats - } - c.lock.RLock() - connSnapshots := make([]connStatSnapshot, 0, c.connectionStats.Len()) - for _, d := range c.connectionStats.Keys() { - if stats, ok := c.connectionStats.Peek(d); ok { - connSnapshots = append(connSnapshots, connStatSnapshot{key: d, stats: *stats}) - } - } - c.lock.RUnlock() - - // Aggregate connection stats by enriched key to avoid duplicate metrics - // when multiple IPs resolve to the same FQDN (e.g., Google's shared IPs) - aggregatedStats := map[common.DestinationKey]*ConnectionStats{} - for _, snap := range connSnapshots { - enrichedKey := c.enrichDestinationKey(snap.key) - if existing, ok := aggregatedStats[enrichedKey]; ok { - existing.Count += snap.stats.Count - existing.TotalTime += snap.stats.TotalTime - existing.Retransmissions += snap.stats.Retransmissions - existing.BytesSent += snap.stats.BytesSent - existing.BytesReceived += snap.stats.BytesReceived - } else { - aggregatedStats[enrichedKey] = &ConnectionStats{ - Count: snap.stats.Count, - TotalTime: snap.stats.TotalTime, - Retransmissions: snap.stats.Retransmissions, - BytesSent: snap.stats.BytesSent, - BytesReceived: snap.stats.BytesReceived, - } - } - } - for enrichedKey, stats := range aggregatedStats { - workload_src := c.srcWorkload - workload_dest := enrichedKey.GetDestinationWorkload() - actualDestWorkload := enrichedKey.GetActualDestinationWorkload() + // --- TCP/connection metrics: push-model, no c.lock --- + c.tcpMetrics.collect(ch) - ch <- c.counter(metrics.NetConnectionsSuccessful, float64(stats.Count), enrichedKey.DestinationLabelValue(), enrichedKey.ActualDestinationLabelValue(), workload_src.Name, workload_src.Namespace, workload_src.Kind, workload_dest.Name, workload_dest.Namespace, workload_dest.Kind, actualDestWorkload.Name, actualDestWorkload.Namespace, actualDestWorkload.Kind) - ch <- c.counter(metrics.NetConnectionsTotalTime, stats.TotalTime.Seconds(), enrichedKey.DestinationLabelValue(), enrichedKey.ActualDestinationLabelValue(), workload_src.Name, workload_src.Namespace, workload_src.Kind, workload_dest.Name, workload_dest.Namespace, workload_dest.Kind, actualDestWorkload.Name, actualDestWorkload.Namespace, actualDestWorkload.Kind) - if stats.Retransmissions > 0 { - ch <- c.counter(metrics.NetRetransmits, float64(stats.Retransmissions), enrichedKey.DestinationLabelValue(), enrichedKey.ActualDestinationLabelValue(), workload_src.Name, workload_src.Namespace, workload_src.Kind, workload_dest.Name, workload_dest.Namespace, workload_dest.Kind, actualDestWorkload.Name, actualDestWorkload.Namespace, actualDestWorkload.Kind) - } - ch <- c.counter(metrics.NetBytesSent, float64(stats.BytesSent), enrichedKey.DestinationLabelValue(), enrichedKey.ActualDestinationLabelValue(), workload_src.Name, workload_src.Namespace, workload_src.Kind, workload_dest.Name, workload_dest.Namespace, workload_dest.Kind, actualDestWorkload.Name, actualDestWorkload.Namespace, actualDestWorkload.Kind) - ch <- c.counter(metrics.NetBytesReceived, float64(stats.BytesReceived), enrichedKey.DestinationLabelValue(), enrichedKey.ActualDestinationLabelValue(), workload_src.Name, workload_src.Namespace, workload_src.Kind, workload_dest.Name, workload_dest.Namespace, workload_dest.Kind, actualDestWorkload.Name, actualDestWorkload.Namespace, actualDestWorkload.Kind) - } - // Copy failedConnectionAttempts under read lock to avoid concurrent map access - c.lock.RLock() - failedConnCopy := make(map[common.HostPort]int64, len(c.failedConnectionAttempts)) - for k, v := range c.failedConnectionAttempts { - failedConnCopy[k] = v - } - c.lock.RUnlock() - - for dst, count := range failedConnCopy { - workload := c.ip_resolver.ResolveIP(dst.IP().String()) - ch <- c.counter(metrics.NetConnectionsFailed, float64(count), dst.String(), workload.Name, workload.Namespace, workload.Kind, workload.Name, workload.Namespace, workload.Kind) - } - - connections := map[common.DestinationKey]int{} - c.lock.RLock() - for _, conn := range c.activeConnections { - if !conn.Closed.IsZero() { - continue - } - connections[c.enrichDestinationKey(conn.DestinationKey)]++ - } - c.lock.RUnlock() - for enrichedKey, count := range connections { - actualDestWorkload := enrichedKey.GetActualDestinationWorkload() - destWorkload := enrichedKey.GetDestinationWorkload() - ch <- c.gauge(metrics.NetConnectionsActive, float64(count), enrichedKey.DestinationLabelValue(), enrichedKey.ActualDestinationLabelValue(), c.srcWorkload.Name, c.srcWorkload.Namespace, c.srcWorkload.Kind, destWorkload.Name, destWorkload.Namespace, destWorkload.Kind, actualDestWorkload.Name, actualDestWorkload.Namespace, actualDestWorkload.Kind) - } - - // Copy logParsers under read lock to avoid concurrent map access + // --- Log metrics: use sync.Map for samples, brief c.lock for logParsers snapshot --- c.lock.RLock() logParsersCopy := make(map[string]*LogParser, len(c.logParsers)) for k, v := range c.logParsers { @@ -505,19 +397,8 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) { for source, p := range logParsersCopy { for _, ctr := range p.parser.GetCounters() { if ctr.Level == logparser.LevelCritical || ctr.Level == logparser.LevelError { - c.lock.RLock() - sample, ok := c.logSamples[ctr.Hash] - c.lock.RUnlock() - if !ok { - c.lock.Lock() - sample, ok = c.logSamples[ctr.Hash] - if !ok { - sample = common.TruncateUtf8(ctr.Sample, *flags.MaxLabelLength) - c.logSamples[ctr.Hash] = sample - } - c.lock.Unlock() - } - ch <- c.counter(metrics.LogMessages, float64(ctr.Messages), source, ctr.Level.String(), ctr.Hash, sample) + sample, _ := c.logSamples.LoadOrStore(ctr.Hash, common.TruncateUtf8(ctr.Sample, *flags.MaxLabelLength)) + ch <- c.counter(metrics.LogMessages, float64(ctr.Messages), source, ctr.Level.String(), ctr.Hash, sample.(string)) } } for _, sc := range p.parser.GetSensitiveCounters() { @@ -525,11 +406,11 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) { } } + // --- Process-level metrics: brief snapshot under c.lock --- appTypes := map[string]struct{}{} seenJvms := map[string]bool{} seenDotNetApps := map[string]bool{} - // Copy processes map under read lock to avoid concurrent map access c.lock.RLock() processesCopy := make(map[uint32]*Process, len(c.processes)) for pid, p := range c.processes { @@ -608,6 +489,7 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) { ch <- c.counter(metrics.NodejsEventLoopBlockedTime, c.nodejsStats.EventLoopBlockedTime.Seconds()) } + // --- L7 metrics: push-model, own lock --- c.l7Stats.collect(ch) if !*flags.DisablePinger { @@ -650,7 +532,7 @@ func (c *Container) onProcessStart(pid uint32) *Process { } } if min.After(c.startedAt) { - c.restarts++ + c.tcpMetrics.ObserveRestart() c.startedAt = min } } @@ -669,7 +551,7 @@ func (c *Container) onProcessExit(pid uint32, oomKill bool) { } delete(c.delaysByPid, pid) if oomKill { - c.oomKills++ + c.tcpMetrics.ObserveOOMKill() } } @@ -812,23 +694,18 @@ func (c *Container) onConnectionOpen(pid uint32, fd uint64, src, dst, actualDst return } - c.lock.Lock() - defer c.lock.Unlock() - - // Fast Path (or slow path that found no match): - // Proceed with creating the key and updating stats as usual. key := common.NewDestinationKey(dst, actualDst, c.registry.getDomain(dst.IP()), dstWorkload, actualDstWorkload) if failed { - c.failedConnectionAttempts[key.Destination()]++ + c.tcpMetrics.ObserveConnectionFailed(key.Destination(), dstWorkload) } else { - stats, _ := c.connectionStats.Get(key) - if stats == nil { - stats = &ConnectionStats{} - c.connectionStats.Add(key, stats) - } - stats.Count++ - stats.TotalTime += duration + c.tcpMetrics.ObserveConnectionOpen(key, srcWorkload, duration.Seconds()) + } + + c.lock.Lock() + defer c.lock.Unlock() + + if !failed { connection := &ActiveConnection{ DestinationKey: key, Pid: pid, @@ -940,19 +817,18 @@ func (c *Container) updateConnectionTrafficStats(ac *ActiveConnection, sent, rec return } c.migrateConnectionKeyIfNeeded(ac) - stats, _ := c.connectionStats.Get(ac.DestinationKey) - if stats == nil { - stats = &ConnectionStats{} - c.connectionStats.Add(ac.DestinationKey, stats) - } + var sentDelta, recvDelta uint64 if sent > ac.BytesSent { - stats.BytesSent += sent - ac.BytesSent + sentDelta = sent - ac.BytesSent } if received > ac.BytesReceived { - stats.BytesReceived += received - ac.BytesReceived + recvDelta = received - ac.BytesReceived } ac.BytesSent = sent ac.BytesReceived = received + if sentDelta > 0 || recvDelta > 0 { + c.tcpMetrics.ObserveTraffic(ac.DestinationKey, ac.srcWorkload, sentDelta, recvDelta) + } } func (c *Container) trackLLMRequest(provider LLMProvider, host, path, payloadBase64, responseBase64 string, duration time.Duration) { @@ -1114,37 +990,15 @@ func (c *Container) enrichDestinationKey(key common.DestinationKey) common.Desti return key } -// migrateConnectionKeyToFQDN updates conn.DestinationKey from IP to the given FQDN -// and migrates the corresponding connectionStats entry. +// migrateConnectionKeyToFQDN updates conn.DestinationKey from IP to the given FQDN. // Must be called under c.lock. func (c *Container) migrateConnectionKeyToFQDN(conn *ActiveConnection, fqdn string) { - oldKey := conn.DestinationKey - newKey := oldKey.WithResolvedDomain(fqdn) - if oldKey == newKey { - return - } - - // Migrate connectionStats from old key to new key - oldStats, _ := c.connectionStats.Get(oldKey) - if oldStats != nil { - c.connectionStats.Remove(oldKey) - newStats, _ := c.connectionStats.Get(newKey) - if newStats != nil { - newStats.Count += oldStats.Count - newStats.TotalTime += oldStats.TotalTime - newStats.Retransmissions += oldStats.Retransmissions - newStats.BytesSent += oldStats.BytesSent - newStats.BytesReceived += oldStats.BytesReceived - } else { - c.connectionStats.Add(newKey, oldStats) - } - } - + newKey := conn.DestinationKey.WithResolvedDomain(fqdn) conn.DestinationKey = newKey } // migrateConnectionKeyIfNeeded updates conn.DestinationKey from IP to FQDN when -// DNS becomes available, and migrates the corresponding connectionStats entry. +// DNS becomes available. // Must be called under c.lock. func (c *Container) migrateConnectionKeyIfNeeded(conn *ActiveConnection) { if conn == nil { @@ -1578,19 +1432,49 @@ func (c *Container) processHTTP2WithoutConnection(pid uint32, fd uint64, r *l7.R return nil, L7RequestProcessed } +// refreshActiveConnections snapshots active connections under c.lock and +// pushes the gauge values to TCPMetrics. Called periodically from the +// event handler goroutine (not from Collect). +func (c *Container) refreshActiveConnections() { + c.lock.RLock() + counts := map[common.DestinationKey]activeConnAgg{} + for _, conn := range c.activeConnections { + if !conn.Closed.IsZero() { + continue + } + enrichedKey := c.enrichDestinationKey(conn.DestinationKey) + agg, ok := counts[enrichedKey] + if !ok { + agg = activeConnAgg{src: conn.srcWorkload} + } + agg.count++ + counts[enrichedKey] = agg + } + c.lock.RUnlock() + + entries := make([]activeEntry, 0, len(counts)) + for key, agg := range counts { + entries = append(entries, activeEntry{ + labels: tcpLabels(key, agg.src), + count: agg.count, + }) + } + c.tcpMetrics.resetAndSetActive(entries) +} + +type activeConnAgg struct { + src common.Workload + count int +} + func (c *Container) onRetransmission(src netaddr.IPPort, dst netaddr.IPPort) bool { - c.lock.Lock() - defer c.lock.Unlock() + c.lock.RLock() conn, ok := c.activeConnections[ConnectionKey{src: src, dst: dst}] + c.lock.RUnlock() if !ok { return false } - stats, _ := c.connectionStats.Get(conn.DestinationKey) - if stats == nil { - stats = &ConnectionStats{} - c.connectionStats.Add(conn.DestinationKey, stats) - } - stats.Retransmissions++ + c.tcpMetrics.ObserveRetransmission(conn.DestinationKey, conn.srcWorkload) return true } @@ -1702,14 +1586,12 @@ func (c *Container) getMounts() map[string]map[string]*proc.FSStat { return res } +// getListens must be called with c.lock held (at least RLock). func (c *Container) getListens() map[netaddr.IPPort]int { - // Copy processes under read lock — c.processes is mutated by handleEvents goroutine - c.lock.RLock() processesCopy := make(map[uint32]*Process, len(c.processes)) for pid, p := range c.processes { processesCopy[pid] = p } - c.lock.RUnlock() res := map[netaddr.IPPort]int{} for addr, byPid := range c.listens { @@ -1823,16 +1705,13 @@ func (c *Container) ping() map[netaddr.IP]float64 { } ips := map[netaddr.IP]struct{}{} - for _, d := range c.connectionStats.Keys() { - if ip := d.ActualDestination().IP(); !ip.IsZero() { + c.lock.RLock() + for _, conn := range c.activeConnections { + if ip := conn.DestinationKey.ActualDestinationIfKnown().IP(); !ip.IsZero() { ips[ip] = struct{}{} } } - for dst := range c.failedConnectionAttempts { - if ip := dst.IP(); !ip.IsZero() { - ips[dst.IP()] = struct{}{} - } - } + c.lock.RUnlock() if len(ips) == 0 { return nil } @@ -1984,12 +1863,6 @@ func (c *Container) gc(now time.Time) { _, active := establishedDst[dst] if !active && !at.IsZero() && now.Sub(at) > gcInterval { delete(c.lastConnectionAttempts, dst) - delete(c.failedConnectionAttempts, dst) - for _, d := range c.connectionStats.Keys() { - if d.Destination() == dst { - c.connectionStats.Remove(d) - } - } c.l7Stats.delete(dst) } } diff --git a/containers/registry.go b/containers/registry.go index 1e7bb8d7..891d71dd 100644 --- a/containers/registry.go +++ b/containers/registry.go @@ -83,12 +83,6 @@ type Registry struct { processInfoCh chan<- ProcessInfo ip_resolver IPResolver - ebpfStatsLastUpdated time.Time - ebpfStatsLock sync.Mutex - trafficStatsUpdateCh chan *TrafficStatsUpdate - nodejsStatsUpdateCh chan *NodejsStatsUpdate - pythonStatsUpdateCh chan *PythonStatsUpdate - gpuProcessUsageSampleChan chan gpu.ProcessUsageSample // nodeConstLabels holds node-level label values for embedding directly @@ -156,12 +150,9 @@ func NewRegistry(reg prometheus.Registerer, rawReg prometheus.Registerer, proces containersByPidIgnored: map[uint32]*time.Time{}, ip2fqdn: map[netaddr.IP]*common.Domain{}, - processInfoCh: processInfoCh, - ip_resolver: ip_resolver, - tracer: ebpftracer.NewTracer(hostNetNs, selfNetNs, *flags.DisableL7Tracing), - trafficStatsUpdateCh: make(chan *TrafficStatsUpdate), - nodejsStatsUpdateCh: make(chan *NodejsStatsUpdate), - pythonStatsUpdateCh: make(chan *PythonStatsUpdate), + processInfoCh: processInfoCh, + ip_resolver: ip_resolver, + tracer: ebpftracer.NewTracer(hostNetNs, selfNetNs, *flags.DisableL7Tracing), gpuProcessUsageSampleChan: gpuProcessUsageSampleChan, nodeConstLabels: NodeConstLabels{MachineID: machineId, SystemUUID: systemUuid, AZ: az, Region: region}, @@ -185,7 +176,6 @@ func (r *Registry) Describe(ch chan<- *prometheus.Desc) { } func (r *Registry) Collect(ch chan<- prometheus.Metric) { - r.updateStatsFromEbpfMapsIfNecessary() r.ip2fqdnLock.RLock() defer r.ip2fqdnLock.RUnlock() for ip, domain := range r.ip2fqdn { @@ -201,8 +191,12 @@ func (r *Registry) Close() { func (r *Registry) handleEvents(ch <-chan ebpftracer.Event) { gcTicker := time.NewTicker(gcInterval) defer gcTicker.Stop() + ebpfStatsTicker := time.NewTicker(MinTrafficStatsUpdateInterval) + defer ebpfStatsTicker.Stop() for { select { + case <-ebpfStatsTicker.C: + r.updateEbpfStatsAndActiveConns() case now := <-gcTicker.C: for pid, c := range r.containersByPid { cg, err := proc.ReadCgroup(pid) @@ -272,36 +266,6 @@ func (r *Registry) handleEvents(ch <-chan ebpftracer.Event) { } } r.ip2fqdnLock.Unlock() - case u := <-r.trafficStatsUpdateCh: - if u == nil { - continue - } - r.containerLock.RLock() - c := r.containersByPid[u.Pid] - r.containerLock.RUnlock() - if c != nil { - c.updateTrafficStats(u) - } - case u := <-r.nodejsStatsUpdateCh: - if u == nil { - continue - } - r.containerLock.RLock() - c := r.containersByPid[u.Pid] - r.containerLock.RUnlock() - if c != nil { - c.updateNodejsStats(*u) - } - case u := <-r.pythonStatsUpdateCh: - if u == nil { - continue - } - r.containerLock.RLock() - c := r.containersByPid[u.Pid] - r.containerLock.RUnlock() - if c != nil { - c.updatePythonStats(*u) - } case sample := <-r.gpuProcessUsageSampleChan: r.containerLock.RLock() c := r.containersByPid[sample.Pid] @@ -687,70 +651,72 @@ func (r *Registry) getOrCreateContainer(pid uint32) *Container { return c } -func (r *Registry) updateStatsFromEbpfMapsIfNecessary() { - if !r.ebpfStatsLock.TryLock() { - return // Skip update if another one is already in progress. - } - defer r.ebpfStatsLock.Unlock() - - if time.Now().Sub(r.ebpfStatsLastUpdated) < MinTrafficStatsUpdateInterval { - return - } - - r.updateTrafficStats() - r.updateNodejsStats() - r.updatePythonStats() - - r.ebpfStatsLastUpdated = time.Now() -} - -func (r *Registry) updateTrafficStats() { +// updateEbpfStatsAndActiveConns reads eBPF maps and updates container stats +// directly from the event handler goroutine. Also refreshes active connection gauges. +func (r *Registry) updateEbpfStatsAndActiveConns() { + // Traffic stats from eBPF maps iter := r.tracer.ActiveConnectionsIterator() cid := ebpftracer.ConnectionId{} stats := ebpftracer.Connection{} for iter.Next(&cid, &stats) { - r.trafficStatsUpdateCh <- &TrafficStatsUpdate{ - Pid: cid.PID, - FD: cid.FD, - BytesSent: stats.BytesSent, - BytesReceived: stats.BytesReceived, - Protocol: stats.Protocol, + r.containerLock.RLock() + c := r.containersByPid[cid.PID] + r.containerLock.RUnlock() + if c != nil { + c.updateTrafficStats(&TrafficStatsUpdate{ + Pid: cid.PID, + FD: cid.FD, + BytesSent: stats.BytesSent, + BytesReceived: stats.BytesReceived, + Protocol: stats.Protocol, + }) } } if err := iter.Err(); err != nil { klog.Warningln(err) } - r.trafficStatsUpdateCh <- nil -} -func (r *Registry) updateNodejsStats() { - iter := r.tracer.NodejsStatsIterator() + // Node.js stats + njsIter := r.tracer.NodejsStatsIterator() var pid uint64 - stats := ebpftracer.NodejsStats{} - - for iter.Next(&pid, &stats) { - r.nodejsStatsUpdateCh <- &NodejsStatsUpdate{Pid: uint32(pid), Stats: stats} + njsStats := ebpftracer.NodejsStats{} + for njsIter.Next(&pid, &njsStats) { + r.containerLock.RLock() + c := r.containersByPid[uint32(pid)] + r.containerLock.RUnlock() + if c != nil { + c.updateNodejsStats(NodejsStatsUpdate{Pid: uint32(pid), Stats: njsStats}) + } } - - if err := iter.Err(); err != nil { + if err := njsIter.Err(); err != nil { klog.Warningln(err) } - r.nodejsStatsUpdateCh <- nil -} -func (r *Registry) updatePythonStats() { - iter := r.tracer.PythonStatsIterator() - var pid uint64 - stats := ebpftracer.PythonStats{} - - for iter.Next(&pid, &stats) { - r.pythonStatsUpdateCh <- &PythonStatsUpdate{Pid: uint32(pid), Stats: stats} + // Python stats + pyIter := r.tracer.PythonStatsIterator() + pyStats := ebpftracer.PythonStats{} + for pyIter.Next(&pid, &pyStats) { + r.containerLock.RLock() + c := r.containersByPid[uint32(pid)] + r.containerLock.RUnlock() + if c != nil { + c.updatePythonStats(PythonStatsUpdate{Pid: uint32(pid), Stats: pyStats}) + } } - - if err := iter.Err(); err != nil { + if err := pyIter.Err(); err != nil { klog.Warningln(err) } - r.pythonStatsUpdateCh <- nil + + // Refresh active connection gauges for all containers + 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 { + c.refreshActiveConnections() + } } func (r *Registry) getDomain(ip netaddr.IP) *common.Domain { diff --git a/containers/tcp_metrics.go b/containers/tcp_metrics.go new file mode 100644 index 00000000..d3e20fdd --- /dev/null +++ b/containers/tcp_metrics.go @@ -0,0 +1,172 @@ +package containers + +import ( + "sync" + + "github.com/coroot/coroot-node-agent/common" + "github.com/prometheus/client_golang/prometheus" +) + +// TCPMetrics holds pre-registered CounterVec/GaugeVec for TCP connection metrics. +// 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 + + successful *prometheus.CounterVec + totalTime *prometheus.CounterVec + failed *prometheus.CounterVec + retransmits *prometheus.CounterVec + bytesSent *prometheus.CounterVec + bytesRecv *prometheus.CounterVec + active *prometheus.GaugeVec + restarts prometheus.Counter + oomKills prometheus.Counter + + constLabels prometheus.Labels + initialized bool +} + +// tcpVarLabels are shared across most TCP connection metrics (11 labels). +var tcpVarLabels = []string{ + "destination", "actual_destination", + "src_workload_name", "src_workload_namespace", "src_workload_kind", + "destination_workload_name", "destination_workload_namespace", "destination_workload_kind", + "actual_destination_workload_name", "actual_destination_workload_namespace", "actual_destination_workload_kind", +} + +// tcpFailedVarLabels are used for failed connection metrics (7 labels). +var tcpFailedVarLabels = []string{ + "destination", + "destination_workload_name", "destination_workload_namespace", "destination_workload_kind", + "actual_destination_workload_name", "actual_destination_workload_namespace", "actual_destination_workload_kind", +} + +func NewTCPMetrics(constLabels prometheus.Labels) *TCPMetrics { + return &TCPMetrics{ + constLabels: constLabels, + } +} + +func (t *TCPMetrics) ensureInitialized() { + t.mu.RLock() + if t.initialized { + t.mu.RUnlock() + return + } + t.mu.RUnlock() + + t.mu.Lock() + defer t.mu.Unlock() + if t.initialized { + return + } + + cl := t.constLabels + + t.successful = newCounterVec("container_net_tcp_successful_connects_total", "Total number of successful TCP connects", cl, tcpVarLabels...) + t.totalTime = newCounterVec("container_net_tcp_connection_time_seconds_total", "Time spent on TCP connections", cl, tcpVarLabels...) + t.failed = newCounterVec("container_net_tcp_failed_connects_total", "Total number of failed TCP connects", cl, tcpFailedVarLabels...) + t.retransmits = newCounterVec("container_net_tcp_retransmits_total", "Total number of retransmitted TCP segments", cl, tcpVarLabels...) + t.bytesSent = newCounterVec("container_net_tcp_bytes_sent_total", "Total number of bytes sent to the peer", cl, tcpVarLabels...) + t.bytesRecv = newCounterVec("container_net_tcp_bytes_received_total", "Total number of bytes received from the peer", cl, tcpVarLabels...) + t.active = newGaugeVec("container_net_tcp_active_connections", "Number of active outbound connections used by the container", cl, tcpVarLabels...) + 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.initialized = true +} + +// tcpLabels builds the 11-label value slice for TCP metrics. +func tcpLabels(key common.DestinationKey, src common.Workload) []string { + dest := key.GetDestinationWorkload() + actualDest := key.GetActualDestinationWorkload() + return []string{ + key.DestinationLabelValue(), key.ActualDestinationLabelValue(), + src.Name, src.Namespace, src.Kind, + dest.Name, dest.Namespace, dest.Kind, + actualDest.Name, actualDest.Namespace, actualDest.Kind, + } +} + +// ObserveConnectionOpen records a successful connection open. +func (t *TCPMetrics) ObserveConnectionOpen(key common.DestinationKey, src common.Workload, durationSeconds float64) { + t.ensureInitialized() + labels := tcpLabels(key, src) + 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( + dst.String(), + workload.Name, workload.Namespace, workload.Kind, + workload.Name, workload.Namespace, workload.Kind, + ).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() +} + +// 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) + if sentDelta > 0 { + t.bytesSent.WithLabelValues(labels...).Add(float64(sentDelta)) + } + if recvDelta > 0 { + t.bytesRecv.WithLabelValues(labels...).Add(float64(recvDelta)) + } +} + +// resetAndSetActive replaces all active connection gauge values. +// Called from the event handler goroutine periodically. +func (t *TCPMetrics) resetAndSetActive(entries []activeEntry) { + t.ensureInitialized() + t.active.Reset() + for _, e := range entries { + t.active.WithLabelValues(e.labels...).Set(float64(e.count)) + } +} + +// activeEntry holds pre-computed label values and count for active connections. +type activeEntry struct { + labels []string + count int +} + +// ObserveRestart increments the restart counter. +func (t *TCPMetrics) ObserveRestart() { + t.ensureInitialized() + t.restarts.Inc() +} + +// ObserveOOMKill increments the OOM kill counter. +func (t *TCPMetrics) ObserveOOMKill() { + t.ensureInitialized() + t.oomKills.Inc() +} + +// collect forwards all pre-built metrics to the channel. No c.lock needed. +func (t *TCPMetrics) collect(ch chan<- prometheus.Metric) { + t.mu.RLock() + defer t.mu.RUnlock() + if !t.initialized { + return + } + t.successful.Collect(ch) + t.totalTime.Collect(ch) + t.failed.Collect(ch) + t.retransmits.Collect(ch) + t.bytesSent.Collect(ch) + t.bytesRecv.Collect(ch) + t.active.Collect(ch) + t.restarts.Collect(ch) + t.oomKills.Collect(ch) +} From 0b01e86ba07c0a1c090e058c07ea6a1ed42e5313 Mon Sep 17 00:00:00 2001 From: mayankpande88 Date: Sun, 8 Mar 2026 10:59:26 +0530 Subject: [PATCH 5/5] fix: use single containerCollector instead of per-container registration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit prometheus.Registry.Unregister does not work for unchecked collectors (empty Describe). It only searches collectorsByID, never the uncheckedCollectors slice. This meant every container replacement (process restart within same pod) leaked the old Container object in the registry — Gather() collected duplicate metrics and returned HTTP 500. Register a single containerCollector on the raw registry that iterates containersById under RLock. Individual container Register/Unregister calls removed entirely. --- containers/registry.go | 45 ++++++++++++++++++++++++++++++------------ 1 file changed, 32 insertions(+), 13 deletions(-) diff --git a/containers/registry.go b/containers/registry.go index 891d71dd..88820dc4 100644 --- a/containers/registry.go +++ b/containers/registry.go @@ -66,8 +66,7 @@ func (n NodeConstLabels) Values() []string { } type Registry struct { - reg prometheus.Registerer // wrapped registerer for ip2fqdn and LLM metrics - rawReg prometheus.Registerer // raw registry for containers (no wrapping overhead) + reg prometheus.Registerer // wrapped registerer for ip2fqdn and LLM metrics tracer *ebpftracer.Tracer events chan ebpftracer.Event @@ -142,7 +141,6 @@ func NewRegistry(reg prometheus.Registerer, rawReg prometheus.Registerer, proces r := &Registry{ reg: reg, - rawReg: rawReg, events: make(chan ebpftracer.Event, 2000), containersById: map[ContainerID]*Container{}, containersByCgroupId: map[string]*Container{}, @@ -162,6 +160,12 @@ func NewRegistry(reg prometheus.Registerer, rawReg prometheus.Registerer, proces if err = reg.Register(r); err != nil { return nil, err } + // Register a single collector for all containers on the raw registry. + // Individual containers are NOT registered/unregistered because prometheus.Registry + // does not support Unregister for unchecked collectors (empty Describe). + if err = rawReg.Register(&containerCollector{registry: r}); err != nil { + return nil, err + } go r.handleEvents(r.events) if err = r.tracer.Run(r.events); err != nil { close(r.events) @@ -183,6 +187,31 @@ func (r *Registry) Collect(ch chan<- prometheus.Metric) { } } +// containerCollector is registered on the raw Prometheus registry and collects +// metrics from all known containers. This avoids registering each Container as +// an individual unchecked collector, which is needed because prometheus.Registry +// does not support Unregister for unchecked collectors (Describe returns nothing). +type containerCollector struct { + registry *Registry +} + +func (cc *containerCollector) Describe(chan<- *prometheus.Desc) { + // Unchecked collector: container metrics are dynamic +} + +func (cc *containerCollector) Collect(ch chan<- prometheus.Metric) { + cc.registry.containerLock.RLock() + containers := make([]*Container, 0, len(cc.registry.containersById)) + for _, c := range cc.registry.containersById { + containers = append(containers, c) + } + cc.registry.containerLock.RUnlock() + + for _, c := range containers { + c.Collect(ch) + } +} + func (r *Registry) Close() { r.tracer.Close() close(r.events) @@ -250,11 +279,6 @@ func (r *Registry) handleEvents(ch <-chan ebpftracer.Event) { } } delete(r.containersById, c.id) - - // Unregister from Prometheus (do this after removing from maps) - if ok := r.rawReg.Unregister(c); !ok { - klog.Warningf("failed to unregister container: %s", c.id) - } c.Close() } r.containerLock.Unlock() @@ -617,7 +641,6 @@ func (r *Registry) getOrCreateContainer(pid uint32) *Container { } if c := r.containersById[id]; c != nil { klog.Warningln("id conflict, replacing container:", id) - r.rawReg.Unregister(c) delete(r.containersById, c.id) for cgid, container := range r.containersByCgroupId { if container == c { @@ -639,10 +662,6 @@ func (r *Registry) getOrCreateContainer(pid uint32) *Container { return nil } klog.InfoS("detected a new container", "pid", pid, "cg", cg.Id, "id", id, "app", c.appId) - if err := r.rawReg.Register(c); err != nil { - klog.Warningf("failed to register container %s: %v", id, err) - return nil - } // Update all maps atomically while holding the lock r.containersByPid[pid] = c