diff --git a/containers/container.go b/containers/container.go index e0af5531..c997c42a 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 @@ -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 { diff --git a/containers/l7.go b/containers/l7.go index 1436961d..4292b4ae 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 3b6674ce..c2369f62 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 55c45106..0b15bf67 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(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 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/tls.go b/ebpftracer/tls.go index f895a418..b5eb7094 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,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 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/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..e18cb71e 100644 --- a/main.go +++ b/main.go @@ -3,12 +3,15 @@ package main import ( "bytes" "flag" + "fmt" "log" "net/http" _ "net/http/pprof" "os" "os/signal" "runtime" + "runtime/debug" + "strconv" "strings" "syscall" "time" @@ -124,6 +127,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) + debug.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 +275,27 @@ 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" { + if limit, err := strconv.ParseInt(s, 10, 64); 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)) + 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 + } + } + } + return 0, fmt.Errorf("unable to read cgroup memory limit") +} 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