subc_daemon/
observability.rs

1use std::{
2    collections::{HashMap, VecDeque},
3    sync::{
4        atomic::{AtomicU64, Ordering},
5        Arc, Mutex, MutexGuard,
6    },
7    time::Duration,
8};
9
10use serde_json::{json, Value};
11use tracing::debug;
12
13use crate::registry::ConnectionId;
14
15/// The only keys the route.open refusal counter may carry. A module's own
16/// error code on a rejected bind is counted under `module_rejected` and rides
17/// the log event as a separate field: if the module's string were the key, a
18/// module could grow this map for the daemon's lifetime and push terminal
19/// control sequences through `ck daemon` to an operator's screen. The
20/// `&'static str` increment signature plus the debug assertion keep the set
21/// closed at the call site, not just here.
22const ROUTE_OPEN_REFUSAL_COUNTER_CODES: &[&str] = &[
23    "module_warming",
24    ROUTE_OPEN_REFUSED_DECLARED_NOT_READY,
25    ROUTE_OPEN_REFUSED_REQUIRED_CAPABILITY_UNPROVIDED,
26    "target_unavailable",
27    subc_protocol::error_codes::TARGET_FLOW_UNSUPPORTED,
28    subc_protocol::error_codes::TARGET_AGENT_RUN_UNSUPPORTED,
29    "module_removed",
30    "module_no_protocol",
31    "unknown_module",
32    "module_reloading",
33    "op_not_allowed",
34    "bad_consumer_identity",
35    "capability_forbidden",
36    "admission_facts_not_permitted",
37    "admission_facts_target_not_allowed",
38    "route_limit",
39    "forwarding_error",
40    "module_timeout",
41    ROUTE_OPEN_REFUSED_BREAKER_OPEN,
42    "module_rejected",
43    subc_protocol::error_codes::SCOPE_EPOCH_REQUIRED,
44    subc_protocol::error_codes::SCOPE_NOT_SYNCED,
45    subc_protocol::error_codes::SCOPE_NOT_LIVE,
46    subc_protocol::error_codes::SCOPE_ENDED,
47    subc_protocol::error_codes::SCOPE_NOT_CARRIER,
48    subc_protocol::error_codes::SCOPE_BIND_PROOF_REQUIRED,
49    subc_protocol::error_codes::SCOPE_BIND_PROOF_MISMATCH,
50    subc_protocol::error_codes::SCOPE_CHANGED,
51    subc_protocol::error_codes::INVALID_REQUEST,
52];
53
54/// Counter key for a `route.open` refused by the per-module bind-relay breaker
55/// before any relay was attempted.
56///
57/// The frame the caller receives carries `module_timeout`, because both SDKs
58/// already classify that as retryable with capped backoff and inventing a new
59/// wire code would need a change in each of them. The COUNTER is deliberately a
60/// different key: "this module burned the full bind budget" and "this module is
61/// being refused in microseconds because it already did that repeatedly" are
62/// the two states an operator most needs to tell apart, and they are
63/// indistinguishable from the client side, where both look like one retryable
64/// error that the next attempt may well satisfy.
65pub(crate) const ROUTE_OPEN_REFUSED_BREAKER_OPEN: &str = "module_timeout_breaker_open";
66
67/// Counter key for a registered module that declared itself not ready.
68///
69/// The caller still receives `module_warming`, but operators must be able to
70/// distinguish declared readiness from a supervised process that has not
71/// registered yet.
72pub(crate) const ROUTE_OPEN_REFUSED_DECLARED_NOT_READY: &str = "module_warming_declared_not_ready";
73
74/// Counter key for a registered, declared-ready module held not-ready because
75/// a capability it declares `need: required` has no registered provider.
76///
77/// The caller still receives `module_warming`; a separate key lets an operator
78/// tell "the module says it is warming" from "the module is waiting on a
79/// provider that is not running", which point at different fixes.
80pub(crate) const ROUTE_OPEN_REFUSED_REQUIRED_CAPABILITY_UNPROVIDED: &str =
81    "module_warming_required_capability_unprovided";
82
83/// Shared count of authenticated socket connections accepted by the daemon.
84#[derive(Debug, Clone, Default)]
85pub struct ConnectedClients {
86    count: Arc<AtomicU64>,
87}
88
89impl ConnectedClients {
90    pub fn new() -> Self {
91        Self::default()
92    }
93
94    pub fn count(&self) -> u64 {
95        self.count.load(Ordering::SeqCst)
96    }
97
98    pub(crate) fn open(&self, connection_id: ConnectionId) -> ConnectedClientGuard {
99        let previous = self.count.fetch_add(1, Ordering::SeqCst);
100        let current = previous + 1;
101        // DEBUG, not INFO: this pair fired on every connect and disconnect and was
102        // measured at 43-74% of the daemon's own log (issue #114), burying the
103        // lines an operator opens the file for. The count itself is not lost: it
104        // is served live as `connected_clients` on server.describe (`ck daemon`),
105        // and a connection that opens a route still names its connection_id on the
106        // `route.open accepted` line, so per-connection forensics survive where
107        // they matter. Raise the filter (`CK_LOG=subc=debug`) to see every churn.
108        debug!(
109            connection_id = connection_id.get(),
110            connected_clients = current,
111            previous_connected_clients = previous,
112            "authenticated connection count changed"
113        );
114        ConnectedClientGuard {
115            clients: self.clone(),
116            connection_id,
117        }
118    }
119}
120
121pub(crate) struct ConnectedClientGuard {
122    clients: ConnectedClients,
123    connection_id: ConnectionId,
124}
125
126impl Drop for ConnectedClientGuard {
127    fn drop(&mut self) {
128        let previous = self.clients.count.fetch_sub(1, Ordering::SeqCst);
129        let current = previous.saturating_sub(1);
130        // DEBUG for the same reason as the open side above.
131        debug!(
132            connection_id = self.connection_id.get(),
133            connected_clients = current,
134            previous_connected_clients = previous,
135            "authenticated connection count changed"
136        );
137    }
138}
139
140/// Lock-free counters for route lifecycle drops and delivery failures.
141#[derive(Debug, Clone, Default)]
142pub struct DaemonCounters {
143    /// Every non-request frame a module sent on a (channel, epoch) the daemon
144    /// holds no bound route for, dropped. Counts both kinds below: orphan
145    /// traffic on a route the daemon released, and frames on a (channel,
146    /// epoch) that was never allocated on that module connection.
147    module_frames_dropped_no_route: Arc<AtomicU64>,
148    /// The orphan-traffic part of `module_frames_dropped_no_route`: the daemon
149    /// had allocated that (channel, epoch) on the sending module connection and
150    /// has since released it, so the module is still holding a route the
151    /// daemon closed. `module_frames_dropped_no_route` minus this is the count
152    /// of frames on a (channel, epoch) that never existed on the connection.
153    module_frames_dropped_released_route: Arc<AtomicU64>,
154    module_frames_dropped_released_route_by_module: Arc<Mutex<HashMap<String, u64>>>,
155    /// Route GOODBYEs the daemon enqueued to a module in answer to its frames
156    /// on a (channel, epoch) the daemon holds no route for, telling the module
157    /// to drop the route. Rate-limited per module connection and channel, so
158    /// this counts answers, not orphan frames.
159    module_orphan_route_goodbyes_sent: Arc<AtomicU64>,
160    // Per-module maps and the rate window are daemon-lifetime diagnostics only:
161    // they deliberately reset on restart instead of becoming durable daemon state.
162    module_frames_dropped_no_route_by_module: Arc<Mutex<HashMap<String, u64>>>,
163    route_open_refused_by_code: Arc<Mutex<HashMap<String, u64>>>,
164    route_open_accepted_by_principal: Arc<Mutex<HashMap<String, u64>>>,
165    module_frames_dropped_no_route_window: Arc<Mutex<DropWindow>>,
166    module_requests_dropped_stale_route: Arc<AtomicU64>,
167    client_frames_dropped_stale_route: Arc<AtomicU64>,
168    client_egress_close_delivery_failed: Arc<AtomicU64>,
169    goodbye_relay_client_failed: Arc<AtomicU64>,
170    goodbye_relay_module_dropped: Arc<AtomicU64>,
171    goodbye_relay_module_dropped_by_module: Arc<Mutex<HashMap<String, u64>>>,
172    route_released_epoch_fenced: Arc<AtomicU64>,
173    route_release_stale_skipped: Arc<AtomicU64>,
174    drains_with_undeclared_gauge: Arc<AtomicU64>,
175}
176
177/// Ten one-minute buckets make sustained module-to-client route drops visible
178/// without retaining one record for every dropped frame.
179#[derive(Debug)]
180struct DropWindow {
181    started_at: tokio::time::Instant,
182    buckets: VecDeque<DropBucket>,
183}
184
185#[derive(Debug)]
186struct DropBucket {
187    minute: u64,
188    count: u64,
189}
190
191impl Default for DropWindow {
192    fn default() -> Self {
193        Self {
194            started_at: tokio::time::Instant::now(),
195            buckets: VecDeque::new(),
196        }
197    }
198}
199
200impl DropWindow {
201    const MINUTE: Duration = Duration::from_secs(60);
202    const BUCKETS: u64 = 10;
203
204    fn record(&mut self, now: tokio::time::Instant) {
205        let minute = self.minute_at(now);
206        self.prune_before(minute);
207        match self.buckets.back_mut() {
208            Some(bucket) if bucket.minute == minute => bucket.count += 1,
209            _ => self.buckets.push_back(DropBucket { minute, count: 1 }),
210        }
211    }
212
213    fn count_last_10m(&mut self, now: tokio::time::Instant) -> u64 {
214        let minute = self.minute_at(now);
215        self.prune_before(minute);
216        self.buckets.iter().map(|bucket| bucket.count).sum()
217    }
218
219    fn nonzero_minutes_last_10m(&mut self, now: tokio::time::Instant) -> u64 {
220        let minute = self.minute_at(now);
221        self.prune_before(minute);
222        self.buckets.len() as u64
223    }
224
225    fn minute_at(&self, now: tokio::time::Instant) -> u64 {
226        now.saturating_duration_since(self.started_at).as_secs() / Self::MINUTE.as_secs()
227    }
228
229    fn prune_before(&mut self, current_minute: u64) {
230        while self
231            .buckets
232            .front()
233            .is_some_and(|bucket| current_minute.saturating_sub(bucket.minute) >= Self::BUCKETS)
234        {
235            self.buckets.pop_front();
236        }
237    }
238}
239
240impl DaemonCounters {
241    pub fn new() -> Self {
242        Self::default()
243    }
244
245    /// Returns a JSON snapshot whose stable, additive schema keeps the
246    /// `server.describe` diagnostic endpoint backward-compatible.
247    pub fn snapshot(&self) -> Value {
248        let mut snapshot = serde_json::Map::new();
249        snapshot.insert(
250            "module_frames_dropped_no_route".into(),
251            self.module_frames_dropped_no_route
252                .load(Ordering::Relaxed)
253                .into(),
254        );
255        let mut drop_window = self
256            .module_frames_dropped_no_route_window
257            .lock()
258            .expect("drop-rate window mutex poisoned");
259        let now = tokio::time::Instant::now();
260        snapshot.insert(
261            "module_frames_dropped_no_route_last_10m".into(),
262            drop_window.count_last_10m(now).into(),
263        );
264        snapshot.insert(
265            "module_frames_dropped_no_route_nonzero_minutes_last_10m".into(),
266            drop_window.nonzero_minutes_last_10m(now).into(),
267        );
268        insert_nonempty_counts(
269            &mut snapshot,
270            "module_frames_dropped_no_route_by_module",
271            &self.module_frames_dropped_no_route_by_module,
272        );
273        snapshot.insert(
274            "module_frames_dropped_released_route".into(),
275            self.module_frames_dropped_released_route
276                .load(Ordering::Relaxed)
277                .into(),
278        );
279        insert_nonempty_counts(
280            &mut snapshot,
281            "module_frames_dropped_released_route_by_module",
282            &self.module_frames_dropped_released_route_by_module,
283        );
284        snapshot.insert(
285            "module_orphan_route_goodbyes_sent".into(),
286            self.module_orphan_route_goodbyes_sent
287                .load(Ordering::Relaxed)
288                .into(),
289        );
290        insert_nonempty_counts(
291            &mut snapshot,
292            "route_open_refused_by_code",
293            &self.route_open_refused_by_code,
294        );
295        insert_nonempty_counts(
296            &mut snapshot,
297            "route_open_accepted_by_principal",
298            &self.route_open_accepted_by_principal,
299        );
300        snapshot.insert(
301            "module_requests_dropped_stale_route".into(),
302            self.module_requests_dropped_stale_route
303                .load(Ordering::Relaxed)
304                .into(),
305        );
306        snapshot.insert(
307            "client_frames_dropped_stale_route".into(),
308            self.client_frames_dropped_stale_route
309                .load(Ordering::Relaxed)
310                .into(),
311        );
312        snapshot.insert(
313            "client_egress_close_delivery_failed".into(),
314            self.client_egress_close_delivery_failed
315                .load(Ordering::Relaxed)
316                .into(),
317        );
318        snapshot.insert(
319            "goodbye_relay_client_failed".into(),
320            self.goodbye_relay_client_failed
321                .load(Ordering::Relaxed)
322                .into(),
323        );
324        snapshot.insert(
325            "goodbye_relay_module_dropped".into(),
326            self.goodbye_relay_module_dropped
327                .load(Ordering::Relaxed)
328                .into(),
329        );
330        insert_nonempty_counts(
331            &mut snapshot,
332            "goodbye_relay_module_dropped_by_module",
333            &self.goodbye_relay_module_dropped_by_module,
334        );
335        snapshot.insert(
336            "route_released_epoch_fenced".into(),
337            self.route_released_epoch_fenced
338                .load(Ordering::Relaxed)
339                .into(),
340        );
341        snapshot.insert(
342            "route_release_stale_skipped".into(),
343            self.route_release_stale_skipped
344                .load(Ordering::Relaxed)
345                .into(),
346        );
347        snapshot.insert(
348            "drains_with_undeclared_gauge".into(),
349            self.drains_with_undeclared_gauge
350                .load(Ordering::Relaxed)
351                .into(),
352        );
353        Value::Object(snapshot)
354    }
355
356    pub(crate) fn increment_module_frames_dropped_no_route(&self, module_id: Option<&str>) {
357        self.module_frames_dropped_no_route
358            .fetch_add(1, Ordering::Relaxed);
359        if let Some(module_id) = module_id {
360            increment_keyed_count(&self.module_frames_dropped_no_route_by_module, module_id);
361        }
362        self.module_frames_dropped_no_route_window
363            .lock()
364            .expect("drop-rate window mutex poisoned")
365            .record(tokio::time::Instant::now());
366    }
367
368    /// Count a dropped module frame whose (channel, epoch) the daemon had
369    /// allocated on that connection and since released. Called in addition to
370    /// [`Self::increment_module_frames_dropped_no_route`], never instead of it.
371    pub(crate) fn increment_module_frames_dropped_released_route(&self, module_id: Option<&str>) {
372        self.module_frames_dropped_released_route
373            .fetch_add(1, Ordering::Relaxed);
374        if let Some(module_id) = module_id {
375            increment_keyed_count(
376                &self.module_frames_dropped_released_route_by_module,
377                module_id,
378            );
379        }
380    }
381
382    pub(crate) fn increment_module_orphan_route_goodbyes_sent(&self) {
383        self.module_orphan_route_goodbyes_sent
384            .fetch_add(1, Ordering::Relaxed);
385    }
386
387    pub(crate) fn increment_route_open_refused(&self, code: &'static str) {
388        debug_assert!(ROUTE_OPEN_REFUSAL_COUNTER_CODES.contains(&code));
389        increment_keyed_count(&self.route_open_refused_by_code, code);
390    }
391
392    /// Count an accepted route.open by the principal the daemon stamped.
393    ///
394    /// THE KEY SPACE IS CLOSED BY CONSTRUCTION, unlike the refusal counter which
395    /// needs an explicit allowlist: a principal is `direct` or
396    /// `reserved:<module_id>`, and a module id was already refused at HELLO
397    /// unless it is a single path component free of control characters. So an
398    /// untrusted string cannot expand this map without first passing module-id
399    /// validation, and the bound is the number of modules rather than the number
400    /// of distinct strings a caller can invent.
401    pub(crate) fn increment_route_open_accepted(&self, principal: &str) {
402        increment_keyed_count(&self.route_open_accepted_by_principal, principal);
403    }
404
405    pub(crate) fn increment_module_requests_dropped_stale_route(&self) {
406        self.module_requests_dropped_stale_route
407            .fetch_add(1, Ordering::Relaxed);
408    }
409
410    pub(crate) fn increment_client_frames_dropped_stale_route(&self) {
411        self.client_frames_dropped_stale_route
412            .fetch_add(1, Ordering::Relaxed);
413    }
414
415    pub(crate) fn increment_client_egress_close_delivery_failed(&self) {
416        self.client_egress_close_delivery_failed
417            .fetch_add(1, Ordering::Relaxed);
418    }
419
420    pub(crate) fn increment_goodbye_relay_client_failed(&self) {
421        self.goodbye_relay_client_failed
422            .fetch_add(1, Ordering::Relaxed);
423    }
424
425    pub(crate) fn increment_goodbye_relay_module_dropped(&self, module_id: Option<&str>) {
426        self.goodbye_relay_module_dropped
427            .fetch_add(1, Ordering::Relaxed);
428        if let Some(module_id) = module_id {
429            increment_keyed_count(&self.goodbye_relay_module_dropped_by_module, module_id);
430        }
431    }
432
433    pub(crate) fn increment_route_released_epoch_fenced(&self) {
434        self.route_released_epoch_fenced
435            .fetch_add(1, Ordering::Relaxed);
436    }
437
438    pub(crate) fn increment_route_release_stale_skipped(&self) {
439        self.route_release_stale_skipped
440            .fetch_add(1, Ordering::Relaxed);
441    }
442
443    pub(crate) fn increment_drains_with_undeclared_gauge(&self) {
444        self.drains_with_undeclared_gauge
445            .fetch_add(1, Ordering::Relaxed);
446    }
447}
448
449fn increment_keyed_count(counts: &Mutex<HashMap<String, u64>>, key: &str) {
450    *counts
451        .lock()
452        .expect("keyed counter mutex poisoned")
453        .entry(key.to_string())
454        .or_default() += 1;
455}
456
457fn insert_nonempty_counts(
458    snapshot: &mut serde_json::Map<String, Value>,
459    key: &str,
460    counts: &Mutex<HashMap<String, u64>>,
461) {
462    let counts: MutexGuard<'_, HashMap<String, u64>> =
463        counts.lock().expect("keyed counter mutex poisoned");
464    if !counts.is_empty() {
465        snapshot.insert(key.to_string(), json!(&*counts));
466    }
467}
468
469#[cfg(test)]
470mod tests {
471    use super::*;
472
473    #[test]
474    fn counter_snapshot_includes_zero_rate_and_omits_empty_module_maps() {
475        let counters = DaemonCounters::new();
476        let snapshot = counters.snapshot();
477
478        assert_eq!(snapshot["module_frames_dropped_no_route_last_10m"], 0);
479        assert_eq!(
480            snapshot["module_frames_dropped_no_route_nonzero_minutes_last_10m"],
481            0
482        );
483        assert!(snapshot
484            .get("module_frames_dropped_no_route_by_module")
485            .is_none());
486        assert!(snapshot
487            .get("goodbye_relay_module_dropped_by_module")
488            .is_none());
489    }
490
491    #[tokio::test(start_paused = true)]
492    async fn module_frame_drops_are_attributed_to_the_emitting_module() {
493        let counters = DaemonCounters::new();
494        counters.increment_module_frames_dropped_no_route(Some("alpha"));
495        counters.increment_module_frames_dropped_no_route(Some("alpha"));
496
497        let snapshot = counters.snapshot();
498        assert_eq!(snapshot["module_frames_dropped_no_route"], 2);
499        assert_eq!(
500            snapshot["module_frames_dropped_no_route_by_module"],
501            json!({ "alpha": 2 })
502        );
503        assert_eq!(snapshot["module_frames_dropped_no_route_last_10m"], 2);
504    }
505
506    #[tokio::test(start_paused = true)]
507    async fn frame_drop_rate_ages_out_after_ten_minute_buckets() {
508        let counters = DaemonCounters::new();
509        counters.increment_module_frames_dropped_no_route(Some("alpha"));
510
511        tokio::time::advance(Duration::from_secs(9 * 60)).await;
512        assert_eq!(
513            counters.snapshot()["module_frames_dropped_no_route_last_10m"],
514            1
515        );
516
517        tokio::time::advance(Duration::from_secs(60)).await;
518        assert_eq!(
519            counters.snapshot()["module_frames_dropped_no_route_last_10m"],
520            0
521        );
522    }
523
524    #[tokio::test(start_paused = true)]
525    async fn frame_drop_window_counts_only_nonzero_minutes() {
526        let counters = DaemonCounters::new();
527        for minute in 0..10 {
528            if minute > 0 {
529                tokio::time::advance(Duration::from_secs(60)).await;
530            }
531            if minute != 4 {
532                counters.increment_module_frames_dropped_no_route(Some("alpha"));
533            }
534        }
535
536        assert_eq!(
537            counters.snapshot()["module_frames_dropped_no_route_nonzero_minutes_last_10m"],
538            9
539        );
540    }
541
542    #[test]
543    fn goodbye_relay_drops_are_attributed_to_the_target_module() {
544        let counters = DaemonCounters::new();
545        counters.increment_goodbye_relay_module_dropped(Some("alpha"));
546
547        assert_eq!(
548            counters.snapshot()["goodbye_relay_module_dropped_by_module"],
549            json!({ "alpha": 1 })
550        );
551    }
552}
subc_daemon/observability.rs

subc_daemon/
observability.rs
