mz_compute/compute_state/peek_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//! What an index peek reports about the walk that answers it.
7//!
8//! A walk runs on one driver or two, and the numbers a [`PeekScan`] carries are cumulative over
9//! every slice either drove. Each histogram is observed once per walk, by the driver that produces
10//! the terminal outcome. An offload is not terminal, so a slice that offloads reports nothing and
11//! the task that finishes reports the whole set. A cancelled walk reports nothing, so the substrate
12//! counters count walks that ended.
13//!
14//! [`PeekScan`]: super::peek_scan::PeekScan
15
16use std::time::{Duration, Instant};
17
18use mz_ore::cast::CastLossy;
19use mz_ore::metrics::UIntGauge;
20use prometheus::{Histogram, IntCounter};
21
22use crate::compute_state::peek_scan::WalkPhases;
23use crate::metrics::WorkerMetrics;
24
25/// The metrics an index peek walk reports, on either substrate.
26///
27/// Cloned into an offloaded walk's task, so that a walk reports the same phases wherever it ran.
28#[derive(Clone, Debug)]
29pub(super) struct PeekWalkMetrics {
30 /// Counts walks that ended on the timely worker.
31 walks_inline: IntCounter,
32 /// Counts walks that ended away from the timely worker.
33 walks_offloaded: IntCounter,
34 /// Counts walks that answered from the peek response stash.
35 walks_stashed: IntCounter,
36 error_scan_seconds: Histogram,
37 cursor_setup_seconds: Histogram,
38 row_iteration_seconds: Histogram,
39 row_iteration_rows: Histogram,
40 result_sort_seconds: Histogram,
41 result_sort_rows: Histogram,
42 row_collection_seconds: Histogram,
43 /// How many offloaded walks are waiting for a permit. Reported by the offloaded driver alone,
44 /// which is the only one that queues.
45 permit_queue_depth: UIntGauge,
46 /// How long an offloaded walk waited for its permit. Reported by the offloaded driver alone.
47 permit_wait_seconds: Histogram,
48}
49
50impl PeekWalkMetrics {
51 pub(super) fn new(metrics: &WorkerMetrics) -> Self {
52 Self {
53 walks_inline: metrics.index_peek_walks_inline.clone(),
54 walks_offloaded: metrics.index_peek_walks_offloaded.clone(),
55 walks_stashed: metrics.index_peek_stashed_total.clone(),
56 error_scan_seconds: metrics.index_peek_error_scan_seconds.clone(),
57 cursor_setup_seconds: metrics.index_peek_cursor_setup_seconds.clone(),
58 row_iteration_seconds: metrics.index_peek_row_iteration_seconds.clone(),
59 row_iteration_rows: metrics.index_peek_row_iteration_rows.clone(),
60 result_sort_seconds: metrics.index_peek_result_sort_seconds.clone(),
61 result_sort_rows: metrics.index_peek_result_sort_rows.clone(),
62 row_collection_seconds: metrics.index_peek_row_collection_seconds.clone(),
63 permit_queue_depth: metrics.index_peek_permit_queue_depth.clone(),
64 permit_wait_seconds: metrics.index_peek_permit_wait_seconds.clone(),
65 }
66 }
67
68 /// Accounts for an offloaded walk joining the queue for a permit.
69 ///
70 /// The returned guard leaves the queue however the wait ends, a cancellation and an abort
71 /// included, so the depth cannot drift up over a process's life.
72 pub(super) fn queued_for_permit(&self) -> PermitWait {
73 self.permit_queue_depth.inc();
74 PermitWait {
75 queue_depth: self.permit_queue_depth.clone(),
76 wait_seconds: self.permit_wait_seconds.clone(),
77 since: Instant::now(),
78 }
79 }
80
81 /// Counts a walk that the timely worker drove to an outcome.
82 ///
83 /// A walk that suspends leaves the worker rather than finishing here, so a peek answered from
84 /// the peek stash never counts here: the driver that writes to the stash is the offloaded one.
85 pub(super) fn walked_inline(&self) {
86 self.walks_inline.inc();
87 }
88
89 /// Counts a walk that an offloaded task drove to an outcome, whatever that outcome is.
90 ///
91 /// A walk cancelled while queued or while running counts on neither substrate, as does one
92 /// whose task died without an outcome, which the worker answers with an error of its own. The
93 /// two therefore sum to the walks that reached an outcome rather than to the peeks answered.
94 pub(super) fn walked_offloaded(&self) {
95 self.walks_offloaded.inc();
96 }
97
98 /// Counts a walk that answered with a handle to the peek response stash.
99 ///
100 /// Counted alongside [`Self::walked_offloaded`] rather than instead of it, so the two
101 /// substrates still sum to the walks that ended.
102 pub(super) fn walked_to_stash(&self) {
103 self.walks_stashed.inc();
104 }
105
106 /// Reports the phases that precede the walk over the ok trace.
107 ///
108 /// Reported for every terminal outcome, a hand-off to the peek stash included, because both
109 /// phases are over by then whatever the walk does next.
110 pub(super) fn observe_error_phase(&self, phases: &WalkPhases) {
111 // A peek its error trace answered reports neither number: that walk stopped where the
112 // error was, and the cursor the second number times was never used.
113 if !phases.error_trace_clean {
114 return;
115 }
116
117 self.error_scan_seconds
118 .observe(phases.error_scan.as_secs_f64());
119 self.cursor_setup_seconds
120 .observe(phases.cursor_setup.as_secs_f64());
121 }
122
123 /// Reports the walk over the ok trace, for a walk that completed it. A walk that ends any
124 /// other way reports nothing here: the rows it examined are not the rows an answer took.
125 pub(super) fn observe_ok_phase(&self, phases: &WalkPhases) {
126 self.row_iteration_seconds
127 .observe(phases.row_iteration.as_secs_f64());
128 self.row_iteration_rows
129 .observe(f64::cast_lossy(phases.rows_processed));
130 self.result_sort_seconds
131 .observe(phases.thinning.as_secs_f64());
132 self.result_sort_rows
133 .observe(f64::cast_lossy(phases.rows_thinned));
134 }
135
136 /// Reports the time [`rows_response`](super::peek_scan::rows_response) took.
137 pub(super) fn observe_row_collection(&self, elapsed: Duration) {
138 self.row_collection_seconds.observe(elapsed.as_secs_f64());
139 }
140}
141
142/// An offloaded walk's place in the queue for a permit, for as long as it holds one.
143///
144/// Leaving the queue is a drop rather than a call, so a cancelled or aborted walk leaves it as
145/// surely as an admitted one.
146pub(super) struct PermitWait {
147 queue_depth: UIntGauge,
148 wait_seconds: Histogram,
149 since: Instant,
150}
151
152impl PermitWait {
153 /// Reports the wait of a walk that was admitted. A walk that leaves the queue any other way
154 /// reports nothing, so the histogram describes waits that ended in a permit.
155 pub(super) fn admitted(self) {
156 self.wait_seconds
157 .observe(self.since.elapsed().as_secs_f64());
158 }
159}
160
161impl Drop for PermitWait {
162 fn drop(&mut self) {
163 self.queue_depth.dec();
164 }
165}
166
167/// The metrics an index peek reports from the worker that owns it.
168///
169/// These time the worker's own handling of a peek, which no offloaded walk repeats, so unlike the
170/// walk's metrics they have a single observer by construction.
171pub(super) struct IndexPeekMetrics<'a> {
172 pub seek_fulfillment_seconds: &'a Histogram,
173 pub frontier_check_seconds: &'a Histogram,
174 /// The metrics of the walk itself, which an offloaded walk reports too.
175 pub walk: &'a PeekWalkMetrics,
176}