[usm] add regular and raw tracepoints /sched_process_exit

pkg/collector/corechecks/ebpf/probe/ebpfcheck/probe.go

@@ -965,15 +965,3 @@ func hashMapNumberOfEntriesWithHelper(mp *ebpf.Map, mapid ebpf.MapID, mphCache *
 
 	return int64(res), nil
 }
-
-// HashMapNumberOfEntries returns the number of entries in the map
-func HashMapNumberOfEntries(mp *ebpf.Map) (int64, error) {
-	if isPerCPU(mp.Type()) {
-		return -1, fmt.Errorf("unsupported map type: %s", mp.String())
-	}
-	buffers := entryCountBuffers{
-		keysBufferSizeLimit:   0, // No limit
-		valuesBufferSizeLimit: 0, // No limit
-	}
-	return hashMapNumberOfEntriesWithIteration(mp, &buffers, 1)
-}

pkg/network/ebpf/c/protocols/flush.h

@@ -28,9 +28,7 @@ int tracepoint__net__netif_receive_skb(void *ctx) {
     return 0;
 }
 
-SEC("tracepoint/sched/sched_process_exit")
-int tracepoint__sched__sched_process_exit(void *ctx) {
-    CHECK_BPF_PROGRAM_BYPASSED()
+static __always_inline void delete_pid_in_maps() {
     u64 pid_tgid = bpf_get_current_pid_tgid();
 
     bpf_map_delete_elem(&ssl_read_args, &pid_tgid);
@@ -39,22 +37,19 @@ int tracepoint__sched__sched_process_exit(void *ctx) {
     bpf_map_delete_elem(&ssl_write_ex_args, &pid_tgid);
     bpf_map_delete_elem(&ssl_ctx_by_pid_tgid, &pid_tgid);
     bpf_map_delete_elem(&bio_new_socket_args, &pid_tgid);
+}
 
+SEC("tracepoint/sched/sched_process_exit")
+int tracepoint__sched__sched_process_exit(void *ctx) {
+    CHECK_BPF_PROGRAM_BYPASSED()
+    delete_pid_in_maps();
     return 0;
 }
 
 SEC("raw_tracepoint/sched_process_exit")
 int raw_tracepoint__sched_process_exit(void *ctx) {
     CHECK_BPF_PROGRAM_BYPASSED()
-    u64 pid_tgid = bpf_get_current_pid_tgid();
-
-    bpf_map_delete_elem(&ssl_read_args, &pid_tgid);
-    bpf_map_delete_elem(&ssl_read_ex_args, &pid_tgid);
-    bpf_map_delete_elem(&ssl_write_args, &pid_tgid);
-    bpf_map_delete_elem(&ssl_write_ex_args, &pid_tgid);
-    bpf_map_delete_elem(&ssl_ctx_by_pid_tgid, &pid_tgid);
-    bpf_map_delete_elem(&bio_new_socket_args, &pid_tgid);
-
+    delete_pid_in_maps();
     return 0;
 }
 

pkg/network/usm/ebpf_main.go

@@ -34,7 +34,6 @@ import (
 	"github.com/DataDog/datadog-agent/pkg/network/tracer/offsetguess"
 	"github.com/DataDog/datadog-agent/pkg/network/usm/buildmode"
 	"github.com/DataDog/datadog-agent/pkg/network/usm/utils"
-	"github.com/DataDog/datadog-agent/pkg/util/kernel"
 	"github.com/DataDog/datadog-agent/pkg/util/log"
 	manager "github.com/DataDog/ebpf-manager"
 )
@@ -150,29 +149,6 @@ func newEBPFProgram(c *config.Config, connectionProtocolMap *ebpf.Map) (*ebpfPro
 		}
 	}
 
-	if rawTracepointsSupported() {
-		// use a raw tracepoint on a supported kernel to intercept terminated threads and clear the corresponding maps.
-		mgr.Probes = append(mgr.Probes, []*manager.Probe{
-			{
-				ProbeIdentificationPair: manager.ProbeIdentificationPair{
-					EBPFFuncName: "raw_tracepoint__sched_process_exit",
-					UID:          probeUID,
-				},
-				TracepointName: "sched_process_exit",
-			},
-		}...)
-	} else {
-		// use a regular tracepoint to intercept terminated threads.
-		mgr.Probes = append(mgr.Probes, []*manager.Probe{
-			{
-				ProbeIdentificationPair: manager.ProbeIdentificationPair{
-					EBPFFuncName: "tracepoint__sched__sched_process_exit",
-					UID:          probeUID,
-				},
-			},
-		}...)
-	}
-
 	program := &ebpfProgram{
 		Manager:               ddebpf.NewManager(mgr, "usm", &ebpftelemetry.ErrorsTelemetryModifier{}),
 		cfg:                   c,
@@ -486,14 +462,6 @@ func (e *ebpfProgram) init(buf bytecode.AssetReader, options manager.Options) er
 		}
 	}
 
