metrale_telemetry/
hub.rs

1// SPDX-License-Identifier: MIT OR Apache-2.0
2
3//! 2026-09-26: [`Telemetry`]: one instrument set behind one level gate.
4//!
5//! Every hot-path entry point first loads one relaxed atomic (the level, or
6//! for `kernel_spans_armed` the armed flag) and at `Off` returns before any
7//! clock read, allocation or atomic write. `tests/zero_cost_off.rs` checks it
8//! with a counting allocator and a counting clock.
9//!
10//! The serving process uses [`global`], a `static`: the HTTP handlers for
11//! `/metrics` and `/v1/events`, the GPU backend and the device sampler thread
12//! all reach it without a handle from the scheduler. Tests build their own
13//! instance.
14//!
15//! Owner: telemetry.
16//! Invariants: at `Off`, no hot-path entry point reads the clock, allocates
17//! or writes an atomic.
18
19use std::sync::OnceLock;
20use std::sync::atomic::{AtomicU8, AtomicU32, Ordering};
21use std::time::Instant;
22
23use crate::clock::{Clock, MonotonicClock};
24use crate::device::{DeviceReading, SAMPLE_WORDS};
25use crate::energy::{
26    Advance, CounterTracker, EnergyPoint, EnergyRing, RING_SLOTS, joules_per_token,
27    request_millijoules,
28};
29use crate::instrument::{Counter, Gauge, GaugeF64};
30use crate::kernel::{KernelInstruments, LANES};
31use crate::level::{Level, TelemetryConfig};
32use crate::request::{RequestInstruments, RequestTiming};
33use crate::sched::{MAX_PHASES, SchedInstruments, SchedShape, SpecMatrix};
34use crate::seqcell::SeqCell;
35
36/// 2026-09-26: Device samples the live J/token spans (1 s at the serving cadence).
37pub const LIVE_WINDOW: u64 = 10;
38
39/// 2026-09-26: Whether the device layer is reading.
40#[derive(Clone, Copy, Debug, PartialEq, Eq)]
41#[repr(u8)]
42pub enum DeviceState {
43    /// 2026-09-26: No sampler has started (telemetry off, or not yet configured).
44    NotStarted = 0,
45    Available = 1,
46    /// 2026-09-26: The device layer cannot read: NVML is missing, has no such
47    /// device, or the sampler thread did not start. See
48    /// [`Telemetry::device_unavailable_reason`].
49    Unavailable = 2,
50}
51
52pub struct Telemetry {
53    level: AtomicU8,
54    clock: &'static dyn Clock,
55    kernel_every: AtomicU32,
56    phase_names: OnceLock<&'static [&'static str]>,
57    pub(crate) device: SeqCell<SAMPLE_WORDS>,
58    device_state: AtomicU8,
59    device_reason: OnceLock<String>,
60    pub device_read_errors: Gauge,
61    pub energy_mj: Counter,
62    pub energy_counter_raw: Gauge,
63    pub energy_resets: Counter,
64    pub energy_wraps: Counter,
65    pub tokens: Counter,
66    /// 2026-09-26: `tokens` at the first energy sample.
67    energy_token_base: Gauge,
68    pub joules_per_token_live: GaugeF64,
69    ring: OnceLock<EnergyRing>,
70    pub kernel: KernelInstruments,
71    pub sched: SchedInstruments,
72    pub spec: SpecMatrix,
73    pub spec_confidence: crate::spec_confidence::ConfidenceOutcomes,
74    pub requests: RequestInstruments,
75}
76
77impl std::fmt::Debug for Telemetry {
78    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
79        f.debug_struct("Telemetry")
80            .field("level", &self.level())
81            .finish_non_exhaustive()
82    }
83}
84
85static GLOBAL: Telemetry = Telemetry::new(&MonotonicClock);
86
87/// 2026-09-26: The serving process's instrument set.
88pub fn global() -> &'static Telemetry {
89    &GLOBAL
90}
91
92impl Telemetry {
93    /// 2026-09-26: A set at level `Off` over `clock`. Allocates nothing.
94    pub const fn new(clock: &'static dyn Clock) -> Self {
95        Self {
96            level: AtomicU8::new(Level::Off as u8),
97            clock,
98            kernel_every: AtomicU32::new(u32::MAX),
99            phase_names: OnceLock::new(),
100            device: SeqCell::new(),
101            device_state: AtomicU8::new(DeviceState::NotStarted as u8),
102            device_reason: OnceLock::new(),
103            device_read_errors: Gauge::new(),
104            energy_mj: Counter::new(),
105            energy_counter_raw: Gauge::new(),
106            energy_resets: Counter::new(),
107            energy_wraps: Counter::new(),
108            tokens: Counter::new(),
109            energy_token_base: Gauge::new(),
110            joules_per_token_live: GaugeF64::new(),
111            ring: OnceLock::new(),
112            kernel: KernelInstruments::new(),
113            sched: SchedInstruments::new(),
114            spec: SpecMatrix::new(),
115            spec_confidence: crate::spec_confidence::ConfidenceOutcomes::new(),
116            requests: RequestInstruments::new(),
117        }
118    }
119
120    /// 2026-09-26: Apply a validated configuration and disarm kernel spans.
121    /// The energy ring is allocated here, on the first call at `Basic` or
122    /// above, and never again.
123    pub fn configure(&self, cfg: &TelemetryConfig) {
124        if cfg.level() >= Level::Basic {
125            self.ring
126                .get_or_init(|| EnergyRing::with_capacity(RING_SLOTS));
127        }
128        self.kernel_every
129            .store(cfg.kernel_span_every(), Ordering::Relaxed);
130        self.kernel.armed.store(false, Ordering::Relaxed);
131        self.level.store(cfg.level() as u8, Ordering::Release);
132    }
133
134    #[inline]
135    pub fn level(&self) -> Level {
136        Level::from_u8(self.level.load(Ordering::Relaxed))
137    }
138
139    #[inline]
140    fn on(&self) -> bool {
141        self.level.load(Ordering::Relaxed) != Level::Off as u8
142    }
143
144    /// 2026-09-26: Name the loop phases `phase` indexes, keeping at most
145    /// [`MAX_PHASES`]. The first call wins.
146    pub fn set_phase_names(&self, names: &'static [&'static str]) {
147        let _ = self.phase_names.set(&names[..names.len().min(MAX_PHASES)]);
148    }
149
150    pub fn phase_names(&self) -> &'static [&'static str] {
151        self.phase_names.get().copied().unwrap_or(&[])
152    }
153
154    /// 2026-09-26: `n` more tokens were produced.
155    #[inline]
156    pub fn tokens(&self, n: u64) {
157        if !self.on() {
158            return;
159        }
160        self.tokens.add(n);
161    }
162
163    /// 2026-09-26: Loop phase `idx` ran from `since` until now. An `idx` of
164    /// [`MAX_PHASES`] or more lands in the last histogram.
165    #[inline]
166    pub fn phase(&self, idx: usize, since: Instant) {
167        if !self.on() {
168            return;
169        }
170        let ns = self.clock.elapsed_ns(since);
171        self.sched.phase[idx.min(MAX_PHASES - 1)].observe(ns);
172    }
173
174    /// 2026-09-26: This tick's scheduler shape.
175    #[inline]
176    pub fn sched_tick(&self, shape: &SchedShape) {
177        if !self.on() {
178            return;
179        }
180        self.sched.record(shape);
181    }
182
183    #[inline]
184    pub fn stream_sync(&self) {
185        if self.on() {
186            self.sched.stream_syncs.add(1);
187        }
188    }
189
190    #[inline]
191    pub fn blocking_d2h(&self) {
192        if self.on() {
193            self.sched.blocking_d2h.add(1);
194        }
195    }
196
197    #[inline]
198    pub fn graph_replay(&self) {
199        if self.on() {
200            self.sched.graph_replays.add(1);
201        }
202    }
203
204    #[inline]
205    pub fn graph_capture(&self) {
206        if self.on() {
207            self.sched.graph_captures.add(1);
208        }
209    }
210
211    /// 2026-09-26: A verify step checked `drafts` drafts and accepted `accepted`.
212    #[inline]
213    pub fn spec_verified(&self, drafts: usize, accepted: usize) {
214        if !self.on() {
215            return;
216        }
217        self.spec.record(drafts, accepted);
218    }
219
220    /// 2026-10-04: A verify step reached a draft whose drafter top-1 log-probability was
221    /// `lp`, and accepted it or not.
222    #[inline]
223    pub fn spec_draft_confidence(&self, lp: f32, accepted: bool) {
224        if !self.on() {
225            return;
226        }
227        self.spec_confidence.record(lp, accepted);
228    }
229
230    /// 2026-09-26: A request finished. Its energy is attributed from the ring
231    /// window its lifetime spans, only while the device layer is available.
232    pub fn request_finished(&self, r: &RequestTiming) {
233        if !self.on() {
234            return;
235        }
236        let end = self.clock.now_ns();
237        let start = end.saturating_sub(r.e2e_ns());
238        let energy = self
239            .ring
240            .get()
241            .filter(|_| self.device_state() == DeviceState::Available)
242            .and_then(|ring| request_millijoules(ring, start, end, r.output_tokens));
243        self.requests.record(r, energy);
244    }
245
246    /// 2026-09-26: A scheduler step begins. At level `Kernel`, arms per-kernel
247    /// spans on one step in `kernel_span_every` and returns whether this step
248    /// is armed; below `Kernel`, returns `false` and changes nothing.
249    #[inline]
250    pub fn step_begin(&self) -> bool {
251        if self.level() != Level::Kernel {
252            return false;
253        }
254        let n = self.kernel.step.fetch_add(1, Ordering::Relaxed);
255        let every = u64::from(self.kernel_every.load(Ordering::Relaxed).max(1));
256        let arm = n.is_multiple_of(every);
257        self.kernel.armed.store(arm, Ordering::Relaxed);
258        arm
259    }
260
261    /// 2026-09-26: Whether the kernel launch path should bracket launches this
262    /// step. Set by `step_begin`, cleared by `configure`.
263    #[inline]
264    pub fn kernel_spans_armed(&self) -> bool {
265        self.kernel.armed.load(Ordering::Relaxed)
266    }
267
268    /// 2026-09-26: Whether lane spans are recorded (level `Kernel`).
269    #[inline]
270    pub fn lane_spans(&self) -> bool {
271        self.level() == Level::Kernel
272    }
273
274    pub fn lane_names(&self) -> &'static [&'static str] {
275        &LANES
276    }
277
278    pub fn device_state(&self) -> DeviceState {
279        match self.device_state.load(Ordering::Acquire) {
280            1 => DeviceState::Available,
281            2 => DeviceState::Unavailable,
282            _ => DeviceState::NotStarted,
283        }
284    }
285
286    pub fn device_unavailable_reason(&self) -> Option<&str> {
287        self.device_reason.get().map(String::as_str)
288    }
289
290    /// 2026-09-26: The device layer cannot read. The first `why` is kept for
291    /// [`Telemetry::device_unavailable_reason`].
292    pub fn mark_device_unavailable(&self, why: String) {
293        let _ = self.device_reason.set(why);
294        self.device_state
295            .store(DeviceState::Unavailable as u8, Ordering::Release);
296    }
297
298    /// 2026-09-26: Absorb one device reading: publish it and, when it carries
299    /// the energy counter, advance the energy total; once the ring exists, also
300    /// push a ring point pairing it with the tokens so far and update the live
301    /// J/token.
302    pub fn absorb_device(
303        &self,
304        tracker: &mut CounterTracker,
305        reading: &DeviceReading,
306        read_errors: u64,
307    ) {
308        let at_ns = self.clock.now_ns();
309        self.device.store(reading.encode(at_ns));
310        self.device_read_errors.set(read_errors);
311        self.device_state
312            .store(DeviceState::Available as u8, Ordering::Release);
313        let Some(raw) = reading.energy_counter_mj else {
314            return;
315        };
316        self.energy_counter_raw.set(raw);
317        let step = tracker.advance(raw);
318        match step {
319            Advance::Wrapped(_) => self.energy_wraps.add(1),
320            Advance::Reset(_) => self.energy_resets.add(1),
321            Advance::First => self.energy_token_base.set(self.tokens.get()),
322            Advance::Delta(_) => {}
323        }
324        self.energy_mj.add(step.millijoules());
325        let Some(ring) = self.ring.get() else { return };
326        let now = EnergyPoint {
327            at_ns,
328            energy_mj: self.energy_mj.get(),
329            tokens: self.tokens.get(),
330        };
331        ring.push(now);
332        let live = (1..=LIVE_WINDOW)
333            .rev()
334            .find_map(|k| ring.back(k))
335            .and_then(|then| {
336                joules_per_token(now.energy_mj - then.energy_mj, now.tokens - then.tokens)
337            });
338        self.joules_per_token_live.set(live);
339    }
340
341    /// 2026-09-26: The latest device reading, its time, and how many readings
342    /// have been stored; `None` before the first.
343    pub fn device_reading(&self) -> Option<(DeviceReading, u64, u64)> {
344        let (stores, words) = self.device.load();
345        (stores > 0).then(|| {
346            let (r, at) = DeviceReading::decode(&words);
347            (r, at, stores)
348        })
349    }
350
351    /// 2026-09-26: GPU-rail J/token since the first energy sample.
352    pub fn joules_per_token_cumulative(&self) -> Option<f64> {
353        joules_per_token(self.energy_mj.get(), self.tokens_since_first_sample())
354    }
355
356    /// 2026-09-26: Tokens counted since the first energy sample: the
357    /// denominator for [`Telemetry::energy_mj`], which also starts there.
358    pub fn tokens_since_first_sample(&self) -> u64 {
359        self.tokens
360            .get()
361            .saturating_sub(self.energy_token_base.get())
362    }
363
364    pub fn now_ns(&self) -> u64 {
365        self.clock.now_ns()
366    }
367}
368
369#[cfg(test)]
370#[path = "hub_tests.rs"]
371mod tests;
metrale_telemetry/hub.rs

metrale_telemetry/
hub.rs
