mz_storage/
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 all things storage.
11//!
12//! The structure of this module is designed to make adding new metrics as easy as possible. The
13//! structure and naming conventions are as follows:
14//!
15//! Metrics for X end up in the `x.rs` submodule, unless X fits into one of the existing
16//! submodules. The struct `XMetricsDefs` defines the `CounterVec/GaugeVec/etc`'s that must be
17//! registered with the `MetricsRegistry` to create new metrics. `XMetricsDefs` should be a
18//! sub-field of `StorageMetrics` (or recursively a sub-field). `XMetricsDefs` has a
19//! `register_with` function to create it using a `MetricsRegistry`.
20//!
21//! `XMetrics` contains the actual gauges/counters/etc that are created using `XMetricsDefs`.
22//! Typically these are created with `new` functions that takes a `&XMetricsDefs`, a `GlobalId`,
23//! and a worker id, but sometimes more complex schemes are used, for metrics that are globally
24//! shared, or have some other shape to their labels.
25//!
26//! `StorageMetrics` is the main entry-point to this module, and for convenience, typically
27//! provides a `get_x_metrics` to obtain an `XMetrics` struct. This is to prevent users from
28//! needing to interact with metrics _definitions_ into the code that actually bumps those
29//! metrics.
30
31use mz_ore::metrics::MetricsRegistry;
32use mz_repr::GlobalId;
33
34use crate::statistics::{SinkStatisticsMetricDefs, SourceStatisticsMetricDefs};
35
36pub mod decode;
37pub mod sink;
38pub mod source;
39pub mod upsert;
40
41/// A top-level struct holding all various _definitions_ of all metrics
42/// use by the `mz-storage` crate.
43///
44/// Created by registering it with a `MetricsRegistry`, it also provides helpers
45/// to obtain various _instantiated_ time-series, either per-worker, shared globally,
46/// or some more specific labeling scheme.
47///
48/// This struct can be cloned, and the various definitions are shared.
49#[derive(Clone, Debug)]
50pub struct StorageMetrics {
51    pub(crate) source_defs: source::SourceMetricDefs,
52    pub(crate) decode_defs: decode::DecodeMetricDefs,
53    pub(crate) upsert_defs: upsert::UpsertMetricDefs,
54    pub(crate) upsert_backpressure_defs: upsert::UpsertBackpressureMetricDefs,
55    pub(crate) sink_defs: sink::SinkMetricDefs,
56
57    // Defined in the `statistics` module, as they are kept in sync with
58    // user-facing data.
59    pub(crate) source_statistics: SourceStatisticsMetricDefs,
60    pub(crate) sink_statistics: SinkStatisticsMetricDefs,
61}
62
63impl StorageMetrics {
64    /// Register all metrics with the `MetricsRegistry`.
65    pub fn register_with(registry: &MetricsRegistry) -> Self {
66        Self {
67            source_defs: source::SourceMetricDefs::register_with(registry),
68            decode_defs: decode::DecodeMetricDefs::register_with(registry),
69            upsert_defs: upsert::UpsertMetricDefs::register_with(registry),
70            upsert_backpressure_defs: upsert::UpsertBackpressureMetricDefs::register_with(registry),
71            sink_defs: sink::SinkMetricDefs::register_with(registry),
72            source_statistics: SourceStatisticsMetricDefs::register_with(registry),
73            sink_statistics: SinkStatisticsMetricDefs::register_with(registry),
74        }
75    }
76
77    /// Get the backpressure series for the given id and worker id.
78    pub(crate) fn get_backpressure_metrics(
79        &self,
80        id: GlobalId,
81        index: usize,
82    ) -> upsert::UpsertBackpressureMetrics {
83        upsert::UpsertBackpressureMetrics::new(&self.upsert_backpressure_defs, id, index)
84    }
85
86    /// Get an `UpsertMetrics` for the given id and worker id (and optional
87    /// `UpsertBackpressureMetrics`).
88    pub(crate) fn get_upsert_metrics(
89        &self,
90        id: GlobalId,
91        worker_id: usize,
92        backpressure_metrics: Option<upsert::UpsertBackpressureMetrics>,
93    ) -> upsert::UpsertMetrics {
94        upsert::UpsertMetrics::new(&self.upsert_defs, id, worker_id, backpressure_metrics)
95    }
96
97    /// Get a `SourcePersistSinkMetrics` for the given configuration.
98    pub(crate) fn get_source_persist_sink_metrics(
99        &self,
100        export_id: GlobalId,
101        primary_source_id: GlobalId,
102        worker_id: usize,
103        data_shard: &mz_persist_client::ShardId,
104    ) -> source::SourcePersistSinkMetrics {
105        source::SourcePersistSinkMetrics::new(
106            &self.source_defs.source_defs,
107            export_id,
108            primary_source_id,
109            worker_id,
110            data_shard,
111        )
112    }
113
114    /// Get a `SourceMetrics` for the given id and worker id.
115    pub(crate) fn get_source_metrics(
116        &self,
117        id: GlobalId,
118        worker_id: usize,
119    ) -> source::SourceMetrics {
120        source::SourceMetrics::new(&self.source_defs.source_defs, id, worker_id)
121    }
122
123    /// Get a `PgSourceMetrics` for the given id.
124    pub(crate) fn get_postgres_source_metrics(
125        &self,
126        id: GlobalId,
127    ) -> source::postgres::PgSourceMetrics {
128        source::postgres::PgSourceMetrics::new(&self.source_defs.postgres_defs, id)
129    }
130
131    /// Get a `MySqlSourceMetrics` for the given id.
132    pub(crate) fn get_mysql_source_metrics(
133        &self,
134        id: GlobalId,
135    ) -> source::mysql::MySqlSourceMetrics {
136        source::mysql::MySqlSourceMetrics::new(&self.source_defs.mysql_defs, id)
137    }
138
139    /// Get a `SqlServerSourceMetrics` for the given id.
140    pub(crate) fn get_sql_server_source_metrics(
141        &self,
142        source_id: GlobalId,
143        worker_id: usize,
144    ) -> source::sql_server::SqlServerSourceMetrics {
145        source::sql_server::SqlServerSourceMetrics::new(
146            &self.source_defs.sql_server_defs,
147            source_id,
148            worker_id,
149        )
150    }
151
152    /// Get an `OffsetCommitMetrics` for the given id.
153    pub(crate) fn get_offset_commit_metrics(&self, id: GlobalId) -> source::OffsetCommitMetrics {
154        source::OffsetCommitMetrics::new(&self.source_defs.source_defs, id)
155    }
156
157    /// Get an `KafkaSourceMetrics` for the given configuration.
158    pub(crate) fn get_kafka_source_metrics(
159        &self,
160        ids: Vec<i32>,
161        topic: String,
162        source_id: GlobalId,
163    ) -> source::kafka::KafkaSourceMetrics {
164        source::kafka::KafkaSourceMetrics::new(
165            &self.source_defs.kafka_source_defs,
166            ids,
167            topic,
168            source_id,
169        )
170    }
171
172    /// Get an `KafkaSinkMetrics` for the given configuration.
173    pub(crate) fn get_kafka_sink_metrics(
174        &self,
175        sink_id: GlobalId,
176    ) -> sink::kafka::KafkaSinkMetrics {
177        sink::kafka::KafkaSinkMetrics::new(&self.sink_defs.kafka_defs, sink_id)
178    }
179
180    /// Get an `IcebergSinkMetrics` for the given configuration.
181    pub(crate) fn get_iceberg_sink_metrics(
182        &self,
183        sink_id: GlobalId,
184        worker_id: usize,
185    ) -> sink::iceberg::IcebergSinkMetrics {
186        sink::iceberg::IcebergSinkMetrics::new(&self.sink_defs.iceberg_defs, sink_id, worker_id)
187    }
188}
mz_storage/metrics.rs

mz_storage/
metrics.rs
