diff --git a/containers/container.go b/containers/container.go index 25508fb1..ea5640d1 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") + + // 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) } - 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) cid := string(id) appId := common.ContainerIdToOtelServiceName(cid) @@ -436,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() @@ -1123,18 +1141,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) { 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) 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 { diff --git a/ebpftracer/tracer.go b/ebpftracer/tracer.go index 160aecb8..1277aee5 100644 --- a/ebpftracer/tracer.go +++ b/ebpftracer/tracer.go @@ -8,11 +8,14 @@ import ( "errors" "fmt" "io" + "log" "os" "path" "runtime" "strconv" "strings" + "sync" + "sync/atomic" "time" "github.com/cilium/ebpf" @@ -258,7 +261,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 +414,40 @@ 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 { + // 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) + } + } + } +} + 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 +461,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 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)