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