mz_storage_operators/
metrics.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10//! Metrics for storage dataflow operators
11
12use std::sync::atomic::{AtomicU64, Ordering};
13
14use mz_ore::metrics::{IntCounter, UIntGauge};
15
16/// Metric handles for one instance of the `backpressure` operator.
17///
18/// The series behind the handles belong to whoever constructed them and stay
19/// registered for as long as that owner keeps them: the persist client's
20/// process-level backpressure series for `persist_source`, or a source's own
21/// per-worker series for upsert. Several operator instances may share one
22/// gauge, which is why the gauge handle is a [GaugeContribution].
23#[derive(Debug)]
24pub struct BackpressureOperatorMetrics {
25    /// Bytes this operator emitted.
26    pub emitted_bytes: IntCounter,
27    /// This operator's share of the gauge: the inflight bytes it most recently
28    /// stalled on.
29    pub last_backpressured_bytes: GaugeContribution,
30    /// Bytes retired by processing downstream of this operator.
31    pub retired_bytes: IntCounter,
32}
33
34impl BackpressureOperatorMetrics {
35    pub fn new(
36        emitted_bytes: IntCounter,
37        last_backpressured_bytes: UIntGauge,
38        retired_bytes: IntCounter,
39    ) -> Self {
40        BackpressureOperatorMetrics {
41            emitted_bytes,
42            last_backpressured_bytes: GaugeContribution::new(last_backpressured_bytes),
43            retired_bytes,
44        }
45    }
46}
47
48/// One contributor's share of a gauge that reads as the sum over all
49/// contributors. Dropping the contribution withdraws it.
50#[derive(Debug)]
51pub struct GaugeContribution {
52    gauge: UIntGauge,
53    contributed: AtomicU64,
54}
55
56impl GaugeContribution {
57    pub fn new(gauge: UIntGauge) -> Self {
58        GaugeContribution {
59            gauge,
60            contributed: AtomicU64::new(0),
61        }
62    }
63
64    /// Replaces this contributor's share with `value`.
65    pub fn set(&self, value: u64) {
66        let previous = self.contributed.swap(value, Ordering::AcqRel);
67        // Add before subtracting so the shared total never dips below the sum
68        // of the other contributions.
69        self.gauge.add(value);
70        self.gauge.sub(previous);
71    }
72}
73
74impl Drop for GaugeContribution {
75    fn drop(&mut self) {
76        self.gauge.sub(self.contributed.load(Ordering::Acquire));
77    }
78}
79
80#[cfg(test)]
81mod tests {
82    use mz_ore::metrics::UIntGauge;
83
84    use super::GaugeContribution;
85
86    #[mz_ore::test]
87    fn gauge_contribution_sums_live_shares_and_withdraws_on_drop() {
88        let gauge = UIntGauge::new("gauge", "help").expect("valid metric");
89        let a = GaugeContribution::new(gauge.clone());
90        let b = GaugeContribution::new(gauge.clone());
91
92        a.set(5);
93        b.set(3);
94        assert_eq!(gauge.get(), 8);
95
96        a.set(2);
97        assert_eq!(gauge.get(), 5);
98
99        drop(a);
100        assert_eq!(gauge.get(), 3);
101
102        b.set(0);
103        assert_eq!(gauge.get(), 0);
104    }
105}
mz_storage_operators/metrics.rs

mz_storage_operators/
metrics.rs
