Support monitoring short-lived processes in the access log module

CHANGES.md

@@ -37,6 +37,7 @@ Release Notes.
 * Report a periodic info-level summary of un-resolved remote addresses, including the conntrack and ztunnel correlation statistics, in the access log module.
 * Attach all matched ztunnel `track_outbound` symbol copies when attaching the uprobe in the access log module.
 * Add success/failure counters to the conntrack real peer address queries.
+* Support monitoring short-lived processes in the access log module.
 
 #### Bug Fixes
 * Fix the base image cannot run in the arm64.

bpf/accesslog/process/process.c

@@ -24,6 +24,13 @@ struct {
 
 struct process_execute_event {
     __u32 pid;
+    __u32 pad0;
+    __u64 cgroup_id;
+    // comm is the kernel's task name. User space prefers /proc/<pid>/cmdline for the process
+    // name and only falls back to this when the process is already gone - which is the whole
+    // point of reporting a short-lived process, so its identity must not depend on /proc still
+    // being readable by the time the event is consumed.
+    char comm[16];
 };
 
 struct trace_event_raw_sched_process_fork {
@@ -35,20 +42,95 @@ struct trace_event_raw_sched_process_fork {
         char __data[0];
 }  __attribute__((preserve_access_index)) ;
 
-SEC("tracepoint/sched/sched_process_fork")
-int tracepoint_sched_process_fork(struct trace_event_raw_sched_process_fork* ctx) {
-    // ctx->parent_pid is the forking task's TID (thread id), not the process id.
-    // For multi-threaded apps (e.g. JVM) a worker thread forking would report its
-    // thread TID, which the periodic /proc scan never lists, causing the process to
-    // flip between detected and dead. Use the real TGID instead.
+struct trace_event_raw_sched_process_exec {
+        struct trace_entry ent;
+        __u32 __data_loc_filename;
+        __u32 pid;
+        __u32 old_pid;
+        char __data[0];
+}  __attribute__((preserve_access_index)) ;
+
+// report_running_process starts tracing the current process and tells user space about it, unless
+// it is filtered out or already known. Shared by the fork and the exec tracepoint, which differ
+// only in *when* they notice a process, never in how it is judged.
+static __always_inline void report_running_process(void *ctx) {
     __u32 tgid = bpf_get_current_pid_tgid() >> 32;
-    // adding to the monitor
+
+    // Reject what we are not interested in before emitting anything. This is the check that keeps
+    // the machine-wide fork rate off the perf queue and out of user space, so it runs first, ahead
+    // of the process_monitor_control lookup below: the host processes it rejects fork far more than
+    // the monitored pods do, and this way they cost a single lookup rather than two.
+    __u64 cgroup_id = 0;
+    if (cgroup_filter_enabled) {
+        // Preferred: the cgroup is known up front, so a host process never reaches user space.
+        cgroup_id = bpf_get_current_cgroup_id();
+        if (!bpf_map_lookup_elem(&cgroup_monitor_allowlist, &cgroup_id)) {
+            return;
+        }
+    } else {
+        // Fallback: nothing is known up front, so every process gets reported once and only the
+        // ones user space rejected are filtered out from then on.
+        if (bpf_map_lookup_elem(&process_not_monitor, &tgid)) {
+            return;
+        }
+    }
+
+    // Already decided for this process - nothing new to tell user space. Without this a
+    // thread-happy process (a JVM spawning workers) would emit an event and force a user-space
+    // re-check on every single clone; with it, each process costs one event per lifetime.
+    if (bpf_map_lookup_elem(&process_monitor_control, &tgid)) {
+        return;
+    }
+
+    // Start tracing immediately, optimistically: the process may be gone long before the
+    // periodic discovery would have found it. User space still gets the final say and removes
+    // the entry if the process turns out not to be monitorable.
     __u32 v = 1;
     bpf_map_update_elem(&process_monitor_control, &tgid, &v, 0);
 
-    // send to the user-space to check the pid should monitor or not
     struct process_execute_event event = {};
     event.pid = tgid;
+    event.cgroup_id = cgroup_id;
+    bpf_get_current_comm(&event.comm, sizeof(event.comm));
     bpf_perf_event_output(ctx, &process_execute_queue, BPF_F_CURRENT_CPU, &event, sizeof(event));
+}
+
+// A process that execs is reported here rather than at fork, because only here is it the thing it
+// is going to be: the runtime forks a helper from its *own* cgroup (containerd-shim sits in a host
+// cgroup, e.g. /podruntime/docker), and that child only joins the container's cgroup and namespaces
+// afterwards, right before execve. Judging at fork would therefore test the shim's cgroup and
+// reject every process started through `kubectl exec` - exactly the short-lived kind this exists
+// for. At exec the process already carries its final cgroup, pid and name.
+SEC("tracepoint/sched/sched_process_exec")
+int tracepoint_sched_process_exec(struct trace_event_raw_sched_process_exec* ctx) {
+    report_running_process(ctx);
+    return 0;
+}
+
+SEC("tracepoint/sched/sched_process_fork")
+int tracepoint_sched_process_fork(struct trace_event_raw_sched_process_fork* ctx) {
+    // Reclaim any entry left behind by a dead task that used to own the child's pid.
+    //
+    // The kernel allocates every task's pid - thread (TID) and process (TGID) alike - from one
+    // global allocator, so a pid identifies at most one task at a time. ctx->child_pid was just
+    // handed out by copy_process(), which proves the pid was free a moment ago, which in turn
+    // proves any entry still keyed by it belongs to a task that has already exited: a live task
+    // never releases its pid, so the child could not have been given the number. Deleting here
+    // is therefore safe (it can never evict a live process) and sufficient (a pid can only be
+    // reused by going through a fork, so this runs before the new owner executes a single
+    // instruction). That removes the need for an exit tracepoint or a periodic sweep.
+    //
+    // This has to happen for every fork, before any of the filtering below can return early.
+    __u32 child = ctx->child_pid;
+    bpf_map_delete_elem(&process_monitor_control, &child);
+    bpf_map_delete_elem(&process_not_monitor, &child);
+
+    // Report the forking process itself. The exec tracepoint covers anything that starts a new
+    // program; this additionally catches a process that only ever forks - and, because
+    // bpf_get_current_pid_tgid() is read inside, it reports the real TGID rather than
+    // ctx->parent_pid, which for a multi-threaded app (e.g. a JVM worker thread forking) is the
+    // thread's TID that the periodic /proc scan never lists, making the process flip between
+    // detected and dead.
+    report_running_process(ctx);
     return 0;
 }

bpf/accesslog/process/process.h

@@ -24,6 +24,44 @@ struct {
 	__type(value, __u32);
 } process_monitor_control SEC(".maps");
 
+// cgroup_monitor_allowlist holds the cgroup id of every cgroup a monitored process was found in.
+// User space fills it from the periodic process discovery (which reads /proc and is therefore
+// never gated by this map), and the fork tracepoint reads it to decide, entirely in kernel space,
+// whether a forking process is worth reporting. A node runs far more unmonitored host processes
+// (kubelet, containerd, systemd, ...) than monitored pod ones, so rejecting them here - before any
+// perf event is emitted and before user space touches /proc - is what makes the tracepoint cheap.
+//
+// Only consulted when cgroup_filter_enabled is set; see process_not_monitor for the fallback.
+struct {
+	__uint(type, BPF_MAP_TYPE_HASH);
+	__uint(max_entries, 10000);
+	__type(key, __u64);
+	__type(value, __u32);
+} cgroup_monitor_allowlist SEC(".maps");
+
+// process_not_monitor is the fallback filter for hosts where the cgroup allowlist cannot be built
+// (cgroup v1, or a cgroup tree user space failed to walk). There the tracepoint cannot tell a pod
+// process from a host one up front, so it reports every process once and user space records the
+// ones it rejected here, which keeps them from being reported again.
+//
+// It is weaker than the allowlist - every process costs one event and one /proc read instead of
+// zero - but still collapses the machine-wide *fork* rate down to one event per process lifetime,
+// which is the cost that made this tracepoint unaffordable before.
+//
+// LRU rather than HASH: entries accumulate for every unmonitored process on the node and nothing
+// enumerates them for removal, so eviction has to be the kernel's job. Evicting one only costs a
+// re-check, whereas a full map would silently stop recording rejections.
+struct {
+	__uint(type, BPF_MAP_TYPE_LRU_HASH);
+	__uint(max_entries, 10240);
+	__type(key, __u32);
+	__type(value, __u32);
+} process_not_monitor SEC(".maps");
+
+// cgroup_filter_enabled selects between the two filters above. User space sets it at startup after
+// probing the host, and it never changes afterwards.
+volatile __u32 cgroup_filter_enabled;
+
 volatile __u32 ztunnel_process_pid;
 
 static __inline bool tgid_should_trace(__u32 tgid) {

pkg/accesslog/collector/process.go

@@ -18,13 +18,33 @@
 package collector
 
 import (
+	"bytes"
+
 	"github.com/apache/skywalking-rover/pkg/accesslog/common"
+	"github.com/apache/skywalking-rover/pkg/logger"
 	"github.com/apache/skywalking-rover/pkg/module"
 )
 
+var processLog = logger.GetLogger("access_log", "collector", "process")
+
+// processExecuteQueueParallels reads the process events with more than one goroutine. The queue is
+// low volume by design - the kernel filters before emitting - but a process that is reported and
+// then handled late is a process whose access logs sit waiting, so keep the drain prompt.
+const processExecuteQueueParallels = 2
+
+// processExecuteQueuePerCPUBuffer sizes the per-CPU perf buffer. A burst of process starts (a pod
+// rolling out, a CronJob firing) must not be dropped, and the events are 32 bytes each.
+const processExecuteQueuePerCPUBuffer = 32 * 1024
+
 var processCollectInstance = NewProcessCollector()
 
-// ProcessCollector Management all processes which needs to be monitored
+// ProcessCollector notices processes as the kernel starts them, rather than waiting for the
+// periodic /proc scan to come round.
+//
+// The scan cannot see a process that starts and exits between two of its rounds, so anything short
+// lived - a `kubectl exec`, a CronJob, a CGI worker - produces socket traffic that is dropped in the
+// kernel because the process was never marked as monitored. This collector closes that window: the
+// kernel marks a process on sight and tells user space, which then confirms or removes it.
 type ProcessCollector struct {
 }
 
@@ -33,12 +53,25 @@ func NewProcessCollector() *ProcessCollector {
 }
 
 func (p *ProcessCollector) Start(_ *module.Manager, context *common.AccessLogContext) error {
-	// monitor process been execute
+	// Nothing is attached unless user space can actually judge what the kernel reports. Leaving the
+	// tracepoints off is not a degraded mode - it is exactly the behavior before this collector
+	// existed, with the periodic scan still discovering every process that lives long enough.
+	if !context.ConnectionMgr.ProcessExecuteMonitorReady() {
+		processLog.Info("not monitoring process execution; short-lived processes will only be " +
+			"discovered by the periodic process scan")
+		return nil
+	}
+
+	// exec is what catches a short-lived process: only there does it carry its own pid, its final
+	// cgroup and its final name. fork additionally keeps the monitor map free of pids left behind by
+	// dead processes, and catches a process that only ever forks.
+	context.BPF.AddTracePoint("sched", "sched_process_exec", context.BPF.TracepointSchedProcessExec)
 	context.BPF.AddTracePoint("sched", "sched_process_fork", context.BPF.TracepointSchedProcessFork)
 
-	context.BPF.ReadEventAsync(context.BPF.ProcessExecuteQueue, func(data interface{}) {
-		context.ConnectionMgr.OnNewProcessExecuting(int32(data.(*ProcessExecuteEvent).PID))
-	}, func() interface{} {
+	context.BPF.ReadEventAsyncWithBufferSize(context.BPF.ProcessExecuteQueue, func(data interface{}) {
+		event := data.(*ProcessExecuteEvent)
+		context.ConnectionMgr.OnNewProcessExecuting(int32(event.PID), event.CgroupID, event.CommString())
+	}, processExecuteQueuePerCPUBuffer, processExecuteQueueParallels, func() interface{} {
 		return &ProcessExecuteEvent{}
 	})
 
@@ -48,6 +81,23 @@ func (p *ProcessCollector) Start(_ *module.Manager, context *common.AccessLogCon
 func (p *ProcessCollector) Stop() {
 }
 
+// ProcessExecuteEvent must keep the exact memory layout of struct process_execute_event in
+// bpf/accesslog/process/process.c: the perf record is decoded with binary.Read, which lays the
+// fields out back to back and would silently shift every field after an implicit C alignment gap.
+// PAD0 is the explicit stand-in for the padding the compiler inserts before the 8-byte aligned
+// CgroupID.
 type ProcessExecuteEvent struct {
-	PID uint32
+	PID      uint32
+	PAD0     uint32
+	CgroupID uint64
+	Comm     [16]byte
+}
+
+// CommString returns the kernel task name as a string, dropping the NUL padding that
+// bpf_get_current_comm leaves in the fixed size buffer.
+func (e *ProcessExecuteEvent) CommString() string {
+	if idx := bytes.IndexByte(e.Comm[:], 0); idx >= 0 {
+		return string(e.Comm[:idx])
+	}
+	return string(e.Comm[:])
 }