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#[derive(Debug)]
11pub struct OperatorMetrics {
12    records_in: Counter,
13    records_out: Counter,
14    dropped_filtered: Counter,
15    dropped_skip: Counter,
16    dropped_unrouted: Counter,
17    err_retryable: Counter,
18    err_record: Counter,
19    err_fatal: Counter,
20    batch_duration: Histogram,
21}
22
23impl OperatorMetrics {
24    /// Resolve all operator handles.
25    pub fn new(labels: &ComponentLabels) -> Self {
26        OperatorMetrics {
27            records_in: labels.counter(names::OPERATOR_RECORDS_IN_TOTAL),
28            records_out: labels.counter(names::OPERATOR_RECORDS_OUT_TOTAL),
29            dropped_filtered: labels.counter1(
30                names::OPERATOR_RECORDS_DROPPED_TOTAL,
31                names::L_REASON,
32                "filtered",
33            ),
34            dropped_skip: labels.counter1(
35                names::OPERATOR_RECORDS_DROPPED_TOTAL,
36                names::L_REASON,
37                "skip_policy",
38            ),
39            dropped_unrouted: labels.counter1(
40                names::OPERATOR_RECORDS_DROPPED_TOTAL,
41                names::L_REASON,
42                "unrouted",
43            ),
44            err_retryable: labels.counter1(
45                names::OPERATOR_ERRORS_TOTAL,
46                names::L_ERROR_TYPE,
47                ErrorClass::Retryable.label(),
48            ),
49            err_record: labels.counter1(
50                names::OPERATOR_ERRORS_TOTAL,
51                names::L_ERROR_TYPE,
52                ErrorClass::RecordLevel.label(),
53            ),
54            err_fatal: labels.counter1(
55                names::OPERATOR_ERRORS_TOTAL,
56                names::L_ERROR_TYPE,
57                ErrorClass::Fatal.label(),
58            ),
59            batch_duration: labels.histogram(names::OPERATOR_BATCH_DURATION_SECONDS),
60        }
61    }
62
63    /// Record one processed batch.
64    #[inline]
65    pub fn batch(&self, records_in: u64, records_out: u64, d: Duration) {
66        self.records_in.increment(records_in);
67        self.records_out.increment(records_out);
68        self.batch_duration.record(d.as_secs_f64());
69    }
70
71    /// Count records removed by a predicate.
72    #[inline]
73    pub fn filtered(&self, n: u64) {
74        self.dropped_filtered.increment(n);
75    }
76
77    /// Count records dropped by the Skip error policy.
78    #[inline]
79    pub fn skipped(&self, n: u64) {
80        self.dropped_skip.increment(n);
81    }
82
83    /// Count records dropped because they matched no split-sink branch and
84    /// the split's `unmatched` policy is `Skip`.
85    #[inline]
86    pub fn unrouted(&self, n: u64) {
87        self.dropped_unrouted.increment(n);
88    }
89
90    /// Count user-code errors of one taxonomy class.
91    #[inline]
92    pub fn errors(&self, class: ErrorClass, n: u64) {
93        match class {
94            ErrorClass::Retryable => self.err_retryable.increment(n),
95            ErrorClass::RecordLevel => self.err_record.increment(n),
96            ErrorClass::Fatal => self.err_fatal.increment(n),
97        }
98    }
99}
spate_core/metrics/operator.rs

spate_core/metrics/
operator.rs
