mz_metrics/
lib.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License in the LICENSE file at the
6// root of this repository, or online at
7//
8//     http://www.apache.org/licenses/LICENSE-2.0
9//
10// Unless required by applicable law or agreed to in writing, software
11// distributed under the License is distributed on an "AS IS" BASIS,
12// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13// See the License for the specific language governing permissions and
14// limitations under the License.
15
16//! Internal metrics libraries for Materialize.
17
18#![warn(missing_docs, missing_debug_implementations)]
19
20use std::path::PathBuf;
21use std::time::Duration;
22
23use mz_dyncfg::{ConfigSet, ConfigUpdates};
24use mz_ore::metrics::MetricsRegistry;
25use tokio::time::Interval;
26
27pub use dyncfgs::all_dyncfgs;
28
29mod dyncfgs;
30pub mod lgalloc;
31pub mod rusage;
32pub mod usage;
33
34/// Handle to metrics defined in this crate.
35#[derive(Debug)]
36pub struct Metrics {
37    config_set: ConfigSet,
38    lgalloc: MetricsTask,
39    lgalloc_map: MetricsTask,
40    rusage: MetricsTask,
41    usage: MetricsTask,
42}
43
44static METRICS: std::sync::Mutex<Option<Metrics>> = std::sync::Mutex::new(None);
45
46/// Register all metrics into the provided registry.
47///
48/// We do not recommend calling this function multiple times. It is safe to call this function,
49/// but it might delete previous metrics. If we ever want to change this, we should
50/// remove the shared static mutex and make this function return a handle to the metrics.
51///
52/// This function is async, because it needs to be called from a tokio runtime context.
53///
54/// `disk_root` is a directory whose filesystem usage should be tracked, or `None` for processes
55/// that do not use disk.
56#[allow(clippy::unused_async)]
57pub async fn register_metrics_into(
58    metrics_registry: &MetricsRegistry,
59    config_set: ConfigSet,
60    disk_root: Option<PathBuf>,
61) {
62    let update_duration_metric = metrics_registry.register(mz_ore::metric!(
63        name: "mz_metrics_update_duration",
64        help: "The time it took to update lgalloc stats",
65        var_labels: ["name"],
66        buckets: mz_ore::stats::histogram_seconds_buckets(0.000_500, 32.),
67    ));
68
69    let lgalloc = Metrics::new_metrics_task(
70        metrics_registry,
71        lgalloc::register_metrics_into,
72        dyncfgs::MZ_METRICS_LGALLOC_REFRESH_INTERVAL,
73        &update_duration_metric,
74    );
75    let lgalloc_map = Metrics::new_metrics_task(
76        metrics_registry,
77        lgalloc::register_map_metrics_into,
78        dyncfgs::MZ_METRICS_LGALLOC_MAP_REFRESH_INTERVAL,
79        &update_duration_metric,
80    );
81    let rusage = Metrics::new_metrics_task(
82        metrics_registry,
83        rusage::register_metrics_into,
84        dyncfgs::MZ_METRICS_RUSAGE_REFRESH_INTERVAL,
85        &update_duration_metric,
86    );
87
88    let usage = Metrics::new_metrics_task(
89        metrics_registry,
90        |registry| usage::register_metrics_into(registry, disk_root),
91        dyncfgs::MZ_METRICS_USAGE_REFRESH_INTERVAL,
92        &update_duration_metric,
93    );
94
95    *METRICS.lock().expect("lock poisoned") = Some(Metrics {
96        lgalloc,
97        lgalloc_map,
98        rusage,
99        usage,
100        config_set,
101    });
102}
103
104/// Returns the `(name, help, labels, source)` of every metric this crate
105/// registers through a `metric!`-wrapping macro (lgalloc and rusage).
106///
107/// The metrics catalog (`mz-metrics-catalog`) builds the user-facing metrics
108/// docs by scraping `metric!` invocations out of the source with `syn`. These
109/// metrics are invisible to that scraper because their names are assembled at
110/// macro-expansion time, so the catalog
111/// imports them from here instead: it registers them into a throwaway registry
112/// and reads their descriptors back out.
113pub fn describe_metrics() -> Vec<(String, String, Vec<String>, &'static str)> {
114    let registry = MetricsRegistry::new();
115
116    let tag = |descs: Vec<(String, String, Vec<String>)>, src: &'static str| {
117        descs
118            .into_iter()
119            .map(move |(name, help, labels)| (name, help, labels, src))
120    };
121
122    let mut out = Vec::new();
123    out.extend(tag(
124        lgalloc::register_metrics_into(&registry).descs(),
125        lgalloc::SOURCE,
126    ));
127    out.extend(tag(
128        lgalloc::register_map_metrics_into(&registry).descs(),
129        lgalloc::SOURCE,
130    ));
131    out.extend(tag(
132        rusage::register_metrics_into(&registry).descs(),
133        rusage::SOURCE,
134    ));
135    out
136}
137
138/// Extracts a metric's label keys from its Prometheus descriptor
139pub(crate) fn desc_labels(desc: &prometheus::core::Desc) -> Vec<String> {
140    let mut labels: Vec<String> = desc
141        .variable_labels
142        .iter()
143        .cloned()
144        .chain(desc.const_label_pairs.iter().map(|p| p.name().to_owned()))
145        .collect();
146    labels.sort();
147    labels.dedup();
148    labels
149}
150
151/// Update the configuration of the metrics.
152pub fn update_dyncfg(config_updates: &ConfigUpdates) {
153    if let Some(metrics) = METRICS.lock().expect("lock poisoned").as_mut() {
154        metrics.apply_dyncfg_updates(config_updates);
155    }
156}
157
158impl Metrics {
159    /// Update the dynamic configuration.
160    pub fn apply_dyncfg_updates(&mut self, config_updates: &ConfigUpdates) {
161        // Update the config set.
162        config_updates.apply(&self.config_set);
163        // Notify tasks about updated configuration.
164        self.lgalloc.update_dyncfg(&self.config_set);
165        self.lgalloc_map.update_dyncfg(&self.config_set);
166        self.rusage.update_dyncfg(&self.config_set);
167        self.usage.update_dyncfg(&self.config_set);
168    }
169
170    fn new_metrics_task<T: MetricsUpdate>(
171        metrics_registry: &MetricsRegistry,
172        constructor: impl FnOnce(&MetricsRegistry) -> T,
173        interval_config: mz_dyncfg::Config<Duration>,
174        update_duration_metric: &mz_ore::metrics::HistogramVec,
175    ) -> MetricsTask {
176        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
177
178        // Start disabled.
179        let mut interval: Option<Interval> = None;
180
181        let update_duration_metric =
182            update_duration_metric.get_delete_on_drop_metric(&[T::NAME][..]);
183
184        let mut metrics = constructor(metrics_registry);
185
186        let mut update_metrics = move || {
187            tracing::debug!(metrics = T::NAME, "updating metrics");
188            let start = std::time::Instant::now();
189            if let Err(err) = metrics.update() {
190                tracing::error!(metrics = T::NAME, ?err, "metrics update failed");
191            }
192            let elapsed = start.elapsed();
193            update_duration_metric.observe(elapsed.as_secs_f64());
194        };
195
196        let update_interval = |new_interval, interval: &mut Option<Interval>| {
197            // Zero duration disables ticking.
198            if new_interval == Duration::ZERO {
199                *interval = None;
200                return;
201            }
202            // Prevent no-op changes.
203            if Some(new_interval) == interval.as_ref().map(Interval::period) {
204                return;
205            }
206            tracing::debug!(metrics = T::NAME, ?new_interval, "updating interval");
207            let mut new_interval = tokio::time::interval(new_interval);
208            new_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
209            *interval = Some(new_interval);
210        };
211
212        mz_ore::task::spawn(|| format!("mz_metrics_update({})", T::NAME), async move {
213            loop {
214                tokio::select! {
215                    _ = async {
216                        interval.as_mut().unwrap().tick().await
217                    }, if interval.is_some() => {
218                        update_metrics()
219                    }
220                    new_interval = rx.recv() => match new_interval {
221                        Some(new_interval) => update_interval(new_interval, &mut interval),
222                        None => break,
223                    }
224                }
225            }
226        });
227
228        MetricsTask {
229            tx,
230            interval_config,
231        }
232    }
233}
234
235/// Behavior to update metrics.
236pub trait MetricsUpdate: Send + Sync + 'static {
237    /// Error type to indicate updating failed.
238    type Error: std::fmt::Debug;
239    /// A human-readable name.
240    const NAME: &'static str;
241    /// Update the metrics.
242    fn update(&mut self) -> Result<(), Self::Error>;
243}
244
245#[derive(Debug)]
246struct MetricsTask {
247    interval_config: mz_dyncfg::Config<Duration>,
248    tx: tokio::sync::mpsc::UnboundedSender<Duration>,
249}
250
251impl MetricsTask {
252    pub(crate) fn update_dyncfg(&self, config_set: &ConfigSet) {
253        self.tx
254            .send(self.interval_config.get(config_set))
255            .expect("Receiver exists");
256    }
257}
mz_metrics/lib.rs

mz_metrics/
lib.rs
