spate-core 0.1.0

Engine for the Spate framework: records, operator chains, source/sink abstractions, checkpointing, backpressure, config, metrics, and the pipeline runtime. Applications should depend on the `spate` facade crate instead.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99

//! Operator-stage handles (`spate_operator_*`).

use super::labels::ComponentLabels;
use super::names;
use crate::error::ErrorClass;
use metrics::{Counter, Histogram};
use std::time::Duration;

/// Operator-stage handles (`spate_operator_*`).
#[derive(Debug)]
pub struct OperatorMetrics {
    records_in: Counter,
    records_out: Counter,
    dropped_filtered: Counter,
    dropped_skip: Counter,
    dropped_unrouted: Counter,
    err_retryable: Counter,
    err_record: Counter,
    err_fatal: Counter,
    batch_duration: Histogram,
}

impl OperatorMetrics {
    /// Resolve all operator handles.
    pub fn new(labels: &ComponentLabels) -> Self {
        OperatorMetrics {
            records_in: labels.counter(names::OPERATOR_RECORDS_IN_TOTAL),
            records_out: labels.counter(names::OPERATOR_RECORDS_OUT_TOTAL),
            dropped_filtered: labels.counter1(
                names::OPERATOR_RECORDS_DROPPED_TOTAL,
                names::L_REASON,
                "filtered",
            ),
            dropped_skip: labels.counter1(
                names::OPERATOR_RECORDS_DROPPED_TOTAL,
                names::L_REASON,
                "skip_policy",
            ),
            dropped_unrouted: labels.counter1(
                names::OPERATOR_RECORDS_DROPPED_TOTAL,
                names::L_REASON,
                "unrouted",
            ),
            err_retryable: labels.counter1(
                names::OPERATOR_ERRORS_TOTAL,
                names::L_ERROR_TYPE,
                ErrorClass::Retryable.label(),
            ),
            err_record: labels.counter1(
                names::OPERATOR_ERRORS_TOTAL,
                names::L_ERROR_TYPE,
                ErrorClass::RecordLevel.label(),
            ),
            err_fatal: labels.counter1(
                names::OPERATOR_ERRORS_TOTAL,
                names::L_ERROR_TYPE,
                ErrorClass::Fatal.label(),
            ),
            batch_duration: labels.histogram(names::OPERATOR_BATCH_DURATION_SECONDS),
        }
    }

    /// Record one processed batch.
    #[inline]
    pub fn batch(&self, records_in: u64, records_out: u64, d: Duration) {
        self.records_in.increment(records_in);
        self.records_out.increment(records_out);
        self.batch_duration.record(d.as_secs_f64());
    }

    /// Count records removed by a predicate.
    #[inline]
    pub fn filtered(&self, n: u64) {
        self.dropped_filtered.increment(n);
    }

    /// Count records dropped by the Skip error policy.
    #[inline]
    pub fn skipped(&self, n: u64) {
        self.dropped_skip.increment(n);
    }

    /// Count records dropped because they matched no split-sink branch and
    /// the split's `unmatched` policy is `Skip`.
    #[inline]
    pub fn unrouted(&self, n: u64) {
        self.dropped_unrouted.increment(n);
    }

    /// Count user-code errors of one taxonomy class.
    #[inline]
    pub fn errors(&self, class: ErrorClass, n: u64) {
        match class {
            ErrorClass::Retryable => self.err_retryable.increment(n),
            ErrorClass::RecordLevel => self.err_record.increment(n),
            ErrorClass::Fatal => self.err_fatal.increment(n),
        }
    }
}