Merge pull request #152 from parca-dev/rtld

Use rtld map_complete usdt probe to trigger process sync

Author: gnurizen

go.mod

@@ -8,6 +8,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/service/s3 v1.87.0
 	github.com/cespare/xxhash/v2 v2.3.0
 	github.com/cilium/ebpf v0.19.0
+	github.com/coreos/pkg v0.0.0-20240122114842-bbd7aa9bf6fb
 	github.com/docker/go-connections v0.5.0
 	github.com/elastic/go-freelru v0.16.0
 	github.com/elastic/go-perf v0.0.0-20241029065020-30bec95324b8

go.sum

@@ -59,6 +59,8 @@ github.com/containerd/platforms v0.2.1 h1:zvwtM3rz2YHPQsF2CHYM8+KtB5dvhISiXh5ZpS
 github.com/containerd/platforms v0.2.1/go.mod h1:XHCb+2/hzowdiut9rkudds9bE5yJ7npe7dG/wG+uFPw=
 github.com/coreos/go-systemd/v22 v22.5.0 h1:RrqgGjYQKalulkV8NGVIfkXQf6YYmOyiJKk8iXXhfZs=
 github.com/coreos/go-systemd/v22 v22.5.0/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc=
+github.com/coreos/pkg v0.0.0-20240122114842-bbd7aa9bf6fb h1:GIzvVQ9UkUlOhSDlqmrQAAAUd6R3E+caIisNEyWXvNE=
+github.com/coreos/pkg v0.0.0-20240122114842-bbd7aa9bf6fb/go.mod h1:E3G3o1h8I7cfcXa63jLwjI0eiQQMgzzUDFVpN/nH/eA=
 github.com/cpuguy83/dockercfg v0.3.2 h1:DlJTyZGBDlXqUZ2Dk2Q3xHs/FtnooJJVaad2S9GKorA=
 github.com/cpuguy83/dockercfg v0.3.2/go.mod h1:sugsbF4//dDlL/i+S+rtpIWp+5h0BHJHfjj5/jFyUJc=
 github.com/creack/pty v1.1.18 h1:n56/Zwd5o6whRC5PMGretI4IdRLlmBXYNjScPaBgsbY=

internal/controller/config.go

