phantom_protocol/observability/
mod.rs

1//! Phantom Protocol observability subsystem.
2//!
3//! Replaced the Phase 4.5 hand-rolled metrics module (`transport::metrics`),
4//! which has since been deleted. Lock-free hot-path atomics for per-packet
5//! recording plus opt-in OpenTelemetry instruments (metrics + traces) gated
6//! behind the `telemetry-otel` Cargo feature.
7//!
8//! See `docs/observability/refactor-plan.md` for the full design and
9//! `docs/observability/metrics-catalog.md` for the instrument inventory.
10//! This file is the public module surface; concrete types live in the
11//! submodules below.
12//!
13//! ## Layout
14//!
15//! - `atomics` — lock-free hot-path counters (`HotPathAtomics`).
16//! - [`config`] — [`ObservabilityConfig`] (namespace, histogram buckets).
17//! - `instruments` — OTel instrument holder; ZST no-op when the feature
18//!   is off.
19//! - `bridge` — registers `ObservableCounter` callbacks over the atomics.
20//! - [`attrs`] — typed attribute-value enums (the cardinality contract).
21//! - [`snapshot`] — [`MetricsSnapshot`], a cold-path read for FFI / debug.
22
23pub(crate) mod atomics;
24pub mod attrs;
25pub(crate) mod bridge;
26pub mod config;
27pub(crate) mod instruments;
28pub mod snapshot;
29
30pub use attrs::{
31    leg_str, AeadAlgorithm, CookieOutcome, Direction, EarlyDataOutcome, FallbackReason,
32    HandshakeOutcome, PathValidationOutcome, PowOutcome, ProtocolVersion, ReplayReason,
33    ResumptionMode,
34};
35pub use config::{HistogramConfig, ObservabilityConfig, ObservabilityConfigBuilder};
36pub use snapshot::MetricsSnapshot;
37
38use crate::transport::types::LegType;
39use atomics::HotPathAtomics;
40use instruments::PhantomInstruments;
41use std::sync::Arc;
42
43/// Public observability facade.
44///
45/// Wraps the lock-free atomic counters (always present) and the opt-in
46/// OpenTelemetry instrument holder — metrics + traces, active when the
47/// `telemetry-otel` Cargo feature is enabled and a zero-cost ZST otherwise.
48/// Recording sites in `transport`, `api`, and `crypto` call methods on this
49/// struct via an `Arc<Observability>` borrowed from `PhantomListener` /
50/// `PhantomSession`.
51#[derive(Debug)]
52pub struct Observability {
53    config: ObservabilityConfig,
54    atomics: Arc<HotPathAtomics>,
55    instruments: PhantomInstruments,
56}
57
58impl Observability {
59    /// Construct a new observability handle.
60    ///
61    /// Returns an `Arc` because the handle is shared between recording sites
62    /// (in `Session`, `Listener`, handshake code paths) and the OTel
63    /// observable callbacks.
64    ///
65    /// **Call once per process.** When `telemetry-otel` is on, this
66    /// registers observable-instrument callbacks against the *global* OTel
67    /// meter; those callbacks are never unregistered (the meter owns them
68    /// for process life). Constructing many `Observability` instances would
69    /// therefore accumulate callbacks. `PhantomListener` / `PhantomSession`
70    /// each hold one shared `Arc<Observability>` — that is the intended
71    /// usage.
72    pub fn new(config: ObservabilityConfig) -> Arc<Self> {
73        let instruments = PhantomInstruments::new(&config);
74        let atomics = Arc::new(HotPathAtomics::new());
75        // Register OTel observable callbacks that read the atomic counters
76        // on each export tick. The atomics live behind `Arc` so the
77        // callbacks own a strong ref independent of `Observability`'s
78        // lifetime.
79        bridge::register_observables(&atomics, &config);
80        Arc::new(Self {
81            config,
82            atomics,
83            instruments,
84        })
85    }
86
87    /// Borrow the captured configuration.
88    pub fn config(&self) -> &ObservabilityConfig {
89        &self.config
90    }
91
92    /// Capture a cold-path snapshot of all counters and gauges.
93    pub fn snapshot(&self) -> MetricsSnapshot {
94        MetricsSnapshot::capture(&self.atomics)
95    }
96
97    // --- Hot path recording ---
98
99    #[inline]
100    pub fn record_send(&self, bytes: usize, leg: LegType) {
101        self.atomics.record_send(bytes, leg);
102    }
103
104    #[inline]
105    pub fn record_recv(&self, bytes: usize, leg: LegType) {
106        self.atomics.record_recv(bytes, leg);
107    }
108
109    #[inline]
110    pub fn record_encrypt_ns(&self, duration_ns: u64) {
111        self.atomics.record_encrypt_ns(duration_ns);
112    }
113
114    #[inline]
115    pub fn record_decrypt_ns(&self, duration_ns: u64) {
116        self.atomics.record_decrypt_ns(duration_ns);
117    }
118
119    #[inline]
120    pub fn record_rtt_us(&self, rtt_us: u64, path_id: u8) {
121        self.atomics.record_rtt_us(rtt_us, path_id);
122    }
123
124    // --- Gauges ---
125
126    /// Mark a new session as opened. Updates both the lock-free gauge and
127    /// the OTel `UpDownCounter` (when the `telemetry-otel` feature is on).
128    #[inline]
129    pub fn session_opened(&self, leg: LegType) {
130        self.atomics.session_opened();
131        self.instruments.session_opened(leg);
132    }
133
134    #[inline]
135    pub fn session_closed(&self, leg: LegType) {
136        self.atomics.session_closed();
137        self.instruments.session_closed(leg);
138    }
139
140    #[inline]
141    pub fn stream_opened(&self) {
142        self.atomics.stream_opened();
143        self.instruments.stream_opened();
144    }
145
146    #[inline]
147    pub fn stream_closed(&self) {
148        self.atomics.stream_closed();
149        self.instruments.stream_closed();
150    }
151
152    /// Record a successful handshake completion with its duration (ns) into
153    /// the lock-free atomics only — no OTel attribution.
154    ///
155    /// This is the snapshot-only path: the duration accumulates in atomic
156    /// sum+count fields surfaced via [`Self::snapshot`]. Callers that also
157    /// want the labeled OTel `Histogram` use [`Self::record_handshake`]
158    /// (which calls this internally for the success case).
159    pub fn record_handshake_success(&self, duration_ns: u64) {
160        self.atomics.record_handshake_success(duration_ns);
161    }
162
163    /// Record a handshake failure into the lock-free atomics only.
164    ///
165    /// Cause attribution (cookie / signature / transcript / KEM) is carried
166    /// by the labeled OTel path in [`Self::record_handshake`], not here.
167    #[inline]
168    pub fn record_handshake_failure(&self) {
169        self.atomics.record_handshake_failure();
170    }
171
172    // --- Labeled OTel event recorders ---
173
174    /// Record a handshake outcome and its latency with full OTel
175    /// attribution. The lock-free atomic counters and the OTel
176    /// `Histogram` (`{ns}.handshake.duration`) are updated together; the
177    /// histogram's `_count` series — sliced by the `outcome` attribute —
178    /// is the canonical handshake count, so there is no separate counter.
179    pub fn record_handshake(
180        &self,
181        duration: std::time::Duration,
182        outcome: HandshakeOutcome,
183        leg: LegType,
184        cipher: AeadAlgorithm,
185        version: ProtocolVersion,
186    ) {
187        let duration_ns = u64::try_from(duration.as_nanos()).unwrap_or(u64::MAX);
188        match outcome {
189            HandshakeOutcome::Success => self.atomics.record_handshake_success(duration_ns),
190            HandshakeOutcome::Failure => self.atomics.record_handshake_failure(),
191        }
192        self.instruments.record_handshake_duration(
193            duration.as_secs_f64(),
194            outcome,
195            leg,
196            cipher,
197            version,
198        );
199    }
200
201    /// Record a `PATH_VALIDATION` exchange latency.
202    pub fn record_path_validation(
203        &self,
204        duration: std::time::Duration,
205        path_id: u8,
206        outcome: PathValidationOutcome,
207    ) {
208        self.instruments
209            .record_path_validation_duration(duration.as_secs_f64(), path_id, outcome);
210    }
211
212    pub fn record_resumption(&self, mode: ResumptionMode, accepted: bool) {
213        self.instruments.record_resumption(mode, accepted);
214    }
215
216    #[inline]
217    pub fn record_replay_rejected(&self, reason: ReplayReason) {
218        self.instruments.record_replay_rejected(reason);
219    }
220
221    #[inline]
222    pub fn record_aead_failure(&self, leg: LegType, algorithm: AeadAlgorithm) {
223        self.instruments.record_aead_failure(leg, algorithm);
224    }
225
226    #[inline]
227    pub fn record_unencrypted_dropped(&self, leg: LegType) {
228        self.instruments.record_unencrypted_dropped(leg);
229    }
230
231    pub fn record_path_migration(&self, from: u8, to: u8) {
232        self.instruments.record_path_migration(from, to);
233    }
234
235    pub fn record_cookie(&self, outcome: CookieOutcome) {
236        self.instruments.record_cookie(outcome);
237    }
238
239    pub fn record_pow(&self, outcome: PowOutcome, difficulty: u8) {
240        self.instruments.record_pow(outcome, difficulty);
241    }
242
243    pub fn record_early_data(&self, outcome: EarlyDataOutcome) {
244        self.instruments.record_early_data(outcome);
245    }
246
247    pub fn record_rekey(&self, direction: Direction) {
248        self.instruments.record_rekey(direction);
249    }
250
251    pub fn record_fallback(&self, from_leg: LegType, to_leg: LegType, reason: FallbackReason) {
252        self.instruments.record_fallback(from_leg, to_leg, reason);
253    }
254}
255
256#[cfg(test)]
257mod tests {
258    use super::*;
259
260    #[test]
261    fn default_observability_has_default_config() {
262        let obs = Observability::new(ObservabilityConfig::default());
263        assert_eq!(obs.config().namespace.as_ref(), "phantom");
264    }
265
266    #[test]
267    fn new_returns_arc_with_provided_config() {
268        let cfg = ObservabilityConfig::builder().namespace("myapp").build();
269        let obs = Observability::new(cfg);
270        assert_eq!(obs.config().namespace.as_ref(), "myapp");
271        // Cloning the Arc preserves identity.
272        let obs2 = obs.clone();
273        assert_eq!(obs2.config().namespace.as_ref(), "myapp");
274    }
275
276    #[test]
277    fn record_send_round_trips_through_snapshot() {
278        let obs = Observability::new(ObservabilityConfig::default());
279        obs.record_send(1024, LegType::Tcp);
280        obs.record_send(2048, LegType::Tcp);
281        obs.record_recv(512, LegType::Kcp);
282        obs.record_encrypt_ns(100);
283        obs.record_encrypt_ns(300);
284        obs.session_opened(LegType::Tcp);
285        obs.stream_opened();
286        obs.stream_opened();
287        obs.record_rtt_us(5_000, 0);
288
289        let s = obs.snapshot();
290        assert_eq!(s.packets_sent, 2);
291        assert_eq!(s.packets_recv, 1);
292        assert_eq!(s.bytes_sent, 3072);
293        assert_eq!(s.bytes_recv, 512);
294        assert_eq!(s.avg_encrypt_ns, 200);
295        assert_eq!(s.encrypt_count, 2);
296        assert_eq!(s.active_sessions, 1);
297        assert_eq!(s.active_streams, 2);
298        assert_eq!(s.rtt_us_path_0, 5_000);
299    }
300}
phantom_protocol/observability/mod.rs

phantom_protocol/observability/
mod.rs
