This is an automated email from the ASF dual-hosted git repository.
mrproliu pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-rover.git
The following commit(s) were added to refs/heads/main by this push:
new 6a23532 Support monitoring short-lived processes in the access log
module (#214)
6a23532 is described below
commit 6a235327800ea84ab2062c8b1896d8188da895ff
Author: mrproliu <[email protected]>
AuthorDate: Thu Jul 16 19:29:01 2026 +0800
Support monitoring short-lived processes in the access log module (#214)
---
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 | 183 ++++++++++++++-
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 | 205 +++++++++++++++++
pkg/tools/cgroup/resolver_test.go | 312 ++++++++++++++++++++++++++
pkg/tools/host/file.go | 18 ++
pkg/tools/host/file_test.go | 50 +++++
18 files changed, 1440 insertions(+), 28 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 1f2ca99..12fd2a8 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 3658c5c..dc45392 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/<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;
}
\ No newline at end of file
diff --git a/bpf/accesslog/process/process.h b/bpf/accesslog/process/process.h
index a9dd2f4..ee19d66 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 e4dbdce..27c2932 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 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[:])
}
diff --git a/pkg/accesslog/common/connection.go
b/pkg/accesslog/common/connection.go
index decfce9..ad5d5ee 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 ca11907..7964e46 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/<pid> 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 75a09ba..a440141 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 d4ba8b1..6fd8640 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/<pid> 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 e73c16a..4c3d7f1 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 2edfbbf..a7380fd 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,46 @@ var log = logger.GetLogger("process", "finder",
"kubernetes")
var (
kubepodsRegex =
regexp.MustCompile(`cri-containerd-(?P<Group>\w+)\.scope`)
openShiftPodsRegex = regexp.MustCompile(`crio-(?P<Group>\w+)\.scope`)
+ dockerPodsRegex = regexp.MustCompile(`docker-(?P<Group>\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
+// it holds, or "" when it holds none. It mirrors the naming GetProcessCGroup
already parses out of
+// /proc/<pid>/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 containerScopeRegexes {
+ 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 +111,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 +164,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 +230,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 +302,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 behavior 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 +328,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 +345,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 +362,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 +502,96 @@ 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.
+// 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 {
+ 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
+ }
+ containerID, exist :=
f.cgroupResolver.ContainerByCgroupID(exec.CgroupID)
+ if !exist {
+ return false
+ }
+ pc, exist := containers[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 617f82b..ab73280 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 a252416..dc8eb37 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 0000000..23aa636
--- /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 6a62632..8081e9d 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 0000000..d4d89c4
--- /dev/null
+++ b/pkg/tools/cgroup/resolver.go
@@ -0,0 +1,205 @@
+// 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-<qos>.slice/kubepods-<qos>-pod<uid>.slice/<id>.scope,
cgroupfs:
+// kubepods/<qos>/pod<uid>/<id>)
+// - 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-<node>.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/<pid>/cgroup - is the whole
+// point: the paths in /proc/<pid>/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::/../<container-id>", 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 {
+ // 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
+ }
+ 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 := 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 0000000..2f06bf3
--- /dev/null
+++ b/pkg/tools/cgroup/resolver_test.go
@@ -0,0 +1,312 @@
+// 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 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<uid>.slice/cri-containerd-<id>.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<uid>/<id>
+ 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)
+ }
+}
+
+// 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) {
+ 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 b376288..0513ffc 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 0000000..354cd4d
--- /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)
+ }
+ })
+ }
+}
