diff --git a/internal/controller/controller.go b/internal/controller/controller.go index 62e1d0daaf..a46bc36cde 100644 --- a/internal/controller/controller.go +++ b/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) } diff --git a/libpf/trace.go b/libpf/trace.go index 702bd02c34..93814e9ecb 100644 --- a/libpf/trace.go +++ b/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 diff --git a/metrics/ids.go b/metrics/ids.go index 7721efcaea..2a20149e7b 100644 --- a/metrics/ids.go +++ b/metrics/ids.go @@ -686,6 +686,27 @@ const ( // Number of times we didn't find an entry for this process in the LuaJIT process info array IDUnwindLuaJITErrNoProcInfo = 299 + // Number of thread switch-out events observed by the off-CPU time probe + IDOffCPUTimeSwitchOuts = 300 + + // Number of thread switch-in events observed by the off-CPU time probe + IDOffCPUTimeSwitchIns = 301 + + // Number of off-CPU time switch-ins without a recorded switch-out + IDOffCPUTimeUnmatchedSwitchIns = 302 + + // Number of off-CPU time samples filtered by duration or sampling + IDOffCPUTimeFiltered = 303 + + // Number of off-CPU time samples submitted for unwinding + IDOffCPUTimeEmitted = 304 + + // Number of failures recording off-CPU time switch-out state + IDOffCPUTimeStateUpdateFailures = 305 + + // Number of failures reading off-CPU time eBPF metrics + IDOffCPUTimeMetricReadFailures = 306 + // max number of ID values, keep this as *last entry* - IDMax = 300 + IDMax = 307 ) diff --git a/metrics/metrics.json b/metrics/metrics.json index 9455fc8fb3..677af1fd80 100644 --- a/metrics/metrics.json +++ b/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 } ] diff --git a/probes/offcpu/README.md b/probes/offcpu/README.md new file mode 100644 index 0000000000..6a76072407 --- /dev/null +++ b/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. diff --git a/probes/offcpu/metrics.go b/probes/offcpu/metrics.go new file mode 100644 index 0000000000..01a6935786 --- /dev/null +++ b/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()) +} diff --git a/probes/offcpu/metrics_test.go b/probes/offcpu/metrics_test.go new file mode 100644 index 0000000000..6defd00c64 --- /dev/null +++ b/probes/offcpu/metrics_test.go @@ -0,0 +1,86 @@ +// Copyright The OpenTelemetry Authors +// SPDX-License-Identifier: Apache-2.0 + +package offcpu + +import ( + "testing" + "unsafe" + + "github.com/stretchr/testify/require" + "go.opentelemetry.io/ebpf-profiler/metrics" +) + +func TestOffCPUTimeMetricValues(t *testing.T) { + previous := offCPUTimeBPFMetrics{ + SwitchOuts: 100, + SwitchIns: 90, + UnmatchedSwitchIns: 5, + Filtered: 60, + Emitted: 25, + StateUpdateFailures: 1, + } + current := offCPUTimeBPFMetrics{ + SwitchOuts: 150, + SwitchIns: 140, + UnmatchedSwitchIns: 6, + Filtered: 95, + Emitted: 39, + StateUpdateFailures: 2, + } + + got := make(map[metrics.MetricID]metrics.MetricValue) + for _, metric := range offCPUTimeMetricValues(current, previous, 3) { + got[metric.ID] = metric.Value + } + require.Equal(t, map[metrics.MetricID]metrics.MetricValue{ + metrics.IDOffCPUTimeSwitchOuts: 50, + metrics.IDOffCPUTimeSwitchIns: 50, + metrics.IDOffCPUTimeUnmatchedSwitchIns: 1, + metrics.IDOffCPUTimeFiltered: 35, + metrics.IDOffCPUTimeEmitted: 14, + metrics.IDOffCPUTimeStateUpdateFailures: 1, + metrics.IDOffCPUTimeMetricReadFailures: 3, + }, got) +} + +func TestOffCPUTimeBPFMetricsLayout(t *testing.T) { + var value offCPUTimeBPFMetrics + require.Equal(t, uintptr(0), unsafe.Offsetof(value.SwitchOuts)) + require.Equal(t, uintptr(8), unsafe.Offsetof(value.SwitchIns)) + require.Equal(t, uintptr(16), unsafe.Offsetof(value.UnmatchedSwitchIns)) + require.Equal(t, uintptr(24), unsafe.Offsetof(value.Filtered)) + require.Equal(t, uintptr(32), unsafe.Offsetof(value.Emitted)) + require.Equal(t, uintptr(40), unsafe.Offsetof(value.StateUpdateFailures)) + require.Equal(t, uintptr(48), unsafe.Sizeof(value)) +} + +type testMetricsSubscription struct { + closed bool + metrics []metrics.Metric +} + +func (subscription *testMetricsSubscription) Close() error { + subscription.closed = true + return nil +} + +func (subscription *testMetricsSubscription) CloseWithMetrics(final []metrics.Metric) error { + subscription.closed = true + subscription.metrics = final + return nil +} + +func TestFinalMetricsCloserPreservesDeltas(t *testing.T) { + subscription := &testMetricsSubscription{} + closer := &finalMetricsCloser{ + subscription: subscription, + collect: func() []metrics.Metric { + return []metrics.Metric{{ID: metrics.IDOffCPUTimeEmitted, Value: 3}} + }, + } + require.NoError(t, closer.Close()) + require.True(t, subscription.closed) + require.Equal(t, []metrics.Metric{{ID: metrics.IDOffCPUTimeEmitted, Value: 3}}, + subscription.metrics) +} diff --git a/probes/offcpu/offcpu.go b/probes/offcpu/offcpu.go new file mode 100644 index 0000000000..17380e7648 --- /dev/null +++ b/probes/offcpu/offcpu.go @@ -0,0 +1,439 @@ +// Copyright The OpenTelemetry Authors +// SPDX-License-Identifier: Apache-2.0 + +// Package offcpu profiles how long stacks spend off CPU. +// +// It measures the wall-clock nanoseconds between a thread being switched out +// and being switched back in, and attributes that duration to the blocked +// stack. This is the classic offcputime semantics: the value answers "how +// long was this stack waiting" rather than "how often did this stack leave +// the CPU". +package offcpu // import "go.opentelemetry.io/ebpf-profiler/probes/offcpu" + +import ( + "bufio" + "errors" + "fmt" + "os" + "path/filepath" + "strconv" + "strings" + "time" + + cebpf "github.com/cilium/ebpf" + "github.com/cilium/ebpf/link" + "go.opentelemetry.io/ebpf-profiler/internal/log" + "go.opentelemetry.io/ebpf-profiler/libpf" + "go.opentelemetry.io/ebpf-profiler/probes/internal/probeutil" + "go.opentelemetry.io/ebpf-profiler/reporter/samples" + "go.opentelemetry.io/ebpf-profiler/rlimit" + "go.opentelemetry.io/ebpf-profiler/tracer" +) + +// SampleType identifies off-CPU time samples. It is fixed rather than +// descriptor-supplied so downstream duration classification cannot drift. +const SampleType = "off_cpu_time" + +const ( + startsMapName = "off_cpu_time_starts" + metricsMapName = "off_cpu_time_metrics" + switchOutProgramName = "off_cpu_time_switch_out" + switchInProgramName = "off_cpu_time_switch_in" + originVariableName = "off_cpu_time_origin" + minNSVariableName = "off_cpu_time_min_ns" + sampleVariableName = "off_cpu_time_sample_threshold" + stateEnabledVariable = "off_cpu_time_state_enabled" + + // The profiler supports only 64-bit x86_64/aarch64 kernels. Both expose + // prev_state at byte 32 as an 8-byte signed long in sched_switch; this + // layout is validated against tracefs before direct context access is enabled. + expectedStateOffset = uint32(32) + expectedStateSize = uint32(8) + + // stateUnavailable is the sentinel the eBPF side reports when no thread + // state was captured for a sample. Must match OFF_CPU_TIME_STATE_UNAVAILABLE. + stateUnavailable = ^uint64(0) + + // threadStateLabelKey is the per-sample label carrying the switch-out + // thread state. It separates voluntary blocking (interruptible, + // uninterruptible, ...) from runnable threads descheduled by CPU + // pressure (running, preempted) so profiles can be filtered on either. + threadStateLabelKey = "thread.state" + + // switchOutTracepointGroup/Name identify the tracepoint that observes + // threads leaving the CPU. It fires in the outgoing thread's context. + switchOutTracepointGroup = "sched" + switchOutTracepointName = "sched_switch" + + // switchInSymbolPrefix is the kernel function that runs in the incoming + // thread's context right after a context switch. Compilers emit optimized + // variants (for example finish_task_switch.isra.0), so attachment resolves + // the symbol by prefix from kallsyms, mirroring the built-in off-CPU hook. + switchInSymbolPrefix = "finish_task_switch" + + // A non-zero default prevents an empty configuration from unwinding every + // scheduler wakeup on the host. Users may explicitly set 0s for exhaustive + // capture after accepting the overhead. + defaultMinDuration = time.Millisecond + + // The pending set is one entry per sleeping thread system-wide, so the + // default is larger than the per-function latency probes. + defaultMaxEntries = uint32(64 * 1024) + maximumMaxEntries = uint32(1024 * 1024) +) + +// kallsymsPath is a variable to allow tests to substitute a fixture. +var kallsymsPath = "/proc/kallsyms" + +// tracefsRoots lists the mount points searched for tracepoint format files. +// A variable to allow tests to substitute fixtures. +var tracefsRoots = []string{"/sys/kernel/tracing", "/sys/kernel/debug/tracing"} + +type offCPUTimeProbe struct { + minDuration time.Duration + sampleThreshold uint32 + maxEntries uint32 +} + +// New creates an off-CPU time probe from collector configuration. Supported +// fields are min_duration (a Go duration string), sample_rate in (0, 1], and +// max_entries. +// +// min_duration is the primary overhead and volume control: durations are +// always measured exactly, and only wakeups whose measured off-CPU time +// reaches the threshold are eligible for unwinding. sample_rate then applies +// to qualifying wakeups, not to switch-outs. +func New(rawConfig any) (tracer.Probe, error) { + rawConfig = configWithDefaultMinDuration(rawConfig) + cfg, err := probeutil.ParseLatencyConfig( + rawConfig, "off-cpu time", defaultMaxEntries, maximumMaxEntries) + if err != nil { + return nil, err + } + + return &offCPUTimeProbe{ + minDuration: cfg.MinDuration, + sampleThreshold: cfg.SampleThreshold, + maxEntries: cfg.MaxEntries, + }, nil +} + +func configWithDefaultMinDuration(rawConfig any) any { + if rawConfig == nil { + return map[string]any{"min_duration": defaultMinDuration} + } + config, ok := rawConfig.(map[string]any) + if !ok { + return rawConfig + } + if _, configured := config["min_duration"]; configured { + return rawConfig + } + withDefault := make(map[string]any, len(config)+1) + for key, value := range config { + withDefault[key] = value + } + withDefault["min_duration"] = defaultMinDuration + return withDefault +} + +func (p *offCPUTimeProbe) ReportMetadata() *samples.TypeMetadata { + return &samples.TypeMetadata{ + SampleType: SampleType, + SampleUnit: "nanoseconds", + ReportValues: true, + } +} + +func failLoad(resources *probeutil.Resources, err error) (link.Link, error) { + return nil, errors.Join(err, resources.Close()) +} + +func (p *offCPUTimeProbe) Load(origin uint16, ctx *tracer.ProbeContext) (link.Link, error) { + restoreRlimit, err := rlimit.MaximizeMemlock() + if err != nil { + return nil, fmt.Errorf("failed to adjust memlock rlimit: %w", err) + } + defer restoreRlimit() + + collection, err := ctx.CollectionSpecWith( + []string{startsMapName, metricsMapName}, + []string{switchOutProgramName, switchInProgramName}, + []string{ + originVariableName, minNSVariableName, sampleVariableName, + stateEnabledVariable, + }, + ) + if err != nil { + return nil, err + } + collection.Maps[startsMapName].MaxEntries = p.maxEntries + + // The tracepoint encoding of prev_state is not a stable ABI, so the field + // location is resolved from the tracefs format file rather than assumed. + // Failure only disables the thread.state label; durations are unaffected. + stateEnabled := uint32(0) + stateOffset, stateSize, stateErr := resolveSchedSwitchStateField() + if stateErr != nil { + log.Infof("Off-CPU time probe continues without thread state: %v", stateErr) + } else if stateOffset != expectedStateOffset || stateSize != expectedStateSize { + log.Infof( + "Off-CPU time probe continues without thread state: unexpected sched_switch "+ + "prev_state layout offset=%d size=%d (expected offset=%d size=%d)", + stateOffset, stateSize, expectedStateOffset, expectedStateSize) + } else { + stateEnabled = 1 + log.Infof("Off-CPU time thread-state capture enabled: prev_state offset=%d size=%d", + stateOffset, stateSize) + } + + variables := map[string]any{ + originVariableName: origin, + minNSVariableName: uint64(p.minDuration), + sampleVariableName: p.sampleThreshold, + stateEnabledVariable: stateEnabled, + } + for name, value := range variables { + if err := collection.Variables[name].Set(value); err != nil { + return nil, fmt.Errorf("failed to set eBPF variable %q: %w", name, err) + } + } + + resources := &probeutil.Resources{} + ownedMaps, err := probeutil.LoadMaps( + collection, []string{startsMapName, metricsMapName}, resources) + if err != nil { + return failLoad(resources, err) + } + if err := ctx.RewriteMaps(collection, ownedMaps); err != nil { + return failLoad(resources, fmt.Errorf("failed to rewrite off-cpu time maps: %w", err)) + } + + programs := make(map[string]*cebpf.Program, 2) + if err := ctx.LoadProbeUnwinders( + collection, + programs, + []tracer.ProgLoaderHelper{ + {Name: switchOutProgramName, NoTailCallTarget: true, Enable: true}, + {Name: switchInProgramName, NoTailCallTarget: true, Enable: true}, + }, + 0, + ); err != nil { + for _, program := range programs { + resources.Add(program) + } + return failLoad(resources, fmt.Errorf("failed to load off-cpu time programs: %w", err)) + } + for _, program := range programs { + resources.Add(program) + } + + collector, err := newOffCPUTimeMetricsCollector(ownedMaps[metricsMapName]) + if err != nil { + return failLoad(resources, err) + } + subscription, err := ctx.RegisterMetricsCollector(collector.Collect) + if err != nil { + return failLoad(resources, fmt.Errorf("register off-cpu time metrics: %w", err)) + } + resources.Add(&finalMetricsCloser{subscription: subscription, collect: collector.Collect}) + + // Attach the consumer (switch-in) before the producer so activation can't + // leave unconsumable start records, mirroring the other latency probes. + if err := attachSwitchIn(programs[switchInProgramName], resources); err != nil { + return failLoad(resources, err) + } + + switchOutLink, err := link.Tracepoint( + switchOutTracepointGroup, switchOutTracepointName, + programs[switchOutProgramName], nil) + if err != nil { + return failLoad(resources, fmt.Errorf("failed to attach %s/%s tracepoint: %w", + switchOutTracepointGroup, switchOutTracepointName, err)) + } + return resources.WrapLink(switchOutLink), nil +} + +// attachSwitchIn attaches the switch-in program to every finish_task_switch +// variant present in kallsyms. At least one attachment must succeed. +func attachSwitchIn(program *cebpf.Program, resources *probeutil.Resources) error { + symbols, err := resolveSwitchInSymbols() + if err != nil || len(symbols) == 0 { + // kallsyms may be unreadable in restricted environments; fall back to + // the plain symbol and let the kernel resolve it. + if err != nil { + log.Debugf("Failed to resolve %s from kallsyms: %v", switchInSymbolPrefix, err) + } + symbols = []string{switchInSymbolPrefix} + } + + var errs []error + attached := 0 + for _, symbol := range symbols { + switchInLink, linkErr := link.Kprobe(symbol, program, nil) + if linkErr != nil { + errs = append(errs, fmt.Errorf("attach kprobe %q: %w", symbol, linkErr)) + continue + } + resources.Add(switchInLink) + attached++ + } + if attached == 0 { + return fmt.Errorf("failed to attach to any of %d %s symbols: %w", + len(symbols), switchInSymbolPrefix, errors.Join(errs...)) + } + if len(errs) != 0 { + log.Warnf("Off-CPU time attached to %d of %d %s symbols; partial failures: %v", + attached, len(symbols), switchInSymbolPrefix, errors.Join(errs...)) + } + return nil +} + +// LabelTrace implements tracer.TraceLabeler. It converts the switch-out +// thread state carried in the trace's user-data slot into the low-cardinality +// thread.state sample label, merging with any runtime labels already present. +func (p *offCPUTimeProbe) LabelTrace(trace *libpf.EbpfTrace) { + if trace.ProbeUserData == stateUnavailable { + return + } + if trace.CustomLabels == nil { + trace.CustomLabels = make(map[libpf.String]libpf.String, 1) + } + trace.CustomLabels[libpf.Intern(threadStateLabelKey)] = + libpf.Intern(threadStateName(trace.ProbeUserData)) +} + +// threadStateName decodes the sched_switch tracepoint prev_state encoding +// used since kernel 4.14: 0 for a still-runnable thread that entered the +// scheduler voluntarily, single TASK_REPORT bits decoded from +// task_state_index, and TASK_REPORT_MAX (0x100) for involuntary preemption. +// Unrecognized encodings collapse to "other" to bound label cardinality. +func threadStateName(state uint64) string { + switch state { + case 0x0000: + return "running" + case 0x0001: + return "interruptible" + case 0x0002: + return "uninterruptible" + case 0x0004: + return "stopped" + case 0x0008: + return "traced" + case 0x0010: + return "exit_dead" + case 0x0020: + return "zombie" + case 0x0040: + return "parked" + case 0x0080: + return "idle" + case 0x0100: + return "preempted" + default: + return "other" + } +} + +// resolveSchedSwitchStateField locates the prev_state field in the raw +// sched_switch tracepoint buffer by parsing the tracefs format file, e.g.: +// +// field:long prev_state; offset:32; size:8; signed:1; +func resolveSchedSwitchStateField() (offset, size uint32, err error) { + var firstErr error + for _, root := range tracefsRoots { + path := filepath.Join(root, "events", switchOutTracepointGroup, + switchOutTracepointName, "format") + offset, size, err := parseTracepointField(path, "prev_state") + if err == nil { + return offset, size, nil + } + if firstErr == nil { + firstErr = err + } + } + return 0, 0, firstErr +} + +func parseTracepointField(path, field string) (offset, size uint32, err error) { + file, err := os.Open(path) + if err != nil { + return 0, 0, err + } + defer file.Close() + + scanner := bufio.NewScanner(file) + for scanner.Scan() { + line := scanner.Text() + if !strings.Contains(line, "field:") || + !strings.Contains(line, " "+field+";") { + continue + } + parsedOffset, offsetErr := tracepointFieldAttribute(line, "offset") + parsedSize, sizeErr := tracepointFieldAttribute(line, "size") + if offsetErr != nil || sizeErr != nil { + return 0, 0, fmt.Errorf("malformed format line %q in %s", line, path) + } + return parsedOffset, parsedSize, nil + } + if err := scanner.Err(); err != nil { + return 0, 0, err + } + return 0, 0, fmt.Errorf("field %q not found in %s", field, path) +} + +func tracepointFieldAttribute(line, attribute string) (uint32, error) { + marker := attribute + ":" + start := strings.Index(line, marker) + if start < 0 { + return 0, fmt.Errorf("attribute %q not found", attribute) + } + rest := line[start+len(marker):] + end := strings.IndexByte(rest, ';') + if end < 0 { + return 0, fmt.Errorf("attribute %q not terminated", attribute) + } + value, err := strconv.ParseUint(strings.TrimSpace(rest[:end]), 10, 32) + if err != nil { + return 0, err + } + return uint32(value), nil +} + +// resolveSwitchInSymbols returns the text symbols from kallsyms that are +// finish_task_switch or an optimized clone of it (finish_task_switch.). +func resolveSwitchInSymbols() ([]string, error) { + file, err := os.Open(kallsymsPath) + if err != nil { + return nil, err + } + defer file.Close() + + seen := make(map[string]struct{}) + var symbols []string + scanner := bufio.NewScanner(file) + for scanner.Scan() { + // Format: address type name [module] + fields := strings.Fields(scanner.Text()) + if len(fields) < 3 { + continue + } + if fields[1] != "t" && fields[1] != "T" { + continue + } + name := fields[2] + if name != switchInSymbolPrefix && + !strings.HasPrefix(name, switchInSymbolPrefix+".") { + continue + } + if _, duplicate := seen[name]; duplicate { + continue + } + seen[name] = struct{}{} + symbols = append(symbols, name) + } + if err := scanner.Err(); err != nil { + return nil, err + } + return symbols, nil +} diff --git a/probes/offcpu/offcpu_integration_test.go b/probes/offcpu/offcpu_integration_test.go new file mode 100644 index 0000000000..d1a54cc8d4 --- /dev/null +++ b/probes/offcpu/offcpu_integration_test.go @@ -0,0 +1,160 @@ +//go:build integration && linux + +// Copyright The OpenTelemetry Authors +// SPDX-License-Identifier: Apache-2.0 + +package offcpu + +import ( + "context" + "os" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" + "go.opentelemetry.io/ebpf-profiler/interpreter/interpreterconfig" + "go.opentelemetry.io/ebpf-profiler/libpf" + "go.opentelemetry.io/ebpf-profiler/metrics" + "go.opentelemetry.io/ebpf-profiler/tracer" + "go.opentelemetry.io/otel/metric/noop" +) + +type integrationIntervals struct{} + +var startMetricsOnce sync.Once + +func (integrationIntervals) MonitorInterval() time.Duration { return time.Second } +func (integrationIntervals) TracePollInterval() time.Duration { return 10 * time.Millisecond } +func (integrationIntervals) PIDCleanupInterval() time.Duration { return time.Second } +func (integrationIntervals) ExecutableUnloadDelay() time.Duration { return time.Second } + +func newIntegrationTracer(t *testing.T) (*tracer.Tracer, context.CancelFunc, chan *libpf.EbpfTrace) { + t.Helper() + startMetricsOnce.Do(func() { metrics.Start(noop.Meter{}) }) + + ctx, cancel := context.WithCancel(t.Context()) + + trc, err := tracer.NewTracer(ctx, &tracer.Config{ + Intervals: integrationIntervals{}, + InterpretersConfig: interpreterconfig.AllInterpreters(), + SamplesPerSecond: 20, + FrameCacheSize: 1024, + ProbabilisticInterval: time.Minute, + ProbabilisticThreshold: tracer.ProbabilisticThresholdMax, + LoadProbe: true, + DynamicProbeResourcePrefixes: []string{SampleType}, + }) + require.NoError(t, err) + + trc.StartPIDEventProcessor(ctx) + traces := make(chan *libpf.EbpfTrace, 4096) + require.NoError(t, trc.StartMapMonitors(ctx, traces)) + return trc, cancel, traces +} + +func TestOffCPUTimeProbe(t *testing.T) { + trc, cancel, traces := newIntegrationTracer(t) + defer cancel() + defer trc.Close() + + minDuration := 20 * time.Millisecond + probe, err := New(map[string]any{"min_duration": minDuration.String()}) + require.NoError(t, err) + labeler, ok := probe.(tracer.TraceLabeler) + require.True(t, ok, "off-cpu probe must implement tracer.TraceLabeler") + require.NoError(t, trc.Enable(probe)) + + // Block a thread of this process well past min_duration in a loop. Each + // timer wakeup switches the sleeping thread back in and emits a sample + // carrying the measured off-CPU nanoseconds. + sleeperDone := make(chan struct{}) + defer close(sleeperDone) + go func() { + for { + select { + case <-sleeperDone: + return + default: + time.Sleep(3 * minDuration) + } + } + }() + + // Drain the channel at full speed: the probe observes the whole system, + // so this process's samples arrive interleaved with everything else's. + deadline := time.NewTimer(10 * time.Second) + defer deadline.Stop() + pid := libpf.PID(os.Getpid()) + for { + select { + case trace := <-traces: + if trace == nil || trace.PID != pid { + continue + } + // CPU-sampling traces from this process carry no value; only + // off-CPU time samples report a duration past the threshold. + if trace.Value < int64(minDuration) { + continue + } + require.NotEmpty(t, trace.KernelFrames) + require.NotEmpty(t, trace.FrameData) + require.Positive(t, trace.NumFrames) + + // The sleeper blocks in nanosleep, so its switch-out state must + // have been captured as TASK_INTERRUPTIBLE. Wait for one such + // sample; other threads of this process may block differently. + if trace.ProbeUserData != 0x0001 { + continue + } + // The raw monitor channel taps traces before Tracer.HandleTrace + // runs the labeler, so apply it directly to prove the state + // reaches the sample label. + labeler.LabelTrace(trace) + require.Equal(t, libpf.Intern("interruptible"), + trace.CustomLabels[libpf.Intern("thread.state")]) + return + case <-deadline.C: + t.Fatal("timed out waiting for an off-cpu time trace") + } + } +} + +func TestOffCPUTimeProbeFiltersShortWaits(t *testing.T) { + trc, cancel, traces := newIntegrationTracer(t) + defer cancel() + defer trc.Close() + + // A threshold far above any wait this workload produces: nothing may be + // emitted for this process. + probe, err := New(map[string]any{"min_duration": "10m"}) + require.NoError(t, err) + require.NoError(t, trc.Enable(probe)) + + sleeperDone := make(chan struct{}) + defer close(sleeperDone) + go func() { + for { + select { + case <-sleeperDone: + return + default: + time.Sleep(50 * time.Millisecond) + } + } + }() + + pid := libpf.PID(os.Getpid()) + quiet := time.NewTimer(3 * time.Second) + defer quiet.Stop() + for { + select { + case trace := <-traces: + if trace != nil && trace.PID == pid && trace.Value > 0 { + t.Fatalf("unexpected off-cpu sample below threshold: %d ns", trace.Value) + } + case <-quiet.C: + return + } + } +} diff --git a/probes/offcpu/offcpu_test.go b/probes/offcpu/offcpu_test.go new file mode 100644 index 0000000000..77de35f4a3 --- /dev/null +++ b/probes/offcpu/offcpu_test.go @@ -0,0 +1,248 @@ +// Copyright The OpenTelemetry Authors +// SPDX-License-Identifier: Apache-2.0 + +package offcpu + +import ( + "math" + "os" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/require" + "go.opentelemetry.io/ebpf-profiler/libpf" +) + +func TestNewDefaults(t *testing.T) { + created, err := New(map[string]any{}) + require.NoError(t, err) + + probe := created.(*offCPUTimeProbe) + require.Equal(t, defaultMinDuration, probe.minDuration) + require.Equal(t, uint32(math.MaxUint32), probe.sampleThreshold) + require.Equal(t, defaultMaxEntries, probe.maxEntries) + + metadata := probe.ReportMetadata() + require.Equal(t, "off_cpu_time", metadata.SampleType) + require.Equal(t, "nanoseconds", metadata.SampleUnit) + require.True(t, metadata.ReportValues) +} + +func TestNewAllowsExplicitExhaustiveCapture(t *testing.T) { + created, err := New(map[string]any{"min_duration": "0s"}) + require.NoError(t, err) + require.Zero(t, created.(*offCPUTimeProbe).minDuration) +} + +func TestNewConfigured(t *testing.T) { + created, err := New(map[string]any{ + "min_duration": "5ms", + "sample_rate": 0.5, + "max_entries": 128 * 1024, + }) + require.NoError(t, err) + + probe := created.(*offCPUTimeProbe) + require.Equal(t, 5*time.Millisecond, probe.minDuration) + require.Equal(t, uint32(math.MaxUint32/2), probe.sampleThreshold) + require.Equal(t, uint32(128*1024), probe.maxEntries) +} + +func TestNewRejectsInvalidConfiguration(t *testing.T) { + tests := map[string]struct { + config any + errorText string + }{ + "not a map": { + config: "bad", + errorText: "off-cpu time configuration must be a map, got string", + }, + "unknown field": { + config: map[string]any{"unknown": true}, + errorText: "unknown off-cpu time configuration field \"unknown\"", + }, + "bad duration": { + config: map[string]any{"min_duration": "soon"}, + errorText: "invalid min_duration \"soon\"", + }, + "negative duration": { + config: map[string]any{"min_duration": -time.Second}, + errorText: "min_duration must not be negative", + }, + "zero sample rate": { + config: map[string]any{"sample_rate": 0.0}, + errorText: "sample_rate must be in the range (0, 1]", + }, + "sample rate above one": { + config: map[string]any{"sample_rate": 1.1}, + errorText: "sample_rate must be in the range (0, 1]", + }, + "zero max entries": { + config: map[string]any{"max_entries": 0}, + errorText: "max_entries must be in the range", + }, + "too many entries": { + config: map[string]any{"max_entries": float64(maximumMaxEntries + 1)}, + errorText: "max_entries must be in the range", + }, + } + + for name, test := range tests { + t.Run(name, func(t *testing.T) { + _, err := New(test.config) + require.ErrorContains(t, err, test.errorText) + }) + } +} + +func withKallsymsFixture(t *testing.T, content string) { + t.Helper() + path := filepath.Join(t.TempDir(), "kallsyms") + require.NoError(t, os.WriteFile(path, []byte(content), 0o600)) + previous := kallsymsPath + kallsymsPath = path + t.Cleanup(func() { kallsymsPath = previous }) +} + +func TestResolveSwitchInSymbolsMatchesVariants(t *testing.T) { + withKallsymsFixture(t, `ffffffff81000000 T _stext +ffffffff810bd2a0 t finish_task_switch.isra.0 +ffffffff810bd2a0 t finish_task_switch.isra.0 +ffffffff810be000 T finish_task_switch +ffffffff810bf000 t finish_task_switching_helper +ffffffff810c0000 d finish_task_switch.data +ffffffff810c1000 t prefix_finish_task_switch +`) + + symbols, err := resolveSwitchInSymbols() + require.NoError(t, err) + require.Equal(t, []string{"finish_task_switch.isra.0", "finish_task_switch"}, symbols) +} + +func TestResolveSwitchInSymbolsEmptyWhenAbsent(t *testing.T) { + withKallsymsFixture(t, `ffffffff81000000 T _stext +ffffffff810bd2a0 t some_other_symbol +`) + + symbols, err := resolveSwitchInSymbols() + require.NoError(t, err) + require.Empty(t, symbols) +} + +func TestThreadStateName(t *testing.T) { + expected := map[uint64]string{ + 0x0000: "running", + 0x0001: "interruptible", + 0x0002: "uninterruptible", + 0x0004: "stopped", + 0x0008: "traced", + 0x0010: "exit_dead", + 0x0020: "zombie", + 0x0040: "parked", + 0x0080: "idle", + 0x0100: "preempted", + 0x0003: "other", + 0x0200: "other", + } + for state, name := range expected { + require.Equal(t, name, threadStateName(state), "state 0x%x", state) + } +} + +func TestLabelTraceAttachesThreadState(t *testing.T) { + probe := &offCPUTimeProbe{} + key := libpf.Intern("thread.state") + + trace := &libpf.EbpfTrace{ProbeUserData: 0x0001} + probe.LabelTrace(trace) + require.Equal(t, libpf.Intern("interruptible"), trace.CustomLabels[key]) + + // Merges with runtime labels instead of replacing them. + existing := libpf.Intern("request_id") + trace = &libpf.EbpfTrace{ + ProbeUserData: 0x0100, + CustomLabels: map[libpf.String]libpf.String{ + existing: libpf.Intern("abc"), + }, + } + probe.LabelTrace(trace) + require.Equal(t, libpf.Intern("preempted"), trace.CustomLabels[key]) + require.Equal(t, libpf.Intern("abc"), trace.CustomLabels[existing]) + + // No label when the state was unavailable. + trace = &libpf.EbpfTrace{ProbeUserData: stateUnavailable} + probe.LabelTrace(trace) + require.NotContains(t, trace.CustomLabels, key) +} + +const schedSwitchFormatFixture = `name: sched_switch +ID: 315 +format: + field:unsigned short common_type; offset:0; size:2; signed:0; + field:unsigned char common_flags; offset:2; size:1; signed:0; + + field:char prev_comm[16]; offset:8; size:16; signed:0; + field:pid_t prev_pid; offset:24; size:4; signed:1; + field:int prev_prio; offset:28; size:4; signed:1; + field:long prev_state; offset:32; size:8; signed:1; + field:char next_comm[16]; offset:40; size:16; signed:0; +` + +func withTracefsFixture(t *testing.T, roots []string) { + t.Helper() + previous := tracefsRoots + tracefsRoots = roots + t.Cleanup(func() { tracefsRoots = previous }) +} + +func writeFormatFixture(t *testing.T, root, content string) { + t.Helper() + dir := filepath.Join(root, "events", "sched", "sched_switch") + require.NoError(t, os.MkdirAll(dir, 0o755)) + require.NoError(t, os.WriteFile(filepath.Join(dir, "format"), []byte(content), 0o600)) +} + +func TestResolveSchedSwitchStateField(t *testing.T) { + root := t.TempDir() + writeFormatFixture(t, root, schedSwitchFormatFixture) + // The first root is missing; resolution falls back to the second. + withTracefsFixture(t, []string{filepath.Join(t.TempDir(), "absent"), root}) + + offset, size, err := resolveSchedSwitchStateField() + require.NoError(t, err) + require.Equal(t, uint32(32), offset) + require.Equal(t, uint32(8), size) +} + +func TestResolveSchedSwitchStateFieldMissingField(t *testing.T) { + root := t.TempDir() + writeFormatFixture(t, root, "name: sched_switch\n\tfield:pid_t prev_pid;\toffset:24;\tsize:4;\tsigned:1;\n") + withTracefsFixture(t, []string{root}) + + _, _, err := resolveSchedSwitchStateField() + require.ErrorContains(t, err, "prev_state") +} + +func TestParseTracepointFieldIgnoresNameSuffixMatches(t *testing.T) { + root := t.TempDir() + // prev_state_extra must not satisfy a prev_state lookup. + writeFormatFixture(t, root, + "\tfield:long prev_state_extra;\toffset:48;\tsize:8;\tsigned:1;\n"+ + "\tfield:long prev_state;\toffset:32;\tsize:8;\tsigned:1;\n") + withTracefsFixture(t, []string{root}) + + offset, size, err := resolveSchedSwitchStateField() + require.NoError(t, err) + require.Equal(t, uint32(32), offset) + require.Equal(t, uint32(8), size) +} + +func TestResolveSwitchInSymbolsMissingFile(t *testing.T) { + previous := kallsymsPath + kallsymsPath = filepath.Join(t.TempDir(), "does-not-exist") + t.Cleanup(func() { kallsymsPath = previous }) + + _, err := resolveSwitchInSymbols() + require.Error(t, err) +} diff --git a/support/ebpf/off_cpu_time.ebpf.c b/support/ebpf/off_cpu_time.ebpf.c new file mode 100644 index 0000000000..408893bab5 --- /dev/null +++ b/support/ebpf/off_cpu_time.ebpf.c @@ -0,0 +1,190 @@ +// Off-CPU time probe: measures how long each stack was off CPU. +// +// This follows the classic offcputime model (Gregg, BPF Performance Tools, +// Ch06 offcputime.bt): when a thread leaves the CPU its departure time is +// recorded, and when it is switched back in the elapsed nanoseconds are +// attributed to the stack that blocked. The switch-in unwind observes the +// blocking stack: the kernel stack still contains the path that entered the +// scheduler (…→schedule→finish_task_switch) and the user state is unchanged +// since the thread went off CPU. +// +// It intentionally differs from the built-in off-CPU support (off_cpu.ebpf.c) +// by following the exit-filtered latency-probe model used by the lock-wait +// and function-latency probes instead of entry-probability sampling: +// - Every switch-out is timestamped, so long waits are never lost to an +// entry-side coin flip. min_duration and sample_rate are applied to the +// measured duration at switch-in, where the cost of a decision is paid +// once per wakeup instead of once per context switch. +// - The start map is a global LRU hash rather than per-CPU: a thread that +// blocks on one CPU and resumes on another is still measured. +// - Per-probe metrics account for every drop path. + +#include "bpfdefs.h" +#include "tracemgmt.h" +#include "types.h" + +typedef struct OffCpuTimeMetrics { + u64 switch_outs; + u64 switch_ins; + u64 unmatched_switch_ins; + u64 filtered; + u64 emitted; + u64 state_update_failures; +} OffCpuTimeMetrics; + +// OFF_CPU_TIME_STATE_UNAVAILABLE marks a start whose thread state could not +// be read, either because the tracepoint field offset was not resolved or +// because the read failed. Userspace attaches no state label in that case. +#define OFF_CPU_TIME_STATE_UNAVAILABLE (~0ULL) + +typedef struct OffCpuTimeStart { + u64 timestamp; + // prev_state as reported by the sched_switch tracepoint at switch-out: + // 0 for a runnable thread that gave up the CPU, TASK_REPORT bits for + // voluntary blocking, TASK_REPORT_MAX for involuntary preemption. + u64 state; +} OffCpuTimeStart; + +// off_cpu_time_starts records when each thread left the CPU. The pending set +// is every sleeping thread in the system, and a thread that exits while off +// CPU leaves its entry behind with no deterministic cleanup hook (the final +// switch-out of a dying task happens after every exit tracepoint). It is +// therefore an LRU hash: leaked and coldest entries are recycled under +// pressure instead of failing new inserts. Unlike the built-in off-CPU +// profiling map this is not per-CPU, so migrated wakeups still match. +struct off_cpu_time_starts_t { + __uint(type, BPF_MAP_TYPE_LRU_HASH); + __type(key, u64); // pid_tgid + __type(value, OffCpuTimeStart); + __uint(max_entries, 65536); // resized at load time from probe configuration +} off_cpu_time_starts SEC(".maps"); + +struct off_cpu_time_metrics_t { + __uint(type, BPF_MAP_TYPE_PERCPU_ARRAY); + __type(key, u32); + __type(value, OffCpuTimeMetrics); + __uint(max_entries, 1); +} off_cpu_time_metrics SEC(".maps"); + +// These values are set independently for each loaded probe instance. +BPF_RODATA_VAR(u16, off_cpu_time_origin, 0) +BPF_RODATA_VAR(u64, off_cpu_time_min_ns, 0) +BPF_RODATA_VAR(u32, off_cpu_time_sample_threshold, 0xffffffff) + +// Regular tracepoint programs must access their context fields directly; +// bpf_probe_read_kernel() against the tracepoint context can fail even when +// the field lies in range. We still validate this layout against tracefs at +// load time before enabling the read. The profiler supports only 64-bit +// x86_64/aarch64 kernels, whose sched_switch format places the signed long +// prev_state at byte 32 with size 8. +typedef struct SchedSwitchContext { + u8 common_and_previous_task[32]; + s64 prev_state; +} SchedSwitchContext; + +BPF_RODATA_VAR(u32, off_cpu_time_state_enabled, 0) + +static EBPF_INLINE u64 off_cpu_time_read_state(const SchedSwitchContext *ctx) +{ + if (!off_cpu_time_state_enabled) { + return OFF_CPU_TIME_STATE_UNAVAILABLE; + } + return (u64)ctx->prev_state; +} + +static EBPF_INLINE OffCpuTimeMetrics *off_cpu_time_get_metrics() +{ + u32 zero = 0; + return bpf_map_lookup_elem(&off_cpu_time_metrics, &zero); +} + +// off_cpu_time_switch_out runs in the context of the thread that is leaving +// the CPU and records its departure time. It runs on every context switch on +// the system, so it must stay minimal: all filtering and the unwind happen at +// switch-in, once the off-CPU duration is known. +SEC("tracepoint/sched/sched_switch") +int off_cpu_time_switch_out(SchedSwitchContext *ctx) +{ + u64 pid_tgid = bpf_get_current_pid_tgid(); + u32 pid = pid_tgid >> 32; + u32 tid = pid_tgid & 0xffffffff; + if (pid == 0 || tid == 0) { + // Never track the idle task. + return 0; + } + + OffCpuTimeMetrics *metrics = off_cpu_time_get_metrics(); + if (metrics) { + metrics->switch_outs++; + } + + OffCpuTimeStart start = { + .timestamp = bpf_ktime_get_ns(), + // The tracepoint runs in the outgoing thread's context, so prev_state + // describes exactly the thread being recorded: it distinguishes threads + // that blocked voluntarily from runnable threads descheduled by CPU + // pressure, which would otherwise be indistinguishable in the profile. + .state = off_cpu_time_read_state(ctx), + }; + // BPF_ANY: switch-out and switch-in strictly alternate per thread, so an + // existing entry means the matching switch-in was missed (attach race or a + // missed kprobe). The newer timestamp is the correct base either way. + if (bpf_map_update_elem(&off_cpu_time_starts, &pid_tgid, &start, BPF_ANY) != 0) { + if (metrics) { + metrics->state_update_failures++; + } + } + return 0; +} + +// off_cpu_time_switch_in runs in the context of the thread that was just +// scheduled back in, right after the scheduler restored its registers. The +// unwound trace is the stack that was off CPU. Filtering happens here, on the +// measured duration: a wait shorter than min_ns is dropped, and sampling is +// applied per wakeup rather than per switch-out so long waits are never lost. +SEC("kprobe/finish_task_switch") +int off_cpu_time_switch_in(struct pt_regs *ctx) +{ + u64 pid_tgid = bpf_get_current_pid_tgid(); + u32 pid = pid_tgid >> 32; + u32 tid = pid_tgid & 0xffffffff; + if (pid == 0 || tid == 0) { + return 0; + } + + OffCpuTimeMetrics *metrics = off_cpu_time_get_metrics(); + if (metrics) { + metrics->switch_ins++; + } + + OffCpuTimeStart *stored = bpf_map_lookup_elem(&off_cpu_time_starts, &pid_tgid); + if (!stored) { + // First sighting of this thread, or its start was evicted. + if (metrics) { + metrics->unmatched_switch_ins++; + } + return 0; + } + OffCpuTimeStart start = *stored; + bpf_map_delete_elem(&off_cpu_time_starts, &pid_tgid); + + u64 timestamp = bpf_ktime_get_ns(); + if ( + timestamp < start.timestamp || timestamp - start.timestamp < off_cpu_time_min_ns || + (off_cpu_time_sample_threshold != 0xffffffff && + bpf_get_prandom_u32() > off_cpu_time_sample_threshold)) { + if (metrics) { + metrics->filtered++; + } + return 0; + } + + if (metrics) { + metrics->emitted++; + } + DEBUG_PRINT("off-cpu time wakeup pid %d tid %d", pid, tid); + // The switch-out thread state rides in the trace's user-data slot; the + // probe's userspace labeler turns it into a thread.state sample label. + return collect_trace_with_user_data( + ctx, off_cpu_time_origin, pid, tid, timestamp, timestamp - start.timestamp, start.state); +} diff --git a/support/ebpf/tracemgmt.h b/support/ebpf/tracemgmt.h index 3f86dd8a93..20ef2603f8 100644 --- a/support/ebpf/tracemgmt.h +++ b/support/ebpf/tracemgmt.h @@ -929,8 +929,14 @@ get_usermode_regs(struct pt_regs *ctx, UnwindState *state, bool *has_usermode_re #endif // TESTING_COREDUMP -static inline EBPF_INLINE int -collect_trace(struct pt_regs *ctx, u16 origin, u32 pid, u32 tid, u64 trace_timestamp, u64 value) +static inline EBPF_INLINE int collect_trace_internal( + struct pt_regs *ctx, + u16 origin, + u32 pid, + u32 tid, + u64 trace_timestamp, + u64 value, + u64 probe_user_data) { // Only continue processing the trace with a valid origin. if (origin == 0) { @@ -945,12 +951,13 @@ collect_trace(struct pt_regs *ctx, u16 origin, u32 pid, u32 tid, u64 trace_times return -1; } - Trace *trace = &record->trace; - trace->origin = origin; - trace->pid = pid; - trace->tid = tid; - trace->ktime = trace_timestamp; - trace->value = value; + Trace *trace = &record->trace; + trace->origin = origin; + trace->pid = pid; + trace->tid = tid; + trace->ktime = trace_timestamp; + trace->value = value; + trace->probe_user_data = probe_user_data; if (bpf_get_current_comm(&(trace->comm), sizeof(trace->comm)) < 0) { increment_metric(metricID_ErrBPFCurrentComm); } @@ -1000,4 +1007,18 @@ collect_trace(struct pt_regs *ctx, u16 origin, u32 pid, u32 tid, u64 trace_times return -1; } +static inline EBPF_INLINE int +collect_trace(struct pt_regs *ctx, u16 origin, u32 pid, u32 tid, u64 trace_timestamp, u64 value) +{ + return collect_trace_internal(ctx, origin, pid, tid, trace_timestamp, value, 0); +} + +// collect_trace_with_user_data keeps the trace synchronous while carrying a +// probe-defined value to its optional userspace TraceLabeler. +static inline EBPF_INLINE int collect_trace_with_user_data( + struct pt_regs *ctx, u16 origin, u32 pid, u32 tid, u64 trace_timestamp, u64 value, u64 user_data) +{ + return collect_trace_internal(ctx, origin, pid, tid, trace_timestamp, value, user_data); +} + #endif diff --git a/support/ebpf/tracer.ebpf.amd64 b/support/ebpf/tracer.ebpf.amd64 index 851a1bda91..334c3a153b 100644 Binary files a/support/ebpf/tracer.ebpf.amd64 and b/support/ebpf/tracer.ebpf.amd64 differ diff --git a/support/ebpf/tracer.ebpf.arm64 b/support/ebpf/tracer.ebpf.arm64 index 367e23ea0c..76b580aa0c 100644 Binary files a/support/ebpf/tracer.ebpf.arm64 and b/support/ebpf/tracer.ebpf.arm64 differ diff --git a/support/ebpf/types.h b/support/ebpf/types.h index 7c71bf22f5..20f0017897 100644 --- a/support/ebpf/types.h +++ b/support/ebpf/types.h @@ -671,6 +671,10 @@ typedef struct Trace { // e.g. time in nanoseconds for off-CPU traces u64 value; + // probe_user_data carries an optional probe-defined value for userspace + // per-origin labelers. It is not part of the profile sample value. + u64 probe_user_data; + // The CPU that captured this trace. u32 cpu_id; diff --git a/support/types.go b/support/types.go index 2197541632..b32deb7d81 100644 --- a/support/types.go +++ b/support/types.go @@ -160,6 +160,7 @@ type Trace struct { Num_kernel_frames uint16 Origin uint16 Value uint64 + Probe_user_data uint64 Cpu_id uint32 Frame_data [3072]uint64 } @@ -334,7 +335,7 @@ type LuaJITProcInfo struct { const ( Sizeof_StackDelta = 0x4 - Sizeof_Trace = 0x62d8 + Sizeof_Trace = 0x62e0 sizeof_ApmIntProcInfo = 0x8 sizeof_DotnetProcInfo = 0x4 diff --git a/tracer/probes.go b/tracer/probes.go index 33d41f08e6..d6173be0e7 100644 --- a/tracer/probes.go +++ b/tracer/probes.go @@ -14,6 +14,7 @@ import ( cebpf "github.com/cilium/ebpf" "github.com/cilium/ebpf/link" "go.opentelemetry.io/ebpf-profiler/internal/log" + "go.opentelemetry.io/ebpf-profiler/libpf" "go.opentelemetry.io/ebpf-profiler/metrics" "go.opentelemetry.io/ebpf-profiler/reporter/samples" "go.opentelemetry.io/ebpf-profiler/support" @@ -435,6 +436,13 @@ func (c *ProbeContext) LoadProbeUnwinders( bpfVerifierLogLevel, perfProgs.FD(), perCPURecords.FD(), perCPURecordsKp) } +// TraceLabeler is an optional interface for probes. LabelTrace runs before +// symbolization for each trace emitted by the probe's origin and may merge +// low-cardinality, probe-specific sample labels. +type TraceLabeler interface { + LabelTrace(trace *libpf.EbpfTrace) +} + // Probe defines the interface that allows custom stack unwinding trigger points. type Probe interface { // Load attaches a probe that triggers stack unwinding. @@ -493,6 +501,9 @@ func (t *Tracer) Enable(p Probe) error { t.origins.unregister(originID) return errors.New("probe returned a nil link") } + if labeler, ok := p.(TraceLabeler); ok { + t.traceLabelers.Store(originID, labeler) + } t.hooks[hookPoint{group: "probe", name: fmt.Sprintf("%d", originID)}] = lnk return nil diff --git a/tracer/tracer.go b/tracer/tracer.go index ea600b1ae4..2a3f07e65b 100644 --- a/tracer/tracer.go +++ b/tracer/tracer.go @@ -165,6 +165,9 @@ type Tracer struct { // profile type metadata is looked up by. origins *originRegistry + // traceLabelers maps dynamic origin IDs to optional probe labelers. + traceLabelers sync.Map + // probeMetricCollectors are registered dynamically by custom probes. probeMetricsMu sync.RWMutex probeMetricCollectors map[uint64]*probeMetricCollector @@ -1214,6 +1217,7 @@ func (t *Tracer) loadBpfTrace(raw []byte) (*libpf.EbpfTrace, error) { TID: libpf.PID(ptr.Tid), Origin: ptr.Origin, Value: int64(ptr.Value), + ProbeUserData: ptr.Probe_user_data, KTime: int64(ptr.Ktime), CpuID: ptr.Cpu_id, EnvVars: procMeta.EnvVariables, @@ -1514,6 +1518,9 @@ func (t *Tracer) AttachProbes(probes []string) error { } func (t *Tracer) HandleTrace(bpfTrace *libpf.EbpfTrace) { + if labeler, ok := t.traceLabelers.Load(bpfTrace.Origin); ok { + labeler.(TraceLabeler).LabelTrace(bpfTrace) + } t.processManager.HandleTrace(bpfTrace, t.origins.lookup(bpfTrace.Origin)) // Reclaim the EbpfTrace