brokapi_runtime/
metrics.rs

1//! Runtime counters (structure only, no exporter).
2
3use std::time::Duration;
4
5/// A plain set of counters about the connector runtime.
6///
7/// It is a snapshot-style struct for the owner to fill and read. The runtime does not update it by
8/// itself and there is no exporter. All counts start at zero and only grow.
9///
10/// # Examples
11///
12/// ```
13/// use brokapi_runtime::RuntimeMetrics;
14///
15/// let mut m = RuntimeMetrics::new();
16/// m.record_rate_limit("orders");
17/// m.record_rate_limit("orders");
18/// m.record_rate_limit("global");
19/// assert_eq!(m.rate_limit_hits, 3);
20/// assert_eq!(m.rate_limit_by_bucket, vec![("orders".to_string(), 2), ("global".to_string(), 1)]);
21/// ```
22#[derive(Debug, Clone, Default, PartialEq, Eq)]
23pub struct RuntimeMetrics {
24    /// Number of resyncs recorded with [`record_resync`](Self::record_resync).
25    pub resync_count: u64,
26    /// Number of sequence gaps recorded with [`record_seq_gap`](Self::record_seq_gap).
27    pub seq_gaps: u64,
28    /// Number of orders that entered the `Unknown` state. The owner counts them, no method here does.
29    pub unknown_orders: u64,
30    /// Milliseconds spent resolving `Unknown` orders. The owner decides how to sum it, the runtime
31    /// does not fill it.
32    pub unknown_resolved_ms: u64,
33    /// Number of calls refused by the rate limiter.
34    pub rate_limit_hits: u64,
35    /// The same refusals split by bucket name, in order of first appearance.
36    pub rate_limit_by_bucket: Vec<(String, u64)>,
37    /// Events replaced by newer ones in `Conflate` buffers, added with [`record_conflated`](Self::record_conflated).
38    pub buffer_conflated: u64,
39    /// Events thrown away by `DropNewest` buffers, added with [`record_dropped`](Self::record_dropped).
40    pub buffer_dropped: u64,
41    /// Overflows of `Lossless` buffers, counted with [`record_overflow`](Self::record_overflow).
42    pub buffer_overflow: u64,
43    /// Median time from parsing an event to handing it to the consumer. Not filled by the runtime.
44    pub egress_latency_p50_ms: Duration,
45    /// 99th percentile of the same time. Not filled by the runtime.
46    pub egress_latency_p99_ms: Duration,
47}
48
49impl RuntimeMetrics {
50    /// Creates a zeroed set. Same as [`Default::default`].
51    pub fn new() -> Self {
52        Self::default()
53    }
54
55    /// Adds one to [`resync_count`](Self::resync_count).
56    pub fn record_resync(&mut self) {
57        self.resync_count += 1;
58    }
59
60    /// Adds one to [`seq_gaps`](Self::seq_gaps).
61    pub fn record_seq_gap(&mut self) {
62        self.seq_gaps += 1;
63    }
64
65    /// Counts one rate-limit refusal, in total and for `bucket`.
66    pub fn record_rate_limit(&mut self, bucket: &str) {
67        self.rate_limit_hits += 1;
68        if let Some(entry) = self
69            .rate_limit_by_bucket
70            .iter_mut()
71            .find(|(b, _)| b == bucket)
72        {
73            entry.1 += 1;
74        } else {
75            self.rate_limit_by_bucket.push((bucket.to_string(), 1));
76        }
77    }
78
79    /// Adds `n` to [`buffer_conflated`](Self::buffer_conflated).
80    pub fn record_conflated(&mut self, n: u64) {
81        self.buffer_conflated += n;
82    }
83
84    /// Adds `n` to [`buffer_dropped`](Self::buffer_dropped).
85    pub fn record_dropped(&mut self, n: u64) {
86        self.buffer_dropped += n;
87    }
88
89    /// Adds one to [`buffer_overflow`](Self::buffer_overflow).
90    pub fn record_overflow(&mut self) {
91        self.buffer_overflow += 1;
92    }
93}
brokapi_runtime/metrics.rs

brokapi_runtime/
metrics.rs
