chore: rebase with upstream

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\.([^/]+)`)
 )
@@ -196,7 +196,7 @@ 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":
 		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)

containers/container.go

@@ -1351,7 +1351,7 @@
 			return
 		}
 		ch := make(chan logparser.LogEntry)
-		parser := logparser.NewParser(ch, nil, logs.OtelLogEmitter(containerId), multilineCollectorTimeout, *flags.DisableSensitiveLogParsing)
+		parser := logparser.NewParser(ch, nil, logs.OtelLogEmitter(containerId), multilineCollectorTimeout, *flags.LogPatternsPerContainer, false)
 		reader, err := logs.NewTailReader(proc.HostPath(logPath), ch)
 		if err != nil {
 			klog.Warningln(err)
@@ -1371,7 +1371,7 @@
 			klog.Warningln(err)
 			return
 		}
-		parser := logparser.NewParser(ch, nil, logs.OtelLogEmitter(containerId), multilineCollectorTimeout, *flags.DisableSensitiveLogParsing)
+		parser := logparser.NewParser(ch, nil, logs.OtelLogEmitter(containerId), multilineCollectorTimeout, *flags.LogPatternsPerContainer, false)
 		stop := func() {
 			JournaldUnsubscribe(c.cgroup)
 		}
@@ -1388,14 +1388,14 @@
 			delete(c.logParsers, "stdout/stderr")
 		}
 		ch := make(chan logparser.LogEntry)
-		parser := logparser.NewParser(ch, c.metadata.logDecoder, logs.OtelLogEmitter(containerId), multilineCollectorTimeout, *flags.DisableSensitiveLogParsing)
+		parser := logparser.NewParser(ch, c.metadata.logDecoder, logs.OtelLogEmitter(containerId), multilineCollectorTimeout, *flags.LogPatternsPerContainer, false)
 		reader, err := logs.NewTailReader(proc.HostPath(c.metadata.logPath), ch)
 		if err != nil {
 			klog.Warningln(err)
 			parser.Stop()
 			return
 		}
 		klog.InfoS("started logparser for container", c.id, "log", c.metadata.logPath)
 		klog.InfoS("started container logparser", "cg", c.cgroup.Id)
 		c.logParsers["stdout/stderr"] = &LogParser{parser: parser, stop: reader.Stop}
 	}
@@ -1415,7 +1415,7 @@
 		sockets, err := proc.GetSockets(p.Pid)
 		if err != nil {
 			continue
 		}
 		for _, s := range sockets {
 			if s.Listen {
 				listens[s.SAddr] = s.Inode
@@ -1432,7 +1432,7 @@
 	for k, conn := range c.activeConnections {
 		pidFd := PidFd{Pid: conn.Pid, Fd: conn.Fd}
 		if _, ok := established[k]; !ok {
 			delete(c.activeConnections, k)
 			if conn == c.connectionsByPidFd[pidFd] {
 				delete(c.connectionsByPidFd, pidFd)
 			}

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
 	}

ebpftracer/elf.go

@@ -0,0 +1,169 @@
+package ebpftracer
+
+import (
+	"debug/elf"
+	"fmt"
+	"io"
+
+	"github.com/cilium/ebpf"
+	"github.com/cilium/ebpf/link"
+	"golang.org/x/arch/arm64/arm64asm"
+	"golang.org/x/arch/x86/x86asm"
+)
+
+type Symbol struct {
+	s       *elf.Symbol
+	f       *ELFFile
+	address uint64
+}
+
+func (s *Symbol) Name() string {
+	return s.s.Name
+}
+
+func (s *Symbol) Address() uint64 {
+	if s.address == 0 {
+		s.address = s.s.Value
+		for _, p := range s.f.elf.Progs {
+			if p.Type != elf.PT_LOAD || (p.Flags&elf.PF_X) == 0 {
+				continue
+			}
+			if p.Vaddr <= s.s.Value && s.s.Value < (p.Vaddr+p.Memsz) {
+				s.address = s.s.Value - p.Vaddr + p.Off
+				break
+			}
+		}
+	}
+	return s.address
+}
+
+func (s *Symbol) ReturnOffsets() ([]int, error) {
+	text, reader, err := s.f.getTextSectionAndReader()
+	if err != nil {
+		return nil, err
+	}
+
+	sStart := s.s.Value - text.Addr
+	_, err = reader.Seek(int64(sStart), io.SeekStart)
+	if err != nil {
+		return nil, err
+	}
+	sBytes := make([]byte, s.s.Size)
+	_, err = reader.Read(sBytes)
+	if err != nil {
+		return nil, err
+	}
+
+	offsets := getReturnOffsets(s.f.elf.Machine, sBytes)
+	if len(offsets) == 0 {
+		return nil, fmt.Errorf("no offsets found")
+	}
+	return offsets, nil
+}
+
+func (s *Symbol) AttachUprobe(exe *link.Executable, prog *ebpf.Program, pid uint32) (link.Link, error) {
+	return exe.Uprobe(s.Name(), prog, &link.UprobeOptions{Address: s.Address(), PID: int(pid)})
+}
+
+func (s *Symbol) AttachUretprobes(exe *link.Executable, prog *ebpf.Program, pid uint32) ([]link.Link, error) {
+	returnOffsets, err := s.ReturnOffsets()
+	if err != nil {
+		return nil, err
+	}
+	var links []link.Link
+	for _, offset := range returnOffsets {
+		l, err := exe.Uprobe("pthread_cond_timedwait", prog, &link.UprobeOptions{Address: s.Address(), Offset: uint64(offset), PID: int(pid)})
+		if err != nil {
+			return links, err
+		}
+		links = append(links, l)
+	}
+
+	return links, nil
+}
+
+type ELFFile struct {
+	elf               *elf.File
+	symbols           []elf.Symbol
+	textSection       *elf.Section
+	textSectionReader io.ReadSeeker
+}
+
+func OpenELFFile(path string) (*ELFFile, error) {
+	file, err := elf.Open(path)
+	if err != nil {
+		return nil, err
+	}
+	return &ELFFile{elf: file}, nil
+}
+
+func (f *ELFFile) readSymbols() error {
+	symbols, _ := f.elf.Symbols()
+	dyn, _ := f.elf.DynamicSymbols()
+
+	if len(symbols) == 0 && len(dyn) == 0 {
+		return fmt.Errorf("no symbols found")
+	}
+	f.symbols = append(symbols, dyn...)
+	return nil
+}
+
+func (f *ELFFile) GetSymbol(name string) (*Symbol, error) {
+	if f.symbols == nil {
+		if err := f.readSymbols(); err != nil {
+			return nil, err
+		}
+	}
+	var es *elf.Symbol
+	for _, s := range f.symbols {
+		if elf.ST_TYPE(s.Info) != elf.STT_FUNC || s.Size == 0 || s.Value == 0 {
+			continue
+		}
+		if s.Name == name && s.VersionIndex&0x8000 == 0 {
+			es = &s
+			break
+		}
+	}
+	if es == nil {
+		return nil, fmt.Errorf("symbol %s not found", name)
+	}
+	return &Symbol{s: es, f: f}, nil
+}
+
+func (f *ELFFile) getTextSectionAndReader() (*elf.Section, io.ReadSeeker, error) {
+	if f.textSection == nil {
+		f.textSection = f.elf.Section(".text")
+		if f.textSection == nil {
+			return nil, nil, fmt.Errorf("no .text")
+		}
+		f.textSectionReader = f.textSection.Open()
+	}
+	return f.textSection, f.textSectionReader, nil
+}
+
+func (f *ELFFile) Close() error {
+	return f.elf.Close()
+}
+
+func getReturnOffsets(machine elf.Machine, instructions []byte) []int {
+	var res []int
+	switch machine {
+	case elf.EM_X86_64:
+		for i := 0; i < len(instructions); {
+			ins, err := x86asm.Decode(instructions[i:], 64)
+			if err == nil && ins.Op == x86asm.RET {
+				res = append(res, i)
+			}
+			i += ins.Len
+		}
+	case elf.EM_AARCH64:
+		for i := 0; i < len(instructions); {
+			ins, err := arm64asm.Decode(instructions[i:])
+			if err == nil && ins.Op == arm64asm.RET {
+				res = append(res, i)
+			}
+			i += 4
+		}
+	}
+	return res
+}

ebpftracer/nodejs.go

@@ -25,55 +25,65 @@ func (t *Tracer) AttachNodejsProbes(pid uint32, exe string) []link.Link {
 		klog.InfofDepth(1, "pid=%d lib=%s: %s", pid, libPath, msg)
 	}
 
-	var (
-		lastErr error
-		links   []link.Link
-		libPath string
-	)
-	for _, libPath = range append(getLibuv(pid), proc.Path(pid, "root", exe)) {
-		exe, err := link.OpenExecutable(libPath)
-		if err != nil {
-			log(libPath, "failed to open executable", err)
-			return nil
-		}
-		options := &link.UprobeOptions{PID: int(pid)}
-		var uprobe, uretprobe link.Link
-		uprobe, lastErr = exe.Uprobe("uv__io_poll", t.uprobes["uv_io_poll_enter"], options)
-		if lastErr != nil {
-			continue
+	for _, libPath := range append(getLibuv(pid), proc.Path(pid, "root", exe)) {
+		if links, err := t.attachNodejsUprobes(libPath, pid); err == nil {
+			log(libPath, "nodejs uprobes attached", nil)
+			return links
+		} else {
+			log(libPath, "failed to attach nodejs uprobes", err)
 		}
+	}
+	return nil
+}
 
-		links = append(links, uprobe)
-		uretprobe, lastErr = exe.Uretprobe("uv__io_poll", t.uprobes["uv_io_poll_exit"], options)
-		if lastErr != nil {
-			continue
-		}
+func (t *Tracer) attachNodejsUprobes(libPath string, pid uint32) ([]link.Link, error) {
+	exe, err := link.OpenExecutable(libPath)
+	if err != nil {
+		return nil, err
+	}
+	ef, err := OpenELFFile(libPath)
+	if err != nil {
+		return nil, err
+	}
+	defer ef.Close()
+
+	s, err := ef.GetSymbol("uv__io_poll")
+	if err != nil {
+		return nil, err
+	}
+	l, err := s.AttachUprobe(exe, t.uprobes["uv_io_poll_enter"], pid)
+	if err != nil {
+		return nil, err
+	}
+	var links []link.Link
+	links = append(links, l)
 
-		links = append(links, uretprobe)
+	ls, err := s.AttachUretprobes(exe, t.uprobes["uv_io_poll_exit"], pid)
+	links = append(links, ls...)
+	if err != nil {
+		for _, l := range links {
+			_ = l.Close()
+		}
+		return nil, err
+	}
 
-		for _, cb := range []string{"uv__stream_io", "uv__async_io", "uv__poll_io", "uv__server_io", "uv__udp_io"} {
-			uprobe, lastErr = exe.Uprobe(cb, t.uprobes["uv_io_cb_enter"], options)
-			if lastErr != nil {
-				break
-			}
-			links = append(links, uprobe)
-			uretprobe, lastErr = exe.Uretprobe(cb, t.uprobes["uv_io_cb_exit"], options)
-			if lastErr != nil {
-				break
-			}
-			links = append(links, uretprobe)
+	for _, cb := range []string{"uv__stream_io", "uv__async_io", "uv__poll_io", "uv__server_io", "uv__udp_io"} {
+		s, err = ef.GetSymbol(cb)
+		if err != nil {
+			break
 		}
-		if lastErr != nil {
-			continue
+		l, err = s.AttachUprobe(exe, t.uprobes["uv_io_cb_enter"], pid)
+		if err != nil {
+			break
+		}
+		links = append(links, l)
+		ls, err = s.AttachUretprobes(exe, t.uprobes["uv_io_cb_exit"], pid)
+		links = append(links, ls...)
+		if err != nil {
+			break
 		}
-
-		log(libPath, "nodejs uprobes attached", nil)
-		break
-	}
-	if lastErr != nil {
-		log(libPath, "failed to attach uprobe", lastErr)
 	}
-	return links
+	return links, nil
 }
 
 func getLibuv(pid uint32) []string {