etl_core/metrics/
handles.rs

1//! Pre-registered metric handle structs, one per pipeline stage.
2//!
3//! Handles are resolved once at pipeline build time; the record loop only
4//! touches `Counter`/`Gauge`/`Histogram` handles — never name or label
5//! lookups. Methods take **per-batch aggregates**, enforcing the hot-path
6//! counting discipline by API shape (see `docs/METRICS.md`).
7
8use super::E2eBasis;
9use super::names;
10use crate::error::ErrorClass;
11use crate::record::PartitionId;
12use metrics::{Counter, Gauge, Histogram, SharedString, counter, gauge, histogram};
13use std::collections::HashMap;
14use std::sync::Mutex;
15use std::time::Duration;
16
17/// The standard label set attached to every framework metric.
18#[derive(Clone, Debug)]
19pub struct ComponentLabels {
20    /// Pipeline name.
21    pub pipeline: SharedString,
22    /// Component instance id from config/builder (e.g. `orders_kafka`).
23    pub component: SharedString,
24    /// Component implementation (e.g. `kafka`, `clickhouse`, `map`).
25    pub component_type: SharedString,
26}
27
28impl ComponentLabels {
29    /// Build the standard label set.
30    pub fn new(
31        pipeline: impl Into<SharedString>,
32        component: impl Into<SharedString>,
33        component_type: impl Into<SharedString>,
34    ) -> Self {
35        ComponentLabels {
36            pipeline: pipeline.into(),
37            component: component.into(),
38            component_type: component_type.into(),
39        }
40    }
41
42    fn counter(&self, name: &'static str) -> Counter {
43        counter!(name,
44            names::L_PIPELINE => self.pipeline.clone(),
45            names::L_COMPONENT => self.component.clone(),
46            names::L_COMPONENT_TYPE => self.component_type.clone(),
47        )
48    }
49
50    fn counter1(&self, name: &'static str, k: &'static str, v: impl Into<SharedString>) -> Counter {
51        counter!(name,
52            names::L_PIPELINE => self.pipeline.clone(),
53            names::L_COMPONENT => self.component.clone(),
54            names::L_COMPONENT_TYPE => self.component_type.clone(),
55            k => v.into(),
56        )
57    }
58
59    fn counter2(
60        &self,
61        name: &'static str,
62        k1: &'static str,
63        v1: impl Into<SharedString>,
64        k2: &'static str,
65        v2: impl Into<SharedString>,
66    ) -> Counter {
67        counter!(name,
68            names::L_PIPELINE => self.pipeline.clone(),
69            names::L_COMPONENT => self.component.clone(),
70            names::L_COMPONENT_TYPE => self.component_type.clone(),
71            k1 => v1.into(),
72            k2 => v2.into(),
73        )
74    }
75
76    fn gauge(&self, name: &'static str) -> Gauge {
77        gauge!(name,
78            names::L_PIPELINE => self.pipeline.clone(),
79            names::L_COMPONENT => self.component.clone(),
80            names::L_COMPONENT_TYPE => self.component_type.clone(),
81        )
82    }
83
84    fn gauge1(&self, name: &'static str, k: &'static str, v: impl Into<SharedString>) -> Gauge {
85        gauge!(name,
86            names::L_PIPELINE => self.pipeline.clone(),
87            names::L_COMPONENT => self.component.clone(),
88            names::L_COMPONENT_TYPE => self.component_type.clone(),
89            k => v.into(),
90        )
91    }
92
93    fn gauge2(
94        &self,
95        name: &'static str,
96        k1: &'static str,
97        v1: impl Into<SharedString>,
98        k2: &'static str,
99        v2: impl Into<SharedString>,
100    ) -> Gauge {
101        gauge!(name,
102            names::L_PIPELINE => self.pipeline.clone(),
103            names::L_COMPONENT => self.component.clone(),
104            names::L_COMPONENT_TYPE => self.component_type.clone(),
105            k1 => v1.into(),
106            k2 => v2.into(),
107        )
108    }
109
110    fn histogram(&self, name: &'static str) -> Histogram {
111        histogram!(name,
112            names::L_PIPELINE => self.pipeline.clone(),
113            names::L_COMPONENT => self.component.clone(),
114            names::L_COMPONENT_TYPE => self.component_type.clone(),
115        )
116    }
117
118    fn histogram1(
119        &self,
120        name: &'static str,
121        k: &'static str,
122        v: impl Into<SharedString>,
123    ) -> Histogram {
124        histogram!(name,
125            names::L_PIPELINE => self.pipeline.clone(),
126            names::L_COMPONENT => self.component.clone(),
127            names::L_COMPONENT_TYPE => self.component_type.clone(),
128            k => v.into(),
129        )
130    }
131}
132
133impl ErrorClass {
134    fn label(self) -> &'static str {
135        match self {
136            ErrorClass::Retryable => "retryable",
137            ErrorClass::RecordLevel => "record_level",
138            ErrorClass::Fatal => "fatal",
139        }
140    }
141}
142
143/// Dynamic per-partition gauge family, gated by `per_partition_detail`.
144/// Registration happens on the control plane (rebalance/commit paths), so a
145/// mutex is acceptable; the hot path never touches this.
146#[derive(Debug)]
147struct PartitionGauges {
148    name: &'static str,
149    labels: ComponentLabels,
150    gauges: Mutex<HashMap<u32, Gauge>>,
151}
152
153impl PartitionGauges {
154    fn set(&self, partition: PartitionId, value: f64) {
155        let mut gauges = self.gauges.lock().expect("partition gauge lock");
156        gauges
157            .entry(partition.0)
158            .or_insert_with(|| {
159                self.labels
160                    .gauge1(self.name, names::L_PARTITION, partition.0.to_string())
161            })
162            .set(value);
163    }
164
165    /// Drops handles for revoked partitions so they are no longer updated
166    /// and don't accumulate across rebalances. The exporter may keep
167    /// rendering the last value of a dropped series until its own idle
168    /// timeout; that staleness is harmless and expected.
169    fn retain(&self, keep: &[PartitionId]) {
170        let mut gauges = self.gauges.lock().expect("partition gauge lock");
171        gauges.retain(|p, _| keep.iter().any(|k| k.0 == *p));
172    }
173}
174
175/// Source-stage handles (`etl_source_*`).
176#[derive(Debug)]
177pub struct SourceMetrics {
178    records: Counter,
179    bytes: Counter,
180    poll_duration: Histogram,
181    lag_max: Gauge,
182    rebalance_assign: Counter,
183    rebalance_revoke: Counter,
184    lanes_active: Gauge,
185    partition_lag: Option<PartitionGauges>,
186}
187
188impl SourceMetrics {
189    /// Resolve all source handles. `per_partition_detail` gates the
190    /// cardinality-sensitive per-partition lag series.
191    pub fn new(labels: &ComponentLabels, per_partition_detail: bool) -> Self {
192        SourceMetrics {
193            records: labels.counter(names::SOURCE_RECORDS_TOTAL),
194            bytes: labels.counter(names::SOURCE_BYTES_TOTAL),
195            poll_duration: labels.histogram(names::SOURCE_POLL_DURATION_SECONDS),
196            lag_max: labels.gauge(names::SOURCE_LAG_RECORDS),
197            rebalance_assign: labels.counter1(
198                names::SOURCE_REBALANCES_TOTAL,
199                names::L_EVENT,
200                "assign",
201            ),
202            rebalance_revoke: labels.counter1(
203                names::SOURCE_REBALANCES_TOTAL,
204                names::L_EVENT,
205                "revoke",
206            ),
207            lanes_active: labels.gauge(names::SOURCE_LANES_ACTIVE),
208            partition_lag: per_partition_detail.then(|| PartitionGauges {
209                name: names::SOURCE_LAG_RECORDS,
210                labels: labels.clone(),
211                gauges: Mutex::new(HashMap::new()),
212            }),
213        }
214    }
215
216    /// Record one polled batch.
217    #[inline]
218    pub fn batch(&self, records: u64, bytes: u64) {
219        self.records.increment(records);
220        self.bytes.increment(bytes);
221    }
222
223    /// Observe one `poll` call's duration.
224    #[inline]
225    pub fn poll_duration(&self, d: Duration) {
226        self.poll_duration.record(d.as_secs_f64());
227    }
228
229    /// Set the max consumer lag across partitions.
230    pub fn set_lag_max(&self, lag: u64) {
231        self.lag_max.set(lag as f64);
232    }
233
234    /// Set one partition's lag. No-op unless `per_partition_detail`.
235    pub fn set_partition_lag(&self, partition: PartitionId, lag: u64) {
236        if let Some(pg) = &self.partition_lag {
237            pg.set(partition, lag as f64);
238        }
239    }
240
241    /// Drop per-partition series for revoked partitions.
242    pub fn retain_partitions(&self, keep: &[PartitionId]) {
243        if let Some(pg) = &self.partition_lag {
244            pg.retain(keep);
245        }
246    }
247
248    /// Count a rebalance assignment event.
249    pub fn rebalance_assigned(&self) {
250        self.rebalance_assign.increment(1);
251    }
252
253    /// Count a rebalance revocation event.
254    pub fn rebalance_revoked(&self) {
255        self.rebalance_revoke.increment(1);
256    }
257
258    /// Set the number of currently assigned lanes.
259    pub fn set_lanes_active(&self, lanes: usize) {
260        self.lanes_active.set(lanes as f64);
261    }
262}
263
264/// Deserializer-stage handles (`etl_deser_*`).
265#[derive(Debug)]
266pub struct DeserMetrics {
267    ok: Counter,
268    errors: Counter,
269    dropped_skip: Counter,
270    not_ready: Counter,
271    batch_duration: Histogram,
272}
273
274impl DeserMetrics {
275    /// Resolve all deserializer handles.
276    pub fn new(labels: &ComponentLabels) -> Self {
277        DeserMetrics {
278            ok: labels.counter1(names::DESER_RECORDS_TOTAL, names::L_OUTCOME, "ok"),
279            errors: labels.counter1(names::DESER_RECORDS_TOTAL, names::L_OUTCOME, "error"),
280            dropped_skip: labels.counter1(
281                names::DESER_RECORDS_DROPPED_TOTAL,
282                names::L_REASON,
283                "skip_policy",
284            ),
285            not_ready: labels.counter(names::DESER_NOT_READY_TOTAL),
286            batch_duration: labels.histogram(names::DESER_BATCH_DURATION_SECONDS),
287        }
288    }
289
290    /// Record one deserialized batch: emitted records, failed payloads,
291    /// and time spent.
292    #[inline]
293    pub fn batch(&self, ok: u64, errors: u64, d: Duration) {
294        self.ok.increment(ok);
295        if errors > 0 {
296            self.errors.increment(errors);
297        }
298        self.batch_duration.record(d.as_secs_f64());
299    }
300
301    /// Count payloads dropped by the Skip policy.
302    #[inline]
303    pub fn dropped(&self, n: u64) {
304        self.dropped_skip.increment(n);
305    }
306
307    /// Count not-ready replays (payloads waiting on an upstream
308    /// dependency such as a schema fetch).
309    #[inline]
310    pub fn not_ready(&self, n: u64) {
311        self.not_ready.increment(n);
312    }
313}
314
315/// Operator-stage handles (`etl_operator_*`).
316#[derive(Debug)]
317pub struct OperatorMetrics {
318    records_in: Counter,
319    records_out: Counter,
320    dropped_filtered: Counter,
321    dropped_skip: Counter,
322    err_retryable: Counter,
323    err_record: Counter,
324    err_fatal: Counter,
325    batch_duration: Histogram,
326}
327
328impl OperatorMetrics {
329    /// Resolve all operator handles.
330    pub fn new(labels: &ComponentLabels) -> Self {
331        OperatorMetrics {
332            records_in: labels.counter(names::OPERATOR_RECORDS_IN_TOTAL),
333            records_out: labels.counter(names::OPERATOR_RECORDS_OUT_TOTAL),
334            dropped_filtered: labels.counter1(
335                names::OPERATOR_RECORDS_DROPPED_TOTAL,
336                names::L_REASON,
337                "filtered",
338            ),
339            dropped_skip: labels.counter1(
340                names::OPERATOR_RECORDS_DROPPED_TOTAL,
341                names::L_REASON,
342                "skip_policy",
343            ),
344            err_retryable: labels.counter1(
345                names::OPERATOR_ERRORS_TOTAL,
346                names::L_ERROR_TYPE,
347                ErrorClass::Retryable.label(),
348            ),
349            err_record: labels.counter1(
350                names::OPERATOR_ERRORS_TOTAL,
351                names::L_ERROR_TYPE,
352                ErrorClass::RecordLevel.label(),
353            ),
354            err_fatal: labels.counter1(
355                names::OPERATOR_ERRORS_TOTAL,
356                names::L_ERROR_TYPE,
357                ErrorClass::Fatal.label(),
358            ),
359            batch_duration: labels.histogram(names::OPERATOR_BATCH_DURATION_SECONDS),
360        }
361    }
362
363    /// Record one processed batch.
364    #[inline]
365    pub fn batch(&self, records_in: u64, records_out: u64, d: Duration) {
366        self.records_in.increment(records_in);
367        self.records_out.increment(records_out);
368        self.batch_duration.record(d.as_secs_f64());
369    }
370
371    /// Count records removed by a predicate.
372    #[inline]
373    pub fn filtered(&self, n: u64) {
374        self.dropped_filtered.increment(n);
375    }
376
377    /// Count records dropped by the Skip error policy.
378    #[inline]
379    pub fn skipped(&self, n: u64) {
380        self.dropped_skip.increment(n);
381    }
382
383    /// Count user-code errors of one taxonomy class.
384    #[inline]
385    pub fn errors(&self, class: ErrorClass, n: u64) {
386        match class {
387            ErrorClass::Retryable => self.err_retryable.increment(n),
388            ErrorClass::RecordLevel => self.err_record.increment(n),
389            ErrorClass::Fatal => self.err_fatal.increment(n),
390        }
391    }
392}
393
394/// Queue-edge handles (`etl_queue_*`).
395#[derive(Debug)]
396pub struct QueueMetrics {
397    depth: Gauge,
398    full_events: Counter,
399}
400
401impl QueueMetrics {
402    /// Resolve handles for one queue edge (e.g. `chain->sink/shard-3`) and
403    /// publish its configured capacity.
404    pub fn new(labels: &ComponentLabels, queue: &str, capacity: usize) -> Self {
405        let queue: SharedString = queue.to_owned().into();
406        labels
407            .gauge1(names::QUEUE_CAPACITY, names::L_QUEUE, queue.clone())
408            .set(capacity as f64);
409        QueueMetrics {
410            depth: labels.gauge1(names::QUEUE_DEPTH, names::L_QUEUE, queue.clone()),
411            full_events: labels.counter1(names::QUEUE_FULL_EVENTS_TOTAL, names::L_QUEUE, queue),
412        }
413    }
414
415    /// Set the current queue depth.
416    #[inline]
417    pub fn set_depth(&self, depth: usize) {
418        self.depth.set(depth as f64);
419    }
420
421    /// Count `try_send` rejections.
422    #[inline]
423    pub fn full_events(&self, n: u64) {
424        self.full_events.increment(n);
425    }
426}
427
428/// Backpressure handles (`etl_backpressure_*`).
429#[derive(Debug)]
430pub struct BackpressureMetrics {
431    paused: Gauge,
432    paused_seconds: Gauge,
433    pause_events: Counter,
434    inflight_bytes: Gauge,
435}
436
437impl BackpressureMetrics {
438    /// Resolve all backpressure handles.
439    pub fn new(labels: &ComponentLabels) -> Self {
440        BackpressureMetrics {
441            paused: labels.gauge(names::BACKPRESSURE_PAUSED),
442            paused_seconds: labels.gauge(names::BACKPRESSURE_PAUSED_SECONDS_TOTAL),
443            pause_events: labels.counter(names::BACKPRESSURE_PAUSE_EVENTS_TOTAL),
444            inflight_bytes: labels.gauge(names::BACKPRESSURE_INFLIGHT_BYTES),
445        }
446    }
447
448    /// Record a pause transition.
449    pub fn pause_started(&self) {
450        self.paused.set(1.0);
451        self.pause_events.increment(1);
452    }
453
454    /// Record a resume transition and the time spent paused.
455    pub fn pause_ended(&self, paused_for: Duration) {
456        self.paused.set(0.0);
457        // Monotonic accumulator; gauge because the facade counter is
458        // integer-only (see names.rs).
459        self.paused_seconds.increment(paused_for.as_secs_f64());
460    }
461
462    /// Set the current in-flight byte budget usage.
463    #[inline]
464    pub fn set_inflight_bytes(&self, bytes: usize) {
465        self.inflight_bytes.set(bytes as f64);
466    }
467}
468
469/// Why a sink batch was sealed and flushed.
470#[derive(Clone, Copy, Debug, PartialEq, Eq)]
471pub enum FlushReason {
472    /// `max_rows` reached.
473    Rows,
474    /// `max_bytes` reached.
475    Bytes,
476    /// Linger deadline expired.
477    Linger,
478    /// Drain (shutdown or revocation) forced the seal.
479    Drain,
480}
481
482impl FlushReason {
483    fn label(self) -> &'static str {
484        match self {
485            FlushReason::Rows => "rows",
486            FlushReason::Bytes => "bytes",
487            FlushReason::Linger => "linger",
488            FlushReason::Drain => "drain",
489        }
490    }
491}
492
493/// Per-replica handles inside one shard.
494#[derive(Debug)]
495struct ReplicaMetrics {
496    healthy: Gauge,
497    breaker_opens: Counter,
498}
499
500/// Sink-shard handles (`etl_sink_*`), one struct per shard worker.
501#[derive(Debug)]
502pub struct SinkShardMetrics {
503    records: Counter,
504    bytes: Counter,
505    batch_rows: Histogram,
506    batch_bytes: Histogram,
507    flush_rows: Counter,
508    flush_bytes: Counter,
509    flush_linger: Counter,
510    flush_drain: Counter,
511    flush_duration: Histogram,
512    retries: Counter,
513    err_retryable: Counter,
514    err_record: Counter,
515    err_fatal: Counter,
516    inflight: Gauge,
517    abandoned: Counter,
518    e2e: Histogram,
519    e2e_basis: E2eBasis,
520    replicas: Vec<ReplicaMetrics>,
521}
522
523impl SinkShardMetrics {
524    /// Resolve all handles for one shard. `replicas` are display names used
525    /// as the `replica` label (bounded by cluster topology). `e2e_basis`
526    /// selects the time base for `etl_e2e_latency_seconds` (see
527    /// `docs/METRICS.md`).
528    ///
529    /// Call **after** [`install`](crate::metrics::install): handles bind to
530    /// the recorder present at construction, and a handle built before the
531    /// exporter exists silently records into the void.
532    pub fn new(
533        labels: &ComponentLabels,
534        shard: u32,
535        replicas: &[String],
536        e2e_basis: E2eBasis,
537    ) -> Self {
538        let shard: SharedString = shard.to_string().into();
539        let replicas = replicas
540            .iter()
541            .map(|replica| {
542                let m = ReplicaMetrics {
543                    healthy: labels.gauge2(
544                        names::SINK_REPLICA_HEALTHY,
545                        names::L_SHARD,
546                        shard.clone(),
547                        names::L_REPLICA,
548                        replica.clone(),
549                    ),
550                    breaker_opens: labels.counter2(
551                        names::SINK_BREAKER_OPENS_TOTAL,
552                        names::L_SHARD,
553                        shard.clone(),
554                        names::L_REPLICA,
555                        replica.clone(),
556                    ),
557                };
558                m.healthy.set(1.0);
559                m
560            })
561            .collect();
562        SinkShardMetrics {
563            records: labels.counter1(names::SINK_RECORDS_TOTAL, names::L_SHARD, shard.clone()),
564            bytes: labels.counter1(names::SINK_BYTES_TOTAL, names::L_SHARD, shard.clone()),
565            batch_rows: labels.histogram(names::SINK_BATCH_ROWS),
566            batch_bytes: labels.histogram(names::SINK_BATCH_BYTES),
567            flush_rows: labels.counter2(
568                names::SINK_FLUSHES_TOTAL,
569                names::L_SHARD,
570                shard.clone(),
571                names::L_REASON,
572                FlushReason::Rows.label(),
573            ),
574            flush_bytes: labels.counter2(
575                names::SINK_FLUSHES_TOTAL,
576                names::L_SHARD,
577                shard.clone(),
578                names::L_REASON,
579                FlushReason::Bytes.label(),
580            ),
581            flush_linger: labels.counter2(
582                names::SINK_FLUSHES_TOTAL,
583                names::L_SHARD,
584                shard.clone(),
585                names::L_REASON,
586                FlushReason::Linger.label(),
587            ),
588            flush_drain: labels.counter2(
589                names::SINK_FLUSHES_TOTAL,
590                names::L_SHARD,
591                shard.clone(),
592                names::L_REASON,
593                FlushReason::Drain.label(),
594            ),
595            flush_duration: labels.histogram1(
596                names::SINK_FLUSH_DURATION_SECONDS,
597                names::L_SHARD,
598                shard.clone(),
599            ),
600            retries: labels.counter1(names::SINK_RETRIES_TOTAL, names::L_SHARD, shard.clone()),
601            err_retryable: labels.counter2(
602                names::SINK_ERRORS_TOTAL,
603                names::L_SHARD,
604                shard.clone(),
605                names::L_ERROR_TYPE,
606                ErrorClass::Retryable.label(),
607            ),
608            err_record: labels.counter2(
609                names::SINK_ERRORS_TOTAL,
610                names::L_SHARD,
611                shard.clone(),
612                names::L_ERROR_TYPE,
613                ErrorClass::RecordLevel.label(),
614            ),
615            err_fatal: labels.counter2(
616                names::SINK_ERRORS_TOTAL,
617                names::L_SHARD,
618                shard.clone(),
619                names::L_ERROR_TYPE,
620                ErrorClass::Fatal.label(),
621            ),
622            inflight: labels.gauge1(names::SINK_INFLIGHT_BATCHES, names::L_SHARD, shard.clone()),
623            abandoned: labels.counter1(names::SINK_ABANDONED_BATCHES_TOTAL, names::L_SHARD, shard),
624            e2e: labels.histogram(names::E2E_LATENCY_SECONDS),
625            e2e_basis,
626            replicas,
627        }
628    }
629
630    /// Observe end-to-end latency for one durably written batch, from its
631    /// oldest record. `ingest_age` is time since that record entered the
632    /// terminal stage; `oldest_event_ms` is its source event time. The
633    /// configured basis picks which one lands in the histogram (event
634    /// basis falls back to ingest when no event time was available).
635    #[inline]
636    pub fn e2e_observed(&self, ingest_age: Duration, oldest_event_ms: i64) {
637        let latency = match self.e2e_basis {
638            E2eBasis::Event if oldest_event_ms != i64::MAX => {
639                let now_ms = std::time::SystemTime::now()
640                    .duration_since(std::time::UNIX_EPOCH)
641                    .map(|d| i64::try_from(d.as_millis()).unwrap_or(i64::MAX))
642                    .unwrap_or(0);
643                Duration::from_millis(u64::try_from(now_ms - oldest_event_ms).unwrap_or(0))
644            }
645            _ => ingest_age,
646        };
647        self.e2e.record(latency.as_secs_f64());
648    }
649
650    /// Record one durably acknowledged flush.
651    #[inline]
652    pub fn flushed(&self, reason: FlushReason, rows: u64, bytes: u64, d: Duration) {
653        self.records.increment(rows);
654        self.bytes.increment(bytes);
655        self.batch_rows.record(rows as f64);
656        self.batch_bytes.record(bytes as f64);
657        self.flush_duration.record(d.as_secs_f64());
658        match reason {
659            FlushReason::Rows => self.flush_rows.increment(1),
660            FlushReason::Bytes => self.flush_bytes.increment(1),
661            FlushReason::Linger => self.flush_linger.increment(1),
662            FlushReason::Drain => self.flush_drain.increment(1),
663        }
664    }
665
666    /// Count flush attempts beyond the first.
667    #[inline]
668    pub fn retries(&self, n: u64) {
669        self.retries.increment(n);
670    }
671
672    /// Count write errors of one taxonomy class.
673    #[inline]
674    pub fn errors(&self, class: ErrorClass, n: u64) {
675        match class {
676            ErrorClass::Retryable => self.err_retryable.increment(n),
677            ErrorClass::RecordLevel => self.err_record.increment(n),
678            ErrorClass::Fatal => self.err_fatal.increment(n),
679        }
680    }
681
682    /// Set the number of sealed batches currently in flight.
683    #[inline]
684    pub fn set_inflight(&self, batches: usize) {
685        self.inflight.set(batches as f64);
686    }
687
688    /// Mark one replica healthy (circuit closed) or quarantined (open).
689    pub fn set_replica_healthy(&self, replica: usize, healthy: bool) {
690        if let Some(r) = self.replicas.get(replica) {
691            r.healthy.set(if healthy { 1.0 } else { 0.0 });
692        }
693    }
694
695    /// Count a circuit-breaker open transition on one replica.
696    pub fn breaker_opened(&self, replica: usize) {
697        if let Some(r) = self.replicas.get(replica) {
698            r.breaker_opens.increment(1);
699        }
700    }
701
702    /// Count batches abandoned at the drain deadline.
703    pub fn abandoned(&self, n: u64) {
704        self.abandoned.increment(n);
705    }
706}
707
708/// Checkpointer handles (`etl_checkpoint_*` and end-to-end latency).
709#[derive(Debug)]
710pub struct CheckpointMetrics {
711    pending_max: Gauge,
712    commits_ok: Counter,
713    commits_err: Counter,
714    commit_duration: Histogram,
715    watermark_age: Gauge,
716    partition_pending: Option<PartitionGauges>,
717}
718
719impl CheckpointMetrics {
720    /// Resolve all checkpointer handles.
721    pub fn new(labels: &ComponentLabels, per_partition_detail: bool) -> Self {
722        CheckpointMetrics {
723            pending_max: labels.gauge(names::CHECKPOINT_PENDING_BATCHES),
724            commits_ok: labels.counter1(names::CHECKPOINT_COMMITS_TOTAL, names::L_OUTCOME, "ok"),
725            commits_err: labels.counter1(
726                names::CHECKPOINT_COMMITS_TOTAL,
727                names::L_OUTCOME,
728                "error",
729            ),
730            commit_duration: labels.histogram(names::CHECKPOINT_COMMIT_DURATION_SECONDS),
731            watermark_age: labels.gauge(names::CHECKPOINT_WATERMARK_AGE_SECONDS),
732            partition_pending: per_partition_detail.then(|| PartitionGauges {
733                name: names::CHECKPOINT_PENDING_BATCHES,
734                labels: labels.clone(),
735                gauges: Mutex::new(HashMap::new()),
736            }),
737        }
738    }
739
740    /// Set the max pending-batch count across partitions.
741    pub fn set_pending_max(&self, pending: usize) {
742        self.pending_max.set(pending as f64);
743    }
744
745    /// Set one partition's pending count. No-op unless
746    /// `per_partition_detail`.
747    pub fn set_partition_pending(&self, partition: PartitionId, pending: usize) {
748        if let Some(pg) = &self.partition_pending {
749            pg.set(partition, pending as f64);
750        }
751    }
752
753    /// Drop per-partition series for revoked partitions.
754    pub fn retain_partitions(&self, keep: &[PartitionId]) {
755        if let Some(pg) = &self.partition_pending {
756            pg.retain(keep);
757        }
758    }
759
760    /// Record one source commit call.
761    pub fn commit(&self, ok: bool, d: Duration) {
762        if ok {
763            self.commits_ok.increment(1);
764        } else {
765            self.commits_err.increment(1);
766        }
767        self.commit_duration.record(d.as_secs_f64());
768    }
769
770    /// Set the age of the oldest unacknowledged batch.
771    pub fn set_watermark_age(&self, age: Duration) {
772        self.watermark_age.set(age.as_secs_f64());
773    }
774}
775
776/// Lifecycle state of the pipeline, exported via `etl_pipeline_state`.
777#[derive(Clone, Copy, Debug, PartialEq, Eq)]
778pub enum PipelineState {
779    /// Starting up: connecting, awaiting assignment.
780    Starting,
781    /// Processing records.
782    Running,
783    /// Draining after SIGTERM or a full revocation.
784    Draining,
785    /// Failed; the process will exit non-zero.
786    Failed,
787}
788
789/// Pipeline-level handles (`etl_pipeline_*`).
790#[derive(Debug)]
791pub struct PipelineMetrics {
792    starting: Gauge,
793    running: Gauge,
794    draining: Gauge,
795    failed: Gauge,
796    threads: Gauge,
797}
798
799impl PipelineMetrics {
800    /// Resolve pipeline handles and publish the info series.
801    pub fn new(labels: &ComponentLabels, version: &str) -> Self {
802        labels
803            .gauge1(names::PIPELINE_INFO, names::L_VERSION, version.to_owned())
804            .set(1.0);
805        let state = |s: &'static str| labels.gauge1(names::PIPELINE_STATE, names::L_STATE, s);
806        let m = PipelineMetrics {
807            starting: state("starting"),
808            running: state("running"),
809            draining: state("draining"),
810            failed: state("failed"),
811            threads: labels.gauge(names::PIPELINE_THREADS),
812        };
813        m.set_state(PipelineState::Starting);
814        m
815    }
816
817    /// Flip the state gauges so exactly the current state reads 1.
818    pub fn set_state(&self, state: PipelineState) {
819        self.starting.set(if state == PipelineState::Starting {
820            1.0
821        } else {
822            0.0
823        });
824        self.running.set(if state == PipelineState::Running {
825            1.0
826        } else {
827            0.0
828        });
829        self.draining.set(if state == PipelineState::Draining {
830            1.0
831        } else {
832            0.0
833        });
834        self.failed.set(if state == PipelineState::Failed {
835            1.0
836        } else {
837            0.0
838        });
839    }
840
841    /// Publish the pinned pipeline thread count.
842    pub fn set_threads(&self, threads: usize) {
843        self.threads.set(threads as f64);
844    }
845}
etl_core/metrics/handles.rs

etl_core/metrics/
handles.rs
