connpool/observability/
stats.rs

1//! `PoolStats`: an immutable metrics snapshot, and the collector that feeds it.
2
3use std::collections::BTreeMap;
4use std::error::Error;
5use std::sync::Mutex;
6
7use crate::core::connection::{CloseReason, PooledConnection};
8use crate::observability::events::PoolListener;
9use crate::resilience::breaker::BreakerState;
10
11/// Caps how many recent acquire-wait / create-latency samples are kept. avg and all
12/// percentiles are computed over this same window, so they always describe the same bounded
13/// population.
14const ACQUIRE_WAIT_WINDOW: usize = 1024;
15
16/// A value in the [`PoolStats::to_dictionary`] wire representation.
17#[derive(Debug, Clone, PartialEq)]
18pub enum StatValue {
19    /// A string value.
20    Str(String),
21    /// An integer value.
22    Int(i64),
23    /// A floating-point value.
24    Float(f64),
25    /// A map of close-reason -> count.
26    Map(BTreeMap<String, i64>),
27    /// A null value (used for an absent breaker state).
28    Null,
29}
30
31/// Linear-interpolated percentile of an already-sorted slice. `q` in `[0, 1]`. Empty -> `0.0`.
32fn percentile(sorted: &[f64], q: f64) -> f64 {
33    if sorted.is_empty() {
34        return 0.0;
35    }
36    let k = (sorted.len() - 1) as f64 * q;
37    let lo = k as usize;
38    let hi = (lo + 1).min(sorted.len() - 1);
39    sorted[lo] + (sorted[hi] - sorted[lo]) * (k - lo as f64)
40}
41
42fn average(values: &[f64]) -> f64 {
43    if values.is_empty() {
44        return 0.0;
45    }
46    values.iter().sum::<f64>() / values.len() as f64
47}
48
49/// An immutable snapshot of pool metrics. The keys of [`PoolStats::to_dictionary`] are
50/// snake_case wire strings identical across every poolyglot language.
51#[derive(Debug, Clone, PartialEq)]
52pub struct PoolStats {
53    /// Pool name.
54    pub name: String,
55    /// Configured hard cap.
56    pub max_size: usize,
57    /// idle + in use + being created.
58    pub size: usize,
59    /// Available for checkout right now.
60    pub idle: usize,
61    /// Checked out, or briefly reserved by a background idle check.
62    pub in_use: usize,
63    /// Callers currently blocked in `acquire()`.
64    pub waiting: usize,
65
66    /// Total connections created.
67    pub created: u64,
68    /// Total create failures.
69    pub create_failures: u64,
70    /// Total acquires.
71    pub acquired: u64,
72    /// Total releases.
73    pub released: u64,
74    /// Total acquire timeouts.
75    pub timeouts: u64,
76    /// Total connections closed.
77    pub closed: u64,
78    /// Borrowed connections flagged as possibly leaked.
79    pub leaks: u64,
80    /// Close counts by reason wire string.
81    pub closed_by_reason: BTreeMap<String, i64>,
82
83    /// Average acquire-wait seconds over the window.
84    pub avg_acquire_wait: f64,
85    /// p50 acquire-wait.
86    pub p50_acquire_wait: f64,
87    /// p95 acquire-wait.
88    pub p95_acquire_wait: f64,
89    /// p99 acquire-wait.
90    pub p99_acquire_wait: f64,
91
92    /// Average create-latency seconds over the window.
93    pub avg_create_latency: f64,
94    /// p50 create-latency.
95    pub p50_create_latency: f64,
96    /// p95 create-latency.
97    pub p95_create_latency: f64,
98    /// p99 create-latency.
99    pub p99_create_latency: f64,
100
101    /// Current breaker state wire string, if a breaker is enabled.
102    pub breaker_state: Option<String>,
103}
104
105impl PoolStats {
106    /// The snake_case wire representation, identical across every poolyglot language.
107    pub fn to_dictionary(&self) -> BTreeMap<&'static str, StatValue> {
108        let mut m = BTreeMap::new();
109        m.insert("name", StatValue::Str(self.name.clone()));
110        m.insert("max_size", StatValue::Int(self.max_size as i64));
111        m.insert("size", StatValue::Int(self.size as i64));
112        m.insert("idle", StatValue::Int(self.idle as i64));
113        m.insert("in_use", StatValue::Int(self.in_use as i64));
114        m.insert("waiting", StatValue::Int(self.waiting as i64));
115        m.insert("created", StatValue::Int(self.created as i64));
116        m.insert("create_failures", StatValue::Int(self.create_failures as i64));
117        m.insert("acquired", StatValue::Int(self.acquired as i64));
118        m.insert("released", StatValue::Int(self.released as i64));
119        m.insert("timeouts", StatValue::Int(self.timeouts as i64));
120        m.insert("closed", StatValue::Int(self.closed as i64));
121        m.insert("leaks", StatValue::Int(self.leaks as i64));
122        m.insert("closed_by_reason", StatValue::Map(self.closed_by_reason.clone()));
123        m.insert("avg_acquire_wait", StatValue::Float(self.avg_acquire_wait));
124        m.insert("p50_acquire_wait", StatValue::Float(self.p50_acquire_wait));
125        m.insert("p95_acquire_wait", StatValue::Float(self.p95_acquire_wait));
126        m.insert("p99_acquire_wait", StatValue::Float(self.p99_acquire_wait));
127        m.insert("avg_create_latency", StatValue::Float(self.avg_create_latency));
128        m.insert("p50_create_latency", StatValue::Float(self.p50_create_latency));
129        m.insert("p95_create_latency", StatValue::Float(self.p95_create_latency));
130        m.insert("p99_create_latency", StatValue::Float(self.p99_create_latency));
131        m.insert(
132            "breaker_state",
133            match &self.breaker_state {
134                Some(s) => StatValue::Str(s.clone()),
135                None => StatValue::Null,
136            },
137        );
138        m
139    }
140}
141
142/// The point-in-time gauges a collector doesn't itself track.
143pub(crate) struct StatsGauges {
144    pub name: String,
145    pub max_size: usize,
146    pub size: usize,
147    pub idle: usize,
148    pub in_use: usize,
149    pub waiting: usize,
150    pub breaker_state: Option<String>,
151}
152
153struct Counters {
154    created: u64,
155    create_failures: u64,
156    acquired: u64,
157    released: u64,
158    timeouts: u64,
159    closed: u64,
160    leaks: u64,
161    by_reason: BTreeMap<String, i64>,
162    waits: Vec<f64>,
163    creates: Vec<f64>,
164}
165
166/// A [`PoolListener`] that counts events and keeps bounded windows of acquire-wait and
167/// create-latency samples for percentile reporting.
168pub(crate) struct StatsCollector<T> {
169    inner: Mutex<Counters>,
170    _marker: std::marker::PhantomData<fn(T)>,
171}
172
173fn push_window(window: &mut Vec<f64>, value: f64) {
174    window.push(value);
175    if window.len() > ACQUIRE_WAIT_WINDOW {
176        let overflow = window.len() - ACQUIRE_WAIT_WINDOW;
177        window.drain(0..overflow);
178    }
179}
180
181impl<T> StatsCollector<T> {
182    pub(crate) fn new() -> Self {
183        Self {
184            inner: Mutex::new(Counters {
185                created: 0,
186                create_failures: 0,
187                acquired: 0,
188                released: 0,
189                timeouts: 0,
190                closed: 0,
191                leaks: 0,
192                by_reason: BTreeMap::new(),
193                waits: Vec::new(),
194                creates: Vec::new(),
195            }),
196            _marker: std::marker::PhantomData,
197        }
198    }
199
200    pub(crate) fn snapshot(&self, g: StatsGauges) -> PoolStats {
201        let c = self.inner.lock().unwrap();
202        let mut waits = c.waits.clone();
203        waits.sort_by(|a, b| a.partial_cmp(b).unwrap());
204        let mut creates = c.creates.clone();
205        creates.sort_by(|a, b| a.partial_cmp(b).unwrap());
206        PoolStats {
207            name: g.name,
208            max_size: g.max_size,
209            size: g.size,
210            idle: g.idle,
211            in_use: g.in_use,
212            waiting: g.waiting,
213            created: c.created,
214            create_failures: c.create_failures,
215            acquired: c.acquired,
216            released: c.released,
217            timeouts: c.timeouts,
218            closed: c.closed,
219            leaks: c.leaks,
220            closed_by_reason: c.by_reason.clone(),
221            avg_acquire_wait: average(&waits),
222            p50_acquire_wait: percentile(&waits, 0.50),
223            p95_acquire_wait: percentile(&waits, 0.95),
224            p99_acquire_wait: percentile(&waits, 0.99),
225            avg_create_latency: average(&creates),
226            p50_create_latency: percentile(&creates, 0.50),
227            p95_create_latency: percentile(&creates, 0.95),
228            p99_create_latency: percentile(&creates, 0.99),
229            breaker_state: g.breaker_state,
230        }
231    }
232}
233
234impl<T> PoolListener<T> for StatsCollector<T> {
235    fn on_connection_created(&self, _conn: &PooledConnection<T>, duration: f64) {
236        let mut c = self.inner.lock().unwrap();
237        c.created += 1;
238        push_window(&mut c.creates, duration);
239    }
240
241    fn on_connection_closed(&self, _conn: &PooledConnection<T>, reason: CloseReason) {
242        let mut c = self.inner.lock().unwrap();
243        c.closed += 1;
244        *c.by_reason.entry(reason.as_str().to_string()).or_insert(0) += 1;
245    }
246
247    fn on_create_failed(&self, _error: &(dyn Error + 'static)) {
248        self.inner.lock().unwrap().create_failures += 1;
249    }
250
251    fn on_acquired(&self, _conn: &PooledConnection<T>, wait_time: f64) {
252        let mut c = self.inner.lock().unwrap();
253        c.acquired += 1;
254        push_window(&mut c.waits, wait_time);
255    }
256
257    fn on_released(&self, _conn: &PooledConnection<T>, _hold_time: f64) {
258        self.inner.lock().unwrap().released += 1;
259    }
260
261    fn on_acquire_timeout(&self, _waited: f64) {
262        self.inner.lock().unwrap().timeouts += 1;
263    }
264
265    fn on_connection_leak(&self, _conn: &PooledConnection<T>, _held_for: f64) {
266        self.inner.lock().unwrap().leaks += 1;
267    }
268
269    fn on_breaker_state_change(&self, _old: BreakerState, _new: BreakerState) {}
270}
271
272#[cfg(test)]
273mod tests {
274    use super::*;
275
276    fn gauges() -> StatsGauges {
277        StatsGauges {
278            name: "p".into(),
279            max_size: 10,
280            size: 2,
281            idle: 1,
282            in_use: 1,
283            waiting: 0,
284            breaker_state: None,
285        }
286    }
287
288    #[test]
289    fn percentile_and_average_edge_cases() {
290        assert_eq!(percentile(&[], 0.5), 0.0);
291        assert_eq!(average(&[]), 0.0);
292        let v = vec![1.0, 2.0, 3.0, 4.0];
293        assert_eq!(percentile(&v, 0.0), 1.0);
294        assert_eq!(percentile(&v, 1.0), 4.0);
295        assert_eq!(percentile(&v, 0.5), 2.5); // interpolated
296        assert_eq!(average(&v), 2.5);
297    }
298
299    #[test]
300    fn collector_counts_and_snapshots() {
301        let c = StatsCollector::<u32>::new();
302        let pc = PooledConnection::new(1u32, 0.0, None);
303        c.on_connection_created(&pc, 0.2);
304        c.on_connection_created(&pc, 0.4);
305        c.on_acquired(&pc, 0.1);
306        c.on_released(&pc, 1.0);
307        c.on_acquire_timeout(2.0);
308        c.on_connection_closed(&pc, CloseReason::Idle);
309        c.on_connection_closed(&pc, CloseReason::Idle);
310        c.on_connection_leak(&pc, 5.0);
311        let e: Box<dyn Error + Send + Sync> = "x".into();
312        c.on_create_failed(&*e);
313        c.on_breaker_state_change(BreakerState::Closed, BreakerState::Open);
314
315        let s = c.snapshot(gauges());
316        assert_eq!(s.created, 2);
317        assert_eq!(s.acquired, 1);
318        assert_eq!(s.released, 1);
319        assert_eq!(s.timeouts, 1);
320        assert_eq!(s.closed, 2);
321        assert_eq!(s.leaks, 1);
322        assert_eq!(s.create_failures, 1);
323        assert_eq!(s.closed_by_reason.get("idle"), Some(&2));
324        assert!((s.avg_create_latency - 0.3).abs() < 1e-9);
325        assert_eq!(s.name, "p");
326        assert_eq!(s.max_size, 10);
327    }
328
329    #[test]
330    fn to_dictionary_has_snakecase_keys() {
331        let c = StatsCollector::<u32>::new();
332        let mut g = gauges();
333        g.breaker_state = Some("closed".into());
334        let d = c.snapshot(g).to_dictionary();
335        for key in [
336            "name",
337            "max_size",
338            "size",
339            "idle",
340            "in_use",
341            "waiting",
342            "created",
343            "create_failures",
344            "acquired",
345            "released",
346            "timeouts",
347            "closed",
348            "leaks",
349            "closed_by_reason",
350            "avg_acquire_wait",
351            "p50_acquire_wait",
352            "p95_acquire_wait",
353            "p99_acquire_wait",
354            "avg_create_latency",
355            "p50_create_latency",
356            "p95_create_latency",
357            "p99_create_latency",
358            "breaker_state",
359        ] {
360            assert!(d.contains_key(key), "missing key {key}");
361        }
362        assert_eq!(d["breaker_state"], StatValue::Str("closed".into()));
363        assert_eq!(d["name"], StatValue::Str("p".into()));
364        assert_eq!(d["max_size"], StatValue::Int(10));
365        // Null breaker when absent.
366        let d2 = c.snapshot(gauges()).to_dictionary();
367        assert_eq!(d2["breaker_state"], StatValue::Null);
368    }
369
370    #[test]
371    fn window_is_bounded() {
372        let c = StatsCollector::<u32>::new();
373        let pc = PooledConnection::new(1u32, 0.0, None);
374        for i in 0..(ACQUIRE_WAIT_WINDOW + 50) {
375            c.on_acquired(&pc, i as f64);
376        }
377        let inner = c.inner.lock().unwrap();
378        assert_eq!(inner.waits.len(), ACQUIRE_WAIT_WINDOW);
379        // oldest 50 dropped -> first sample is 50.0
380        assert_eq!(inner.waits[0], 50.0);
381    }
382}
connpool/observability/stats.rs

connpool/observability/
stats.rs
