mz_cluster_client/
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 shared by both compute and storage.
11
12use mz_ore::cast::CastLossy;
13use mz_ore::metric;
14use mz_ore::metrics::{
15    CounterVec, DeleteOnDropCounter, DeleteOnDropGauge, GaugeVec, IntCounterVec, MetricTag,
16    MetricVisibility, MetricsRegistry,
17};
18use mz_ore::stats::SlidingMinMax;
19use prometheus::core::{AtomicF64, AtomicU64};
20
21/// Controller metrics.
22#[derive(Debug, Clone)]
23pub struct ControllerMetrics {
24    dataflow_wallclock_lag_seconds: GaugeVec,
25    dataflow_wallclock_lag_seconds_sum: CounterVec,
26    dataflow_wallclock_lag_seconds_count: IntCounterVec,
27}
28
29impl ControllerMetrics {
30    /// Create a metrics instance registered into the given registry.
31    pub fn new(metrics_registry: &MetricsRegistry) -> Self {
32        Self {
33            // The next three metrics immitate a summary metric type. The `prometheus` crate lacks
34            // support for summaries, so we roll our own. Note that we also only expose the 0- and
35            // the 1-quantile, i.e., minimum and maximum lag values.
36            dataflow_wallclock_lag_seconds: metrics_registry.register(metric!(
37                name: "mz_dataflow_wallclock_lag_seconds",
38                help: "A summary of the second-by-second lag of the dataflow frontier relative \
39                       to wallclock time, aggregated over the last minute.",
40                var_labels: ["instance_id", "replica_id", "collection_id", "quantile"],
41                visibility: MetricVisibility::Public,
42                tags: [MetricTag::Compute, MetricTag::Source, MetricTag::Sink],
43            )),
44            dataflow_wallclock_lag_seconds_sum: metrics_registry.register(metric!(
45                name: "mz_dataflow_wallclock_lag_seconds_sum",
46                help: "The total sum of dataflow wallclock lag measurements.",
47                var_labels: ["instance_id", "replica_id", "collection_id"],
48            )),
49            dataflow_wallclock_lag_seconds_count: metrics_registry.register(metric!(
50                name: "mz_dataflow_wallclock_lag_seconds_count",
51                help: "The total count of dataflow wallclock lag measurements.",
52                var_labels: ["instance_id", "replica_id", "collection_id"],
53            )),
54        }
55    }
56
57    /// Return an object that tracks wallclock lag metrics for the given collection on the given
58    /// cluster and replica.
59    pub fn wallclock_lag_metrics(
60        &self,
61        collection_id: String,
62        instance_id: Option<String>,
63        replica_id: Option<String>,
64    ) -> WallclockLagMetrics {
65        let labels = vec![
66            instance_id.unwrap_or_default(),
67            replica_id.unwrap_or_default(),
68            collection_id,
69        ];
70
71        let labels_with_quantile = |quantile: &str| {
72            labels
73                .iter()
74                .cloned()
75                .chain([quantile.to_string()])
76                .collect()
77        };
78
79        let wallclock_lag_seconds_min = self
80            .dataflow_wallclock_lag_seconds
81            .get_delete_on_drop_metric(labels_with_quantile("0"));
82        let wallclock_lag_seconds_max = self
83            .dataflow_wallclock_lag_seconds
84            .get_delete_on_drop_metric(labels_with_quantile("1"));
85        let wallclock_lag_seconds_sum = self
86            .dataflow_wallclock_lag_seconds_sum
87            .get_delete_on_drop_metric(labels.clone());
88        let wallclock_lag_seconds_count = self
89            .dataflow_wallclock_lag_seconds_count
90            .get_delete_on_drop_metric(labels);
91        let wallclock_lag_minmax = SlidingMinMax::new(60);
92
93        WallclockLagMetrics {
94            wallclock_lag_seconds_min,
95            wallclock_lag_seconds_max,
96            wallclock_lag_seconds_sum,
97            wallclock_lag_seconds_count,
98            wallclock_lag_minmax,
99        }
100    }
101}
102
103/// Metrics tracking frontier wallclock lag for a collection.
104#[derive(Debug)]
105pub struct WallclockLagMetrics {
106    /// Gauge tracking minimum dataflow wallclock lag.
107    wallclock_lag_seconds_min: DeleteOnDropGauge<AtomicF64, Vec<String>>,
108    /// Gauge tracking maximum dataflow wallclock lag.
109    wallclock_lag_seconds_max: DeleteOnDropGauge<AtomicF64, Vec<String>>,
110    /// Counter tracking the total sum of dataflow wallclock lag.
111    wallclock_lag_seconds_sum: DeleteOnDropCounter<AtomicF64, Vec<String>>,
112    /// Counter tracking the total count of dataflow wallclock lag measurements.
113    wallclock_lag_seconds_count: DeleteOnDropCounter<AtomicU64, Vec<String>>,
114
115    /// State maintaining minimum and maximum wallclock lag.
116    wallclock_lag_minmax: SlidingMinMax<f32>,
117}
118
119impl WallclockLagMetrics {
120    /// Observe a new wallclock lag measurement.
121    pub fn observe(&mut self, lag_secs: u64) {
122        let lag_secs = f32::cast_lossy(lag_secs);
123
124        self.wallclock_lag_minmax.add_sample(lag_secs);
125
126        let (&min, &max) = self
127            .wallclock_lag_minmax
128            .get()
129            .expect("just added a sample");
130
131        self.wallclock_lag_seconds_min.set(min.into());
132        self.wallclock_lag_seconds_max.set(max.into());
133        self.wallclock_lag_seconds_sum.inc_by(lag_secs.into());
134        self.wallclock_lag_seconds_count.inc();
135    }
136}
mz_cluster_client/metrics.rs

mz_cluster_client/
metrics.rs
