[off-CPU stack 3/3] Add duration-filtered off-CPU profiling with task-state labels

internal/controller/controller.go

@@ -14,6 +14,7 @@ import (
 	"go.opentelemetry.io/ebpf-profiler/internal/linux"
 	"go.opentelemetry.io/ebpf-profiler/internal/log"
 	"go.opentelemetry.io/ebpf-profiler/probes/generic"
+	"go.opentelemetry.io/ebpf-profiler/probes/offcpu"
 
 	"go.opentelemetry.io/ebpf-profiler/libpf"
 	"go.opentelemetry.io/ebpf-profiler/metrics"
@@ -210,6 +211,8 @@ func createCustomProbe(name string, cfg any) (tracer.Probe, error) {
 			return nil, fmt.Errorf("decoding generic probe config: %w", err)
 		}
 		return generic.New(gcfg)
+	case offcpu.SampleType:
+		return offcpu.New(cfg)
 	default:
 		return nil, fmt.Errorf("unknown custom probe: %q", name)
 	}

libpf/trace.go

@@ -140,6 +140,7 @@ type EbpfTrace struct {
 	KernelFrames     Frames
 	FrameDataBuf     [3072]uint64
 	Value            int64
+	ProbeUserData    uint64
 	KTime            int64
 	CpuID            uint32
 	TID              PID

metrics/metrics.json

@@ -2188,5 +2188,54 @@
     "name": "UnwindLuaJITErrNoProcInfo",
     "field": "bpf.luajit.errors.no_proc_info",
     "id": 299
+  },
+  {
+    "description": "Number of thread switch-out events observed by the off-CPU time probe",
+    "type": "counter",
+    "name": "OffCPUTimeSwitchOuts",
+    "field": "agent.off_cpu_time.switch_outs",
+    "id": 300
+  },
+  {
+    "description": "Number of thread switch-in events observed by the off-CPU time probe",
+    "type": "counter",
+    "name": "OffCPUTimeSwitchIns",
+    "field": "agent.off_cpu_time.switch_ins",
+    "id": 301
+  },
+  {
+    "description": "Number of off-CPU time switch-ins without a recorded switch-out",
+    "type": "counter",
+    "name": "OffCPUTimeUnmatchedSwitchIns",
+    "field": "agent.off_cpu_time.switch_ins.unmatched",
+    "id": 302
+  },
+  {
+    "description": "Number of off-CPU time samples filtered by duration or sampling",
+    "type": "counter",
+    "name": "OffCPUTimeFiltered",
+    "field": "agent.off_cpu_time.filtered",
+    "id": 303
+  },
+  {
+    "description": "Number of off-CPU time samples submitted for unwinding",
+    "type": "counter",
+    "name": "OffCPUTimeEmitted",
+    "field": "agent.off_cpu_time.emitted",
+    "id": 304
+  },
+  {
+    "description": "Number of failures recording off-CPU time switch-out state",
+    "type": "counter",
+    "name": "OffCPUTimeStateUpdateFailures",
+    "field": "agent.off_cpu_time.state.update_failures",
+    "id": 305
+  },
+  {
+    "description": "Number of failures reading off-CPU time eBPF metrics",
+    "type": "counter",
+    "name": "OffCPUTimeMetricReadFailures",
+    "field": "agent.off_cpu_time.metrics.read_failures",
+    "id": 306
   }
 ]

probes/offcpu/README.md

