spate_core/metrics/
queue.rs

1//! Queue-edge handles (`spate_queue_*`).
2
3use super::MetricsError;
4use super::labels::{ComponentLabels, OwnedGauge};
5use super::names;
6use super::ownership::{SeriesClaim, series_key};
7use metrics::{Counter, SharedString};
8
9/// Queue-edge handles (`spate_queue_*`).
10#[derive(Debug)]
11pub struct QueueMetrics {
12    depth: OwnedGauge,
13    full_events: Counter,
14    _claim: Option<SeriesClaim>,
15}
16
17impl QueueMetrics {
18    /// Resolve handles for one queue edge (e.g. `chain->sink/shard-3`) and
19    /// publish its configured capacity.
20    ///
21    /// Claims this edge's series (the labels plus the queue name); a second
22    /// live handle set for the same edge logs and becomes a shadow, counting
23    /// but publishing neither depth nor capacity (see "Series ownership" in
24    /// `docs/METRICS.md`).
25    pub fn new(labels: &ComponentLabels, queue: &str, capacity: usize) -> Self {
26        let claim = SeriesClaim::claim_or_shadow(Self::key(labels, queue));
27        Self::build(labels, queue, capacity, claim)
28    }
29
30    /// Resolve handles for one queue edge, failing when another live handle
31    /// set already owns it. The pipeline builder's path.
32    ///
33    /// # Errors
34    ///
35    /// [`MetricsError::DuplicateSeries`] on a collision.
36    pub fn try_new(
37        labels: &ComponentLabels,
38        queue: &str,
39        capacity: usize,
40    ) -> Result<Self, MetricsError> {
41        let claim = SeriesClaim::try_claim(Self::key(labels, queue))?;
42        Ok(Self::build(labels, queue, capacity, Some(claim)))
43    }
44
45    fn key(labels: &ComponentLabels, queue: &str) -> String {
46        series_key("queue", labels, &format!("queue={queue}"))
47    }
48
49    fn build(
50        labels: &ComponentLabels,
51        queue: &str,
52        capacity: usize,
53        claim: Option<SeriesClaim>,
54    ) -> Self {
55        let owned = claim.is_some();
56        let queue: SharedString = queue.to_owned().into();
57        OwnedGauge::new(
58            labels.gauge1(names::QUEUE_CAPACITY, names::L_QUEUE, queue.clone()),
59            owned,
60        )
61        .set(capacity as f64);
62        QueueMetrics {
63            depth: OwnedGauge::new(
64                labels.gauge1(names::QUEUE_DEPTH, names::L_QUEUE, queue.clone()),
65                owned,
66            ),
67            full_events: labels.counter1(names::QUEUE_FULL_EVENTS_TOTAL, names::L_QUEUE, queue),
68            _claim: claim,
69        }
70    }
71
72    /// Set the current queue depth.
73    #[inline]
74    pub fn set_depth(&self, depth: usize) {
75        self.depth.set(depth as f64);
76    }
77
78    /// Count `try_send` rejections.
79    #[inline]
80    pub fn full_events(&self, n: u64) {
81        self.full_events.increment(n);
82    }
83}
spate_core/metrics/queue.rs

spate_core/metrics/
queue.rs
