laminar_connectors/postgres/cdc/metrics/
mod.rs

1//! `PostgreSQL` CDC source connector metrics.
2//!
3//! Prometheus-backed counters/gauges for tracking CDC replication performance.
4
5use prometheus::{IntCounter, IntGauge, Registry};
6
7use crate::prom::reg_or_local;
8
9/// Metrics for the `PostgreSQL` CDC source connector.
10///
11/// All counters are prometheus-backed and will appear in the scrape
12/// output when a shared registry is provided.
13#[derive(Debug, Clone)]
14pub struct PostgresCdcMetrics {
15    /// Total change events received (insert + update + delete).
16    pub events_received: IntCounter,
17
18    /// Total bytes received from the WAL stream.
19    pub bytes_received: IntCounter,
20
21    /// Total batches produced for downstream.
22    pub batches_produced: IntCounter,
23
24    /// Total INSERT operations received.
25    pub inserts: IntCounter,
26
27    /// Total UPDATE operations received.
28    pub updates: IntCounter,
29
30    /// Total DELETE operations received.
31    pub deletes: IntCounter,
32
33    /// Total transactions (commit messages) received.
34    pub transactions: IntCounter,
35
36    /// Total rows emitted by initial snapshots.
37    pub snapshot_rows: IntCounter,
38
39    /// Current confirmed flush LSN (as raw u64).
40    pub confirmed_flush_lsn: IntGauge,
41
42    /// Current replication lag in bytes (`write_lsn` - `confirmed_flush_lsn`).
43    pub replication_lag_bytes: IntGauge,
44
45    /// Inactive slots of other claims under the `slot.name` prefix, last counted.
46    pub orphaned_slots: IntGauge,
47}
48
49impl PostgresCdcMetrics {
50    /// Creates a new metrics instance with all counters at zero.
51    ///
52    /// # Panics
53    ///
54    /// Panics if a built-in metric descriptor is invalid.
55    #[must_use]
56    pub fn new(registry: Option<&Registry>) -> Self {
57        let mut local = None;
58        let reg = reg_or_local(registry, &mut local);
59
60        Self {
61            events_received: reg.counter(
62                "postgres_cdc_events_received_total",
63                "Total CDC change events received",
64            ),
65            bytes_received: reg.counter(
66                "postgres_cdc_bytes_received_total",
67                "Total bytes from WAL stream",
68            ),
69            batches_produced: reg.counter(
70                "postgres_cdc_batches_produced_total",
71                "Total batches produced",
72            ),
73            inserts: reg.counter("postgres_cdc_inserts_total", "Total INSERT events"),
74            updates: reg.counter("postgres_cdc_updates_total", "Total UPDATE events"),
75            deletes: reg.counter("postgres_cdc_deletes_total", "Total DELETE events"),
76            transactions: reg.counter(
77                "postgres_cdc_transactions_total",
78                "Total transactions received",
79            ),
80            snapshot_rows: reg.counter(
81                "postgres_cdc_snapshot_rows_total",
82                "Total rows emitted by initial snapshots",
83            ),
84            confirmed_flush_lsn: reg.gauge(
85                "postgres_cdc_confirmed_flush_lsn",
86                "Current confirmed flush LSN",
87            ),
88            replication_lag_bytes: reg.gauge(
89                "postgres_cdc_replication_lag_bytes",
90                "Replication lag in bytes",
91            ),
92            orphaned_slots: reg.gauge(
93                "postgres_cdc_orphaned_slots",
94                "Inactive replication slots of other claims under the slot.name prefix",
95            ),
96        }
97    }
98
99    /// Records a received INSERT event.
100    pub fn record_insert(&self) {
101        self.inserts.inc();
102        self.events_received.inc();
103    }
104
105    /// Records a received UPDATE event.
106    pub fn record_update(&self) {
107        self.updates.inc();
108        self.events_received.inc();
109    }
110
111    /// Records a received DELETE event.
112    pub fn record_delete(&self) {
113        self.deletes.inc();
114        self.events_received.inc();
115    }
116
117    /// Records a received transaction commit.
118    pub fn record_transaction(&self) {
119        self.transactions.inc();
120    }
121
122    /// Records bytes received from the WAL stream.
123    pub fn record_bytes(&self, bytes: u64) {
124        self.bytes_received.inc_by(bytes);
125    }
126
127    /// Records rows emitted by an initial snapshot.
128    pub fn record_snapshot_rows(&self, rows: u64) {
129        self.snapshot_rows.inc_by(rows);
130    }
131
132    /// Records a batch produced for downstream.
133    pub fn record_batch(&self) {
134        self.batches_produced.inc();
135    }
136
137    /// Updates the confirmed flush LSN.
138    pub fn set_confirmed_flush_lsn(&self, lsn: u64) {
139        self.confirmed_flush_lsn
140            .set(i64::from_ne_bytes(lsn.to_ne_bytes()));
141    }
142
143    /// Updates the replication lag in bytes.
144    pub fn set_replication_lag_bytes(&self, lag: u64) {
145        self.replication_lag_bytes
146            .set(i64::from_ne_bytes(lag.to_ne_bytes()));
147    }
148
149    /// Updates the orphaned slot count.
150    pub fn set_orphaned_slots(&self, count: usize) {
151        self.orphaned_slots
152            .set(i64::try_from(count).unwrap_or(i64::MAX));
153    }
154}
155
156impl Default for PostgresCdcMetrics {
157    fn default() -> Self {
158        Self::new(None)
159    }
160}
161
162#[cfg(test)]
163mod tests;
laminar_connectors/postgres/cdc/metrics/mod.rs

laminar_connectors/postgres/cdc/metrics/
mod.rs
