usm: process monitor: reuse consumers.NewProcessConsumer (DataDog#31864)

cmd/system-probe/modules/eventmonitor.go

@@ -85,14 +85,9 @@ func createEventMonitorModule(_ *sysconfigtypes.Config, deps module.FactoryDepen
 
 	netconfig := netconfig.New()
 	if netconfig.EnableUSMEventStream {
-		procmonconsumer, err := createProcessMonitorConsumer(evm, netconfig)
-		if err != nil {
+		if err := createProcessMonitorConsumer(evm, netconfig); err != nil {
 			return nil, err
 		}
-		if procmonconsumer != nil {
-			evm.RegisterEventConsumer(procmonconsumer)
-			log.Info("USM process monitoring consumer initialized")
-		}
 	}
 
 	gpucfg := gpuconfig.New()

cmd/system-probe/modules/eventmonitor_linux.go

@@ -11,11 +11,13 @@ import (
 	"github.com/DataDog/datadog-agent/cmd/system-probe/api/module"
 	"github.com/DataDog/datadog-agent/cmd/system-probe/config"
 	"github.com/DataDog/datadog-agent/pkg/eventmonitor"
+	"github.com/DataDog/datadog-agent/pkg/eventmonitor/consumers"
 	netconfig "github.com/DataDog/datadog-agent/pkg/network/config"
 	usmconfig "github.com/DataDog/datadog-agent/pkg/network/usm/config"
 	usmstate "github.com/DataDog/datadog-agent/pkg/network/usm/state"
-	procmon "github.com/DataDog/datadog-agent/pkg/process/monitor"
+	"github.com/DataDog/datadog-agent/pkg/process/monitor"
 	secconfig "github.com/DataDog/datadog-agent/pkg/security/config"
+	"github.com/DataDog/datadog-agent/pkg/util/log"
 )
 
 // EventMonitor - Event monitor Factory
@@ -28,10 +30,28 @@ var EventMonitor = module.Factory{
 	},
 }
 
-func createProcessMonitorConsumer(evm *eventmonitor.EventMonitor, config *netconfig.Config) (eventmonitor.EventConsumer, error) {
+const (
+	eventMonitorID          = "PROCESS_MONITOR"
+	eventMonitorChannelSize = 500
+)
+
+var (
+	eventTypes = []consumers.ProcessConsumerEventTypes{
+		consumers.ExecEventType,
+		consumers.ExitEventType,
+	}
+)
+
+func createProcessMonitorConsumer(evm *eventmonitor.EventMonitor, config *netconfig.Config) error {
 	if !usmconfig.IsUSMSupportedAndEnabled(config) || !usmconfig.NeedProcessMonitor(config) || usmstate.Get() != usmstate.Running {
-		return nil, nil
+		return nil
 	}
 
-	return procmon.NewProcessMonitorEventConsumer(evm)
+	consumer, err := consumers.NewProcessConsumer(eventMonitorID, eventMonitorChannelSize, eventTypes, evm)
+	if err != nil {
+		return err
+	}
+	monitor.InitializeEventConsumer(consumer)
+	log.Info("USM process monitoring consumer initialized")
+	return nil
 }

cmd/system-probe/modules/eventmonitor_windows.go

@@ -21,8 +21,8 @@ var EventMonitor = module.Factory{
 	Fn:               createEventMonitorModule,
 }
 
-func createProcessMonitorConsumer(_ *eventmonitor.EventMonitor, _ *netconfig.Config) (eventmonitor.EventConsumer, error) {
-	return nil, nil
+func createProcessMonitorConsumer(_ *eventmonitor.EventMonitor, _ *netconfig.Config) error {
+	return nil
 }
 
 func createGPUProcessEventConsumer(_ *eventmonitor.EventMonitor) error {

pkg/ebpf/uprobes/attacher_test.go

@@ -27,14 +27,12 @@ import (
 	"github.com/DataDog/datadog-agent/pkg/ebpf/bytecode"
 	"github.com/DataDog/datadog-agent/pkg/ebpf/ebpftest"
 	"github.com/DataDog/datadog-agent/pkg/ebpf/prebuilt"
-	eventmonitortestutil "github.com/DataDog/datadog-agent/pkg/eventmonitor/testutil"
+	"github.com/DataDog/datadog-agent/pkg/eventmonitor/consumers/testutil"
 	"github.com/DataDog/datadog-agent/pkg/network/go/bininspect"
 	"github.com/DataDog/datadog-agent/pkg/network/usm/sharedlibraries"
 	fileopener "github.com/DataDog/datadog-agent/pkg/network/usm/sharedlibraries/testutil"
 	"github.com/DataDog/datadog-agent/pkg/network/usm/utils"
 	"github.com/DataDog/datadog-agent/pkg/process/monitor"
-	procmontestutil "github.com/DataDog/datadog-agent/pkg/process/monitor/testutil"
-	secutils "github.com/DataDog/datadog-agent/pkg/security/utils"
 	"github.com/DataDog/datadog-agent/pkg/util/kernel"
 )
 
@@ -800,8 +798,7 @@ func launchProcessMonitor(t *testing.T, useEventStream bool) *monitor.ProcessMon
 	t.Cleanup(pm.Stop)
 	require.NoError(t, pm.Initialize(useEventStream))
 	if useEventStream {
-		secutils.SetCachedHostname("test-hostname")
-		eventmonitortestutil.StartEventMonitor(t, procmontestutil.RegisterProcessMonitorEventConsumer)
+		monitor.InitializeEventConsumer(testutil.NewTestProcessConsumer(t))
 	}
 
 	return pm

pkg/network/usm/monitor_tls_test.go

@@ -31,7 +31,7 @@ import (
 
 	"github.com/DataDog/datadog-agent/pkg/ebpf/ebpftest"
 	"github.com/DataDog/datadog-agent/pkg/ebpf/prebuilt"
-	eventmonitortestutil "github.com/DataDog/datadog-agent/pkg/eventmonitor/testutil"
+	consumerstestutil "github.com/DataDog/datadog-agent/pkg/eventmonitor/consumers/testutil"
 	"github.com/DataDog/datadog-agent/pkg/network"
 	"github.com/DataDog/datadog-agent/pkg/network/config"
 	"github.com/DataDog/datadog-agent/pkg/network/protocols"
@@ -44,8 +44,7 @@ import (
 	"github.com/DataDog/datadog-agent/pkg/network/usm/consts"
 	usmtestutil "github.com/DataDog/datadog-agent/pkg/network/usm/testutil"
 	"github.com/DataDog/datadog-agent/pkg/network/usm/utils"
-	procmontestutil "github.com/DataDog/datadog-agent/pkg/process/monitor/testutil"
-	secutils "github.com/DataDog/datadog-agent/pkg/security/utils"
+	"github.com/DataDog/datadog-agent/pkg/process/monitor"
 	globalutils "github.com/DataDog/datadog-agent/pkg/util/testutil"
 	dockerutils "github.com/DataDog/datadog-agent/pkg/util/testutil/docker"
 )
@@ -870,8 +869,7 @@ func setupUSMTLSMonitor(t *testing.T, cfg *config.Config) *Monitor {
 	require.NoError(t, err)
 	require.NoError(t, usmMonitor.Start())
 	if cfg.EnableUSMEventStream && usmconfig.NeedProcessMonitor(cfg) {
-		secutils.SetCachedHostname("test-hostname")
-		eventmonitortestutil.StartEventMonitor(t, procmontestutil.RegisterProcessMonitorEventConsumer)
+		monitor.InitializeEventConsumer(consumerstestutil.NewTestProcessConsumer(t))
 	}
 	t.Cleanup(usmMonitor.Stop)
 	t.Cleanup(utils.ResetDebugger)

pkg/network/usm/sharedlibraries/watcher_test.go

@@ -26,13 +26,11 @@ import (
 
 	"github.com/DataDog/datadog-agent/pkg/ebpf/ebpftest"
 	"github.com/DataDog/datadog-agent/pkg/ebpf/prebuilt"
-	eventmonitortestutil "github.com/DataDog/datadog-agent/pkg/eventmonitor/testutil"
+	"github.com/DataDog/datadog-agent/pkg/eventmonitor/consumers/testutil"
 	usmconfig "github.com/DataDog/datadog-agent/pkg/network/usm/config"
 	fileopener "github.com/DataDog/datadog-agent/pkg/network/usm/sharedlibraries/testutil"
 	"github.com/DataDog/datadog-agent/pkg/network/usm/utils"
 	"github.com/DataDog/datadog-agent/pkg/process/monitor"
-	procmontestutil "github.com/DataDog/datadog-agent/pkg/process/monitor/testutil"
-	secutils "github.com/DataDog/datadog-agent/pkg/security/utils"
 	"github.com/DataDog/datadog-agent/pkg/util/kernel"
 	"github.com/DataDog/datadog-agent/pkg/util/log"
 )
@@ -42,8 +40,7 @@ func launchProcessMonitor(t *testing.T, useEventStream bool) {
 	t.Cleanup(pm.Stop)
 	require.NoError(t, pm.Initialize(useEventStream))
 	if useEventStream {
-		secutils.SetCachedHostname("test-hostname")
-		eventmonitortestutil.StartEventMonitor(t, procmontestutil.RegisterProcessMonitorEventConsumer)
+		monitor.InitializeEventConsumer(testutil.NewTestProcessConsumer(t))
 	}
 }
 

pkg/process/monitor/process_monitor.go

@@ -18,10 +18,9 @@ import (
 	"github.com/vishvananda/netlink"
 	"go.uber.org/atomic"
 
-	"github.com/DataDog/datadog-agent/pkg/eventmonitor"
+	"github.com/DataDog/datadog-agent/pkg/eventmonitor/consumers"
 	"github.com/DataDog/datadog-agent/pkg/network/protocols/telemetry"
 	"github.com/DataDog/datadog-agent/pkg/runtime"
-	"github.com/DataDog/datadog-agent/pkg/security/secl/model"
 	"github.com/DataDog/datadog-agent/pkg/util/kernel"
 	"github.com/DataDog/datadog-agent/pkg/util/log"
 )
@@ -490,76 +489,20 @@ func (pm *ProcessMonitor) Stop() {
 	pm.processExitCallbacksMutex.Unlock()
 }
 
-// Event defines the event used by the process monitor
-type Event struct {
-	Type model.EventType
-	Pid  uint32
-}
-
-// EventConsumer defines an event consumer to handle event monitor events in the
-// process monitor
-type EventConsumer struct{}
-
-// NewProcessMonitorEventConsumer returns a new process monitor event consumer
-func NewProcessMonitorEventConsumer(em *eventmonitor.EventMonitor) (*EventConsumer, error) {
-	consumer := &EventConsumer{}
-	err := em.AddEventConsumerHandler(consumer)
-	return consumer, err
-}
-
-// ChanSize returns the channel size used by this consumer
-func (ec *EventConsumer) ChanSize() int {
-	return 500
-}
-
-// ID returns the ID of this consumer
-func (ec *EventConsumer) ID() string {
-	return "PROCESS_MONITOR"
-}
-
-// Start the consumer
-func (ec *EventConsumer) Start() error {
-	return nil
-}
-
-// Stop the consumer
-func (ec *EventConsumer) Stop() {
-}
-
-// EventTypes returns the event types handled by this consumer
-func (ec *EventConsumer) EventTypes() []model.EventType {
-	return []model.EventType{
-		model.ExecEventType,
-		model.ExitEventType,
-	}
-}
-
-// HandleEvent handles events received from the event monitor
-func (ec *EventConsumer) HandleEvent(event any) {
-	sevent, ok := event.(*Event)
-	if !ok {
-		return
-	}
-
-	processMonitor.tel.events.Add(1)
-	switch sevent.Type {
-	case model.ExecEventType:
+// InitializeEventConsumer initializes the event consumer with the event handling.
+func InitializeEventConsumer(consumer *consumers.ProcessConsumer) {
+	consumer.SubscribeExec(func(pid uint32) {
+		processMonitor.tel.events.Add(1)
 		processMonitor.tel.exec.Add(1)
 		if processMonitor.hasExecCallbacks.Load() {
-			processMonitor.handleProcessExec(sevent.Pid)
+			processMonitor.handleProcessExec(pid)
 		}
-	case model.ExitEventType:
+	})
+	consumer.SubscribeExit(func(pid uint32) {
+		processMonitor.tel.events.Add(1)
 		processMonitor.tel.exit.Add(1)
 		if processMonitor.hasExitCallbacks.Load() {
-			processMonitor.handleProcessExit(sevent.Pid)
+			processMonitor.handleProcessExit(pid)
 		}
-	}
-}
-
-// Copy should copy the given event or return nil to discard it
-func (ec *EventConsumer) Copy(event *model.Event) any {
-	return &Event{
-		Type: event.GetEventType(),
-		Pid:  event.GetProcessPid(),
-	}
+	})
 }

pkg/process/monitor/process_monitor_test.go

@@ -19,13 +19,10 @@ import (
 	"github.com/vishvananda/netns"
 	"go.uber.org/atomic"
 
-	"github.com/DataDog/datadog-agent/pkg/eventmonitor"
-	eventmonitortestutil "github.com/DataDog/datadog-agent/pkg/eventmonitor/testutil"
+	"github.com/DataDog/datadog-agent/pkg/eventmonitor/consumers/testutil"
 	"github.com/DataDog/datadog-agent/pkg/network/protocols/telemetry"
-	"github.com/DataDog/datadog-agent/pkg/security/utils"
 	"github.com/DataDog/datadog-agent/pkg/util"
 	"github.com/DataDog/datadog-agent/pkg/util/kernel"
-	"github.com/DataDog/datadog-agent/pkg/util/log"
 )
 
 func getProcessMonitor(t *testing.T) *ProcessMonitor {
@@ -40,12 +37,12 @@ func getProcessMonitor(t *testing.T) *ProcessMonitor {
 func waitForProcessMonitor(t *testing.T, pm *ProcessMonitor) {
 	execCounter := atomic.NewInt32(0)
 	execCallback := func(_ uint32) { execCounter.Inc() }
-	registerCallback(t, pm, true, (*ProcessCallback)(&execCallback))
+	registerCallback(t, pm, true, &execCallback)
 
 	exitCounter := atomic.NewInt32(0)
 	// Sanity subscribing a callback.
 	exitCallback := func(_ uint32) { exitCounter.Inc() }
-	registerCallback(t, pm, false, (*ProcessCallback)(&exitCallback))
+	registerCallback(t, pm, false, &exitCallback)
 
 	require.Eventually(t, func() bool {
 		_ = exec.Command("/bin/echo").Run()
@@ -56,14 +53,7 @@ func waitForProcessMonitor(t *testing.T, pm *ProcessMonitor) {
 func initializePM(t *testing.T, pm *ProcessMonitor, useEventStream bool) {
 	require.NoError(t, pm.Initialize(useEventStream))
 	if useEventStream {
-		utils.SetCachedHostname("test-hostname")
-		eventmonitortestutil.StartEventMonitor(t, func(t *testing.T, evm *eventmonitor.EventMonitor) {
-			// Can't use the implementation in procmontestutil due to import cycles
-			procmonconsumer, err := NewProcessMonitorEventConsumer(evm)
-			require.NoError(t, err)
-			evm.RegisterEventConsumer(procmonconsumer)
-			log.Info("process monitoring test consumer initialized")
-		})
+		InitializeEventConsumer(testutil.NewTestProcessConsumer(t))
 	}
 	waitForProcessMonitor(t, pm)
 }
@@ -113,7 +103,7 @@ func (s *processMonitorSuite) TestProcessMonitorSanity() {
 		defer execsMutex.Unlock()
 		execs[pid] = struct{}{}
 	}
-	registerCallback(t, pm, true, (*ProcessCallback)(&callback))
+	registerCallback(t, pm, true, &callback)
 
 	exitMutex := sync.RWMutex{}
 	exits := make(map[uint32]struct{})
@@ -122,7 +112,7 @@ func (s *processMonitorSuite) TestProcessMonitorSanity() {
 		defer exitMutex.Unlock()
 		exits[pid] = struct{}{}
 	}
-	registerCallback(t, pm, false, (*ProcessCallback)(&exitCallback))
+	registerCallback(t, pm, false, &exitCallback)
 
 	initializePM(t, pm, s.useEventStream)
 	cmd := exec.Command(testBinaryPath, "test")
@@ -182,7 +172,7 @@ func (s *processMonitorSuite) TestProcessRegisterMultipleCallbacks() {
 			defer execCountersMutexes[i].Unlock()
 			c[pid] = struct{}{}
 		}
-		registerCallback(t, pm, true, (*ProcessCallback)(&callback))
+		registerCallback(t, pm, true, &callback)
 
 		exitCountersMutexes[i] = sync.RWMutex{}
 		exitCounters[i] = make(map[uint32]struct{})
@@ -193,7 +183,7 @@ func (s *processMonitorSuite) TestProcessRegisterMultipleCallbacks() {
 			defer exitCountersMutexes[i].Unlock()
 			exitc[pid] = struct{}{}
 		}
-		registerCallback(t, pm, false, (*ProcessCallback)(&exitCallback))
+		registerCallback(t, pm, false, &exitCallback)
 	}
 
 	initializePM(t, pm, s.useEventStream)
@@ -252,10 +242,10 @@ func (s *processMonitorSuite) TestProcessMonitorInNamespace() {
 	pm := getProcessMonitor(t)
 
 	callback := func(pid uint32) { execSet.Store(pid, struct{}{}) }
-	registerCallback(t, pm, true, (*ProcessCallback)(&callback))
+	registerCallback(t, pm, true, &callback)
 
 	exitCallback := func(pid uint32) { exitSet.Store(pid, struct{}{}) }
-	registerCallback(t, pm, false, (*ProcessCallback)(&exitCallback))
+	registerCallback(t, pm, false, &exitCallback)
 
 	monNs, err := netns.New()
 	require.NoError(t, err, "could not create network namespace for process monitor")