fix: fix for unexpected token error

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

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)

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 {

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

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)