From ed0bb8cf7f927360fadb0f0359281f4ccdb089c6 Mon Sep 17 00:00:00 2001 From: zuub-don Date: Fri, 7 Aug 2026 16:59:38 -0700 Subject: [PATCH 1/3] Attribute ring-buffer loss to the CPU and probe that lost it A reserve failure told a reader only that the capture had a hole somewhere. The kernel already held finer detail and userspace threw it away: the telemetry counters live in a per-CPU array, and the readout summed across CPUs before the numbers reached the stream. Report two independent projections of the same total rather than their cross product. `kernel_loss_by_cpu` bounds a hole to a core and costs nothing, being the map contents the lookup already returned. `kernel_loss_by_probe` names the kernel function or TC/XDP program that could not emit, which is what identifies the leg of a path a hole belongs to. Only CPUs and probes that lost something appear; an empty array positively states that none did. The per-probe attribution needs new kernel state: a preallocated per-CPU hash keyed by probe site, so the failure path never allocates and needs no atomic. All sixteen reserve-failure sites now route through one helper that bumps the authoritative total first and adds attribution second. A probe plan that overflows the map degrades to `kernel_unattributed_reserve_failures` rather than misfiling loss against the wrong probe. Because the total and the attribution are separate map reads taken while probes still fire, userspace reads them in the opposite order to the kernel's writes. That keeps the breakdown an undercount of the total it refines rather than, as measured before the ordering was fixed, a count that could exceed it. Reliability gains three optional fields. They are defaulted on read and absent from the schema's required list, so captures recorded before this change still parse and replay. Co-Authored-By: Claude Opus 5 (1M context) --- crates/skbx-cli/src/main.rs | 247 ++++++++++++++++++++++++++++- crates/skbx-contract/src/lib.rs | 255 +++++++++++++++++++++++++++++- crates/skbx-sensor/bpf/skbx.bpf.c | 130 +++++++++++---- crates/skbx-sensor/src/lib.rs | 20 +-- crates/skbx-sensor/src/live.rs | 68 ++++++-- crates/skbx-sensor/src/raw.rs | 219 ++++++++++++++++++++++++- docs/architecture.md | 33 ++++ 7 files changed, 907 insertions(+), 65 deletions(-) diff --git a/crates/skbx-cli/src/main.rs b/crates/skbx-cli/src/main.rs index c5d9478..6f220a5 100644 --- a/crates/skbx-cli/src/main.rs +++ b/crates/skbx-cli/src/main.rs @@ -4,9 +4,9 @@ use nix::sched::{CloneFlags, setns}; use skbx_contract::{ BpfMapOperation, BpfMapOperationKind, BpfProgramAction, BpfProgramKind, BpfProgramPhase, BpfProgramRef, BtfDump, CONTRACT_VERSION, CaptureEnd, CaptureFilters, CaptureLimits, - CaptureStart, Describe, Envelope, EventAssociation, FunctionRef, MatchOrigin, MetadataEncoding, - MetadataScalar, MetadataValue, PacketMeta, PacketTuple, PresentedTimestamp, Reliability, - StopReason, TimestampMode, TraceEvent, + CaptureStart, Describe, Envelope, EventAssociation, FunctionRef, KernelCpuLoss, + KernelProbeLoss, MatchOrigin, MetadataEncoding, MetadataScalar, MetadataValue, PacketMeta, + PacketTuple, PresentedTimestamp, Reliability, StopReason, TimestampMode, TraceEvent, }; use skbx_core::{ BoundedMap, DEFAULT_BTF_PATH, DropReasonTable, SymbolTable, build_dynamic_probe_plan, @@ -543,6 +543,35 @@ fn run(cli: Cli) -> Result { summary.reliability.userspace_decode_failures, summary.reliability.output_failures ); + if !summary.reliability.kernel_loss_by_cpu.is_empty() { + let cpus = summary + .reliability + .kernel_loss_by_cpu + .iter() + .map(|loss| { + format!( + "cpu{}=reserve:{},read:{}", + loss.cpu, loss.reserve_failures, loss.read_failures + ) + }) + .collect::>() + .join(" "); + println!("Kernel loss by CPU: {cpus}"); + } + for loss in &summary.reliability.kernel_loss_by_probe { + let site = match (&loss.function, loss.program_id) { + (Some(function), _) => display_function(function), + (None, Some(id)) => format!("bpf_program:{id}"), + (None, None) => "unknown".into(), + }; + println!("Kernel loss at {site}: reserve={}", loss.reserve_failures); + } + if summary.reliability.kernel_unattributed_reserve_failures != 0 { + println!( + "Kernel loss unattributed: reserve={} (probe-site map full; per-probe list undercounts)", + summary.reliability.kernel_unattributed_reserve_failures + ); + } } } Ok(if summary.complete { 0 } else { 3 }) @@ -1105,11 +1134,18 @@ fn capture( } CaptureWriter::Segmented(writer) => { writer.write_event(&event, || { - let current = sensor.stats()?.into_reliability( + // Attribution first, total second. The kernel bumps + // the total before the per-probe map, so reading in + // that same order lets a failure land in between and + // be attributed without being counted, leaving the + // breakdown larger than the total it refines. + let probes = kernel_loss_by_probe(&sensor, &symbols)?; + let mut current = sensor.stats()?.into_reliability( sensor.recursion_misses()?, sensor.decode_failures(), sensor.enrichment_failures(), ); + current.kernel_loss_by_probe = probes; let segment = reliability_delta(¤t, &reliability_checkpoint); reliability_checkpoint = current; Ok(segment) @@ -1120,12 +1156,16 @@ fn capture( } } + // Attribution before total, so the breakdown can only ever undercount + // the number it refines. See the segment boundary above. + let probes = kernel_loss_by_probe(&sensor, &symbols)?; let stats = sensor.stats()?; - let reliability = stats.into_reliability( + let mut reliability = stats.into_reliability( sensor.recursion_misses()?, sensor.decode_failures(), sensor.enrichment_failures(), ); + reliability.kernel_loss_by_probe = probes; let complete = reliability.complete(); let end = CaptureEnd { schema: CONTRACT_VERSION.into(), @@ -1677,9 +1717,104 @@ fn reliability_delta(current: &Reliability, previous: &Reliability) -> Reliabili output_failures: current .output_failures .saturating_sub(previous.output_failures), + kernel_loss_by_cpu: kernel_loss_by_cpu_delta( + ¤t.kernel_loss_by_cpu, + &previous.kernel_loss_by_cpu, + ), + kernel_loss_by_probe: kernel_loss_by_probe_delta( + ¤t.kernel_loss_by_probe, + &previous.kernel_loss_by_probe, + ), + kernel_unattributed_reserve_failures: current + .kernel_unattributed_reserve_failures + .saturating_sub(previous.kernel_unattributed_reserve_failures), } } +/// Resolves each losing probe site to a name. A kernel site carries its +/// instruction pointer symbolized against kallsyms; a TC/XDP site carries the +/// BPF program id, which needs no lookup. +fn kernel_loss_by_probe( + sensor: &skbx_sensor::LiveSensor, + symbols: &SymbolTable, +) -> Result> { + Ok(sensor + .probe_loss()? + .into_iter() + .map(|loss| KernelProbeLoss { + function: (loss.site.function_ip != 0).then(|| FunctionRef { + address: format!("{:#x}", loss.site.function_ip), + symbol: symbols.resolve(loss.site.function_ip).map(str::to_owned), + }), + program_id: (loss.site.program_id != 0).then_some(loss.site.program_id), + reserve_failures: loss.reserve_failures, + }) + .collect()) +} + +/// Probes are matched on their site, not their position: the list is ordered by +/// loss, so a probe can move between checkpoints as other probes overtake it. +fn kernel_loss_by_probe_delta( + current: &[KernelProbeLoss], + previous: &[KernelProbeLoss], +) -> Vec { + let site = |loss: &KernelProbeLoss| { + ( + loss.function + .as_ref() + .map(|function| function.address.clone()), + loss.program_id, + ) + }; + let mut segment: Vec = current + .iter() + .filter_map(|now| { + let before = previous.iter().find(|earlier| site(earlier) == site(now)); + let reserve_failures = now + .reserve_failures + .saturating_sub(before.map_or(0, |earlier| earlier.reserve_failures)); + (reserve_failures != 0).then(|| KernelProbeLoss { + function: now.function.clone(), + program_id: now.program_id, + reserve_failures, + }) + }) + .collect(); + segment.sort_by(|a, b| { + b.reserve_failures + .cmp(&a.reserve_failures) + .then_with(|| site(a).cmp(&site(b))) + }); + segment +} + +/// A CPU absent from `previous` lost nothing before this segment, so its +/// current count is entirely this segment's. A CPU that lost nothing new is +/// dropped rather than carried as a zero, matching the live readout's rule +/// that only CPUs with loss appear. +fn kernel_loss_by_cpu_delta( + current: &[KernelCpuLoss], + previous: &[KernelCpuLoss], +) -> Vec { + current + .iter() + .filter_map(|now| { + let before = previous.iter().find(|earlier| earlier.cpu == now.cpu); + let reserve_failures = now + .reserve_failures + .saturating_sub(before.map_or(0, |earlier| earlier.reserve_failures)); + let read_failures = now + .read_failures + .saturating_sub(before.map_or(0, |earlier| earlier.read_failures)); + (reserve_failures != 0 || read_failures != 0).then_some(KernelCpuLoss { + cpu: now.cpu, + reserve_failures, + read_failures, + }) + }) + .collect() +} + fn packet_tuple(raw: &skbx_sensor::RawPacketTuple) -> Option { let (source, destination) = match raw.l3_protocol { 0x0800 => ( @@ -2037,6 +2172,108 @@ fn install_signal_handlers() { mod tests { use super::*; + fn loss(cpu: u32, reserve_failures: u64, read_failures: u64) -> KernelCpuLoss { + KernelCpuLoss { + cpu, + reserve_failures, + read_failures, + } + } + + #[test] + fn segment_loss_is_attributed_per_cpu_not_just_totalled() { + let previous = Reliability { + kernel_reserve_failures: 4, + kernel_loss_by_cpu: vec![loss(1, 4, 0)], + ..Reliability::default() + }; + let current = Reliability { + kernel_reserve_failures: 11, + kernel_loss_by_cpu: vec![loss(1, 6, 0), loss(5, 5, 3)], + ..Reliability::default() + }; + + let segment = reliability_delta(¤t, &previous); + + assert_eq!(segment.kernel_reserve_failures, 7); + assert_eq!( + segment.kernel_loss_by_cpu, + vec![loss(1, 2, 0), loss(5, 5, 3)] + ); + } + + fn probe(symbol: &str, reserve_failures: u64) -> KernelProbeLoss { + KernelProbeLoss { + function: Some(FunctionRef { + address: format!("0x{:x}", symbol.len()), + symbol: Some(symbol.into()), + }), + program_id: None, + reserve_failures, + } + } + + #[test] + fn segment_probe_loss_matches_sites_across_a_reordering() { + // `nf_hook_slow` overtakes `ip_rcv` between checkpoints, so positional + // matching would subtract each probe's count from the other's. + let previous = Reliability { + kernel_loss_by_probe: vec![probe("ip_rcv", 10), probe("nf_hook_slow", 3)], + ..Reliability::default() + }; + let current = Reliability { + kernel_loss_by_probe: vec![probe("nf_hook_slow", 40), probe("ip_rcv", 12)], + ..Reliability::default() + }; + + let segment = reliability_delta(¤t, &previous); + + assert_eq!( + segment.kernel_loss_by_probe, + vec![probe("nf_hook_slow", 37), probe("ip_rcv", 2)] + ); + } + + #[test] + fn a_probe_absent_from_the_previous_checkpoint_counts_in_full() { + let current = Reliability { + kernel_loss_by_probe: vec![probe("ip_rcv", 5)], + ..Reliability::default() + }; + + let segment = reliability_delta(¤t, &Reliability::default()); + + assert_eq!(segment.kernel_loss_by_probe, vec![probe("ip_rcv", 5)]); + } + + #[test] + fn a_probe_with_no_new_loss_leaves_the_segment_breakdown() { + let previous = Reliability { + kernel_loss_by_probe: vec![probe("ip_rcv", 5)], + ..Reliability::default() + }; + + assert!( + reliability_delta(&previous, &previous) + .kernel_loss_by_probe + .is_empty() + ); + } + + #[test] + fn a_cpu_with_no_new_loss_leaves_the_segment_breakdown() { + let previous = Reliability { + kernel_reserve_failures: 4, + kernel_loss_by_cpu: vec![loss(1, 4, 0)], + ..Reliability::default() + }; + + let segment = reliability_delta(&previous, &previous); + + assert_eq!(segment.kernel_reserve_failures, 0); + assert!(segment.kernel_loss_by_cpu.is_empty()); + } + #[test] fn pwru_text_presentation_flags_parse_explicit_booleans() { let cli = Cli::try_parse_from([ diff --git a/crates/skbx-contract/src/lib.rs b/crates/skbx-contract/src/lib.rs index e42b563..d708790 100644 --- a/crates/skbx-contract/src/lib.rs +++ b/crates/skbx-contract/src/lib.rs @@ -631,6 +631,53 @@ pub struct CaptureSegmentEnd { pub next_seq: Option, } +/// Observation loss attributed to the CPU that could not emit it. +/// +/// The kernel keeps its counters in a per-CPU map, so this attribution is +/// already paid for; summing it away on readout discarded the only locality +/// the kernel side carries. It bounds a hole to a core, not to a probe: a +/// non-zero entry says this CPU failed to emit something, not which function +/// was being observed at the time. +/// +/// Only CPUs that lost something appear. An absent CPU lost nothing — it is +/// not an unobserved CPU. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct KernelCpuLoss { + pub cpu: u32, + pub reserve_failures: u64, + pub read_failures: u64, +} + +/// Observation loss attributed to the probe that could not emit it. +/// +/// This is the finer of the two attributions: it names the leg of the path a +/// hole belongs to, where [`KernelCpuLoss`] only bounds it to a core. Exactly +/// one of `function` and `program_id` is set — a kernel-function probe carries +/// the function it was attached to, a TC/XDP probe the BPF program id. +/// +/// The list is not exhaustive, for two reasons, so a reader that needs the true +/// number of holes must use `kernel_reserve_failures` rather than summing it. +/// When the kernel's probe-site map is full, further failures are counted in +/// `Reliability::kernel_unattributed_reserve_failures` instead. And the totals +/// and the breakdown are separate map reads taken while probes are still +/// firing, so failures landing between them are counted without being +/// attributed. Both effects run the same direction: over a whole capture this +/// list undercounts, and never exceeds, the total it refines. +/// +/// A single segment footer is the one exception. It carries the difference +/// between two checkpoints of two independently sampled series, so if the +/// earlier checkpoint lagged further behind than the later one, that segment's +/// breakdown can exceed the segment's own total by the difference. It is left +/// as measured rather than clamped, because clamping would invent a number the +/// kernel never reported. The undercount still holds across the segments summed +/// together. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct KernelProbeLoss { + pub function: Option, + pub program_id: Option, + pub reserve_failures: u64, +} + #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct Reliability { pub kernel_reserve_failures: u64, @@ -643,6 +690,23 @@ pub struct Reliability { #[serde(default)] pub userspace_enrichment_failures: u64, pub output_failures: u64, + /// Per-CPU breakdown of the kernel reserve/read failures totalled above. + /// + /// Always emitted, so an empty array positively states that no CPU lost + /// anything rather than leaving the reader to guess whether the writer + /// knew. Defaulted on read so captures written before this field parse. + #[serde(default)] + pub kernel_loss_by_cpu: Vec, + /// Per-probe breakdown of the same reserve failures, ordered by loss + /// descending so the worst leg reads first. Independent of the per-CPU + /// view above rather than a cross product of it. + #[serde(default)] + pub kernel_loss_by_probe: Vec, + /// Reserve failures that happened but could not be filed against a probe, + /// because the kernel's probe-site map was full. Non-zero means + /// `kernel_loss_by_probe` undercounts by exactly this much. + #[serde(default)] + pub kernel_unattributed_reserve_failures: u64, } impl Reliability { @@ -1142,7 +1206,38 @@ pub fn json_schema() -> serde_json::Value { "kernel_recursion_misses": {"type": "integer", "minimum": 0}, "userspace_decode_failures": {"type": "integer", "minimum": 0}, "userspace_enrichment_failures": {"type": "integer", "minimum": 0}, - "output_failures": {"type": "integer", "minimum": 0} + "output_failures": {"type": "integer", "minimum": 0}, + "kernel_loss_by_cpu": { + "type": "array", + "items": {"$ref": "#/$defs/KernelCpuLoss"} + }, + "kernel_loss_by_probe": { + "type": "array", + "items": {"$ref": "#/$defs/KernelProbeLoss"} + }, + "kernel_unattributed_reserve_failures": {"type": "integer", "minimum": 0} + }, + "additionalProperties": false + }, + "KernelProbeLoss": { + "type": "object", + "required": ["function", "program_id", "reserve_failures"], + "properties": { + "function": { + "oneOf": [{"$ref": "#/$defs/FunctionRef"}, {"type": "null"}] + }, + "program_id": {"type": ["integer", "null"], "minimum": 0}, + "reserve_failures": {"type": "integer", "minimum": 0} + }, + "additionalProperties": false + }, + "KernelCpuLoss": { + "type": "object", + "required": ["cpu", "reserve_failures", "read_failures"], + "properties": { + "cpu": {"type": "integer", "minimum": 0}, + "reserve_failures": {"type": "integer", "minimum": 0}, + "read_failures": {"type": "integer", "minimum": 0} }, "additionalProperties": false } @@ -1208,4 +1303,162 @@ mod tests { fn schema_is_version_pinned() { assert_eq!(json_schema()["$id"], EVENT_SCHEMA); } + + #[test] + fn per_cpu_loss_survives_a_serialization_round_trip() { + let reliability = Reliability { + kernel_reserve_failures: 9, + kernel_loss_by_cpu: vec![ + KernelCpuLoss { + cpu: 3, + reserve_failures: 4, + read_failures: 0, + }, + KernelCpuLoss { + cpu: 11, + reserve_failures: 5, + read_failures: 2, + }, + ], + ..Reliability::default() + }; + let encoded = serde_json::to_string(&reliability).expect("reliability serializes"); + + assert_eq!( + serde_json::from_str::(&encoded).expect("reliability parses"), + reliability + ); + } + + #[test] + fn a_lossless_capture_states_an_empty_breakdown_rather_than_omitting_it() { + let encoded = serde_json::to_value(Reliability::default()).expect("reliability serializes"); + + assert_eq!( + encoded["kernel_loss_by_cpu"], + serde_json::json!([]), + "an omitted field would be ambiguous between no loss and an older writer" + ); + } + + #[test] + fn captures_written_before_per_cpu_attribution_still_parse() { + let legacy = serde_json::json!({ + "kernel_reserve_failures": 2, + "kernel_read_failures": 0, + "kernel_filtered_events": 0, + "userspace_decode_failures": 0, + "userspace_enrichment_failures": 0, + "output_failures": 0 + }); + + let reliability: Reliability = + serde_json::from_value(legacy).expect("legacy reliability parses"); + assert_eq!(reliability.kernel_reserve_failures, 2); + assert!(reliability.kernel_loss_by_cpu.is_empty()); + assert!(!reliability.complete()); + } + + #[test] + fn per_probe_loss_names_the_leg_and_admits_undercounting() { + let reliability = Reliability { + kernel_reserve_failures: 12, + kernel_unattributed_reserve_failures: 5, + kernel_loss_by_probe: vec![ + KernelProbeLoss { + function: Some(FunctionRef { + address: "0xffffffff81234560".into(), + symbol: Some("nf_hook_slow".into()), + }), + program_id: None, + reserve_failures: 7, + }, + KernelProbeLoss { + function: None, + program_id: Some(42), + reserve_failures: 0, + }, + ], + ..Reliability::default() + }; + let encoded = serde_json::to_string(&reliability).expect("reliability serializes"); + + assert_eq!( + serde_json::from_str::(&encoded).expect("reliability parses"), + reliability + ); + // The per-probe list is a subset, so the total is the only number a + // reader may treat as the true count of holes. Deliberately `<=` and + // not `==`: the breakdown is read from a different map than the total, + // while probes are still firing, so it may legitimately lag. + let attributed: u64 = reliability + .kernel_loss_by_probe + .iter() + .map(|loss| loss.reserve_failures) + .sum(); + assert!( + attributed + reliability.kernel_unattributed_reserve_failures + <= reliability.kernel_reserve_failures + ); + assert!(!reliability.complete()); + } + + #[test] + fn captures_written_before_per_probe_attribution_still_parse() { + let legacy = serde_json::json!({ + "kernel_reserve_failures": 2, + "kernel_read_failures": 0, + "kernel_filtered_events": 0, + "userspace_decode_failures": 0, + "userspace_enrichment_failures": 0, + "output_failures": 0, + "kernel_loss_by_cpu": [{"cpu": 1, "reserve_failures": 2, "read_failures": 0}] + }); + + let reliability: Reliability = + serde_json::from_value(legacy).expect("legacy reliability parses"); + assert_eq!(reliability.kernel_loss_by_cpu.len(), 1); + assert!(reliability.kernel_loss_by_probe.is_empty()); + assert_eq!(reliability.kernel_unattributed_reserve_failures, 0); + } + + #[test] + fn schema_admits_the_per_probe_loss_breakdown() { + let schema = json_schema(); + let reliability = &schema["$defs"]["Reliability"]; + + assert_eq!( + reliability["properties"]["kernel_loss_by_probe"]["items"]["$ref"], + "#/$defs/KernelProbeLoss" + ); + assert!( + reliability["properties"]["kernel_unattributed_reserve_failures"]["type"] == "integer" + ); + // The site reference must resolve, or every lossy capture fails + // validation on a dangling $ref. + assert!(schema["$defs"]["FunctionRef"].is_object()); + assert!(schema["$defs"]["KernelProbeLoss"].is_object()); + } + + #[test] + fn schema_admits_the_per_cpu_loss_breakdown() { + let schema = json_schema(); + let reliability = &schema["$defs"]["Reliability"]; + + // Reliability is additionalProperties:false, so an emitted field that + // the schema does not name would make every capture fail validation. + assert_eq!( + reliability["properties"]["kernel_loss_by_cpu"]["items"]["$ref"], + "#/$defs/KernelCpuLoss" + ); + assert!(schema["$defs"]["KernelCpuLoss"].is_object()); + // Not required: older captures legitimately lack it. + assert!( + !reliability["required"] + .as_array() + .expect("required is an array") + .iter() + .any(|field| field == "kernel_loss_by_cpu") + ); + } } diff --git a/crates/skbx-sensor/bpf/skbx.bpf.c b/crates/skbx-sensor/bpf/skbx.bpf.c index 00b2f6f..3569651 100644 --- a/crates/skbx-sensor/bpf/skbx.bpf.c +++ b/crates/skbx-sensor/bpf/skbx.bpf.c @@ -295,8 +295,27 @@ struct kernel_stats { __u64 reserve_failures; __u64 read_failures; __u64 filtered_events; + __u64 unattributed_reserve_failures; }; +_Static_assert(sizeof(struct kernel_stats) == 32, + "kernel stats ABI changed"); + +/* Identifies the probe that could not emit. A kernel-function probe is named + * by its instruction pointer; a TC/XDP program probe by the BPF program id. + * They occupy separate fields rather than one tagged word so that nothing has + * to assume a program id can never look like a kernel text address. */ +struct probe_site_key { + __u64 function_ip; + __u32 program_id; + __u32 _pad; +}; + +_Static_assert(sizeof(struct probe_site_key) == 16, + "probe site key ABI changed"); + +#define MAX_PROBE_SITES 512 + struct metadata_access { __u32 offsets[MAX_METADATA_ACCESS_STEPS]; __u8 dereference_mask; @@ -371,6 +390,18 @@ struct { __type(value, struct kernel_stats); } telemetry SEC(".maps"); +/* Per-CPU so the increment needs no atomic, and so a hole keeps both the probe + * it belongs to and the core it happened on. Preallocated by the verifier at + * load time, so the failure path never allocates. Sized well above any probe + * plan; a plan that outgrows it degrades to the unattributed counter rather + * than to a wrong attribution. */ +struct { + __uint(type, BPF_MAP_TYPE_PERCPU_HASH); + __uint(max_entries, MAX_PROBE_SITES); + __type(key, struct probe_site_key); + __type(value, __u64); +} probe_site_loss SEC(".maps"); + struct { __uint(type, BPF_MAP_TYPE_STACK_TRACE); __uint(max_entries, 1024); @@ -457,6 +488,51 @@ static __always_inline struct kernel_stats *stats(void) return bpf_map_lookup_elem(&telemetry, &key); } +static __always_inline struct probe_site_key kernel_probe_site( + struct pt_regs *ctx) +{ + struct probe_site_key site; + + __builtin_memset(&site, 0, sizeof(site)); + site.function_ip = (__u64)PT_REGS_IP(ctx); + return site; +} + +static __always_inline struct probe_site_key program_probe_site(void) +{ + struct probe_site_key site; + + __builtin_memset(&site, 0, sizeof(site)); + site.program_id = CONFIG.dynamic_program_id; + return site; +} + +/* The total stays authoritative: it is bumped unconditionally, and the + * attribution is added on top. A capture whose probe plan overflowed the map + * therefore still reports the right number of holes, with the surplus landing + * in unattributed_reserve_failures instead of being silently misfiled. */ +static __always_inline void record_reserve_failure( + struct kernel_stats *counters, struct probe_site_key site) +{ + __u64 *attributed; + __u64 first = 1; + + if (counters) + counters->reserve_failures++; + attributed = bpf_map_lookup_elem(&probe_site_loss, &site); + if (attributed) { + *attributed += 1; + return; + } + /* BPF_ANY rather than BPF_NOEXIST: another CPU may have inserted this key + * between the lookup and here, and this CPU's own slot is zero either way, + * so overwriting it with 1 is correct. The update then fails only when the + * map is genuinely full, which is the one case worth counting. */ + if (bpf_map_update_elem(&probe_site_loss, &site, &first, BPF_ANY) && + counters) + counters->unattributed_reserve_failures++; +} + static __always_inline int read_netns(struct sk_buff *skb, __u32 *netns_id) { struct net_device *device = 0; @@ -1585,8 +1661,7 @@ static __always_inline int trace_skb_associated(struct pt_regs *ctx, bpf_map_lookup_elem(&btf_scratch, &key); if (!record) { - if (counters) - counters->reserve_failures++; + record_reserve_failure(counters, kernel_probe_site(ctx)); return 0; } __builtin_memset(&record->record, 0, sizeof(record->record)); @@ -1598,9 +1673,8 @@ static __always_inline int trace_skb_associated(struct pt_regs *ctx, record->components |= BTF_RECORD_COMPONENT_METADATA; } fill_btf_dumps(skb, &record->dumps, counters); - if (bpf_ringbuf_output(&events, record, sizeof(*record), 0) && - counters) - counters->reserve_failures++; + if (bpf_ringbuf_output(&events, record, sizeof(*record), 0)) + record_reserve_failure(counters, kernel_probe_site(ctx)); return 0; } if (CONFIG.metadata_count) { @@ -1608,8 +1682,7 @@ static __always_inline int trace_skb_associated(struct pt_regs *ctx, bpf_ringbuf_reserve(&events, sizeof(*record), 0); if (!record) { - if (counters) - counters->reserve_failures++; + record_reserve_failure(counters, kernel_probe_site(ctx)); return 0; } __builtin_memset(record, 0, sizeof(*record)); @@ -1622,8 +1695,7 @@ static __always_inline int trace_skb_associated(struct pt_regs *ctx, bpf_ringbuf_reserve(&events, sizeof(*event), 0); if (!event) { - if (counters) - counters->reserve_failures++; + record_reserve_failure(counters, kernel_probe_site(ctx)); return 0; } fill_trace_event(ctx, skb, association, match, identity, event, @@ -1708,8 +1780,7 @@ int skbx_trace_tc(__u64 *ctx) (struct skbx_program_btf_trace_event *)scratch; if (!record) { - if (counters) - counters->reserve_failures++; + record_reserve_failure(counters, program_probe_site()); return 0; } __builtin_memset(&record->record, 0, sizeof(record->record)); @@ -1721,9 +1792,8 @@ int skbx_trace_tc(__u64 *ctx) record->components |= BTF_RECORD_COMPONENT_METADATA; } fill_btf_dumps(skb, &record->dumps, counters); - if (bpf_ringbuf_output(&events, record, sizeof(*record), 0) && - counters) - counters->reserve_failures++; + if (bpf_ringbuf_output(&events, record, sizeof(*record), 0)) + record_reserve_failure(counters, program_probe_site()); return 0; } if (CONFIG.metadata_count) { @@ -1731,8 +1801,7 @@ int skbx_trace_tc(__u64 *ctx) bpf_ringbuf_reserve(&events, sizeof(*record), 0); if (!record) { - if (counters) - counters->reserve_failures++; + record_reserve_failure(counters, program_probe_site()); return 0; } fill_program_trace_event(ctx, skb, match, identity, @@ -1744,8 +1813,7 @@ int skbx_trace_tc(__u64 *ctx) bpf_ringbuf_reserve(&events, sizeof(*record), 0); if (!record) { - if (counters) - counters->reserve_failures++; + record_reserve_failure(counters, program_probe_site()); return 0; } fill_program_trace_event(ctx, skb, match, identity, record, @@ -1802,8 +1870,7 @@ int skbx_trace_xdp(__u64 *ctx) bpf_ringbuf_reserve(&events, sizeof(*record), 0); if (!record) { - if (counters) - counters->reserve_failures++; + record_reserve_failure(counters, program_probe_site()); return 0; } fill_xdp_program_trace_event(ctx, xdp, match, identity, @@ -1816,8 +1883,7 @@ int skbx_trace_xdp(__u64 *ctx) bpf_ringbuf_reserve(&events, sizeof(*record), 0); if (!record) { - if (counters) - counters->reserve_failures++; + record_reserve_failure(counters, program_probe_site()); return 0; } fill_xdp_program_trace_event(ctx, xdp, match, identity, @@ -1853,8 +1919,7 @@ int skbx_trace_xdp_exit(__u64 *ctx) bpf_ringbuf_reserve(&events, sizeof(*record), 0); if (!record) { - if (counters) - counters->reserve_failures++; + record_reserve_failure(counters, program_probe_site()); return 0; } fill_xdp_program_trace_event(ctx, xdp, match, identity, @@ -1868,8 +1933,7 @@ int skbx_trace_xdp_exit(__u64 *ctx) bpf_ringbuf_reserve(&events, sizeof(*record), 0); if (!record) { - if (counters) - counters->reserve_failures++; + record_reserve_failure(counters, program_probe_site()); return 0; } fill_xdp_program_trace_event(ctx, xdp, match, identity, @@ -1964,8 +2028,7 @@ static __always_inline int trace_map_associated(struct pt_regs *ctx, bpf_map_lookup_elem(&btf_scratch, &key); if (!record) { - if (counters) - counters->reserve_failures++; + record_reserve_failure(counters, kernel_probe_site(ctx)); return 0; } __builtin_memset(&record->record, 0, sizeof(record->record)); @@ -1981,9 +2044,8 @@ static __always_inline int trace_map_associated(struct pt_regs *ctx, } fill_btf_dumps((struct sk_buff *)*skb_addr, &record->dumps, counters); - if (bpf_ringbuf_output(&events, record, sizeof(*record), 0) && - counters) - counters->reserve_failures++; + if (bpf_ringbuf_output(&events, record, sizeof(*record), 0)) + record_reserve_failure(counters, kernel_probe_site(ctx)); return 0; } if (CONFIG.metadata_count) { @@ -1991,8 +2053,7 @@ static __always_inline int trace_map_associated(struct pt_regs *ctx, bpf_ringbuf_reserve(&events, sizeof(*record), 0); if (!record) { - if (counters) - counters->reserve_failures++; + record_reserve_failure(counters, kernel_probe_site(ctx)); return 0; } __builtin_memset(record, 0, sizeof(*record)); @@ -2006,8 +2067,7 @@ static __always_inline int trace_map_associated(struct pt_regs *ctx, bpf_ringbuf_reserve(&events, sizeof(*record), 0); if (!record) { - if (counters) - counters->reserve_failures++; + record_reserve_failure(counters, kernel_probe_site(ctx)); return 0; } __builtin_memset(record, 0, sizeof(*record)); diff --git a/crates/skbx-sensor/src/lib.rs b/crates/skbx-sensor/src/lib.rs index 26d4787..a7c7228 100644 --- a/crates/skbx-sensor/src/lib.rs +++ b/crates/skbx-sensor/src/lib.rs @@ -5,16 +5,16 @@ mod raw; pub use raw::{ ASSOCIATION_DIRECT, ASSOCIATION_STACK, BPF_PROGRAM_PHASE_ENTRY, BPF_PROGRAM_PHASE_EXIT, BPF_PROGRAM_TC, BPF_PROGRAM_XDP, BTF_DUMP_SHARED_INFO, BTF_DUMP_SK_BUFF, - BTF_RECORD_COMPONENT_MAP, BTF_RECORD_COMPONENT_METADATA, KernelStats, MAP_OPERATION_DELETE, - MAP_OPERATION_LOOKUP, MAP_OPERATION_UPDATE, MAP_READ_KEY_FAILED, MAP_READ_METADATA_FAILED, - MAP_READ_VALUE_FAILED, MATCH_FILTER, MATCH_STACK_ASSOCIATION, MATCH_TRACKED_SKB, - MATCH_TRACKED_XDP, MAX_BTF_DUMP_BYTES, MAX_MAP_CAPTURE_BYTES, MAX_METADATA_PROJECTIONS, - READ_CALLER_FAILED, READ_CB_FAILED, READ_DEVICE_FAILED, READ_IFINDEX_FAILED, READ_LEN_FAILED, - READ_MARK_FAILED, READ_MTU_FAILED, READ_NETNS_FAILED, READ_PROTOCOL_FAILED, - READ_TUNNEL_TUPLE_FAILED, READ_TUPLE_FAILED, RawBpfProgram, RawBtfDumps, RawBtfTraceEvent, - RawMapMetadataTraceEvent, RawMapTraceEvent, RawMetadata, RawMetadataTraceEvent, RawObservation, - RawPacketTuple, RawProgramBtfTraceEvent, RawProgramMetadataTraceEvent, RawProgramTraceEvent, - RawTraceEvent, + BTF_RECORD_COMPONENT_MAP, BTF_RECORD_COMPONENT_METADATA, KernelStats, KernelStatsByCpu, + MAP_OPERATION_DELETE, MAP_OPERATION_LOOKUP, MAP_OPERATION_UPDATE, MAP_READ_KEY_FAILED, + MAP_READ_METADATA_FAILED, MAP_READ_VALUE_FAILED, MATCH_FILTER, MATCH_STACK_ASSOCIATION, + MATCH_TRACKED_SKB, MATCH_TRACKED_XDP, MAX_BTF_DUMP_BYTES, MAX_MAP_CAPTURE_BYTES, + MAX_METADATA_PROJECTIONS, ProbeSiteKey, ProbeSiteLoss, READ_CALLER_FAILED, READ_CB_FAILED, + READ_DEVICE_FAILED, READ_IFINDEX_FAILED, READ_LEN_FAILED, READ_MARK_FAILED, READ_MTU_FAILED, + READ_NETNS_FAILED, READ_PROTOCOL_FAILED, READ_TUNNEL_TUPLE_FAILED, READ_TUPLE_FAILED, + RawBpfProgram, RawBtfDumps, RawBtfTraceEvent, RawMapMetadataTraceEvent, RawMapTraceEvent, + RawMetadata, RawMetadataTraceEvent, RawObservation, RawPacketTuple, RawProgramBtfTraceEvent, + RawProgramMetadataTraceEvent, RawProgramTraceEvent, RawTraceEvent, }; #[cfg(feature = "ebpf")] diff --git a/crates/skbx-sensor/src/live.rs b/crates/skbx-sensor/src/live.rs index 7296e32..36c007e 100644 --- a/crates/skbx-sensor/src/live.rs +++ b/crates/skbx-sensor/src/live.rs @@ -1,4 +1,4 @@ -use crate::{KernelStats, RawObservation}; +use crate::{KernelStats, KernelStatsByCpu, ProbeSiteKey, ProbeSiteLoss, RawObservation}; use libbpf_rs::btf::{Btf, BtfType, TypeId}; use libbpf_rs::query::{ProgInfoIter, ProgInfoQueryOptions, ProgramInfo}; use libbpf_rs::{ @@ -161,6 +161,7 @@ pub struct LiveSensor { ring: NonNull, ring_state: Box, telemetry: MapHandle, + probe_site_loss: MapHandle, stack_traces: MapHandle, _links: Vec, _object: Object, @@ -548,6 +549,7 @@ impl LiveSensor { } let events = map_handle(&object, "events")?; let telemetry = map_handle(&object, "telemetry")?; + let probe_site_loss = map_handle(&object, "probe_site_loss")?; let stack_traces = map_handle(&object, "stack_traces")?; let mut ring_state = Box::::default(); let ring = create_ring(&events, ring_state.as_mut())?; @@ -556,6 +558,7 @@ impl LiveSensor { ring, ring_state, telemetry, + probe_site_loss, stack_traces, _links: links, _object: object, @@ -581,14 +584,16 @@ impl LiveSensor { Ok(self.ring_state.events.drain(..).collect()) } - pub fn stats(&self) -> Result { + /// Values come back one per possible CPU in CPU-id order, and are kept that + /// way. Totalling here would discard which core failed to emit. + pub fn stats(&self) -> Result { let key = 0_u32.to_ne_bytes(); let values = self .telemetry .lookup_percpu(&key, MapFlags::ANY) .map_err(|error| LiveError::Map(error.to_string()))? .ok_or_else(|| LiveError::Map("telemetry key 0 not found".into()))?; - let mut total = KernelStats::default(); + let mut per_cpu = Vec::with_capacity(values.len()); for value in values { let Some(stats) = kernel_stats_from_bytes(&value) else { return Err(LiveError::Map(format!( @@ -596,13 +601,57 @@ impl LiveSensor { value.len() ))); }; - total.reserve_failures = total - .reserve_failures - .saturating_add(stats.reserve_failures); - total.read_failures = total.read_failures.saturating_add(stats.read_failures); - total.filtered_events = total.filtered_events.saturating_add(stats.filtered_events); + per_cpu.push(stats); } - Ok(total) + Ok(KernelStatsByCpu::new(per_cpu)) + } + + /// Reserve failures per probe, summed across CPUs. + /// + /// Ordered by loss descending then by site, because hash iteration order is + /// not stable and a capture footer has to be reproducible. + pub fn probe_loss(&self) -> Result, LiveError> { + let mut losses = Vec::new(); + for key in self.probe_site_loss.keys() { + let Some(site) = ProbeSiteKey::from_bytes(&key) else { + return Err(LiveError::Map(format!( + "probe site key has unexpected size {}", + key.len() + ))); + }; + let Some(values) = self + .probe_site_loss + .lookup_percpu(&key, MapFlags::ANY) + .map_err(|error| LiveError::Map(error.to_string()))? + else { + // The key was evicted between iterating and reading it. The + // global counter already covers this failure, so skipping the + // attribution loses detail, not the fact of the loss. + continue; + }; + let mut reserve_failures = 0_u64; + for value in values { + let bytes: [u8; 8] = value.as_slice().try_into().map_err(|_| { + LiveError::Map(format!( + "probe site value has unexpected size {}", + value.len() + )) + })?; + reserve_failures = reserve_failures.saturating_add(u64::from_ne_bytes(bytes)); + } + if reserve_failures != 0 { + losses.push(ProbeSiteLoss { + site, + reserve_failures, + }); + } + } + losses.sort_by(|a, b| { + b.reserve_failures + .cmp(&a.reserve_failures) + .then_with(|| a.site.cmp(&b.site)) + }); + Ok(losses) } pub fn decode_failures(&self) -> u64 { @@ -970,6 +1019,7 @@ fn kernel_stats_from_bytes(bytes: &[u8]) -> Option { reserve_failures: u64::from_ne_bytes(bytes[0..8].try_into().ok()?), read_failures: u64::from_ne_bytes(bytes[8..16].try_into().ok()?), filtered_events: u64::from_ne_bytes(bytes[16..24].try_into().ok()?), + unattributed_reserve_failures: u64::from_ne_bytes(bytes[24..32].try_into().ok()?), }) } diff --git a/crates/skbx-sensor/src/raw.rs b/crates/skbx-sensor/src/raw.rs index a91512b..ee48369 100644 --- a/crates/skbx-sensor/src/raw.rs +++ b/crates/skbx-sensor/src/raw.rs @@ -1,4 +1,4 @@ -use skbx_contract::Reliability; +use skbx_contract::{KernelCpuLoss, Reliability}; pub const READ_LEN_FAILED: u16 = 1 << 0; pub const READ_PROTOCOL_FAILED: u16 = 1 << 1; @@ -485,23 +485,121 @@ pub struct KernelStats { pub reserve_failures: u64, pub read_failures: u64, pub filtered_events: u64, + /// Reserve failures the kernel could not file against a probe because the + /// probe-site map was full. Counted here rather than dropped so the + /// per-probe breakdown never has to be read as exhaustive. + pub unattributed_reserve_failures: u64, } -impl KernelStats { +/// Identifies the probe that failed to emit, mirroring `struct probe_site_key` +/// in the BPF object. Exactly one of the two fields is set: a kernel-function +/// probe carries its instruction pointer, a TC/XDP program probe its program id. +#[repr(C)] +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, PartialOrd, Ord)] +pub struct ProbeSiteKey { + pub function_ip: u64, + pub program_id: u32, + pub _pad: u32, +} + +impl ProbeSiteKey { + pub const BYTE_LEN: usize = std::mem::size_of::(); + + pub fn from_bytes(bytes: &[u8]) -> Option { + if bytes.len() != Self::BYTE_LEN { + return None; + } + Some(Self { + function_ip: u64::from_ne_bytes(bytes[0..8].try_into().ok()?), + program_id: u32::from_ne_bytes(bytes[8..12].try_into().ok()?), + _pad: 0, + }) + } +} + +/// Reserve failures attributed to one probe, summed across CPUs. +/// +/// The kernel keeps this per-CPU as well, but the emitted contract reports +/// probe and CPU as two independent projections rather than their cross +/// product: a probe-by-core matrix grows with both and answers a question +/// nobody asked. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct ProbeSiteLoss { + pub site: ProbeSiteKey, + pub reserve_failures: u64, +} + +/// Kernel counters as the per-CPU array actually holds them, indexed by CPU id. +/// +/// The counters live in a `BPF_MAP_TYPE_PERCPU_ARRAY`, so the breakdown is free +/// to carry — it is what the lookup already returns. Totalling it at readout +/// threw away the only attribution the kernel side has for a lost observation. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct KernelStatsByCpu { + per_cpu: Vec, +} + +impl KernelStatsByCpu { + /// `per_cpu` is ordered by CPU id, one entry per possible CPU. + pub fn new(per_cpu: Vec) -> Self { + Self { per_cpu } + } + + pub fn per_cpu(&self) -> &[KernelStats] { + &self.per_cpu + } + + pub fn total(&self) -> KernelStats { + self.per_cpu + .iter() + .fold(KernelStats::default(), |mut total, stats| { + total.reserve_failures = total + .reserve_failures + .saturating_add(stats.reserve_failures); + total.read_failures = total.read_failures.saturating_add(stats.read_failures); + total.filtered_events = total.filtered_events.saturating_add(stats.filtered_events); + total.unattributed_reserve_failures = total + .unattributed_reserve_failures + .saturating_add(stats.unattributed_reserve_failures); + total + }) + } + + /// Only CPUs that lost something. A CPU that filtered events but lost none + /// is not loss, so it is omitted rather than reported as a hole. + pub fn loss_by_cpu(&self) -> Vec { + self.per_cpu + .iter() + .enumerate() + .filter(|(_, stats)| stats.reserve_failures != 0 || stats.read_failures != 0) + .map(|(cpu, stats)| KernelCpuLoss { + cpu: u32::try_from(cpu).unwrap_or(u32::MAX), + reserve_failures: stats.reserve_failures, + read_failures: stats.read_failures, + }) + .collect() + } + pub fn into_reliability( self, recursion_misses: u64, decode_failures: u64, enrichment_failures: u64, ) -> Reliability { + let total = self.total(); Reliability { - kernel_reserve_failures: self.reserve_failures, - kernel_read_failures: self.read_failures, - kernel_filtered_events: self.filtered_events, + kernel_reserve_failures: total.reserve_failures, + kernel_read_failures: total.read_failures, + kernel_filtered_events: total.filtered_events, kernel_recursion_misses: recursion_misses, userspace_decode_failures: decode_failures, userspace_enrichment_failures: enrichment_failures, output_failures: 0, + kernel_loss_by_cpu: self.loss_by_cpu(), + kernel_unattributed_reserve_failures: total.unattributed_reserve_failures, + // Filled in by the caller, which owns the symbol table needed to + // turn a probe site address into a function name. + kernel_loss_by_probe: Vec::new(), } } } @@ -666,4 +764,115 @@ mod tests { assert!(dumps.is_some()); assert_eq!(program.expect("program").id, 43); } + + fn stats_by_cpu(per_cpu: &[(u64, u64, u64)]) -> KernelStatsByCpu { + KernelStatsByCpu::new( + per_cpu + .iter() + .map( + |&(reserve_failures, read_failures, filtered_events)| KernelStats { + reserve_failures, + read_failures, + filtered_events, + unattributed_reserve_failures: 0, + }, + ) + .collect(), + ) + } + + #[test] + fn reliability_totals_still_match_the_per_cpu_breakdown() { + let stats = stats_by_cpu(&[(2, 0, 10), (0, 0, 40), (3, 1, 0)]); + let reliability = stats.into_reliability(0, 0, 0); + + assert_eq!(reliability.kernel_reserve_failures, 5); + assert_eq!(reliability.kernel_read_failures, 1); + assert_eq!(reliability.kernel_filtered_events, 50); + assert_eq!( + reliability + .kernel_loss_by_cpu + .iter() + .map(|loss| loss.reserve_failures) + .sum::(), + reliability.kernel_reserve_failures + ); + } + + #[test] + fn loss_is_attributed_to_the_cpu_that_could_not_emit() { + let stats = stats_by_cpu(&[(0, 0, 0), (0, 0, 0), (7, 2, 0)]); + + assert_eq!( + stats.loss_by_cpu(), + vec![KernelCpuLoss { + cpu: 2, + reserve_failures: 7, + read_failures: 2, + }] + ); + } + + #[test] + fn a_cpu_that_only_filtered_is_not_reported_as_loss() { + // Filtering is the probe declining to emit, not a hole in the capture. + let stats = stats_by_cpu(&[(0, 0, 900), (0, 0, 12)]); + + assert!(stats.loss_by_cpu().is_empty()); + assert_eq!(stats.into_reliability(0, 0, 0).kernel_filtered_events, 912); + } + + #[test] + fn unattributed_reserve_failures_total_across_cpus() { + let stats = KernelStatsByCpu::new(vec![ + KernelStats { + reserve_failures: 5, + unattributed_reserve_failures: 2, + ..KernelStats::default() + }, + KernelStats { + reserve_failures: 4, + unattributed_reserve_failures: 4, + ..KernelStats::default() + }, + ]); + + let reliability = stats.into_reliability(0, 0, 0); + + assert_eq!(reliability.kernel_reserve_failures, 9); + assert_eq!(reliability.kernel_unattributed_reserve_failures, 6); + // The total is authoritative; attribution is only ever a subset of it. + assert!( + reliability.kernel_unattributed_reserve_failures <= reliability.kernel_reserve_failures + ); + } + + #[test] + fn probe_site_keys_decode_to_exactly_one_kind_of_site() { + let mut kernel = [0_u8; ProbeSiteKey::BYTE_LEN]; + kernel[0..8].copy_from_slice(&0xffff_ffff_8123_4560_u64.to_ne_bytes()); + let kernel = ProbeSiteKey::from_bytes(&kernel).expect("kernel site decodes"); + assert_eq!(kernel.function_ip, 0xffff_ffff_8123_4560); + assert_eq!(kernel.program_id, 0); + + let mut program = [0_u8; ProbeSiteKey::BYTE_LEN]; + program[8..12].copy_from_slice(&77_u32.to_ne_bytes()); + let program = ProbeSiteKey::from_bytes(&program).expect("program site decodes"); + assert_eq!(program.function_ip, 0); + assert_eq!(program.program_id, 77); + } + + #[test] + fn probe_site_key_rejects_a_wrong_sized_key() { + assert_eq!(ProbeSiteKey::BYTE_LEN, 16); + assert!(ProbeSiteKey::from_bytes(&[0_u8; 12]).is_none()); + } + + #[test] + fn a_lossless_capture_reports_an_empty_breakdown_not_a_missing_one() { + let reliability = stats_by_cpu(&[(0, 0, 0), (0, 0, 0)]).into_reliability(0, 0, 0); + + assert!(reliability.complete()); + assert!(reliability.kernel_loss_by_cpu.is_empty()); + } } diff --git a/docs/architecture.md b/docs/architecture.md index 06c147a..90f51bb 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -26,6 +26,39 @@ The footer carries kernel loss, user-space decode failures, output failures and the stop reason. This prevents a partial trace from silently looking authoritative. +Kernel loss is reported as a total plus two independent breakdowns, not their +cross product. `kernel_loss_by_cpu` attributes a hole to the core it happened +on; the kernel keeps those counters in a per-CPU array, so that view costs +nothing to carry. `kernel_loss_by_probe` attributes it to the probe that could +not emit — the kernel function, or the TC/XDP program id — which is what names +the leg of the path a hole belongs to. Only CPUs and probes that lost something +appear, and an empty array positively states that none did. + +The per-probe attribution is bounded by a fixed-size map. When a probe plan +overflows it, the surplus lands in `kernel_unattributed_reserve_failures` +rather than being misfiled against another probe. The breakdown is also a +separate map read from the totals, taken while probes are still firing, so +failures landing between the two reads are counted without being attributed. +The kernel bumps the total before the attribution and userspace reads them in +the opposite order, which keeps both effects pointing the same way: +`kernel_loss_by_probe` undercounts and never exceeds +`kernel_reserve_failures`, which remains the only authoritative total. + +A rotated segment footer is the exception: it holds the difference between two +checkpoints of two separately sampled series, so a single segment's per-probe +breakdown can exceed that segment's own total by however far the earlier +checkpoint lagged. Measured values are reported rather than clamped, and the +undercount still holds across all segments summed together. + +The per-CPU breakdown has no such gap — it comes from the same map read as the +totals, so it sums to `kernel_reserve_failures` exactly, in whole captures and +in segments alike. + +A hole that is attributed to a probe still says only that this probe failed to +emit at least once. It does not identify which packet lost that observation, so +`complete: false` continues to downgrade every absence in the capture — the +attribution narrows where to look, it does not restore the absence as evidence. + ## Pipeline ```text From 215d8e26dbe5748584bbfc2a8e95611c3c7765a1 Mon Sep 17 00:00:00 2001 From: zuub-don Date: Fri, 7 Aug 2026 19:17:43 -0700 Subject: [PATCH 2/3] Address review: dedupe probe sites and require the breakdown Two findings from CodeRabbit on this branch, both real. probe_loss collected into a list. keys() is bpf_map_get_next_key iteration, which can hand back the same key more than once when entries are inserted while it runs, and probes keep firing throughout the read. Every revisit became a second entry for the same probe and overstated that probe's share of the loss. Collect through a map keyed by site instead, so a revisit overwrites an equivalent reading rather than duplicating it. into_reliability returned a Reliability with kernel_loss_by_probe empty and left the caller to patch it in afterwards. A future third call site that forgot would drop the attribution silently, and a dropped breakdown reads exactly like a capture that lost nothing per probe. Take it as a parameter so the compiler requires it. A third finding, replacing the two optional site fields with an enum so that "exactly one of function and program_id" is unrepresentable, is correct but changes the wire shape. Deferred: the append-only schema is what lets captures recorded before these fields still parse. Co-Authored-By: Claude Opus 5 (1M context) --- crates/skbx-cli/src/main.rs | 8 ++++---- crates/skbx-sensor/src/live.rs | 23 +++++++++++++++++------ crates/skbx-sensor/src/raw.rs | 13 +++++++++---- 3 files changed, 30 insertions(+), 14 deletions(-) diff --git a/crates/skbx-cli/src/main.rs b/crates/skbx-cli/src/main.rs index 6f220a5..f6996c1 100644 --- a/crates/skbx-cli/src/main.rs +++ b/crates/skbx-cli/src/main.rs @@ -1140,12 +1140,12 @@ fn capture( // be attributed without being counted, leaving the // breakdown larger than the total it refines. let probes = kernel_loss_by_probe(&sensor, &symbols)?; - let mut current = sensor.stats()?.into_reliability( + let current = sensor.stats()?.into_reliability( sensor.recursion_misses()?, sensor.decode_failures(), sensor.enrichment_failures(), + probes, ); - current.kernel_loss_by_probe = probes; let segment = reliability_delta(¤t, &reliability_checkpoint); reliability_checkpoint = current; Ok(segment) @@ -1160,12 +1160,12 @@ fn capture( // the number it refines. See the segment boundary above. let probes = kernel_loss_by_probe(&sensor, &symbols)?; let stats = sensor.stats()?; - let mut reliability = stats.into_reliability( + let reliability = stats.into_reliability( sensor.recursion_misses()?, sensor.decode_failures(), sensor.enrichment_failures(), + probes, ); - reliability.kernel_loss_by_probe = probes; let complete = reliability.complete(); let end = CaptureEnd { schema: CONTRACT_VERSION.into(), diff --git a/crates/skbx-sensor/src/live.rs b/crates/skbx-sensor/src/live.rs index 36c007e..8b94c6a 100644 --- a/crates/skbx-sensor/src/live.rs +++ b/crates/skbx-sensor/src/live.rs @@ -6,7 +6,7 @@ use libbpf_rs::{ ProgramType, }; use skbx_contract::{BpfProgramKind, BpfProgramRef, ProbeSpec}; -use std::collections::{BTreeSet, VecDeque}; +use std::collections::{BTreeMap, BTreeSet, VecDeque}; use std::ffi::{OsStr, c_void}; use std::mem; use std::os::fd::{AsFd, AsRawFd}; @@ -610,8 +610,15 @@ impl LiveSensor { /// /// Ordered by loss descending then by site, because hash iteration order is /// not stable and a capture footer has to be reproducible. + /// + /// Collected through a map keyed by site rather than pushed to a list. + /// `keys()` is `bpf_map_get_next_key()` iteration, which may hand back a + /// key more than once when entries are inserted while it runs — and probes + /// keep firing throughout. A list would turn each revisit into a duplicate + /// entry and overstate that probe's share; keying by site makes a revisit + /// simply overwrite an equivalent reading. pub fn probe_loss(&self) -> Result, LiveError> { - let mut losses = Vec::new(); + let mut losses = BTreeMap::new(); for key in self.probe_site_loss.keys() { let Some(site) = ProbeSiteKey::from_bytes(&key) else { return Err(LiveError::Map(format!( @@ -640,12 +647,16 @@ impl LiveSensor { reserve_failures = reserve_failures.saturating_add(u64::from_ne_bytes(bytes)); } if reserve_failures != 0 { - losses.push(ProbeSiteLoss { - site, - reserve_failures, - }); + losses.insert(site, reserve_failures); } } + let mut losses: Vec = losses + .into_iter() + .map(|(site, reserve_failures)| ProbeSiteLoss { + site, + reserve_failures, + }) + .collect(); losses.sort_by(|a, b| { b.reserve_failures .cmp(&a.reserve_failures) diff --git a/crates/skbx-sensor/src/raw.rs b/crates/skbx-sensor/src/raw.rs index ee48369..a9b9fb0 100644 --- a/crates/skbx-sensor/src/raw.rs +++ b/crates/skbx-sensor/src/raw.rs @@ -1,4 +1,4 @@ -use skbx_contract::{KernelCpuLoss, Reliability}; +use skbx_contract::{KernelCpuLoss, KernelProbeLoss, Reliability}; pub const READ_LEN_FAILED: u16 = 1 << 0; pub const READ_PROTOCOL_FAILED: u16 = 1 << 1; @@ -580,11 +580,16 @@ impl KernelStatsByCpu { .collect() } + /// `loss_by_probe` is a parameter rather than a field the caller patches in + /// afterwards: leaving it to the caller means a future third call site can + /// drop the attribution silently, and a dropped breakdown reads exactly + /// like a capture that lost nothing per probe. pub fn into_reliability( self, recursion_misses: u64, decode_failures: u64, enrichment_failures: u64, + loss_by_probe: Vec, ) -> Reliability { let total = self.total(); Reliability { @@ -597,9 +602,9 @@ impl KernelStatsByCpu { output_failures: 0, kernel_loss_by_cpu: self.loss_by_cpu(), kernel_unattributed_reserve_failures: total.unattributed_reserve_failures, - // Filled in by the caller, which owns the symbol table needed to - // turn a probe site address into a function name. - kernel_loss_by_probe: Vec::new(), + // Resolved by the caller, which owns the symbol table needed to turn + // a probe site address into a function name. + kernel_loss_by_probe: loss_by_probe, } } } From efe14cba56fcae3b6a07e77ce19fa7f6321cc8c4 Mon Sep 17 00:00:00 2001 From: zuub-don Date: Fri, 7 Aug 2026 19:18:30 -0700 Subject: [PATCH 3/3] Fix the test call sites the signature change broke into_reliability gained a parameter and four unit tests still passed three arguments. cargo build does not compile tests, so building rather than running make check hid it; the previous commit was pushed red. Also assert the breakdown reaches the footer, which is the behaviour the new parameter exists to guarantee. Co-Authored-By: Claude Opus 5 (1M context) --- crates/skbx-sensor/src/raw.rs | 28 ++++++++++++++++++++++++---- 1 file changed, 24 insertions(+), 4 deletions(-) diff --git a/crates/skbx-sensor/src/raw.rs b/crates/skbx-sensor/src/raw.rs index a9b9fb0..4433b38 100644 --- a/crates/skbx-sensor/src/raw.rs +++ b/crates/skbx-sensor/src/raw.rs @@ -789,7 +789,7 @@ mod tests { #[test] fn reliability_totals_still_match_the_per_cpu_breakdown() { let stats = stats_by_cpu(&[(2, 0, 10), (0, 0, 40), (3, 1, 0)]); - let reliability = stats.into_reliability(0, 0, 0); + let reliability = stats.into_reliability(0, 0, 0, Vec::new()); assert_eq!(reliability.kernel_reserve_failures, 5); assert_eq!(reliability.kernel_read_failures, 1); @@ -824,7 +824,12 @@ mod tests { let stats = stats_by_cpu(&[(0, 0, 900), (0, 0, 12)]); assert!(stats.loss_by_cpu().is_empty()); - assert_eq!(stats.into_reliability(0, 0, 0).kernel_filtered_events, 912); + assert_eq!( + stats + .into_reliability(0, 0, 0, Vec::new()) + .kernel_filtered_events, + 912 + ); } #[test] @@ -842,7 +847,7 @@ mod tests { }, ]); - let reliability = stats.into_reliability(0, 0, 0); + let reliability = stats.into_reliability(0, 0, 0, Vec::new()); assert_eq!(reliability.kernel_reserve_failures, 9); assert_eq!(reliability.kernel_unattributed_reserve_failures, 6); @@ -873,9 +878,24 @@ mod tests { assert!(ProbeSiteKey::from_bytes(&[0_u8; 12]).is_none()); } + #[test] + fn the_probe_breakdown_is_carried_through_rather_than_dropped() { + // The parameter exists so this cannot be forgotten; assert it lands. + let probes = vec![KernelProbeLoss { + function: None, + program_id: Some(7), + reserve_failures: 3, + }]; + + let reliability = stats_by_cpu(&[(3, 0, 0)]).into_reliability(0, 0, 0, probes.clone()); + + assert_eq!(reliability.kernel_loss_by_probe, probes); + } + #[test] fn a_lossless_capture_reports_an_empty_breakdown_not_a_missing_one() { - let reliability = stats_by_cpu(&[(0, 0, 0), (0, 0, 0)]).into_reliability(0, 0, 0); + let reliability = + stats_by_cpu(&[(0, 0, 0), (0, 0, 0)]).into_reliability(0, 0, 0, Vec::new()); assert!(reliability.complete()); assert!(reliability.kernel_loss_by_cpu.is_empty());