subms_otel/
bridge.rs

1//! Always-available one-shot bridge from a typed [`SubMsBenchSummary`] /
2//! [`SubMsTimer`] to OpenTelemetry instruments. No background threads, no
3//! per-sample channel - just walks the summary and emits.
4
5use std::collections::BTreeMap;
6use std::time::SystemTime;
7
8use opentelemetry::KeyValue;
9use opentelemetry::metrics::Meter;
10use opentelemetry::trace::{Span, SpanBuilder, Tracer};
11
12use subms::{ObservationCtx, SubMsBenchSummary, SubMsStageKind, SubMsStageSummary, SubMsTimer};
13
14/// Canonical name for the per-stage latency histogram. Consumers filter on
15/// this in Grafana / dashboards.
16pub const HISTOGRAM_NAME: &str = "subms.latency";
17
18/// Canonical unit for the latency histogram. Seconds is OTEL convention for
19/// duration measurements - we emit ns / 1e9 to satisfy it.
20pub const HISTOGRAM_UNIT: &str = "s";
21
22/// Histogram bucket boundaries (in seconds) keyed off the stage kind. Adapter
23/// code applies these when first constructing the `Histogram` instrument for
24/// a given stage kind.
25///
26/// - `HotPath` covers 50ns .. 1ms, the sub-ms p99 budget recipes target.
27/// - `BatchOp` / `OneShot` cover 1us .. 1s, where whole-structure ops live.
28/// - `Unspecified` returns an empty slice so OTEL's default boundaries apply.
29pub fn histogram_boundaries(kind: SubMsStageKind) -> &'static [f64] {
30    const HOT: &[f64] = &[
31        5e-8, 1e-7, 2e-7, 5e-7, 1e-6, 2e-6, 5e-6, 1e-5, 5e-5, 1e-4, 5e-4, 1e-3,
32    ];
33    const BATCH: &[f64] = &[1e-6, 1e-5, 1e-4, 1e-3, 1e-2, 1e-1, 1.0];
34    match kind {
35        SubMsStageKind::HotPath => HOT,
36        SubMsStageKind::BatchOp | SubMsStageKind::OneShot => BATCH,
37        SubMsStageKind::Unspecified => &[],
38        _ => &[],
39    }
40}
41
42/// Build the OTEL attribute set for a hot-path record. Just the harness
43/// identity the [`ObservationCtx`] carries - inputs and meta arrive later via
44/// the summary.
45pub fn attributes_from_ctx(ctx: &ObservationCtx<'_>) -> Vec<KeyValue> {
46    vec![
47        KeyValue::new("subms.workload", ctx.workload.to_string()),
48        KeyValue::new("subms.lang", ctx.lang.to_string()),
49        KeyValue::new("subms.stage", ctx.stage.to_string()),
50        KeyValue::new("subms.stage.kind", ctx.stage_kind.as_str()),
51    ]
52}
53
54/// Build the full attribute set for a stage when we have the post-bench
55/// summary. Pulls workload + lang + stage from the summary and folds in any
56/// of the canonical inputs/meta keys that are present.
57pub fn attributes_from_summary(
58    summary: &SubMsBenchSummary,
59    stage: &SubMsStageSummary,
60    kind: SubMsStageKind,
61) -> Vec<KeyValue> {
62    let mut attrs = vec![
63        KeyValue::new("subms.workload", summary.workload.clone()),
64        KeyValue::new("subms.lang", summary.lang.clone()),
65        KeyValue::new("subms.stage", stage.name.clone()),
66        KeyValue::new("subms.stage.kind", kind.as_str()),
67    ];
68    push_meta_attrs(&mut attrs, &summary.inputs, &summary.meta);
69    attrs
70}
71
72/// Fold inputs + meta entries into an attribute set. Public so the async
73/// observer can use the same logic when draining cached summary state.
74pub(crate) fn push_meta_attrs(
75    attrs: &mut Vec<KeyValue>,
76    inputs: &BTreeMap<String, String>,
77    meta: &BTreeMap<String, String>,
78) {
79    fn put(
80        attrs: &mut Vec<KeyValue>,
81        map: &BTreeMap<String, String>,
82        src: &str,
83        dst: &'static str,
84    ) {
85        if let Some(v) = map.get(src) {
86            attrs.push(KeyValue::new(dst, v.clone()));
87        }
88    }
89    put(attrs, meta, "subms.recipe.slug", "subms.recipe.slug");
90    put(
91        attrs,
92        meta,
93        "subms.recipe.category",
94        "subms.recipe.category",
95    );
96    put(
97        attrs,
98        meta,
99        "subms.workload.feature",
100        "subms.workload.feature",
101    );
102    put(attrs, inputs, "entries", "subms.workload.entries");
103    put(attrs, inputs, "seed", "subms.workload.seed");
104    put(attrs, meta, "host", "subms.host");
105    put(attrs, meta, "hardware_tier", "subms.hardware.tier");
106    put(attrs, meta, "crate_version", "subms.crate.version");
107}
108
109/// Export every stage of a [`SubMsBenchSummary`] as a Histogram emission on
110/// the given [`Meter`]. Each percentile (p50, p99, p999, max, mean, stddev)
111/// and the downsampled chronological samples are recorded as histogram
112/// values under the stage's full attribute set.
113///
114/// The histogram name is [`HISTOGRAM_NAME`] (one instrument per
115/// (workload, stage) - the attribute set carries the stage identity).
116///
117/// Stage kind is not carried on the summary; this function defaults to
118/// [`SubMsStageKind::Unspecified`] for boundary selection. Callers that know
119/// the per-stage kind ahead of time should use [`OtelObserver`] / the
120/// kind-tracking observer surface instead.
121pub fn export_summary(summary: &SubMsBenchSummary, meter: &Meter) {
122    for stage in &summary.stages {
123        let kind = SubMsStageKind::Unspecified;
124        let attrs = attributes_from_summary(summary, stage, kind);
125        let histogram = histogram_for_stage(meter, kind);
126        // Emit the headline percentiles as discrete records. This is the
127        // "summary mode" - downstream systems can read p50/p99/etc. straight
128        // off the histogram buckets that contain these values.
129        record_seconds(&histogram, stage.p50_ns, &attrs);
130        record_seconds(&histogram, stage.p99_ns, &attrs);
131        record_seconds(&histogram, stage.p999_ns, &attrs);
132        record_seconds(&histogram, stage.max_ns, &attrs);
133        record_seconds(&histogram, stage.mean_ns, &attrs);
134        // Replay the downsampled timeline so backends with native histogram
135        // aggregation see the actual distribution shape.
136        if let Some(samples) = &stage.samples_ns {
137            for ns in samples {
138                record_seconds(&histogram, *ns, &attrs);
139            }
140        }
141    }
142}
143
144fn record_seconds(histogram: &opentelemetry::metrics::Histogram<f64>, ns: u64, attrs: &[KeyValue]) {
145    histogram.record(ns as f64 / 1e9, attrs);
146}
147
148/// Build a `Histogram<f64>` whose bucket boundaries match the given stage
149/// kind. Public-ish helper used by both [`export_summary`] and the observer
150/// modules.
151pub(crate) fn histogram_for_stage(
152    meter: &Meter,
153    kind: SubMsStageKind,
154) -> opentelemetry::metrics::Histogram<f64> {
155    let bounds = histogram_boundaries(kind);
156    let mut builder = meter
157        .f64_histogram(HISTOGRAM_NAME)
158        .with_unit(HISTOGRAM_UNIT)
159        .with_description("subms per-stage latency histogram");
160    if !bounds.is_empty() {
161        builder = builder.with_boundaries(bounds.to_vec());
162    }
163    builder.build()
164}
165
166/// Export a [`SubMsTimer`] timeline as a parent span + one child span per
167/// checkpoint. Span start times are reconstructed as `now - elapsed_ns` so
168/// the timing offsets the timer captured are preserved end-to-end.
169///
170/// The parent span's name is the timer's name (or `"subms.timer"` for an
171/// unnamed timer). Each checkpoint becomes a child span whose duration is
172/// the checkpoint's `since_last_ns`.
173pub fn export_timer<T: Tracer>(timer: &SubMsTimer, tracer: &T)
174where
175    T::Span: Span + Send + Sync + 'static,
176{
177    let total_ns = timer.elapsed_ns();
178    let now = SystemTime::now();
179    let parent_start = now
180        .checked_sub(std::time::Duration::from_nanos(total_ns))
181        .unwrap_or(now);
182
183    let parent_name: std::borrow::Cow<'static, str> = if timer.name().is_empty() {
184        "subms.timer".into()
185    } else {
186        timer.name().to_string().into()
187    };
188    let mut parent = SpanBuilder::from_name(parent_name)
189        .with_start_time(parent_start)
190        .with_attributes(vec![
191            KeyValue::new("subms.timer.total_ns", total_ns as i64),
192            KeyValue::new("subms.timer.checkpoints", timer.checkpoints().len() as i64),
193        ])
194        .start(tracer);
195
196    for cp in timer.checkpoints() {
197        let child_start = parent_start
198            .checked_add(std::time::Duration::from_nanos(
199                cp.since_start_ns.saturating_sub(cp.since_last_ns),
200            ))
201            .unwrap_or(parent_start);
202        let child_end = parent_start
203            .checked_add(std::time::Duration::from_nanos(cp.since_start_ns))
204            .unwrap_or(parent_start);
205
206        let mut child = SpanBuilder::from_name(cp.label.clone())
207            .with_start_time(child_start)
208            .with_attributes(vec![
209                KeyValue::new("subms.timer.since_last_ns", cp.since_last_ns as i64),
210                KeyValue::new("subms.timer.since_start_ns", cp.since_start_ns as i64),
211                KeyValue::new("subms.timer.is_stop", cp.is_stop),
212            ])
213            .start(tracer);
214        child.end_with_timestamp(child_end);
215    }
216
217    parent.end_with_timestamp(now);
218}
subms_otel/bridge.rs

subms_otel/
bridge.rs
