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