rightkit_daemon/
metrics.rs

1//! Request, latency, in-flight and connection-table counters of a running host (rightkit#1280).
2//!
3//! Read them as numbers with [`Hosted::metrics`](crate::Hosted::metrics): requests per method (and
4//! per method per minute over the last hour), a latency histogram per method, error codes, the
5//! most requests that were in flight at once, and the connection table's occupancy, its peak and
6//! its cap. Counts and durations only, never request content. Recording takes one short lock per
7//! request; nothing polls.
8
9use std::collections::{BTreeMap, VecDeque};
10use std::sync::atomic::{AtomicUsize, Ordering};
11use std::sync::Mutex;
12use std::time::{Duration, SystemTime, UNIX_EPOCH};
13
14/// Upper bounds of the latency histogram buckets, in milliseconds. A request slower than the last
15/// bound is counted in the final, open bucket (one more count than bounds).
16pub const LATENCY_BOUNDS_MS: [u64; 10] = [1, 5, 10, 25, 50, 100, 250, 500, 1_000, 5_000];
17/// Distinct method names counted apart; further names are counted under [`OTHER_METHOD`].
18pub const MAX_METHODS: usize = 256;
19/// The bucket for methods past [`MAX_METHODS`].
20pub const OTHER_METHOD: &str = "(other)";
21/// Minutes of per-minute history kept per method.
22const MINUTES_KEPT: usize = 60;
23
24/// What the counters hold for one method.
25#[derive(Debug, Clone, PartialEq, Eq, Default)]
26pub struct MethodMetrics {
27    /// Requests that reached the handler.
28    pub requests: u64,
29    /// Requests that returned an error, by error code.
30    pub errors: BTreeMap<String, u64>,
31    /// Requests per latency bucket: index `i` counts latencies up to
32    /// [`LATENCY_BOUNDS_MS`]`[i]`, the last index everything slower.
33    pub latency_buckets: [u64; LATENCY_BOUNDS_MS.len() + 1],
34    /// Sum of the latencies, microseconds.
35    pub total_micros: u64,
36    /// The slowest request, microseconds.
37    pub max_micros: u64,
38    /// `(minute since the epoch, requests in it)`, oldest first, at most the last hour.
39    pub per_minute: Vec<(u64, u64)>,
40}
41
42impl MethodMetrics {
43    /// Requests that returned an error, in all.
44    pub fn error_count(&self) -> u64 {
45        self.errors.values().sum()
46    }
47
48    /// Requests in the minute that contains `now` (wall clock), for a rate reading.
49    pub fn requests_in_minute_of(&self, now: SystemTime) -> u64 {
50        let minute = minute_of(now);
51        self.per_minute
52            .iter()
53            .find(|(m, _)| *m == minute)
54            .map_or(0, |(_, n)| *n)
55    }
56}
57
58/// A reading of the counters, from [`Hosted::metrics`](crate::Hosted::metrics).
59#[derive(Debug, Clone, PartialEq, Eq, Default)]
60pub struct HostMetrics {
61    /// Counters per method.
62    pub methods: BTreeMap<String, MethodMetrics>,
63    /// Requests being run now.
64    pub inflight: usize,
65    /// The most requests that were in flight at once.
66    pub inflight_max: usize,
67    /// Connections held now (admitted, handshake or not).
68    pub connections: usize,
69    /// The most connections held at once.
70    pub connections_peak: usize,
71    /// The cap of the connection table ([`HostOptions::max_connections`](crate::HostOptions)); on
72    /// Windows the pipe-instance cap bounds the table instead, and this is that cap.
73    pub connection_limit: usize,
74    /// Connections refused because the table was full.
75    pub connections_refused: u64,
76}
77
78impl HostMetrics {
79    /// Every request counted, across methods.
80    pub fn total_requests(&self) -> u64 {
81        self.methods.values().map(|m| m.requests).sum()
82    }
83}
84
85#[derive(Default)]
86struct PerMethod {
87    stats: MethodMetrics,
88    minutes: VecDeque<(u64, u64)>,
89}
90
91/// The live counters. One per host.
92pub(crate) struct Metrics {
93    methods: Mutex<BTreeMap<String, PerMethod>>,
94    inflight_max: AtomicUsize,
95    connections_peak: AtomicUsize,
96    refused: std::sync::atomic::AtomicU64,
97    connection_limit: usize,
98}
99
100fn minute_of(now: SystemTime) -> u64 {
101    now.duration_since(UNIX_EPOCH)
102        .map(|d| d.as_secs() / 60)
103        .unwrap_or(0)
104}
105
106impl Metrics {
107    pub(crate) fn new(connection_limit: usize) -> Self {
108        Self {
109            methods: Mutex::new(BTreeMap::new()),
110            inflight_max: AtomicUsize::new(0),
111            connections_peak: AtomicUsize::new(0),
112            refused: std::sync::atomic::AtomicU64::new(0),
113            connection_limit,
114        }
115    }
116
117    /// A request started; `now_inflight` is the count including it.
118    pub(crate) fn note_inflight(&self, now_inflight: usize) {
119        self.inflight_max.fetch_max(now_inflight, Ordering::SeqCst);
120    }
121
122    /// The table now holds `live` connections.
123    pub(crate) fn note_connections(&self, live: usize) {
124        self.connections_peak.fetch_max(live, Ordering::SeqCst);
125    }
126
127    pub(crate) fn note_refused(&self) {
128        self.refused.fetch_add(1, Ordering::SeqCst);
129    }
130
131    /// One finished request.
132    pub(crate) fn record(&self, method: &str, took: Duration, error_code: Option<&str>) {
133        let mut methods = self.methods.lock().unwrap_or_else(|e| e.into_inner());
134        let key = if methods.contains_key(method) || methods.len() < MAX_METHODS {
135            method
136        } else {
137            OTHER_METHOD
138        };
139        let entry = methods.entry(key.to_string()).or_default();
140        entry.stats.requests += 1;
141        if let Some(code) = error_code {
142            *entry.stats.errors.entry(code.to_string()).or_insert(0) += 1;
143        }
144        let ms = took.as_millis() as u64;
145        let bucket = LATENCY_BOUNDS_MS
146            .iter()
147            .position(|bound| ms <= *bound)
148            .unwrap_or(LATENCY_BOUNDS_MS.len());
149        entry.stats.latency_buckets[bucket] += 1;
150        let micros = took.as_micros().min(u128::from(u64::MAX)) as u64;
151        entry.stats.total_micros = entry.stats.total_micros.saturating_add(micros);
152        entry.stats.max_micros = entry.stats.max_micros.max(micros);
153        let minute = minute_of(SystemTime::now());
154        match entry.minutes.back_mut() {
155            Some((m, n)) if *m == minute => *n += 1,
156            _ => {
157                entry.minutes.push_back((minute, 1));
158                while entry.minutes.len() > MINUTES_KEPT {
159                    entry.minutes.pop_front();
160                }
161            }
162        }
163    }
164
165    pub(crate) fn snapshot(&self, inflight: usize, connections: usize) -> HostMetrics {
166        let methods = self
167            .methods
168            .lock()
169            .unwrap_or_else(|e| e.into_inner())
170            .iter()
171            .map(|(name, m)| {
172                let mut stats = m.stats.clone();
173                stats.per_minute = m.minutes.iter().copied().collect();
174                (name.clone(), stats)
175            })
176            .collect();
177        HostMetrics {
178            methods,
179            inflight,
180            inflight_max: self.inflight_max.load(Ordering::SeqCst),
181            connections,
182            connections_peak: self.connections_peak.load(Ordering::SeqCst),
183            connection_limit: self.connection_limit,
184            connections_refused: self.refused.load(Ordering::SeqCst),
185        }
186    }
187}
rightkit_daemon/metrics.rs

rightkit_daemon/
metrics.rs
