etl_core/metrics/
names.rs

1//! Metric and label name constants — the single source of truth for the
2//! taxonomy documented in `docs/METRICS.md`.
3//!
4//! Every framework metric is registered through these constants; nothing
5//! else may hard-code a metric name. Names follow Prometheus conventions:
6//! `_total` suffix on counters, unit suffixes (`_seconds`, `_bytes`,
7//! `_rows`) on everything measured in a unit, `etl_` prefix throughout.
8
9// Standard labels attached to every framework metric.
10
11/// Pipeline name label.
12pub const L_PIPELINE: &str = "pipeline";
13/// Component instance id label (e.g. `orders_kafka`).
14pub const L_COMPONENT: &str = "component";
15/// Component implementation label (e.g. `kafka`, `clickhouse`, `map`).
16pub const L_COMPONENT_TYPE: &str = "component_type";
17
18// Metric-specific labels.
19
20/// Source partition label (cardinality-gated by `per_partition_detail`).
21pub const L_PARTITION: &str = "partition";
22/// Sink shard label.
23pub const L_SHARD: &str = "shard";
24/// Sink replica label.
25pub const L_REPLICA: &str = "replica";
26/// Drop reason label (`filtered`, `skip_policy`).
27pub const L_REASON: &str = "reason";
28/// Outcome label (`ok`, `error`).
29pub const L_OUTCOME: &str = "outcome";
30/// Rebalance event label (`assign`, `revoke`).
31pub const L_EVENT: &str = "event";
32/// Error taxonomy class label (`retryable`, `record_level`, `fatal`).
33pub const L_ERROR_TYPE: &str = "error_type";
34/// Queue edge label (`<upstream>-><downstream>`).
35pub const L_QUEUE: &str = "queue";
36/// Pipeline state label (`starting`, `running`, `draining`, `failed`).
37pub const L_STATE: &str = "state";
38/// Build version label on `etl_pipeline_info`.
39pub const L_VERSION: &str = "version";
40
41// Source.
42
43/// Records emitted by the source (post-poll, pre-deserialization).
44pub const SOURCE_RECORDS_TOTAL: &str = "etl_source_records_total";
45/// Payload bytes emitted by the source.
46pub const SOURCE_BYTES_TOTAL: &str = "etl_source_bytes_total";
47/// Time spent inside `poll` per call.
48pub const SOURCE_POLL_DURATION_SECONDS: &str = "etl_source_poll_duration_seconds";
49/// Consumer lag; unlabelled series is the max across partitions.
50pub const SOURCE_LAG_RECORDS: &str = "etl_source_lag_records";
51/// Rebalance events observed, labelled by [`L_EVENT`].
52pub const SOURCE_REBALANCES_TOTAL: &str = "etl_source_rebalances_total";
53/// Currently assigned lanes (partitions).
54pub const SOURCE_LANES_ACTIVE: &str = "etl_source_lanes_active";
55
56// Deserializer.
57
58/// Deserialization outputs plus one `error` per failed payload, by
59/// [`L_OUTCOME`].
60pub const DESER_RECORDS_TOTAL: &str = "etl_deser_records_total";
61/// Payloads dropped by the Skip error policy, by [`L_REASON`].
62pub const DESER_RECORDS_DROPPED_TOTAL: &str = "etl_deser_records_dropped_total";
63/// Deserialization time per source batch.
64pub const DESER_BATCH_DURATION_SECONDS: &str = "etl_deser_batch_duration_seconds";
65/// Counter: payload replays awaiting an upstream dependency (e.g. a schema
66/// fetch) — `DeserError::NotReady` occurrences. Not an error and not
67/// backpressure.
68pub const DESER_NOT_READY_TOTAL: &str = "etl_deser_not_ready_total";
69
70// Operators.
71
72/// Records entering the operator.
73pub const OPERATOR_RECORDS_IN_TOTAL: &str = "etl_operator_records_in_total";
74/// Records emitted downstream by the operator.
75pub const OPERATOR_RECORDS_OUT_TOTAL: &str = "etl_operator_records_out_total";
76/// Records intentionally removed, by [`L_REASON`].
77pub const OPERATOR_RECORDS_DROPPED_TOTAL: &str = "etl_operator_records_dropped_total";
78/// User-code errors by [`L_ERROR_TYPE`].
79pub const OPERATOR_ERRORS_TOTAL: &str = "etl_operator_errors_total";
80/// Processing time per batch through this operator.
81pub const OPERATOR_BATCH_DURATION_SECONDS: &str = "etl_operator_batch_duration_seconds";
82
83// Queues.
84
85/// Items currently queued, by [`L_QUEUE`].
86pub const QUEUE_DEPTH: &str = "etl_queue_depth";
87/// Configured queue bound, by [`L_QUEUE`].
88pub const QUEUE_CAPACITY: &str = "etl_queue_capacity";
89/// `try_send` rejections (each is a backpressure signal, never a block).
90pub const QUEUE_FULL_EVENTS_TOTAL: &str = "etl_queue_full_events_total";
91
92// Backpressure.
93
94/// 1 while the source is paused by the watermark controller.
95pub const BACKPRESSURE_PAUSED: &str = "etl_backpressure_paused";
96/// Cumulative paused time in seconds. Monotonically increasing; exported
97/// as a gauge because the `metrics` counter type is integer-only.
98pub const BACKPRESSURE_PAUSED_SECONDS_TOTAL: &str = "etl_backpressure_paused_seconds_total";
99/// Pause transitions (flapping indicator when high).
100pub const BACKPRESSURE_PAUSE_EVENTS_TOTAL: &str = "etl_backpressure_pause_events_total";
101/// Current global in-flight byte budget usage.
102pub const BACKPRESSURE_INFLIGHT_BYTES: &str = "etl_backpressure_inflight_bytes";
103
104// Sink.
105
106/// Records durably written (acknowledged flushes only), by [`L_SHARD`].
107pub const SINK_RECORDS_TOTAL: &str = "etl_sink_records_total";
108/// Bytes durably written, by [`L_SHARD`].
109pub const SINK_BYTES_TOTAL: &str = "etl_sink_bytes_total";
110/// Rows per sealed batch.
111pub const SINK_BATCH_ROWS: &str = "etl_sink_batch_rows";
112/// Bytes per sealed batch.
113pub const SINK_BATCH_BYTES: &str = "etl_sink_batch_bytes";
114/// Flushes by trigger, by [`L_SHARD`] and [`L_REASON`].
115pub const SINK_FLUSHES_TOTAL: &str = "etl_sink_flushes_total";
116/// Write round-trip per flush including retries, by [`L_SHARD`].
117pub const SINK_FLUSH_DURATION_SECONDS: &str = "etl_sink_flush_duration_seconds";
118/// Flush attempts beyond the first, by [`L_SHARD`].
119pub const SINK_RETRIES_TOTAL: &str = "etl_sink_retries_total";
120/// Write errors, by [`L_SHARD`] and [`L_ERROR_TYPE`].
121pub const SINK_ERRORS_TOTAL: &str = "etl_sink_errors_total";
122/// Sealed batches currently in flight, by [`L_SHARD`].
123pub const SINK_INFLIGHT_BATCHES: &str = "etl_sink_inflight_batches";
124/// 1 = circuit closed, 0 = open, by [`L_SHARD`] and [`L_REPLICA`].
125pub const SINK_REPLICA_HEALTHY: &str = "etl_sink_replica_healthy";
126/// Circuit-breaker open transitions, by [`L_SHARD`] and [`L_REPLICA`].
127pub const SINK_BREAKER_OPENS_TOTAL: &str = "etl_sink_breaker_opens_total";
128/// Batches abandoned at the drain deadline (replayed after restart).
129pub const SINK_ABANDONED_BATCHES_TOTAL: &str = "etl_sink_abandoned_batches_total";
130
131// Checkpointing.
132
133/// Unacknowledged batches tracked; unlabelled series is the max across
134/// partitions.
135pub const CHECKPOINT_PENDING_BATCHES: &str = "etl_checkpoint_pending_batches";
136/// Source commit calls, by [`L_OUTCOME`].
137pub const CHECKPOINT_COMMITS_TOTAL: &str = "etl_checkpoint_commits_total";
138/// Commit round-trip time.
139pub const CHECKPOINT_COMMIT_DURATION_SECONDS: &str = "etl_checkpoint_commit_duration_seconds";
140/// Age of the oldest unacknowledged batch — the primary "stuck pipeline"
141/// alert signal.
142pub const CHECKPOINT_WATERMARK_AGE_SECONDS: &str = "etl_checkpoint_watermark_age_seconds";
143
144// End to end.
145
146/// Source-to-durable-write latency, observed per acknowledged batch.
147pub const E2E_LATENCY_SECONDS: &str = "etl_e2e_latency_seconds";
148
149// Pipeline.
150
151/// Constant 1; carries build metadata via [`L_VERSION`].
152pub const PIPELINE_INFO: &str = "etl_pipeline_info";
153/// 1 for the current state, 0 otherwise, by [`L_STATE`].
154pub const PIPELINE_STATE: &str = "etl_pipeline_state";
155/// Pinned pipeline thread count.
156pub const PIPELINE_THREADS: &str = "etl_pipeline_threads";
157
158/// Every counter name (must end in `_total`).
159pub const COUNTERS: &[&str] = &[
160    SOURCE_RECORDS_TOTAL,
161    SOURCE_BYTES_TOTAL,
162    SOURCE_REBALANCES_TOTAL,
163    DESER_RECORDS_TOTAL,
164    DESER_RECORDS_DROPPED_TOTAL,
165    DESER_NOT_READY_TOTAL,
166    OPERATOR_RECORDS_IN_TOTAL,
167    OPERATOR_RECORDS_OUT_TOTAL,
168    OPERATOR_RECORDS_DROPPED_TOTAL,
169    OPERATOR_ERRORS_TOTAL,
170    QUEUE_FULL_EVENTS_TOTAL,
171    BACKPRESSURE_PAUSE_EVENTS_TOTAL,
172    SINK_RECORDS_TOTAL,
173    SINK_BYTES_TOTAL,
174    SINK_FLUSHES_TOTAL,
175    SINK_RETRIES_TOTAL,
176    SINK_ERRORS_TOTAL,
177    SINK_BREAKER_OPENS_TOTAL,
178    SINK_ABANDONED_BATCHES_TOTAL,
179    CHECKPOINT_COMMITS_TOTAL,
180];
181
182/// Every gauge name.
183pub const GAUGES: &[&str] = &[
184    SOURCE_LAG_RECORDS,
185    SOURCE_LANES_ACTIVE,
186    QUEUE_DEPTH,
187    QUEUE_CAPACITY,
188    BACKPRESSURE_PAUSED,
189    BACKPRESSURE_PAUSED_SECONDS_TOTAL,
190    BACKPRESSURE_INFLIGHT_BYTES,
191    SINK_INFLIGHT_BATCHES,
192    SINK_REPLICA_HEALTHY,
193    CHECKPOINT_PENDING_BATCHES,
194    CHECKPOINT_WATERMARK_AGE_SECONDS,
195    PIPELINE_INFO,
196    PIPELINE_STATE,
197    PIPELINE_THREADS,
198];
199
200/// Every histogram name (must carry a unit suffix).
201pub const HISTOGRAMS: &[&str] = &[
202    SOURCE_POLL_DURATION_SECONDS,
203    DESER_BATCH_DURATION_SECONDS,
204    OPERATOR_BATCH_DURATION_SECONDS,
205    SINK_BATCH_ROWS,
206    SINK_BATCH_BYTES,
207    SINK_FLUSH_DURATION_SECONDS,
208    CHECKPOINT_COMMIT_DURATION_SECONDS,
209    E2E_LATENCY_SECONDS,
210];
211
212#[cfg(test)]
213mod tests {
214    use super::*;
215    use std::collections::HashSet;
216
217    #[test]
218    fn names_follow_prometheus_conventions() {
219        for name in COUNTERS {
220            assert!(
221                name.ends_with("_total"),
222                "counter `{name}` must end in _total"
223            );
224        }
225        for name in HISTOGRAMS {
226            assert!(
227                name.ends_with("_seconds") || name.ends_with("_rows") || name.ends_with("_bytes"),
228                "histogram `{name}` must carry a unit suffix"
229            );
230        }
231        for name in GAUGES {
232            // One documented exception: the paused-time accumulator keeps
233            // its counter-style name but is exported as a gauge because the
234            // facade's counter type is integer-only.
235            if *name == BACKPRESSURE_PAUSED_SECONDS_TOTAL {
236                continue;
237            }
238            assert!(
239                !name.ends_with("_total"),
240                "gauge `{name}` must not end in _total"
241            );
242        }
243    }
244
245    #[test]
246    fn names_are_prefixed_and_unique() {
247        let all: Vec<&str> = COUNTERS
248            .iter()
249            .chain(GAUGES)
250            .chain(HISTOGRAMS)
251            .copied()
252            .collect();
253        let unique: HashSet<&str> = all.iter().copied().collect();
254        assert_eq!(unique.len(), all.len(), "duplicate metric name");
255        for name in &all {
256            assert!(name.starts_with("etl_"), "`{name}` must be etl_-prefixed");
257        }
258    }
259}
etl_core/metrics/names.rs

etl_core/metrics/
names.rs
