Merge upstream coroot/coroot-node-agent changes

Dockerfile

@@ -1,5 +1,15 @@
-FROM golang:1.24.7-trixie AS builder
-RUN apt update && apt install -y libsystemd-dev
+FROM debian:bullseye AS builder
+# Using Debian instead of the official Golang image because it’s based on newer OS versions
+# with newer glibc, which causes compatibility issues.
+
+RUN apt-get update && apt-get install -y \
+    curl git build-essential pkg-config libsystemd-dev
+
+ARG GO_VERSION=1.24.9
+RUN curl -fsSL https://go.dev/dl/go${GO_VERSION}.linux-$(dpkg --print-architecture).tar.gz -o go.tar.gz && \
+    tar -C /usr/local -xzf go.tar.gz && rm go.tar.gz
+ENV PATH="/usr/local/go/bin:${PATH}"
+
 WORKDIR /tmp/src
 COPY go.mod .
 COPY go.sum .

cgroup/cgroup.go

@@ -23,7 +23,7 @@ var (
 	crioIdRegexp        = regexp.MustCompile(`crio-([a-z0-9]{64})`)
 	containerdIdRegexp  = regexp.MustCompile(`cri-containerd[-:]([a-z0-9]{64})`)
 	lxcIdRegexp         = regexp.MustCompile(`/lxc/([^/]+)`)
-	systemSliceIdRegexp = regexp.MustCompile(`(/(system|runtime|reserved)\.slice/([^/]+))`)
+	systemSliceIdRegexp = regexp.MustCompile(`(/(system|runtime|reserved|kube|azure)\.slice/([^/]+))`)
 	talosIdRegexp       = regexp.MustCompile(`/(system|podruntime)/([^/]+)`)
 	lxcPayloadRegexp    = regexp.MustCompile(`/lxc\.payload\.([^/]+)`)
 )