@@ -11,6 +11,7 @@ import (
 
 	"go.opentelemetry.io/ebpf-profiler/reporter"
 	"go.opentelemetry.io/ebpf-profiler/tracer"
+	"go.opentelemetry.io/ebpf-profiler/util"
 )
 
 type Config struct {
@@ -95,7 +96,7 @@ func (cfg *Config) Validate() error {
 	}
 
 	if !cfg.NoKernelVersionCheck {
-		major, minor, patch, err := tracer.GetCurrentKernelVersion()
+		major, minor, patch, err := util.GetCurrentKernelVersion()
 		if err != nil {
 			return fmt.Errorf("failed to get kernel version: %v", err)
 		}

interpreter/luajit/mappings_test.go

@@ -15,10 +15,12 @@ import (
 	"testing"
 	"unsafe"
 
+	"github.com/cilium/ebpf/link"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/ebpf-profiler/host"
 	"go.opentelemetry.io/ebpf-profiler/interpreter"
 	"go.opentelemetry.io/ebpf-profiler/libpf"
+	"go.opentelemetry.io/ebpf-profiler/libpf/pfelf"
 	"go.opentelemetry.io/ebpf-profiler/lpm"
 	"go.opentelemetry.io/ebpf-profiler/process"
 	"go.opentelemetry.io/ebpf-profiler/util"
@@ -65,6 +67,11 @@ func (m *ebpfMapsMockup) DeleteProcData(libpf.InterpreterType, libpf.PID) error
 	return nil
 }
 
+func (mockup *ebpfMapsMockup) AttachUSDTProbe(_ libpf.PID, _ string, _ pfelf.USDTProbe,
+	_ string) (link.Link, error) {
+	return nil, nil
+}
+
 // TestSynchronizeMappings tests that if a mapping is realloc'd we do the right thing.
 func TestSynchronizeMappings(t *testing.T) {
 	for _, tc := range []struct {

interpreter/rtld/rtld.go

@@ -0,0 +1,98 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package rtld // import "go.opentelemetry.io/ebpf-profiler/interpreter/rtld"
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/cilium/ebpf/link"
+	log "github.com/sirupsen/logrus"
+	"go.opentelemetry.io/ebpf-profiler/interpreter"
+	"go.opentelemetry.io/ebpf-profiler/libpf"
+	"go.opentelemetry.io/ebpf-profiler/libpf/pfelf"
+	"go.opentelemetry.io/ebpf-profiler/remotememory"
+)
+
+// data holds the Uprobe link to keep it in memory
+type data struct {
+	probe     pfelf.USDTProbe
+	path      string
+	probeLink link.Link
+}
+
+// instance represents a per-PID instance of the rtld interpreter
+type instance struct {
+	interpreter.InstanceStubs
+	link link.Link
+}
+
+// Loader detects if the ELF file contains the rtld:map_complete USDT probe
+func Loader(_ interpreter.EbpfHandler, info *interpreter.LoaderInfo) (interpreter.Data, error) {
+	// Check if this is ld.so by examining the filename
+	fileName := info.FileName()
+	if !strings.Contains(fileName, "ld-") && !strings.Contains(fileName, "ld.so") {
+		return nil, nil
+	}
+
+	ef, err := info.GetELF()
+	if err != nil {
+		return nil, err
+	}
+
+	// Look for .note.stapsdt section which contains USDT probes
+	sec := ef.Section(".note.stapsdt")
+	if sec == nil {
+		log.Debugf("No .note.stapsdt section found in %s", fileName)
+		return nil, nil
+	}
+
+	// Parse USDT probes from the section
+	probes, err := pfelf.ParseUSDTProbes(sec)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse USDT probes: %w", err)
+	}
+
+	// Look for rtld:map_complete probe
+	for _, probe := range probes {
+		if probe.Provider == "rtld" && probe.Name == "map_complete" {
+			log.Infof("Found rtld:map_complete USDT probe in %s at 0x%x",
+				fileName, probe.Location)
+			return &data{
+				path:  fileName,
+				probe: probe,
+			}, nil
+		}
+	}
+
+	return nil, nil
+}
+
+// Attach attaches the uprobe to the rtld:map_complete USDT probe
+func (d *data) Attach(ebpf interpreter.EbpfHandler, pid libpf.PID, _ libpf.Address,
+	_ remotememory.RemoteMemory) (interpreter.Instance, error) {
+	link, err := ebpf.AttachUSDTProbe(pid, d.path, d.probe, "usdt_rtld_map_complete")
+	if err != nil {
+		return nil, fmt.Errorf("failed to attach uprobe to rtld:map_complete usdt: %w", err)
+	}
+	log.Infof("Attached uprobe to rtld:map_complete usdt in PID %d", pid)
+	return &instance{link: link}, nil
+}
+
+// Detach removes the uprobe
+func (i *instance) Detach(_ interpreter.EbpfHandler, pid libpf.PID) error {
+	log.Debugf("[rtld] Detach called for PID %d", pid)
+	return nil
+}
+
+// Unload cleans up the uprobe link
+func (d *data) Unload(_ interpreter.EbpfHandler) {
+	if d.probeLink != nil {
+		if err := d.probeLink.Close(); err != nil {
+			log.Errorf("[rtld] Failed to close uprobe link: %v", err)
+		}
+		d.probeLink = nil
+	}
+	log.Debugf("[rtld] Unloaded uprobe for %s", d.path)
+}

interpreter/rtld/rtld_test.go

@@ -0,0 +1,105 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+//go:build amd64 && !integration
+
+package rtld_test
+
+import (
+	"context"
+	"testing"
+	"time"
+	"unsafe"
+
+	"github.com/coreos/pkg/dlopen"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/ebpf-profiler/metrics"
+	"go.opentelemetry.io/ebpf-profiler/support"
+	"go.opentelemetry.io/ebpf-profiler/testutils"
+	"go.opentelemetry.io/ebpf-profiler/tracer"
+	tracertypes "go.opentelemetry.io/ebpf-profiler/tracer/types"
+)
+
+func TestIntegration(t *testing.T) {
+	if !testutils.IsRoot() {
+		t.Skip("This test requires root privileges")
+	}
+
+	// Create a context for the tracer
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	// Start the tracer with all tracers enabled
+	traceCh, trc := testutils.StartTracer(ctx, t,
+		tracertypes.AllTracers(),
+		&testutils.MockReporter{},
+		false)
+	defer trc.Close()
+
+	// Consume traces to prevent blocking
+	go func() {
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-traceCh:
+				// Discard traces
+			}
+		}
+	}()
+
+	// retry a few times to get the metric, our process has to be detected and
+	// the rtld interpreter has to attach.
+	require.Eventually(t, func() bool {
+		// Get the initial metric value
+		initialCount := getEBPFMetricValue(trc, metrics.IDRtldMapCompleteHits)
+		t.Logf("Initial rtld:map_complete metric count: %d", initialCount)
+
+		// Use dlopen to load a shared library
+		// libm is a standard math library that's always present
+		lib, err := dlopen.GetHandle([]string{
+			"/lib/x86_64-linux-gnu/libm.so.6",
+			"libm.so.6",
+		})
+		require.NoError(t, err, "Failed to open libm.so.6")
+		defer lib.Close()
+
+		// Get the metrics after dlopen
+		finalCount := getEBPFMetricValue(trc, metrics.IDRtldMapCompleteHits)
+		t.Logf("Final rtld:map_complete metric count: %d", finalCount)
+
+		// Check that the metric was incremented
+		return finalCount > initialCount
+	}, 10*time.Second, 50*time.Millisecond)
+}
+
+func getEBPFMetricValue(trc *tracer.Tracer, metricID metrics.MetricID) uint64 {
+	// Access the eBPF maps directly using the public method
+	ebpfMaps := trc.GetEbpfMaps()
+	metricsMap, ok := ebpfMaps["metrics"]
+	if !ok {
+		return 0
+	}
+
+	// Find the eBPF metric ID that corresponds to our metrics.MetricID
+	var ebpfMetricID uint32
+	for ebpfID, id := range support.MetricsTranslation {
+		if id == metricID {
+			ebpfMetricID = uint32(ebpfID)
+			break
+		}
+	}
+
+	// Read the per-CPU values
+	var perCPUValues []uint64
+	if err := metricsMap.Lookup(unsafe.Pointer(&ebpfMetricID), &perCPUValues); err != nil {
+		return 0
+	}
+
+	// Sum all per-CPU values
+	var total uint64
+	for _, val := range perCPUValues {
+		total += val
+	}
+	return total
+}

