dynomite/stats/mod.rs
1//! Pool, server, and peer metrics with histograms and a JSON snapshot.
2//!
3//! The stats subsystem is split into small modules:
4//!
5//! * [`Histogram`] - Cassandra-style estimated histogram.
6//! * [`PoolField`] / [`ServerField`] - typed metric handles.
7//! * [`Snapshot`] - aggregate value rendered to JSON.
8//! * [`StatsServer`] - REST endpoint serving the latest snapshot.
9//!
10//! [`Stats`] glues the pieces together: a writer accumulates counters,
11//! gauges, and histogram observations; a periodic aggregator publishes
12//! a fresh [`Snapshot`] that the REST endpoint serves.
13
14mod codec;
15mod failure;
16mod histogram;
17mod numeric;
18mod prometheus;
19mod rest;
20mod snapshot;
21
22use std::sync::Arc;
23use std::time::{SystemTime, UNIX_EPOCH};
24
25use parking_lot::Mutex;
26use tokio::time::{Duration, Instant};
27use tokio_util::sync::CancellationToken;
28
29pub use crate::stats::codec::{
30 MetricSpec, PoolField, ServerField, StatsMetricType, POOL_CODEC, SERVER_CODEC,
31};
32pub use crate::stats::failure::{
33 FailureMetrics, FailureSnapshot, NoTargetsEntry, PeerEntry, PeerStateEntry, PhiEntry,
34 TimeoutEntry, TransitionEntry,
35};
36pub use crate::stats::histogram::{Histogram, BUCKET_COUNT};
37pub use crate::stats::prometheus::render_prometheus;
38pub use crate::stats::rest::{
39 ClusterInfoProvider, RingProvider, StatsServer, MAX_HEADERS, MAX_REQUEST_BYTES,
40};
41pub use crate::stats::snapshot::{
42 describe_stats, HistogramSummary, PeerStats, PoolStats, ServerStats, ServiceInfo, Snapshot,
43};
44
45/// Live, mutable counters and histograms for a single engine instance.
46///
47/// `Stats` is the writer side; readers consume frozen [`Snapshot`]
48/// values produced by [`Stats::snapshot`].
49///
50/// # Examples
51///
52/// ```
53/// use dynomite::stats::{PoolStats, ServerStats, ServiceInfo, Stats};
54/// let stats = Stats::new(
55/// ServiceInfo::default(),
56/// PoolStats::new("dyn_o_mite"),
57/// ServerStats::new("redis"),
58/// );
59/// assert_eq!(stats.snapshot().pool.name, "dyn_o_mite");
60/// ```
61#[derive(Debug)]
62pub struct Stats {
63 inner: Arc<Mutex<StatsInner>>,
64 failure: Arc<FailureMetrics>,
65 started: Instant,
66}
67
68#[derive(Debug)]
69struct StatsInner {
70 info: ServiceInfo,
71 pool: PoolStats,
72 server: ServerStats,
73 latency: Histogram,
74 payload_size: Histogram,
75 cross_region_latency: Histogram,
76 cross_zone_latency: Histogram,
77 server_latency: Histogram,
78 cross_region_queue_wait: Histogram,
79 cross_zone_queue_wait: Histogram,
80 server_queue_wait: Histogram,
81 client_out_queue: Histogram,
82 server_in_queue: Histogram,
83 server_out_queue: Histogram,
84 dnode_client_out_queue: Histogram,
85 peer_in_queue: Histogram,
86 peer_out_queue: Histogram,
87 remote_peer_in_queue: Histogram,
88 remote_peer_out_queue: Histogram,
89 alloc_msgs: i64,
90 free_msgs: i64,
91 alloc_mbufs: i64,
92 free_mbufs: i64,
93 dyn_memory: i64,
94}
95
96impl StatsInner {
97 fn new(info: ServiceInfo, pool: PoolStats, server: ServerStats) -> Self {
98 Self {
99 info,
100 pool,
101 server,
102 latency: Histogram::new(),
103 payload_size: Histogram::new(),
104 cross_region_latency: Histogram::new(),
105 cross_zone_latency: Histogram::new(),
106 server_latency: Histogram::new(),
107 cross_region_queue_wait: Histogram::new(),
108 cross_zone_queue_wait: Histogram::new(),
109 server_queue_wait: Histogram::new(),
110 client_out_queue: Histogram::new(),
111 server_in_queue: Histogram::new(),
112 server_out_queue: Histogram::new(),
113 dnode_client_out_queue: Histogram::new(),
114 peer_in_queue: Histogram::new(),
115 peer_out_queue: Histogram::new(),
116 remote_peer_in_queue: Histogram::new(),
117 remote_peer_out_queue: Histogram::new(),
118 alloc_msgs: 0,
119 free_msgs: 0,
120 alloc_mbufs: 0,
121 free_mbufs: 0,
122 dyn_memory: 0,
123 }
124 }
125}
126
127/// Channels used to mutate histogram observations.
128///
129/// # Examples
130///
131/// ```
132/// use dynomite::stats::Latency;
133/// assert_ne!(Latency::Request, Latency::Server);
134/// assert_eq!(Latency::Request, Latency::Request);
135/// // The variant set is small and copy-able.
136/// let copied = Latency::CrossRegion;
137/// assert_eq!(copied, Latency::CrossRegion);
138/// ```
139#[derive(Copy, Clone, Eq, PartialEq, Debug)]
140pub enum Latency {
141 /// Top-level request latency.
142 Request,
143 /// Cross-region peer round-trip time.
144 CrossRegion,
145 /// Cross-zone peer latency.
146 CrossZone,
147 /// Backing-server response latency.
148 Server,
149}
150
151/// Channels used for queue-wait-time observations.
152///
153/// # Examples
154///
155/// ```
156/// use dynomite::stats::QueueWait;
157/// assert_ne!(QueueWait::CrossRegion, QueueWait::CrossZone);
158/// assert_ne!(QueueWait::CrossZone, QueueWait::Server);
159/// // Variants implement Copy, so a binding survives a move.
160/// let original = QueueWait::Server;
161/// let copy = original;
162/// assert_eq!(original, copy);
163/// ```
164#[derive(Copy, Clone, Eq, PartialEq, Debug)]
165pub enum QueueWait {
166 /// Cross-region queue wait time.
167 CrossRegion,
168 /// Cross-zone queue wait time.
169 CrossZone,
170 /// Backing-server queue wait time.
171 Server,
172}
173
174/// Channels used for queue-length observations (observed at sample
175/// time, not events).
176///
177/// # Examples
178///
179/// ```
180/// use dynomite::stats::QueueGauge;
181/// // Each variant is distinct so the dispatch in `record_queue_len`
182/// // routes to a unique histogram.
183/// let all = [
184/// QueueGauge::ClientOut,
185/// QueueGauge::ServerIn,
186/// QueueGauge::ServerOut,
187/// QueueGauge::DnodeClientOut,
188/// QueueGauge::PeerIn,
189/// QueueGauge::PeerOut,
190/// QueueGauge::RemotePeerIn,
191/// QueueGauge::RemotePeerOut,
192/// ];
193/// for (i, lhs) in all.iter().enumerate() {
194/// for rhs in &all[i + 1..] {
195/// assert_ne!(lhs, rhs);
196/// }
197/// }
198/// ```
199#[derive(Copy, Clone, Eq, PartialEq, Debug)]
200pub enum QueueGauge {
201 /// Client out-queue length.
202 ClientOut,
203 /// Server in-queue length.
204 ServerIn,
205 /// Server out-queue length.
206 ServerOut,
207 /// Dnode client out-queue length.
208 DnodeClientOut,
209 /// Local-DC peer in-queue length.
210 PeerIn,
211 /// Local-DC peer out-queue length.
212 PeerOut,
213 /// Remote-DC peer in-queue length.
214 RemotePeerIn,
215 /// Remote-DC peer out-queue length.
216 RemotePeerOut,
217}
218
219impl Stats {
220 /// Construct a new `Stats` with empty counters and histograms.
221 ///
222 /// # Examples
223 ///
224 /// ```
225 /// use dynomite::stats::{PoolStats, ServerStats, ServiceInfo, Stats};
226 ///
227 /// let info = ServiceInfo {
228 /// source: "node-a".into(),
229 /// version: "0.0.1".into(),
230 /// rack: "r1".into(),
231 /// dc: "dc1".into(),
232 /// };
233 /// let stats = Stats::new(
234 /// info,
235 /// PoolStats::new("dyn_o_mite"),
236 /// ServerStats::new("redis_local"),
237 /// );
238 /// let snap = stats.snapshot();
239 /// assert_eq!(snap.pool.name, "dyn_o_mite");
240 /// ```
241 pub fn new(info: ServiceInfo, pool: PoolStats, server: ServerStats) -> Self {
242 Self {
243 inner: Arc::new(Mutex::new(StatsInner::new(info, pool, server))),
244 failure: Arc::new(FailureMetrics::new()),
245 started: Instant::now(),
246 }
247 }
248
249 /// Borrow the failure-cause metrics handle.
250 ///
251 /// The dispatcher and the gossip handler clone this `Arc`
252 /// so they can record per-cause errors and per-peer state
253 /// transitions. The handle is created at construction time
254 /// and lives for the lifetime of the [`Stats`] value.
255 ///
256 /// # Examples
257 ///
258 /// ```
259 /// use dynomite::stats::{PoolStats, ServerStats, ServiceInfo, Stats};
260 /// let s = Stats::new(
261 /// ServiceInfo::default(),
262 /// PoolStats::new("p"),
263 /// ServerStats::new("s"),
264 /// );
265 /// let m = s.failure_metrics();
266 /// assert!(m.snapshot().is_empty());
267 /// ```
268 #[must_use]
269 pub fn failure_metrics(&self) -> Arc<FailureMetrics> {
270 self.failure.clone()
271 }
272
273 /// Record a latency observation in the matching histogram.
274 ///
275 /// # Examples
276 ///
277 /// ```
278 /// use dynomite::stats::{Latency, PoolStats, ServerStats, ServiceInfo, Stats};
279 /// let stats = Stats::new(
280 /// ServiceInfo::default(),
281 /// PoolStats::new("p"),
282 /// ServerStats::new("s"),
283 /// );
284 /// stats.record_latency(Latency::Request, 100);
285 /// ```
286 pub fn record_latency(&self, channel: Latency, value: u64) {
287 let mut inner = self.inner.lock();
288 match channel {
289 Latency::Request => inner.latency.record(value),
290 Latency::CrossRegion => inner.cross_region_latency.record(value),
291 Latency::CrossZone => inner.cross_zone_latency.record(value),
292 Latency::Server => inner.server_latency.record(value),
293 }
294 }
295
296 /// Record a payload-size observation.
297 ///
298 /// # Examples
299 ///
300 /// ```
301 /// use dynomite::stats::{PoolStats, ServerStats, ServiceInfo, Stats};
302 /// let stats = Stats::new(
303 /// ServiceInfo::default(),
304 /// PoolStats::new("p"),
305 /// ServerStats::new("s"),
306 /// );
307 /// stats.record_payload_size(2048);
308 /// ```
309 pub fn record_payload_size(&self, value: u64) {
310 self.inner.lock().payload_size.record(value);
311 }
312
313 /// Record a queue wait time observation.
314 ///
315 /// # Examples
316 ///
317 /// ```
318 /// use dynomite::stats::{PoolStats, QueueWait, ServerStats, ServiceInfo, Stats};
319 /// let stats = Stats::new(
320 /// ServiceInfo::default(),
321 /// PoolStats::new("p"),
322 /// ServerStats::new("s"),
323 /// );
324 /// stats.record_queue_wait(QueueWait::Server, 12);
325 /// ```
326 pub fn record_queue_wait(&self, channel: QueueWait, value: u64) {
327 let mut inner = self.inner.lock();
328 match channel {
329 QueueWait::CrossRegion => inner.cross_region_queue_wait.record(value),
330 QueueWait::CrossZone => inner.cross_zone_queue_wait.record(value),
331 QueueWait::Server => inner.server_queue_wait.record(value),
332 }
333 }
334
335 /// Record a queue-length sample.
336 ///
337 /// # Examples
338 ///
339 /// ```
340 /// use dynomite::stats::{PoolStats, QueueGauge, ServerStats, ServiceInfo, Stats};
341 /// let stats = Stats::new(
342 /// ServiceInfo::default(),
343 /// PoolStats::new("p"),
344 /// ServerStats::new("s"),
345 /// );
346 /// stats.record_queue_len(QueueGauge::ClientOut, 4);
347 /// ```
348 pub fn record_queue_len(&self, channel: QueueGauge, value: u64) {
349 let mut inner = self.inner.lock();
350 match channel {
351 QueueGauge::ClientOut => inner.client_out_queue.record(value),
352 QueueGauge::ServerIn => inner.server_in_queue.record(value),
353 QueueGauge::ServerOut => inner.server_out_queue.record(value),
354 QueueGauge::DnodeClientOut => inner.dnode_client_out_queue.record(value),
355 QueueGauge::PeerIn => inner.peer_in_queue.record(value),
356 QueueGauge::PeerOut => inner.peer_out_queue.record(value),
357 QueueGauge::RemotePeerIn => inner.remote_peer_in_queue.record(value),
358 QueueGauge::RemotePeerOut => inner.remote_peer_out_queue.record(value),
359 }
360 }
361
362 /// Increment a pool counter or gauge by one.
363 ///
364 /// # Examples
365 ///
366 /// ```
367 /// use dynomite::stats::{PoolField, PoolStats, ServerStats, ServiceInfo, Stats};
368 /// let stats = Stats::new(
369 /// ServiceInfo::default(),
370 /// PoolStats::new("p"),
371 /// ServerStats::new("s"),
372 /// );
373 /// stats.pool_incr(PoolField::ClientEof);
374 /// assert_eq!(stats.pool_get(PoolField::ClientEof), 1);
375 /// ```
376 pub fn pool_incr(&self, field: PoolField) {
377 self.pool_incr_by(field, 1);
378 }
379
380 /// Decrement a pool gauge by one.
381 ///
382 /// # Examples
383 ///
384 /// ```
385 /// use dynomite::stats::{PoolField, PoolStats, ServerStats, ServiceInfo, Stats};
386 /// let stats = Stats::new(
387 /// ServiceInfo::default(),
388 /// PoolStats::new("p"),
389 /// ServerStats::new("s"),
390 /// );
391 /// stats.pool_set(PoolField::ClientConnections, 5);
392 /// stats.pool_decr(PoolField::ClientConnections);
393 /// assert_eq!(stats.pool_get(PoolField::ClientConnections), 4);
394 /// ```
395 pub fn pool_decr(&self, field: PoolField) {
396 self.pool_incr_by(field, -1);
397 }
398
399 /// Add `delta` to a pool counter or gauge.
400 ///
401 /// Wraps on overflow. Counters are 64-bit signed and never reach
402 /// the wrap
403 /// boundary under realistic workloads.
404 pub fn pool_incr_by(&self, field: PoolField, delta: i64) {
405 let mut inner = self.inner.lock();
406 let slot = &mut inner.pool.metrics[field.index()];
407 *slot = slot.wrapping_add(delta);
408 }
409
410 /// Set a pool gauge or timestamp to an absolute value.
411 ///
412 /// # Examples
413 ///
414 /// ```
415 /// use dynomite::stats::{PoolField, PoolStats, ServerStats, ServiceInfo, Stats};
416 /// let stats = Stats::new(
417 /// ServiceInfo::default(),
418 /// PoolStats::new("p"),
419 /// ServerStats::new("s"),
420 /// );
421 /// stats.pool_set(PoolField::PeerEjectedAt, 1_700_000_000);
422 /// assert_eq!(stats.pool_get(PoolField::PeerEjectedAt), 1_700_000_000);
423 /// ```
424 pub fn pool_set(&self, field: PoolField, value: i64) {
425 self.inner.lock().pool.metrics[field.index()] = value;
426 }
427
428 /// Read the current value of a pool metric.
429 ///
430 /// # Examples
431 ///
432 /// ```
433 /// use dynomite::stats::{PoolField, PoolStats, ServerStats, ServiceInfo, Stats};
434 /// let stats = Stats::new(
435 /// ServiceInfo::default(),
436 /// PoolStats::new("p"),
437 /// ServerStats::new("s"),
438 /// );
439 /// assert_eq!(stats.pool_get(PoolField::ClientEof), 0);
440 /// ```
441 pub fn pool_get(&self, field: PoolField) -> i64 {
442 self.inner.lock().pool.metrics[field.index()]
443 }
444
445 /// Increment a server counter or gauge by one.
446 ///
447 /// # Examples
448 ///
449 /// ```
450 /// use dynomite::stats::{PoolStats, ServerField, ServerStats, ServiceInfo, Stats};
451 /// let stats = Stats::new(
452 /// ServiceInfo::default(),
453 /// PoolStats::new("p"),
454 /// ServerStats::new("s"),
455 /// );
456 /// stats.server_incr(ServerField::ReadRequests);
457 /// assert_eq!(stats.server_get(ServerField::ReadRequests), 1);
458 /// ```
459 pub fn server_incr(&self, field: ServerField) {
460 self.server_incr_by(field, 1);
461 }
462
463 /// Decrement a server gauge by one.
464 ///
465 /// # Examples
466 ///
467 /// ```
468 /// use dynomite::stats::{PoolStats, ServerField, ServerStats, ServiceInfo, Stats};
469 /// let stats = Stats::new(
470 /// ServiceInfo::default(),
471 /// PoolStats::new("p"),
472 /// ServerStats::new("s"),
473 /// );
474 /// stats.server_set(ServerField::InQueue, 3);
475 /// stats.server_decr(ServerField::InQueue);
476 /// assert_eq!(stats.server_get(ServerField::InQueue), 2);
477 /// ```
478 pub fn server_decr(&self, field: ServerField) {
479 self.server_incr_by(field, -1);
480 }
481
482 /// Add `delta` to a server counter or gauge.
483 ///
484 /// Wraps on overflow. Counters are 64-bit signed and never reach
485 /// the wrap
486 /// boundary under realistic workloads.
487 pub fn server_incr_by(&self, field: ServerField, delta: i64) {
488 let mut inner = self.inner.lock();
489 let slot = &mut inner.server.metrics[field.index()];
490 *slot = slot.wrapping_add(delta);
491 }
492
493 /// Set a server gauge or timestamp to an absolute value.
494 ///
495 /// # Examples
496 ///
497 /// ```
498 /// use dynomite::stats::{PoolStats, ServerField, ServerStats, ServiceInfo, Stats};
499 /// let stats = Stats::new(
500 /// ServiceInfo::default(),
501 /// PoolStats::new("p"),
502 /// ServerStats::new("s"),
503 /// );
504 /// stats.server_set(ServerField::ServerEjectedAt, 1_700_000_000);
505 /// assert_eq!(stats.server_get(ServerField::ServerEjectedAt), 1_700_000_000);
506 /// ```
507 pub fn server_set(&self, field: ServerField, value: i64) {
508 self.inner.lock().server.metrics[field.index()] = value;
509 }
510
511 /// Read the current value of a server metric.
512 ///
513 /// # Examples
514 ///
515 /// ```
516 /// use dynomite::stats::{PoolStats, ServerField, ServerStats, ServiceInfo, Stats};
517 /// let stats = Stats::new(
518 /// ServiceInfo::default(),
519 /// PoolStats::new("p"),
520 /// ServerStats::new("s"),
521 /// );
522 /// assert_eq!(stats.server_get(ServerField::ReadRequests), 0);
523 /// ```
524 pub fn server_get(&self, field: ServerField) -> i64 {
525 self.inner.lock().server.metrics[field.index()]
526 }
527
528 /// Set the resource usage gauges sampled
529 /// once per aggregation cycle.
530 ///
531 /// # Examples
532 ///
533 /// ```
534 /// use dynomite::stats::{PoolStats, ServerStats, ServiceInfo, Stats};
535 /// let stats = Stats::new(
536 /// ServiceInfo::default(),
537 /// PoolStats::new("p"),
538 /// ServerStats::new("s"),
539 /// );
540 /// stats.set_resource_usage(0, 0, 0, 0, 0);
541 /// assert_eq!(stats.snapshot().alloc_msgs, 0);
542 /// ```
543 pub fn set_resource_usage(
544 &self,
545 alloc_msgs: i64,
546 free_msgs: i64,
547 alloc_mbufs: i64,
548 free_mbufs: i64,
549 dyn_memory: i64,
550 ) {
551 let mut inner = self.inner.lock();
552 inner.alloc_msgs = alloc_msgs;
553 inner.free_msgs = free_msgs;
554 inner.alloc_mbufs = alloc_mbufs;
555 inner.free_mbufs = free_mbufs;
556 inner.dyn_memory = dyn_memory;
557 }
558
559 /// Build an immutable snapshot of every counter, gauge, and
560 /// histogram quantile at the current instant.
561 ///
562 /// # Examples
563 ///
564 /// ```
565 /// use dynomite::stats::{PoolStats, ServerStats, ServiceInfo, Stats};
566 /// let stats = Stats::new(
567 /// ServiceInfo::default(),
568 /// PoolStats::new("p"),
569 /// ServerStats::new("s"),
570 /// );
571 /// let snap = stats.snapshot();
572 /// assert_eq!(snap.pool.name, "p");
573 /// ```
574 pub fn snapshot(&self) -> Snapshot {
575 let inner = self.inner.lock();
576 let elapsed = self.started.elapsed();
577 let timestamp = SystemTime::now()
578 .duration_since(UNIX_EPOCH)
579 .map_or(0, |d| d.as_secs());
580 Snapshot {
581 info: inner.info.clone(),
582 uptime: i64::try_from(elapsed.as_secs()).unwrap_or(i64::MAX),
583 timestamp: i64::try_from(timestamp).unwrap_or(i64::MAX),
584 latency: HistogramSummary::from_histogram(&inner.latency),
585 payload_size: HistogramSummary::from_histogram(&inner.payload_size),
586 cross_region_latency: HistogramSummary::from_histogram(&inner.cross_region_latency),
587 cross_zone_latency: HistogramSummary::from_histogram(&inner.cross_zone_latency),
588 server_latency: HistogramSummary::from_histogram(&inner.server_latency),
589 cross_region_queue_wait: HistogramSummary::from_histogram(
590 &inner.cross_region_queue_wait,
591 ),
592 cross_zone_queue_wait: HistogramSummary::from_histogram(&inner.cross_zone_queue_wait),
593 server_queue_wait: HistogramSummary::from_histogram(&inner.server_queue_wait),
594 client_out_queue_p99: queue_p99(&inner.client_out_queue),
595 server_in_queue_p99: queue_p99(&inner.server_in_queue),
596 server_out_queue_p99: queue_p99(&inner.server_out_queue),
597 dnode_client_out_queue_p99: queue_p99(&inner.dnode_client_out_queue),
598 peer_in_queue_p99: queue_p99(&inner.peer_in_queue),
599 peer_out_queue_p99: queue_p99(&inner.peer_out_queue),
600 remote_peer_in_queue_p99: queue_p99(&inner.remote_peer_in_queue),
601 remote_peer_out_queue_p99: queue_p99(&inner.remote_peer_out_queue),
602 alloc_msgs: inner.alloc_msgs,
603 free_msgs: inner.free_msgs,
604 alloc_mbufs: inner.alloc_mbufs,
605 free_mbufs: inner.free_mbufs,
606 dyn_memory: inner.dyn_memory,
607 pool: inner.pool.clone(),
608 server: inner.server.clone(),
609 failure: self.failure.snapshot(),
610 }
611 }
612
613 /// Reset every histogram. This runs every
614 /// five minutes from inside the aggregation loop.
615 ///
616 /// # Examples
617 ///
618 /// ```
619 /// use dynomite::stats::{Latency, PoolStats, ServerStats, ServiceInfo, Stats};
620 /// let stats = Stats::new(
621 /// ServiceInfo::default(),
622 /// PoolStats::new("p"),
623 /// ServerStats::new("s"),
624 /// );
625 /// stats.record_latency(Latency::Request, 50);
626 /// stats.reset_histograms();
627 /// assert_eq!(stats.snapshot().latency.max, 0);
628 /// ```
629 pub fn reset_histograms(&self) {
630 let mut inner = self.inner.lock();
631 inner.latency.reset();
632 inner.payload_size.reset();
633 inner.cross_region_latency.reset();
634 inner.cross_zone_latency.reset();
635 inner.server_latency.reset();
636 inner.cross_region_queue_wait.reset();
637 inner.cross_zone_queue_wait.reset();
638 inner.server_queue_wait.reset();
639 inner.client_out_queue.reset();
640 inner.server_in_queue.reset();
641 inner.server_out_queue.reset();
642 inner.dnode_client_out_queue.reset();
643 inner.peer_in_queue.reset();
644 inner.peer_out_queue.reset();
645 inner.remote_peer_in_queue.reset();
646 inner.remote_peer_out_queue.reset();
647 }
648}
649
650/// Returns the queue p99 from `h`, or `0` when the histogram is
651/// overflowing. Percentile publishing is suppressed in the overflow
652/// path so overflow values do not leak into the JSON output as
653/// `u64::MAX`.
654fn queue_p99(h: &Histogram) -> u64 {
655 if h.is_overflowing() {
656 0
657 } else {
658 h.percentile(0.99)
659 }
660}
661
662/// Async aggregator handle: snapshots at a fixed interval into a
663/// shared cell that the REST server reads from.
664///
665/// # Examples
666///
667/// ```no_run
668/// use std::sync::Arc;
669/// use std::time::Duration;
670/// use dynomite::stats::{Aggregator, PoolStats, ServerStats, ServiceInfo, Snapshot, Stats};
671/// use parking_lot::Mutex;
672/// use tokio_util::sync::CancellationToken;
673///
674/// # async fn _example() {
675/// let stats = Arc::new(Stats::new(
676/// ServiceInfo::default(),
677/// PoolStats::new("dyn_o_mite"),
678/// ServerStats::new("redis"),
679/// ));
680/// let sink = Arc::new(Mutex::new(Snapshot::default()));
681/// let token = CancellationToken::new();
682/// let agg = Aggregator::new(stats, sink, Duration::from_secs(1), Duration::from_secs(300));
683/// let _ = tokio::spawn({ let token = token.clone(); async move { agg.run(token).await } });
684/// token.cancel();
685/// # }
686/// ```
687pub struct Aggregator {
688 stats: Arc<Stats>,
689 sink: Arc<Mutex<Snapshot>>,
690 interval: Duration,
691 histogram_reset: Duration,
692}
693
694impl Aggregator {
695 /// Create a new aggregator. The aggregation loop reads from
696 /// `stats` and publishes to `sink` once every `interval`.
697 /// Histograms are reset every `histogram_reset` elapsed time;
698 /// the default cadence is five minutes.
699 ///
700 /// # Examples
701 ///
702 /// ```
703 /// use std::sync::Arc;
704 /// use std::time::Duration;
705 /// use dynomite::stats::{Aggregator, PoolStats, ServerStats, ServiceInfo, Snapshot, Stats};
706 /// use parking_lot::Mutex;
707 ///
708 /// let stats = Arc::new(Stats::new(
709 /// ServiceInfo::default(),
710 /// PoolStats::new("dyn_o_mite"),
711 /// ServerStats::new("redis"),
712 /// ));
713 /// let sink = Arc::new(Mutex::new(Snapshot::default()));
714 /// let _agg = Aggregator::new(stats, sink, Duration::from_secs(1), Duration::from_secs(300));
715 /// ```
716 pub fn new(
717 stats: Arc<Stats>,
718 sink: Arc<Mutex<Snapshot>>,
719 interval: Duration,
720 histogram_reset: Duration,
721 ) -> Self {
722 Self {
723 stats,
724 sink,
725 interval,
726 histogram_reset,
727 }
728 }
729
730 /// Run the aggregation loop until `cancel` is triggered. The future
731 /// returns `()` after observing cancellation; callers that want a
732 /// clean shutdown should clone the token and call
733 /// [`CancellationToken::cancel`] on it.
734 ///
735 /// # Examples
736 ///
737 /// ```no_run
738 /// use std::sync::Arc;
739 /// use std::time::Duration;
740 /// use dynomite::stats::{Aggregator, PoolStats, ServerStats, ServiceInfo, Snapshot, Stats};
741 /// use parking_lot::Mutex;
742 /// use tokio_util::sync::CancellationToken;
743 ///
744 /// # async fn _example() {
745 /// let stats = Arc::new(Stats::new(
746 /// ServiceInfo::default(),
747 /// PoolStats::new("dyn_o_mite"),
748 /// ServerStats::new("redis"),
749 /// ));
750 /// let sink = Arc::new(Mutex::new(Snapshot::default()));
751 /// let token = CancellationToken::new();
752 /// let agg = Aggregator::new(stats, sink, Duration::from_secs(1), Duration::from_secs(300));
753 /// let cancel = token.clone();
754 /// let handle = tokio::spawn(async move { agg.run(cancel).await });
755 /// token.cancel();
756 /// let _ = handle.await;
757 /// # }
758 /// ```
759 pub async fn run(self, cancel: CancellationToken) {
760 let mut ticker = tokio::time::interval(self.interval);
761 ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
762 let mut last_reset = Instant::now();
763 loop {
764 tokio::select! {
765 biased;
766 () = cancel.cancelled() => return,
767 _ = ticker.tick() => {}
768 }
769 let snap = self.stats.snapshot();
770 *self.sink.lock() = snap;
771 if last_reset.elapsed() >= self.histogram_reset {
772 self.stats.reset_histograms();
773 last_reset = Instant::now();
774 }
775 }
776 }
777}
778
779#[cfg(test)]
780mod tests {
781 use super::*;
782
783 fn fresh() -> Stats {
784 Stats::new(
785 ServiceInfo {
786 source: "node".into(),
787 version: "0.0.1".into(),
788 rack: "r".into(),
789 dc: "d".into(),
790 },
791 PoolStats::new("dyn_o_mite"),
792 ServerStats::new("redis"),
793 )
794 }
795
796 #[test]
797 fn counter_incr_and_get() {
798 let s = fresh();
799 s.pool_incr(PoolField::ClientEof);
800 s.pool_incr(PoolField::ClientEof);
801 assert_eq!(s.pool_get(PoolField::ClientEof), 2);
802 }
803
804 #[test]
805 fn gauge_set_and_decrement() {
806 let s = fresh();
807 s.pool_set(PoolField::ClientConnections, 5);
808 s.pool_decr(PoolField::ClientConnections);
809 assert_eq!(s.pool_get(PoolField::ClientConnections), 4);
810 }
811
812 #[test]
813 fn server_metric_round_trip() {
814 let s = fresh();
815 s.server_incr_by(ServerField::ReadRequests, 42);
816 s.server_set(ServerField::ServerEjectedAt, 1_700_000_000);
817 assert_eq!(s.server_get(ServerField::ReadRequests), 42);
818 assert_eq!(s.server_get(ServerField::ServerEjectedAt), 1_700_000_000);
819 }
820
821 #[test]
822 fn snapshot_reflects_writes() {
823 let s = fresh();
824 s.pool_incr(PoolField::StatsCount);
825 s.record_latency(Latency::Request, 100);
826 s.record_payload_size(2048);
827 let snap = s.snapshot();
828 assert_eq!(snap.pool.metrics[PoolField::StatsCount.index()], 1);
829 assert_eq!(snap.latency.max, 100);
830 assert!(snap.payload_size.max >= 2048);
831 }
832
833 #[test]
834 fn metric_indexes_have_canonical_order() {
835 for (i, variant) in PoolField::ALL.iter().enumerate() {
836 assert_eq!(variant.index(), i);
837 }
838 }
839}