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 (see "Series ownership" in `docs/METRICS.md`).
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    /// Set one partition's pending count. No-op unless
89    /// `per_partition_detail`.
90    pub fn set_partition_pending(&self, partition: PartitionId, pending: usize) {
91        if let Some(pg) = &self.partition_pending {
92            pg.set(partition, pending as f64);
93        }
94    }
95
96    /// Drop per-partition series for revoked partitions.
97    pub fn retain_partitions(&self, keep: &[PartitionId]) {
98        if let Some(pg) = &self.partition_pending {
99            pg.retain(keep);
100        }
101    }
102
103    /// Record one source commit call.
104    pub fn commit(&self, ok: bool, d: Duration) {
105        if ok {
106            self.commits_ok.increment(1);
107        } else {
108            self.commits_err.increment(1);
109        }
110        self.commit_duration.record(d.as_secs_f64());
111    }
112
113    /// Set the age of the oldest unacknowledged batch.
114    pub fn set_watermark_age(&self, age: Duration) {
115        self.watermark_age.set(age.as_secs_f64());
116    }
117}
spate_core/metrics/checkpoint.rs

spate_core/metrics/
checkpoint.rs
