zenkey_fleet/
stats.rs

1//! Windowed per-key statistics (issues #13/#15): message/byte counters and
2//! an exponentially-weighted rate, keyed by wire key. Backs `zenctl topic
3//! hz`/`bw`/`echo --rate` and zengui's tree badges.
4//!
5//! Perf posture (report §14): lookups borrow (`&str` against the `String`
6//! keys — no per-sample allocation on the hot hit path); one allocation per
7//! *new* key is the floor.
8
9use std::collections::HashMap;
10use std::time::{Duration, Instant};
11
12/// One key's running statistics.
13#[derive(Debug, Clone)]
14pub struct KeyStats {
15    pub count: u64,
16    pub bytes: u64,
17    /// EWMA of the instantaneous rate (Hz), time-decayed.
18    pub rate_hz: f64,
19    pub last_seen: Instant,
20    /// Consecutive source-sequence-number gap count, when publishers attach
21    /// SourceInfo (unstable API) — loss visibility, `--loss`.
22    pub sn_gaps: u64,
23    last_sn: Option<u32>,
24}
25
26/// The table. Feed it samples; read it per key or in aggregate.
27#[derive(Debug, Default)]
28pub struct StatsTable {
29    keys: HashMap<String, KeyStats>,
30}
31
32/// EWMA time constant (~2 s: responsive enough for a UI badge, smooth
33/// enough not to flicker); samples older than ~tau contribute e^-1.
34const TAU: Duration = Duration::from_secs(2);
35
36impl StatsTable {
37    pub fn new() -> Self {
38        Self::default()
39    }
40
41    /// Record one sample. `now` is injected for deterministic tests.
42    pub fn record(&mut self, key: &str, payload_len: usize, sn: Option<u32>, now: Instant) {
43        if let Some(s) = self.keys.get_mut(key) {
44            let dt = now.saturating_duration_since(s.last_seen).as_secs_f64();
45            if dt > 0.0 {
46                let alpha = 1.0 - (-dt / TAU.as_secs_f64()).exp();
47                let instant_rate = 1.0 / dt;
48                s.rate_hz += alpha * (instant_rate - s.rate_hz);
49            }
50            s.count += 1;
51            s.bytes += payload_len as u64;
52            s.last_seen = now;
53            if let (Some(prev), Some(cur)) = (s.last_sn, sn)
54                && cur > prev + 1
55            {
56                s.sn_gaps += u64::from(cur - prev - 1);
57            }
58            s.last_sn = sn;
59        } else {
60            self.keys.insert(
61                key.to_string(),
62                KeyStats {
63                    count: 1,
64                    bytes: payload_len as u64,
65                    rate_hz: 0.0,
66                    last_seen: now,
67                    sn_gaps: 0,
68                    last_sn: sn,
69                },
70            );
71        }
72    }
73
74    pub fn get(&self, key: &str) -> Option<&KeyStats> {
75        self.keys.get(key)
76    }
77
78    pub fn iter(&self) -> impl Iterator<Item = (&str, &KeyStats)> {
79        self.keys.iter().map(|(k, v)| (k.as_str(), v))
80    }
81
82    pub fn len(&self) -> usize {
83        self.keys.len()
84    }
85
86    pub fn is_empty(&self) -> bool {
87        self.keys.is_empty()
88    }
89
90    /// Aggregate totals: (samples, bytes, summed EWMA rate).
91    pub fn totals(&self) -> (u64, u64, f64) {
92        self.keys.values().fold((0, 0, 0.0), |(c, b, r), s| {
93            (c + s.count, b + s.bytes, r + s.rate_hz)
94        })
95    }
96}
97
98#[cfg(test)]
99mod tests {
100    use super::*;
101
102    #[test]
103    fn rates_converge_and_gaps_count() {
104        let mut t = StatsTable::new();
105        let t0 = Instant::now();
106        // 10 Hz for 100 samples: the EWMA converges near 10.
107        for i in 0..100u32 {
108            t.record(
109                "v1/h-a/telemetry/x/m",
110                8,
111                Some(i),
112                t0 + Duration::from_millis(100 * u64::from(i)),
113            );
114        }
115        let s = t.get("v1/h-a/telemetry/x/m").unwrap();
116        assert_eq!(s.count, 100);
117        assert_eq!(s.bytes, 800);
118        assert!((s.rate_hz - 10.0).abs() < 1.0, "rate {}", s.rate_hz);
119        assert_eq!(s.sn_gaps, 0);
120
121        // A sequence jump records the gap.
122        t.record(
123            "v1/h-a/telemetry/x/m",
124            8,
125            Some(105),
126            t0 + Duration::from_millis(10_100),
127        );
128        assert_eq!(t.get("v1/h-a/telemetry/x/m").unwrap().sn_gaps, 5);
129    }
130
131    #[test]
132    fn totals_aggregate() {
133        let mut t = StatsTable::new();
134        let now = Instant::now();
135        t.record("a", 10, None, now);
136        t.record("b", 20, None, now);
137        let (count, bytes, _) = t.totals();
138        assert_eq!((count, bytes), (2, 30));
139        assert_eq!(t.len(), 2);
140    }
141}
zenkey_fleet/stats.rs

zenkey_fleet/
stats.rs
