fix: reduce memory usage, fix race conditions, and prevent corrupted eBPF spans

containers/container.go

@@ -33,15 +33,15 @@ import (
 )
 
 var (
-	gcInterval                = 10 * time.Minute
+	gcInterval                = 5 * time.Minute
 	pingTimeout               = 300 * time.Millisecond
 	multilineCollectorTimeout = time.Second
 	payloadThreshold          = 1024 * 1024
 	gpuStatsWindow            = 15 * time.Second
 )
 
 const (
-	connectionStatsCacheSize = 8192 // LRU cache size for connection stats
+	connectionStatsCacheSize = 4096 // LRU cache size for connection stats
 )
 
 type ContainerID string
@@ -316,7 +316,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 {
@@ -349,8 +354,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 {
@@ -377,37 +382,53 @@ 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() {
+		if stats, ok := c.connectionStats.Peek(d); ok {
+			connSnapshots = append(connSnapshots, connStatSnapshot{key: d, stats: *stats})
+		}
+	}
+	c.lock.RUnlock()
+
 	// Aggregate connection stats by enriched key to avoid duplicate metrics
 	// when multiple IPs resolve to the same FQDN (e.g., Google's shared IPs)
 	aggregatedStats := map[common.DestinationKey]*ConnectionStats{}
-	for _, d := range c.connectionStats.Keys() {
-		stats, ok := c.connectionStats.Peek(d)
-		if !ok {
-			continue
-		}
-		enrichedKey := c.enrichDestinationKey(d)
+	for _, snap := range connSnapshots {
+		enrichedKey := c.enrichDestinationKey(snap.key)
 		if existing, ok := aggregatedStats[enrichedKey]; ok {
-			existing.Count += stats.Count
-			existing.TotalTime += stats.TotalTime
-			existing.Retransmissions += stats.Retransmissions
-			existing.BytesSent += stats.BytesSent
-			existing.BytesReceived += stats.BytesReceived
+			existing.Count += snap.stats.Count
+			existing.TotalTime += snap.stats.TotalTime
+			existing.Retransmissions += snap.stats.Retransmissions
+			existing.BytesSent += snap.stats.BytesSent
+			existing.BytesReceived += snap.stats.BytesReceived
 		} else {
 			aggregatedStats[enrichedKey] = &ConnectionStats{
-				Count:           stats.Count,
-				TotalTime:       stats.TotalTime,
-				Retransmissions: stats.Retransmissions,
-				BytesSent:       stats.BytesSent,
-				BytesReceived:   stats.BytesReceived,
+				Count:           snap.stats.Count,
+				TotalTime:       snap.stats.TotalTime,
+				Retransmissions: snap.stats.Retransmissions,
+				BytesSent:       snap.stats.BytesSent,
+				BytesReceived:   snap.stats.BytesReceived,
 			}
 		}
 	}
@@ -463,10 +484,17 @@ 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.logSamples[ctr.Hash] = sample
+					c.lock.Lock()
+					sample, ok = c.logSamples[ctr.Hash]
+					if !ok {
+						sample = common.TruncateUtf8(ctr.Sample, *flags.MaxLabelLength)
+						c.logSamples[ctr.Hash] = sample
+					}
+					c.lock.Unlock()
 				}
 				ch <- counter(metrics.LogMessages, float64(ctr.Messages), source, ctr.Level.String(), ctr.Hash, sample)
 			}
@@ -730,7 +758,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
 	}
@@ -1880,6 +1910,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 {

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

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

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(5).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(3).Infof("L7_EVENT_QUEUE_FULL: dropping event pid=%d fd=%d", e.Pid, e.Fd)
 		return

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

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,17 +176,22 @@ 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.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
 	log := func(msg string, err error) {
 		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.V(3).Infof("GO_TLS_FILTERED: pid=%d exe=%s msg=%s err=%s", pid, exeName, msg, err.Error())
 					return
 				}
@@ -245,6 +256,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

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,