laminar_connectors/kafka/
metrics.rs

1//! Prometheus-backed Kafka source metrics.
2
3use prometheus::{IntCounter, IntGauge, Registry};
4
5use crate::prom::reg_or_local;
6
7/// Prometheus-backed counters/gauges for Kafka source connector statistics.
8#[derive(Debug, Clone)]
9pub struct KafkaSourceMetrics {
10    /// Total records polled from Kafka.
11    pub records_polled: IntCounter,
12    /// Total bytes polled from Kafka.
13    pub bytes_polled: IntCounter,
14    /// Total deserialization or consumer errors.
15    pub errors: IntCounter,
16    /// Total batches returned from `poll_batch()`.
17    pub batches_polled: IntCounter,
18    /// Total offset commits to Kafka.
19    pub commits: IntCounter,
20    /// Broker progress commits that failed locally or remotely.
21    pub commit_failures: IntCounter,
22    /// Total consumer group rebalances.
23    pub rebalances: IntCounter,
24    /// Count of successful Schema Registry discoveries at DDL time.
25    pub sr_discovery_successes: IntCounter,
26    /// Count of Schema Registry discovery failures (HTTP error, parse error).
27    pub sr_discovery_failures: IntCounter,
28    /// Count of Schema Registry discovery timeouts.
29    pub sr_discovery_timeouts: IntCounter,
30    /// Records using a prepared writer schema, observed once per batch.
31    pub schema_cache_hits: IntCounter,
32    /// Unknown writer definitions fetched on the connector runtime.
33    pub schema_cache_misses: IntCounter,
34    /// Writer fetch failures or cancellation, before source progress publication.
35    pub schema_fetch_failures: IntCounter,
36    /// Cumulative queue/fetch latency, including cancelled unknown writer resolutions.
37    pub schema_fetch_microseconds: IntCounter,
38    /// Records currently blocked behind unresolved writer metadata.
39    pub schema_unresolved_records: IntGauge,
40}
41
42impl KafkaSourceMetrics {
43    /// If `registry` is `Some`, counters are registered there (visible
44    /// in the Prometheus scrape); otherwise a throwaway registry is used.
45    #[must_use]
46    #[allow(clippy::missing_panics_doc, clippy::too_many_lines)]
47    pub fn new(registry: Option<&Registry>) -> Self {
48        let mut local = None;
49        let handle = reg_or_local(registry, &mut local);
50
51        Self {
52            records_polled: handle.counter(
53                "kafka_source_records_polled_total",
54                "Total records polled from Kafka",
55            ),
56            bytes_polled: handle.counter(
57                "kafka_source_bytes_polled_total",
58                "Total bytes polled from Kafka",
59            ),
60            errors: handle.counter("kafka_source_errors_total", "Total Kafka consumer errors"),
61            batches_polled: handle.counter(
62                "kafka_source_batches_polled_total",
63                "Total batches polled from Kafka",
64            ),
65            commits: handle.counter(
66                "kafka_source_commits_total",
67                "Total offset commits to Kafka",
68            ),
69            commit_failures: handle.counter(
70                "kafka_source_commit_failures_total",
71                "Kafka broker progress commit failures",
72            ),
73            rebalances: handle.counter(
74                "kafka_source_rebalances_total",
75                "Total consumer group rebalances",
76            ),
77            sr_discovery_successes: handle.counter(
78                "kafka_source_sr_discovery_successes_total",
79                "Schema Registry discovery successes",
80            ),
81            sr_discovery_failures: handle.counter(
82                "kafka_source_sr_discovery_failures_total",
83                "Schema Registry discovery failures",
84            ),
85            schema_cache_hits: handle.counter(
86                "kafka_source_schema_cache_hits_total",
87                "Records decoded with a prepared writer contract",
88            ),
89            schema_cache_misses: handle.counter(
90                "kafka_source_schema_cache_misses_total",
91                "Unknown writer schema resolutions",
92            ),
93            schema_fetch_failures: handle.counter(
94                "kafka_source_schema_fetch_failures_total",
95                "Failed or cancelled writer schema resolutions",
96            ),
97            schema_fetch_microseconds: handle.counter(
98                "kafka_source_schema_fetch_microseconds_total",
99                "Cumulative unresolved writer queue and fetch latency",
100            ),
101            schema_unresolved_records: handle.gauge(
102                "kafka_source_schema_unresolved_records",
103                "Records blocked on writer metadata",
104            ),
105            sr_discovery_timeouts: handle.counter(
106                "kafka_source_sr_discovery_timeouts_total",
107                "Schema Registry discovery timeouts",
108            ),
109        }
110    }
111
112    /// Records a successful poll of `records` records totaling `bytes`.
113    pub fn record_poll(&self, records: u64, bytes: u64) {
114        self.records_polled.inc_by(records);
115        self.bytes_polled.inc_by(bytes);
116        self.batches_polled.inc();
117    }
118
119    /// Records a consumer or deserialization error.
120    pub fn record_error(&self) {
121        self.errors.inc();
122    }
123
124    /// Records a consumer group rebalance event.
125    pub fn record_rebalance(&self) {
126        self.rebalances.inc();
127    }
128
129    /// Records a successful Schema Registry discovery at DDL time.
130    pub fn record_sr_discovery_success(&self) {
131        self.sr_discovery_successes.inc();
132    }
133
134    /// Records a Schema Registry discovery failure.
135    pub fn record_sr_discovery_failure(&self) {
136        self.sr_discovery_failures.inc();
137    }
138
139    /// Records a Schema Registry discovery timeout.
140    pub fn record_sr_discovery_timeout(&self) {
141        self.sr_discovery_timeouts.inc();
142    }
143}
144
145impl Default for KafkaSourceMetrics {
146    fn default() -> Self {
147        Self::new(None)
148    }
149}
150
151#[cfg(test)]
152mod tests {
153    use super::*;
154
155    #[test]
156    fn test_initial_zeros() {
157        let m = KafkaSourceMetrics::new(None);
158        assert_eq!(m.records_polled.get(), 0);
159        assert_eq!(m.bytes_polled.get(), 0);
160        assert_eq!(m.errors.get(), 0);
161    }
162
163    #[test]
164    fn test_record_poll() {
165        let m = KafkaSourceMetrics::new(None);
166        m.record_poll(100, 5000);
167        m.record_poll(200, 10000);
168
169        assert_eq!(m.records_polled.get(), 300);
170        assert_eq!(m.bytes_polled.get(), 15000);
171    }
172
173    #[test]
174    fn test_record_error() {
175        let m = KafkaSourceMetrics::new(None);
176        m.record_error();
177        m.record_error();
178
179        assert_eq!(m.errors.get(), 2);
180    }
181
182    #[test]
183    fn test_record_commit_failure() {
184        let m = KafkaSourceMetrics::new(None);
185        m.commit_failures.inc();
186        assert_eq!(m.commit_failures.get(), 1);
187    }
188
189    #[test]
190    fn test_record_rebalance() {
191        let m = KafkaSourceMetrics::new(None);
192        m.record_rebalance();
193        m.record_rebalance();
194
195        assert_eq!(m.rebalances.get(), 2);
196    }
197
198    #[test]
199    fn test_sr_discovery_counters() {
200        let m = KafkaSourceMetrics::new(None);
201        m.record_sr_discovery_success();
202        m.record_sr_discovery_success();
203        m.record_sr_discovery_failure();
204        m.record_sr_discovery_timeout();
205
206        assert_eq!(m.sr_discovery_successes.get(), 2);
207        assert_eq!(m.sr_discovery_failures.get(), 1);
208        assert_eq!(m.sr_discovery_timeouts.get(), 1);
209    }
210
211    #[test]
212    fn test_registered_on_prometheus_registry() {
213        let reg = Registry::new();
214        let m = KafkaSourceMetrics::new(Some(&reg));
215        m.record_poll(10, 500);
216        m.record_error();
217
218        // Verify the metrics are registered on the registry.
219        let families = reg.gather();
220        let names: Vec<&str> = families
221            .iter()
222            .map(prometheus::proto::MetricFamily::name)
223            .collect();
224        assert!(names.contains(&"kafka_source_records_polled_total"));
225        assert!(names.contains(&"kafka_source_errors_total"));
226    }
227}
laminar_connectors/kafka/metrics.rs

laminar_connectors/kafka/
metrics.rs
