spate_core/metrics/
operator.rs

1//! Operator-stage handles (`spate_operator_*`).
2
3use super::labels::ComponentLabels;
4use super::names;
5use crate::error::ErrorClass;
6use metrics::{Counter, Histogram};
7use std::time::Duration;
8
9/// Operator-stage handles (`spate_operator_*`).
10///
11/// `error_type` names what the stage did with an error. A stage carries no
12/// error class of its own, so the retryable class belongs to
13/// [`SinkShardMetrics`](super::SinkShardMetrics).
14#[derive(Debug)]
15pub struct OperatorMetrics {
16    records_in: Counter,
17    records_out: Counter,
18    dropped_filtered: Counter,
19    dropped_skip: Counter,
20    dropped_unrouted: Counter,
21    err_record: Counter,
22    err_fatal: Counter,
23    batch_duration: Histogram,
24}
25
26impl OperatorMetrics {
27    /// Resolve all operator handles.
28    pub fn new(labels: &ComponentLabels) -> Self {
29        OperatorMetrics {
30            records_in: labels.counter(names::OPERATOR_RECORDS_IN_TOTAL),
31            records_out: labels.counter(names::OPERATOR_RECORDS_OUT_TOTAL),
32            dropped_filtered: labels.counter1(
33                names::OPERATOR_RECORDS_DROPPED_TOTAL,
34                names::L_REASON,
35                "filtered",
36            ),
37            dropped_skip: labels.counter1(
38                names::OPERATOR_RECORDS_DROPPED_TOTAL,
39                names::L_REASON,
40                "skip_policy",
41            ),
42            dropped_unrouted: labels.counter1(
43                names::OPERATOR_RECORDS_DROPPED_TOTAL,
44                names::L_REASON,
45                "unrouted",
46            ),
47            err_record: labels.counter1(
48                names::OPERATOR_ERRORS_TOTAL,
49                names::L_ERROR_TYPE,
50                ErrorClass::RecordLevel.label(),
51            ),
52            err_fatal: labels.counter1(
53                names::OPERATOR_ERRORS_TOTAL,
54                names::L_ERROR_TYPE,
55                ErrorClass::Fatal.label(),
56            ),
57            batch_duration: labels.histogram(names::OPERATOR_BATCH_DURATION_SECONDS),
58        }
59    }
60
61    /// Record one processed batch.
62    #[inline]
63    pub fn batch(&self, records_in: u64, records_out: u64, d: Duration) {
64        self.records_in.increment(records_in);
65        self.records_out.increment(records_out);
66        self.batch_duration.record(d.as_secs_f64());
67    }
68
69    /// Count records removed by a predicate.
70    #[inline]
71    pub fn filtered(&self, n: u64) {
72        self.dropped_filtered.increment(n);
73    }
74
75    /// Count records dropped by the Skip error policy.
76    #[inline]
77    pub fn skipped(&self, n: u64) {
78        self.dropped_skip.increment(n);
79    }
80
81    /// Count records dropped because they matched no split-sink branch and
82    /// the split's `unmatched` policy is `Skip`.
83    #[inline]
84    pub fn unrouted(&self, n: u64) {
85        self.dropped_unrouted.increment(n);
86    }
87
88    /// Count records whose failure the Skip error policy dropped.
89    #[inline]
90    pub fn record_errors(&self, n: u64) {
91        self.err_record.increment(n);
92    }
93
94    /// Count one failure that stopped the pipeline.
95    #[inline]
96    pub fn fatal_error(&self) {
97        self.err_fatal.increment(1);
98    }
99}
spate_core/metrics/operator.rs

spate_core/metrics/
operator.rs