-	if rawTracepointsSupported() {
-		// exclude regular tracepoint if kernel supports raw tracepoint
-		options.ExcludedFunctions = append(options.ExcludedFunctions, "tracepoint__sched__sched_process_exit")
-	} else {
-		//exclude a raw tracepoint if kernel does not support it.
-		options.ExcludedFunctions = append(options.ExcludedFunctions, "raw_tracepoint__sched_process_exit")
-	}
-
 	err := e.InitWithOptions(buf, &options)
 	if err != nil {
 		cleanup()
@@ -506,11 +474,6 @@ func (e *ebpfProgram) init(buf bytecode.AssetReader, options manager.Options) er
 	return err
 }
 
-func rawTracepointsSupported() bool {
-	kversion, err := kernel.HostVersion()
-	return err == nil && kversion >= kernel.VersionCode(4, 17, 0)
-}
-
 func getAssetName(module string, debug bool) string {
 	if debug {
 		return fmt.Sprintf("%s-debug.o", module)

pkg/network/usm/ebpf_ssl.go

@@ -9,6 +9,7 @@ package usm
 
 import (
 	"bytes"
+	"errors"
 	"fmt"
 	"io"
 	"os"
@@ -20,8 +21,8 @@ import (
 	"time"
 	"unsafe"
 
-	manager "github.com/DataDog/ebpf-manager"
 	"github.com/cilium/ebpf"
+	"github.com/cilium/ebpf/features"
 	"github.com/davecgh/go-spew/spew"
 
 	ddebpf "github.com/DataDog/datadog-agent/pkg/ebpf"
@@ -38,6 +39,7 @@ import (
 	"github.com/DataDog/datadog-agent/pkg/util/log"
 	"github.com/DataDog/datadog-agent/pkg/util/safeelf"
 	ddsync "github.com/DataDog/datadog-agent/pkg/util/sync"
+	manager "github.com/DataDog/ebpf-manager"
 )
 
 const (
@@ -69,6 +71,9 @@ const (
 	gnutlsRecordSendRetprobe    = "uretprobe__gnutls_record_send"
 	gnutlsByeProbe              = "uprobe__gnutls_bye"
 	gnutlsDeinitProbe           = "uprobe__gnutls_deinit"
+
+	rawTracepointSchedProcessExit = "raw_tracepoint__sched_process_exit"
+	oldTracepointSchedProcessExit = "tracepoint__sched__sched_process_exit"
 )
 
 var openSSLProbes = []manager.ProbesSelector{
@@ -416,6 +421,16 @@ var opensslSpec = &protocols.ProtocolSpec{
 				EBPFFuncName: gnutlsDeinitProbe,
 			},
 		},
+		{
+			ProbeIdentificationPair: manager.ProbeIdentificationPair{
+				EBPFFuncName: rawTracepointSchedProcessExit,
+			},
+		},
+		{
+			ProbeIdentificationPair: manager.ProbeIdentificationPair{
+				EBPFFuncName: oldTracepointSchedProcessExit,
+			},
+		},
 	},
 }
 
@@ -424,6 +439,7 @@ type sslProgram struct {
 	watcher       *sharedlibraries.Watcher
 	istioMonitor  *istioMonitor
 	nodeJSMonitor *nodeJSMonitor
+	ebpfManager   *manager.Manager
 }
 
 func newSSLProgramProtocolFactory(m *manager.Manager) protocols.ProtocolFactory {
@@ -477,6 +493,7 @@ func newSSLProgramProtocolFactory(m *manager.Manager) protocols.ProtocolFactory
 			watcher:       watcher,
 			istioMonitor:  istio,
 			nodeJSMonitor: nodejs,
+			ebpfManager:   m,
 		}, nil
 	}
 }
@@ -487,7 +504,7 @@ func (o *sslProgram) Name() string {
 }
 
 // ConfigureOptions changes map attributes to the given options.
-func (o *sslProgram) ConfigureOptions(_ *manager.Manager, options *manager.Options) {
+func (o *sslProgram) ConfigureOptions(m *manager.Manager, options *manager.Options) {
 	options.MapSpecEditors[sslSockByCtxMap] = manager.MapSpecEditor{
 		MaxEntries: o.cfg.MaxTrackedConnections,
 		EditorFlag: manager.EditMaxEntries,
@@ -496,12 +513,12 @@ func (o *sslProgram) ConfigureOptions(_ *manager.Manager, options *manager.Optio
 		MaxEntries: o.cfg.MaxTrackedConnections,
 		EditorFlag: manager.EditMaxEntries,
 	}
+	o.addProcessExitProbe(m, options)
 }
 
 // PreStart is called before the start of the provided eBPF manager.
-func (o *sslProgram) PreStart(m *manager.Manager) error {
-	o.watcher.Start()
-	o.watcher.SetEbpfManager(m)
+func (o *sslProgram) PreStart(*manager.Manager) error {
+	o.watcher.Start(o.cleanupDeadPids)
 	o.istioMonitor.Start()
 	o.nodeJSMonitor.Start()
 	return nil
@@ -806,3 +823,87 @@ func getUID(lib utils.PathIdentifier) string {
 func (*sslProgram) IsBuildModeSupported(buildmode.Type) bool {
 	return true
 }
+
+// addProcessExitProbe adds a raw or regular tracepoint program depending on which is supported.
+func (o *sslProgram) addProcessExitProbe(mgr *manager.Manager, options *manager.Options) {
+	if features.HaveProgramType(ebpf.RawTracepoint) == nil {
+		// use a raw tracepoint on a supported kernel to intercept terminated threads and clear the corresponding maps
+		p := &manager.Probe{
+			ProbeIdentificationPair: manager.ProbeIdentificationPair{
+				EBPFFuncName: rawTracepointSchedProcessExit,
+				UID:          probeUID,
+			},
+			TracepointName: "sched_process_exit",
+		}
+		mgr.Probes = append(mgr.Probes, p)
+		options.ActivatedProbes = append(options.ActivatedProbes, &manager.ProbeSelector{ProbeIdentificationPair: p.ProbeIdentificationPair})
+		// exclude regular tracepoint
+		options.ExcludedFunctions = append(options.ExcludedFunctions, oldTracepointSchedProcessExit)
+	} else {
+		// use a regular tracepoint to intercept terminated threads
+		p := &manager.Probe{
+			ProbeIdentificationPair: manager.ProbeIdentificationPair{
+				EBPFFuncName: oldTracepointSchedProcessExit,
+				UID:          probeUID,
+			},
+		}
+		mgr.Probes = append(mgr.Probes, p)
+		options.ActivatedProbes = append(options.ActivatedProbes, &manager.ProbeSelector{ProbeIdentificationPair: p.ProbeIdentificationPair})
+		// exclude a raw tracepoint
+		options.ExcludedFunctions = append(options.ExcludedFunctions, rawTracepointSchedProcessExit)
+	}
+}
+
+var sslPidKeyMaps = []string{
+	"ssl_read_args",
+	"ssl_read_ex_args",
+	"ssl_write_args",
+	"ssl_write_ex_args",
+	"ssl_ctx_by_pid_tgid",
+	"bio_new_socket_args",
+}
+
+// cleanupDeadPids clears maps of terminated processes.
+func (o *sslProgram) cleanupDeadPids(alivePIDs map[uint32]struct{}) {
+	if o.ebpfManager != nil {
+		err := o.deleteDeadPidsInMaps(sslPidKeyMaps, alivePIDs)
+		if err != nil {
+			log.Debugf("SSL maps cleanup error: %v", err)
+		}
+	}
+}
+
+// deleteDeadPidsInMaps deletes dead processes in maps with the key 'pid_tgid'
+func (o *sslProgram) deleteDeadPidsInMaps(mapNames []string, alivePIDs map[uint32]struct{}) error {
+	var errs []error
+	for _, n := range mapNames {
+		err := deleteDeadPidsInMap(o.ebpfManager, n, alivePIDs)
+		if err != nil {
+			errs = append(errs, err)
+		}
+	}
+	return errors.Join(errs...)
+}
+
+// deleteDeadPidsInMap finds a map by name and deletes dead processes.
+func deleteDeadPidsInMap(manager *manager.Manager, mapName string, alivePIDs map[uint32]struct{}) error {
+	emap, _, err := manager.GetMap(mapName)
+	if err != nil {
+		return fmt.Errorf("dead process cleaner failed to get map: %q error: %s", mapName, err)
+	}
+
+	var keysToDelete []uint64
+	var key uint64
+	value := make([]byte, emap.ValueSize())
+	iter := emap.Iterate()
+
+	for iter.Next(unsafe.Pointer(&key), unsafe.Pointer(&value)) {
+		pid := uint32(key >> 32)
+		if _, exists := alivePIDs[pid]; !exists {
+			keysToDelete = append(keysToDelete, key)
+		}
+	}
+	_, err = emap.BatchDelete(keysToDelete, nil)
+
+	return err
+}