@@ -165,7 +165,7 @@ func containerByCgroup(cgroupPath string) (ContainerType, string, error) {
 	switch {
 	case cgroupPath == "/init":
 		return ContainerTypeTalosRuntime, "/talos/init", nil
-	case prefix == "user.slice" || prefix == "init.scope":
+	case prefix == "user.slice" || prefix == "init.scope" || prefix == "systemd":
 		return ContainerTypeStandaloneProcess, "", nil
 	case prefix == "docker" || (prefix == "system.slice" && len(parts) > 1 && strings.HasPrefix(parts[1], "docker-")):
 		matches := dockerIdRegexp.FindStringSubmatch(cgroupPath)
@@ -196,7 +196,10 @@ func containerByCgroup(cgroupPath string) (ContainerType, string, error) {
 			return ContainerTypeUnknown, "", fmt.Errorf("invalid talos runtime cgroup %s", cgroupPath)
 		}
 		return ContainerTypeTalosRuntime, path.Join("/talos/", matches[2]), nil
-	case prefix == "system.slice" || prefix == "runtime.slice" || prefix == "reserved.slice":
+	case prefix == "system.slice" || prefix == "runtime.slice" || prefix == "reserved.slice" || prefix == "kube.slice" || prefix == "azure.slice":
+		if strings.HasSuffix(cgroupPath, ".scope") {
+			return ContainerTypeStandaloneProcess, "", nil
+		}
 		matches := systemSliceIdRegexp.FindStringSubmatch(cgroupPath)
 		if matches == nil {
 			return ContainerTypeUnknown, "", fmt.Errorf("invalid systemd cgroup %s", cgroupPath)

cgroup/cgroup_test.go

@@ -161,6 +161,16 @@ func TestContainerByCgroup(t *testing.T) {
 	as.Equal("/reserved.slice/kubelet.service", id)
 	as.Nil(err)
 
+	typ, id, err = containerByCgroup("/kube.slice/kubelet.service")
+	as.Equal(typ, ContainerTypeSystemdService)
+	as.Equal("/kube.slice/kubelet.service", id)
+	as.Nil(err)
+
+	typ, id, err = containerByCgroup("/azure.slice/walinuxagent.service")
+	as.Equal(typ, ContainerTypeSystemdService)
+	as.Equal("/azure.slice/walinuxagent.service", id)
+	as.Nil(err)
+
 	typ, id, err = containerByCgroup("/system.slice/system-postgresql.slice/postgresql@9.4-main.service")
 	as.Equal(typ, ContainerTypeSystemdService)
 	as.Equal("/system.slice/system-postgresql.slice", id)
@@ -195,4 +205,24 @@ func TestContainerByCgroup(t *testing.T) {
 	as.Equal(ContainerTypeStandaloneProcess, typ)
 	as.Equal("", id)
 	as.Nil(err)
+
+	typ, id, err = containerByCgroup("/systemd/system.slice")
+	as.Equal(ContainerTypeStandaloneProcess, typ)
+	as.Equal("", id)
+	as.Nil(err)
+
+	typ, id, err = containerByCgroup("/system.slice/cri-containerd-69e8ded3c33c9d5e2b93acd74787b17a8629f74d6707bc5bb9b2e095337d0263.scope")
+	as.Equal(ContainerTypeStandaloneProcess, typ)
+	as.Equal("", id)
+	as.Nil(err)
+
+	typ, id, err = containerByCgroup("/system.slice/run-ra2ddf9594bbf4a1986439b594f89eb0f.scope")
+	as.Equal(ContainerTypeStandaloneProcess, typ)
+	as.Equal("", id)
+	as.Nil(err)
+
+	typ, id, err = containerByCgroup("/system.slice/docker-ba7b10d15d16e10e3de7a2dcd408a3d971169ae303f46cfad4c5453c6326fee2.scope")
+	as.Equal(ContainerTypeDocker, typ)
+	as.Equal("ba7b10d15d16e10e3de7a2dcd408a3d971169ae303f46cfad4c5453c6326fee2", id)
+	as.Nil(err)
 }

containers/container.go

@@ -50,16 +50,16 @@ type ContainerNetwork struct {
 }
 
 type ContainerMetadata struct {
-	name               string
-	labels             map[string]string
-	volumes            map[string]string
-	logPath            string
-	image              string
-	logDecoder         logparser.Decoder
-	hostListens        map[string][]netaddr.IPPort
-	networks           map[string]ContainerNetwork
-	env                map[string]string
-	systemdTriggeredBy string
+	name        string
+	labels      map[string]string
+	volumes     map[string]string
+	logPath     string
+	image       string
+	logDecoder  logparser.Decoder
+	hostListens map[string][]netaddr.IPPort
+	networks    map[string]ContainerNetwork
+	env         map[string]string
+	systemd     SystemdProperties
 }
 
 type Delays struct {
@@ -332,8 +332,8 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) {
 	}
 	atomic.StoreInt64(&c.lastCollectTime, nowNanos)
 
-	if c.metadata.image != "" || c.metadata.systemdTriggeredBy != "" {
-		ch <- gauge(metrics.ContainerInfo, 1, c.metadata.image, c.metadata.systemdTriggeredBy)
+	if c.metadata.image != "" || !c.metadata.systemd.IsEmpty() {
+		ch <- gauge(metrics.ContainerInfo, 1, c.metadata.image, c.metadata.systemd.TriggeredBy, c.metadata.systemd.Type)
 	}
 
 	ch <- counter(metrics.Restarts, float64(c.restarts))
@@ -1798,13 +1798,13 @@ func (c *Container) runLogParser(logPath string) {
 	switch c.cgroup.ContainerType {
 	case cgroup.ContainerTypeSystemdService:
 		ch := make(chan logparser.LogEntry)
-		if err := JournaldSubscribe(c.cgroup, ch); err != nil {
+		if err := JournaldSubscribe(c.metadata.systemd.Unit, ch); err != nil {
 			klog.Warningln(err)
 			return
 		}
 		parser := logparser.NewParser(ch, nil, logs.OtelLogEmitter(containerId), multilineCollectorTimeout, *flags.DisableSensitiveLogParsing)
 		stop := func() {
-			JournaldUnsubscribe(c.cgroup)
+			JournaldUnsubscribe(c.metadata.systemd.Unit)
 		}
 		klog.InfoS("started logparser for container", c.id)
 		klog.InfoS("started journald logparser", "cg", c.cgroup.Id)

containers/containerd.go

@@ -68,7 +68,7 @@ func ContainerdInspect(containerID string) (*ContainerMetadata, error) {
 	}
 
 	var spec oci.Spec
-	if err := json.Unmarshal(c.Spec.Value, &spec); err != nil {
+	if err := json.Unmarshal(c.Spec.GetValue(), &spec); err != nil {
 		klog.Warningln(err)
 	} else {
 		for _, m := range spec.Mounts {
@@ -82,7 +82,7 @@ func ContainerdInspect(containerID string) (*ContainerMetadata, error) {
 				LogPath string
 			}
 		}{}
-		if err := json.Unmarshal(data.Value, &md); err != nil {
+		if err := json.Unmarshal(data.GetValue(), &md); err != nil {
 			klog.Warningln(err)
 		} else {
 			res.logPath = md.Metadata.LogPath

containers/journald.go

@@ -3,7 +3,6 @@ package containers
 import (
 	"fmt"
 
-	"github.com/coroot/coroot-node-agent/cgroup"
 	"github.com/coroot/coroot-node-agent/logs"
 	"github.com/coroot/coroot-node-agent/proc"
 	"github.com/nudgebee/logparser"
@@ -25,20 +24,20 @@ func JournaldInit() error {
 	return nil
 }
 
-func JournaldSubscribe(cg *cgroup.Cgroup, ch chan<- logparser.LogEntry) error {
+func JournaldSubscribe(unit string, ch chan<- logparser.LogEntry) error {
 	if journaldReader == nil {
 		return fmt.Errorf("journald reader not initialized")
 	}
-	err := journaldReader.Subscribe(cg.Id, ch)
+	err := journaldReader.Subscribe(unit, ch)
 	if err != nil {
 		return err
 	}
 	return nil
 }
 
-func JournaldUnsubscribe(cg *cgroup.Cgroup) {
+func JournaldUnsubscribe(unit string) {
 	if journaldReader == nil {
 		return
 	}
-	journaldReader.Unsubscribe(cg.Id)
+	journaldReader.Unsubscribe(unit)
 }

containers/metrics.go

@@ -66,7 +66,7 @@ var metrics = struct {
 	LLMTokensUsed *prometheus.Desc
 	LLMLatency    *prometheus.Desc
 }{
-	ContainerInfo: metric("container_info", "Meta information about the container", "image", "systemd_triggered_by"),
+	ContainerInfo: metric("container_info", "Meta information about the container", "image", "systemd_triggered_by", "systemd_type"),
 
 	Restarts: metric("container_restarts_total", "Number of times the container was restarted"),
 

containers/process.go

@@ -113,7 +113,11 @@ func (p *Process) instrumentPython(cmdline []byte, tracer *ebpftracer.Tracer) {
 	if len(cmd) == 0 {
 		return
 	}
-	cmd = bytes.TrimSuffix(bytes.Fields(cmd)[0], []byte{':'})
+	cmdFields := bytes.Fields(cmd)
+	if len(cmdFields) == 0 {
+		return
+	}
+	cmd = bytes.TrimSuffix(cmdFields[0], []byte{':'})
 	if !pythonCmd.Match(cmd) {
 		return
 	}
@@ -172,7 +176,13 @@ func (p *Process) removeOldGpuUsageSamples(cutoff time.Time) {
 
 func (p *Process) Close() {
 	p.cancelFunc()
-	for _, u := range p.uprobes {
-		_ = u.Close()
+	if len(p.uprobes) > 0 {
+		uprobes := p.uprobes
+		p.uprobes = nil
+		go func() {
+			for _, u := range uprobes {
+				_ = u.Close()
+			}
+		}()
 	}
 }

containers/registry.go

@@ -357,8 +357,6 @@ func (r *Registry) handleEvents(ch <-chan ebpftracer.Event) {
 			case ebpftracer.EventTypeConnectionError:
 				if c := r.getOrCreateContainer(e.Pid); c != nil {
 					c.onConnectionOpen(e.Pid, e.Fd, e.SrcAddr, e.DstAddr, e.ActualDstAddr, 0, true, e.Duration)
-				} else {
-					klog.Infoln("TCP connection error from unknown container", e)
 				}
 			case ebpftracer.EventTypeConnectionClose:
 				r.containerLock.RLock()
@@ -540,6 +538,9 @@ func (r *Registry) getOrCreateContainer(pid uint32) *Container {
 		}
 		return nil
 	}
+	if strings.HasSuffix(cg.Id, "(deleted)") {
+		return nil
+	}
 
 	// Check if container exists by cgroup ID with read lock
 	r.containerLock.RLock()
@@ -593,6 +594,14 @@ func (r *Registry) getOrCreateContainer(pid uint32) *Container {
 		r.containerLock.Unlock()
 		return nil
 	}
+	if cg.ContainerType == cgroup.ContainerTypeSystemdService && *flags.SkipSystemdSystemServices {
+		if md.systemd.IsSystemService() {
+			klog.InfoS("skipping system service", "id", id, "unit", md.systemd.Unit, "type", md.systemd.Type, "triggered_by", md.systemd.TriggeredBy, "pid", pid)
+			t := time.Now()
+			r.containersByPidIgnored[pid] = &t
+			return nil
+		}
+	}
 
 	// Acquire write lock for container creation to prevent race conditions
 	writeLockStart := time.Now()
@@ -827,9 +836,10 @@ func calcId(cg *cgroup.Cgroup, md *ContainerMetadata) ContainerID {
 func getContainerMetadata(cg *cgroup.Cgroup) (*ContainerMetadata, error) {
 	switch cg.ContainerType {
 	case cgroup.ContainerTypeSystemdService:
+		var err error
 		md := &ContainerMetadata{}
-		md.systemdTriggeredBy = SystemdTriggeredBy(cg.ContainerId)
-		return md, nil
+		md.systemd, err = getSystemdProperties(cg.Id)
+		return md, err
 	case cgroup.ContainerTypeDocker, cgroup.ContainerTypeContainerd, cgroup.ContainerTypeSandbox, cgroup.ContainerTypeCrio:
 	default:
 		return &ContainerMetadata{}, nil