1use mz_ore::metric;
11use mz_ore::metrics::{MetricTag, MetricVisibility, MetricsRegistry, UIntGauge};
12use mz_ore::stats::{histogram_milliseconds_buckets, histogram_seconds_buckets};
13use mz_sql::ast::{AstInfo, Statement, StatementKind, SubscribeOutput};
14use mz_sql::session::user::User;
15use mz_sql_parser::ast::statement_kind_label_value;
16use prometheus::core::{AtomicU64, GenericCounter};
17use prometheus::{Histogram, HistogramVec, IntCounter, IntCounterVec, IntGaugeVec};
18
19#[derive(Debug, Clone)]
20pub struct Metrics {
21 pub query_total: IntCounterVec,
22 pub active_sessions: IntGaugeVec,
23 pub active_subscribes: IntGaugeVec,
24 pub active_copy_tos: IntGaugeVec,
25 pub queue_busy_seconds: Histogram,
26 pub determine_timestamp: IntCounterVec,
27 pub timestamp_difference_for_strict_serializable_ms: HistogramVec,
28 pub timestamp_difference_for_bounded_staleness_ms: HistogramVec,
29 pub commands: IntCounterVec,
30 pub storage_usage_collection_time_seconds: Histogram,
31 pub arrangement_sizes_collection_time_seconds: Histogram,
32 pub arrangement_sizes_rows_written: IntCounter,
33 pub subscribe_outputs: IntCounterVec,
34 pub canceled_peeks: IntCounter,
35 pub linearize_message_seconds: HistogramVec,
36 pub time_to_first_row_seconds: HistogramVec,
37 pub statement_logging_records: IntCounterVec,
38 pub statement_logging_unsampled_bytes: IntCounter,
39 pub statement_logging_actual_bytes: IntCounter,
40 pub message_batch: Histogram,
41 pub message_handling: HistogramVec,
42 pub optimization_notices: IntCounterVec,
43 pub append_table_duration_seconds: Histogram,
44 pub webhook_validation_reduce_failures: IntCounterVec,
45 pub webhook_get_appender: IntCounter,
46 pub check_scheduling_policies_seconds: HistogramVec,
47 pub handle_scheduling_decisions_seconds: HistogramVec,
48 pub row_set_finishing_seconds: Histogram,
49 pub session_startup_table_writes_seconds: Histogram,
50 pub parse_seconds: Histogram,
51 pub pgwire_message_processing_seconds: HistogramVec,
52 pub result_rows_first_to_last_byte_seconds: HistogramVec,
53 pub pgwire_ensure_transaction_seconds: HistogramVec,
54 pub catalog_snapshot_seconds: HistogramVec,
55 pub catalog_snapshot_cache: IntCounterVec,
56 pub catalog_arc_strong_count: UIntGauge,
57 pub catalog_arc_weak_count: UIntGauge,
58 pub pgwire_recv_scheduling_delay_ms: HistogramVec,
59 pub catalog_transact_seconds: HistogramVec,
60 pub apply_catalog_implications_seconds: Histogram,
61 pub group_commit_catalog_upper_seconds: Histogram,
62}
63
64impl Metrics {
65 pub(crate) fn register_into(registry: &MetricsRegistry) -> Self {
66 Self {
67 query_total: registry.register(metric!(
68 name: "mz_query_total",
69 help: "The total number of queries issued of the given type since process start.",
70 var_labels: ["session_type", "statement_type"],
71 visibility: MetricVisibility::Public,
72 tags: [MetricTag::Environment],
73 )),
74 active_sessions: registry.register(metric!(
75 name: "mz_active_sessions",
76 help: "The number of active coordinator sessions.",
77 var_labels: ["session_type"],
78 visibility: MetricVisibility::Public,
79 tags: [MetricTag::Environment],
80 )),
81 active_subscribes: registry.register(metric!(
82 name: "mz_active_subscribes",
83 help: "The number of active SUBSCRIBE queries.",
84 var_labels: ["session_type"],
85 visibility: MetricVisibility::Public,
86 tags: [MetricTag::Environment],
87 )),
88 active_copy_tos: registry.register(metric!(
89 name: "mz_active_copy_tos",
90 help: "The number of active COPY TO queries.",
91 var_labels: ["session_type"],
92 )),
93 queue_busy_seconds: registry.register(metric!(
94 name: "mz_coord_queue_busy_seconds",
95 help: "The number of seconds the coord queue was processing before it was empty. This is a sampled metric and does not measure the full coord queue wait/idle times.",
96 buckets: histogram_seconds_buckets(0.000_128, 32.0)
97 )),
98 determine_timestamp: registry.register(metric!(
99 name: "mz_determine_timestamp",
100 help: "The total number of calls to determine_timestamp.",
101 var_labels:["respond_immediately", "isolation_level", "compute_instance"],
102 )),
103 timestamp_difference_for_strict_serializable_ms: registry.register(metric!(
104 name: "mz_timestamp_difference_for_strict_serializable_ms",
105 help: "Difference in timestamp in milliseconds for running in strict serializable vs serializable isolation level.",
106 var_labels:["compute_instance"],
107 buckets: histogram_milliseconds_buckets(1., 8000.),
108 )),
109 timestamp_difference_for_bounded_staleness_ms: registry.register(metric!(
110 name: "mz_timestamp_difference_for_bounded_staleness_ms",
111 help: "How much older bounded-staleness timestamps are compared to serializable, in milliseconds. Measures the actual staleness incurred.",
112 var_labels:["compute_instance"],
113 buckets: histogram_milliseconds_buckets(1., 8000.),
114 )),
115 commands: registry.register(metric!(
116 name: "mz_adapter_commands",
117 help: "The total number of adapter commands issued of the given type since process start.",
118 var_labels: ["command_type", "status", "application_name"],
119 visibility: MetricVisibility::Public,
120 tags: [MetricTag::Environment],
121 )),
122 storage_usage_collection_time_seconds: registry.register(metric!(
123 name: "mz_storage_usage_collection_time_seconds",
124 help: "The number of seconds the coord spends collecting usage metrics from storage.",
125 buckets: histogram_seconds_buckets(0.000_128, 8.0)
126 )),
127 arrangement_sizes_collection_time_seconds: registry.register(metric!(
128 name: "mz_arrangement_sizes_collection_time_seconds",
129 help: "Seconds to read mz_object_arrangement_sizes and prepare history-table updates for one snapshot.",
130 buckets: histogram_seconds_buckets(0.000_128, 8.0)
131 )),
132 arrangement_sizes_rows_written: registry.register(metric!(
133 name: "mz_arrangement_sizes_rows_written_total",
134 help: "Total rows appended to mz_object_arrangement_size_history since process start.",
135 )),
136 subscribe_outputs: registry.register(metric!(
137 name: "mz_subscribe_outputs",
138 help: "The total number of different subscribe outputs used",
139 var_labels: ["session_type", "subscribe_output"],
140 )),
141 canceled_peeks: registry.register(metric!(
142 name: "mz_canceled_peeks_total",
143 help: "The total number of canceled peeks since process start.",
144 )),
145 linearize_message_seconds: registry.register(metric!(
146 name: "mz_linearize_message_seconds",
147 help: "The number of seconds it takes to linearize strict serializable messages",
148 var_labels: ["type", "immediately_handled"],
149 buckets: histogram_seconds_buckets(0.000_128, 8.0),
150 )),
151 time_to_first_row_seconds: registry.register(metric! {
152 name: "mz_time_to_first_row_seconds",
153 help: "Latency of an execute for a successful query from pgwire's perspective",
154 var_labels: ["instance_id", "isolation_level", "strategy", "application_name"],
155 buckets: histogram_seconds_buckets(0.000_128, 32.0)
156 }),
157 statement_logging_records: registry.register(metric! {
158 name: "mz_statement_logging_record_count",
159 help: "The total number of SQL statements tagged with whether or not they were recorded.",
160 var_labels: ["sample"],
161 }),
162 statement_logging_unsampled_bytes: registry.register(metric!(
163 name: "mz_statement_logging_unsampled_bytes",
164 help: "The total amount of SQL text that would have been logged if statement logging were unsampled.",
165 )),
166 statement_logging_actual_bytes: registry.register(metric!(
167 name: "mz_statement_logging_actual_bytes",
168 help: "The total amount of SQL text that was logged by statement logging.",
169 )),
170 message_batch: registry.register(metric!(
171 name: "mz_coordinator_message_batch_size",
172 help: "Message batch size handled by the coordinator.",
173 buckets: vec![0., 1., 2., 3., 4., 6., 8., 12., 16., 24., 32., 48., 64.],
174 )),
175 message_handling: registry.register(metric!(
176 name: "mz_slow_message_handling",
177 help: "Latency for ALL coordinator messages. 'slow' is in the name for legacy reasons, but is not accurate.",
178 var_labels: ["message_kind"],
179 buckets: histogram_seconds_buckets(0.000_128, 512.0),
180 )),
181 optimization_notices: registry.register(metric!(
182 name: "mz_optimization_notices",
183 help: "Number of optimization notices per notice type.",
184 var_labels: ["notice_type"],
185 )),
186 append_table_duration_seconds: registry.register(metric!(
187 name: "mz_append_table_duration_seconds",
188 help: "Latency for appending to any (user or system) table.",
189 buckets: histogram_seconds_buckets(0.128, 32.0),
190 )),
191 webhook_validation_reduce_failures: registry.register(metric!(
192 name: "mz_webhook_validation_reduce_failures",
193 help: "Count of how many times we've failed to reduce a webhook source's CHECK statement.",
194 var_labels: ["reason"],
195 )),
196 webhook_get_appender: registry.register(metric!(
197 name: "mz_webhook_get_appender_count",
198 help: "Count of getting a webhook appender from the Coordinator.",
199 )),
200 check_scheduling_policies_seconds: registry.register(metric!(
201 name: "mz_check_scheduling_policies_seconds",
202 help: "The time each policy in `check_scheduling_policies` takes.",
203 var_labels: ["policy", "thread"],
204 buckets: histogram_seconds_buckets(0.000_128, 8.0),
205 )),
206 handle_scheduling_decisions_seconds: registry.register(metric!(
207 name: "mz_handle_scheduling_decisions_seconds",
208 help: "The time `handle_scheduling_decisions` takes.",
209 var_labels: ["altered_a_cluster"],
210 buckets: histogram_seconds_buckets(0.000_128, 8.0),
211 )),
212 row_set_finishing_seconds: registry.register(metric!(
213 name: "mz_row_set_finishing_seconds",
214 help: "The time it takes to run RowSetFinishing::finish.",
215 buckets: histogram_seconds_buckets(0.000_128, 16.0),
216 )),
217 session_startup_table_writes_seconds: registry.register(metric!(
218 name: "mz_session_startup_table_writes_seconds",
219 help: "If we had to wait for builtin table writes before processing a query, how long did we wait for.",
220 buckets: histogram_seconds_buckets(0.000_008, 4.0),
221 )),
222 parse_seconds: registry.register(metric!(
223 name: "mz_parse_seconds",
224 help: "The time it takes to parse a SQL statement. (Works for both Simple Queries and the Extended Query protocol.)",
225 buckets: histogram_seconds_buckets(0.001, 8.0),
226 )),
227 pgwire_message_processing_seconds: registry.register(metric!(
228 name: "mz_pgwire_message_processing_seconds",
229 help: "The time it takes to process each of the pgwire message types, measured in the Adapter frontend",
230 var_labels: ["message_type"],
231 buckets: histogram_seconds_buckets(0.001, 512.0),
232 )),
233 result_rows_first_to_last_byte_seconds: registry.register(metric!(
234 name: "mz_result_rows_first_to_last_byte_seconds",
235 help: "The time from just before sending the first result row to sending a final response message after having successfully flushed the last result row to the connection. (This can span multiple FETCH statements.) (This is never observed for unbounded SUBSCRIBEs, i.e., which have no last result row.)",
236 var_labels: ["statement_type"],
237 buckets: histogram_seconds_buckets(0.001, 8192.0),
238 )),
239 pgwire_ensure_transaction_seconds: registry.register(metric!(
240 name: "mz_pgwire_ensure_transaction_seconds",
241 help: "The time it takes to run `ensure_transactions` when processing pgwire messages.",
242 var_labels: ["message_type"],
243 buckets: histogram_seconds_buckets(0.001, 512.0),
244 )),
245 catalog_snapshot_seconds: registry.register(metric!(
246 name: "mz_catalog_snapshot_seconds",
247 help: "The time it takes to fetch a catalog snapshot from the Coordinator. \
248 Only observed on session snapshot cache misses.",
249 var_labels: ["context"],
250 buckets: histogram_seconds_buckets(0.001, 512.0),
251 )),
252 catalog_snapshot_cache: registry.register(metric!(
253 name: "mz_catalog_snapshot_cache",
254 help: "Hits and misses of the session-side catalog snapshot cache. A miss \
255 costs a Coordinator round-trip.",
256 var_labels: ["context", "result"],
257 )),
258 catalog_arc_strong_count: registry.register(metric!(
259 name: "mz_catalog_arc_strong_count",
260 help: "The number of strong references to the current catalog snapshot: roughly, \
261 in-flight users plus a small constant baseline.",
262 )),
263 catalog_arc_weak_count: registry.register(metric!(
264 name: "mz_catalog_arc_weak_count",
265 help: "The number of weak references to the current catalog snapshot: sessions \
266 whose snapshot cache points at the current catalog version (older \
267 versions are not counted). Drops on catalog changes and recovers as \
268 session caches repopulate.",
269 )),
270 pgwire_recv_scheduling_delay_ms: registry.register(metric!(
271 name: "mz_pgwire_recv_scheduling_delay_ms",
272 help: "The time between a pgwire connection's receiver task being woken up by incoming data and getting polled.",
273 var_labels: ["message_type"],
274 buckets: histogram_milliseconds_buckets(0.128, 512000.),
275 )),
276 catalog_transact_seconds: registry.register(metric!(
277 name: "mz_catalog_transact_seconds",
278 help: "The time it takes to run various catalog transact methods.",
279 var_labels: ["method"],
280 buckets: histogram_seconds_buckets(0.001, 32.0),
281 )),
282 apply_catalog_implications_seconds: registry.register(metric!(
283 name: "mz_apply_catalog_implications_seconds",
284 help: "The time it takes to apply catalog implications.",
285 buckets: histogram_seconds_buckets(0.001, 32.0),
286 )),
287 group_commit_catalog_upper_seconds: registry.register(metric!(
288 name: "mz_group_commit_catalog_upper_seconds",
289 help: "The time it takes to advance the catalog shard upper during group commit.",
290 buckets: histogram_seconds_buckets(0.001, 32.0),
291 )),
292 }
293 }
294
295 pub(crate) fn row_set_finishing_seconds(&self) -> Histogram {
296 self.row_set_finishing_seconds.clone()
297 }
298
299 pub(crate) fn session_metrics(&self) -> SessionMetrics {
300 SessionMetrics {
301 row_set_finishing_seconds: self.row_set_finishing_seconds(),
302 session_startup_table_writes_seconds: self.session_startup_table_writes_seconds.clone(),
303 query_total: self.query_total.clone(),
304 determine_timestamp: self.determine_timestamp.clone(),
305 timestamp_difference_for_strict_serializable_ms: self
306 .timestamp_difference_for_strict_serializable_ms
307 .clone(),
308 timestamp_difference_for_bounded_staleness_ms: self
309 .timestamp_difference_for_bounded_staleness_ms
310 .clone(),
311 optimization_notices: self.optimization_notices.clone(),
312 statement_logging_records: self.statement_logging_records.clone(),
313 statement_logging_unsampled_bytes: self.statement_logging_unsampled_bytes.clone(),
314 statement_logging_actual_bytes: self.statement_logging_actual_bytes.clone(),
315 }
316 }
317}
318
319#[derive(Debug, Clone)]
321pub struct SessionMetrics {
322 row_set_finishing_seconds: Histogram,
323 session_startup_table_writes_seconds: Histogram,
324 query_total: IntCounterVec,
325 determine_timestamp: IntCounterVec,
326 timestamp_difference_for_strict_serializable_ms: HistogramVec,
327 timestamp_difference_for_bounded_staleness_ms: HistogramVec,
328 optimization_notices: IntCounterVec,
329 statement_logging_records: IntCounterVec,
330 statement_logging_unsampled_bytes: IntCounter,
331 statement_logging_actual_bytes: IntCounter,
332}
333
334impl SessionMetrics {
335 pub(crate) fn row_set_finishing_seconds(&self) -> &Histogram {
336 &self.row_set_finishing_seconds
337 }
338
339 pub(crate) fn session_startup_table_writes_seconds(&self) -> &Histogram {
340 &self.session_startup_table_writes_seconds
341 }
342
343 pub(crate) fn query_total(&self, label_values: &[&str]) -> GenericCounter<AtomicU64> {
344 self.query_total.with_label_values(label_values)
345 }
346
347 pub(crate) fn determine_timestamp(&self, label_values: &[&str]) -> GenericCounter<AtomicU64> {
348 self.determine_timestamp.with_label_values(label_values)
349 }
350
351 pub(crate) fn timestamp_difference_for_strict_serializable_ms(
352 &self,
353 label_values: &[&str],
354 ) -> Histogram {
355 self.timestamp_difference_for_strict_serializable_ms
356 .with_label_values(label_values)
357 }
358
359 pub(crate) fn timestamp_difference_for_bounded_staleness_ms(
360 &self,
361 label_values: &[&str],
362 ) -> Histogram {
363 self.timestamp_difference_for_bounded_staleness_ms
364 .with_label_values(label_values)
365 }
366
367 pub(crate) fn optimization_notices(&self, label_values: &[&str]) -> GenericCounter<AtomicU64> {
368 self.optimization_notices.with_label_values(label_values)
369 }
370
371 pub(crate) fn statement_logging_records(
372 &self,
373 label_values: &[&str],
374 ) -> GenericCounter<AtomicU64> {
375 self.statement_logging_records
376 .with_label_values(label_values)
377 }
378
379 pub(crate) fn statement_logging_unsampled_bytes(&self) -> &IntCounter {
380 &self.statement_logging_unsampled_bytes
381 }
382
383 pub(crate) fn statement_logging_actual_bytes(&self) -> &IntCounter {
384 &self.statement_logging_actual_bytes
385 }
386}
387
388pub(crate) fn session_type_label_value(user: &User) -> &'static str {
389 match user.is_internal() {
390 true => "system",
391 false => "user",
392 }
393}
394
395pub fn statement_type_label_value<T>(stmt: &Statement<T>) -> &'static str
396where
397 T: AstInfo,
398{
399 statement_kind_label_value(StatementKind::from(stmt))
400}
401
402pub(crate) fn subscribe_output_label_value<T>(output: &SubscribeOutput<T>) -> &'static str
403where
404 T: AstInfo,
405{
406 match output {
407 SubscribeOutput::Diffs => "diffs",
408 SubscribeOutput::WithinTimestampOrderBy { .. } => "within_timestamp_order_by",
409 SubscribeOutput::EnvelopeUpsert { .. } => "envelope_upsert",
410 SubscribeOutput::EnvelopeDebezium { .. } => "envelope_debezium",
411 }
412}