interpreter/types.go

@@ -7,8 +7,10 @@ import (
 	"errors"
 	"unsafe"
 
+	"github.com/cilium/ebpf/link"
 	"go.opentelemetry.io/ebpf-profiler/host"
 	"go.opentelemetry.io/ebpf-profiler/libpf"
+	"go.opentelemetry.io/ebpf-profiler/libpf/pfelf"
 	"go.opentelemetry.io/ebpf-profiler/lpm"
 	"go.opentelemetry.io/ebpf-profiler/metrics"
 	"go.opentelemetry.io/ebpf-profiler/process"
@@ -112,6 +114,10 @@ type EbpfHandler interface {
 
 	// If unwinder needs special behavior for coredump mode to work use this.
 	CoredumpTest() bool
+
+	// AttachUSDTProbe attaches a uprobe to the given USDT probe in the given process.
+	AttachUSDTProbe(pid libpf.PID, path string, probe pfelf.USDTProbe,
+		progName string) (link.Link, error)
 }
 
 // Loader is a function to detect and load data from given interpreter ELF file.

libpf/pfelf/usdt.go

@@ -0,0 +1,89 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package pfelf // import "go.opentelemetry.io/ebpf-profiler/libpf/pfelf"
+
+import (
+	"encoding/binary"
+	"strings"
+)
+
+// USDTProbe represents a USDT probe found in ELF
+type USDTProbe struct {
+	Provider        string
+	Name            string
+	Location        uint64
+	Base            uint64
+	SemaphoreOffset uint64
+	Arguments       string
+}
+
+// ParseUSDTProbes reads USDT probe information from ELF .note.stapsdt section
+func ParseUSDTProbes(section *Section) ([]USDTProbe, error) {
+	var probes []USDTProbe
+
+	// Find .note.stapsdt section
+	data, err := section.Data(16 * 1024)
+	if err != nil {
+		return nil, err
+	}
+
+	// Parse note entries
+	offset := 0
+	for offset < len(data) {
+		if offset+12 > len(data) {
+			break
+		}
+
+		// Note header: namesz(4) + descsz(4) + type(4)
+		namesz := binary.LittleEndian.Uint32(data[offset : offset+4])
+		descsz := binary.LittleEndian.Uint32(data[offset+4 : offset+8])
+		noteType := binary.LittleEndian.Uint32(data[offset+8 : offset+12])
+		offset += 12
+
+		if noteType != 3 { // NT_STAPSDT
+			// Skip this note
+			nameEnd := offset + int((namesz+3)&^3) // align to 4 bytes
+			descEnd := nameEnd + int((descsz+3)&^3)
+			offset = descEnd
+			continue
+		}
+
+		// Skip owner name (should be "stapsdt")
+		nameEnd := offset + int((namesz+3)&^3)
+
+		if nameEnd+int(descsz) > len(data) {
+			break
+		}
+
+		// Parse descriptor
+		desc := data[nameEnd : nameEnd+int(descsz)]
+		if len(desc) < 24 { // 3 uint64 values
+			offset = nameEnd + int((descsz+3)&^3)
+			continue
+		}
+
+		location := binary.LittleEndian.Uint64(desc[0:8])
+		base := binary.LittleEndian.Uint64(desc[8:16])
+		semaphore := binary.LittleEndian.Uint64(desc[16:24])
+
+		// Parse strings: provider\0probe\0arguments\0
+		stringData := desc[24:]
+		strings := strings.Split(string(stringData), "\x00")
+		if len(strings) >= 3 {
+			probe := USDTProbe{
+				Provider:        strings[0],
+				Name:            strings[1],
+				Location:        location,
+				Base:            base,
+				SemaphoreOffset: semaphore,
+				Arguments:       strings[2],
+			}
+			probes = append(probes, probe)
+		}
+
+		offset = nameEnd + int((descsz+3)&^3)
+	}
+
+	return probes, nil
+}

metrics/ids.go

@@ -2211,5 +2211,12 @@
     "name": "UnwindNodeAsyncIdErrReadIdDouble",
     "field": "bpf.nodejs_async_id.errors.read_id_double",
     "id": 307
+  },
+  {
+    "description": "Number of times rtld:map_complete USDT probe was fired",
+    "type": "counter",
+    "name": "RtldMapCompleteHits",
+    "field": "bpf.rtld.map_complete_hits",
+    "id": 308
   }
 ]