metrale_telemetry/export/
events.rs

1// SPDX-License-Identifier: MIT OR Apache-2.0
2
3//! 2026-09-26: The `/v1/events` stream: one JSON object per SSE `data:` line
4//! (JSONL); the server sends one per device-sample period.
5//!
6//! Schema (version [`EVENT_SCHEMA_VERSION`]); every key is always present,
7//! and a quantity the device does not report is `null`, never 0:
8//!
9//! ```text
10//! { "v": 1, "seq": <u64, per-stream>, "kind": "snapshot",
11//!   "level": "off"|"basic"|"kernel", "at_ns": <telemetry clock>,
12//!   "device_state": "not_started"|"available"|"unavailable",
13//!   "device": null | { "power_mw", "energy_counter_mj", "graphics_clock_mhz",
14//!       "sm_clock_mhz", "mem_clock_mhz", "temperature_c",
15//!       "clocks_event_reasons", "mem_used_bytes", "mem_total_bytes",
16//!       "pcie_tx_kbps", "pcie_rx_kbps", "nvlink_active_links",
17//!       "samples", "age_ns", "read_errors" },
18//!   "energy": { "gpu_mj_total", "gpu_counter_mj", "tokens_total",
19//!       "joules_per_token_live", "joules_per_token_cumulative",
20//!       "counter_resets", "counter_wraps" },
21//!   "sched": { "pending", "active", "prefilling", "swapped",
22//!       "kv_blocks_free", "kv_blocks_total", "ssm_slots_used",
23//!       "ssm_slots_total", "ticks", "stream_syncs", "blocking_d2h",
24//!       "graph_replays", "graph_captures" },
25//!   "spec": [ { "drafts", "steps", "acceptance_by_depth": [f64] } ],
26//!   "requests": { "finished", "ttft_mean_ms", "tpot_mean_ms", "e2e_mean_ms",
27//!       "energy_mean_j", "energy_unattributed" } }
28//! ```
29//!
30//! Owner: telemetry.
31//! Invariants: an event line holds no newline.
32
33use serde::Serialize;
34
35use crate::snapshot::TelemetrySnapshot;
36
37pub const EVENT_SCHEMA_VERSION: u32 = 1;
38
39#[derive(Serialize)]
40struct Event<'a> {
41    v: u32,
42    seq: u64,
43    kind: &'static str,
44    #[serde(flatten)]
45    snap: &'a TelemetrySnapshot,
46}
47
48/// 2026-09-26: One event line, with no trailing newline (SSE framing adds it).
49pub fn snapshot_line(seq: u64, snap: &TelemetrySnapshot) -> String {
50    serde_json::to_string(&Event {
51        v: EVENT_SCHEMA_VERSION,
52        seq,
53        kind: "snapshot",
54        snap,
55    })
56    .expect("a snapshot is plain data and always serialises")
57}
58
59#[cfg(test)]
60mod tests {
61    use super::*;
62    use crate::clock::MonotonicClock;
63    use crate::hub::Telemetry;
64    use crate::level::{Level, TelemetryConfig};
65
66    #[test]
67    fn a_line_is_one_json_object_carrying_every_schema_key() {
68        static CLOCK: MonotonicClock = MonotonicClock;
69        let t = Telemetry::new(&CLOCK);
70        t.configure(&TelemetryConfig::serving(Level::Basic, 0));
71        t.tokens(3);
72        let line = snapshot_line(7, &t.snapshot());
73        assert!(!line.contains('\n'), "one line per event");
74        let v: serde_json::Value = serde_json::from_str(&line).unwrap();
75        assert_eq!(v["v"], 1);
76        assert_eq!(v["seq"], 7);
77        assert_eq!(v["kind"], "snapshot");
78        assert_eq!(v["level"], "basic");
79        assert_eq!(v["device_state"], "not_started");
80        assert!(v["device"].is_null(), "no sample yet is null, not zeros");
81        assert!(v["energy"]["joules_per_token_live"].is_null());
82        for k in ["pending", "kv_blocks_free", "graph_replays", "stream_syncs"] {
83            assert!(v["sched"].get(k).is_some(), "sched.{k}");
84        }
85        for k in ["finished", "tpot_mean_ms", "energy_unattributed"] {
86            assert!(v["requests"].get(k).is_some(), "requests.{k}");
87        }
88    }
89}
metrale_telemetry/export/events.rs

metrale_telemetry/export/
events.rs
