From a74ad13fe59af488a439570298dbe1eb7508360e Mon Sep 17 00:00:00 2001 From: mayankpande88 Date: Tue, 10 Mar 2026 12:14:51 +0530 Subject: [PATCH 1/7] fix: update logparser to v0.0.0-20260310062405-9d6258bdd771 Picks up fix/json-log-pattern-hashing (PR #16) which stabilizes JSON log pattern hashing by extracting only message fields. --- go.mod | 2 +- go.sum | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/go.mod b/go.mod index 2e8b940..735b76b 100644 --- a/go.mod +++ b/go.mod @@ -21,7 +21,7 @@ require ( github.com/hashicorp/golang-lru/v2 v2.0.7 github.com/jpillora/backoff v1.0.0 github.com/mdlayher/taskstats v0.0.0-20230712191918-387b3d561d14 - github.com/nudgebee/logparser v0.0.0-20260218041043-99ea4437d928 + github.com/nudgebee/logparser v0.0.0-20260310062405-9d6258bdd771 github.com/opencontainers/runtime-spec v1.1.0 github.com/prometheus/client_golang v1.20.5 github.com/prometheus/client_model v0.6.1 diff --git a/go.sum b/go.sum index 35f7c4c..4379105 100644 --- a/go.sum +++ b/go.sum @@ -323,8 +323,8 @@ github.com/morikuni/aec v1.0.0 h1:nP9CBfwrvYnBRgY6qfDQkygYDmYwOilePFkwzv4dU8A= github.com/morikuni/aec v1.0.0/go.mod h1:BbKIizmSmc5MMPqRYbxO4ZU0S0+P200+tUnFx7PXmsc= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= -github.com/nudgebee/logparser v0.0.0-20260218041043-99ea4437d928 h1:KExtH0Z1wo+yJgY2EJbZhF0YRLNcYZVASSN5gtZ513o= -github.com/nudgebee/logparser v0.0.0-20260218041043-99ea4437d928/go.mod h1:oFnM9D6YEjZzb1jy0kJ/NUkTbkZA+BgNYjpLO4n4szA= +github.com/nudgebee/logparser v0.0.0-20260310062405-9d6258bdd771 h1:DfyX+uJ3B1x/Bz0Ie7jtLi9UoIP5PCC5myZCk6JkfmM= +github.com/nudgebee/logparser v0.0.0-20260310062405-9d6258bdd771/go.mod h1:oFnM9D6YEjZzb1jy0kJ/NUkTbkZA+BgNYjpLO4n4szA= github.com/oklog/ulid v1.3.1 h1:EGfNDEx6MqHz8B3uNV6QAib1UR2Lm97sHi3ocA6ESJ4= github.com/oklog/ulid v1.3.1/go.mod h1:CirwcVhetQ6Lv90oh/F+FBtV6XMibvdAFo93nm5qn4U= github.com/onsi/ginkgo v1.16.5 h1:8xi0RTUf59SOSfEtZMvwTvXYMzG4gV23XVHOZiXNtnE= From 21075bc8a06ddf5575319c58319643c240cd273c Mon Sep 17 00:00:00 2001 From: Mayank Pande Date: Fri, 13 Mar 2026 16:12:11 +0530 Subject: [PATCH 2/7] fix: configurable sensitive log detection with optimized logparser (#224) * fix: update logparser and add configurable sensitive log detection flags - Update logparser to v0.0.0-20260313095946-b22168b95d9c with optimized sensitive parsing (anchor pre-filter, confidence tiers, singleton cache) - Add SENSITIVE_LOG_SAMPLE_RATE, SENSITIVE_LOG_MIN_CONFIDENCE, SENSITIVE_LOG_MAX_DETECTIONS flags for fine-grained control - Update all NewParser callsites to use SensitiveConfig struct - Remove local replace directive * fix: deduplicate SensitiveConfig and align flag naming - Extract SensitiveConfig into a single variable reused across all 3 NewParser calls - Rename flag/env var to sensitive-log-max-detections-per-container / SENSITIVE_LOG_MAX_DETECTIONS_PER_CONTAINER to match variable name --- common/log_parser_test.go | 6 +++++- containers/container.go | 12 +++++++++--- flags/flags.go | 25 ++++++++++++++----------- go.mod | 2 +- go.sum | 4 ++-- 5 files changed, 31 insertions(+), 18 deletions(-) diff --git a/common/log_parser_test.go b/common/log_parser_test.go index 34924d5..af2f685 100644 --- a/common/log_parser_test.go +++ b/common/log_parser_test.go @@ -12,7 +12,11 @@ import ( func TestLogParser(t *testing.T) { const defaultPatternsPerLevel = 256 ch := make(chan logparser.LogEntry) - parser := logparser.NewParser(ch, nil, nil, 1*time.Second, defaultPatternsPerLevel, false) + parser := logparser.NewParser(ch, nil, nil, 1*time.Second, defaultPatternsPerLevel, logparser.SensitiveConfig{ + Enabled: true, + MinConfidence: "high", + MaxDetections: 100, + }) ch <- logparser.LogEntry{Timestamp: time.Now(), Content: "INFO:root:AWS access key: AKIAIOSFODNN7EXAMPLE", Level: logparser.LevelInfo} diff --git a/containers/container.go b/containers/container.go index 445a6f0..bb74553 100644 --- a/containers/container.go +++ b/containers/container.go @@ -1746,13 +1746,19 @@ func (c *Container) runLogParser(logPath string) { } containerId := string(c.id) + sensitiveCfg := logparser.SensitiveConfig{ + Enabled: !*flags.DisableSensitiveLogParsing, + SampleRate: *flags.SensitiveLogSampleRate, + MinConfidence: *flags.SensitiveLogMinConfidence, + MaxDetections: *flags.SensitiveLogMaxDetectionsPerContainer, + } if logPath != "" { if c.logParsers[logPath] != nil { return } ch := make(chan logparser.LogEntry) - parser := logparser.NewParser(ch, nil, logs.OtelLogEmitter(containerId), multilineCollectorTimeout, *flags.LogPatternsPerContainer, *flags.DisableSensitiveLogParsing) + parser := logparser.NewParser(ch, nil, logs.OtelLogEmitter(containerId), multilineCollectorTimeout, *flags.LogPatternsPerContainer, sensitiveCfg) reader, err := logs.NewTailReader(proc.HostPath(logPath), ch) if err != nil { klog.Warningln(err) @@ -1772,7 +1778,7 @@ func (c *Container) runLogParser(logPath string) { klog.Warningln(err) return } - parser := logparser.NewParser(ch, nil, logs.OtelLogEmitter(containerId), multilineCollectorTimeout, *flags.LogPatternsPerContainer, *flags.DisableSensitiveLogParsing) + parser := logparser.NewParser(ch, nil, logs.OtelLogEmitter(containerId), multilineCollectorTimeout, *flags.LogPatternsPerContainer, sensitiveCfg) stop := func() { JournaldUnsubscribe(c.metadata.systemd.Unit) } @@ -1789,7 +1795,7 @@ func (c *Container) runLogParser(logPath string) { delete(c.logParsers, "stdout/stderr") } ch := make(chan logparser.LogEntry) - parser := logparser.NewParser(ch, c.metadata.logDecoder, logs.OtelLogEmitter(containerId), multilineCollectorTimeout, *flags.LogPatternsPerContainer, *flags.DisableSensitiveLogParsing) + parser := logparser.NewParser(ch, c.metadata.logDecoder, logs.OtelLogEmitter(containerId), multilineCollectorTimeout, *flags.LogPatternsPerContainer, sensitiveCfg) reader, err := logs.NewTailReader(proc.HostPath(c.metadata.logPath), ch) if err != nil { klog.Warningln(err) diff --git a/flags/flags.go b/flags/flags.go index d460b74..9a256ca 100644 --- a/flags/flags.go +++ b/flags/flags.go @@ -52,17 +52,20 @@ var ( ProfilesEndpoint = kingpin.Flag("profiles-endpoint", "The URL of the endpoint to send profiles to").Envar("PROFILES_ENDPOINT").URL() InsecureSkipVerify = kingpin.Flag("insecure-skip-verify", "whether to skip verifying the certificate or not").Envar("INSECURE_SKIP_VERIFY").Default("false").Bool() - ScrapeInterval = kingpin.Flag("scrape-interval", "How often to gather metrics from the agent").Default("15s").Envar("SCRAPE_INTERVAL").Duration() - WalDir = kingpin.Flag("wal-dir", "Path to where the agent stores data (e.g. the metrics Write-Ahead Log)").Default("/tmp/coroot-node-agent").Envar("WAL_DIR").String() - MaxSpoolSize = kingpin.Flag("max-spool-size", "Maximum size of the on-disk spool used to buffer data when it cannot be sent to collector. Supports size suffixes like KB, MB, or GB.").Default("500MB").Envar("MAX_SPOOL_SIZE").Bytes() - ResolveDns = kingpin.Flag("resolve-dns", "should resolve DNS").Default("false").Envar("RESOLVE_DNS").Bool() - IgnoreControlPlane = kingpin.Flag("ignore-control-plane", "ignore control plane like loki").Default("karpenter,loki,prometheus,grafana,kubelet,etcd,apiserver,victoria,nudgebee-agent,kube-system").Envar("IGNORE_CONTROL_PLANE").String() - SanitizeHeaders = kingpin.Flag("sanitize-headers", "should sanitize headers").Default("true").Envar("SANITIZE_HEADERS").Bool() - SensitiveHeader = kingpin.Flag("sensitive-headers", "sanitize headers using patterns").Default("Authorization, Cookie, X-Action-Token").Envar("SENSITIVE_HEADERS").String() - DisableKubeProbe = kingpin.Flag("disable-kube-probe", "disable kube probe trace").Default("true").Envar("DISABLE_KUBE_PROBE").Bool() - EnableDynamicLogTailing = kingpin.Flag("enable-dynamic-log-tailing", "Enable automatic tailing of log files opened by container processes.").Default("false").Envar("ENABLE_DYNAMIC_LOG_TAILING").Bool() - DisableSensitiveLogParsing = kingpin.Flag("disable-sensitive-log-parsing", "disable sensitive log parsing").Default("true").Envar("DISABLE_SENSITIVE_LOG_PARSING").Bool() - TraceIdHeaders = kingpin.Flag("trace-id-headers", "trace id headers").Default("Traceparent,X-Request-Id").Envar("TRACE_ID_HEADERS").String() + ScrapeInterval = kingpin.Flag("scrape-interval", "How often to gather metrics from the agent").Default("15s").Envar("SCRAPE_INTERVAL").Duration() + WalDir = kingpin.Flag("wal-dir", "Path to where the agent stores data (e.g. the metrics Write-Ahead Log)").Default("/tmp/coroot-node-agent").Envar("WAL_DIR").String() + MaxSpoolSize = kingpin.Flag("max-spool-size", "Maximum size of the on-disk spool used to buffer data when it cannot be sent to collector. Supports size suffixes like KB, MB, or GB.").Default("500MB").Envar("MAX_SPOOL_SIZE").Bytes() + ResolveDns = kingpin.Flag("resolve-dns", "should resolve DNS").Default("false").Envar("RESOLVE_DNS").Bool() + IgnoreControlPlane = kingpin.Flag("ignore-control-plane", "ignore control plane like loki").Default("karpenter,loki,prometheus,grafana,kubelet,etcd,apiserver,victoria,nudgebee-agent,kube-system").Envar("IGNORE_CONTROL_PLANE").String() + SanitizeHeaders = kingpin.Flag("sanitize-headers", "should sanitize headers").Default("true").Envar("SANITIZE_HEADERS").Bool() + SensitiveHeader = kingpin.Flag("sensitive-headers", "sanitize headers using patterns").Default("Authorization, Cookie, X-Action-Token").Envar("SENSITIVE_HEADERS").String() + DisableKubeProbe = kingpin.Flag("disable-kube-probe", "disable kube probe trace").Default("true").Envar("DISABLE_KUBE_PROBE").Bool() + EnableDynamicLogTailing = kingpin.Flag("enable-dynamic-log-tailing", "Enable automatic tailing of log files opened by container processes.").Default("false").Envar("ENABLE_DYNAMIC_LOG_TAILING").Bool() + DisableSensitiveLogParsing = kingpin.Flag("disable-sensitive-log-parsing", "disable sensitive log parsing").Default("true").Envar("DISABLE_SENSITIVE_LOG_PARSING").Bool() + SensitiveLogSampleRate = kingpin.Flag("sensitive-log-sample-rate", "Sample rate for sensitive log detection: check 1-in-N lines (0 or 1 = every line)").Default("100").Envar("SENSITIVE_LOG_SAMPLE_RATE").Int() + SensitiveLogMinConfidence = kingpin.Flag("sensitive-log-min-confidence", "Minimum confidence level for sensitive patterns: high, medium, low").Default("medium").Envar("SENSITIVE_LOG_MIN_CONFIDENCE").String() + SensitiveLogMaxDetectionsPerContainer = kingpin.Flag("sensitive-log-max-detections-per-container", "Max unique sensitive patterns tracked per container (0 = unlimited)").Default("100").Envar("SENSITIVE_LOG_MAX_DETECTIONS_PER_CONTAINER").Int() + TraceIdHeaders = kingpin.Flag("trace-id-headers", "trace id headers").Default("Traceparent,X-Request-Id").Envar("TRACE_ID_HEADERS").String() HttpPathNormalizationRules = kingpin.Flag("http-path-normalization-rules", "Custom HTTP path normalization rules in format 'pattern1:replacement1,pattern2:replacement2'").Envar("HTTP_PATH_NORMALIZATION_RULES").String() diff --git a/go.mod b/go.mod index 735b76b..20244e4 100644 --- a/go.mod +++ b/go.mod @@ -21,7 +21,7 @@ require ( github.com/hashicorp/golang-lru/v2 v2.0.7 github.com/jpillora/backoff v1.0.0 github.com/mdlayher/taskstats v0.0.0-20230712191918-387b3d561d14 - github.com/nudgebee/logparser v0.0.0-20260310062405-9d6258bdd771 + github.com/nudgebee/logparser v0.0.0-20260313095946-b22168b95d9c github.com/opencontainers/runtime-spec v1.1.0 github.com/prometheus/client_golang v1.20.5 github.com/prometheus/client_model v0.6.1 diff --git a/go.sum b/go.sum index 4379105..160a6c5 100644 --- a/go.sum +++ b/go.sum @@ -323,8 +323,8 @@ github.com/morikuni/aec v1.0.0 h1:nP9CBfwrvYnBRgY6qfDQkygYDmYwOilePFkwzv4dU8A= github.com/morikuni/aec v1.0.0/go.mod h1:BbKIizmSmc5MMPqRYbxO4ZU0S0+P200+tUnFx7PXmsc= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= -github.com/nudgebee/logparser v0.0.0-20260310062405-9d6258bdd771 h1:DfyX+uJ3B1x/Bz0Ie7jtLi9UoIP5PCC5myZCk6JkfmM= -github.com/nudgebee/logparser v0.0.0-20260310062405-9d6258bdd771/go.mod h1:oFnM9D6YEjZzb1jy0kJ/NUkTbkZA+BgNYjpLO4n4szA= +github.com/nudgebee/logparser v0.0.0-20260313095946-b22168b95d9c h1:C0Jx8+61ndlWeKrS25W7hFJW/qpxUpG96tBQdjffd5I= +github.com/nudgebee/logparser v0.0.0-20260313095946-b22168b95d9c/go.mod h1:oFnM9D6YEjZzb1jy0kJ/NUkTbkZA+BgNYjpLO4n4szA= github.com/oklog/ulid v1.3.1 h1:EGfNDEx6MqHz8B3uNV6QAib1UR2Lm97sHi3ocA6ESJ4= github.com/oklog/ulid v1.3.1/go.mod h1:CirwcVhetQ6Lv90oh/F+FBtV6XMibvdAFo93nm5qn4U= github.com/onsi/ginkgo v1.16.5 h1:8xi0RTUf59SOSfEtZMvwTvXYMzG4gV23XVHOZiXNtnE= From 031d66b1e2b10c5421a21be06c09651748817ecc Mon Sep 17 00:00:00 2001 From: Mayank Pande Date: Fri, 13 Mar 2026 22:03:49 +0530 Subject: [PATCH 3/7] fix: cap HTTP/2 DATA payload accumulation to prevent OOM (#226) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Http2Parser.Parse appends DATA frame payloads to RequestPayload and ResponsePayload with no size limit. On nodes with high-throughput HTTP/2 traffic (envoy-gateway, gRPC proxies, benchmark-server), these grow unbounded — confirmed via pprof: 149MB (62% of heap) consumed by Http2Parser on a pod that OOMs after ~22h. Cap both payloads at 128KB per stream. They are only used for LLM provider detection and trace spans which need at most a few KB. END_STREAM tracking is unaffected so stream completion still works. --- ebpftracer/l7/http2.go | 24 ++++++++++++++++++++---- 1 file changed, 20 insertions(+), 4 deletions(-) diff --git a/ebpftracer/l7/http2.go b/ebpftracer/l7/http2.go index 6b2169b..fb8871d 100644 --- a/ebpftracer/l7/http2.go +++ b/ebpftracer/l7/http2.go @@ -37,6 +37,12 @@ const ( // Max accumulated header block size (64KB) to prevent unbounded growth maxPendingHeaderBlockSize = 64 * 1024 + + // Max accumulated DATA payload size per stream (128KB). + // Prevents unbounded memory growth on high-throughput HTTP/2 connections. + // Payloads are only used for LLM provider detection and trace spans, + // which need at most a few KB of data. + maxDataPayloadSize = 128 * 1024 ) type Http2FrameHeader struct { @@ -401,20 +407,30 @@ frameLoop: switch method { case MethodHttp2ClientFrames: - // Client DATA frame = request payload + // Client DATA frame = request payload (capped to prevent unbounded growth) req := p.activeRequests[h.StreamId] - if req != nil { + if req != nil && len(req.RequestPayload) < maxDataPayloadSize { + remaining := maxDataPayloadSize - len(req.RequestPayload) + if len(dataPayload) > remaining { + dataPayload = dataPayload[:remaining] + } req.RequestPayload = append(req.RequestPayload, dataPayload...) } case MethodHttp2ServerFrames: - // Server DATA frame = response payload + // Server DATA frame = response payload (capped to prevent unbounded growth) req := p.activeRequests[h.StreamId] if req != nil { // Track first response time for TTFT if req.firstResponseTime == 0 && len(dataPayload) > 0 { req.firstResponseTime = kernelTime } - req.ResponsePayload = append(req.ResponsePayload, dataPayload...) + if len(req.ResponsePayload) < maxDataPayloadSize { + remaining := maxDataPayloadSize - len(req.ResponsePayload) + if len(dataPayload) > remaining { + dataPayload = dataPayload[:remaining] + } + req.ResponsePayload = append(req.ResponsePayload, dataPayload...) + } // Check for END_STREAM flag on DATA frame if h.Flags&http2FlagEndStream != 0 { req.responseEndStream = true From 655f5e40465a50ae2c5a0eeed20ff8686d13155b Mon Sep 17 00:00:00 2001 From: Mayank Pande Date: Fri, 13 Mar 2026 22:04:04 +0530 Subject: [PATCH 4/7] fix: graceful shutdown, L7 panic protection, and bounded strippedGoExeCache (#225) - Replace os.Exit(0) in signal handler with context cancellation so deferred cleanup (cr.Close, profiling.Stop, resolver.StopWatching) runs on shutdown. Previously leaked eBPF programs, uprobes, and perf buffers on every restart. Use http.Server.Shutdown for graceful HTTP server termination. (fixes #207) - Wrap L7 event processing with recover() to prevent a single malformed packet from crashing the entire agent. Covers both processL7Event and pending L7 retry paths. (fixes #210) - Replace unbounded sync.Map with LRU cache (cap 1000) for strippedGoExeCache to prevent slow memory growth on CI/CD nodes with many unique Go binaries. hashicorp/golang-lru is already a dependency. (fixes #214) --- containers/registry.go | 21 ++++++++++++++++++++- ebpftracer/tls.go | 11 ++++++----- main.go | 26 ++++++++++++++++++++------ 3 files changed, 46 insertions(+), 12 deletions(-) diff --git a/containers/registry.go b/containers/registry.go index 88820dc..a17a039 100644 --- a/containers/registry.go +++ b/containers/registry.go @@ -403,6 +403,11 @@ func (r *Registry) handleEvents(ch <-chan ebpftracer.Event) { const maxIP2FQDNEntries = 10000 func (r *Registry) processL7Event(e ebpftracer.Event) { + defer func() { + if p := recover(); p != nil { + klog.Errorf("recovered from panic in L7 event handler: pid=%d fd=%d protocol=%d: %v", e.Pid, e.Fd, e.L7Request.Protocol, p) + } + }() if c := r.containersByPid[e.Pid]; c != nil { klog.V(5).Infof("L7_EVENT_CONTAINER_FOUND: pid=%d container=%s", e.Pid, c.id) ip2fqdn, result := c.onL7RequestWithResult(e.Pid, e.Fd, e.Timestamp, e.L7Request, e.SocketInfo) @@ -491,7 +496,10 @@ func (r *Registry) processPendingL7Events() { continue } - ip2fqdn, result := c.onL7RequestWithResult(p.event.Pid, p.event.Fd, p.event.Timestamp, p.event.L7Request, p.event.SocketInfo) + ip2fqdn, result, panicked := r.safeOnL7Request(c, p.event) + if panicked { + continue + } if result == L7RequestConnNotFound { // Still not found - keep in queue if not exceeded max retries if p.retryCount < maxRetries { @@ -521,6 +529,17 @@ func (r *Registry) processPendingL7Events() { } } +func (r *Registry) safeOnL7Request(c *Container, e ebpftracer.Event) (ip2fqdn map[netaddr.IP]*common.Domain, result L7RequestResult, panicked bool) { + defer func() { + if p := recover(); p != nil { + klog.Errorf("recovered from panic in pending L7 retry: pid=%d fd=%d protocol=%d: %v", e.Pid, e.Fd, e.L7Request.Protocol, p) + panicked = true + } + }() + ip2fqdn, result = c.onL7RequestWithResult(e.Pid, e.Fd, e.Timestamp, e.L7Request, e.SocketInfo) + return +} + func (r *Registry) getOrCreateContainer(pid uint32) *Container { // Fast path: try to find existing container with read lock lockStart := time.Now() diff --git a/ebpftracer/tls.go b/ebpftracer/tls.go index b5eb709..986fe90 100644 --- a/ebpftracer/tls.go +++ b/ebpftracer/tls.go @@ -10,12 +10,12 @@ import ( "os" "regexp" "strings" - "sync" "unsafe" "github.com/cilium/ebpf/link" "github.com/coroot/coroot-node-agent/common" "github.com/coroot/coroot-node-agent/proc" + lru "github.com/hashicorp/golang-lru/v2" "k8s.io/klog/v2" ) @@ -53,8 +53,9 @@ var ( // strippedGoExeCache caches exe paths that are Go binaries but have no TLS symbols. // This avoids expensive ELF scanning for the same stripped binary across many - // short-lived processes (e.g., kubectl invocations). Uses sync.Map for lock-free reads. - strippedGoExeCache = &sync.Map{} // map[string]struct{} + // short-lived processes (e.g., kubectl invocations). Capped at 1000 entries to + // prevent unbounded growth on CI/CD nodes with many unique Go binaries. + strippedGoExeCache, _ = lru.New[string, struct{}](1000) ) func (t *Tracer) AttachOpenSslUprobes(pid uint32) []link.Link { @@ -181,7 +182,7 @@ func (t *Tracer) AttachGoTlsUprobes(pid uint32) ([]link.Link, bool) { // Skip binaries we already know are stripped (no TLS symbols). // This avoids expensive ELF scanning for repeated short-lived processes like kubectl. - if _, stripped := strippedGoExeCache.Load(exeName); stripped { + if strippedGoExeCache.Contains(exeName) { klog.V(3).Infof("GO_TLS_SKIP_STRIPPED: pid=%d exe=%s", pid, exeName) return nil, true // still a Go app, just stripped } @@ -258,7 +259,7 @@ func (t *Tracer) AttachGoTlsUprobes(pid uint32) ([]link.Link, bool) { log("failed to get write symbol", err) // Cache this exe as stripped to skip future attempts if exeName != "" { - strippedGoExeCache.Store(exeName, struct{}{}) + strippedGoExeCache.Add(exeName, struct{}{}) } return nil, isGolangApp } diff --git a/main.go b/main.go index b334db7..b200bfc 100644 --- a/main.go +++ b/main.go @@ -2,6 +2,7 @@ package main import ( "bytes" + "context" "flag" "fmt" "log" @@ -163,16 +164,16 @@ func main() { if err != nil { klog.Errorf("Error starting resolver: %v", err) } + defer resolver.StopWatching() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() go func() { sigChannel := make(chan os.Signal, 1) - defer close(sigChannel) - signal.Notify(sigChannel, os.Interrupt, syscall.SIGTERM) <-sigChannel - klog.Infoln("Received signal, shutting down") - resolver.StopWatching() - os.Exit(0) // Ensure graceful termination + cancel() }() hostname, kv, err := uname() if err != nil { @@ -252,8 +253,21 @@ func main() { "Metrics collection timeout", )) + srv := &http.Server{Addr: *flags.ListenAddress} + go func() { + <-ctx.Done() + shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 5*time.Second) + defer shutdownCancel() + if err := srv.Shutdown(shutdownCtx); err != nil { + klog.Errorf("HTTP server shutdown error: %v", err) + } + }() + klog.Infoln("listening on:", *flags.ListenAddress) - klog.Errorln(http.ListenAndServe(*flags.ListenAddress, nil)) + if err := srv.ListenAndServe(); err != http.ErrServerClosed { + klog.Errorln(err) + } + klog.Infoln("shutdown complete") } func info(name, version string) prometheus.Collector { From 3042078f3cbfff090534273b2c69874ac964a966 Mon Sep 17 00:00:00 2001 From: Mayank Pande Date: Sat, 14 Mar 2026 17:04:00 +0530 Subject: [PATCH 5/7] fix: lightweight HTTP/2 parsing to reduce CPU on busy nodes (#227) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix: lightweight HTTP/2 parsing to reduce CPU on busy nodes HTTP/2 frame parsing was consuming 70% of CPU (profiled). Most HTTP/2 traffic (gRPC, K8s API, service mesh) doesn't need payload capture — only status codes for L7 metrics. - Add lightweight mode to Http2Parser: skips DATA payload accumulation and RequestHeaders map allocation for non-LLM traffic - Auto-upgrade to full mode when :authority matches an LLM provider - Reuse statuses/grpcStatuses maps across Parse() calls (clear instead of alloc) — eliminates ~15% CPU from per-call map allocation - Cap activeRequests at 100 per connection to prevent orphan stream growth - Cap HTTP/2 parsers at 50 per container - Fix parser GC leak: connectionless parsers now cleaned up when pid dies - Eliminate duplicate GetActiveStreamsForLLM() call per event - Skip LLM stream tracker entirely when parser is lightweight * fix: address PR review — GC leak for closed connections, deduplicate END_STREAM - Fix parser GC: closed connection parsers in long-running processes were not cleaned up. Now uses heuristic: if pid is alive but no connection exists, clean up parsers with zero active requests and no partial data. - Add HasPartialData() to Http2Parser for safe GC decisions. - Deduplicate END_STREAM handling: extract it before lightweight/full branch instead of duplicating in both paths. * fix: skip-set for non-LLM HTTP/2 traffic to eliminate CPU waste processHTTP2WithoutConnection produces zero L7 metrics (no connection = no DestinationKey). Its only purpose is LLM detection, yet it was parsing every HTTP/2 frame from gRPC/K8s API traffic — consuming 70% of CPU on busy nodes. Two-phase skip-set approach: 1. If dstIP is known, check DNS cache — skip immediately if not LLM. 2. Otherwise, parse first event per pid:fd in lightweight mode to extract :authority. If not LLM, add to skip-set and never parse that fd again. Also adds GC for skip-set entries when the pid dies. --- containers/container.go | 136 +++++++++++++++++++++++++++------------- containers/llm.go | 6 ++ ebpftracer/l7/http2.go | 110 ++++++++++++++++++++++---------- 3 files changed, 178 insertions(+), 74 deletions(-) diff --git a/containers/container.go b/containers/container.go index bb74553..7f6b361 100644 --- a/containers/container.go +++ b/containers/container.go @@ -38,6 +38,12 @@ var ( gpuStatsWindow = 15 * time.Second ) +const ( + // Max per-connection HTTP/2 parsers per container. + // Each parser holds HPACK decoders and active request state. + maxHTTP2ParsersPerContainer = 50 +) + type ContainerID string type ContainerNetwork struct { @@ -128,6 +134,7 @@ type Container struct { 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) + http2SkipSet map[PidFd]struct{} // pid:fd pairs confirmed as non-LLM — skip all future HTTP/2 events l7Stats L7Stats tcpMetrics *TCPMetrics @@ -1158,9 +1165,14 @@ func (c *Container) onL7RequestWithResult(pid uint32, fd uint64, timestamp uint6 } pidFd := PidFd{Pid: pid, Fd: fd} if c.googleHTTP2Parsers[pidFd] == nil { - c.googleHTTP2Parsers[pidFd] = l7.NewHttp2Parser() + if len(c.googleHTTP2Parsers) >= maxHTTP2ParsersPerContainer { + return nil, L7RequestProcessed + } + p := l7.NewHttp2Parser() + p.Lightweight = true + p.LLMHostChecker = isLLMRelevantHost + c.googleHTTP2Parsers[pidFd] = p } - // Use per-connection parser (each HTTP/2 connection has its own HPACK dynamic table) parser := c.googleHTTP2Parsers[pidFd] conn.http2Parser = parser // Keep reference on connection for compatibility requests := parser.Parse(r.Method, r.Payload, uint64(r.Duration)) @@ -1188,7 +1200,7 @@ func (c *Container) onL7RequestWithResult(pid uint32, fd uint64, timestamp uint6 // Feed active streams to LLM stream tracker for SSE-based completion detection // This handles streaming LLM responses that don't send END_STREAM until complete - if c.llmStreamTracker != nil { + if c.llmStreamTracker != nil && !parser.Lightweight { // Resolve host for LLM detection host := conn.DestinationKey.GetDestinationWorkload().Name if isIPAddress(host) { @@ -1196,18 +1208,16 @@ func (c *Container) onL7RequestWithResult(pid uint32, fd uint64, timestamp uint6 host = domain.FQDN } } + activeStreams := parser.GetActiveStreamsForLLM() // Also try :authority header if host is still an IP if host == "" || isIPAddress(host) { - for _, update := range parser.GetActiveStreamsForLLM() { + for _, update := range activeStreams { if update.Authority != "" && !isIPAddress(update.Authority) { host = update.Authority break } } } - - // Process active streams for LLM tracking - activeStreams := parser.GetActiveStreamsForLLM() for _, update := range activeStreams { streamHost := host if streamHost == "" && update.Authority != "" { @@ -1345,49 +1355,70 @@ func (c *Container) onL7RequestWithResult(pid uint32, fd uint64, timestamp uint6 // This is common for Go TLS connections where goroutines switch threads between // TCP connect and TLS write, causing fd_by_pid_tgid lookup to fail in eBPF. // -// We can still extract useful info from HTTP/2 frames: -// - :authority pseudo-header for host resolution (LLM detection) -// - :path for request path matching -// - Request/response payloads for LLM token extraction -// -// Each HTTP/2 connection has its own HPACK dynamic table, so we use per-fd parsers -// to avoid header decoding corruption when a process has multiple connections. -// -// Note: Metrics that require destination IP will be skipped since we don't have it. +// Since this path produces NO L7 metrics (no connection = no DestinationKey), the only +// purpose is LLM detection. To avoid wasting CPU on non-LLM traffic (gRPC, K8s API, etc.), +// we use a two-phase approach: +// 1. If dstIP is known, check DNS cache first — skip if not LLM-relevant. +// 2. Otherwise, parse only the first event per pid:fd to extract :authority. +// If not LLM-relevant, add to skip-set and never parse that fd again. func (c *Container) processHTTP2WithoutConnection(pid uint32, fd uint64, r *l7.RequestData, dstIP ...netaddr.IP) (map[netaddr.IP]*common.Domain, L7RequestResult) { - // Use per-connection parser: each HTTP/2 connection has its own HPACK dynamic table, - // so sharing a parser across connections corrupts header decoding. + pidFd := PidFd{Pid: pid, Fd: fd} + + // Fast path: already determined this fd is not LLM-relevant + if c.http2SkipSet != nil { + if _, skip := c.http2SkipSet[pidFd]; skip { + return nil, L7RequestProcessed + } + } + + // If we have a destination IP, check DNS cache before parsing + if len(dstIP) > 0 && !dstIP[0].IsZero() { + host := "" + if domain := c.registry.getDomain(dstIP[0]); domain != nil { + host = domain.FQDN + } + if host == "" || !isLLMRelevantHost(host) { + // Not an LLM destination — skip this and all future events + c.addToHTTP2SkipSet(pidFd) + return nil, L7RequestProcessed + } + } + + // Parse the HTTP/2 frames to extract :authority for LLM detection if c.googleHTTP2Parsers == nil { c.googleHTTP2Parsers = make(map[PidFd]*l7.Http2Parser) } - - pidFd := PidFd{Pid: pid, Fd: fd} parser := c.googleHTTP2Parsers[pidFd] if parser == nil { + if len(c.googleHTTP2Parsers) >= maxHTTP2ParsersPerContainer { + return nil, L7RequestProcessed + } parser = l7.NewHttp2Parser() + parser.Lightweight = true + parser.LLMHostChecker = isLLMRelevantHost c.googleHTTP2Parsers[pidFd] = parser } requests := parser.Parse(r.Method, r.Payload, uint64(r.Duration)) - if len(requests) == 0 { - klog.V(3).Infof("HTTP2_CONNECTIONLESS_PARSE: pid=%d fd=%d method=%d payloadLen=%d completed=0 active=%d", - pid, fd, r.Method, len(r.Payload), parser.ActiveRequestCount()) + // If parser is still in lightweight mode after parsing, no LLM host was found. + // Once we've seen at least one completed request (or enough HEADERS to decide), + // add to skip-set. We check ActiveRequestCount==0 to ensure we've seen a full + // request-response cycle (not just the first DATA frame before any HEADERS). + if parser.Lightweight && len(requests) > 0 { + // Parsed completed requests but none were LLM — this fd is non-LLM + c.addToHTTP2SkipSet(pidFd) + delete(c.googleHTTP2Parsers, pidFd) + return nil, L7RequestProcessed } for _, req := range requests { - // Extract host from :authority header (set by HTTP/2 parser) host := req.Authority if host == "" && len(dstIP) > 0 && !dstIP[0].IsZero() { - // Fallback: resolve destination IP via DNS cache. - // This handles mid-stream HPACK join where :authority is in the - // dynamic table and can't be decoded. if domain := c.registry.getDomain(dstIP[0]); domain != nil { host = domain.FQDN - klog.V(4).Infof("HTTP2_CONNECTIONLESS: resolved host from DNS cache: ip=%s host=%s path=%s", dstIP[0], host, req.Path) } } - // Detect LLM provider: try host first, then path, then response structure provider := ProviderUnknown if host != "" { provider = DetectLLMProvider(host) @@ -1395,8 +1426,6 @@ func (c *Container) processHTTP2WithoutConnection(pid uint32, fd uint64, r *l7.R if provider == ProviderUnknown && req.Path != "" { provider, _ = DetectLLMProviderFromPath(req.Path) } - // Last resort: analyze request/response structure for LLM patterns. - // Skip for gRPC (protobuf data can contain strings that match JSON patterns). isGRPC := strings.HasPrefix(req.ContentType, "application/grpc") if provider == ProviderUnknown && !isGRPC && len(req.ResponsePayload) > 0 { provider = detectProviderFromResponseStructure(req.ResponsePayload) @@ -1405,26 +1434,18 @@ func (c *Container) processHTTP2WithoutConnection(pid uint32, fd uint64, r *l7.R provider = detectProviderFromRequestStructure(req.RequestPayload) } - if host == "" && provider == ProviderUnknown { - klog.V(3).Infof("HTTP2_CONNECTIONLESS: no authority/path and no LLM detected for pid=%d fd=%d", pid, fd) - continue - } - if provider != ProviderUnknown { - // Set a default server address if host is unknown (detected from payload structure) if host == "" { host = providerDefaultHost(provider) } - - klog.V(4).Infof("HTTP2_CONNECTIONLESS_LLM_DETECTED: pid=%d fd=%d provider=%s host=%s path=%s reqLen=%d respLen=%d", - pid, fd, provider, host, req.Path, len(req.RequestPayload), len(req.ResponsePayload)) + klog.V(4).Infof("HTTP2_CONNECTIONLESS_LLM_DETECTED: pid=%d fd=%d provider=%s host=%s path=%s", + pid, fd, provider, host, req.Path) requestPayloadBase64 := base64.StdEncoding.EncodeToString(req.RequestPayload) responsePayloadBase64 := "" if len(req.ResponsePayload) > 0 { responsePayloadBase64 = base64.StdEncoding.EncodeToString(req.ResponsePayload) } - c.trackLLMRequest(provider, host, req.Path, requestPayloadBase64, responsePayloadBase64, req.Duration) } } @@ -1432,6 +1453,13 @@ func (c *Container) processHTTP2WithoutConnection(pid uint32, fd uint64, r *l7.R return nil, L7RequestProcessed } +func (c *Container) addToHTTP2SkipSet(pidFd PidFd) { + if c.http2SkipSet == nil { + c.http2SkipSet = make(map[PidFd]struct{}) + } + c.http2SkipSet[pidFd] = struct{}{} +} + // refreshActiveConnections snapshots active connections under c.lock and // pushes the gauge values to TCPMetrics. Called periodically from the // event handler goroutine (not from Collect). @@ -1857,14 +1885,38 @@ func (c *Container) gc(now time.Time) { // Clean up HTTP/2 parsers for closed/dead connections. // Parsers hold HPACK decoders, partial frame buffers, and active request maps // that accumulate memory over time if not cleaned up. + // Two cases: (1) connection-tracked parsers — delete when connection is gone, + // (2) connectionless parsers (processHTTP2WithoutConnection) — delete when pid dies. if c.googleHTTP2Parsers != nil { for pidFd := range c.googleHTTP2Parsers { - if _, alive := c.connectionsByPidFd[pidFd]; !alive { + if _, hasConn := c.connectionsByPidFd[pidFd]; hasConn { + continue // connection still alive + } + if _, hasProc := c.processes[pidFd.Pid]; !hasProc { + delete(c.googleHTTP2Parsers, pidFd) // pid dead — connectionless parser cleanup + continue + } + // Pid is alive but no connection — could be a connectionless parser + // (still needed) or a closed connection's parser (stale). + // Check if this parser was created via the connectionless path by + // seeing if there was ever a tracked connection for this pidFd. + // If activeConnections had it at some point, it's stale. + // Use a simple heuristic: if the parser has zero active requests + // and no partial frame data, it's safe to clean up. + p := c.googleHTTP2Parsers[pidFd] + if p.ActiveRequestCount() == 0 && !p.HasPartialData() { delete(c.googleHTTP2Parsers, pidFd) } } } + // Clean up skip-set entries for dead pids + for pidFd := range c.http2SkipSet { + if _, hasProc := c.processes[pidFd.Pid]; !hasProc { + delete(c.http2SkipSet, pidFd) + } + } + for dst, at := range c.lastConnectionAttempts { _, active := establishedDst[dst] if !active && !at.IsZero() && now.Sub(at) > gcInterval { diff --git a/containers/llm.go b/containers/llm.go index 1e04a0f..47a9e72 100644 --- a/containers/llm.go +++ b/containers/llm.go @@ -96,6 +96,12 @@ func providerDefaultHost(provider LLMProvider) string { } } +// isLLMRelevantHost returns true if the host matches a known LLM provider. +// Used by Http2Parser to auto-upgrade from lightweight to full mode. +func isLLMRelevantHost(host string) bool { + return DetectLLMProvider(host) != ProviderUnknown +} + // DetectLLMProvider identifies if a hostname belongs to an LLM provider func DetectLLMProvider(hostname string) LLMProvider { hostname = strings.ToLower(hostname) diff --git a/ebpftracer/l7/http2.go b/ebpftracer/l7/http2.go index fb8871d..7c484d5 100644 --- a/ebpftracer/l7/http2.go +++ b/ebpftracer/l7/http2.go @@ -43,6 +43,10 @@ const ( // Payloads are only used for LLM provider detection and trace spans, // which need at most a few KB of data. maxDataPayloadSize = 128 * 1024 + + // Max concurrent HTTP/2 streams tracked per connection. + // Prevents unbounded memory growth when responses never complete (orphan streams). + maxActiveRequests = 100 ) type Http2FrameHeader struct { @@ -96,6 +100,10 @@ type Http2Parser struct { activeRequests map[uint32]*Http2Request lastGcTime uint64 + // Reusable maps cleared on each Parse() call to avoid per-call allocation + statuses map[uint32]Status + grpcStatuses map[uint32]Status + // Buffers for partial frame reassembly across Read() calls // Needed because S2A/TLS returns small chunks that may split HTTP/2 frames clientPartialFrame []byte @@ -110,13 +118,26 @@ type Http2Parser struct { // Static table headers still work; dynamic table rebuilds over time. clientDecoderDegraded bool serverDecoderDegraded bool + + // Lightweight mode skips DATA payload accumulation and RequestHeaders map. + // Used for non-LLM HTTP/2 traffic (gRPC, K8s API, etc.) where only status + // codes are needed for L7 metrics. Auto-upgrades to full mode when an + // LLM-relevant :authority header is detected. + Lightweight bool + + // LLMHostChecker is called with :authority values to determine if this + // connection serves LLM traffic. When it returns true, Lightweight is + // set to false and full payload capture begins. + LLMHostChecker func(host string) bool } func NewHttp2Parser() *Http2Parser { return &Http2Parser{ clientDecoder: hpack.NewDecoder(4096, nil), serverDecoder: hpack.NewDecoder(4096, nil), - activeRequests: map[uint32]*Http2Request{}, + activeRequests: make(map[uint32]*Http2Request), + statuses: make(map[uint32]Status), + grpcStatuses: make(map[uint32]Status), } } @@ -141,6 +162,13 @@ func (p *Http2Parser) ActiveRequestCount() int { return len(p.activeRequests) } +// HasPartialData returns true if the parser has buffered partial frame data +// that would be lost if the parser were garbage collected. +func (p *Http2Parser) HasPartialData() bool { + return len(p.clientPartialFrame) > 0 || len(p.serverPartialFrame) > 0 || + p.clientPendingHeaders != nil || p.serverPendingHeaders != nil +} + // Http2StreamUpdate contains information about an active HTTP/2 stream // Used for notifying LLM stream tracker about streaming responses type Http2StreamUpdate struct { @@ -234,14 +262,20 @@ func (p *Http2Parser) decodeHeaderBlock( case MethodHttp2ClientFrames: req := p.activeRequests[streamId] if req == nil { + if len(p.activeRequests) >= maxActiveRequests { + // Too many active streams; skip to prevent unbounded growth + break + } req = &Http2Request{ - kernelTime: kernelTime, - RequestHeaders: make(map[string]string), + kernelTime: kernelTime, + } + if !p.Lightweight { + req.RequestHeaders = make(map[string]string) } p.activeRequests[streamId] = req } decoder.SetEmitFunc(func(hf hpack.HeaderField) { - // Store all headers for trace correlation + // Store all headers for trace correlation (full mode only) if req.RequestHeaders != nil { req.RequestHeaders[hf.Name] = hf.Value } @@ -262,6 +296,13 @@ func (p *Http2Parser) decodeHeaderBlock( case ":authority": if req.Authority == "" && hf.Value != "" { req.Authority = hf.Value + // Auto-upgrade: if authority matches an LLM host, switch to full mode + if p.Lightweight && p.LLMHostChecker != nil && p.LLMHostChecker(hf.Value) { + p.Lightweight = false + if req.RequestHeaders == nil { + req.RequestHeaders = make(map[string]string) + } + } } case "content-type": if req.ContentType == "" && hf.Value != "" { @@ -336,8 +377,10 @@ func (p *Http2Parser) Parse(method Method, payload []byte, kernelTime uint64) [] } var decoder *hpack.Decoder - statuses := map[uint32]Status{} - grpcStatuses := map[uint32]Status{} + clear(p.statuses) + clear(p.grpcStatuses) + statuses := p.statuses + grpcStatuses := p.grpcStatuses // Prepend any saved partial frame data from previous Parse() call // This handles HTTP/2 frames split across multiple S2A/TLS Read() calls @@ -403,37 +446,40 @@ frameLoop: offset = frameStart break frameLoop } - dataPayload := payload[offset : offset+h.Length] - - switch method { - case MethodHttp2ClientFrames: - // Client DATA frame = request payload (capped to prevent unbounded growth) - req := p.activeRequests[h.StreamId] - if req != nil && len(req.RequestPayload) < maxDataPayloadSize { - remaining := maxDataPayloadSize - len(req.RequestPayload) - if len(dataPayload) > remaining { - dataPayload = dataPayload[:remaining] - } - req.RequestPayload = append(req.RequestPayload, dataPayload...) + + // Track END_STREAM on server DATA frames unconditionally (needed for both modes) + if method == MethodHttp2ServerFrames && h.Flags&http2FlagEndStream != 0 { + if req := p.activeRequests[h.StreamId]; req != nil { + req.responseEndStream = true } - case MethodHttp2ServerFrames: - // Server DATA frame = response payload (capped to prevent unbounded growth) - req := p.activeRequests[h.StreamId] - if req != nil { - // Track first response time for TTFT - if req.firstResponseTime == 0 && len(dataPayload) > 0 { - req.firstResponseTime = kernelTime - } - if len(req.ResponsePayload) < maxDataPayloadSize { - remaining := maxDataPayloadSize - len(req.ResponsePayload) + } + + // Payload accumulation only in full mode + if !p.Lightweight { + dataPayload := payload[offset : offset+h.Length] + switch method { + case MethodHttp2ClientFrames: + req := p.activeRequests[h.StreamId] + if req != nil && len(req.RequestPayload) < maxDataPayloadSize { + remaining := maxDataPayloadSize - len(req.RequestPayload) if len(dataPayload) > remaining { dataPayload = dataPayload[:remaining] } - req.ResponsePayload = append(req.ResponsePayload, dataPayload...) + req.RequestPayload = append(req.RequestPayload, dataPayload...) } - // Check for END_STREAM flag on DATA frame - if h.Flags&http2FlagEndStream != 0 { - req.responseEndStream = true + case MethodHttp2ServerFrames: + req := p.activeRequests[h.StreamId] + if req != nil { + if req.firstResponseTime == 0 && len(dataPayload) > 0 { + req.firstResponseTime = kernelTime + } + if len(req.ResponsePayload) < maxDataPayloadSize { + remaining := maxDataPayloadSize - len(req.ResponsePayload) + if len(dataPayload) > remaining { + dataPayload = dataPayload[:remaining] + } + req.ResponsePayload = append(req.ResponsePayload, dataPayload...) + } } } } From 0ce81e819d164284d8500b221f359d6e2c6db1bb Mon Sep 17 00:00:00 2001 From: Mayank Pande Date: Mon, 16 Mar 2026 16:25:17 +0530 Subject: [PATCH 6/7] fix: prevent runtime crash from uint64-to-int overflow in event readers (#228) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix: prevent runtime crash from uint64-to-int overflow in event readers Both perf and ring buffer L7 event readers convert eBPF-reported payload sizes from uint64 to int without safe clamping. If eBPF sends corrupted data with high-bit-set sizes (e.g. 0x8000000000000001), int() produces a negative value that bypasses the MaxPayloadSize cap, causing make([]byte, negative) to crash the reader goroutine. Also upgrades cilium/ebpf v0.20.0 → v0.21.0 which includes ring buffer and memory management fixes relevant to Go 1.24. Fixes runtime crashes: "fatal error: stack not a power of 2" observed on 12/19 pods after ~7 hours of operation. * fix: revert cilium/ebpf upgrade, keep clampSize fix only cilium/ebpf v0.21.0 has breaking API changes (removed Sym, RewriteConstants, MapName) that affect transitive dependencies. Revert to v0.20.0 and keep just the uint64-to-int overflow fix. --- ebpftracer/tracer.go | 30 +++++++++++++----------------- go.mod | 10 +++++----- go.sum | 24 ++++++++++++------------ 3 files changed, 30 insertions(+), 34 deletions(-) diff --git a/ebpftracer/tracer.go b/ebpftracer/tracer.go index 0f00b30..26ec9de 100644 --- a/ebpftracer/tracer.go +++ b/ebpftracer/tracer.go @@ -448,6 +448,15 @@ func safeDuration(ns uint64) time.Duration { return time.Duration(ns) } +// clampSize safely converts a uint64 size to int, capping at maxSize. +// Prevents negative int from uint64 values with the high bit set. +func clampSize(size uint64, maxSize int) int { + if size > uint64(maxSize) { + return maxSize + } + return int(size) +} + func (t *lostSamplesTracker) recordLostSamples(name string, count uint64, cpu int) { t.count.Add(count) now := time.Now().Unix() @@ -492,14 +501,8 @@ func runEventsReader(name string, r *perf.Reader, ch chan<- Event, typ perfMapTy } payloadSize := binary.LittleEndian.Uint64(data[40:48]) responseSize := binary.LittleEndian.Uint64(data[48:56]) - payloadLen := int(payloadSize) - if payloadLen > MaxPayloadSize { - payloadLen = MaxPayloadSize - } - responseLen := int(responseSize) - if responseLen > MaxPayloadSize { - responseLen = MaxPayloadSize - } + payloadLen := clampSize(payloadSize, MaxPayloadSize) + responseLen := clampSize(responseSize, MaxPayloadSize) payloadData := make([]byte, payloadLen) if l7EventHeaderSize+payloadLen <= len(data) { @@ -607,15 +610,8 @@ func runRingbufEventsReader(name string, r *ringbuf.Reader, ch chan<- Event) { payloadSize := binary.LittleEndian.Uint64(data[40:48]) responseSize := binary.LittleEndian.Uint64(data[48:56]) - - payloadLen := int(payloadSize) - if payloadLen > MaxPayloadSize { - payloadLen = MaxPayloadSize - } - responseLen := int(responseSize) - if responseLen > MaxPayloadSize { - responseLen = MaxPayloadSize - } + payloadLen := clampSize(payloadSize, MaxPayloadSize) + responseLen := clampSize(responseSize, MaxPayloadSize) payloadData := make([]byte, payloadLen) if l7EventHeaderSize+payloadLen <= len(data) { diff --git a/go.mod b/go.mod index 20244e4..b122fc9 100644 --- a/go.mod +++ b/go.mod @@ -127,14 +127,14 @@ require ( github.com/josharian/intern v1.0.0 // indirect github.com/josharian/native v1.1.0 // indirect github.com/json-iterator/go v1.1.12 // indirect - github.com/klauspost/compress v1.17.11 // indirect + github.com/klauspost/compress v1.18.0 // indirect github.com/liggitt/tabwriter v0.0.0-20181228230101-89fcab3d43de // indirect github.com/mackerelio/go-osstat v0.2.5 // indirect github.com/magiconair/properties v1.8.7 // indirect github.com/mailru/easyjson v0.7.7 // indirect github.com/mdlayher/genetlink v1.3.2 // indirect github.com/mdlayher/netlink v1.7.2 // indirect - github.com/mdlayher/socket v0.4.1 // indirect + github.com/mdlayher/socket v0.5.1 // indirect github.com/mitchellh/go-wordwrap v1.0.1 // indirect github.com/mitchellh/mapstructure v1.5.0 // indirect github.com/moby/docker-image-spec v1.3.1 // indirect @@ -149,7 +149,7 @@ require ( github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/oklog/ulid v1.3.1 // indirect github.com/opencontainers/go-digest v1.0.0 // indirect - github.com/opencontainers/image-spec v1.1.0 // indirect + github.com/opencontainers/image-spec v1.1.1 // indirect github.com/opencontainers/selinux v1.13.0 // indirect github.com/opentracing/opentracing-go v1.2.1-0.20220228012449-10b1cf09e00b // indirect github.com/pelletier/go-toml/v2 v2.2.2 // indirect @@ -167,8 +167,8 @@ require ( github.com/sourcegraph/conc v0.3.0 // indirect github.com/spf13/afero v1.11.0 // indirect github.com/spf13/cast v1.7.0 // indirect - github.com/spf13/cobra v1.8.1 // indirect - github.com/spf13/pflag v1.0.6-0.20210604193023-d5e0c0615ace // indirect + github.com/spf13/cobra v1.9.1 // indirect + github.com/spf13/pflag v1.0.7 // indirect github.com/spf13/viper v1.19.0 // indirect github.com/subosito/gotenv v1.6.0 // indirect github.com/ulikunitz/xz v0.5.14 // indirect diff --git a/go.sum b/go.sum index 160a6c5..991435d 100644 --- a/go.sum +++ b/go.sum @@ -88,7 +88,7 @@ github.com/coroot/dotnetdiag v1.2.2 h1:PVP/By8o+xhPjfVolJYcjHLbFQInM7pkaD6/otPLc github.com/coroot/dotnetdiag v1.2.2/go.mod h1:veXCMlFzm1yNl7wwJb/ZLxO4WbzhDBoy1VG1XtkH2ls= github.com/coroot/pyroscope/ebpf v0.0.0-20250418092207-a70610b6df72 h1:NFEMXMOdUzQFG+OwRVecv/yc75/s8MmLkNquMm6hk3I= github.com/coroot/pyroscope/ebpf v0.0.0-20250418092207-a70610b6df72/go.mod h1:IepHM9FJ0n3n3k+ZV23Y7vNAfvWI7LDuLqWPO4rB6sQ= -github.com/cpuguy83/go-md2man/v2 v2.0.4/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o= +github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g= github.com/cyphar/filepath-securejoin v0.6.0 h1:BtGB77njd6SVO6VztOHfPxKitJvd/VPT+OFBFMOi1Is= github.com/cyphar/filepath-securejoin v0.6.0/go.mod h1:A8hd4EnAeyujCJRrICiOWqjS1AX0a9kM5XL+NwKoYSc= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= @@ -254,8 +254,8 @@ github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnr github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8= github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= -github.com/klauspost/compress v1.17.11 h1:In6xLpyWOi1+C7tXUUWv2ot1QvBjxevKAaI6IXrJmUc= -github.com/klauspost/compress v1.17.11/go.mod h1:pMDklpSncoRMuLFrf1W9Ss9KT+0rH90U12bZKk7uwG0= +github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo= +github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ= github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= @@ -290,8 +290,8 @@ github.com/mdlayher/netlink v1.4.1/go.mod h1:e4/KuJ+s8UhfUpO9z00/fDZZmhSrs+oxyqA github.com/mdlayher/netlink v1.7.2 h1:/UtM3ofJap7Vl4QWCPDGXY8d3GIY2UGSDbK+QWmY8/g= github.com/mdlayher/netlink v1.7.2/go.mod h1:xraEF7uJbxLhc5fpHL4cPe221LI2bdttWlU+ZGLfQSw= github.com/mdlayher/socket v0.0.0-20210307095302-262dc9984e00/go.mod h1:GAFlyu4/XV68LkQKYzKhIo/WW7j3Zi0YRAz/BOoanUc= -github.com/mdlayher/socket v0.4.1 h1:eM9y2/jlbs1M615oshPQOHZzj6R6wMT7bX5NPiQvn2U= -github.com/mdlayher/socket v0.4.1/go.mod h1:cAqeGjoufqdxWkD7DkpyS+wcefOtmu5OQ8KuoJGIReA= +github.com/mdlayher/socket v0.5.1 h1:VZaqt6RkGkt2OE9l3GcC6nZkqD3xKeQLyfleW/uBcos= +github.com/mdlayher/socket v0.5.1/go.mod h1:TjPLHI1UgwEv5J1B5q0zTZq12A/6H7nKmtTanQE37IQ= github.com/mdlayher/taskstats v0.0.0-20230712191918-387b3d561d14 h1:eKehnW2s+3DQYZLAa/Pm04sk1G+k8LlZt0OUDbyYmrI= github.com/mdlayher/taskstats v0.0.0-20230712191918-387b3d561d14/go.mod h1:hDhp1SgOluLtKhnB65Wb/j3f7ghQWdOl+XIrbH9yqWc= github.com/mitchellh/go-wordwrap v1.0.1 h1:TLuKupo69TCn6TQSyGxwI1EblZZEsQ0vMlAFQflz0v0= @@ -334,8 +334,8 @@ github.com/onsi/gomega v1.35.1 h1:Cwbd75ZBPxFSuZ6T+rN/WCb/gOc6YgFBXLlZLhC7Ds4= github.com/onsi/gomega v1.35.1/go.mod h1:PvZbdDc8J6XJEpDK4HCuRBm8a6Fzp9/DmhC9C7yFlog= github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U= github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM= -github.com/opencontainers/image-spec v1.1.0 h1:8SG7/vwALn54lVB/0yZ/MMwhFrPYtpEHQb2IpWsCzug= -github.com/opencontainers/image-spec v1.1.0/go.mod h1:W4s4sFTMaBeK1BQLXbG4AdM2szdn85PY75RI83NrTrM= +github.com/opencontainers/image-spec v1.1.1 h1:y0fUlFfIZhPF1W537XOLg0/fcx6zcHCJwooC2xJA040= +github.com/opencontainers/image-spec v1.1.1/go.mod h1:qpqAh3Dmcf36wStyyWU+kCeDgrGnAve2nCC8+7h8Q0M= github.com/opencontainers/runtime-spec v1.1.0 h1:HHUyrt9mwHUjtasSbXSMvs4cyFxh+Bll4AjJ9odEGpg= github.com/opencontainers/runtime-spec v1.1.0/go.mod h1:jwyrGlmzljRJv/Fgzds9SsS/C5hL+LL3ko9hs6T5lQ0= github.com/opencontainers/selinux v1.13.0 h1:Zza88GWezyT7RLql12URvoxsbLfjFx988+LGaWfbL84= @@ -385,11 +385,11 @@ github.com/spf13/afero v1.11.0 h1:WJQKhtpdm3v2IzqG8VMqrr6Rf3UYpEF239Jy9wNepM8= github.com/spf13/afero v1.11.0/go.mod h1:GH9Y3pIexgf1MTIWtNGyogA5MwRIDXGUr+hbWNoBjkY= github.com/spf13/cast v1.7.0 h1:ntdiHjuueXFgm5nzDRdOS4yfT43P5Fnud6DH50rz/7w= github.com/spf13/cast v1.7.0/go.mod h1:ancEpBxwJDODSW/UG4rDrAqiKolqNNh2DX3mk86cAdo= -github.com/spf13/cobra v1.8.1 h1:e5/vxKd/rZsfSJMUX1agtjeTDf+qv1/JdBF8gg5k9ZM= -github.com/spf13/cobra v1.8.1/go.mod h1:wHxEcudfqmLYa8iTfL+OuZPbBZkmvliBWKIezN3kD9Y= -github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= -github.com/spf13/pflag v1.0.6-0.20210604193023-d5e0c0615ace h1:9PNP1jnUjRhfmGMlkXHjYPishpcw4jpSt/V/xYY3FMA= -github.com/spf13/pflag v1.0.6-0.20210604193023-d5e0c0615ace/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= +github.com/spf13/cobra v1.9.1 h1:CXSaggrXdbHK9CF+8ywj8Amf7PBRmPCOJugH954Nnlo= +github.com/spf13/cobra v1.9.1/go.mod h1:nDyEzZ8ogv936Cinf6g1RU9MRY64Ir93oCnqb9wxYW0= +github.com/spf13/pflag v1.0.6/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= +github.com/spf13/pflag v1.0.7 h1:vN6T9TfwStFPFM5XzjsvmzZkLuaLX+HS+0SeFLRgU6M= +github.com/spf13/pflag v1.0.7/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= github.com/spf13/viper v1.19.0 h1:RWq5SEjt8o25SROyN3z2OrDB9l7RPd3lwTWU8EcEdcI= github.com/spf13/viper v1.19.0/go.mod h1:GQUN9bilAbhU/jgc1bKs99f/suXKeUMct8Adx5+Ntkg= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= From 5ce5228cbc68f4157b72f7c1a33f87b7030b6437 Mon Sep 17 00:00:00 2001 From: Mayank Pande Date: Thu, 26 Mar 2026 11:21:29 +0530 Subject: [PATCH 7/7] fix: evict stale CounterVec label combinations to prevent series explosion (#231) 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()