etl_core/metrics/
mod.rs

1//! Metrics: exporter installation and pre-registered handle structs for
2//! every pipeline stage.
3//!
4//! `etl-rs` instruments through the [`metrics`] facade — pipeline authors
5//! register custom metrics with the same macros and they are exported
6//! alongside the framework's. [`install`] wires the exporter selected by
7//! configuration; the taxonomy contract lives in `docs/METRICS.md` and its
8//! names in [`names`].
9//!
10//! # One pipeline per process
11//!
12//! The exporter installs a **process-global** recorder (the `metrics`
13//! facade has one global recorder), matching the framework's
14//! one-pipeline-per-process deployment model. [`install`] therefore
15//! succeeds at most once per process; a second call returns
16//! [`MetricsError::AlreadyInstalled`].
17//!
18//! # Hot-path discipline
19//!
20//! All handles are pre-registered at pipeline build time via the structs in
21//! this module ([`SourceMetrics`], [`SinkShardMetrics`], ...). The record
22//! loop only ever touches resolved `Counter`/`Gauge`/`Histogram` handles,
23//! and methods take per-batch aggregates.
24
25mod handles;
26pub mod names;
27
28pub use handles::{
29    BackpressureMetrics, CheckpointMetrics, ComponentLabels, DeserMetrics, FlushReason,
30    OperatorMetrics, PipelineMetrics, PipelineState, QueueMetrics, SinkShardMetrics, SourceMetrics,
31};
32
33use metrics_exporter_prometheus::{BuildError, Matcher, PrometheusBuilder, PrometheusHandle};
34use std::net::{Ipv4Addr, SocketAddr};
35use std::sync::Arc;
36use std::time::Duration;
37
38/// Buckets for `*_duration_seconds` histograms and
39/// `etl_e2e_latency_seconds` (1 ms .. 60 s, roughly exponential).
40pub const DURATION_SECONDS_BUCKETS: &[f64] = &[
41    0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0,
42];
43
44/// Buckets for `etl_sink_batch_rows` (powers of 4, 64 .. 1Mi rows).
45pub const BATCH_ROWS_BUCKETS: &[f64] = &[
46    64.0, 256.0, 1024.0, 4096.0, 16384.0, 65536.0, 262144.0, 1048576.0,
47];
48
49/// Buckets for `etl_sink_batch_bytes` (powers of 4, 4 KiB .. 256 MiB).
50pub const BATCH_BYTES_BUCKETS: &[f64] = &[
51    4096.0,
52    16384.0,
53    65536.0,
54    262144.0,
55    1048576.0,
56    4194304.0,
57    16777216.0,
58    67108864.0,
59    268435456.0,
60];
61
62/// Which exporter to install.
63#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
64#[non_exhaustive]
65pub enum Exporter {
66    /// Prometheus scrape endpoint, served by the admin server.
67    #[default]
68    Prometheus,
69    /// No export; all handles become no-ops.
70    None,
71}
72
73/// Time basis for `etl_e2e_latency_seconds`.
74#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
75#[non_exhaustive]
76pub enum E2eBasis {
77    /// Framework ingest time — clock-skew free (default).
78    #[default]
79    Ingest,
80    /// The record's event time (e.g. Kafka message timestamp) —
81    /// clock-skew sensitive but reflects true upstream delay.
82    Event,
83}
84
85/// Exporter settings, mapped from the `metrics` config section by the
86/// pipeline runtime. Defined here (not in `config`) so this module has no
87/// config dependency.
88#[derive(Clone, Debug, PartialEq, Eq)]
89pub struct MetricsSettings {
90    /// Which exporter to install.
91    pub exporter: Exporter,
92    /// Admin-server listen address (`/metrics`, `/healthz`, `/readyz`).
93    pub listen: SocketAddr,
94    /// Enable cardinality-sensitive per-partition series.
95    pub per_partition_detail: bool,
96    /// Time basis for end-to-end latency.
97    pub e2e_basis: E2eBasis,
98}
99
100impl Default for MetricsSettings {
101    fn default() -> Self {
102        MetricsSettings {
103            exporter: Exporter::Prometheus,
104            listen: SocketAddr::from((Ipv4Addr::UNSPECIFIED, 9090)),
105            per_partition_detail: false,
106            e2e_basis: E2eBasis::Ingest,
107        }
108    }
109}
110
111/// Exporter installation failed.
112#[derive(Debug, thiserror::Error)]
113#[non_exhaustive]
114pub enum MetricsError {
115    /// A global recorder is already installed in this process (one pipeline
116    /// per process).
117    #[error("a metrics recorder is already installed in this process")]
118    AlreadyInstalled,
119    /// The exporter rejected its configuration.
120    #[error("failed to build the metrics exporter: {0}")]
121    Build(String),
122}
123
124/// Handle to the installed exporter. Cheap to clone.
125#[derive(Clone, Debug)]
126pub struct MetricsHandle {
127    inner: Inner,
128    process: Option<Arc<metrics_process::Collector>>,
129}
130
131#[derive(Clone, Debug)]
132enum Inner {
133    Prometheus(PrometheusHandle),
134    Noop,
135}
136
137impl MetricsHandle {
138    /// Render the current exposition-format snapshot (empty for the no-op
139    /// exporter).
140    #[must_use]
141    pub fn render(&self) -> String {
142        match &self.inner {
143            Inner::Prometheus(handle) => {
144                if let Some(process) = &self.process {
145                    process.collect();
146                }
147                handle.render()
148            }
149            Inner::Noop => String::new(),
150        }
151    }
152
153    /// The render function seam handed to the admin server, keeping it
154    /// independent of exporter internals.
155    #[must_use]
156    pub fn render_fn(&self) -> Arc<dyn Fn() -> String + Send + Sync> {
157        let this = self.clone();
158        Arc::new(move || this.render())
159    }
160
161    /// One maintenance tick: drains histogram state and refreshes process
162    /// metrics. Cheap; call on an interval.
163    pub fn upkeep_tick(&self) {
164        if let Inner::Prometheus(handle) = &self.inner {
165            handle.run_upkeep();
166        }
167        if let Some(process) = &self.process {
168            process.collect();
169        }
170    }
171
172    /// Spawn the periodic upkeep task on the current tokio runtime.
173    #[must_use]
174    pub fn spawn_upkeep(&self, period: Duration) -> tokio::task::JoinHandle<()> {
175        let this = self.clone();
176        tokio::spawn(async move {
177            let mut tick = tokio::time::interval(period);
178            tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
179            loop {
180                tick.tick().await;
181                this.upkeep_tick();
182            }
183        })
184    }
185}
186
187/// A Prometheus builder pre-configured with the bucket layout from
188/// `docs/METRICS.md`.
189fn configured_builder() -> Result<PrometheusBuilder, BuildError> {
190    PrometheusBuilder::new()
191        .set_buckets_for_metric(
192            Matcher::Suffix("_duration_seconds".into()),
193            DURATION_SECONDS_BUCKETS,
194        )?
195        .set_buckets_for_metric(
196            Matcher::Full(names::E2E_LATENCY_SECONDS.into()),
197            DURATION_SECONDS_BUCKETS,
198        )?
199        .set_buckets_for_metric(
200            Matcher::Full(names::SINK_BATCH_ROWS.into()),
201            BATCH_ROWS_BUCKETS,
202        )?
203        .set_buckets_for_metric(
204            Matcher::Full(names::SINK_BATCH_BYTES.into()),
205            BATCH_BYTES_BUCKETS,
206        )
207}
208
209/// The handle from this process's successful [`install`]. Installation is
210/// once-per-process (the recorder is global); later `install` calls reuse
211/// this handle instead of failing, so assembly code can install the
212/// exporter *before* pre-registering metric handles and the runtime's own
213/// install becomes a no-op.
214static INSTALLED: std::sync::OnceLock<MetricsHandle> = std::sync::OnceLock::new();
215
216/// The settings of the first successful [`install`], kept so later calls
217/// with different settings can warn that theirs are ignored.
218static INSTALLED_SETTINGS: std::sync::OnceLock<MetricsSettings> = std::sync::OnceLock::new();
219
220/// Install the configured exporter as this process's global recorder and
221/// return the handle the admin server renders from.
222///
223/// **Call this before constructing any metric handle structs**
224/// ([`SinkShardMetrics`](crate::metrics::SinkShardMetrics) and friends):
225/// handles bind to the recorder present at construction, and handles built
226/// earlier record into the void. Idempotent — a second call returns the
227/// first call's handle (with a warning when the requested settings differ).
228/// [`MetricsError::AlreadyInstalled`] is only returned when a *foreign*
229/// global recorder (not installed through this function) already exists.
230///
231/// For [`Exporter::Prometheus`] this also registers the `process_*`
232/// collector (CPU, memory, fds). No HTTP listener is spawned here — the
233/// admin server owns the socket.
234pub fn install(settings: &MetricsSettings) -> Result<MetricsHandle, MetricsError> {
235    // Exporter::None installs no global recorder at all, so it neither
236    // claims nor consults the once-per-process slot — a later Prometheus
237    // install still works (and tests with metrics disabled stay isolated).
238    if settings.exporter == Exporter::None {
239        return Ok(MetricsHandle {
240            inner: Inner::Noop,
241            process: None,
242        });
243    }
244    if let Some(existing) = INSTALLED.get() {
245        if INSTALLED_SETTINGS
246            .get()
247            .is_some_and(|first| first != settings)
248        {
249            tracing::warn!(
250                requested = ?settings,
251                active = ?INSTALLED_SETTINGS.get(),
252                "metrics exporter already installed with different settings; \
253                 the first install's exporter stays in effect"
254            );
255        }
256        return Ok(existing.clone());
257    }
258    let builder = configured_builder().map_err(|e| MetricsError::Build(e.to_string()))?;
259    let handle = builder.install_recorder().map_err(|e| match e {
260        BuildError::FailedToSetGlobalRecorder(_) => MetricsError::AlreadyInstalled,
261        other => MetricsError::Build(other.to_string()),
262    })?;
263    let process = metrics_process::Collector::new("process_");
264    process.describe();
265    process.collect();
266    let handle = MetricsHandle {
267        inner: Inner::Prometheus(handle),
268        process: Some(Arc::new(process)),
269    };
270    let _ = INSTALLED_SETTINGS.set(settings.clone());
271    Ok(INSTALLED.get_or_init(|| handle).clone())
272}
273
274#[cfg(all(test, not(loom)))] // exporter internals (quanta) are loom-aware; not our model
275mod tests {
276    use super::*;
277    use crate::error::ErrorClass;
278    use crate::record::PartitionId;
279
280    /// Build a local (non-global) recorder with the production bucket
281    /// configuration, run `f` against it, and return the rendered output.
282    fn render_with_local_recorder(f: impl FnOnce()) -> String {
283        let recorder = configured_builder()
284            .expect("bucket config must be valid")
285            .build_recorder();
286        let handle = recorder.handle();
287        metrics::with_local_recorder(&recorder, f);
288        handle.run_upkeep();
289        handle.render()
290    }
291
292    fn labels() -> ComponentLabels {
293        ComponentLabels::new("orders", "orders_kafka", "kafka")
294    }
295
296    #[test]
297    fn handle_structs_register_and_render_the_taxonomy() {
298        let rendered = render_with_local_recorder(|| {
299            let src = SourceMetrics::new(&labels(), true);
300            src.batch(512, 131_072);
301            src.poll_duration(Duration::from_millis(3));
302            src.set_lag_max(42);
303            src.set_partition_lag(PartitionId(7), 40);
304            src.rebalance_assigned();
305            src.set_lanes_active(4);
306
307            let deser = DeserMetrics::new(&labels());
308            deser.batch(510, 2, Duration::from_millis(1));
309            deser.dropped(2);
310
311            let op = OperatorMetrics::new(&labels());
312            op.batch(510, 380, Duration::from_micros(600));
313            op.filtered(130);
314            op.errors(ErrorClass::RecordLevel, 1);
315
316            let q = QueueMetrics::new(&labels(), "chain->sink/0", 4096);
317            q.set_depth(17);
318            q.full_events(1);
319
320            let bp = BackpressureMetrics::new(&labels());
321            bp.pause_started();
322            bp.pause_ended(Duration::from_millis(250));
323            bp.set_inflight_bytes(1 << 20);
324
325            let shard = SinkShardMetrics::new(
326                &labels(),
327                3,
328                &["ch-3-0".into(), "ch-3-1".into()],
329                E2eBasis::Ingest,
330            );
331            shard.flushed(
332                FlushReason::Rows,
333                500_000,
334                64 << 20,
335                Duration::from_millis(90),
336            );
337            shard.retries(1);
338            shard.errors(ErrorClass::Retryable, 1);
339            shard.set_inflight(2);
340            shard.set_replica_healthy(1, false);
341            shard.breaker_opened(1);
342            shard.abandoned(0);
343
344            let cp = CheckpointMetrics::new(&labels(), false);
345            cp.set_pending_max(12);
346            cp.commit(true, Duration::from_millis(4));
347            cp.set_watermark_age(Duration::from_secs(1));
348
349            let pl = PipelineMetrics::new(&labels(), "0.1.0");
350            pl.set_state(PipelineState::Running);
351            pl.set_threads(4);
352        });
353
354        // Spot-check one series per stage, with labels.
355        for needle in [
356            r#"etl_source_records_total{pipeline="orders",component="orders_kafka",component_type="kafka"} 512"#,
357            r#"partition="7""#,
358            r#"etl_deser_records_total{pipeline="orders",component="orders_kafka",component_type="kafka",outcome="ok"} 510"#,
359            r#"etl_operator_records_dropped_total{pipeline="orders",component="orders_kafka",component_type="kafka",reason="filtered"} 130"#,
360            r#"etl_queue_capacity{pipeline="orders",component="orders_kafka",component_type="kafka",queue="chain->sink/0"} 4096"#,
361            r#"etl_backpressure_pause_events_total{pipeline="orders",component="orders_kafka",component_type="kafka"} 1"#,
362            r#"etl_sink_flushes_total{pipeline="orders",component="orders_kafka",component_type="kafka",shard="3",reason="rows"} 1"#,
363            r#"etl_sink_replica_healthy{pipeline="orders",component="orders_kafka",component_type="kafka",shard="3",replica="ch-3-1"} 0"#,
364            r#"etl_checkpoint_commits_total{pipeline="orders",component="orders_kafka",component_type="kafka",outcome="ok"} 1"#,
365            r#"etl_pipeline_state{pipeline="orders",component="orders_kafka",component_type="kafka",state="running"} 1"#,
366            r#"etl_pipeline_info{pipeline="orders",component="orders_kafka",component_type="kafka",version="0.1.0"} 1"#,
367        ] {
368            assert!(
369                rendered.contains(needle),
370                "rendered output missing `{needle}`:\n{rendered}"
371            );
372        }
373    }
374
375    #[test]
376    fn duration_histograms_use_configured_buckets() {
377        let rendered = render_with_local_recorder(|| {
378            let src = SourceMetrics::new(&labels(), false);
379            src.poll_duration(Duration::from_millis(3));
380        });
381        assert!(
382            rendered.contains(r#"le="0.005""#),
383            "expected a 5ms bucket boundary:\n{rendered}"
384        );
385        assert!(
386            rendered.contains("etl_source_poll_duration_seconds_bucket"),
387            "expected histogram exposition:\n{rendered}"
388        );
389    }
390
391    #[test]
392    fn per_partition_series_are_gated_and_retained() {
393        let rendered = render_with_local_recorder(|| {
394            let gated = SourceMetrics::new(&labels(), false);
395            gated.set_partition_lag(PartitionId(1), 5);
396
397            let detailed = CheckpointMetrics::new(&labels(), true);
398            detailed.set_partition_pending(PartitionId(1), 5);
399            detailed.set_partition_pending(PartitionId(2), 9);
400            detailed.retain_partitions(&[PartitionId(2)]);
401            // Re-set after retention: only partition 2 should re-register.
402            detailed.set_partition_pending(PartitionId(2), 11);
403        });
404        let gated_series_leaked = rendered
405            .lines()
406            .any(|l| l.starts_with("etl_source_lag_records") && l.contains("partition="));
407        assert!(
408            !gated_series_leaked,
409            "per-partition lag must be gated off:\n{rendered}"
410        );
411        assert!(rendered.contains(
412            r#"etl_checkpoint_pending_batches{pipeline="orders",component="orders_kafka",component_type="kafka",partition="2"} 11"#
413        ));
414    }
415
416    #[test]
417    fn state_gauge_flips_exactly_one_state() {
418        let rendered = render_with_local_recorder(|| {
419            let pl = PipelineMetrics::new(&labels(), "0.1.0");
420            pl.set_state(PipelineState::Draining);
421        });
422        assert!(rendered.contains(r#"state="draining"} 1"#));
423        for other in ["starting", "running", "failed"] {
424            assert!(
425                rendered.contains(&format!(r#"state="{other}"}} 0"#)),
426                "state `{other}` should read 0:\n{rendered}"
427            );
428        }
429    }
430
431    #[test]
432    fn noop_exporter_renders_empty() {
433        let handle = install(&MetricsSettings {
434            exporter: Exporter::None,
435            ..MetricsSettings::default()
436        })
437        .expect("noop install");
438        assert_eq!(handle.render(), "");
439        handle.upkeep_tick(); // must not panic
440    }
441
442    /// The single test that installs the process-global recorder: install,
443    /// register, render, upkeep. Kept as ONE test because a global recorder
444    /// can only be installed once per test process; all other tests use
445    /// local recorders.
446    #[test]
447    fn install_prometheus_end_to_end() {
448        let handle = install(&MetricsSettings::default()).expect("first install succeeds");
449
450        let pl = PipelineMetrics::new(&labels(), "0.1.0");
451        pl.set_threads(4);
452
453        let rendered = handle.render();
454        assert!(rendered.contains("etl_pipeline_info"));
455        assert!(rendered.contains("etl_pipeline_threads"));
456        assert!(
457            rendered.contains("process_cpu_seconds_total"),
458            "process collector wired:\n{rendered}"
459        );
460
461        handle.upkeep_tick();
462        let render_fn = handle.render_fn();
463        assert!(render_fn().contains("etl_pipeline_threads"));
464
465        // Install is idempotent: a second call returns the SAME exporter,
466        // so handles registered between the two calls stay visible. This is
467        // the assembly-order guarantee: user code installs early, registers
468        // sink handles, and the runtime's own install() reuses the exporter
469        // (the flagship-example pattern).
470        let shard = SinkShardMetrics::new(&labels(), 7, &["reuse-7-0".into()], E2eBasis::Ingest);
471        shard.flushed(FlushReason::Rows, 10, 1_000, Duration::from_millis(3));
472        shard.e2e_observed(Duration::from_millis(25), i64::MAX);
473        let second = install(&MetricsSettings::default()).expect("second install reuses");
474        let rendered = second.render();
475        assert!(
476            rendered.contains("etl_sink_records_total"),
477            "handles registered before the second install render through it:\n{rendered}"
478        );
479        assert!(rendered.contains("etl_e2e_latency_seconds"));
480
481        // Exporter::None never claims the process slot.
482        let noop = install(&MetricsSettings {
483            exporter: Exporter::None,
484            ..MetricsSettings::default()
485        })
486        .expect("noop install");
487        assert!(noop.render().is_empty());
488        assert!(
489            install(&MetricsSettings::default())
490                .expect("prometheus still reusable")
491                .render()
492                .contains("etl_pipeline_info")
493        );
494    }
495}
etl_core/metrics/mod.rs

etl_core/metrics/
mod.rs
