From 9f7bcf299e2edbbe8219847519c82479e0b934b7 Mon Sep 17 00:00:00 2001 From: mrproliu <741550557@qq.com> Date: Thu, 16 Jul 2026 14:25:53 +0800 Subject: [PATCH 1/3] Support monitoring short-lived processes in the access log module --- CHANGES.md | 1 + bpf/accesslog/process/process.c | 98 ++++++- bpf/accesslog/process/process.h | 38 +++ pkg/accesslog/collector/process.go | 62 ++++- pkg/accesslog/common/connection.go | 226 +++++++++++++++- pkg/process/api.go | 25 ++ pkg/process/api/process.go | 16 ++ pkg/process/finders/base/finder.go | 13 + pkg/process/finders/base/template.go | 29 +- pkg/process/finders/kubernetes/finder.go | 164 +++++++++++- pkg/process/finders/kubernetes/template.go | 5 +- pkg/process/finders/manager.go | 64 +++++ pkg/process/finders/manager_executing_test.go | 107 ++++++++ pkg/process/module.go | 16 ++ pkg/tools/cgroup/resolver.go | 199 ++++++++++++++ pkg/tools/cgroup/resolver_test.go | 249 ++++++++++++++++++ pkg/tools/host/file.go | 18 ++ pkg/tools/host/file_test.go | 50 ++++ 18 files changed, 1352 insertions(+), 28 deletions(-) create mode 100644 pkg/process/finders/manager_executing_test.go create mode 100644 pkg/tools/cgroup/resolver.go create mode 100644 pkg/tools/cgroup/resolver_test.go create mode 100644 pkg/tools/host/file_test.go diff --git a/CHANGES.md b/CHANGES.md index 1f2ca996..12fd2a8c 100644 --- a/CHANGES.md +++ b/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. diff --git a/bpf/accesslog/process/process.c b/bpf/accesslog/process/process.c index 3658c5c1..dc453926 100644 --- a/bpf/accesslog/process/process.c +++ b/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//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; } \ No newline at end of file diff --git a/bpf/accesslog/process/process.h b/bpf/accesslog/process/process.h index a9dd2f48..ee19d66a 100644 --- a/bpf/accesslog/process/process.h +++ b/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) { diff --git a/pkg/accesslog/collector/process.go b/pkg/accesslog/collector/process.go index e4dbdce9..a9d535a8 100644 --- a/pkg/accesslog/collector/process.go +++ b/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 behaviour 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[:]) } diff --git a/pkg/accesslog/common/connection.go b/pkg/accesslog/common/connection.go index decfce90..ad5d5eed 100644 --- a/pkg/accesslog/common/connection.go +++ b/pkg/accesslog/common/connection.go @@ -138,6 +138,20 @@ type ConnectionManager struct { // monitoring process map in BPF processMonitorMap *ebpf.Map activeConnectionMap *ebpf.Map + // processNotMonitorMap and cgroupAllowlistMap are the two ways the fork/exec tracepoints decide + // what to skip; which one is in force is settled once at startup, see processExecuteMonitorReady + processNotMonitorMap *ebpf.Map + cgroupAllowlistMap *ebpf.Map + // processExecuteMonitorReady records whether the tracepoints were attached at all + processExecuteMonitorReady bool + // cgroupSeeder resolves a container to its cgroup id; nil when the host cannot resolve them + cgroupSeeder process.CgroupOperator + // allowedCgroups counts which monitored pids live in each allowed cgroup, so a cgroup stays + // allowed until the last of them is gone + allowedCgroups map[uint64]map[int32]bool + // processVerdicts remembers what was concluded about a process, so its access logs can outlive + // it just long enough to be flushed + processVerdicts *cache.Expiring monitorFilter MonitorFilter @@ -232,6 +246,10 @@ func NewConnectionManager(config *Config, moduleMgr *module.Manager, bpfLoader * monitoringProcesses: make(map[int32][]api.ProcessInterface), processMonitorMap: bpfLoader.ProcessMonitorControl, activeConnectionMap: bpfLoader.ActiveConnectionMap, + processNotMonitorMap: bpfLoader.ProcessNotMonitor, + cgroupAllowlistMap: bpfLoader.CgroupMonitorAllowlist, + processVerdicts: cache.NewExpiring(), + allowedCgroups: make(map[uint64]map[int32]bool), monitorFilter: filter, flushListeners: make([]FlusherListener, 0), connectTracker: track, @@ -240,9 +258,52 @@ func NewConnectionManager(config *Config, moduleMgr *module.Manager, bpfLoader * unresolvedByReason: make(map[string]int64), resolutionGracePeriod: resolutionGraceFor(flushPeriod), } + mgr.setupProcessExecuteMonitor(bpfLoader) return mgr } +// setupProcessExecuteMonitor settles, once, how the kernel should filter the processes it reports, +// and whether reporting them is worth doing at all. +// +// The cgroup allowlist is the preferred filter because it rejects a host process without emitting +// anything, but it needs the process discovery to resolve cgroup ids, which in turn needs a cgroup +// v2 tree. Where that is missing the per-process rejection list still keeps the cost to one report +// per process, which is what makes these tracepoints affordable in the first place. +func (c *ConnectionManager) setupProcessExecuteMonitor(bpfLoader *bpf.Loader) { + // Nothing is reported unless some active finder can actually act on it. This is not a + // micro-optimisation: a finder that only ever monitors long-lived processes(the VM finder takes + // a process only if it holds a listening port, which one living for milliseconds never does) + // would reject every reported process *after* paying for a /proc read - all cost, no coverage. + // That cost is precisely why these tracepoints were removed from this agent once before. + if !c.processOP.ExecutingProcessesWanted() { + log.Infof("not monitoring process execution: no active process finder can use it, so " + + "short-lived processes stay discoverable only by the periodic scan") + return + } + + seeder, canSeedCgroups := c.processOP.(process.CgroupOperator) + if canSeedCgroups { + canSeedCgroups = seeder.CgroupResolvable() + } + c.cgroupSeeder = seeder + + enabled := uint32(0) + if canSeedCgroups { + enabled = 1 + } + if err := bpfLoader.CgroupFilterEnabled.Set(enabled); err != nil { + log.Warnf("cannot select the process execution filter, not monitoring process execution: %v", err) + return + } + c.processExecuteMonitorReady = true + if canSeedCgroups { + log.Infof("monitoring process execution, filtering by the cgroups of the monitored containers") + } else { + log.Infof("monitoring process execution, filtering per process: no cgroup v2 tree is available, " + + "so every process is reported once before it can be skipped") + } +} + // resolutionGraceFor bounds the resolution-defer grace so it stays below connectionDeleteDelayTime // (with one flush period of headroom). A connection is removed connectionDeleteDelayTime after it // closes, and a deferred log can only be emitted at a flush cycle at or after its resolution @@ -377,11 +438,159 @@ func (c *ConnectionManager) Stop() { c.processOP.DeleteListener(c) } -func (c *ConnectionManager) OnNewProcessExecuting(pid int32) { - // if the process should not be monitoring, then delete in the map - if !c.processOP.ShouldMonitor(pid) { - c.updateMonitorStatusForProcess(pid, false) +// ProcessVerdict is what user space concluded about a process the kernel reported. +type ProcessVerdict int + +const ( + // ProcessVerdictUnknown means no conclusion has been reached yet, either because the report is + // still being handled or because the process vanished before it could be identified. + ProcessVerdictUnknown ProcessVerdict = iota + // ProcessVerdictMonitored means the process is currently monitored. + ProcessVerdictMonitored + // ProcessVerdictApproved means the process passed every filter and has since exited. Its logs + // are still worth sending: the process being short-lived is the reason it is interesting. + ProcessVerdictApproved + // ProcessVerdictRejected means the process was positively identified as one we must not report. + ProcessVerdictRejected +) + +// processVerdictKeepTime is how long a verdict outlives the process. It only has to cover the gap +// between a process exiting and its last buffered access logs being flushed. +const processVerdictKeepTime = time.Minute + +// ProcessExecuteMonitorReady reports whether user space can judge the processes the kernel would +// report, and therefore whether the fork/exec tracepoints are worth attaching at all. +func (c *ConnectionManager) ProcessExecuteMonitorReady() bool { + return c.processExecuteMonitorReady +} + +// OnNewProcessExecuting handles a process the kernel has just started and optimistically marked as +// monitored. Everything the kernel could capture is passed in, because the process may already be +// gone - that is precisely the case this exists for. +func (c *ConnectionManager) OnNewProcessExecuting(pid int32, cgroupID uint64, comm string) { + // The cgroup id here is the kernel's own(bpf_get_current_cgroup_id), while the allowlist that + // let this event through was seeded from a cgroup id user space read as a directory inode. The + // two must be the same number for any of this to work, so log the kernel's side of it: paired + // with the "cgroup mapping: id=... path=..." lines from the walk, it shows whether they agree. + log.Debugf("process execution reported by the kernel: pid=%d cgroup=%d comm=%s", pid, cgroupID, comm) + + if c.processOP.ShouldMonitorExecuting(&api.ProcessExecuteContext{Pid: pid, CgroupID: cgroupID, Comm: comm}) { + // the finder registered it; AddNewProcess arrives through the listener and records the + // verdict, so there is nothing to undo here + return + } + + // Stop tracing it. Whether this counts as a positive rejection decides if its already buffered + // access logs may be sent, so be careful about the difference: a process we could still see is + // one we truly judged, while a process that vanished before we looked was never judged at all + // and its logs must not be trusted. + c.updateMonitorStatusForProcess(pid, false) + if !path.Exists(host.GetHostProcInHost(fmt.Sprintf("%d", pid))) { + log.Debugf("the process %d exited before it could be identified, its access logs will be dropped", pid) + return + } + c.recordProcessVerdict(pid, ProcessVerdictRejected) + // Remember the rejection in the kernel so the process stops being reported on every exec/fork. + // Only meaningful in the fallback mode; with the cgroup allowlist the process never gets here. + c.rejectProcessInBPF(pid) +} + +// allowCgroupsOfProcesses lets the kernel report the processes that start inside these processes' +// cgroups, which is how a short-lived sibling - a `kubectl exec`, a CronJob - gets monitored from +// its very first syscall instead of waiting for the next discovery round to notice it. +// +// The pids sharing a cgroup are counted, because a container's cgroup must stay allowed until the +// last of its processes is gone; dropping it when any one process exits would blind the rest. +func (c *ConnectionManager) allowCgroupsOfProcesses(pid int32, processes []api.ProcessInterface) { + if c.cgroupAllowlistMap == nil || c.cgroupSeeder == nil { + return + } + for _, cgroupID := range c.cgroupIDsOf(processes) { + pids := c.allowedCgroups[cgroupID] + if pids == nil { + pids = make(map[int32]bool) + c.allowedCgroups[cgroupID] = pids + if err := c.cgroupAllowlistMap.Update(cgroupID, uint32(1), ebpf.UpdateAny); err != nil { + log.Warnf("failed to allow the cgroup %d, short-lived processes in it will only be "+ + "found by the periodic scan: %v", cgroupID, err) + delete(c.allowedCgroups, cgroupID) + continue + } + log.Debugf("allowing process execution reports from cgroup %d", cgroupID) + } + pids[pid] = true + } +} + +// disallowCgroupsOfProcesses drops a pid from its cgroups and stops the kernel reporting a cgroup +// once nothing monitored is left in it. +func (c *ConnectionManager) disallowCgroupsOfProcesses(pid int32, processes []api.ProcessInterface) { + if c.cgroupAllowlistMap == nil || c.cgroupSeeder == nil { + return + } + for _, cgroupID := range c.cgroupIDsOf(processes) { + pids := c.allowedCgroups[cgroupID] + if pids == nil { + continue + } + delete(pids, pid) + if len(pids) > 0 { + continue + } + delete(c.allowedCgroups, cgroupID) + if err := c.cgroupAllowlistMap.Delete(cgroupID); err != nil && !errors.Is(err, ebpf.ErrKeyNotExist) { + log.Warnf("failed to disallow the cgroup %d: %v", cgroupID, err) + } + } +} + +// cgroupIDsOf resolves the cgroups the given processes live in. Only Kubernetes processes have a +// container to resolve through, and a container whose cgroup is not mapped yet simply yields +// nothing, leaving the periodic scan to cover it. +func (c *ConnectionManager) cgroupIDsOf(processes []api.ProcessInterface) []uint64 { + ids := make([]uint64, 0, 1) + for _, p := range processes { + if p.DetectType() != api.Kubernetes { + continue + } + k8sProcess, ok := p.DetectProcess().(*kubernetes.Process) + if !ok { + continue + } + cgroupID, exist := c.cgroupSeeder.CgroupIDByContainer(k8sProcess.PodContainer().CGroupID()) + if !exist { + continue + } + ids = append(ids, cgroupID) + } + return ids +} + +// rejectProcessInBPF records a process as not worth reporting, so the tracepoints skip it from now +// on. Without it a busy unmonitored process would be re-reported and re-checked forever. +func (c *ConnectionManager) rejectProcessInBPF(pid int32) { + if c.processNotMonitorMap == nil { + return + } + if err := c.processNotMonitorMap.Update(uint32(pid), uint32(1), ebpf.UpdateAny); err != nil { + log.Warnf("failed to record the process %d as not monitored: %v", pid, err) + } +} + +func (c *ConnectionManager) recordProcessVerdict(pid int32, verdict ProcessVerdict) { + c.processVerdicts.Set(pid, verdict, processVerdictKeepTime) +} + +// ProcessVerdictOf reports what was concluded about a process, so a caller can tell a process that +// is genuinely not ours from one we simply have not judged yet. +func (c *ConnectionManager) ProcessVerdictOf(pid uint32) ProcessVerdict { + if c.ProcessIsMonitor(pid) { + return ProcessVerdictMonitored + } + if v, exist := c.processVerdicts.Get(int32(pid)); exist { + return v.(ProcessVerdict) } + return ProcessVerdictUnknown } func (c *ConnectionManager) GetExcludeNamespaces() []string { @@ -778,6 +987,11 @@ func (c *ConnectionManager) AddNewProcess(pid int32, entities []api.ProcessInter } c.monitoringProcesses[pid] = monitorProcesses c.updateMonitorStatusForProcess(pid, true) + c.recordProcessVerdict(pid, ProcessVerdictMonitored) + // Seeding the allowlist here, and nowhere earlier, is deliberate: this is the one point both the + // finder and the monitor filter(exclude namespaces, clusters) have passed, so a cgroup can only + // be allowed once everything agrees its processes are ours to report. + c.allowCgroupsOfProcesses(pid, monitorProcesses) for _, entity := range monitorProcesses { for _, host := range entity.ExposeHosts() { c.localIPWithPid[host] = pid @@ -845,8 +1059,12 @@ func (c *ConnectionManager) RemoveProcess(pid int32, _ []api.ProcessInterface) { c.monitoringProcessLock.Lock() defer c.monitoringProcessLock.Unlock() // delete monitoring process and IP addresses + c.disallowCgroupsOfProcesses(pid, c.monitoringProcesses[pid]) delete(c.monitoringProcesses, pid) c.updateMonitorStatusForProcess(pid, false) + // The process passed every filter while it lived, so its access logs are still worth sending + // once it is gone - a short-lived process being gone is the normal case, not a reason to drop. + c.recordProcessVerdict(pid, ProcessVerdictApproved) c.rebuildLocalIPWithPID() c.printTotalAddressesWithPid("remove monitoring process") for _, l := range c.processListeners { diff --git a/pkg/process/api.go b/pkg/process/api.go index ca119075..7964e46d 100644 --- a/pkg/process/api.go +++ b/pkg/process/api.go @@ -32,6 +32,31 @@ type Operator interface { DeleteListener(listener api.ProcessListener) // ShouldMonitor check the process should be monitored ShouldMonitor(pid int32) bool + // ShouldMonitorExecuting check a process the kernel has just started should be monitored, + // using what the kernel captured about it so the answer does not depend on /proc/ still + // existing - a short-lived process may already be gone by the time it is asked. + ShouldMonitorExecuting(exec *api.ProcessExecuteContext) bool + // ExecutingProcessesWanted reports whether any active finder can act on a process reported the + // moment it starts, and therefore whether reporting them is worth its cost at all. + // + // It is false when every active finder only ever monitors long-lived processes: the VM finder, + // for instance, takes a process only if it holds a listening port, which a process that lives + // for milliseconds never does. Reporting to such a finder buys nothing and costs a /proc read + // per process, so the caller must not ask the kernel for the reports in the first place. + ExecutingProcessesWanted() bool +} + +// CgroupOperator is an optional interface the process module satisfies when it can resolve the +// cgroup of a container, which lets a caller filter processes in kernel space by cgroup id. +// +// It is optional because it needs a cgroup v2 tree the agent can read; a caller must handle a +// missing implementation, or a CgroupResolvable of false, by filtering some other way rather than +// by monitoring nothing. +type CgroupOperator interface { + // CgroupResolvable reports whether cgroup ids can be resolved on this host at all + CgroupResolvable() bool + // CgroupIDByContainer returns the cgroup id of a container, false when it is not(yet) known + CgroupIDByContainer(containerID string) (uint64, bool) } type K8sOperator interface { diff --git a/pkg/process/api/process.go b/pkg/process/api/process.go index 75a09ba2..a4401417 100644 --- a/pkg/process/api/process.go +++ b/pkg/process/api/process.go @@ -86,6 +86,22 @@ type DetectedProcess interface { ExposeHosts() []string } +// ProcessExecuteContext carries what the kernel captured about a process the moment it started, +// for finders that are asked to judge it before the periodic /proc scan would have seen it. +// +// Everything here is read in kernel space precisely because a short-lived process may already be +// gone by the time the event is handled, taking its /proc entry with it. A finder should still +// prefer /proc while it is there - it is richer - and only lean on these fields as the fallback. +type ProcessExecuteContext struct { + // Pid of the process in the host pid namespace + Pid int32 + // CgroupID as reported by bpf_get_current_cgroup_id, which is the inode of the process's + // cgroup v2 directory. Zero when the host could not provide one (cgroup v1). + CgroupID uint64 + // Comm is the kernel's task name, a coarser stand-in for the command line + Comm string +} + // ProcessEntity is related to backend entity concept type ProcessEntity struct { Layer string diff --git a/pkg/process/finders/base/finder.go b/pkg/process/finders/base/finder.go index d4ba8b18..6fd86408 100644 --- a/pkg/process/finders/base/finder.go +++ b/pkg/process/finders/base/finder.go @@ -56,6 +56,19 @@ type ProcessFinder interface { ParseProcessID(process api.DetectedProcess, downstream *v3.EBPFProcessDownstream) string } +// ExecutingProcessFinder is an optional interface a ProcessFinder may implement to judge a process +// the kernel has just started, using what the kernel captured about it. +// +// It exists because ShouldMonitor(pid) can only work while /proc/ is readable, and the whole +// reason to look at a process this early is that it may not live long enough for that. A finder +// that cannot make use of the kernel-side data simply does not implement this, and the manager +// falls back to ShouldMonitor. +type ExecutingProcessFinder interface { + // ShouldMonitorExecuting validates a just-started process and, if true, adds it to the storage + // exactly like ShouldMonitor does. + ShouldMonitorExecuting(exec *api.ProcessExecuteContext) bool +} + // ProcessManager is an API work for help ProcessFinder synchronized process with backend type ProcessManager interface { GetModuleManager() *module.Manager diff --git a/pkg/process/finders/base/template.go b/pkg/process/finders/base/template.go index e73c16a5..4c3d7f1a 100644 --- a/pkg/process/finders/base/template.go +++ b/pkg/process/finders/base/template.go @@ -60,7 +60,14 @@ func NewTemplateRover(manager *module.Manager) *TemplateRover { // NewTemplateProcess is generated the process context for render func NewTemplateProcess(_ *module.Manager, p *process.Process) *TemplateProcess { - return &TemplateProcess{p} + return &TemplateProcess{Process: p} +} + +// NewTemplateProcessWithFallbackName is like NewTemplateProcess, but names the process from +// fallbackName(the kernel task name) when its command line cannot be read - which is what happens +// to a process that has already exited, the very case a short-lived process is reported for. +func NewTemplateProcessWithFallbackName(_ *module.Manager, p *process.Process, fallbackName string) *TemplateProcess { + return &TemplateProcess{Process: p, fallbackName: fallbackName} } type TemplateRover struct { @@ -97,6 +104,26 @@ func (t *TemplateRover) HostName() string { type TemplateProcess struct { *process.Process + // fallbackName stands in for the command line when /proc no longer holds one + fallbackName string +} + +// Cmdline is the process's command line, falling back to the kernel task name when /proc cannot +// supply one. It shadows the embedded gopsutil method so that every caller - the periodic scan and +// the kernel-driven one alike - names a process by the same rule, otherwise the same pid could be +// registered under two different names and flip between detected and dead. +// +// An empty command line falls back too: a zombie still has a readable /proc entry but an empty +// cmdline, which would otherwise name the process "". +func (p *TemplateProcess) Cmdline() (string, error) { + cmdline, err := p.Process.Cmdline() + if err == nil && cmdline != "" { + return cmdline, nil + } + if p.fallbackName != "" { + return p.fallbackName, nil + } + return cmdline, err } // ExeFilePath Execute file path diff --git a/pkg/process/finders/kubernetes/finder.go b/pkg/process/finders/kubernetes/finder.go index 2edfbbfa..f440f359 100644 --- a/pkg/process/finders/kubernetes/finder.go +++ b/pkg/process/finders/kubernetes/finder.go @@ -51,6 +51,7 @@ import ( "github.com/apache/skywalking-rover/pkg/logger" "github.com/apache/skywalking-rover/pkg/process/api" "github.com/apache/skywalking-rover/pkg/process/finders/base" + "github.com/apache/skywalking-rover/pkg/tools/cgroup" "github.com/apache/skywalking-rover/pkg/tools/host" ) @@ -59,10 +60,42 @@ var log = logger.GetLogger("process", "finder", "kubernetes") var ( kubepodsRegex = regexp.MustCompile(`cri-containerd-(?P\w+)\.scope`) openShiftPodsRegex = regexp.MustCompile(`crio-(?P\w+)\.scope`) + dockerPodsRegex = regexp.MustCompile(`docker-(?P\w+)\.scope`) ipExistTimeout = time.Minute * 10 ipSearchParallel = 10 ) +// containerIDFromCgroupDir turns the base name of a cgroup directory into the id of the container +// it holds, or "" when it holds none. It mirrors the naming GetProcessCGroup already parses out of +// /proc//cgroup, so that a container resolved by walking the tree and one resolved by reading +// a process's cgroup line come out with the same id. +func containerIDFromCgroupDir(dirName string) string { + for _, re := range []*regexp.Regexp{kubepodsRegex, openShiftPodsRegex, dockerPodsRegex} { + if m := re.FindStringSubmatch(dirName); len(m) > 1 { + return m[1] + } + } + // the cgroupfs driver, unlike the systemd one, names the directory after the container id + if isContainerID(dirName) { + return dirName + } + return "" +} + +// isContainerID reports whether a bare cgroup directory name is a container id, which keeps the +// host's own slices(system.slice, init.scope, ...) out of the mapping. +func isContainerID(name string) bool { + if len(name) < 12 { + return false + } + for _, c := range name { + if (c < '0' || c > '9') && (c < 'a' || c > 'f') { + return false + } + } + return true +} + type ProcessFinder struct { conf *Config @@ -74,6 +107,9 @@ type ProcessFinder struct { stopChan chan struct{} processCache *lru.Cache + // cgroupResolver maps cgroup ids to containers; nil when the host has no usable cgroup v2 tree + cgroupResolver *cgroup.Resolver + // k8s clients k8sConfig *rest.Config registry Registry @@ -124,6 +160,24 @@ func (f *ProcessFinder) InitWithRegistry(ctx context.Context, conf base.FinderBa } f.processCache = processCache + // The cgroup mapping is what lets an already-exited process still be attributed to its pod, and + // what lets the access log module filter in kernel space. It needs a cgroup v2 tree; on a v1 + // host the finder simply keeps working the way it always has, through the periodic /proc scan. + cgroupMountPoint := cgroup.DefaultMountPoint() + if cgroup.Available(cgroupMountPoint) { + f.cgroupResolver = cgroup.NewResolver(cgroupMountPoint, containerIDFromCgroupDir) + if err := f.cgroupResolver.Refresh(); err != nil { + log.Warnf("cannot read the cgroup tree at %s, falling back to process discovery alone: %v", + cgroupMountPoint, err) + f.cgroupResolver = nil + } else { + log.Infof("resolving containers by cgroup id, %d container cgroups mapped", f.cgroupResolver.Size()) + } + } else { + log.Infof("no cgroup v2 hierarchy at %s, containers will only be discovered by the periodic scan", + cgroupMountPoint) + } + return nil } @@ -172,6 +226,10 @@ func (f *ProcessFinder) Start() { } func (f *ProcessFinder) analyzeProcesses() error { + // keep the cgroup mapping in step with the containers that come and go on this node; this scan + // reads /proc directly, so it can never be held back by a stale mapping + f.RefreshCgroupResolver() + // find out all containers containers := f.registry.BuildPodContainers() if len(containers) == 0 { @@ -240,6 +298,14 @@ func (f *ProcessFinder) buildProcess(p *process.Process, detectedProcesses []api } func (f *ProcessFinder) BuildProcesses(p *process.Process, pc *PodContainer) ([]*Process, error) { + return f.BuildProcessesWithFallbackName(p, pc, "") +} + +// BuildProcessesWithFallbackName is BuildProcesses for a process whose /proc entry may already be +// gone: fallbackName(the kernel task name) then stands in for the command line. Passing "" gives +// exactly the behaviour of BuildProcesses. +func (f *ProcessFinder) BuildProcessesWithFallbackName(p *process.Process, pc *PodContainer, + fallbackName string) ([]*Process, error) { // find builder builders := make([]*ProcessBuilder, 0) for _, b := range f.conf.Analyzers { @@ -258,8 +324,16 @@ func (f *ProcessFinder) BuildProcesses(p *process.Process, pc *PodContainer) ([] } cmdline, err := p.Cmdline() - if err != nil { - return nil, err + if err != nil || cmdline == "" { + // the process is gone(or is a zombie, which keeps a readable but empty cmdline); the + // kernel task name is all that is left to identify it by + if fallbackName == "" { + if err != nil { + return nil, fmt.Errorf("cannot read the command line of process %d: %w", p.Pid, err) + } + return nil, fmt.Errorf("the command line of process %d is empty", p.Pid) + } + cmdline = fallbackName } // build process @@ -267,9 +341,9 @@ func (f *ProcessFinder) BuildProcesses(p *process.Process, pc *PodContainer) ([] for _, builder := range builders { entity := &api.ProcessEntity{} entity.Layer = builder.Layer - entity.ServiceName, err = f.buildEntity(err, p, pc, builder.ServiceNameBuilder) - entity.InstanceName, err = f.buildEntity(err, p, pc, builder.InstanceNameBuilder) - entity.ProcessName, err = f.buildEntity(err, p, pc, builder.ProcessNameBuilder) + entity.ServiceName, err = f.buildEntity(nil, p, pc, builder.ServiceNameBuilder, fallbackName) + entity.InstanceName, err = f.buildEntity(err, p, pc, builder.InstanceNameBuilder, fallbackName) + entity.ProcessName, err = f.buildEntity(err, p, pc, builder.ProcessNameBuilder, fallbackName) entity.Labels = builder.Labels if err != nil { return nil, err @@ -284,11 +358,12 @@ func (f *ProcessFinder) BuildProcesses(p *process.Process, pc *PodContainer) ([] return processes, nil } -func (f *ProcessFinder) buildEntity(err error, ps *process.Process, pc *PodContainer, entity *base.TemplateBuilder) (string, error) { +func (f *ProcessFinder) buildEntity(err error, ps *process.Process, pc *PodContainer, entity *base.TemplateBuilder, + fallbackName string) (string, error) { if err != nil { return "", err } - return renderTemplate(entity, ps, pc, f) + return renderTemplate(entity, ps, pc, f, fallbackName) } func (f *ProcessFinder) GetProcessCGroup(pid int32) ([]string, error) { @@ -423,6 +498,81 @@ func (f *ProcessFinder) ShouldMonitor(pid int32) bool { return true } +// ShouldMonitorExecuting judges a process the kernel has just started. +// +// It prefers the ordinary /proc based path, which is richer and is exactly what the periodic scan +// does. Only when /proc has nothing left to say - the process already exited, which is the whole +// reason it is worth catching this early - does it fall back to what the kernel handed us: the +// cgroup id identifies the container, and the task name stands in for the command line. +func (f *ProcessFinder) ShouldMonitorExecuting(exec *api.ProcessExecuteContext) bool { + if f.ShouldMonitor(exec.Pid) { + return true + } + if exec.CgroupID == 0 || f.cgroupResolver == nil { + // no kernel side identity to fall back on(cgroup v1, or the tree could not be walked) + return false + } + containerID, exist := f.cgroupResolver.ContainerByCgroupID(exec.CgroupID) + if !exist { + return false + } + pc, exist := f.registry.BuildPodContainers()[containerID] + if !exist || pc == nil { + return false + } + // the process is gone, so nothing can be read from /proc; a bare Process carries the pid, which + // is all the entity building still needs from it. + processes, err := f.BuildProcessesWithFallbackName(&process.Process{Pid: exec.Pid}, pc, exec.Comm) + if err != nil || len(processes) == 0 { + log.Debugf("cannot build the exited process %d in container %s: %v", exec.Pid, containerID, err) + return false + } + detected := make([]api.DetectedProcess, 0, len(processes)) + for _, p := range processes { + detected = append(detected, p) + } + f.manager.AddDetectedProcess(detected) + return true +} + +// RefreshCgroupResolver rebuilds the cgroup id -> container mapping. The caller drives it from the +// periodic scan, which reads /proc and is therefore never blocked by whatever the mapping says. +func (f *ProcessFinder) RefreshCgroupResolver() { + if f.cgroupResolver == nil { + return + } + if err := f.cgroupResolver.Refresh(); err != nil { + log.Warnf("cannot refresh the cgroup mapping, already started short-lived processes may not "+ + "be attributed until the next refresh: %v", err) + } +} + +// CgroupResolvable reports whether this host lets cgroup ids be resolved at all, which decides +// whether a caller may filter by cgroup or has to filter some other way. +func (f *ProcessFinder) CgroupResolvable() bool { + return f.cgroupResolver != nil +} + +// ContainerByCgroupID names the container a cgroup id belongs to. This is the direction that lets +// an already-exited process still be attributed: the cgroup id comes with the kernel event, so +// nothing has to be read from the process's /proc entry. +func (f *ProcessFinder) ContainerByCgroupID(cgroupID uint64) (string, bool) { + if f.cgroupResolver == nil { + return "", false + } + return f.cgroupResolver.ContainerByCgroupID(cgroupID) +} + +// CgroupIDByContainer exposes the cgroup id of a container so the access log module can seed the +// kernel side allowlist with it. The second result is false when the container's cgroup is not(yet) +// mapped, which the caller must read as "cannot filter on this one", never as "monitor nothing". +func (f *ProcessFinder) CgroupIDByContainer(containerID string) (uint64, bool) { + if f.cgroupResolver == nil { + return 0, false + } + return f.cgroupResolver.CgroupIDByContainer(containerID) +} + func (f *ProcessFinder) IsPodIP(ip string) (bool, error) { val, exist := f.podIPChecker.Get(ip) if exist { diff --git a/pkg/process/finders/kubernetes/template.go b/pkg/process/finders/kubernetes/template.go index 617f82b9..ab73280b 100644 --- a/pkg/process/finders/kubernetes/template.go +++ b/pkg/process/finders/kubernetes/template.go @@ -59,11 +59,12 @@ func executeFilter(builders []*base.TemplateBuilder, p *process.Process, pc *Pod return true, nil } -func renderTemplate(builder *base.TemplateBuilder, p *process.Process, pc *PodContainer, finder *ProcessFinder) (string, error) { +func renderTemplate(builder *base.TemplateBuilder, p *process.Process, pc *PodContainer, finder *ProcessFinder, + fallbackName string) (string, error) { moduleManager := finder.manager.GetModuleManager() return builder.Execute(&EntityRenderContext{ Rover: base.NewTemplateRover(moduleManager), - Process: base.NewTemplateProcess(moduleManager, p), + Process: base.NewTemplateProcessWithFallbackName(moduleManager, p, fallbackName), Pod: &TemplatePod{pc, finder}, Container: &TemplateContainer{pc}, }) diff --git a/pkg/process/finders/manager.go b/pkg/process/finders/manager.go index a2524162..dc8eb37e 100644 --- a/pkg/process/finders/manager.go +++ b/pkg/process/finders/manager.go @@ -167,3 +167,67 @@ func (m *ProcessManager) ShouldMonitor(pid int32) bool { } return monitor } + +// cgroupResolvingFinder is what a finder must offer for cgroup ids to be resolvable; only the +// Kubernetes finder knows about containers, so only it implements this. +type cgroupResolvingFinder interface { + CgroupResolvable() bool + CgroupIDByContainer(containerID string) (uint64, bool) +} + +// CgroupResolvable reports whether any finder can resolve cgroup ids on this host. +func (m *ProcessManager) CgroupResolvable() bool { + for _, finder := range m.finders { + if resolver, ok := finder.(cgroupResolvingFinder); ok && resolver.CgroupResolvable() { + return true + } + } + return false +} + +// CgroupIDByContainer returns the cgroup id of a container as resolved by whichever finder knows it. +func (m *ProcessManager) CgroupIDByContainer(containerID string) (uint64, bool) { + for _, finder := range m.finders { + if resolver, ok := finder.(cgroupResolvingFinder); ok { + if id, exist := resolver.CgroupIDByContainer(containerID); exist { + return id, true + } + } + } + return 0, false +} + +// ExecutingProcessesWanted reports whether any *active* finder implements ExecutingProcessFinder, +// which is what makes reporting a process the instant it starts worth its cost. +// +// Implementing that interface is the finder's way of saying it can act on a process that may +// already be gone; a finder without it can only fall back to the /proc based check, which for a +// short-lived process answers "no" after having paid for the lookup. m.finders holds only the +// active finders, so an inactive one cannot keep the reports switched on. +func (m *ProcessManager) ExecutingProcessesWanted() bool { + for _, finder := range m.finders { + if _, ok := finder.(base.ExecutingProcessFinder); ok { + return true + } + } + return false +} + +// ShouldMonitorExecuting judges a process the kernel has just started. Finders that can use the +// kernel-side context take it; the rest fall back to the /proc based check, which is all they could +// have done anyway. +func (m *ProcessManager) ShouldMonitorExecuting(exec *api.ProcessExecuteContext) bool { + monitor := false + for _, finder := range m.finders { + if executing, ok := finder.(base.ExecutingProcessFinder); ok { + if executing.ShouldMonitorExecuting(exec) { + monitor = true + } + continue + } + if finder.ShouldMonitor(exec.Pid) { + monitor = true + } + } + return monitor +} diff --git a/pkg/process/finders/manager_executing_test.go b/pkg/process/finders/manager_executing_test.go new file mode 100644 index 00000000..23aa6363 --- /dev/null +++ b/pkg/process/finders/manager_executing_test.go @@ -0,0 +1,107 @@ +// Licensed to Apache Software Foundation (ASF) under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Apache Software Foundation (ASF) licenses this file to you under +// the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package finders + +import ( + "context" + "testing" + + commonv3 "skywalking.apache.org/repo/goapi/collect/common/v3" + v3 "skywalking.apache.org/repo/goapi/collect/ebpf/profiling/process/v3" + + "github.com/apache/skywalking-rover/pkg/process/api" + "github.com/apache/skywalking-rover/pkg/process/finders/base" +) + +// plainFinder monitors only long-lived processes, the way the VM finder does: it has no use for a +// process reported the instant it starts, so it does not implement ExecutingProcessFinder. +type plainFinder struct{} + +func (p *plainFinder) Init(_ context.Context, _ base.FinderBaseConfig, _ base.ProcessManager) error { + return nil +} +func (p *plainFinder) Start() {} +func (p *plainFinder) Stop() error { return nil } +func (p *plainFinder) DetectType() api.ProcessDetectType { return api.Kubernetes } +func (p *plainFinder) ShouldMonitor(_ int32) bool { return false } +func (p *plainFinder) ValidateProcessIsSame(_, _ api.DetectedProcess) bool { return false } +func (p *plainFinder) BuildEBPFProcess(_ *base.BuildEBPFProcessContext, _ api.DetectedProcess) *v3.EBPFProcessProperties { + return nil +} +func (p *plainFinder) BuildNecessaryProperties(_ api.DetectedProcess) []*commonv3.KeyStringValuePair { + return nil +} +func (p *plainFinder) ParseProcessID(_ api.DetectedProcess, _ *v3.EBPFProcessDownstream) string { + return "" +} + +// executingFinder additionally takes processes as they start, the way the Kubernetes finder does. +type executingFinder struct{ plainFinder } + +func (e *executingFinder) ShouldMonitorExecuting(_ *api.ProcessExecuteContext) bool { return true } + +// TestExecutingProcessesWanted pins the gate that decides whether the kernel is asked to report +// processes at all. +// +// The cost of getting this wrong is asymmetric and silent. Reporting when nothing can use the +// reports is what made these tracepoints unaffordable before: a finder that only monitors +// long-lived processes rejects every reported one *after* a /proc read, paying the full price for +// no coverage. Nothing fails; the agent just quietly burns CPU on every process the machine starts. +func TestExecutingProcessesWanted(t *testing.T) { + tests := []struct { + name string + finders []base.ProcessFinder + want bool + }{ + { + name: "no finder can use the reports - do not ask for them", + finders: []base.ProcessFinder{&plainFinder{}}, + want: false, + }, + { + name: "a finder takes processes as they start", + finders: []base.ProcessFinder{&executingFinder{}}, + want: true, + }, + { + name: "one of several can use them", + finders: []base.ProcessFinder{&plainFinder{}, &executingFinder{}}, + want: true, + }, + { + name: "no active finder at all", + finders: nil, + want: false, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + m := &ProcessManager{finders: make(map[base.FinderBaseConfig]base.ProcessFinder)} + for i, f := range tt.finders { + m.finders[&stubConfig{id: i}] = f + } + if got := m.ExecutingProcessesWanted(); got != tt.want { + t.Fatalf("ExecutingProcessesWanted() = %v, want %v", got, tt.want) + } + }) + } +} + +type stubConfig struct{ id int } + +func (s *stubConfig) ActiveFinder() bool { return true } diff --git a/pkg/process/module.go b/pkg/process/module.go index 6a62632b..8081e9da 100644 --- a/pkg/process/module.go +++ b/pkg/process/module.go @@ -103,6 +103,22 @@ func (m *Module) ShouldMonitor(pid int32) bool { return m.manager.ShouldMonitor(pid) } +func (m *Module) ShouldMonitorExecuting(exec *api.ProcessExecuteContext) bool { + return m.manager.ShouldMonitorExecuting(exec) +} + +func (m *Module) ExecutingProcessesWanted() bool { + return m.manager.ExecutingProcessesWanted() +} + +func (m *Module) CgroupResolvable() bool { + return m.manager.CgroupResolvable() +} + +func (m *Module) CgroupIDByContainer(containerID string) (uint64, bool) { + return m.manager.CgroupIDByContainer(containerID) +} + func (m *Module) NodeName() string { return m.config.Kubernetes.NodeName } diff --git a/pkg/tools/cgroup/resolver.go b/pkg/tools/cgroup/resolver.go new file mode 100644 index 00000000..0e299414 --- /dev/null +++ b/pkg/tools/cgroup/resolver.go @@ -0,0 +1,199 @@ +// Licensed to Apache Software Foundation (ASF) under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Apache Software Foundation (ASF) licenses this file to you under +// the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Package cgroup resolves the cgroup ids that eBPF programs report(through +// bpf_get_current_cgroup_id) to the container they belong to, and back. +package cgroup + +import ( + "fmt" + "io/fs" + "os" + "path/filepath" + "strings" + "sync" + "syscall" + + "github.com/apache/skywalking-rover/pkg/logger" + "github.com/apache/skywalking-rover/pkg/tools/host" +) + +var log = logger.GetLogger("tools", "cgroup") + +// DefaultMountPoint is where the unified(v2) cgroup hierarchy lives, as seen from inside the agent +// container. +// +// It has to be the *host's* /sys: the agent's own only exposes its own cgroup subtree, and the +// containers whose ids we need are not in it. Where the host's /sys is mounted is a property of the +// deployment - the helm chart bind-mounts it at /sys, an environment where the node is itself a +// container may put it elsewhere - so the prefix comes from ROVER_HOST_SYS_MAPPING rather than +// being assumed, exactly as the /proc prefix does. +func DefaultMountPoint() string { + return host.GetHostSysInHost("fs/cgroup") +} + +const ( + // controllersFile only exists on the unified(v2) hierarchy, which makes its presence the + // probe for "this host runs cgroup v2". On a v1-only host bpf_get_current_cgroup_id() reports + // the id from the unified hierarchy, where the containers are not, so every lookup would miss + // and the caller must disable the whole cgroup based path instead. + controllersFile = "cgroup.controllers" +) + +// There is deliberately no bound on how deep the walk goes, because there is no depth to bound it +// to. Where a container's cgroup sits is not fixed by anything: +// +// - the cgroup driver decides both the naming and the number of levels(systemd: +// kubepods.slice/kubepods-.slice/kubepods--pod.slice/.scope, cgroupfs: +// kubepods//pod/) +// - the QoS class adds a level for burstable/besteffort that guaranteed pods do not have +// - wherever the kubelet itself runs inside a container(kind, docker-in-docker) the whole +// hierarchy hangs under that container's cgroup, e.g. /system.slice/docker-.scope/ +// kubelet.slice/kubelet-kubepods.slice/..., which is three levels further down +// - the kubelet can be pointed at an arbitrary cgroup root +// +// If the layout were knowable the path could simply be built and stat'ed, and none of this would +// exist. It is not, which is why the tree is walked and matched by name - so what makes a directory +// a container is its name, never how deep it happens to be. An earlier version guessed a depth that +// fit a bare node; on a nested one it matched nothing at all and did so silently. +// +// The walk itself is cheap: the tree holds at most a few thousand directories, only their names are +// read, and a directory is stat'ed only once its name says it is a container. + +// NameNormalizer turns the base name of a cgroup directory into the id of the container it holds, +// or "" when the directory does not belong to a container. It is injected rather than implemented +// here because the naming is container runtime specific and the process finder already owns those +// rules. +type NameNormalizer func(dirName string) string + +// Resolver maps cgroup ids to container ids and back, by walking the host's cgroup v2 tree. +// +// The kernel identifies a cgroup v2 by the inode of its directory, and that inode is precisely +// what bpf_get_current_cgroup_id() returns, so the mapping is built by stat'ing the tree. +// +// Reading the tree from the mounted cgroupfs - rather than from /proc//cgroup - is the whole +// point: the paths in /proc//cgroup are rendered relative to the cgroup namespace of the +// *reading* process, so an agent living in its own namespace reads unresolvable paths such as +// "0::/../", whereas a bind-mounted host cgroupfs exposes the full tree with real +// inodes no matter which namespace the reader sits in. +type Resolver struct { + mountPoint string + normalize NameNormalizer + + mu sync.RWMutex + idByContainer map[string]uint64 + containerByID map[uint64]string +} + +// NewResolver builds a Resolver over mountPoint(use DefaultMountPoint unless testing). It does not +// walk anything yet; call Refresh for that. +func NewResolver(mountPoint string, normalize NameNormalizer) *Resolver { + return &Resolver{ + mountPoint: mountPoint, + normalize: normalize, + idByContainer: make(map[string]uint64), + containerByID: make(map[uint64]string), + } +} + +// Available reports whether mountPoint holds a usable unified(v2) hierarchy. A caller must treat a +// false here as "the cgroup id of a process cannot be resolved on this host" and fall back to +// whatever it did before, never as a reason to fail. +func Available(mountPoint string) bool { + st, err := os.Stat(filepath.Join(mountPoint, controllersFile)) + return err == nil && !st.IsDir() +} + +// Refresh rebuilds the mapping from the current state of the tree. It is meant to be driven by the +// periodic process discovery: a cgroup that appears between two refreshes is simply resolved on the +// next one, and until then the caller behaves as it did before this package existed. +func (r *Resolver) Refresh() error { + idByContainer := make(map[string]uint64) + containerByID := make(map[uint64]string) + + root := filepath.Clean(r.mountPoint) + scanned := 0 + err := filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error { + if err != nil { + // cgroups come and go while we walk; a vanished directory is normal churn on a busy + // node, so skip it rather than abandoning the whole refresh. + return nil //nolint:nilerr // deliberate: keep walking past transient errors + } + if !d.IsDir() { + return nil + } + scanned++ + container := r.normalize(d.Name()) + if container == "" { + return nil + } + info, err := d.Info() + if err != nil { + return nil //nolint:nilerr // the directory disappeared mid-walk + } + stat, ok := info.Sys().(*syscall.Stat_t) + if !ok { + return nil + } + id := uint64(stat.Ino) + idByContainer[container] = id + containerByID[id] = container + // The path is logged with the id because this mapping is the one thing that has to agree + // with what the kernel reports: bpf_get_current_cgroup_id() returns the inode of exactly + // this directory. If the two ever disagree, the id and the path it was read from are what + // is needed to see why. + log.Debugf("cgroup mapping: id=%d container=%s path=%s", + id, container, strings.TrimPrefix(path, root)) + return nil + }) + if err != nil { + return fmt.Errorf("walking the cgroup tree at %s: %w", root, err) + } + log.Debugf("cgroup walk over %s finished: %d directories scanned, %d containers mapped", + root, scanned, len(containerByID)) + + r.mu.Lock() + defer r.mu.Unlock() + r.idByContainer = idByContainer + r.containerByID = containerByID + return nil +} + +// CgroupIDByContainer returns the cgroup id of a container, for seeding the BPF allowlist. +func (r *Resolver) CgroupIDByContainer(containerID string) (uint64, bool) { + r.mu.RLock() + defer r.mu.RUnlock() + id, exist := r.idByContainer[containerID] + return id, exist +} + +// ContainerByCgroupID returns the container a cgroup id belongs to. This is the direction that lets +// a process which has already exited still be attributed: the cgroup id arrives with the kernel +// event, so nothing has to be read from the process's /proc entry. +func (r *Resolver) ContainerByCgroupID(id uint64) (string, bool) { + r.mu.RLock() + defer r.mu.RUnlock() + container, exist := r.containerByID[id] + return container, exist +} + +// Size reports how many container cgroups are currently mapped, for logging. +func (r *Resolver) Size() int { + r.mu.RLock() + defer r.mu.RUnlock() + return len(r.containerByID) +} diff --git a/pkg/tools/cgroup/resolver_test.go b/pkg/tools/cgroup/resolver_test.go new file mode 100644 index 00000000..c793e78d --- /dev/null +++ b/pkg/tools/cgroup/resolver_test.go @@ -0,0 +1,249 @@ +// Licensed to Apache Software Foundation (ASF) under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Apache Software Foundation (ASF) licenses this file to you under +// the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package cgroup + +import ( + "fmt" + "os" + "path/filepath" + "regexp" + "strings" + "syscall" + "testing" +) + +// containerdScope mirrors the systemd-driver naming the Kubernetes finder already parses. +var containerdScope = regexp.MustCompile(`cri-containerd-(\w+)\.scope`) + +// testNormalizer applies the same rules the Kubernetes finder uses: a systemd scope carries the +// container id inside its name, while the cgroupfs driver names the directory after the id itself. +func testNormalizer(dirName string) string { + if m := containerdScope.FindStringSubmatch(dirName); len(m) > 1 { + return m[1] + } + if len(dirName) == 64 && !strings.ContainsAny(dirName, ".-") { + return dirName + } + return "" +} + +// inodeOf reads the inode the kernel gave a directory. On a real cgroup v2 tree this is the value +// bpf_get_current_cgroup_id() reports, which is the equivalence the Resolver relies on. +func inodeOf(t *testing.T, path string) uint64 { + t.Helper() + info, err := os.Stat(path) + if err != nil { + t.Fatalf("stat %s: %v", path, err) + } + stat, ok := info.Sys().(*syscall.Stat_t) + if !ok { + t.Fatalf("no Stat_t for %s", path) + } + return uint64(stat.Ino) +} + +func mkdirAll(t *testing.T, parts ...string) string { + t.Helper() + path := filepath.Join(parts...) + if err := os.MkdirAll(path, 0o755); err != nil { + t.Fatalf("mkdir %s: %v", path, err) + } + return path +} + +// buildTree lays out a cgroup v2 tree shaped like a real node: the kubepods slices a container +// lives under, plus host slices that must never be mistaken for containers. +func buildTree(t *testing.T) (root, systemdContainer, cgroupfsContainer string) { + t.Helper() + root = t.TempDir() + if err := os.WriteFile(filepath.Join(root, controllersFile), []byte("cpu memory\n"), 0o600); err != nil { + t.Fatalf("write controllers: %v", err) + } + + // systemd driver: kubepods.slice/kubepods-burstable.slice/kubepods-burstable-pod.slice/cri-containerd-.scope + systemdContainer = strings.Repeat("a", 64) + mkdirAll(t, root, "kubepods.slice", "kubepods-burstable.slice", + "kubepods-burstable-pod1234.slice", "cri-containerd-"+systemdContainer+".scope") + + // cgroupfs driver: kubepods/besteffort/pod/ + cgroupfsContainer = strings.Repeat("b", 64) + mkdirAll(t, root, "kubepods", "besteffort", "pod5678", cgroupfsContainer) + + // host slices - these must not show up as containers + mkdirAll(t, root, "system.slice", "kubelet.service") + mkdirAll(t, root, "init.scope") + return root, systemdContainer, cgroupfsContainer +} + +func TestResolverMapsContainersToTheirCgroupInode(t *testing.T) { + root, systemdContainer, cgroupfsContainer := buildTree(t) + + r := NewResolver(root, testNormalizer) + if err := r.Refresh(); err != nil { + t.Fatalf("refresh: %v", err) + } + + // exactly the two containers, none of the host slices + if got := r.Size(); got != 2 { + t.Fatalf("expected 2 container cgroups, got %d", got) + } + + cases := []struct { + name string + container string + dir string + }{ + {"systemd driver", systemdContainer, filepath.Join(root, "kubepods.slice", "kubepods-burstable.slice", + "kubepods-burstable-pod1234.slice", "cri-containerd-"+systemdContainer+".scope")}, + {"cgroupfs driver", cgroupfsContainer, filepath.Join(root, "kubepods", "besteffort", "pod5678", cgroupfsContainer)}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + want := inodeOf(t, c.dir) + + // forward: container -> cgroup id, which is what seeds the BPF allowlist + id, exist := r.CgroupIDByContainer(c.container) + if !exist { + t.Fatalf("container %s not mapped", c.container) + } + if id != want { + t.Fatalf("cgroup id = %d, want the directory inode %d", id, want) + } + + // reverse: cgroup id -> container, which is what attributes an already-exited process + container, exist := r.ContainerByCgroupID(want) + if !exist { + t.Fatalf("cgroup id %d not mapped back", want) + } + if container != c.container { + t.Fatalf("container = %s, want %s", container, c.container) + } + }) + } +} + +func TestResolverIgnoresHostSlices(t *testing.T) { + root, _, _ := buildTree(t) + r := NewResolver(root, testNormalizer) + if err := r.Refresh(); err != nil { + t.Fatalf("refresh: %v", err) + } + // a host slice's inode must not resolve to any container: this is what keeps kubelet and + // friends out of the allowlist. + hostInode := inodeOf(t, filepath.Join(root, "system.slice", "kubelet.service")) + if container, exist := r.ContainerByCgroupID(hostInode); exist { + t.Fatalf("host slice resolved to container %s", container) + } +} + +// TestResolverFindsNestedKubeletLayout is the regression guard for the layout that appears wherever +// the kubelet itself runs inside a container(kind, docker-in-docker): the pod cgroups hang under +// the node container's own cgroup, which puts a container seven levels down instead of four. A +// depth bound sized for a bare node maps nothing here, and the failure is silent - the mapping is +// simply empty, so nothing is ever attributed. +func TestResolverFindsNestedKubeletLayout(t *testing.T) { + root := t.TempDir() + if err := os.WriteFile(filepath.Join(root, controllersFile), []byte("cpu memory\n"), 0o600); err != nil { + t.Fatalf("write controllers: %v", err) + } + container := strings.Repeat("d", 64) + dir := mkdirAll(t, root, "system.slice", "docker-"+strings.Repeat("e", 64)+".scope", + "kubelet.slice", "kubelet-kubepods.slice", "kubelet-kubepods-burstable.slice", + "kubelet-kubepods-burstable-pod6fc4a17a_9b0a_4ab4_833e_7004a16f89ce.slice", + "cri-containerd-"+container+".scope") + + r := NewResolver(root, testNormalizer) + if err := r.Refresh(); err != nil { + t.Fatalf("refresh: %v", err) + } + id, exist := r.CgroupIDByContainer(container) + if !exist { + t.Fatal("a container under a nested kubelet hierarchy must still be mapped") + } + if want := inodeOf(t, dir); id != want { + t.Fatalf("cgroup id = %d, want the directory inode %d", id, want) + } +} + +// TestResolverIsIndifferentToDepth pins the invariant that replaced the old depth bound: what makes +// a directory a container is its name, never how deep it sits. There is no layout to bound the walk +// to - the driver, the QoS class and a containerised kubelet each move things around - so a resolver +// that stops at some depth maps nothing on the layouts it did not anticipate, silently. +func TestResolverIsIndifferentToDepth(t *testing.T) { + root := t.TempDir() + if err := os.WriteFile(filepath.Join(root, controllersFile), []byte("cpu memory\n"), 0o600); err != nil { + t.Fatalf("write controllers: %v", err) + } + // far deeper than any real hierarchy, to show the walk is not the thing deciding + container := strings.Repeat("f", 64) + parts := []string{root} + for i := 0; i < 20; i++ { + parts = append(parts, fmt.Sprintf("nest%d", i)) + } + dir := mkdirAll(t, append(parts, "cri-containerd-"+container+".scope")...) + + r := NewResolver(root, testNormalizer) + if err := r.Refresh(); err != nil { + t.Fatalf("refresh: %v", err) + } + id, exist := r.CgroupIDByContainer(container) + if !exist { + t.Fatal("a container must be mapped no matter how deep it sits") + } + if want := inodeOf(t, dir); id != want { + t.Fatalf("cgroup id = %d, want the directory inode %d", id, want) + } +} + +func TestAvailableProbesUnifiedHierarchy(t *testing.T) { + root, _, _ := buildTree(t) + if !Available(root) { + t.Fatal("a tree with cgroup.controllers must be reported available") + } + + // a v1 tree has no cgroup.controllers at its root, and there the cgroup id reported by BPF + // would never match anything, so the caller must fall back instead. + v1 := t.TempDir() + mkdirAll(t, v1, "memory") + if Available(v1) { + t.Fatal("a tree without cgroup.controllers must not be reported available") + } +} + +// TestRefreshReplacesPreviousState makes sure a container that goes away stops resolving, rather +// than lingering and attributing a future process to a dead pod. +func TestRefreshReplacesPreviousState(t *testing.T) { + root, systemdContainer, _ := buildTree(t) + r := NewResolver(root, testNormalizer) + if err := r.Refresh(); err != nil { + t.Fatalf("refresh: %v", err) + } + if _, exist := r.CgroupIDByContainer(systemdContainer); !exist { + t.Fatal("container should be mapped before removal") + } + + if err := os.RemoveAll(filepath.Join(root, "kubepods.slice")); err != nil { + t.Fatalf("remove: %v", err) + } + if err := r.Refresh(); err != nil { + t.Fatalf("refresh after removal: %v", err) + } + if _, exist := r.CgroupIDByContainer(systemdContainer); exist { + t.Fatal("a removed container must stop resolving after a refresh") + } +} diff --git a/pkg/tools/host/file.go b/pkg/tools/host/file.go index b3762883..0513ffc0 100644 --- a/pkg/tools/host/file.go +++ b/pkg/tools/host/file.go @@ -26,6 +26,7 @@ var ( hostProcMappingPath string hostEtcMappingPath string hostVarLogPodsMappingPath string + hostSysMappingPath string ) func init() { @@ -36,6 +37,7 @@ func init() { } hostEtcMappingPath = os.Getenv("ROVER_HOST_ETC_MAPPING") hostVarLogPodsMappingPath = os.Getenv("ROVER_HOST_VAR_LOG_PODS_MAPPING") + hostSysMappingPath = os.Getenv("ROVER_HOST_SYS_MAPPING") } func GetHostProcInHost(procSubPath string) string { @@ -65,6 +67,22 @@ func GetHostVarLogPodsInHost(subPath string) string { return cleanPath("/var/log/pods/" + subPath) } +// GetHostSysInHost resolves a path under the host's /sys as seen from inside the agent container. +// +// The cgroup tree lives under here, and reading the *host's* copy of it is what makes a cgroup id +// resolvable at all: the agent's own /sys shows only its own cgroup subtree. Where the host's /sys +// is mounted is a property of the deployment, not something that can be assumed - the helm chart +// bind-mounts the host /sys at /sys, while an environment where the node is itself a container +// (kind) has to reach it through another prefix - so the mount point is injected through +// ROVER_HOST_SYS_MAPPING(the same pattern as ROVER_HOST_PROC_MAPPING); when unset the real host +// path /sys is used. +func GetHostSysInHost(subPath string) string { + if hostSysMappingPath != "" { + return cleanPath(hostSysMappingPath + "/" + subPath) + } + return cleanPath("/sys/" + subPath) +} + func cleanPath(p string) string { return path.Clean(p) } diff --git a/pkg/tools/host/file_test.go b/pkg/tools/host/file_test.go new file mode 100644 index 00000000..354cd4dd --- /dev/null +++ b/pkg/tools/host/file_test.go @@ -0,0 +1,50 @@ +// Licensed to Apache Software Foundation (ASF) under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Apache Software Foundation (ASF) licenses this file to you under +// the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package host + +import "testing" + +// TestGetHostSysInHost guards the prefix the cgroup tree is read through. +// +// This matters more than it looks: if the mapping were ignored and /sys used directly, the agent +// would read its *own* cgroup subtree, where none of the node's containers are. Nothing would +// error - the walk would just map no containers, and every cgroup lookup would miss - so the +// symptom would be "no short-lived process is ever reported", which is indistinguishable from +// there being no traffic. +func TestGetHostSysInHost(t *testing.T) { + original := hostSysMappingPath + defer func() { hostSysMappingPath = original }() + + tests := []struct { + name string + mapping string + want string + }{ + {"unmapped, the real host path", "", "/sys/fs/cgroup"}, + {"mapped by the deployment", "/host/sys", "/host/sys/fs/cgroup"}, + {"mapping with a trailing slash", "/host/sys/", "/host/sys/fs/cgroup"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + hostSysMappingPath = tt.mapping + if got := GetHostSysInHost("fs/cgroup"); got != tt.want { + t.Fatalf("GetHostSysInHost = %s, want %s", got, tt.want) + } + }) + } +} From 346fc486d39eeae36d32625922050026f4fc77b3 Mon Sep 17 00:00:00 2001 From: mrproliu <741550557@qq.com> Date: Thu, 16 Jul 2026 15:10:54 +0800 Subject: [PATCH 2/3] fix comments --- pkg/process/finders/kubernetes/finder.go | 27 ++++++++-- pkg/tools/cgroup/resolver.go | 12 +++-- pkg/tools/cgroup/resolver_test.go | 63 ++++++++++++++++++++++++ 3 files changed, 95 insertions(+), 7 deletions(-) diff --git a/pkg/process/finders/kubernetes/finder.go b/pkg/process/finders/kubernetes/finder.go index f440f359..45189aed 100644 --- a/pkg/process/finders/kubernetes/finder.go +++ b/pkg/process/finders/kubernetes/finder.go @@ -63,6 +63,10 @@ var ( dockerPodsRegex = regexp.MustCompile(`docker-(?P\w+)\.scope`) ipExistTimeout = time.Minute * 10 ipSearchParallel = 10 + + // containerScopeRegexes is built once rather than per call: containerIDFromCgroupDir is the + // normalizer for a full cgroup tree walk, so a per-call slice would allocate once per directory. + containerScopeRegexes = []*regexp.Regexp{kubepodsRegex, openShiftPodsRegex, dockerPodsRegex} ) // containerIDFromCgroupDir turns the base name of a cgroup directory into the id of the container @@ -70,7 +74,7 @@ var ( // /proc//cgroup, so that a container resolved by walking the tree and one resolved by reading // a process's cgroup line come out with the same id. func containerIDFromCgroupDir(dirName string) string { - for _, re := range []*regexp.Regexp{kubepodsRegex, openShiftPodsRegex, dockerPodsRegex} { + for _, re := range containerScopeRegexes { if m := re.FindStringSubmatch(dirName); len(m) > 1 { return m[1] } @@ -504,10 +508,25 @@ func (f *ProcessFinder) ShouldMonitor(pid int32) bool { // does. Only when /proc has nothing left to say - the process already exited, which is the whole // reason it is worth catching this early - does it fall back to what the kernel handed us: the // cgroup id identifies the container, and the task name stands in for the command line. +// This runs once per process the kernel reports, so the pod/container map is built once and shared +// by both paths below. Calling ShouldMonitor here instead would rebuild it a second time for every +// event, and rebuilding means walking every pod of every informer - the kind of per-event cost that +// made these tracepoints too expensive to keep the last time round. func (f *ProcessFinder) ShouldMonitorExecuting(exec *api.ProcessExecuteContext) bool { - if f.ShouldMonitor(exec.Pid) { - return true + containers := f.registry.BuildPodContainers() + if len(containers) == 0 { + return false } + + // the ordinary path: the process is still alive, so /proc answers just as it does for the + // periodic scan + if alive, err := process.NewProcess(exec.Pid); err == nil { + if processes, monitor := f.buildProcess(alive, nil, containers); monitor && len(processes) > 0 { + f.manager.AddDetectedProcess(processes) + return true + } + } + if exec.CgroupID == 0 || f.cgroupResolver == nil { // no kernel side identity to fall back on(cgroup v1, or the tree could not be walked) return false @@ -516,7 +535,7 @@ func (f *ProcessFinder) ShouldMonitorExecuting(exec *api.ProcessExecuteContext) if !exist { return false } - pc, exist := f.registry.BuildPodContainers()[containerID] + pc, exist := containers[containerID] if !exist || pc == nil { return false } diff --git a/pkg/tools/cgroup/resolver.go b/pkg/tools/cgroup/resolver.go index 0e299414..18839156 100644 --- a/pkg/tools/cgroup/resolver.go +++ b/pkg/tools/cgroup/resolver.go @@ -129,9 +129,15 @@ func (r *Resolver) Refresh() error { scanned := 0 err := filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error { if err != nil { - // cgroups come and go while we walk; a vanished directory is normal churn on a busy - // node, so skip it rather than abandoning the whole refresh. - return nil //nolint:nilerr // deliberate: keep walking past transient errors + // A cgroup that vanished mid-walk is normal churn on a busy node, so skip it. Anything + // else - a permission or an IO error - must not be swallowed: it would leave the mapping + // quietly incomplete, and an incomplete mapping does not look like a failure. It looks + // exactly like the containers it missed having no short-lived processes. Fail instead, + // so the caller keeps its previous mapping, or falls back, knowingly. + if os.IsNotExist(err) { + return nil + } + return fmt.Errorf("reading %s: %w", path, err) } if !d.IsDir() { return nil diff --git a/pkg/tools/cgroup/resolver_test.go b/pkg/tools/cgroup/resolver_test.go index c793e78d..731c34ad 100644 --- a/pkg/tools/cgroup/resolver_test.go +++ b/pkg/tools/cgroup/resolver_test.go @@ -210,6 +210,69 @@ func TestResolverIsIndifferentToDepth(t *testing.T) { } } +// TestRefreshFailsOnUnreadableSubtree pins the difference between the two kinds of walk error. +// +// A cgroup vanishing mid-walk is routine and must be skipped. Anything else - here an unreadable +// directory - has to fail the refresh instead, because a mapping that is quietly missing entries is +// indistinguishable from those containers simply having no short-lived processes: nothing errors, +// nothing is attributed, and there is no signal anywhere. Failing lets the caller keep its previous +// mapping, or fall back, knowing that it did. +func TestRefreshFailsOnUnreadableSubtree(t *testing.T) { + if os.Geteuid() == 0 { + t.Skip("root ignores directory permissions, so the error cannot be provoked") + } + root, _, _ := buildTree(t) + locked := mkdirAll(t, root, "kubepods.slice", "locked.slice") + mkdirAll(t, locked, "cri-containerd-"+strings.Repeat("9", 64)+".scope") + if err := os.Chmod(locked, 0o000); err != nil { + t.Fatalf("chmod: %v", err) + } + // restored so t.TempDir can clean the tree up + t.Cleanup(func() { os.Chmod(locked, 0o755) }) //nolint:errcheck // best effort + + r := NewResolver(root, testNormalizer) + err := r.Refresh() + if err == nil { + t.Fatal("an unreadable subtree must fail the refresh, not silently shrink the mapping") + } + if !strings.Contains(err.Error(), "locked.slice") { + t.Fatalf("the error should name the path it could not read, got: %v", err) + } +} + +// TestRefreshKeepsPreviousMappingOnFailure guards what a failed refresh must NOT do: throw away a +// mapping that was good. The periodic refresh only logs its error, so if a failure emptied the +// mapping the agent would stop attributing anything until some later walk happened to succeed. +func TestRefreshKeepsPreviousMappingOnFailure(t *testing.T) { + if os.Geteuid() == 0 { + t.Skip("root ignores directory permissions, so the error cannot be provoked") + } + root, systemdContainer, _ := buildTree(t) + r := NewResolver(root, testNormalizer) + if err := r.Refresh(); err != nil { + t.Fatalf("first refresh: %v", err) + } + before, exist := r.CgroupIDByContainer(systemdContainer) + if !exist { + t.Fatal("the container should be mapped after a good refresh") + } + + locked := mkdirAll(t, root, "kubepods.slice", "locked2.slice") + if err := os.Chmod(locked, 0o000); err != nil { + t.Fatalf("chmod: %v", err) + } + t.Cleanup(func() { os.Chmod(locked, 0o755) }) //nolint:errcheck // best effort + + if err := r.Refresh(); err == nil { + t.Fatal("expected the refresh to fail") + } + after, exist := r.CgroupIDByContainer(systemdContainer) + if !exist || after != before { + t.Fatalf("a failed refresh must leave the previous mapping intact, got exist=%v id=%d want id=%d", + exist, after, before) + } +} + func TestAvailableProbesUnifiedHierarchy(t *testing.T) { root, _, _ := buildTree(t) if !Available(root) { From 8b926b5a1bf0094c571081f68284b4aecc2b7d67 Mon Sep 17 00:00:00 2001 From: mrproliu <741550557@qq.com> Date: Thu, 16 Jul 2026 15:44:14 +0800 Subject: [PATCH 3/3] fix CI --- pkg/accesslog/collector/process.go | 2 +- pkg/process/finders/kubernetes/finder.go | 2 +- pkg/tools/cgroup/resolver.go | 2 +- pkg/tools/cgroup/resolver_test.go | 2 +- 4 files changed, 4 insertions(+), 4 deletions(-) diff --git a/pkg/accesslog/collector/process.go b/pkg/accesslog/collector/process.go index a9d535a8..27c2932c 100644 --- a/pkg/accesslog/collector/process.go +++ b/pkg/accesslog/collector/process.go @@ -54,7 +54,7 @@ func NewProcessCollector() *ProcessCollector { func (p *ProcessCollector) Start(_ *module.Manager, context *common.AccessLogContext) error { // 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 behaviour before this collector + // 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 " + diff --git a/pkg/process/finders/kubernetes/finder.go b/pkg/process/finders/kubernetes/finder.go index 45189aed..a7380fd0 100644 --- a/pkg/process/finders/kubernetes/finder.go +++ b/pkg/process/finders/kubernetes/finder.go @@ -307,7 +307,7 @@ func (f *ProcessFinder) BuildProcesses(p *process.Process, pc *PodContainer) ([] // BuildProcessesWithFallbackName is BuildProcesses for a process whose /proc entry may already be // gone: fallbackName(the kernel task name) then stands in for the command line. Passing "" gives -// exactly the behaviour of BuildProcesses. +// exactly the behavior of BuildProcesses. func (f *ProcessFinder) BuildProcessesWithFallbackName(p *process.Process, pc *PodContainer, fallbackName string) ([]*Process, error) { // find builder diff --git a/pkg/tools/cgroup/resolver.go b/pkg/tools/cgroup/resolver.go index 18839156..d4d89c41 100644 --- a/pkg/tools/cgroup/resolver.go +++ b/pkg/tools/cgroup/resolver.go @@ -155,7 +155,7 @@ func (r *Resolver) Refresh() error { if !ok { return nil } - id := uint64(stat.Ino) + id := stat.Ino idByContainer[container] = id containerByID[id] = container // The path is logged with the id because this mapping is the one thing that has to agree diff --git a/pkg/tools/cgroup/resolver_test.go b/pkg/tools/cgroup/resolver_test.go index 731c34ad..2f06bf31 100644 --- a/pkg/tools/cgroup/resolver_test.go +++ b/pkg/tools/cgroup/resolver_test.go @@ -54,7 +54,7 @@ func inodeOf(t *testing.T, path string) uint64 { if !ok { t.Fatalf("no Stat_t for %s", path) } - return uint64(stat.Ino) + return stat.Ino } func mkdirAll(t *testing.T, parts ...string) string {