From 281e60f5cbfe5a6f3c67f86e7f90e2446853b0bb Mon Sep 17 00:00:00 2001 From: mayankpande88 Date: Tue, 17 Feb 2026 16:03:45 +0530 Subject: [PATCH 1/8] fix: add missing patternsPerLevelLimit param to logparser.NewParser calls --- common/log_parser_test.go | 2 +- containers/container.go | 6 +++--- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/common/log_parser_test.go b/common/log_parser_test.go index 75daf2bf..d90aac1d 100644 --- a/common/log_parser_test.go +++ b/common/log_parser_test.go @@ -12,7 +12,7 @@ import ( func TestLogParser(t *testing.T) { ch := make(chan logparser.LogEntry) - parser := logparser.NewParser(ch, nil, nil, 1*time.Second, false) + parser := logparser.NewParser(ch, nil, nil, 1*time.Second, 256, false) ch <- logparser.LogEntry{Timestamp: time.Now(), Content: "INFO:root:AWS access key: AKIAUCTZOIG66SPQV67B", Level: logparser.LevelInfo} diff --git a/containers/container.go b/containers/container.go index 2c287c0c..040573ca 100644 --- a/containers/container.go +++ b/containers/container.go @@ -1782,7 +1782,7 @@ func (c *Container) runLogParser(logPath string) { return } ch := make(chan logparser.LogEntry) - parser := logparser.NewParser(ch, nil, logs.OtelLogEmitter(containerId), multilineCollectorTimeout, *flags.DisableSensitiveLogParsing) + parser := logparser.NewParser(ch, nil, logs.OtelLogEmitter(containerId), multilineCollectorTimeout, *flags.LogPatternsPerContainer, *flags.DisableSensitiveLogParsing) reader, err := logs.NewTailReader(proc.HostPath(logPath), ch) if err != nil { klog.Warningln(err) @@ -1802,7 +1802,7 @@ func (c *Container) runLogParser(logPath string) { klog.Warningln(err) return } - parser := logparser.NewParser(ch, nil, logs.OtelLogEmitter(containerId), multilineCollectorTimeout, *flags.DisableSensitiveLogParsing) + parser := logparser.NewParser(ch, nil, logs.OtelLogEmitter(containerId), multilineCollectorTimeout, *flags.LogPatternsPerContainer, *flags.DisableSensitiveLogParsing) stop := func() { JournaldUnsubscribe(c.metadata.systemd.Unit) } @@ -1819,7 +1819,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.DisableSensitiveLogParsing) + parser := logparser.NewParser(ch, c.metadata.logDecoder, logs.OtelLogEmitter(containerId), multilineCollectorTimeout, *flags.LogPatternsPerContainer, *flags.DisableSensitiveLogParsing) reader, err := logs.NewTailReader(proc.HostPath(c.metadata.logPath), ch) if err != nil { klog.Warningln(err) From eef39b9e436db669ce272e154c4178376dd72c97 Mon Sep 17 00:00:00 2001 From: mayankpande88 Date: Tue, 17 Feb 2026 16:36:48 +0530 Subject: [PATCH 2/8] fix: update logparser dep, sanitize log paths, replace test credential --- common/log_parser_test.go | 6 +++--- containers/container.go | 15 +++++++++------ go.mod | 3 ++- go.sum | 6 ++++-- 4 files changed, 18 insertions(+), 12 deletions(-) diff --git a/common/log_parser_test.go b/common/log_parser_test.go index d90aac1d..34924d5b 100644 --- a/common/log_parser_test.go +++ b/common/log_parser_test.go @@ -10,11 +10,11 @@ import ( ) func TestLogParser(t *testing.T) { - + const defaultPatternsPerLevel = 256 ch := make(chan logparser.LogEntry) - parser := logparser.NewParser(ch, nil, nil, 1*time.Second, 256, false) + parser := logparser.NewParser(ch, nil, nil, 1*time.Second, defaultPatternsPerLevel, false) - ch <- logparser.LogEntry{Timestamp: time.Now(), Content: "INFO:root:AWS access key: AKIAUCTZOIG66SPQV67B", Level: logparser.LevelInfo} + ch <- logparser.LogEntry{Timestamp: time.Now(), Content: "INFO:root:AWS access key: AKIAIOSFODNN7EXAMPLE", Level: logparser.LevelInfo} // wait for 10 seconds time.Sleep(10 * time.Second) diff --git a/containers/container.go b/containers/container.go index 040573ca..e4bb2a4c 100644 --- a/containers/container.go +++ b/containers/container.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "os" + "path/filepath" "sort" "strings" "sync" @@ -1995,12 +1996,14 @@ func resolveFd(pid uint32, fd uint64) (mntId string, logPath string) { } mntId = info.MntId - if info.Flags&os.O_WRONLY != 0 && strings.HasPrefix(info.Dest, "/var/log/") && - !strings.HasPrefix(info.Dest, "/var/log/pods/") && - !strings.HasPrefix(info.Dest, "/var/log/containers/") && - !strings.HasPrefix(info.Dest, "/var/log/journal/") { - - logPath = info.Dest + if info.Flags&os.O_WRONLY != 0 { + cleaned := filepath.Clean(info.Dest) + if strings.HasPrefix(cleaned, "/var/log/") && + !strings.HasPrefix(cleaned, "/var/log/pods/") && + !strings.HasPrefix(cleaned, "/var/log/containers/") && + !strings.HasPrefix(cleaned, "/var/log/journal/") { + logPath = cleaned + } } return } diff --git a/go.mod b/go.mod index 30f324cb..fbe58abc 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-20250610175416-3edee28c2059 + github.com/nudgebee/logparser v0.0.0-20260105180412-79ec196fcae5 github.com/opencontainers/runtime-spec v1.1.0 github.com/prometheus/client_golang v1.20.5 github.com/prometheus/client_model v0.6.1 @@ -123,6 +123,7 @@ require ( github.com/hashicorp/hcl v1.0.1-vault-5 // indirect github.com/ianlancetaylor/demangle v0.0.0-20240312041847-bd984b5ce465 // indirect github.com/inconshreveable/mousetrap v1.1.0 // indirect + github.com/jaeyo/go-drain3 v0.1.2 // indirect 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 diff --git a/go.sum b/go.sum index d7581be7..f1cd7f43 100644 --- a/go.sum +++ b/go.sum @@ -230,6 +230,8 @@ github.com/ianlancetaylor/demangle v0.0.0-20240312041847-bd984b5ce465 h1:KwWnWVW github.com/ianlancetaylor/demangle v0.0.0-20240312041847-bd984b5ce465/go.mod h1:gx7rwoVhcfuVKG5uya9Hs3Sxj7EIvldVofAWIUtGouw= github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8= github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw= +github.com/jaeyo/go-drain3 v0.1.2 h1:fY21wgbwhzzaoRNSQ+6HVbpYw4KkAYjCFCoERYozIJ8= +github.com/jaeyo/go-drain3 v0.1.2/go.mod h1:6xr/0Dmq3BglAIZ5tDKiQiZvXevU1rE+qpfYZic9h9Y= github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY= github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y= github.com/josharian/native v0.0.0-20200817173448-b6b71def0850/go.mod h1:7X/raswPFr05uY3HiLlYeyQntB6OO7E/d2Cu7qoaN2w= @@ -321,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-20250610175416-3edee28c2059 h1:uBl8oxU66pg5348GTrIvRiNz1er9Wc7Ild+Kg0Qq+fA= -github.com/nudgebee/logparser v0.0.0-20250610175416-3edee28c2059/go.mod h1:ddmUa8en2sToRL+bBvFkuS1AHvA6Msn56v3BpLMbKoc= +github.com/nudgebee/logparser v0.0.0-20260105180412-79ec196fcae5 h1:2WScOnrFoCGTn6TBRo+inJhycp8t1zTh3yFsbqRiv+s= +github.com/nudgebee/logparser v0.0.0-20260105180412-79ec196fcae5/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 060e11d58b3c7d1f9613f0d27483e3af85731be4 Mon Sep 17 00:00:00 2001 From: mayankpande88 Date: Wed, 18 Feb 2026 14:10:44 +0530 Subject: [PATCH 3/8] fix: update logparser dep to fix sensitive pattern loading The previous version had a corrupted sensitive_patterns.json that caused "cannot unmarshal object into Go value of type []SensitivePattern" errors, breaking sensitive data detection in log parsing. --- go.mod | 2 +- go.sum | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/go.mod b/go.mod index fbe58abc..f1c571c8 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-20260105180412-79ec196fcae5 + github.com/nudgebee/logparser v0.0.0-20260218041043-99ea4437d928 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 f1cd7f43..b44b42ff 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-20260105180412-79ec196fcae5 h1:2WScOnrFoCGTn6TBRo+inJhycp8t1zTh3yFsbqRiv+s= -github.com/nudgebee/logparser v0.0.0-20260105180412-79ec196fcae5/go.mod h1:oFnM9D6YEjZzb1jy0kJ/NUkTbkZA+BgNYjpLO4n4szA= +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/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 f044d6cdbdb80bde98e4fb327222c55e7ea42d3b Mon Sep 17 00:00:00 2001 From: mayankpande88 Date: Sat, 7 Mar 2026 14:14:36 +0530 Subject: [PATCH 4/8] fix: reduce memory usage, fix race conditions, and fd leaks Memory optimizations: - Auto-tune GOMEMLIMIT to 90% of cgroup memory limit - Reduce connectionStats LRU cache from 8192 to 4096 - Reduce event channel buffer from 10000 to 2000 - Cap ip2fqdn map at 10000 entries - Reduce label interner max from 10000 to 5000 - Reduce LLM stream limits (maxActiveStreams 1000->200, buffer 1MB->64KB) - Cache stripped Go binary paths to avoid repeated uprobe attach attempts - Reduce GC interval from 10min to 5min - Clean up HTTP/2 HPACK parsers for dead connections in gc() Race condition fixes in Container.Collect(): - Snapshot connectionStats LRU under RLock (not concurrent-safe) - Call getListens/getProxiedListens under RLock - Protect logSamples map reads/writes with lock - Read restarts/oomKills under RLock - Read c.processes under RLock in onConnectionOpen before lock scope File descriptor leak fixes in logs/tail_reader.go: - Close file on Stat/Seek failure in NewTailReader - Close old file before setting nil in moved() - Cap partial line buffer at 64KB to prevent unbounded growth - Fix prefix accumulation bug (was replacing instead of appending) --- containers/container.go | 60 +++++++++++++++++++++++++++++++++------- containers/l7.go | 8 +++--- containers/llm_stream.go | 4 +-- containers/registry.go | 16 +++++++---- ebpftracer/tls.go | 22 ++++++++++++--- logs/tail_reader.go | 10 ++++++- main.go | 38 +++++++++++++++++++++++++ 7 files changed, 131 insertions(+), 27 deletions(-) diff --git a/containers/container.go b/containers/container.go index e4bb2a4c..016d5260 100644 --- a/containers/container.go +++ b/containers/container.go @@ -33,7 +33,7 @@ import ( ) var ( - gcInterval = 10 * time.Minute + gcInterval = 5 * time.Minute pingTimeout = 300 * time.Millisecond multilineCollectorTimeout = time.Second payloadThreshold = 1024 * 1024 @@ -41,7 +41,7 @@ var ( ) const ( - connectionStatsCacheSize = 8192 // LRU cache size for connection stats + connectionStatsCacheSize = 4096 // LRU cache size for connection stats ) type ContainerID string @@ -337,7 +337,12 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) { ch <- gauge(metrics.ContainerInfo, 1, c.metadata.image, c.metadata.systemd.TriggeredBy, c.metadata.systemd.Type) } - ch <- counter(metrics.Restarts, float64(c.restarts)) + c.lock.RLock() + restarts := c.restarts + oomKills := c.oomKills + c.lock.RUnlock() + + ch <- counter(metrics.Restarts, float64(restarts)) if cpu := c.cgroup.CpuStat(); cpu != nil { if cpu.LimitCores > 0 { @@ -370,8 +375,8 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) { ch <- counter(metrics.PsiIO, psi.IOSecondsFull, "full") } - if c.oomKills > 0 { - ch <- counter(metrics.OOMKills, float64(c.oomKills)) + if oomKills > 0 { + ch <- counter(metrics.OOMKills, float64(oomKills)) } if disks, err := node.GetDisks(); err == nil { @@ -398,20 +403,38 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) { } } - for addr, open := range c.getListens() { + // Snapshot listens under lock since getListens/getProxiedListens access c.listens and c.processes + c.lock.RLock() + listens := c.getListens() + proxiedListens := c.getProxiedListens() + c.lock.RUnlock() + + for addr, open := range listens { ch <- gauge(metrics.NetListenInfo, float64(open), addr.String(), "") } - for proxy, addrs := range c.getProxiedListens() { + for proxy, addrs := range proxiedListens { for addr := range addrs { ch <- gauge(metrics.NetListenInfo, 1, addr.String(), proxy) } } + // 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() { - stats, ok := c.connectionStats.Peek(d) - if !ok { - continue + if stats, ok := c.connectionStats.Peek(d); ok { + connSnapshots = append(connSnapshots, connStatSnapshot{key: d, stats: *stats}) } + } + c.lock.RUnlock() + + for _, snap := range connSnapshots { + d := snap.key + stats := snap.stats workload_src := c.srcWorkload workload_dest := d.GetDestinationWorkload() actualDestWorkload := d.GetActualDestinationWorkload() @@ -463,10 +486,14 @@ 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 { sample = common.TruncateUtf8(ctr.Sample, *flags.MaxLabelLength) + c.lock.Lock() c.logSamples[ctr.Hash] = sample + c.lock.Unlock() } ch <- counter(metrics.LogMessages, float64(ctr.Messages), source, ctr.Level.String(), ctr.Hash, sample) } @@ -730,7 +757,9 @@ func (c *Container) onConnectionOpen(pid uint32, fd uint64, src, dst, actualDst if common.PortFilter.ShouldBeSkipped(dst.Port()) { return } + c.lock.RLock() p := c.processes[pid] + c.lock.RUnlock() if p == nil { return } @@ -1879,6 +1908,17 @@ 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. + if c.googleHTTP2Parsers != nil { + for pidFd := range c.googleHTTP2Parsers { + if _, alive := c.connectionsByPidFd[pidFd]; !alive { + delete(c.googleHTTP2Parsers, pidFd) + } + } + } + for dst, at := range c.lastConnectionAttempts { _, active := establishedDst[dst] if !active && !at.IsZero() && now.Sub(at) > gcInterval { diff --git a/containers/l7.go b/containers/l7.go index 2888c108..5bdb50d1 100644 --- a/containers/l7.go +++ b/containers/l7.go @@ -45,11 +45,11 @@ func (si *stringInterner) intern(s string) string { } // Limit cache size to prevent unbounded growth - if len(si.cache) > 10000 { - // Clear half the cache when it gets too large - newCache := make(map[string]string, 5000) + if len(si.cache) > 5000 { + // Clear most of the cache when it gets too large + newCache := make(map[string]string, 2000) for k, v := range si.cache { - if len(newCache) >= 5000 { + if len(newCache) >= 2000 { break } newCache[k] = v diff --git a/containers/llm_stream.go b/containers/llm_stream.go index ebf7a788..35f4f520 100644 --- a/containers/llm_stream.go +++ b/containers/llm_stream.go @@ -14,8 +14,8 @@ import ( ) const ( - maxActiveStreams = 1000 // Max concurrent streams - maxBufferPerStream = 1 << 20 // 1MB per stream + maxActiveStreams = 200 // Max concurrent streams + maxBufferPerStream = 64 * 1024 // 64KB per stream (only need final usage JSON) streamIdleTimeout = 30 * time.Second // Release abandoned streams streamMaxDuration = 5 * time.Minute // Cap for very long streams streamGCInterval = 10 * time.Second // GC interval diff --git a/containers/registry.go b/containers/registry.go index 2b316721..8b53f72a 100644 --- a/containers/registry.go +++ b/containers/registry.go @@ -131,7 +131,7 @@ func NewRegistry(reg prometheus.Registerer, processInfoCh chan<- ProcessInfo, ip r := &Registry{ reg: reg, - events: make(chan ebpftracer.Event, 10000), + events: make(chan ebpftracer.Event, 2000), containersById: map[ContainerID]*Container{}, containersByCgroupId: map[string]*Container{}, containersByPid: map[uint32]*Container{}, @@ -392,6 +392,8 @@ func (r *Registry) handleEvents(ch <-chan ebpftracer.Event) { } // processL7Event handles an L7 event, queueing it for retry if the connection isn't found yet +const maxIP2FQDNEntries = 10000 + func (r *Registry) processL7Event(e ebpftracer.Event) { if c := r.containersByPid[e.Pid]; c != nil { klog.V(2).Infof("L7_EVENT_CONTAINER_FOUND: pid=%d container=%s", e.Pid, c.id) @@ -405,8 +407,9 @@ func (r *Registry) processL7Event(e ebpftracer.Event) { } r.ip2fqdnLock.Lock() for ip, domain := range ip2fqdn { - r.ip2fqdn[ip] = domain - // Also update IP resolver cache for trace hostname display + if len(r.ip2fqdn) < maxIP2FQDNEntries { + r.ip2fqdn[ip] = domain + } r.ip_resolver.CacheDNS(ip.String(), domain.FQDN) } r.ip2fqdnLock.Unlock() @@ -415,8 +418,9 @@ func (r *Registry) processL7Event(e ebpftracer.Event) { ip2fqdn := r.handleHostDNSRequest(e.L7Request) r.ip2fqdnLock.Lock() for ip, domain := range ip2fqdn { - r.ip2fqdn[ip] = domain - // Also update IP resolver cache for trace hostname display + if len(r.ip2fqdn) < maxIP2FQDNEntries { + r.ip2fqdn[ip] = domain + } r.ip_resolver.CacheDNS(ip.String(), domain.FQDN) } r.ip2fqdnLock.Unlock() @@ -429,7 +433,7 @@ func (r *Registry) queueL7EventForRetry(e ebpftracer.Event) { defer r.pendingL7EventsLock.Unlock() // Limit queue size to prevent memory issues - const maxPendingEvents = 1000 + const maxPendingEvents = 500 if len(r.pendingL7Events) >= maxPendingEvents { klog.V(2).Infof("L7_EVENT_QUEUE_FULL: dropping event pid=%d fd=%d", e.Pid, e.Fd) return diff --git a/ebpftracer/tls.go b/ebpftracer/tls.go index 4545b83c..e8ae0ddd 100644 --- a/ebpftracer/tls.go +++ b/ebpftracer/tls.go @@ -10,6 +10,7 @@ import ( "os" "regexp" "strings" + "sync" "unsafe" "github.com/cilium/ebpf/link" @@ -49,6 +50,11 @@ var ( libCryptoRe = regexp.MustCompile(`libcrypto\.so(\.\d+)*`) // A more specific regex for psycopg2's bundled libs psycopg2LibRe = regexp.MustCompile(`lib(ssl|crypto)-[a-f0-9]+\.so\.\d+`) + + // 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{} ) func (t *Tracer) AttachOpenSslUprobes(pid uint32) []link.Link { @@ -170,9 +176,15 @@ func (t *Tracer) AttachGoTlsUprobes(pid uint32) ([]link.Link, bool) { path := proc.Path(pid, "exe") - // DEBUG: Log every attempt to attach Go TLS uprobes (at Info level for visibility) exeName, _ := os.Readlink(path) - klog.Infof("GO_TLS_ATTACH_ATTEMPT: pid=%d exe=%s", pid, exeName) + klog.V(2).Infof("GO_TLS_ATTACH_ATTEMPT: pid=%d exe=%s", pid, exeName) + + // 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 { + klog.V(3).Infof("GO_TLS_SKIP_STRIPPED: pid=%d exe=%s", pid, exeName) + return nil, true // still a Go app, just stripped + } var err error var name, version string @@ -180,8 +192,6 @@ func (t *Tracer) AttachGoTlsUprobes(pid uint32) ([]link.Link, bool) { if err != nil { for _, s := range []string{"not a Go executable", "no such file or directory", "no such process", "permission denied"} { if strings.HasSuffix(err.Error(), s) { - // DEBUG: Log filtered errors at Info level for visibility - klog.Infof("GO_TLS_FILTERED: pid=%d exe=%s msg=%s err=%s", pid, exeName, msg, err.Error()) return } } @@ -245,6 +255,10 @@ func (t *Tracer) AttachGoTlsUprobes(pid uint32) ([]link.Link, bool) { if err != nil { if writeSymbol == goTlsWriteSymbol { log("failed to get write symbol", err) + // Cache this exe as stripped to skip future attempts + if exeName != "" { + strippedGoExeCache.Store(exeName, struct{}{}) + } return nil, isGolangApp } continue // S2A symbol is optional diff --git a/logs/tail_reader.go b/logs/tail_reader.go index c8176879..57f6eb8e 100644 --- a/logs/tail_reader.go +++ b/logs/tail_reader.go @@ -41,14 +41,17 @@ func NewTailReader(fileName string, ch chan<- logparser.LogEntry) (*TailReader, return nil, err } if r.info, err = r.file.Stat(); err != nil { + _ = r.file.Close() return nil, err } if _, err = r.file.Seek(0, io.SeekEnd); err != nil { + _ = r.file.Close() return nil, err } r.reader = bufio.NewReader(r.file) go func() { + const maxPrefixLen = 64 * 1024 // 64KB cap on partial line buffer var prefix string for { select { @@ -58,7 +61,11 @@ func NewTailReader(fileName string, ch chan<- logparser.LogEntry) (*TailReader, default: line, err := r.reader.ReadString('\n') if err != nil { - prefix = line + if len(prefix)+len(line) > maxPrefixLen { + prefix = "" // drop oversized partial line + } else { + prefix += line + } r.poll(ctx) continue } @@ -130,6 +137,7 @@ func (r *TailReader) moved(info os.FileInfo) bool { if !os.SameFile(r.info, info) { f, err := os.Open(r.fileName) if err != nil { + _ = r.file.Close() r.file = nil return false } diff --git a/main.go b/main.go index bcd4f6a7..ea0de2a0 100644 --- a/main.go +++ b/main.go @@ -3,6 +3,7 @@ package main import ( "bytes" "flag" + "fmt" "log" "net/http" _ "net/http/pprof" @@ -124,6 +125,17 @@ func getClientSet() (*kubernetes.Clientset, error) { } func main() { + // Set GOMEMLIMIT to 90% of cgroup memory limit if available, otherwise use env var. + // This tells the Go GC to be more aggressive when approaching the limit, + // preventing OOMKills by trading CPU for lower memory usage. + if os.Getenv("GOMEMLIMIT") == "" { + if limit, err := readCgroupMemoryLimit(); err == nil && limit > 0 { + softLimit := int64(float64(limit) * 0.9) + runtime.SetMemoryLimit(softLimit) + log.Printf("GOMEMLIMIT set to %dMiB (90%% of cgroup limit %dMiB)", softLimit/1024/1024, limit/1024/1024) + } + } + // Initialize klog flags to prevent duplicate output klog.InitFlags(nil) if v := os.Getenv("KLOG_V"); v != "" { @@ -261,3 +273,29 @@ func (o *RateLimitedLogOutput) Write(data []byte) (int, error) { } return os.Stderr.Write(data) } + +// readCgroupMemoryLimit reads the memory limit from cgroup v2 or v1. +func readCgroupMemoryLimit() (int64, error) { + // Try cgroup v2 first + if data, err := os.ReadFile("/sys/fs/cgroup/memory.max"); err == nil { + s := strings.TrimSpace(string(data)) + if s != "max" { + var limit int64 + if _, err := fmt.Sscanf(s, "%d", &limit); err == nil { + return limit, nil + } + } + } + // Try cgroup v1 + if data, err := os.ReadFile("/sys/fs/cgroup/memory/memory.limit_in_bytes"); err == nil { + s := strings.TrimSpace(string(data)) + var limit int64 + if _, err := fmt.Sscanf(s, "%d", &limit); err == nil { + // cgroup v1 reports a very large number when unlimited + if limit < 1<<62 { + return limit, nil + } + } + } + return 0, fmt.Errorf("unable to read cgroup memory limit") +} From b8e75211a5e10a31888333e9222667b5005f6886 Mon Sep 17 00:00:00 2001 From: mayankpande88 Date: Sat, 7 Mar 2026 14:14:49 +0530 Subject: [PATCH 5/8] fix: validate eBPF durations to prevent corrupted spans causing OTel OOM eBPF kernel probes occasionally produce garbage duration values where the uint64 has the high bit set. Casting to time.Duration (int64) produces negative durations, creating spans where end_time < start_time. These poison spans cause ClickHouse batch rejections, OTel collector retry accumulation, and eventual OOMKill (165 restarts observed). Source-level fix (eBPF event parsing): - Add safeDuration() that returns 0 for values >= 1 hour or == 0 - Replace all uint64->time.Duration casts in perf/ringbuf readers HTTP/2 parser fix: - Add safeKernelDuration() for kernel timestamp subtraction - Prevents unsigned underflow when events arrive out of order Defense-in-depth (span creation): - createSpan: drop spans with duration <= 0 or > 1 hour - LLMRequest: drop spans with zero/invalid timestamps or > 1 hour --- ebpftracer/l7/http2.go | 19 ++++++++++++++++--- ebpftracer/tracer.go | 16 +++++++++++++--- tracing/tracing.go | 9 +++++++++ 3 files changed, 38 insertions(+), 6 deletions(-) diff --git a/ebpftracer/l7/http2.go b/ebpftracer/l7/http2.go index b14f1aa8..6b2169b6 100644 --- a/ebpftracer/l7/http2.go +++ b/ebpftracer/l7/http2.go @@ -12,9 +12,22 @@ import ( "k8s.io/klog/v2" ) +// safeKernelDuration computes the duration between two kernel timestamps, +// returning 0 if the result would underflow or exceed 1 hour. +func safeKernelDuration(end, start uint64) time.Duration { + if end <= start { + return 0 + } + d := end - start + if d >= uint64(time.Hour) { + return 0 + } + return time.Duration(d) +} + const ( http2FrameHeaderLength = 9 - http2DecoderGcInterval = uint64(10 * time.Minute) + http2DecoderGcInterval = uint64(2 * time.Minute) // HTTP/2 flags http2FlagEndStream = 0x01 @@ -521,7 +534,7 @@ frameLoop: r.GrpcStatus = -1 } } - r.Duration = time.Duration(kernelTime - r.kernelTime) + r.Duration = safeKernelDuration(kernelTime, r.kernelTime) res = append(res, *r) delete(p.activeRequests, streamId) } @@ -555,7 +568,7 @@ frameLoop: } else { r.GrpcStatus = -1 } - r.Duration = time.Duration(kernelTime - r.kernelTime) + r.Duration = safeKernelDuration(kernelTime, r.kernelTime) res = append(res, *r) delete(p.activeRequests, streamId) } diff --git a/ebpftracer/tracer.go b/ebpftracer/tracer.go index 126a557d..7d59b6ba 100644 --- a/ebpftracer/tracer.go +++ b/ebpftracer/tracer.go @@ -438,6 +438,16 @@ func getLostSamplesTracker(name string) *lostSamplesTracker { return tracker.(*lostSamplesTracker) } +// safeDuration converts a uint64 nanosecond value from eBPF to time.Duration. +// Returns 0 for values that would overflow int64 (>= 2^63) or exceed 1 hour, +// which indicate corrupted eBPF timestamps. +func safeDuration(ns uint64) time.Duration { + if ns == 0 || ns >= uint64(time.Hour) { + return 0 + } + return time.Duration(ns) +} + func (t *lostSamplesTracker) recordLostSamples(name string, count uint64, cpu int) { t.count.Add(count) now := time.Now().Unix() @@ -506,7 +516,7 @@ func runEventsReader(name string, r *perf.Reader, ch chan<- Event, typ perfMapTy Protocol: l7.Protocol(data[32]), Method: l7.Method(data[33]), Status: l7.Status(int32(binary.LittleEndian.Uint32(data[20:24]))), - Duration: time.Duration(binary.LittleEndian.Uint64(data[24:32])), + Duration: safeDuration(binary.LittleEndian.Uint64(data[24:32])), StatementId: binary.LittleEndian.Uint32(data[36:40]), PayloadSize: payloadSize, ResponseSize: responseSize, @@ -558,7 +568,7 @@ func runEventsReader(name string, r *perf.Reader, ch chan<- Event, typ perfMapTy ActualDstAddr: ipPort(data[86:102], binary.LittleEndian.Uint16(data[52:54])), Fd: binary.LittleEndian.Uint64(data[0:8]), Timestamp: binary.LittleEndian.Uint64(data[8:16]), - Duration: time.Duration(binary.LittleEndian.Uint64(data[16:24])), + Duration: safeDuration(binary.LittleEndian.Uint64(data[16:24])), } if typ == EventTypeConnectionClose { event.TrafficStats = &TrafficStats{ @@ -622,7 +632,7 @@ func runRingbufEventsReader(name string, r *ringbuf.Reader, ch chan<- Event) { Protocol: l7.Protocol(data[32]), Method: l7.Method(data[33]), Status: l7.Status(int32(binary.LittleEndian.Uint32(data[20:24]))), - Duration: time.Duration(binary.LittleEndian.Uint64(data[24:32])), + Duration: safeDuration(binary.LittleEndian.Uint64(data[24:32])), StatementId: binary.LittleEndian.Uint32(data[36:40]), PayloadSize: payloadSize, ResponseSize: responseSize, diff --git a/tracing/tracing.go b/tracing/tracing.go index 00e8a85c..0f176c75 100644 --- a/tracing/tracing.go +++ b/tracing/tracing.go @@ -184,6 +184,9 @@ func (t *Trace) createSpan(name string, duration time.Duration, error bool, trac if t.tracer.otel == nil { return } + if duration <= 0 || duration > time.Hour { + return + } end := time.Now() if !shouldSample() { @@ -456,6 +459,12 @@ func (t *Trace) LLMRequest(info LLMStreamInfo) { if t == nil || t.tracer.otel == nil { return } + if info.RequestTime.IsZero() || info.CompletionTime.IsZero() || !info.CompletionTime.After(info.RequestTime) { + return + } + if info.CompletionTime.Sub(info.RequestTime) > time.Hour { + return + } if !shouldSample() { return From 46eaeeb8ec3fa268eaeb11aec9fbc7d2e27759ed Mon Sep 17 00:00:00 2001 From: mayankpande88 Date: Sat, 7 Mar 2026 14:45:42 +0530 Subject: [PATCH 6/8] fix: address review comments - Use double-checked locking for logSamples to avoid redundant writes - Use strconv.ParseInt instead of fmt.Sscanf for stricter cgroup parsing --- containers/container.go | 7 +++++-- main.go | 7 +++---- 2 files changed, 8 insertions(+), 6 deletions(-) diff --git a/containers/container.go b/containers/container.go index 11868002..c997c42a 100644 --- a/containers/container.go +++ b/containers/container.go @@ -488,9 +488,12 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) { sample, ok := c.logSamples[ctr.Hash] c.lock.RUnlock() if !ok { - sample = common.TruncateUtf8(ctr.Sample, *flags.MaxLabelLength) c.lock.Lock() - c.logSamples[ctr.Hash] = sample + sample, ok = c.logSamples[ctr.Hash] + if !ok { + sample = common.TruncateUtf8(ctr.Sample, *flags.MaxLabelLength) + c.logSamples[ctr.Hash] = sample + } c.lock.Unlock() } ch <- counter(metrics.LogMessages, float64(ctr.Messages), source, ctr.Level.String(), ctr.Hash, sample) diff --git a/main.go b/main.go index ea0de2a0..aa3eb19b 100644 --- a/main.go +++ b/main.go @@ -10,6 +10,7 @@ import ( "os" "os/signal" "runtime" + "strconv" "strings" "syscall" "time" @@ -280,8 +281,7 @@ func readCgroupMemoryLimit() (int64, error) { if data, err := os.ReadFile("/sys/fs/cgroup/memory.max"); err == nil { s := strings.TrimSpace(string(data)) if s != "max" { - var limit int64 - if _, err := fmt.Sscanf(s, "%d", &limit); err == nil { + if limit, err := strconv.ParseInt(s, 10, 64); err == nil { return limit, nil } } @@ -289,8 +289,7 @@ func readCgroupMemoryLimit() (int64, error) { // Try cgroup v1 if data, err := os.ReadFile("/sys/fs/cgroup/memory/memory.limit_in_bytes"); err == nil { s := strings.TrimSpace(string(data)) - var limit int64 - if _, err := fmt.Sscanf(s, "%d", &limit); err == nil { + if limit, err := strconv.ParseInt(s, 10, 64); err == nil { // cgroup v1 reports a very large number when unlimited if limit < 1<<62 { return limit, nil From 4d98ff5c1f1be0a30d855cf8cc6425d603d2fa46 Mon Sep 17 00:00:00 2001 From: mayankpande88 Date: Sat, 7 Mar 2026 15:14:19 +0530 Subject: [PATCH 7/8] fix: gofmt alignment in llm_stream.go --- containers/llm_stream.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/containers/llm_stream.go b/containers/llm_stream.go index da960152..c2369f62 100644 --- a/containers/llm_stream.go +++ b/containers/llm_stream.go @@ -14,7 +14,7 @@ import ( ) const ( - maxActiveStreams = 200 // Max concurrent streams + maxActiveStreams = 200 // Max concurrent streams maxBufferPerStream = 64 * 1024 // 64KB per stream (only need final usage JSON) streamIdleTimeout = 30 * time.Second // Release abandoned streams streamMaxDuration = 5 * time.Minute // Cap for very long streams From f85f5a0cf26d36593c12a3d065a32a8943eb2b53 Mon Sep 17 00:00:00 2001 From: mayankpande88 Date: Sat, 7 Mar 2026 15:40:56 +0530 Subject: [PATCH 8/8] fix: use debug.SetMemoryLimit instead of runtime.SetMemoryLimit --- main.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/main.go b/main.go index aa3eb19b..e18cb71e 100644 --- a/main.go +++ b/main.go @@ -10,6 +10,7 @@ import ( "os" "os/signal" "runtime" + "runtime/debug" "strconv" "strings" "syscall" @@ -132,7 +133,7 @@ func main() { if os.Getenv("GOMEMLIMIT") == "" { if limit, err := readCgroupMemoryLimit(); err == nil && limit > 0 { softLimit := int64(float64(limit) * 0.9) - runtime.SetMemoryLimit(softLimit) + debug.SetMemoryLimit(softLimit) log.Printf("GOMEMLIMIT set to %dMiB (90%% of cgroup limit %dMiB)", softLimit/1024/1024, limit/1024/1024) } }