@@ -0,0 +1,106 @@
+# Off-CPU time probe
+
+`offcpu` measures **how long** each stack was off CPU. Like the built-in
+`off_cpu` profile, each emitted event is weighted by elapsed nanoseconds rather
+than count one. It follows the classic `offcputime` model (Gregg, *BPF
+Performance Tools*, Ch06): the departure time of every thread is recorded as it is
+switched out, and when the thread is switched back in the elapsed nanoseconds
+are attributed to the stack that blocked. Samples carry the measured duration
+(`off_cpu_time` / nanoseconds), so aggregation by stack yields total wait
+time — the flame graph answers "where does wall-clock time go while blocked".
+
+## Hook points
+
+- `tracepoint/sched/sched_switch` runs in the outgoing thread's context and
+  records only a timestamp keyed by `pid_tgid`. It runs on every context
+  switch, so it does nothing else.
+- `kprobe/finish_task_switch` runs in the incoming thread's context right
+  after the scheduler restored its registers. At this point the kernel stack
+  is still the blocking path (`…→schedule→finish_task_switch`) and the user
+  state is unchanged since the thread went off CPU, so the unwind through the
+  shared kprobe unwinder chain observes exactly the stack that was waiting.
+  Optimized symbol variants (`finish_task_switch.isra.0`) are resolved by
+  prefix from kallsyms, mirroring the built-in off-CPU hook. The program that
+  calls `collect_trace` must be a kprobe: tail calls into the kprobe unwinder
+  program array require matching program types, which is why the measurement
+  side cannot be a tracepoint.
+
+## Semantics
+
+- Duration covers **all** off-CPU time: blocking on I/O, locks, timers, and
+  also runqueue wait after involuntary preemption. Every sample carries a
+  **`thread.state` label** with the thread's state at switch-out so the two
+  populations separate cleanly: `interruptible`/`uninterruptible`/`parked`/
+  `idle`/… mean the thread blocked voluntarily, while `running` (entered the
+  scheduler while still runnable, e.g. `cond_resched`/yield) and `preempted`
+  (descheduled involuntarily) mean the thread was denied CPU by saturation.
+  Filter `thread.state` out of `{running, preempted}` for pure blocking
+  analysis, or onto them to quantify CPU pressure.
+- The `prev_state` layout is validated by parsing the sched_switch
+  **tracepoint format file** (`/sys/kernel/tracing/events/sched/sched_switch/format`,
+  debugfs fallback) before direct context access is enabled. Regular tracepoint
+  contexts require direct field access (`bpf_probe_read_kernel` may fail on
+  them), so the supported 64-bit x86_64/aarch64 layout must report offset 32,
+  size 8. A mismatch logs and continues without the label; durations are
+  unaffected. Values use the post-4.14 `task_state_index` encoding;
+  unrecognized encodings collapse to `other` to bound label cardinality.
+- `min_duration` and `sample_rate` are applied at **switch-in**, after the
+  duration is known. Each retained duration is exact. At `sample_rate=1`, all
+  qualifying waits are retained; lower rates can drop qualifying waits and are
+  not inverse-probability scaled. Unlike built-in entry sampling, duration
+  filtering never rejects a wait before knowing its length. `min_duration` is
+  the primary volume control. The safe default is 1ms; explicitly setting 0s
+  requests exhaustive capture and may overwhelm busy hosts.
+- The switch-out map is a **global LRU hash** (`max_entries`, default 64Ki):
+  the pending set is every sleeping thread in the system, and a thread that
+  exits while off CPU leaks its entry (its final switch-out happens after
+  every exit tracepoint, so there is no deterministic cleanup hook). LRU
+  recycles leaked and coldest entries under pressure; the threads evicted
+  first are those sleeping longest. Unlike the built-in `sched_times` map it
+  is not per-CPU, so a thread that blocks on one CPU and wakes on another is
+  still measured.
+- Idle (pid 0) is never tracked. Kernel threads are measured and produce
+  kernel-only stacks.
+
+## Comparison with the built-in off-CPU profiling
+
+| | built-in (`off-cpu-threshold`) | this probe |
+|---|---|---|
+| sample value | exact selected interval duration (ns) | exact selected interval duration (ns) |
+| decision point | random sample at switch-out | duration filter, then sample at switch-in |
+| long waits | missed with probability 1-p | all retained at sample_rate=1 after min_duration |
+| CPU migration | lost (per-CPU LRU map) | measured (global LRU map) |
+| duration filter | none | `min_duration` |
+| drop accounting | none | per-path metrics |
+| thread state | not captured | `thread.state` sample label |
+| activation | static flag at startup | dynamic `custom_probes` entry |
+
+## Configuration
+
+Enabled as a custom probe named `off_cpu_time`:
+
+```yaml
+custom_probes:
+  off_cpu_time:
+    min_duration: 1ms   # emit only waits at least this long (default 1ms)
+    sample_rate: 0.25   # sample qualifying wakeups (default 1.0)
+    max_entries: 65536  # pending switch-out states (default 64Ki)
+```
+
+The sample type is fixed (`off_cpu_time`, nanoseconds) rather than
+descriptor-supplied so downstream duration classification cannot drift.
+
+## Overhead
+
+The switch-out program adds one hash update to every context switch and the
+switch-in program one lookup+delete; this is the unavoidable cost of exact
+durations (identical to `offcputime`). Unwinds are bounded by
+`min_duration`/`sample_rate`. On busy hosts prefer raising `min_duration`
+over lowering `sample_rate`: it suppresses the huge population of micro-waits
+while keeping every meaningful stall.
+
+## Metrics
+
+`agent.off_cpu_time.switch_outs`, `.switch_ins`, `.switch_ins.unmatched`,
+`.filtered`, `.emitted`, `.state.update_failures`, and
+`.metrics.read_failures` account for every event and drop path.

probes/offcpu/metrics.go

