spate_core/metrics/
checkpoint.rs

1//! Checkpointer handles (`spate_checkpoint_*`).
2
3use super::MetricsError;
4use super::labels::{ComponentLabels, OwnedGauge, PartitionGauges};
5use super::names;
6use super::ownership::{SeriesClaim, series_key};
7use crate::record::PartitionId;
8use metrics::{Counter, Histogram};
9use std::collections::HashMap;
10use std::sync::Mutex;
11use std::time::Duration;
12
13/// Checkpointer handles (`spate_checkpoint_*`).
14#[derive(Debug)]
15pub struct CheckpointMetrics {
16    pending_max: OwnedGauge,
17    commits_ok: Counter,
18    commits_err: Counter,
19    commit_duration: Histogram,
20    watermark_age: OwnedGauge,
21    partition_pending: Option<PartitionGauges>,
22    _claim: Option<SeriesClaim>,
23}
24
25impl CheckpointMetrics {
26    /// Resolve all checkpointer handles.
27    ///
28    /// Claims the `spate_checkpoint_*` series for these labels; a second live
29    /// handle set logs and becomes a shadow, counting commits but publishing
30    /// no gauge.
31    pub fn new(labels: &ComponentLabels, per_partition_detail: bool) -> Self {
32        let claim = SeriesClaim::claim_or_shadow(Self::key(labels));
33        Self::build(labels, per_partition_detail, claim)
34    }
35
36    /// Resolve all checkpointer handles, failing when another live handle set
37    /// already owns the series. The pipeline runtime's path.
38    ///
39    /// # Errors
40    ///
41    /// [`MetricsError::DuplicateSeries`] on a collision.
42    pub fn try_new(
43        labels: &ComponentLabels,
44        per_partition_detail: bool,
45    ) -> Result<Self, MetricsError> {
46        let claim = SeriesClaim::try_claim(Self::key(labels))?;
47        Ok(Self::build(labels, per_partition_detail, Some(claim)))
48    }
49
50    fn key(labels: &ComponentLabels) -> String {
51        series_key("checkpoint", labels, "")
52    }
53
54    fn build(
55        labels: &ComponentLabels,
56        per_partition_detail: bool,
57        claim: Option<SeriesClaim>,
58    ) -> Self {
59        let owned = claim.is_some();
60        CheckpointMetrics {
61            pending_max: OwnedGauge::new(labels.gauge(names::CHECKPOINT_PENDING_BATCHES), owned),
62            commits_ok: labels.counter1(names::CHECKPOINT_COMMITS_TOTAL, names::L_OUTCOME, "ok"),
63            commits_err: labels.counter1(
64                names::CHECKPOINT_COMMITS_TOTAL,
65                names::L_OUTCOME,
66                "error",
67            ),
68            commit_duration: labels.histogram(names::CHECKPOINT_COMMIT_DURATION_SECONDS),
69            watermark_age: OwnedGauge::new(
70                labels.gauge(names::CHECKPOINT_WATERMARK_AGE_SECONDS),
71                owned,
72            ),
73            partition_pending: per_partition_detail.then(|| PartitionGauges {
74                name: names::CHECKPOINT_PENDING_BATCHES,
75                labels: labels.clone(),
76                gauges: Mutex::new(HashMap::new()),
77                owned,
78            }),
79            _claim: claim,
80        }
81    }
82
83    /// Set the max pending-batch count across partitions.
84    pub fn set_pending_max(&self, pending: usize) {
85        self.pending_max.set(pending as f64);
86    }
87
88    /// Whether this handle set publishes the per-partition series.
89    ///
90    /// False when `per_partition_detail` was not set, and false for a shadow,
91    /// which publishes no gauge.
92    /// [`set_partition_pending`](Self::set_partition_pending) and
93    /// [`retain_partitions`](Self::retain_partitions) no-op in both cases. A
94    /// caller reads this to skip gathering the counts at all.
95    #[must_use]
96    pub fn publishes_partition_detail(&self) -> bool {
97        self.partition_pending.as_ref().is_some_and(|pg| pg.owned)
98    }
99
100    /// Set one partition's pending count. No-op unless
101    /// `per_partition_detail`.
102    pub fn set_partition_pending(&self, partition: PartitionId, pending: usize) {
103        if let Some(pg) = &self.partition_pending {
104            pg.set(partition, pending as f64);
105        }
106    }
107
108    /// Drop per-partition series for revoked partitions.
109    ///
110    /// Two things differ from
111    /// [`SourceMetrics::retain_partitions`](super::SourceMetrics::retain_partitions),
112    /// and both follow from what this family counts.
113    ///
114    /// This takes the live set on every commit cycle rather than once a new
115    /// assignment is known. Zeroing a partition the runtime is about to be
116    /// handed back costs nothing, because the unlabeled series reads the same
117    /// `0` from the same empty tracker map in the same cycle. The lag family
118    /// has no aggregate to agree with, so it waits instead.
119    ///
120    /// This also zeroes a partition no other member will publish, where a lag
121    /// gauge holds its last value. `0` here is the count of what the
122    /// checkpointer tracks, and it tracks nothing for a partition it has
123    /// dropped, so the reading stays true. A lag of `0` instead claims the
124    /// partition is caught up, which is a claim about the world rather than
125    /// about the handle, and false for a partition still being consumed
126    /// somewhere.
127    pub fn retain_partitions(&self, keep: &[PartitionId]) {
128        if let Some(pg) = &self.partition_pending {
129            pg.retain(keep);
130        }
131    }
132
133    /// Record one source commit call.
134    pub fn commit(&self, ok: bool, d: Duration) {
135        if ok {
136            self.commits_ok.increment(1);
137        } else {
138            self.commits_err.increment(1);
139        }
140        self.commit_duration.record(d.as_secs_f64());
141    }
142
143    /// Set the age of the oldest unacknowledged batch.
144    pub fn set_watermark_age(&self, age: Duration) {
145        self.watermark_age.set(age.as_secs_f64());
146    }
147}
spate_core/metrics/checkpoint.rs

spate_core/metrics/
checkpoint.rs
