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

phantom_protocol/observability/
snapshot.rs
