laminar_connectors/nats/
metrics.rs

1//! NATS connector metrics. No per-subject labels — subjects are
2//! wildcard-addressable and unbounded-cardinality.
3
4use prometheus::core::Collector;
5use prometheus::{Error as PromError, IntCounter, IntGauge, Registry};
6use tracing::warn;
7
8use crate::prom::reg_or_local;
9
10fn register_collector<C: Collector + Clone + 'static>(reg: &Registry, name: &str, c: &C) {
11    match reg.register(Box::new(c.clone())) {
12        Ok(()) => {}
13        // Multiple connectors on a shared registry can collide; the
14        // second registration silently drops so its counts won't scrape.
15        Err(PromError::AlreadyReg) => {
16            warn!(
17                metric = name,
18                "metric already registered; use separate registries per connector"
19            );
20        }
21        Err(e) => warn!(metric = name, error = ?e, "failed to register metric"),
22    }
23}
24
25/// Prometheus counters for the NATS source.
26#[derive(Debug, Clone)]
27#[allow(missing_docs)]
28pub struct NatsSourceMetrics {
29    pub records_total: IntCounter,
30    pub bytes_total: IntCounter,
31    pub fetch_errors_total: IntCounter,
32    pub acks_total: IntCounter,
33    pub ack_errors_total: IntCounter,
34    pub pending_acks: IntGauge,
35    /// Stream messages not yet delivered to the consumer.
36    pub consumer_lag: IntGauge,
37}
38
39impl NatsSourceMetrics {
40    /// Registers on `registry` if provided; otherwise on a local one.
41    #[must_use]
42    #[allow(clippy::missing_panics_doc)]
43    pub fn new(registry: Option<&Registry>) -> Self {
44        let mut local = None;
45        let handle = reg_or_local(registry, &mut local);
46        let reg = handle.registry();
47
48        macro_rules! reg_c {
49            ($name:expr, $help:expr) => {{
50                let c = IntCounter::new($name, $help).unwrap();
51                register_collector(reg, $name, &c);
52                c
53            }};
54        }
55        let pending_acks = IntGauge::new(
56            "nats_source_pending_acks",
57            "JetStream acks queued or in flight",
58        )
59        .unwrap();
60        register_collector(reg, "nats_source_pending_acks", &pending_acks);
61        let consumer_lag = IntGauge::new(
62            "nats_source_consumer_lag",
63            "Stream messages not yet delivered to the consumer",
64        )
65        .unwrap();
66        register_collector(reg, "nats_source_consumer_lag", &consumer_lag);
67
68        Self {
69            records_total: reg_c!("nats_source_records_total", "Records delivered"),
70            bytes_total: reg_c!("nats_source_bytes_total", "Payload bytes delivered"),
71            fetch_errors_total: reg_c!("nats_source_fetch_errors_total", "Fetch errors"),
72            acks_total: reg_c!("nats_source_acks_total", "Successful acks"),
73            ack_errors_total: reg_c!("nats_source_ack_errors_total", "Failed acks"),
74            pending_acks,
75            consumer_lag,
76        }
77    }
78
79    #[allow(missing_docs)]
80    pub fn record_poll(&self, records: u64, bytes: u64) {
81        self.records_total.inc_by(records);
82        self.bytes_total.inc_by(bytes);
83    }
84
85    #[allow(missing_docs)]
86    pub fn record_fetch_error(&self) {
87        self.fetch_errors_total.inc();
88    }
89
90    pub(crate) fn record_ack_enqueued(&self) {
91        self.pending_acks.inc();
92    }
93
94    #[allow(missing_docs)]
95    pub fn record_ack(&self) {
96        self.acks_total.inc();
97        self.pending_acks.dec();
98    }
99
100    #[allow(missing_docs)]
101    pub fn record_ack_error(&self) {
102        self.ack_errors_total.inc();
103        self.pending_acks.dec();
104    }
105
106    pub(crate) fn record_ack_enqueue_errors(&self, n: usize) {
107        self.ack_errors_total
108            .inc_by(u64::try_from(n).unwrap_or(u64::MAX));
109    }
110
111    #[allow(missing_docs, clippy::cast_possible_wrap)]
112    pub fn record_ack_abandoned(&self, n: usize) {
113        self.ack_errors_total
114            .inc_by(u64::try_from(n).unwrap_or(u64::MAX));
115        self.pending_acks.sub(n as i64);
116    }
117
118    pub(crate) fn record_abandoned_acks(&self) {
119        let pending = self.pending_acks.get();
120        if pending > 0 {
121            self.ack_errors_total
122                .inc_by(u64::try_from(pending).unwrap_or(u64::MAX));
123            self.pending_acks.set(0);
124        }
125    }
126
127    #[allow(missing_docs, clippy::cast_possible_wrap)]
128    pub fn set_consumer_lag(&self, n: u64) {
129        self.consumer_lag.set(n as i64);
130    }
131}
132
133/// Prometheus counters for the NATS sink.
134#[derive(Debug, Clone)]
135#[allow(missing_docs)]
136pub struct NatsSinkMetrics {
137    pub records_total: IntCounter,
138    pub bytes_total: IntCounter,
139    pub publish_errors_total: IntCounter,
140    pub ack_errors_total: IntCounter,
141    /// Publishes the broker dropped as `Nats-Msg-Id` duplicates.
142    pub dedup_total: IntCounter,
143    pub pending_futures: IntGauge,
144}
145
146impl NatsSinkMetrics {
147    /// Registers on `registry` if provided; otherwise on a local one.
148    #[must_use]
149    #[allow(clippy::missing_panics_doc)]
150    pub fn new(registry: Option<&Registry>) -> Self {
151        let mut local = None;
152        let handle = reg_or_local(registry, &mut local);
153        let reg = handle.registry();
154
155        macro_rules! reg_c {
156            ($name:expr, $help:expr) => {{
157                let c = IntCounter::new($name, $help).unwrap();
158                register_collector(reg, $name, &c);
159                c
160            }};
161        }
162        let pending_futures =
163            IntGauge::new("nats_sink_pending_futures", "Outstanding PublishAckFutures").unwrap();
164        register_collector(reg, "nats_sink_pending_futures", &pending_futures);
165
166        Self {
167            records_total: reg_c!("nats_sink_records_total", "Records published"),
168            bytes_total: reg_c!("nats_sink_bytes_total", "Payload bytes published"),
169            publish_errors_total: reg_c!("nats_sink_publish_errors_total", "Publish errors"),
170            ack_errors_total: reg_c!("nats_sink_ack_errors_total", "Publish-ack errors"),
171            dedup_total: reg_c!("nats_sink_dedup_total", "Broker-dropped duplicates"),
172            pending_futures,
173        }
174    }
175
176    #[allow(missing_docs)]
177    pub fn record_published_row(&self, bytes: u64) {
178        self.records_total.inc();
179        self.bytes_total.inc_by(bytes);
180    }
181
182    #[allow(missing_docs)]
183    pub fn record_publish_error(&self) {
184        self.publish_errors_total.inc();
185    }
186
187    #[allow(missing_docs)]
188    pub fn record_ack_error(&self) {
189        self.ack_errors_total.inc();
190    }
191
192    #[allow(missing_docs)]
193    pub fn record_dedup(&self) {
194        self.dedup_total.inc();
195    }
196
197    #[allow(missing_docs, clippy::cast_possible_wrap)]
198    pub fn set_pending_futures(&self, n: usize) {
199        self.pending_futures.set(n as i64);
200    }
201}
202
203#[cfg(test)]
204mod tests {
205    use super::*;
206
207    #[test]
208    fn source_initial_zero() {
209        let m = NatsSourceMetrics::new(None);
210        assert_eq!(m.records_total.get(), 0);
211        assert_eq!(m.bytes_total.get(), 0);
212        assert_eq!(m.fetch_errors_total.get(), 0);
213        assert_eq!(m.consumer_lag.get(), 0);
214    }
215
216    #[test]
217    fn source_records_increment() {
218        let m = NatsSourceMetrics::new(None);
219        m.record_poll(100, 4096);
220        m.record_poll(50, 1024);
221        m.record_ack_enqueued();
222        m.record_ack();
223        m.record_ack_enqueued();
224        m.record_ack_error();
225        m.record_ack_enqueued();
226        m.record_ack_enqueued();
227        m.record_ack_abandoned(2);
228        m.record_fetch_error();
229
230        m.set_consumer_lag(42);
231
232        assert_eq!(m.records_total.get(), 150);
233        assert_eq!(m.bytes_total.get(), 5120);
234        assert_eq!(m.fetch_errors_total.get(), 1);
235        assert_eq!(m.consumer_lag.get(), 42);
236        assert_eq!(m.acks_total.get(), 1);
237        assert_eq!(m.ack_errors_total.get(), 3);
238        assert_eq!(m.pending_acks.get(), 0);
239    }
240
241    #[test]
242    fn sink_records_and_dedup() {
243        let m = NatsSinkMetrics::new(None);
244        for _ in 0..10 {
245            m.record_published_row(200);
246        }
247        m.record_dedup();
248        m.record_dedup();
249        m.set_pending_futures(3);
250
251        assert_eq!(m.records_total.get(), 10);
252        assert_eq!(m.bytes_total.get(), 2000);
253        assert_eq!(m.pending_futures.get(), 3);
254        assert_eq!(m.dedup_total.get(), 2);
255    }
256
257    #[test]
258    fn metrics_register_on_shared_registry() {
259        let reg = Registry::new();
260        let _s = NatsSourceMetrics::new(Some(&reg));
261        let _k = NatsSinkMetrics::new(Some(&reg));
262        let names: Vec<String> = reg.gather().iter().map(|f| f.name().to_string()).collect();
263        assert!(names.contains(&"nats_source_records_total".to_string()));
264        assert!(names.contains(&"nats_sink_records_total".to_string()));
265        assert!(names.contains(&"nats_sink_dedup_total".to_string()));
266    }
267}
laminar_connectors/nats/metrics.rs

laminar_connectors/nats/
metrics.rs
