Skip to main content

mz_compute/
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
10use std::sync::{Arc, Mutex};
11
12use mz_compute_client::metrics::{CommandMetrics, HistoryMetrics};
13use mz_ore::cast::CastFrom;
14use mz_ore::metric;
15use mz_ore::metrics::{
16    MakeCollectorOpts, MetricTag, MetricVisibility, MetricsRegistry, UIntGauge, raw,
17};
18use mz_repr::{GlobalId, SharedRow};
19use prometheus::core::{AtomicF64, GenericCounter};
20use prometheus::proto::LabelPair;
21use prometheus::{Histogram, HistogramVec, IntCounter};
22
23/// Metrics exposed by compute replicas.
24//
25// Most of the metrics here use the `raw` implementations, rather than the `DeleteOnDrop` wrappers
26// because their labels are fixed throughout the lifetime of the replica process. For example, any
27// metric labeled only by `worker_id` can be `raw` since the number of workers cannot change.
28//
29// Metrics that are labelled by a dimension that can change throughout the lifetime of the process
30// (such as `collection_id`) MUST NOT use the `raw` metric types and must use the `DeleteOnDrop`
31// types instead, to avoid memory leaks.
32#[derive(Clone, Debug)]
33pub struct ComputeMetrics {
34    // Optional workload class label to apply to all metrics in registry.
35    workload_class: Arc<Mutex<Option<String>>>,
36
37    // command history
38    history_command_count: raw::UIntGaugeVec,
39    history_dataflow_count: raw::UIntGaugeVec,
40
41    // reconciliation
42    reconciliation_reused_dataflows_count_total: raw::IntCounterVec,
43    reconciliation_replaced_dataflows_count_total: raw::IntCounterVec,
44
45    // arrangements
46    arrangement_maintenance_seconds_total: raw::CounterVec,
47    arrangement_maintenance_active_info: raw::UIntGaugeVec,
48
49    // timings
50    //
51    // Note that this particular metric unfortunately takes some care to
52    // interpret. It measures the duration of step_or_park calls, which
53    // undesirably includes the parking. This is probably fine because we
54    // regularly send progress information through persist sources, which likely
55    // means the parking is capped at a second or two in practice. It also
56    // doesn't do anything to let you pinpoint _which_ operator or worker isn't
57    // yielding, but it should hopefully alert us when there is something to
58    // look at.
59    timely_step_duration_seconds: HistogramVec,
60    persist_peek_seconds: HistogramVec,
61    stashed_peek_seconds: HistogramVec,
62    handle_command_duration_seconds: HistogramVec,
63
64    // Index peek timing phases (per-cluster, no worker label)
65    index_peek_total_seconds: Histogram,
66    index_peek_seek_fulfillment_seconds: Histogram,
67    index_peek_error_scan_seconds: Histogram,
68    index_peek_cursor_setup_seconds: Histogram,
69    index_peek_row_iteration_seconds: Histogram,
70    index_peek_row_iteration_rows: Histogram,
71    index_peek_result_sort_seconds: Histogram,
72    index_peek_result_sort_rows: Histogram,
73    index_peek_frontier_check_seconds: Histogram,
74    index_peek_row_collection_seconds: Histogram,
75
76    // memory usage
77    shared_row_heap_capacity_bytes: raw::UIntGaugeVec,
78
79    // replica expiration
80    replica_expiration_timestamp_seconds: raw::UIntGaugeVec,
81    replica_expiration_remaining_seconds: raw::GaugeVec,
82
83    // collections
84    collection_count: raw::UIntGaugeVec,
85
86    // subscribes
87    subscribe_snapshots_skipped_total: IntCounter,
88}
89
90/// Applies the per-role const label to `opts`, unless `role` is `Solo`.
91///
92/// The two named roles (maintenance, interactive) each get a distinct `role` label so a second
93/// compute runtime in the same process registers a distinct series rather than colliding with the
94/// first. `Solo` omits the label so a single-runtime deployment registers exactly as it did before
95/// a second runtime existed.
96fn with_role(
97    mut opts: MakeCollectorOpts,
98    role: crate::server::ComputeRuntimeRole,
99) -> MakeCollectorOpts {
100    if let Some(label) = role.label() {
101        opts.opts = opts.opts.const_label("role", label);
102    }
103    opts
104}
105
106impl ComputeMetrics {
107    /// Registers the compute metrics for `role` into `registry`.
108    ///
109    /// The two named roles carry a `role` const label so that a second compute runtime in the same
110    /// process registers a distinct series rather than colliding with the first. `Solo` carries no
111    /// such label.
112    pub fn register_with(
113        registry: &MetricsRegistry,
114        role: crate::server::ComputeRuntimeRole,
115    ) -> Self {
116        let workload_class = Arc::new(Mutex::new(None));
117        let mut index_peek_row_buckets =
118            prometheus::exponential_buckets(1.0, 2.0, 25).expect("valid parameters");
119        index_peek_row_buckets.insert(0, 0.0);
120
121        // Apply a `workload_class` label to all metrics in the registry when we
122        // have a known workload class.
123        //
124        // The postprocessor rewrites every metric in the whole registry, so only the maintenance
125        // runtime registers it. A second registration from the interactive runtime would push the
126        // label twice onto each metric and produce a duplicate-label scrape error.
127        if role.owns_process_globals() {
128            registry.register_postprocessor({
129                let workload_class = Arc::clone(&workload_class);
130                move |metrics| {
131                    let workload_class: Option<String> =
132                        workload_class.lock().expect("lock poisoned").clone();
133                    let Some(workload_class) = workload_class else {
134                        return;
135                    };
136                    for metric in metrics {
137                        for metric in metric.mut_metric() {
138                            let mut label = LabelPair::default();
139                            label.set_name("workload_class".into());
140                            label.set_value(workload_class.clone());
141
142                            let mut labels = metric.take_label();
143                            labels.push(label);
144                            metric.set_label(labels);
145                        }
146                    }
147                }
148            });
149        }
150
151        Self {
152            workload_class,
153            history_command_count: registry.register(with_role(metric!(
154                name: "mz_compute_replica_history_command_count",
155                help: "The number of commands in the replica's command history.",
156                var_labels: ["worker_id", "command_type"],
157            ), role)),
158            history_dataflow_count: registry.register(with_role(metric!(
159                name: "mz_compute_replica_history_dataflow_count",
160                help: "The number of dataflows in the replica's command history.",
161                var_labels: ["worker_id"],
162                visibility: MetricVisibility::Public,
163                tags: [MetricTag::Compute],
164            ), role)),
165            reconciliation_reused_dataflows_count_total: registry.register(with_role(metric!(
166                name: "mz_compute_reconciliation_reused_dataflows_count_total",
167                help: "The total number of dataflows that were reused during compute reconciliation.",
168                var_labels: ["worker_id"],
169            ), role)),
170            reconciliation_replaced_dataflows_count_total: registry.register(with_role(metric!(
171                name: "mz_compute_reconciliation_replaced_dataflows_count_total",
172                help: "The total number of dataflows that were replaced during compute reconciliation.",
173                var_labels: ["worker_id", "reason"],
174            ), role)),
175            arrangement_maintenance_seconds_total: registry.register(with_role(metric!(
176                name: "mz_arrangement_maintenance_seconds_total",
177                help: "The total time spent maintaining arrangements.",
178                var_labels: ["worker_id"],
179                visibility: MetricVisibility::Public,
180                tags: [MetricTag::Compute],
181            ), role)),
182            arrangement_maintenance_active_info: registry.register(with_role(metric!(
183                name: "mz_arrangement_maintenance_active_info",
184                help: "Whether maintenance is currently occuring.",
185                var_labels: ["worker_id"],
186            ), role)),
187            timely_step_duration_seconds: registry.register(with_role(metric!(
188                name: "mz_timely_step_duration_seconds",
189                help: "The time spent in each compute step_or_park call",
190                const_labels: {"cluster" => "compute"},
191                var_labels: ["worker_id"],
192                buckets: mz_ore::stats::histogram_seconds_buckets(0.000_128, 32.0),
193            ), role)),
194            shared_row_heap_capacity_bytes: registry.register(with_role(metric!(
195                name: "mz_dataflow_shared_row_heap_capacity_bytes",
196                help: "The heap capacity of the shared row.",
197                var_labels: ["worker_id"],
198            ), role)),
199            persist_peek_seconds: registry.register(with_role(metric!(
200                name: "mz_persist_peek_seconds",
201                help: "Time spent in (experimental) Persist fast-path peeks.",
202                var_labels: ["worker_id"],
203                buckets: mz_ore::stats::histogram_seconds_buckets(0.000_128, 8.0),
204            ), role)),
205            stashed_peek_seconds: registry.register(with_role(metric!(
206                name: "mz_stashed_peek_seconds",
207                help: "Time spent reading a peek result and stashing it in the peek result stash (aka. persist blob).",
208                var_labels: ["worker_id"],
209                buckets: mz_ore::stats::histogram_seconds_buckets(0.000_128, 8.0),
210            ), role)),
211            handle_command_duration_seconds: registry.register(with_role(metric!(
212                name: "mz_cluster_handle_command_duration_seconds",
213                help: "Time spent in handling commands.",
214                const_labels: {"cluster" => "compute"},
215                var_labels: ["worker_id", "command_type"],
216                buckets: mz_ore::stats::histogram_seconds_buckets(0.000_128, 8.0),
217            ), role)),
218            index_peek_total_seconds: registry.register(with_role(metric!(
219                name: "mz_index_peek_total_seconds",
220                help: "Total time processing index peeks, from process_peek entry to response. Excluding peeks that use the peek response stash.",
221                buckets: mz_ore::stats::histogram_seconds_buckets(0.000_128, 8.0),
222            ), role)),
223            index_peek_seek_fulfillment_seconds: registry.register(with_role(metric!(
224                name: "mz_index_peek_seek_fulfillment_seconds",
225                help: "Time in seek_fulfillment method including frontier checks and data collection.",
226                buckets: mz_ore::stats::histogram_seconds_buckets(0.000_128, 8.0),
227            ), role)),
228            index_peek_error_scan_seconds: registry.register(with_role(metric!(
229                name: "mz_index_peek_error_scan_seconds",
230                help: "Time scanning the error trace for errors.",
231                buckets: mz_ore::stats::histogram_seconds_buckets(0.000_128, 8.0),
232            ), role)),
233            index_peek_cursor_setup_seconds: registry.register(with_role(metric!(
234                name: "mz_index_peek_cursor_setup_seconds",
235                help: "Time setting up cursor and literal constraints.",
236                buckets: mz_ore::stats::histogram_seconds_buckets(0.000_128, 8.0),
237            ), role)),
238            index_peek_row_iteration_seconds: registry.register(with_role(metric!(
239                name: "mz_index_peek_row_iteration_seconds",
240                help: "Time iterating rows and evaluating MFP.",
241                buckets: mz_ore::stats::histogram_seconds_buckets(0.000_128, 8.0),
242            ), role)),
243            index_peek_row_iteration_rows: registry.register(with_role(metric!(
244                name: "mz_index_peek_row_iteration_rows",
245                help: "Number of arrangement rows evaluated by the index peek result iterator.",
246                buckets: index_peek_row_buckets.clone(),
247            ), role)),
248            index_peek_result_sort_seconds: registry.register(with_role(metric!(
249                name: "mz_index_peek_result_sort_seconds",
250                help: "Time sorting intermediate results during peek collection.",
251                buckets: mz_ore::stats::histogram_seconds_buckets(0.000_128, 8.0),
252            ), role)),
253            index_peek_result_sort_rows: registry.register(with_role(metric!(
254                name: "mz_index_peek_result_sort_rows",
255                help: "Number of intermediate result rows sorted during peek collection, summed across sort operations.",
256                buckets: index_peek_row_buckets,
257            ), role)),
258            index_peek_frontier_check_seconds: registry.register(with_role(metric!(
259                name: "mz_index_peek_frontier_check_seconds",
260                help: "Time checking trace frontiers.",
261                buckets: mz_ore::stats::histogram_seconds_buckets(0.000_128, 8.0),
262            ), role)),
263            index_peek_row_collection_seconds: registry.register(with_role(metric!(
264                name: "mz_index_peek_row_collection_seconds",
265                help: "Time constructing RowCollection from peek results.",
266                buckets: mz_ore::stats::histogram_seconds_buckets(0.000_128, 8.0),
267            ), role)),
268            replica_expiration_timestamp_seconds: registry.register(with_role(metric!(
269                name: "mz_dataflow_replica_expiration_timestamp_seconds",
270                help: "The replica expiration timestamp in seconds since epoch.",
271                var_labels: ["worker_id"],
272            ), role)),
273            replica_expiration_remaining_seconds: registry.register(with_role(metric!(
274                name: "mz_dataflow_replica_expiration_remaining_seconds",
275                help: "The remaining seconds until replica expiration. Can go negative, can lag behind.",
276                var_labels: ["worker_id"],
277            ), role)),
278            collection_count: registry.register(with_role(metric!(
279                name: "mz_compute_collection_count",
280                help: "The number and hydration status of maintained compute collections.",
281                var_labels: ["worker_id", "type", "hydrated"],
282            ), role)),
283            subscribe_snapshots_skipped_total: registry.register(with_role(metric!(
284                name: "mz_subscribe_snapshots_skipped_total",
285                help: "The number of collection snapshots that were skipped by the subscribe snapshot optimization.",
286            ), role)),
287        }
288    }
289
290    /// Sets the workload class for the compute metrics.
291    pub fn set_workload_class(&self, workload_class: Option<String>) {
292        let mut guard = self.workload_class.lock().expect("lock poisoned");
293        *guard = workload_class
294    }
295
296    pub fn for_worker(&self, worker_id: usize) -> WorkerMetrics {
297        let worker = worker_id.to_string();
298        let arrangement_maintenance_seconds_total = self
299            .arrangement_maintenance_seconds_total
300            .with_label_values(&[&worker]);
301        let arrangement_maintenance_active_info = self
302            .arrangement_maintenance_active_info
303            .with_label_values(&[&worker]);
304        let timely_step_duration_seconds = self
305            .timely_step_duration_seconds
306            .with_label_values(&[&worker]);
307        let persist_peek_seconds = self.persist_peek_seconds.with_label_values(&[&worker]);
308        let stashed_peek_seconds = self.stashed_peek_seconds.with_label_values(&[&worker]);
309        let handle_command_duration_seconds = CommandMetrics::build(|typ| {
310            self.handle_command_duration_seconds
311                .with_label_values(&[worker.as_ref(), typ])
312        });
313        let index_peek_total_seconds = self.index_peek_total_seconds.clone();
314        let index_peek_seek_fulfillment_seconds = self.index_peek_seek_fulfillment_seconds.clone();
315        let index_peek_error_scan_seconds = self.index_peek_error_scan_seconds.clone();
316        let index_peek_cursor_setup_seconds = self.index_peek_cursor_setup_seconds.clone();
317        let index_peek_row_iteration_seconds = self.index_peek_row_iteration_seconds.clone();
318        let index_peek_row_iteration_rows = self.index_peek_row_iteration_rows.clone();
319        let index_peek_result_sort_seconds = self.index_peek_result_sort_seconds.clone();
320        let index_peek_result_sort_rows = self.index_peek_result_sort_rows.clone();
321        let index_peek_frontier_check_seconds = self.index_peek_frontier_check_seconds.clone();
322        let index_peek_row_collection_seconds = self.index_peek_row_collection_seconds.clone();
323        let replica_expiration_timestamp_seconds = self
324            .replica_expiration_timestamp_seconds
325            .with_label_values(&[&worker]);
326        let replica_expiration_remaining_seconds = self
327            .replica_expiration_remaining_seconds
328            .with_label_values(&[&worker]);
329        let shared_row_heap_capacity_bytes = self
330            .shared_row_heap_capacity_bytes
331            .with_label_values(&[&worker]);
332
333        WorkerMetrics {
334            worker_label: worker,
335            metrics: self.clone(),
336            arrangement_maintenance_seconds_total,
337            arrangement_maintenance_active_info,
338            timely_step_duration_seconds,
339            persist_peek_seconds,
340            stashed_peek_seconds,
341            handle_command_duration_seconds,
342            index_peek_total_seconds,
343            index_peek_seek_fulfillment_seconds,
344            index_peek_error_scan_seconds,
345            index_peek_cursor_setup_seconds,
346            index_peek_row_iteration_seconds,
347            index_peek_row_iteration_rows,
348            index_peek_result_sort_seconds,
349            index_peek_result_sort_rows,
350            index_peek_frontier_check_seconds,
351            index_peek_row_collection_seconds,
352            replica_expiration_timestamp_seconds,
353            replica_expiration_remaining_seconds,
354            shared_row_heap_capacity_bytes,
355        }
356    }
357}
358
359/// Per-worker metrics.
360#[derive(Clone, Debug)]
361pub struct WorkerMetrics {
362    worker_label: String,
363    metrics: ComputeMetrics,
364
365    /// The amount of time spent in arrangement maintenance.
366    pub(crate) arrangement_maintenance_seconds_total: GenericCounter<AtomicF64>,
367    /// 1 if this worker is currently doing maintenance.
368    ///
369    /// If maintenance turns out to take a very long time, this will allow us
370    /// to gain a sense that Materialize is stuck on maintenance before the
371    /// maintenance completes
372    pub(crate) arrangement_maintenance_active_info: UIntGauge,
373    /// Histogram of Timely step timings.
374    pub(crate) timely_step_duration_seconds: Histogram,
375    /// Histogram of persist peek durations.
376    pub(crate) persist_peek_seconds: Histogram,
377    /// Histogram of stashed peek durations.
378    pub(crate) stashed_peek_seconds: Histogram,
379    /// Histogram of command handling durations.
380    pub(crate) handle_command_duration_seconds: CommandMetrics<Histogram>,
381    /// Histogram of total index peek durations.
382    pub(crate) index_peek_total_seconds: Histogram,
383    /// Histogram of index peek seek_fulfillment durations.
384    pub(crate) index_peek_seek_fulfillment_seconds: Histogram,
385    /// Histogram of index peek error scan durations.
386    pub(crate) index_peek_error_scan_seconds: Histogram,
387    /// Histogram of index peek cursor setup durations.
388    pub(crate) index_peek_cursor_setup_seconds: Histogram,
389    /// Histogram of index peek row iteration durations.
390    pub(crate) index_peek_row_iteration_seconds: Histogram,
391    /// Histogram of index peek rows processed by the result iterator.
392    pub(crate) index_peek_row_iteration_rows: Histogram,
393    /// Histogram of index peek result sort durations.
394    pub(crate) index_peek_result_sort_seconds: Histogram,
395    /// Histogram of index peek rows sorted across all result sort operations.
396    pub(crate) index_peek_result_sort_rows: Histogram,
397    /// Histogram of index peek frontier check durations.
398    pub(crate) index_peek_frontier_check_seconds: Histogram,
399    /// Histogram of index peek row collection construction durations.
400    pub(crate) index_peek_row_collection_seconds: Histogram,
401    /// The timestamp of replica expiration.
402    pub(crate) replica_expiration_timestamp_seconds: UIntGauge,
403    /// Remaining seconds until replica expiration.
404    pub(crate) replica_expiration_remaining_seconds: raw::Gauge,
405    /// Heap capacity of the shared row.
406    shared_row_heap_capacity_bytes: UIntGauge,
407}
408
409impl WorkerMetrics {
410    pub fn for_history(&self) -> HistoryMetrics<UIntGauge> {
411        let command_counts = CommandMetrics::build(|typ| {
412            self.metrics
413                .history_command_count
414                .with_label_values(&[self.worker_label.as_ref(), typ])
415        });
416        let dataflow_count = self
417            .metrics
418            .history_dataflow_count
419            .with_label_values(&[&self.worker_label]);
420
421        HistoryMetrics {
422            command_counts,
423            dataflow_count,
424        }
425    }
426
427    /// Record the reconciliation result for a single dataflow.
428    ///
429    /// Reconciliation is recorded as successful if the given properties all hold. Otherwise it is
430    /// recorded as unsuccessful, with a reason based on the first property that does not hold.
431    ///
432    /// The properties are:
433    ///  * compatible: The old and new dataflow descriptions are compatible.
434    ///  * uncompacted: Collections currently installed for the dataflow exports have not been
435    ///                 allowed to compact beyond that new dataflow as-of.
436    ///  * subscribe_free: The dataflow does not export a subscribe sink.
437    ///  * copy_to_free: The dataflow does not export a copy-to sink.
438    ///  * dependencies_retained: All local inputs to the dataflow were retained by compute
439    ///                           reconciliation.
440    pub fn record_dataflow_reconciliation(
441        &self,
442        compatible: bool,
443        uncompacted: bool,
444        subscribe_free: bool,
445        copy_to_free: bool,
446        dependencies_retained: bool,
447    ) {
448        if !compatible {
449            self.metrics
450                .reconciliation_replaced_dataflows_count_total
451                .with_label_values(&[self.worker_label.as_ref(), "incompatible"])
452                .inc();
453        } else if !uncompacted {
454            self.metrics
455                .reconciliation_replaced_dataflows_count_total
456                .with_label_values(&[self.worker_label.as_ref(), "compacted"])
457                .inc();
458        } else if !subscribe_free {
459            self.metrics
460                .reconciliation_replaced_dataflows_count_total
461                .with_label_values(&[self.worker_label.as_ref(), "subscribe"])
462                .inc();
463        } else if !copy_to_free {
464            self.metrics
465                .reconciliation_replaced_dataflows_count_total
466                .with_label_values(&[self.worker_label.as_ref(), "copy-to"])
467                .inc();
468        } else if !dependencies_retained {
469            self.metrics
470                .reconciliation_replaced_dataflows_count_total
471                .with_label_values(&[self.worker_label.as_ref(), "dependency"])
472                .inc();
473        } else {
474            self.metrics
475                .reconciliation_reused_dataflows_count_total
476                .with_label_values(&[&self.worker_label])
477                .inc();
478        }
479    }
480
481    /// Record the heap capacity of the shared row.
482    pub fn record_shared_row_metrics(&self) {
483        let binding = SharedRow::get();
484        self.shared_row_heap_capacity_bytes
485            .set(u64::cast_from(binding.byte_capacity()));
486    }
487
488    /// Increase the count of maintained collections.
489    fn inc_collection_count(&self, collection_type: &str, hydrated: bool) {
490        let hydrated = if hydrated { "1" } else { "0" };
491        self.metrics
492            .collection_count
493            .with_label_values(&[self.worker_label.as_ref(), collection_type, hydrated])
494            .inc();
495    }
496
497    /// Decrease the count of maintained collections.
498    fn dec_collection_count(&self, collection_type: &str, hydrated: bool) {
499        let hydrated = if hydrated { "1" } else { "0" };
500        self.metrics
501            .collection_count
502            .with_label_values(&[self.worker_label.as_ref(), collection_type, hydrated])
503            .dec();
504    }
505
506    pub fn inc_subscribe_snapshot_optimization(&self) {
507        self.metrics.subscribe_snapshots_skipped_total.inc()
508    }
509
510    /// Sets the workload class for the compute metrics.
511    pub fn set_workload_class(&self, workload_class: Option<String>) {
512        self.metrics.set_workload_class(workload_class);
513    }
514
515    pub fn for_collection(&self, id: GlobalId) -> CollectionMetrics {
516        CollectionMetrics::new(id, self.clone())
517    }
518}
519
520/// Collection metrics.
521///
522/// Note that these metrics do _not_ have a `collection_id` label. We avoid introducing
523/// per-collection, per-worker metrics because the number of resulting time series would
524/// potentially be huge. Instead we count classes of collections, such as hydrated collections.
525#[derive(Clone, Debug)]
526pub struct CollectionMetrics {
527    metrics: WorkerMetrics,
528    collection_type: &'static str,
529    collection_hydrated: bool,
530}
531
532impl CollectionMetrics {
533    pub fn new(collection_id: GlobalId, metrics: WorkerMetrics) -> Self {
534        let collection_type = match collection_id {
535            GlobalId::System(_) => "system",
536            GlobalId::IntrospectionSourceIndex(_) => "log",
537            GlobalId::User(_) => "user",
538            GlobalId::Transient(_) => "transient",
539            GlobalId::Explain => "explain",
540        };
541        let collection_hydrated = false;
542
543        metrics.inc_collection_count(collection_type, collection_hydrated);
544
545        Self {
546            metrics,
547            collection_type,
548            collection_hydrated,
549        }
550    }
551
552    /// Record this collection as hydration.
553    pub fn record_collection_hydrated(&mut self) {
554        if self.collection_hydrated {
555            return;
556        }
557
558        self.metrics
559            .dec_collection_count(self.collection_type, false);
560        self.metrics
561            .inc_collection_count(self.collection_type, true);
562        self.collection_hydrated = true;
563    }
564}
565
566impl Drop for CollectionMetrics {
567    fn drop(&mut self) {
568        self.metrics
569            .dec_collection_count(self.collection_type, self.collection_hydrated);
570    }
571}
572
573#[cfg(test)]
574mod tests {
575    use std::collections::BTreeSet;
576
577    use mz_ore::metrics::MetricsRegistry;
578
579    use super::ComputeMetrics;
580    use crate::server::ComputeRuntimeRole;
581
582    /// The `Solo` (single-runtime) role registers exactly as compute did before a second runtime
583    /// existed: no metric carries a `role` label, so single-runtime dashboards and exact-match
584    /// alerts are byte-unchanged.
585    #[mz_ore::test]
586    fn solo_runtime_omits_role_label() {
587        let registry = MetricsRegistry::new();
588        let metrics = ComputeMetrics::register_with(&registry, ComputeRuntimeRole::Solo);
589        // Instantiate the per-worker children so the `*Vec` families emit rows to inspect.
590        let _worker = metrics.for_worker(0);
591
592        for family in registry.gather() {
593            for metric in family.get_metric() {
594                for label in metric.get_label() {
595                    assert_ne!(
596                        label.name(),
597                        "role",
598                        "solo metric {} unexpectedly carries a role label",
599                        family.name(),
600                    );
601                }
602            }
603        }
604    }
605
606    /// The two named roles each carry their own `role` label, so two runtimes in one process
607    /// register distinct series rather than colliding. Registering both on one registry also
608    /// exercises the non-collision that lets them coexist.
609    #[mz_ore::test]
610    fn named_roles_carry_distinct_role_label() {
611        let registry = MetricsRegistry::new();
612        let maintenance = ComputeMetrics::register_with(&registry, ComputeRuntimeRole::Maintenance);
613        let interactive = ComputeMetrics::register_with(&registry, ComputeRuntimeRole::Interactive);
614        let _maintenance_worker = maintenance.for_worker(0);
615        let _interactive_worker = interactive.for_worker(0);
616
617        let mut roles = BTreeSet::new();
618        for family in registry.gather() {
619            for metric in family.get_metric() {
620                let role = metric
621                    .get_label()
622                    .iter()
623                    .find(|label| label.name() == "role")
624                    .unwrap_or_else(|| panic!("metric {} missing a role label", family.name()));
625                roles.insert(role.value().to_string());
626            }
627        }
628
629        assert!(roles.contains("maintenance"), "roles seen: {roles:?}");
630        assert!(roles.contains("interactive"), "roles seen: {roles:?}");
631    }
632}