phantom_protocol/observability/
snapshot.rs

1//! Cold-path snapshot of the observability state.
2//!
3//! Reads every hot-path atomic with `Ordering::Relaxed` and exposes the
4//! values in a `Clone`-able plain struct suitable for FFI, logging, and
5//! debugging. Per-leg breakdown is preserved so consumers can compute their
6//! own slices.
7//!
8//! In step 3 of the rollout this struct does NOT carry the security signal
9//! counters or histogram (those are added with the OTel instruments in
10//! later steps). It provides at least the same totals as the legacy
11//! `transport::metrics::MetricsSnapshot` so call sites can migrate.
12
13use crate::observability::atomics::{HotPathAtomics, DIR_RECV, DIR_SEND};
14use crate::transport::types::LegType;
15
16/// Immutable cold-path snapshot of the hot-path atomics.
17#[derive(Debug, Clone)]
18pub struct MetricsSnapshot {
19    pub packets_sent: u64,
20    pub packets_recv: u64,
21    pub bytes_sent: u64,
22    pub bytes_recv: u64,
23
24    /// Per-leg packet counts: `(LegType, packets_sent, packets_recv)`.
25    pub per_leg_packets: [(LegType, u64, u64); 4],
26    /// Per-leg byte counts: `(LegType, bytes_sent, bytes_recv)`.
27    pub per_leg_bytes: [(LegType, u64, u64); 4],
28
29    pub avg_encrypt_ns: u64,
30    pub avg_decrypt_ns: u64,
31    pub encrypt_count: u64,
32    pub decrypt_count: u64,
33
34    pub rtt_us_path_0: u64,
35
36    pub active_sessions: i64,
37    pub active_streams: i64,
38
39    pub handshakes_success: u64,
40    pub handshakes_failure: u64,
41    pub handshake_latency_ns_sum: u64,
42    pub handshake_latency_count: u64,
43
44    pub uptime_secs: u64,
45}
46
47impl Default for MetricsSnapshot {
48    fn default() -> Self {
49        Self {
50            packets_sent: 0,
51            packets_recv: 0,
52            bytes_sent: 0,
53            bytes_recv: 0,
54            per_leg_packets: [
55                (LegType::Kcp, 0, 0),
56                (LegType::Tcp, 0, 0),
57                (LegType::FakeTls, 0, 0),
58                (LegType::Udp, 0, 0),
59            ],
60            per_leg_bytes: [
61                (LegType::Kcp, 0, 0),
62                (LegType::Tcp, 0, 0),
63                (LegType::FakeTls, 0, 0),
64                (LegType::Udp, 0, 0),
65            ],
66            avg_encrypt_ns: 0,
67            avg_decrypt_ns: 0,
68            encrypt_count: 0,
69            decrypt_count: 0,
70            rtt_us_path_0: 0,
71            active_sessions: 0,
72            active_streams: 0,
73            handshakes_success: 0,
74            handshakes_failure: 0,
75            handshake_latency_ns_sum: 0,
76            handshake_latency_count: 0,
77            uptime_secs: 0,
78        }
79    }
80}
81
82impl MetricsSnapshot {
83    pub(crate) fn capture(h: &HotPathAtomics) -> Self {
84        let avg_encrypt_ns = avg(h.encrypt_sum_ns(), h.encrypt_count());
85        let avg_decrypt_ns = avg(h.decrypt_sum_ns(), h.decrypt_count());
86
87        let per_leg_packets = [
88            (
89                LegType::Kcp,
90                h.packets_per_leg(DIR_SEND, LegType::Kcp),
91                h.packets_per_leg(DIR_RECV, LegType::Kcp),
92            ),
93            (
94                LegType::Tcp,
95                h.packets_per_leg(DIR_SEND, LegType::Tcp),
96                h.packets_per_leg(DIR_RECV, LegType::Tcp),
97            ),
98            (
99                LegType::FakeTls,
100                h.packets_per_leg(DIR_SEND, LegType::FakeTls),
101                h.packets_per_leg(DIR_RECV, LegType::FakeTls),
102            ),
103            (
104                LegType::Udp,
105                h.packets_per_leg(DIR_SEND, LegType::Udp),
106                h.packets_per_leg(DIR_RECV, LegType::Udp),
107            ),
108        ];
109        let per_leg_bytes = [
110            (
111                LegType::Kcp,
112                h.bytes_per_leg(DIR_SEND, LegType::Kcp),
113                h.bytes_per_leg(DIR_RECV, LegType::Kcp),
114            ),
115            (
116                LegType::Tcp,
117                h.bytes_per_leg(DIR_SEND, LegType::Tcp),
118                h.bytes_per_leg(DIR_RECV, LegType::Tcp),
119            ),
120            (
121                LegType::FakeTls,
122                h.bytes_per_leg(DIR_SEND, LegType::FakeTls),
123                h.bytes_per_leg(DIR_RECV, LegType::FakeTls),
124            ),
125            (
126                LegType::Udp,
127                h.bytes_per_leg(DIR_SEND, LegType::Udp),
128                h.bytes_per_leg(DIR_RECV, LegType::Udp),
129            ),
130        ];
131
132        Self {
133            packets_sent: h.packets_total(DIR_SEND),
134            packets_recv: h.packets_total(DIR_RECV),
135            bytes_sent: h.bytes_total(DIR_SEND),
136            bytes_recv: h.bytes_total(DIR_RECV),
137            per_leg_packets,
138            per_leg_bytes,
139            avg_encrypt_ns,
140            avg_decrypt_ns,
141            encrypt_count: h.encrypt_count(),
142            decrypt_count: h.decrypt_count(),
143            rtt_us_path_0: h.rtt_us(0),
144            active_sessions: h.active_sessions(),
145            active_streams: h.active_streams(),
146            handshakes_success: h.handshake_success_count(),
147            handshakes_failure: h.handshake_failure_count(),
148            handshake_latency_ns_sum: h.handshake_latency_ns_sum(),
149            handshake_latency_count: h.handshake_latency_count(),
150            uptime_secs: h.uptime_secs(),
151        }
152    }
153}
154
155fn avg(sum: u64, count: u64) -> u64 {
156    sum.checked_div(count).unwrap_or(0)
157}
158
159impl std::fmt::Display for MetricsSnapshot {
160    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
161        write!(
162            f,
163            "tx={} rx={} bytes_tx={} bytes_rx={} encrypt={}ns decrypt={}ns sessions={} streams={} up={}s",
164            self.packets_sent,
165            self.packets_recv,
166            self.bytes_sent,
167            self.bytes_recv,
168            self.avg_encrypt_ns,
169            self.avg_decrypt_ns,
170            self.active_sessions,
171            self.active_streams,
172            self.uptime_secs,
173        )
174    }
175}
176
177#[cfg(test)]
178mod tests {
179    use super::*;
180    use crate::observability::atomics::HotPathAtomics;
181
182    #[test]
183    fn snapshot_zero_state() {
184        let h = HotPathAtomics::new();
185        let s = MetricsSnapshot::capture(&h);
186        assert_eq!(s.packets_sent, 0);
187        assert_eq!(s.avg_encrypt_ns, 0);
188        assert_eq!(s.active_sessions, 0);
189    }
190
191    #[test]
192    fn snapshot_after_recording() {
193        let h = HotPathAtomics::new();
194        h.record_send(1024, LegType::Tcp);
195        h.record_send(2048, LegType::Kcp);
196        h.record_recv(512, LegType::Tcp);
197        h.record_encrypt_ns(100);
198        h.record_encrypt_ns(200);
199        h.session_opened();
200        h.stream_opened();
201
202        let s = MetricsSnapshot::capture(&h);
203        assert_eq!(s.packets_sent, 2);
204        assert_eq!(s.packets_recv, 1);
205        assert_eq!(s.bytes_sent, 3072);
206        assert_eq!(s.avg_encrypt_ns, 150);
207        assert_eq!(s.encrypt_count, 2);
208        assert_eq!(s.active_sessions, 1);
209        assert_eq!(s.active_streams, 1);
210    }
211
212    #[test]
213    fn display_is_one_line() {
214        let h = HotPathAtomics::new();
215        let s = MetricsSnapshot::capture(&h);
216        let text = format!("{}", s);
217        assert!(!text.contains('\n'));
218        assert!(text.contains("tx=0"));
219    }
220}
phantom_protocol/observability/snapshot.rs

phantom_protocol/observability/
snapshot.rs
