laminar_connectors/postgres/cdc/
metrics.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    /// Current confirmed flush LSN (as raw u64).
37    pub confirmed_flush_lsn: IntGauge,
38
39    /// Current replication lag in bytes (`write_lsn` - `confirmed_flush_lsn`).
40    pub replication_lag_bytes: IntGauge,
41}
42
43impl PostgresCdcMetrics {
44    /// Creates a new metrics instance with all counters at zero.
45    ///
46    /// # Panics
47    ///
48    /// Panics if a built-in metric descriptor is invalid.
49    #[must_use]
50    pub fn new(registry: Option<&Registry>) -> Self {
51        let mut local = None;
52        let reg = reg_or_local(registry, &mut local);
53
54        Self {
55            events_received: reg.counter(
56                "postgres_cdc_events_received_total",
57                "Total CDC change events received",
58            ),
59            bytes_received: reg.counter(
60                "postgres_cdc_bytes_received_total",
61                "Total bytes from WAL stream",
62            ),
63            batches_produced: reg.counter(
64                "postgres_cdc_batches_produced_total",
65                "Total batches produced",
66            ),
67            inserts: reg.counter("postgres_cdc_inserts_total", "Total INSERT events"),
68            updates: reg.counter("postgres_cdc_updates_total", "Total UPDATE events"),
69            deletes: reg.counter("postgres_cdc_deletes_total", "Total DELETE events"),
70            transactions: reg.counter(
71                "postgres_cdc_transactions_total",
72                "Total transactions received",
73            ),
74            confirmed_flush_lsn: reg.gauge(
75                "postgres_cdc_confirmed_flush_lsn",
76                "Current confirmed flush LSN",
77            ),
78            replication_lag_bytes: reg.gauge(
79                "postgres_cdc_replication_lag_bytes",
80                "Replication lag in bytes",
81            ),
82        }
83    }
84
85    /// Records a received INSERT event.
86    pub fn record_insert(&self) {
87        self.inserts.inc();
88        self.events_received.inc();
89    }
90
91    /// Records a received UPDATE event.
92    pub fn record_update(&self) {
93        self.updates.inc();
94        self.events_received.inc();
95    }
96
97    /// Records a received DELETE event.
98    pub fn record_delete(&self) {
99        self.deletes.inc();
100        self.events_received.inc();
101    }
102
103    /// Records a received transaction commit.
104    pub fn record_transaction(&self) {
105        self.transactions.inc();
106    }
107
108    /// Records bytes received from the WAL stream.
109    pub fn record_bytes(&self, bytes: u64) {
110        self.bytes_received.inc_by(bytes);
111    }
112
113    /// Records a batch produced for downstream.
114    pub fn record_batch(&self) {
115        self.batches_produced.inc();
116    }
117
118    /// Updates the confirmed flush LSN.
119    pub fn set_confirmed_flush_lsn(&self, lsn: u64) {
120        self.confirmed_flush_lsn
121            .set(i64::from_ne_bytes(lsn.to_ne_bytes()));
122    }
123
124    /// Updates the replication lag in bytes.
125    pub fn set_replication_lag_bytes(&self, lag: u64) {
126        self.replication_lag_bytes
127            .set(i64::from_ne_bytes(lag.to_ne_bytes()));
128    }
129}
130
131impl Default for PostgresCdcMetrics {
132    fn default() -> Self {
133        Self::new(None)
134    }
135}
136
137#[cfg(test)]
138mod tests {
139    use super::*;
140
141    #[test]
142    fn test_record_operations() {
143        let m = PostgresCdcMetrics::new(None);
144        m.record_insert();
145        m.record_insert();
146        m.record_update();
147        m.record_delete();
148        m.record_transaction();
149        m.record_bytes(1024);
150        m.record_batch();
151
152        assert_eq!(m.events_received.get(), 4);
153        assert_eq!(m.inserts.get(), 2);
154        assert_eq!(m.updates.get(), 1);
155        assert_eq!(m.deletes.get(), 1);
156        assert_eq!(m.transactions.get(), 1);
157        assert_eq!(m.bytes_received.get(), 1024);
158        assert_eq!(m.batches_produced.get(), 1);
159    }
160
161    #[test]
162    fn test_lsn_and_lag_tracking() {
163        let m = PostgresCdcMetrics::new(None);
164        m.set_confirmed_flush_lsn(0x1234_ABCD);
165        m.set_replication_lag_bytes(4096);
166
167        assert_eq!(m.confirmed_flush_lsn.get(), 0x1234_ABCD_i64);
168        assert_eq!(m.replication_lag_bytes.get(), 4096);
169    }
170}
laminar_connectors/postgres/cdc/metrics.rs

laminar_connectors/postgres/cdc/
metrics.rs