@@ -0,0 +1,119 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package offcpu // import "go.opentelemetry.io/ebpf-profiler/probes/offcpu"
+
+import (
+	"fmt"
+	"sync"
+
+	"github.com/cilium/ebpf"
+	"go.opentelemetry.io/ebpf-profiler/internal/log"
+	"go.opentelemetry.io/ebpf-profiler/metrics"
+	"go.opentelemetry.io/ebpf-profiler/tracer"
+)
+
+// offCPUTimeBPFMetrics mirrors OffCpuTimeMetrics in off_cpu_time.ebpf.c.
+type offCPUTimeBPFMetrics struct {
+	SwitchOuts          uint64
+	SwitchIns           uint64
+	UnmatchedSwitchIns  uint64
+	Filtered            uint64
+	Emitted             uint64
+	StateUpdateFailures uint64
+}
+
+type offCPUTimeMetricsCollector struct {
+	metricsMap   *ebpf.Map
+	perCPUValues []offCPUTimeBPFMetrics
+
+	mu            sync.Mutex
+	previous      offCPUTimeBPFMetrics
+	readFailures  uint64
+	previousReads uint64
+}
+
+func newOffCPUTimeMetricsCollector(metricsMap *ebpf.Map) (*offCPUTimeMetricsCollector, error) {
+	cpuCount, err := ebpf.PossibleCPU()
+	if err != nil {
+		return nil, fmt.Errorf("determine possible CPUs: %w", err)
+	}
+	return &offCPUTimeMetricsCollector{
+		metricsMap:   metricsMap,
+		perCPUValues: make([]offCPUTimeBPFMetrics, cpuCount),
+	}, nil
+}
+
+func (collector *offCPUTimeMetricsCollector) snapshot() (offCPUTimeBPFMetrics, error) {
+	key := uint32(0)
+	if err := collector.metricsMap.Lookup(&key, &collector.perCPUValues); err != nil {
+		return offCPUTimeBPFMetrics{}, err
+	}
+	var result offCPUTimeBPFMetrics
+	for _, value := range collector.perCPUValues {
+		result.SwitchOuts += value.SwitchOuts
+		result.SwitchIns += value.SwitchIns
+		result.UnmatchedSwitchIns += value.UnmatchedSwitchIns
+		result.Filtered += value.Filtered
+		result.Emitted += value.Emitted
+		result.StateUpdateFailures += value.StateUpdateFailures
+	}
+	return result, nil
+}
+
+func counterDelta(current, previous uint64) metrics.MetricValue {
+	return metrics.MetricValue(current - previous)
+}
+
+func offCPUTimeMetricValues(current, previous offCPUTimeBPFMetrics,
+	readFailures uint64,
+) []metrics.Metric {
+	return []metrics.Metric{
+		{ID: metrics.IDOffCPUTimeSwitchOuts,
+			Value: counterDelta(current.SwitchOuts, previous.SwitchOuts)},
+		{ID: metrics.IDOffCPUTimeSwitchIns,
+			Value: counterDelta(current.SwitchIns, previous.SwitchIns)},
+		{ID: metrics.IDOffCPUTimeUnmatchedSwitchIns,
+			Value: counterDelta(current.UnmatchedSwitchIns, previous.UnmatchedSwitchIns)},
+		{ID: metrics.IDOffCPUTimeFiltered,
+			Value: counterDelta(current.Filtered, previous.Filtered)},
+		{ID: metrics.IDOffCPUTimeEmitted,
+			Value: counterDelta(current.Emitted, previous.Emitted)},
+		{ID: metrics.IDOffCPUTimeStateUpdateFailures,
+			Value: counterDelta(current.StateUpdateFailures, previous.StateUpdateFailures)},
+		{ID: metrics.IDOffCPUTimeMetricReadFailures,
+			Value: metrics.MetricValue(readFailures)},
+	}
+}
+
+func (collector *offCPUTimeMetricsCollector) Collect() []metrics.Metric {
+	collector.mu.Lock()
+	defer collector.mu.Unlock()
+
+	current, err := collector.snapshot()
+	if err != nil {
+		collector.readFailures++
+		log.Debugf("Failed to read off-cpu time metrics: %v", err)
+		reads := collector.readFailures
+		delta := reads - collector.previousReads
+		collector.previousReads = reads
+		return []metrics.Metric{{
+			ID: metrics.IDOffCPUTimeMetricReadFailures, Value: metrics.MetricValue(delta),
+		}}
+	}
+	previous := collector.previous
+	collector.previous = current
+	reads := collector.readFailures
+	readDelta := reads - collector.previousReads
+	collector.previousReads = reads
+	return offCPUTimeMetricValues(current, previous, readDelta)
+}
+
+type finalMetricsCloser struct {
+	subscription tracer.MetricsSubscription
+	collect      func() []metrics.Metric
+}
+
+func (closer *finalMetricsCloser) Close() error {
+	return closer.subscription.CloseWithMetrics(closer.collect())
+}