From 77d35602e28cc4eacb7aa77d60aa2fa0ed7f89d0 Mon Sep 17 00:00:00 2001 From: mayankpande88 Date: Sat, 10 Jan 2026 16:33:32 +0530 Subject: [PATCH 1/8] fix: fix for unexpected token error --- containers/container.go | 34 +++++++++++++++++++++++++--------- containers/registry.go | 4 ++++ 2 files changed, 29 insertions(+), 9 deletions(-) diff --git a/containers/container.go b/containers/container.go index 25508fb1..f096eaa5 100644 --- a/containers/container.go +++ b/containers/container.go @@ -198,15 +198,31 @@ func NewContainer(id ContainerID, cg *cgroup.Cgroup, md *ContainerMetadata, pid return nil, err } defer netNs.Close() - split := strings.Split(string(id), "/") - if len(split) < 4 { - klog.Errorf("unexpected container id %s", id) - return nil, errors.New("unexpected container id") - } - namespace := split[2] - podName := split[3] - src_workload := registry.ip_resolver.ResolvePodOwner(podName, namespace) - klog.Infof("Pod %s/%s is owned by %s/%s/%s", namespace, podName, src_workload.Name, src_workload.Namespace, src_workload.Kind) + + // Resolve workload based on container type + var src_workload common.Workload + idStr := string(id) + split := strings.Split(idStr, "/") + + if strings.HasPrefix(idStr, "/k8s/") || strings.HasPrefix(idStr, "/k8s-cronjob/") { + // Kubernetes container: /k8s/namespace/pod/container or /k8s-cronjob/namespace/job/container + if len(split) < 4 { + klog.Errorf("unexpected k8s container id %s", id) + return nil, errors.New("unexpected container id") + } + namespace := split[2] + podName := split[3] + src_workload = registry.ip_resolver.ResolvePodOwner(podName, namespace) + klog.Infof("Pod %s/%s is owned by %s/%s/%s", namespace, podName, src_workload.Name, src_workload.Namespace, src_workload.Kind) + } else { + // Non-k8s containers (docker, systemd, swarm, nomad): use container name as workload + name := "" + if len(split) > 0 { + name = split[len(split)-1] + } + src_workload = common.Workload{Name: name, Kind: "container"} + klog.V(2).Infof("Non-k8s container %s using workload name: %s", id, name) + } cid := string(id) appId := common.ContainerIdToOtelServiceName(cid) diff --git a/containers/registry.go b/containers/registry.go index 4e41fad5..c3ec42b3 100644 --- a/containers/registry.go +++ b/containers/registry.go @@ -650,6 +650,10 @@ func calcId(cg *cgroup.Cgroup, md *ContainerMetadata) ContainerID { if strings.HasPrefix(cg.ContainerId, "/system.slice/crio-conmon-") { return "" } + // Skip systemd services that match the ignore control plane list + if ignoreControlPlane(cg.ContainerId) { + return "" + } return ContainerID(cg.ContainerId) case cgroup.ContainerTypeTalosRuntime: return ContainerID(cg.ContainerId) From 4905c8ce005239992512d9dab2fb734dd95ed07f Mon Sep 17 00:00:00 2001 From: mayankpande88 Date: Sat, 10 Jan 2026 16:55:26 +0530 Subject: [PATCH 2/8] fix: resolve concurrent map access crash in updateDelays and handle systemd services - Fix concurrent map read/write crash in updateDelays by using proper locking: - Get PIDs snapshot under read lock - Make syscalls without holding lock to avoid contention - Update delays under write lock - Fix "unexpected container id" errors for systemd services: - Only resolve pod owner for k8s containers (/k8s/ or /k8s-cronjob/ prefixes) - Use container name as workload for non-k8s containers - Skip systemd services that match IGNORE_CONTROL_PLANE list in calcId Co-Authored-By: Claude Opus 4.5 --- containers/container.go | 38 ++++++++++++++++++++++++++++++++------ 1 file changed, 32 insertions(+), 6 deletions(-) diff --git a/containers/container.go b/containers/container.go index f096eaa5..30a5def6 100644 --- a/containers/container.go +++ b/containers/container.go @@ -1139,18 +1139,44 @@ func (c *Container) onRetransmission(src netaddr.IPPort, dst netaddr.IPPort) boo } func (c *Container) updateDelays() { + // Get a snapshot of PIDs under read lock to avoid concurrent map access + c.lock.RLock() + pids := make([]uint32, 0, len(c.processes)) for pid := range c.processes { + pids = append(pids, pid) + } + c.lock.RUnlock() + + // Make syscalls without holding the lock to avoid contention + type pidDelayStats struct { + pid uint32 + cpuDelay time.Duration + diskDelay time.Duration + } + pidStats := make([]pidDelayStats, 0, len(pids)) + for _, pid := range pids { stats, err := TaskstatsTGID(pid) if err != nil { continue } - d := c.delaysByPid[pid] - c.delays.cpu += stats.CPUDelay - d.cpu - c.delays.disk += stats.BlockIODelay - d.disk - d.cpu = stats.CPUDelay - d.disk = stats.BlockIODelay - c.delaysByPid[pid] = d + pidStats = append(pidStats, pidDelayStats{ + pid: pid, + cpuDelay: stats.CPUDelay, + diskDelay: stats.BlockIODelay, + }) } + + // Update delays under write lock + c.lock.Lock() + for _, ps := range pidStats { + d := c.delaysByPid[ps.pid] + c.delays.cpu += ps.cpuDelay - d.cpu + c.delays.disk += ps.diskDelay - d.disk + d.cpu = ps.cpuDelay + d.disk = ps.diskDelay + c.delaysByPid[ps.pid] = d + } + c.lock.Unlock() } func (c *Container) updateNodejsStats(s NodejsStatsUpdate) { From d64fa04096e582cd2f0905946cf8e5d606aef075 Mon Sep 17 00:00:00 2001 From: mayankpande88 Date: Sat, 10 Jan 2026 17:11:06 +0530 Subject: [PATCH 3/8] fix: reduce l7_events log spam and increase perf buffer - Increase l7_events perf buffer from 64 to 128 pages per CPU to reduce lost samples - Add rate-limited logging for lost samples (aggregate and log every 10s instead of per-event) - Include CPU info in lost samples log for better debugging The duplicate log lines (4x) were caused by per-CPU lost sample records being logged separately. Now they are aggregated and logged periodically. Co-Authored-By: Claude Opus 4.5 --- ebpftracer/tracer.go | 38 ++++++++++++++++++++++++++++++++++++-- 1 file changed, 36 insertions(+), 2 deletions(-) diff --git a/ebpftracer/tracer.go b/ebpftracer/tracer.go index 160aecb8..a166b230 100644 --- a/ebpftracer/tracer.go +++ b/ebpftracer/tracer.go @@ -13,6 +13,8 @@ import ( "runtime" "strconv" "strings" + "sync" + "sync/atomic" "time" "github.com/cilium/ebpf" @@ -258,7 +260,7 @@ func (t *Tracer) ebpf(ch chan<- Event) error { } if !t.disableL7Tracing { - perfMaps = append(perfMaps, perfMap{name: "l7_events", typ: perfMapTypeL7Events, perCPUBufferSizePages: 64}) // Increased for high-volume LLM API tracing + perfMaps = append(perfMaps, perfMap{name: "l7_events", typ: perfMapTypeL7Events, perCPUBufferSizePages: 128}) // Increased for high-volume LLM API tracing } pageSize := os.Getpagesize() @@ -411,7 +413,39 @@ type httpResponseFragment struct { Data [2048]byte // Must match HTTP_FRAGMENT_SIZE in eBPF } +// lostSamplesTracker tracks lost samples per perf map and logs them periodically +type lostSamplesTracker struct { + count atomic.Uint64 + lastLog atomic.Int64 // Unix timestamp in seconds + interval int64 // Log interval in seconds +} + +var lostSamplesTrackers = sync.Map{} // map[string]*lostSamplesTracker + +func getLostSamplesTracker(name string) *lostSamplesTracker { + tracker, ok := lostSamplesTrackers.Load(name) + if !ok { + tracker, _ = lostSamplesTrackers.LoadOrStore(name, &lostSamplesTracker{interval: 10}) + } + return tracker.(*lostSamplesTracker) +} + +func (t *lostSamplesTracker) recordLostSamples(name string, count uint64, cpu int) { + t.count.Add(count) + now := time.Now().Unix() + lastLog := t.lastLog.Load() + if now-lastLog >= t.interval { + if t.lastLog.CompareAndSwap(lastLog, now) { + total := t.count.Swap(0) + if total > 0 { + klog.Errorf("%s lost %d samples total (last on CPU %d) in the last %d seconds", name, total, cpu, t.interval) + } + } + } +} + func runEventsReader(name string, r *perf.Reader, ch chan<- Event, typ perfMapType, readTimeout time.Duration) { + tracker := getLostSamplesTracker(name) if readTimeout == 0 { readTimeout = 100 * time.Millisecond } @@ -425,7 +459,7 @@ func runEventsReader(name string, r *perf.Reader, ch chan<- Event, typ perfMapTy continue } if rec.LostSamples > 0 { - klog.Errorln(name, "lost samples:", rec.LostSamples) + tracker.recordLostSamples(name, rec.LostSamples, rec.CPU) continue } var event Event From f3d17c2278e68fa72e97d75bb123a94239d1b3d9 Mon Sep 17 00:00:00 2001 From: mayankpande88 Date: Sat, 10 Jan 2026 18:10:13 +0530 Subject: [PATCH 4/8] fix: prevent duplicate klog output Initialize klog flags properly to prevent duplicate log output. The issue was that klog was outputting to both the custom writer and stderr by default. Setting logtostderr=false, alsologtostderr=false, and stderrthreshold=FATAL ensures only the custom output is used. Co-Authored-By: Claude Opus 4.5 --- main.go | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/main.go b/main.go index b1fa3bbf..1a538b7c 100644 --- a/main.go +++ b/main.go @@ -2,6 +2,7 @@ package main import ( "bytes" + "flag" "log" "net/http" _ "net/http/pprof" @@ -123,7 +124,11 @@ func getClientSet() (*kubernetes.Clientset, error) { } func main() { - klog.LogToStderr(false) + // Initialize klog flags to prevent duplicate output + klog.InitFlags(nil) + flag.Set("logtostderr", "false") + flag.Set("alsologtostderr", "false") + flag.Set("stderrthreshold", "FATAL") klog.SetOutput(&RateLimitedLogOutput{limiter: rate.NewLimiter(rate.Limit(*flags.LogPerSecond), *flags.LogBurst)}) klog.Infoln("agent version:", version) From c93a2890e311754466c620d2576c821f158befe6 Mon Sep 17 00:00:00 2001 From: mayankpande88 Date: Sat, 10 Jan 2026 21:04:43 +0530 Subject: [PATCH 5/8] fix: use klog.ErrorS to prevent duplicate log output Replace klog.Errorf with klog.ErrorS for lost samples logging. klog.Errorf logs to ERROR, WARNING, and INFO levels causing 3x duplicates. klog.ErrorS uses structured logging and only logs once at the ERROR level. Co-Authored-By: Claude Sonnet 4.5 --- ebpftracer/tracer.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/ebpftracer/tracer.go b/ebpftracer/tracer.go index a166b230..c1703a16 100644 --- a/ebpftracer/tracer.go +++ b/ebpftracer/tracer.go @@ -438,7 +438,8 @@ func (t *lostSamplesTracker) recordLostSamples(name string, count uint64, cpu in if t.lastLog.CompareAndSwap(lastLog, now) { total := t.count.Swap(0) if total > 0 { - klog.Errorf("%s lost %d samples total (last on CPU %d) in the last %d seconds", name, total, cpu, t.interval) + // Use ErrorS for structured logging that only logs once at ERROR level + klog.ErrorS(nil, "lost samples", "map", name, "total", total, "cpu", cpu, "interval_seconds", t.interval) } } } From dcecabd7dbcfa590fa9f5162802b3c758f10faed Mon Sep 17 00:00:00 2001 From: mayankpande88 Date: Sat, 10 Jan 2026 21:36:43 +0530 Subject: [PATCH 6/8] fix: use standard log.Printf to prevent duplicate output Replace klog.ErrorS with log.Printf for lost samples logging. Both klog.Errorf and klog.ErrorS output to multiple severity levels (ERROR, WARNING, INFO) by design, causing 3x duplicates. The standard log package writes once to the configured output (our RateLimitedLogOutput) without duplication. Co-Authored-By: Claude Sonnet 4.5 --- ebpftracer/tracer.go | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/ebpftracer/tracer.go b/ebpftracer/tracer.go index c1703a16..1277aee5 100644 --- a/ebpftracer/tracer.go +++ b/ebpftracer/tracer.go @@ -8,6 +8,7 @@ import ( "errors" "fmt" "io" + "log" "os" "path" "runtime" @@ -438,8 +439,8 @@ func (t *lostSamplesTracker) recordLostSamples(name string, count uint64, cpu in if t.lastLog.CompareAndSwap(lastLog, now) { total := t.count.Swap(0) if total > 0 { - // Use ErrorS for structured logging that only logs once at ERROR level - klog.ErrorS(nil, "lost samples", "map", name, "total", total, "cpu", cpu, "interval_seconds", t.interval) + // Use standard log package to avoid klog's multi-severity output + log.Printf("ERROR: %s lost %d samples total (last on CPU %d) in the last %d seconds", name, total, cpu, t.interval) } } } From 2c1d88f2dd98c73e010ecc9ceea4e5dba4d1991d Mon Sep 17 00:00:00 2001 From: mayankpande88 Date: Sun, 11 Jan 2026 11:37:34 +0530 Subject: [PATCH 7/8] fix: protect activeConnections map iteration in Collect Add read lock when iterating over activeConnections map in the Collect function to prevent concurrent map iteration and map write errors during Prometheus metric collection. The race condition occurs when: - Prometheus calls Collect() which iterates over activeConnections - Concurrently, connection events modify activeConnections Error: fatal error: concurrent map iteration and map write at containers/container.go:455 Co-Authored-By: Claude Sonnet 4.5 --- containers/container.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/containers/container.go b/containers/container.go index 30a5def6..ea5640d1 100644 --- a/containers/container.go +++ b/containers/container.go @@ -452,12 +452,14 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) { phaseStart = time.Now() connections := map[common.DestinationKey]int{} + c.lock.RLock() for _, conn := range c.activeConnections { if !conn.Closed.IsZero() { continue } connections[conn.DestinationKey]++ } + c.lock.RUnlock() for d, count := range connections { actualDestWorkload := d.GetActualDestinationWorkload() destWorkload := d.GetDestinationWorkload() From 19c22a1dd3b1efe843a85fcebefe5ad5cc8daf03 Mon Sep 17 00:00:00 2001 From: mayankpande88 Date: Sun, 11 Jan 2026 16:01:15 +0530 Subject: [PATCH 8/8] fix: add panic recovery to ClickHouse parser Prevent agent crashes when receiving malformed or incomplete ClickHouse protocol packets by adding defer/recover pattern to ParseClickhouse. Co-Authored-By: Claude Sonnet 4.5 --- ebpftracer/l7/clickhouse.go | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/ebpftracer/l7/clickhouse.go b/ebpftracer/l7/clickhouse.go index cab236c8..a31a25fc 100644 --- a/ebpftracer/l7/clickhouse.go +++ b/ebpftracer/l7/clickhouse.go @@ -6,7 +6,14 @@ import ( "github.com/ClickHouse/ch-go/proto" ) -func ParseClickhouse(payload []byte) string { +func ParseClickhouse(payload []byte) (result string) { + // Recover from panics caused by malformed/incomplete packets + defer func() { + if r := recover(); r != nil { + result = "" + } + }() + r := proto.NewReader(bytes.NewReader(payload)) var err error if _, err = r.Byte(); err != nil {