usm: Increase process monitor channel length

pkg/network/config/config.go

@@ -310,6 +310,9 @@ type Config struct {
 	// EnableUSMEventStream enables USM to use the event stream instead
 	// of netlink for receiving process events.
 	EnableUSMEventStream bool
+
+	// ProcessMonitorChannelLen ..
+	ProcessMonitorChannelLen int
 }
 
 func join(pieces ...string) string {
@@ -424,6 +427,7 @@ func New() *Config {
 		EnableUSMConnectionRollup:   cfg.GetBool(join(smNS, "enable_connection_rollup")),
 		EnableUSMRingBuffers:        cfg.GetBool(join(smNS, "enable_ring_buffers")),
 		EnableUSMEventStream:        cfg.GetBool(join(smNS, "enable_event_stream")),
+		ProcessMonitorChannelLen:    cfg.GetInt(join(smNS, "process_monitor_channel_len")),
 	}
 
 	httpRRKey := join(smNS, "http_replace_rules")

pkg/network/usm/ebpf_gotls.go

@@ -22,6 +22,7 @@ import (
 
 	"github.com/cihub/seelog"
 	"github.com/cilium/ebpf"
+	"github.com/shirou/gopsutil/v3/process"
 	"golang.org/x/sys/unix"
 
 	manager "github.com/DataDog/ebpf-manager"
@@ -37,6 +38,7 @@ import (
 	usmconfig "github.com/DataDog/datadog-agent/pkg/network/usm/config"
 	"github.com/DataDog/datadog-agent/pkg/network/usm/utils"
 	"github.com/DataDog/datadog-agent/pkg/process/monitor"
+	"github.com/DataDog/datadog-agent/pkg/util/kernel"
 	"github.com/DataDog/datadog-agent/pkg/util/log"
 )
 
@@ -244,13 +246,97 @@ func (p *goTLSProgram) PreStart(m *manager.Manager) error {
 				for deletedPid := range deletedPids {
 					_ = p.registry.Unregister(deletedPid)
 				}
+				p.registry.Log()
+				p.Check()
 			}
 		}
 	}()
 
 	return nil
 }
 
