1use 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#[derive(Clone, Debug)]
33pub struct ComputeMetrics {
34 workload_class: Arc<Mutex<Option<String>>>,
36
37 history_command_count: raw::UIntGaugeVec,
39 history_dataflow_count: raw::UIntGaugeVec,
40
41 reconciliation_reused_dataflows_count_total: raw::IntCounterVec,
43 reconciliation_replaced_dataflows_count_total: raw::IntCounterVec,
44
45 arrangement_maintenance_seconds_total: raw::CounterVec,
47 arrangement_maintenance_active_info: raw::UIntGaugeVec,
48
49 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_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 shared_row_heap_capacity_bytes: raw::UIntGaugeVec,
78
79 replica_expiration_timestamp_seconds: raw::UIntGaugeVec,
81 replica_expiration_remaining_seconds: raw::GaugeVec,
82
83 collection_count: raw::UIntGaugeVec,
85
86 subscribe_snapshots_skipped_total: IntCounter,
88}
89
90fn 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 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 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 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#[derive(Clone, Debug)]
361pub struct WorkerMetrics {
362 worker_label: String,
363 metrics: ComputeMetrics,
364
365 pub(crate) arrangement_maintenance_seconds_total: GenericCounter<AtomicF64>,
367 pub(crate) arrangement_maintenance_active_info: UIntGauge,
373 pub(crate) timely_step_duration_seconds: Histogram,
375 pub(crate) persist_peek_seconds: Histogram,
377 pub(crate) stashed_peek_seconds: Histogram,
379 pub(crate) handle_command_duration_seconds: CommandMetrics<Histogram>,
381 pub(crate) index_peek_total_seconds: Histogram,
383 pub(crate) index_peek_seek_fulfillment_seconds: Histogram,
385 pub(crate) index_peek_error_scan_seconds: Histogram,
387 pub(crate) index_peek_cursor_setup_seconds: Histogram,
389 pub(crate) index_peek_row_iteration_seconds: Histogram,
391 pub(crate) index_peek_row_iteration_rows: Histogram,
393 pub(crate) index_peek_result_sort_seconds: Histogram,
395 pub(crate) index_peek_result_sort_rows: Histogram,
397 pub(crate) index_peek_frontier_check_seconds: Histogram,
399 pub(crate) index_peek_row_collection_seconds: Histogram,
401 pub(crate) replica_expiration_timestamp_seconds: UIntGauge,
403 pub(crate) replica_expiration_remaining_seconds: raw::Gauge,
405 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 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 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 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 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 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#[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 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 #[mz_ore::test]
586 fn solo_runtime_omits_role_label() {
587 let registry = MetricsRegistry::new();
588 let metrics = ComputeMetrics::register_with(®istry, ComputeRuntimeRole::Solo);
589 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 #[mz_ore::test]
610 fn named_roles_carry_distinct_role_label() {
611 let registry = MetricsRegistry::new();
612 let maintenance = ComputeMetrics::register_with(®istry, ComputeRuntimeRole::Maintenance);
613 let interactive = ComputeMetrics::register_with(®istry, 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}