laminar_connectors/lakehouse/delta_metrics/
mod.rs

1//! Prometheus-backed Delta Lake sink metrics.
2
3use prometheus::{Histogram, HistogramOpts, IntCounter, IntGauge, Registry};
4
5use super::metrics::LakehouseSinkMetrics;
6use crate::prom::reg_or_local;
7
8/// Prometheus-backed counters/gauges for Delta Lake sink connector statistics.
9#[derive(Debug, Clone)]
10pub struct DeltaLakeSinkMetrics {
11    /// Common metrics (rows flushed, bytes written, commits, etc.).
12    pub common: LakehouseSinkMetrics,
13
14    /// Total MERGE operations (upsert mode).
15    pub merge_operations: IntCounter,
16
17    /// Last Delta Lake table version committed.
18    pub last_delta_version: IntGauge,
19
20    /// End-to-end flush duration histogram (concat → write → checkpoint).
21    /// Buckets cover 5ms up to ~160s (0.005 * 2^15).
22    pub flush_duration: Histogram,
23
24    /// Total changelog rows entering collapse (pre-dedup, per upsert flush).
25    pub collapse_rows_in: IntCounter,
26
27    /// Total upsert rows emitted by collapse (`_op = U`).
28    pub collapse_upserts_out: IntCounter,
29
30    /// Total delete rows emitted by collapse (`_op = D`).
31    pub collapse_deletes_out: IntCounter,
32
33    /// Changelog-collapse duration histogram (per upsert flush).
34    /// Buckets cover 100µs up to ~3.3s (0.0001 * 2^15).
35    pub collapse_duration: Histogram,
36}
37
38impl DeltaLakeSinkMetrics {
39    /// Creates a new metrics instance with all counters at zero.
40    #[must_use]
41    #[allow(clippy::missing_panics_doc)]
42    pub fn new(registry: Option<&Registry>) -> Self {
43        let mut local = None;
44        let handle = reg_or_local(registry, &mut local);
45
46        let flush_duration = Histogram::with_opts(
47            HistogramOpts::new(
48                "delta_sink_flush_duration_seconds",
49                "End-to-end Delta Lake flush duration (pre-concat → write → checkpoint)",
50            )
51            .buckets(prometheus::exponential_buckets(0.005, 2.0, 16).unwrap()),
52        )
53        .unwrap();
54        // Best-effort registration (matches `kafka/metrics.rs` pattern):
55        // `AlreadyReg` is benign on re-init; surface anything else so a
56        // dropped histogram doesn't disappear silently.
57        if let Err(e) = handle.registry().register(Box::new(flush_duration.clone())) {
58            tracing::warn!(
59                metric = "delta_sink_flush_duration_seconds",
60                error = %e,
61                "failed to register delta lake flush_duration histogram"
62            );
63        }
64
65        let collapse_duration = Histogram::with_opts(
66            HistogramOpts::new(
67                "delta_sink_collapse_duration_seconds",
68                "Changelog collapse duration per upsert flush (Z-set/CDC dedup)",
69            )
70            .buckets(prometheus::exponential_buckets(0.0001, 2.0, 16).unwrap()),
71        )
72        .unwrap();
73        if let Err(e) = handle
74            .registry()
75            .register(Box::new(collapse_duration.clone()))
76        {
77            tracing::warn!(
78                metric = "delta_sink_collapse_duration_seconds",
79                error = %e,
80                "failed to register delta lake collapse_duration histogram"
81            );
82        }
83
84        Self {
85            common: LakehouseSinkMetrics::new(registry),
86            merge_operations: handle.counter(
87                "delta_sink_merge_operations_total",
88                "Total MERGE operations (upsert)",
89            ),
90            last_delta_version: handle.gauge(
91                "delta_sink_last_version",
92                "Last committed Delta table version",
93            ),
94            flush_duration,
95            collapse_rows_in: handle.counter(
96                "delta_sink_collapse_rows_in_total",
97                "Changelog rows entering collapse (pre-dedup)",
98            ),
99            collapse_upserts_out: handle.counter(
100                "delta_sink_collapse_upserts_out_total",
101                "Upsert rows emitted by collapse (_op = U)",
102            ),
103            collapse_deletes_out: handle.counter(
104                "delta_sink_collapse_deletes_out_total",
105                "Delete rows emitted by collapse (_op = D)",
106            ),
107            collapse_duration,
108        }
109    }
110
111    /// Records a successful flush of `records` rows totaling `bytes`.
112    pub fn record_flush(&self, records: u64, bytes: u64) {
113        self.common.record_flush(records, bytes);
114    }
115
116    /// Records a successful epoch commit.
117    #[allow(clippy::cast_possible_wrap)]
118    pub fn record_commit(&self, delta_version: u64) {
119        self.common.record_commit();
120        self.last_delta_version.set(delta_version as i64);
121    }
122
123    /// Records a write or I/O error.
124    pub fn record_error(&self) {
125        self.common.record_error();
126    }
127
128    /// Records an epoch rollback.
129    pub fn record_rollback(&self) {
130        self.common.record_rollback();
131    }
132
133    /// Records a MERGE operation (upsert mode).
134    pub fn record_merge(&self) {
135        self.merge_operations.inc();
136    }
137
138    /// Records changelog DELETE operations.
139    pub fn record_deletes(&self, count: u64) {
140        self.common.record_deletes(count);
141    }
142
143    /// Records a completed flush duration (seconds).
144    pub fn observe_flush_duration(&self, seconds: f64) {
145        self.flush_duration.observe(seconds);
146    }
147
148    /// Records one changelog-collapse pass: `rows_in` rows folded down to
149    /// `upserts_out` upserts and `deletes_out` deletes in `seconds`.
150    pub fn observe_collapse(&self, rows_in: u64, upserts_out: u64, deletes_out: u64, seconds: f64) {
151        self.collapse_rows_in.inc_by(rows_in);
152        self.collapse_upserts_out.inc_by(upserts_out);
153        self.collapse_deletes_out.inc_by(deletes_out);
154        self.collapse_duration.observe(seconds);
155    }
156}
157
158impl Default for DeltaLakeSinkMetrics {
159    fn default() -> Self {
160        Self::new(None)
161    }
162}
163
164#[cfg(test)]
165mod tests;
laminar_connectors/lakehouse/delta_metrics/mod.rs

laminar_connectors/lakehouse/delta_metrics/
mod.rs