+func (p *goTLSProgram) Check() error {
+	procRoot := kernel.ProcFSRoot()
+	pids, err := process.Pids()
+	if err != nil {
+		return err
+	}
+
+	seenPids := make(map[int]struct{})
+	for _, program := range utils.GetTracedProgramList() {
+		if program.ProgramType != "go-tls" {
+			continue
+		}
+
+		for _, pid := range program.PIDs {
+			seenPids[int(pid)] = struct{}{}
+		}
+	}
+
+	blockedPathIds := make(map[utils.PathIdentifier]struct{})
+	for _, blocked := range utils.GetBlockedPathIDsList() {
+		if blocked.ProgramType != "go-tls" {
+			continue
+		}
+
+		for _, pathId := range blocked.PathIdentifiers {
+			blockedPathIds[pathId.PathIdentifier] = struct{}{}
+		}
+	}
+
+	selfPid := os.Getpid()
+
+	for _, pid := range pids {
+		proc, err := process.NewProcess(pid)
+		if err != nil {
+			continue
+		}
+
+		created, err := proc.CreateTime()
+		if err != nil {
+			continue
+		}
+
+		now := time.Now().UnixMilli()
+		alive := time.Duration((now - created) * int64(time.Millisecond))
+		if alive < 60*time.Second {
+			continue
+		}
+
+		if _, seen := seenPids[int(pid)]; seen {
+			continue
+		}
+
+		if p.cfg.GoTLSExcludeSelf && int(pid) == selfPid {
+			continue
+		}
+
+		pidAsStr := strconv.FormatUint(uint64(pid), 10)
+		exePath := filepath.Join(p.procRoot, pidAsStr, "exe")
+		binPath, err := utils.ResolveSymlink(exePath)
+		if err != nil {
+			continue
+		}
+
+		if internalProcessRegex.MatchString(binPath) {
+			continue
+		}
+
+		path, err := utils.NewFilePath(procRoot, binPath, uint32(pid))
+		if err != nil {
+			continue
+		}
+
+		if _, blocked := blockedPathIds[path.ID]; blocked {
+			continue
+		}
+
+		log.Errorf("go-tls: pid (%v) path (%v) neither hooked nor blocked", pid, path)
+	}
+
+	return nil
+}
+
 // PostStart registers the goTLS program to the attacher list.
 func (p *goTLSProgram) PostStart(*manager.Manager) error {
 	utils.AddAttacher(p.Name(), p)

pkg/network/usm/monitor.go

@@ -134,7 +134,7 @@ func (m *Monitor) Start() error {
 
 	// Need to explicitly save the error in `err` so the defer function could save the startup error.
 	if usmconfig.NeedProcessMonitor(m.cfg) {
-		err = m.processMonitor.Initialize(m.cfg.EnableUSMEventStream)
+		err = m.processMonitor.Initialize2(m.cfg.EnableUSMEventStream, m.cfg.ProcessMonitorChannelLen)
 	}
 
 	return err

pkg/network/usm/nodejs.go

@@ -125,6 +125,7 @@ func newNodeJSMonitor(c *config.Config, mgr *manager.Manager) (*nodeJSMonitor, e
 		EbpfConfig:                     &c.Config,
 		ExcludeTargets:                 uprobes.ExcludeSelf | uprobes.ExcludeInternal | uprobes.ExcludeBuildkit | uprobes.ExcludeContainerdTmp,
 		EnablePeriodicScanNewProcesses: true,
+		ProcessMonitorEventStream:      c.EnableUSMEventStream,
 	}
 
 	attacher, err := uprobes.NewUprobeAttacher(nodeJsAttacherName, attachCfg, mgr, uprobes.NopOnAttachCallback, &uprobes.NativeBinaryInspector{})

pkg/process/monitor/process_monitor.go

@@ -30,7 +30,7 @@ const (
 	// The size of the process events queue of netlink.
 	processMonitorEventQueueSize = 2048
 	// The size of the callbacks queue for pending tasks.
-	pendingCallbacksQueueSize = 1000
+	pendingCallbacksQueueSize = 4000
 )
 
 var (
@@ -220,9 +220,10 @@ func (pm *ProcessMonitor) initNetlinkProcessEventMonitor() error {
 }
 
 // initCallbackRunner runs multiple workers that run tasks sent over a queue.
-func (pm *ProcessMonitor) initCallbackRunner() {
+func (pm *ProcessMonitor) initCallbackRunner(chanLen int) {
 	cpuNum := runtime.NumVCPU()
-	pm.callbackRunner = make(chan func(), pendingCallbacksQueueSize)
+	pm.callbackRunner = make(chan func(), chanLen)
+	log.Info("process monitor callback channel len", chanLen)
 	pm.callbackRunnerStopChannel = make(chan struct{})
 	pm.callbackRunnersWG.Add(cpuNum)
 	for i := 0; i < cpuNum; i++ {
@@ -355,6 +356,14 @@ func (pm *ProcessMonitor) mainEventLoop() {
 //  2. Run the main event loop in a goroutine.
 //  4. Scans already running processes and call the Exec callbacks on them.
 func (pm *ProcessMonitor) Initialize(useEventStream bool) error {
+	return pm.Initialize2(useEventStream, 0)
+}
+
+// Initialize2 is ...
+func (pm *ProcessMonitor) Initialize2(useEventStream bool, chanLen int) error {
+	if chanLen == 0 {
+		chanLen = pendingCallbacksQueueSize
+	}
 	var initErr error
 	pm.initOnce.Do(
 		func() {
@@ -367,7 +376,7 @@ func (pm *ProcessMonitor) Initialize(useEventStream bool) error {
 
 			pm.useEventStream = useEventStream
 			pm.done = make(chan struct{})
-			pm.initCallbackRunner()
+			pm.initCallbackRunner(chanLen)
 
 			if useEventStream {
 				return