1#![warn(missing_docs, missing_debug_implementations)]
19
20use std::time::Duration;
21
22use mz_dyncfg::{ConfigSet, ConfigUpdates};
23use mz_ore::metrics::MetricsRegistry;
24use tokio::time::Interval;
25
26pub use dyncfgs::all_dyncfgs;
27
28mod dyncfgs;
29pub mod lgalloc;
30pub mod rusage;
31
32#[derive(Debug)]
34pub struct Metrics {
35 config_set: ConfigSet,
36 lgalloc: MetricsTask,
37 lgalloc_map: MetricsTask,
38 rusage: MetricsTask,
39}
40
41static METRICS: std::sync::Mutex<Option<Metrics>> = std::sync::Mutex::new(None);
42
43#[allow(clippy::unused_async)]
51pub async fn register_metrics_into(metrics_registry: &MetricsRegistry, config_set: ConfigSet) {
52 let update_duration_metric = metrics_registry.register(mz_ore::metric!(
53 name: "mz_metrics_update_duration",
54 help: "The time it took to update lgalloc stats",
55 var_labels: ["name"],
56 buckets: mz_ore::stats::histogram_seconds_buckets(0.000_500, 32.),
57 ));
58
59 let lgalloc = Metrics::new_metrics_task(
60 metrics_registry,
61 lgalloc::register_metrics_into,
62 dyncfgs::MZ_METRICS_LGALLOC_REFRESH_INTERVAL,
63 &update_duration_metric,
64 );
65 let lgalloc_map = Metrics::new_metrics_task(
66 metrics_registry,
67 lgalloc::register_map_metrics_into,
68 dyncfgs::MZ_METRICS_LGALLOC_MAP_REFRESH_INTERVAL,
69 &update_duration_metric,
70 );
71 let rusage = Metrics::new_metrics_task(
72 metrics_registry,
73 rusage::register_metrics_into,
74 dyncfgs::MZ_METRICS_RUSAGE_REFRESH_INTERVAL,
75 &update_duration_metric,
76 );
77
78 *METRICS.lock().expect("lock poisoned") = Some(Metrics {
79 lgalloc,
80 lgalloc_map,
81 rusage,
82 config_set,
83 });
84}
85
86pub fn update_dyncfg(config_updates: &ConfigUpdates) {
88 if let Some(metrics) = METRICS.lock().expect("lock poisoned").as_mut() {
89 metrics.apply_dyncfg_updates(config_updates);
90 }
91}
92
93impl Metrics {
94 pub fn apply_dyncfg_updates(&mut self, config_updates: &ConfigUpdates) {
96 config_updates.apply(&self.config_set);
98 self.lgalloc.update_dyncfg(&self.config_set);
100 self.lgalloc_map.update_dyncfg(&self.config_set);
101 self.rusage.update_dyncfg(&self.config_set);
102 }
103
104 fn new_metrics_task<T: MetricsUpdate>(
105 metrics_registry: &MetricsRegistry,
106 constructor: impl FnOnce(&MetricsRegistry) -> T,
107 interval_config: mz_dyncfg::Config<Duration>,
108 update_duration_metric: &mz_ore::metrics::HistogramVec,
109 ) -> MetricsTask {
110 let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
111
112 let mut interval: Option<Interval> = None;
114
115 let update_duration_metric =
116 update_duration_metric.get_delete_on_drop_metric(&[T::NAME][..]);
117
118 let mut metrics = constructor(metrics_registry);
119
120 let mut update_metrics = move || {
121 tracing::debug!(metrics = T::NAME, "updating metrics");
122 let start = std::time::Instant::now();
123 if let Err(err) = metrics.update() {
124 tracing::error!(metrics = T::NAME, ?err, "metrics update failed");
125 }
126 let elapsed = start.elapsed();
127 update_duration_metric.observe(elapsed.as_secs_f64());
128 };
129
130 let update_interval = |new_interval, interval: &mut Option<Interval>| {
131 if new_interval == Duration::ZERO {
133 *interval = None;
134 return;
135 }
136 if Some(new_interval) == interval.as_ref().map(Interval::period) {
138 return;
139 }
140 tracing::debug!(metrics = T::NAME, ?new_interval, "updating interval");
141 let mut new_interval = tokio::time::interval(new_interval);
142 new_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
143 *interval = Some(new_interval);
144 };
145
146 mz_ore::task::spawn(|| format!("mz_metrics_update({})", T::NAME), async move {
147 loop {
148 tokio::select! {
149 _ = async { interval.as_mut().unwrap().tick().await }, if interval.is_some() => {
150 update_metrics()
151 }
152 new_interval = rx.recv() => match new_interval {
153 Some(new_interval) => update_interval(new_interval, &mut interval),
154 None => break,
155 }
156 }
157 }
158 });
159
160 MetricsTask {
161 tx,
162 interval_config,
163 }
164 }
165}
166
167pub trait MetricsUpdate: Send + Sync + 'static {
169 type Error: std::fmt::Debug;
171 const NAME: &'static str;
173 fn update(&mut self) -> Result<(), Self::Error>;
175}
176
177#[derive(Debug)]
178struct MetricsTask {
179 interval_config: mz_dyncfg::Config<Duration>,
180 tx: tokio::sync::mpsc::UnboundedSender<Duration>,
181}
182
183impl MetricsTask {
184 pub(crate) fn update_dyncfg(&self, config_set: &ConfigSet) {
185 self.tx
186 .send(self.interval_config.get(config_set))
187 .expect("Receiver exists");
188 }
189}