mz_adapter/
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// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10use std::borrow::Cow;
11
12use mz_controller_types::ClusterId;
13use mz_ore::metric;
14use mz_ore::metrics::{
15    MakeCollector, MakeCollectorOpts, MetricTag, MetricVisibility, MetricsRegistry, UIntGauge,
16    remove_children_with_label,
17};
18use mz_ore::stats::{histogram_milliseconds_buckets, histogram_seconds_buckets};
19use mz_sql::ast::{
20    AsOf, AstInfo, CopyRelation, CopyStatement, SelectStatement, Statement, StatementKind,
21    SubscribeOutput, SubscribeStatement,
22};
23use mz_sql::session::hint::ApplicationNameHint;
24use mz_sql::session::user::User;
25use mz_sql::session::vars::IsolationLevel;
26use mz_sql_parser::ast::statement_kind_label_value;
27use prometheus::core::{AtomicU64, GenericCounter};
28use prometheus::{Histogram, HistogramVec, IntCounter, IntCounterVec, IntGaugeVec};
29
30use crate::statement_logging::StatementExecutionStrategy;
31
32pub(crate) const OCC_CALLER_SESSION: &str = "session";
33pub(crate) const OCC_CALLER_BACKGROUND: &str = "background";
34
35#[derive(Debug, Clone)]
36pub struct Metrics {
37    pub query_total: IntCounterVec,
38    pub active_sessions: IntGaugeVec,
39    pub active_subscribes: IntGaugeVec,
40    pub active_internal_subscribes: IntGaugeVec,
41    pub active_copy_tos: IntGaugeVec,
42    pub queue_busy_seconds: Histogram,
43    pub commands: IntCounterVec,
44    pub storage_usage_collection_time_seconds: Histogram,
45    pub arrangement_sizes_collection_time_seconds: Histogram,
46    pub arrangement_sizes_rows_written: IntCounter,
47    pub hydration_history_mutations: IntCounterVec,
48    pub hydration_history_retention_batch_full: IntCounter,
49    pub hydration_history_rows_affected: IntCounterVec,
50    pub hydration_history_sweep_duration_seconds: Histogram,
51    pub subscribe_outputs: IntCounterVec,
52    pub as_of_queries: IntCounterVec,
53    pub canceled_peeks: IntCounter,
54    pub linearize_message_seconds: HistogramVec,
55    pub linearize_read_immediate_seconds: Histogram,
56    pub linearize_read_delayed_seconds: Histogram,
57    pub linearize_read_fast_path: IntCounter,
58    pub frontend_transaction_completions: IntCounter,
59    pub frontend_transaction_waits: IntCounter,
60    pub statement_logging_records: IntCounterVec,
61    pub statement_logging_unsampled_bytes: IntCounter,
62    pub statement_logging_actual_bytes: IntCounter,
63    pub message_batch: Histogram,
64    pub message_handling: HistogramVec,
65    pub optimization_notices: IntCounterVec,
66    pub append_table_duration_seconds: Histogram,
67    pub webhook_validation_reduce_failures: IntCounterVec,
68    pub webhook_get_appender: IntCounter,
69    pub row_set_finishing_seconds: Histogram,
70    pub session_startup_table_writes_seconds: Histogram,
71    pub parse_seconds: Histogram,
72    pub pgwire_message_processing_seconds: HistogramVec,
73    pub result_rows_first_to_last_byte_seconds: HistogramVec,
74    pub pgwire_ensure_transaction_seconds: HistogramVec,
75    pub catalog_snapshot_seconds: HistogramVec,
76    pub catalog_snapshot_cache: IntCounterVec,
77    pub catalog_arc_strong_count: UIntGauge,
78    pub catalog_arc_weak_count: UIntGauge,
79    pub pgwire_recv_scheduling_delay_ms: HistogramVec,
80    pub catalog_transact_seconds: HistogramVec,
81    pub catalog_transact_phase_seconds: HistogramVec,
82    pub apply_catalog_implications_seconds: Histogram,
83    pub group_commit_catalog_upper_seconds: Histogram,
84    pub occ_retry_count: HistogramVec,
85    pub by_cluster: ClusterLabeledMetrics,
86}
87
88impl Metrics {
89    pub(crate) fn register_into(registry: &MetricsRegistry) -> Self {
90        let linearize_message_seconds: HistogramVec = registry.register(metric!(
91            name: "mz_linearize_message_seconds",
92            help: "The number of seconds it takes to linearize strict serializable messages",
93            var_labels: ["type", "immediately_handled"],
94            buckets: histogram_seconds_buckets(0.000_128, 8.0),
95        ));
96        Self {
97            query_total: registry.register(metric!(
98                name: "mz_query_total",
99                help: "The total number of queries issued of the given type since process start.",
100                var_labels: ["session_type", "statement_type"],
101                visibility: MetricVisibility::Public,
102                tags: [MetricTag::Environment],
103            )),
104            active_sessions: registry.register(metric!(
105                name: "mz_active_sessions",
106                help: "The number of active coordinator sessions.",
107                var_labels: ["session_type"],
108                visibility: MetricVisibility::Public,
109                tags: [MetricTag::Environment],
110            )),
111            active_subscribes: registry.register(metric!(
112                name: "mz_active_subscribes",
113                help: "The number of active SUBSCRIBE queries.",
114                var_labels: ["session_type"],
115                visibility: MetricVisibility::Public,
116                tags: [MetricTag::Environment],
117            )),
118            active_internal_subscribes: registry.register(metric!(
119                name: "mz_active_internal_subscribes",
120                help: "The number of active internal subscribes used by read-then-write operations and background maintenance.",
121                var_labels: ["session_type"],
122            )),
123            active_copy_tos: registry.register(metric!(
124                name: "mz_active_copy_tos",
125                help: "The number of active COPY TO queries.",
126                var_labels: ["session_type"],
127            )),
128            queue_busy_seconds: registry.register(metric!(
129                name: "mz_coord_queue_busy_seconds",
130                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.",
131                buckets: histogram_seconds_buckets(0.000_128, 32.0)
132            )),
133            commands: registry.register(metric!(
134                name: "mz_adapter_commands",
135                help: "The total number of adapter commands issued of the given type since process start.",
136                var_labels: ["command_type", "status", "application_name"],
137                visibility: MetricVisibility::Public,
138                tags: [MetricTag::Environment],
139            )),
140            storage_usage_collection_time_seconds: registry.register(metric!(
141                name: "mz_storage_usage_collection_time_seconds",
142                help: "The number of seconds the coord spends collecting usage metrics from storage.",
143                buckets: histogram_seconds_buckets(0.000_128, 8.0)
144            )),
145            arrangement_sizes_collection_time_seconds: registry.register(metric!(
146                name: "mz_arrangement_sizes_collection_time_seconds",
147                help: "Seconds to read mz_object_arrangement_sizes and prepare history records for one snapshot.",
148                buckets: histogram_seconds_buckets(0.000_128, 8.0)
149            )),
150            arrangement_sizes_rows_written: registry.register(metric!(
151                name: "mz_arrangement_sizes_rows_written_total",
152                help: "Total rows appended to mz_object_arrangement_size_history since process start.",
153            )),
154            hydration_history_mutations: registry.register(metric!(
155                name: "mz_hydration_history_mutations_total",
156                help: "Total hydration-history collection and retention mutations since process start.",
157                var_labels: ["operation", "outcome"],
158            )),
159            hydration_history_retention_batch_full: registry.register(metric!(
160                name: "mz_hydration_history_retention_batch_full_total",
161                help: "Total hydration-history retention batches that were full. Repeated increments mean retention may not be keeping up with its schedule.",
162            )),
163            hydration_history_rows_affected: registry.register(metric!(
164                name: "mz_hydration_history_rows_affected_total",
165                help: "Total rows changed by hydration-history maintenance since process start.",
166                var_labels: ["action"],
167            )),
168            hydration_history_sweep_duration_seconds: registry.register(metric!(
169                name: "mz_hydration_history_sweep_duration_seconds",
170                help: "Wall time of a complete hydration-history collection and retention sweep.",
171                buckets: histogram_seconds_buckets(0.128, 1024.0),
172            )),
173            subscribe_outputs: registry.register(metric!(
174                name: "mz_subscribe_outputs",
175                help: "The total number of different subscribe outputs used",
176                var_labels: ["session_type", "subscribe_output"],
177            )),
178            as_of_queries: registry.register(metric!(
179                name: "mz_as_of_queries_total",
180                help: "The total number of SELECT and SUBSCRIBE executions with an AS OF clause.",
181                var_labels: ["session_type", "kind", "statement", "application_name"],
182            )),
183            canceled_peeks: registry.register(metric!(
184                name: "mz_canceled_peeks_total",
185                help: "The total number of canceled peeks since process start.",
186            )),
187            linearize_read_immediate_seconds: linearize_message_seconds.with_label_values(&["read", "true"]),
188            linearize_read_delayed_seconds: linearize_message_seconds.with_label_values(&["read", "false"]),
189            linearize_message_seconds,
190            linearize_read_fast_path: registry.register(metric!(
191                name: "mz_linearize_read_fast_path_total",
192                help: "The number of ordinary strict serializable reads already linearized at transaction end",
193            )),
194            frontend_transaction_completions: registry.register(metric!(
195                name: "mz_frontend_transaction_completions_total",
196                help: "Read-only transaction completion attempts owned by the session.",
197            )),
198            frontend_transaction_waits: registry.register(metric!(
199                name: "mz_frontend_transaction_waits_total",
200                help: "Session-owned transaction completions that need a fresh oracle check.",
201            )),
202            statement_logging_records: registry.register(metric! {
203                name: "mz_statement_logging_record_count",
204                help: "The total number of SQL statements tagged with whether or not they were recorded.",
205                var_labels: ["sample"],
206            }),
207            statement_logging_unsampled_bytes: registry.register(metric!(
208                name: "mz_statement_logging_unsampled_bytes",
209                help: "The total amount of SQL text that would have been logged if statement logging were unsampled.",
210            )),
211            statement_logging_actual_bytes: registry.register(metric!(
212                name: "mz_statement_logging_actual_bytes",
213                help: "The total amount of SQL text that was logged by statement logging.",
214            )),
215            message_batch: registry.register(metric!(
216                name: "mz_coordinator_message_batch_size",
217                help: "Message batch size handled by the coordinator.",
218                buckets: vec![0., 1., 2., 3., 4., 6., 8., 12., 16., 24., 32., 48., 64.],
219            )),
220            message_handling: registry.register(metric!(
221                name: "mz_slow_message_handling",
222                help: "Latency for ALL coordinator messages. 'slow' is in the name for legacy reasons, but is not accurate.",
223                var_labels: ["message_kind"],
224                buckets: histogram_seconds_buckets(0.000_128, 512.0),
225            )),
226            optimization_notices: registry.register(metric!(
227                name: "mz_optimization_notices",
228                help: "Number of optimization notices per notice type.",
229                var_labels: ["notice_type"],
230            )),
231            append_table_duration_seconds: registry.register(metric!(
232                name: "mz_append_table_duration_seconds",
233                help: "Latency for appending to any (user or system) table.",
234                buckets: histogram_seconds_buckets(0.128, 32.0),
235            )),
236            webhook_validation_reduce_failures: registry.register(metric!(
237                name: "mz_webhook_validation_reduce_failures",
238                help: "Count of how many times we've failed to reduce a webhook source's CHECK statement.",
239                var_labels: ["reason"],
240            )),
241            webhook_get_appender: registry.register(metric!(
242                name: "mz_webhook_get_appender_count",
243                help: "Count of getting a webhook appender from the Coordinator.",
244            )),
245            row_set_finishing_seconds: registry.register(metric!(
246                name: "mz_row_set_finishing_seconds",
247                help: "The time it takes to run RowSetFinishing::finish.",
248                buckets: histogram_seconds_buckets(0.000_128, 16.0),
249            )),
250            session_startup_table_writes_seconds: registry.register(metric!(
251                name: "mz_session_startup_table_writes_seconds",
252                help: "If we had to wait for builtin table writes before processing a query, how long did we wait for.",
253                buckets: histogram_seconds_buckets(0.000_008, 4.0),
254            )),
255            parse_seconds: registry.register(metric!(
256                name: "mz_parse_seconds",
257                help: "The time it takes to parse a SQL statement. (Works for both Simple Queries and the Extended Query protocol.)",
258                buckets: histogram_seconds_buckets(0.001, 8.0),
259            )),
260            pgwire_message_processing_seconds: registry.register(metric!(
261                name: "mz_pgwire_message_processing_seconds",
262                help: "The time it takes to process each of the pgwire message types, measured in the Adapter frontend",
263                var_labels: ["message_type"],
264                buckets: histogram_seconds_buckets(0.001, 512.0),
265            )),
266            result_rows_first_to_last_byte_seconds: registry.register(metric!(
267                name: "mz_result_rows_first_to_last_byte_seconds",
268                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.)",
269                var_labels: ["statement_type"],
270                buckets: histogram_seconds_buckets(0.001, 8192.0),
271            )),
272            pgwire_ensure_transaction_seconds: registry.register(metric!(
273                name: "mz_pgwire_ensure_transaction_seconds",
274                help: "The time it takes to run `ensure_transactions` when processing pgwire messages.",
275                var_labels: ["message_type"],
276                buckets: histogram_seconds_buckets(0.001, 512.0),
277            )),
278            catalog_snapshot_seconds: registry.register(metric!(
279                name: "mz_catalog_snapshot_seconds",
280                help: "The time it takes to fetch a catalog snapshot from the Coordinator. \
281                       Only observed on session snapshot cache misses.",
282                var_labels: ["context"],
283                buckets: histogram_seconds_buckets(0.001, 512.0),
284            )),
285            catalog_snapshot_cache: registry.register(metric!(
286                name: "mz_catalog_snapshot_cache",
287                help: "Hits and misses of the session-side catalog snapshot cache. A miss \
288                       costs a Coordinator round-trip.",
289                var_labels: ["context", "result"],
290            )),
291            catalog_arc_strong_count: registry.register(metric!(
292                name: "mz_catalog_arc_strong_count",
293                help: "The number of strong references to the current catalog snapshot: roughly, \
294                       in-flight users plus a small constant baseline.",
295            )),
296            catalog_arc_weak_count: registry.register(metric!(
297                name: "mz_catalog_arc_weak_count",
298                help: "The number of weak references to the current catalog snapshot: sessions \
299                       whose snapshot cache points at the current catalog version (older \
300                       versions are not counted). Drops on catalog changes and recovers as \
301                       session caches repopulate.",
302            )),
303            pgwire_recv_scheduling_delay_ms: registry.register(metric!(
304                name: "mz_pgwire_recv_scheduling_delay_ms",
305                help: "The time between a pgwire connection's receiver task being woken up by incoming data and getting polled.",
306                var_labels: ["message_type"],
307                buckets: histogram_milliseconds_buckets(0.128, 512000.),
308            )),
309            catalog_transact_seconds: registry.register(metric!(
310                name: "mz_catalog_transact_seconds",
311                help: "The time it takes to run various catalog transact methods.",
312                var_labels: ["method"],
313                buckets: histogram_seconds_buckets(0.001, 32.0),
314            )),
315            catalog_transact_phase_seconds: registry.register(metric!(
316                name: "mz_catalog_transact_phase_seconds",
317                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.",
318                var_labels: ["phase"],
319                buckets: histogram_seconds_buckets(0.000_128, 32.0),
320            )),
321            apply_catalog_implications_seconds: registry.register(metric!(
322                name: "mz_apply_catalog_implications_seconds",
323                help: "The time it takes to apply catalog implications.",
324                buckets: histogram_seconds_buckets(0.001, 32.0),
325            )),
326            group_commit_catalog_upper_seconds: registry.register(metric!(
327                name: "mz_group_commit_catalog_upper_seconds",
328                help: "The time it takes to advance the catalog shard upper for a txns-shard write (group commits and table register/forget).",
329                buckets: histogram_seconds_buckets(0.001, 32.0),
330            )),
331            occ_retry_count: registry.register(metric!(
332                name: "mz_occ_read_then_write_retry_count",
333                help: "Number of OCC retries per read-then-write operation.",
334                var_labels: ["caller"],
335                buckets: vec![
336                    0., 1., 2., 3., 5., 10., 25., 50., 100., 200., 300., 500., 750., 1000.,
337                ],
338            )),
339            by_cluster: ClusterLabeledMetrics::register_into(registry),
340        }
341    }
342
343    pub(crate) fn row_set_finishing_seconds(&self) -> Histogram {
344        self.row_set_finishing_seconds.clone()
345    }
346
347    pub(crate) fn session_metrics(&self) -> SessionMetrics {
348        SessionMetrics {
349            row_set_finishing_seconds: self.row_set_finishing_seconds(),
350            session_startup_table_writes_seconds: self.session_startup_table_writes_seconds.clone(),
351            query_total: self.query_total.clone(),
352            subscribe_outputs: self.subscribe_outputs.clone(),
353            as_of_queries: self.as_of_queries.clone(),
354            by_cluster: self.by_cluster.clone(),
355            optimization_notices: self.optimization_notices.clone(),
356            statement_logging_records: self.statement_logging_records.clone(),
357            statement_logging_unsampled_bytes: self.statement_logging_unsampled_bytes.clone(),
358            statement_logging_actual_bytes: self.statement_logging_actual_bytes.clone(),
359        }
360    }
361}
362
363/// Metrics to be accessed from a [`crate::session::Session`].
364#[derive(Debug, Clone)]
365pub struct SessionMetrics {
366    row_set_finishing_seconds: Histogram,
367    session_startup_table_writes_seconds: Histogram,
368    query_total: IntCounterVec,
369    subscribe_outputs: IntCounterVec,
370    as_of_queries: IntCounterVec,
371    by_cluster: ClusterLabeledMetrics,
372    optimization_notices: IntCounterVec,
373    statement_logging_records: IntCounterVec,
374    statement_logging_unsampled_bytes: IntCounter,
375    statement_logging_actual_bytes: IntCounter,
376}
377
378impl SessionMetrics {
379    pub(crate) fn row_set_finishing_seconds(&self) -> &Histogram {
380        &self.row_set_finishing_seconds
381    }
382
383    pub(crate) fn session_startup_table_writes_seconds(&self) -> &Histogram {
384        &self.session_startup_table_writes_seconds
385    }
386
387    pub(crate) fn query_total(&self, label_values: &[&str]) -> GenericCounter<AtomicU64> {
388        self.query_total.with_label_values(label_values)
389    }
390
391    pub(crate) fn subscribe_outputs(&self, label_values: &[&str]) -> GenericCounter<AtomicU64> {
392        self.subscribe_outputs.with_label_values(label_values)
393    }
394
395    pub(crate) fn as_of_queries(&self, label_values: &[&str]) -> GenericCounter<AtomicU64> {
396        self.as_of_queries.with_label_values(label_values)
397    }
398
399    pub(crate) fn by_cluster(&self) -> &ClusterLabeledMetrics {
400        &self.by_cluster
401    }
402
403    pub(crate) fn optimization_notices(&self, label_values: &[&str]) -> GenericCounter<AtomicU64> {
404        self.optimization_notices.with_label_values(label_values)
405    }
406
407    pub(crate) fn statement_logging_records(
408        &self,
409        label_values: &[&str],
410    ) -> GenericCounter<AtomicU64> {
411        self.statement_logging_records
412            .with_label_values(label_values)
413    }
414
415    pub(crate) fn statement_logging_unsampled_bytes(&self) -> &IntCounter {
416        &self.statement_logging_unsampled_bytes
417    }
418
419    pub(crate) fn statement_logging_actual_bytes(&self) -> &IntCounter {
420        &self.statement_logging_actual_bytes
421    }
422}
423
424pub(crate) fn session_type_label_value(user: &User) -> &'static str {
425    match user.is_internal() {
426        true => "system",
427        false => "user",
428    }
429}
430
431pub fn statement_type_label_value<T>(stmt: &Statement<T>) -> &'static str
432where
433    T: AstInfo,
434{
435    statement_kind_label_value(StatementKind::from(stmt))
436}
437
438pub(crate) fn subscribe_output_label_value<T>(output: &SubscribeOutput<T>) -> &'static str
439where
440    T: AstInfo,
441{
442    match output {
443        SubscribeOutput::Diffs => "diffs",
444        SubscribeOutput::WithinTimestampOrderBy { .. } => "within_timestamp_order_by",
445        SubscribeOutput::EnvelopeUpsert { .. } => "envelope_upsert",
446        SubscribeOutput::EnvelopeDebezium { .. } => "envelope_debezium",
447    }
448}
449
450/// Returns the `mz_as_of_queries_total` label values for an execution of `stmt`, or `None` if
451/// `stmt` is not a `SELECT` or `SUBSCRIBE` with an `AS OF` clause. A `COPY` counts as the
452/// statement it copies.
453pub(crate) fn as_of_query_label_values<T>(
454    session_type: &'static str,
455    application_name: ApplicationNameHint,
456    stmt: &Statement<T>,
457) -> Option<[&'static str; 4]>
458where
459    T: AstInfo,
460{
461    let (as_of, statement) = match stmt {
462        Statement::Select(SelectStatement {
463            as_of: Some(as_of), ..
464        })
465        | Statement::Copy(CopyStatement {
466            relation:
467                CopyRelation::Select(SelectStatement {
468                    as_of: Some(as_of), ..
469                }),
470            ..
471        }) => (as_of, "select"),
472        Statement::Subscribe(SubscribeStatement {
473            as_of: Some(as_of), ..
474        })
475        | Statement::Copy(CopyStatement {
476            relation:
477                CopyRelation::Subscribe(SubscribeStatement {
478                    as_of: Some(as_of), ..
479                }),
480            ..
481        }) => (as_of, "subscribe"),
482        _ => return None,
483    };
484    let kind = match as_of {
485        AsOf::At(_) => "at",
486        AsOf::AtLeast(_) => "at_least",
487    };
488    Some([session_type, kind, statement, application_name.as_str()])
489}
490
491/// Adapter metrics whose series carry a cluster id label.
492///
493/// Series are created on first observation, as for any labeled metric, and
494/// `remove_cluster` deletes a dropped cluster's series from every vec that
495/// `register` put here.
496#[derive(Debug, Clone)]
497pub struct ClusterLabeledMetrics {
498    time_to_first_row_seconds: HistogramVec,
499    determine_timestamp: IntCounterVec,
500    timestamp_difference_for_bounded_staleness_ms: HistogramVec,
501    /// Every vec above with the label that carries its cluster id. `register`
502    /// is the only way in, so a vec registered any other way is not swept.
503    delete_on_cluster_drop: Vec<ClusterLabeledVec>,
504}
505
506/// A metric vec with the name of the label that carries the cluster id.
507#[derive(Debug, Clone)]
508enum ClusterLabeledVec {
509    Histogram(HistogramVec, &'static str),
510    Counter(IntCounterVec, &'static str),
511}
512
513impl ClusterLabeledVec {
514    fn delete_cluster(&self, cluster: &str) {
515        match self {
516            Self::Histogram(vec, label) => remove_children_with_label(vec, label, cluster),
517            Self::Counter(vec, label) => remove_children_with_label(vec, label, cluster),
518        }
519    }
520}
521
522impl ClusterLabeledMetrics {
523    fn register_into(registry: &MetricsRegistry) -> Self {
524        let mut vecs = Vec::new();
525        Self {
526            time_to_first_row_seconds: Self::register(
527                registry,
528                &mut vecs,
529                ClusterLabeledVec::Histogram,
530                "instance_id",
531                metric! {
532                    name: "mz_time_to_first_row_seconds",
533                    help: "Latency of an execute for a successful query from pgwire's perspective",
534                    var_labels: ["instance_id", "isolation_level", "strategy", "application_name"],
535                    // NOTE: Measurements below 512 microseconds are negligible, so omit those buckets.
536                    buckets: histogram_seconds_buckets(0.000_512, 32.0)
537                },
538            ),
539            determine_timestamp: Self::register(
540                registry,
541                &mut vecs,
542                ClusterLabeledVec::Counter,
543                "compute_instance",
544                metric!(
545                    name: "mz_determine_timestamp",
546                    help: "The total number of calls to determine_timestamp.",
547                    var_labels: ["respond_immediately", "isolation_level", "compute_instance"],
548                ),
549            ),
550            timestamp_difference_for_bounded_staleness_ms: Self::register(
551                registry,
552                &mut vecs,
553                ClusterLabeledVec::Histogram,
554                "compute_instance",
555                metric!(
556                    name: "mz_timestamp_difference_for_bounded_staleness_ms",
557                    help: "How much older bounded-staleness timestamps are compared to serializable, in milliseconds. Measures the actual staleness incurred.",
558                    var_labels: ["compute_instance"],
559                    buckets: histogram_milliseconds_buckets(1., 8000.),
560                ),
561            ),
562            delete_on_cluster_drop: vecs,
563        }
564    }
565
566    /// Registers `opts` and records it, wrapped by `wrap`, with `cluster_label`
567    /// as the label carrying its cluster id, so `remove_cluster` covers it.
568    ///
569    /// Panics if `cluster_label` is not a variable label of `opts`.
570    fn register<M: MakeCollector>(
571        registry: &MetricsRegistry,
572        vecs: &mut Vec<ClusterLabeledVec>,
573        wrap: fn(M, &'static str) -> ClusterLabeledVec,
574        cluster_label: &'static str,
575        opts: MakeCollectorOpts,
576    ) -> M {
577        assert!(
578            opts.opts
579                .variable_labels
580                .iter()
581                .any(|label| label == cluster_label),
582            "{cluster_label} is not a label of {}",
583            opts.opts.name
584        );
585        let vec: M = registry.register(opts);
586        // A metric vec is a handle onto shared state, so the clone sees the
587        // same children as the field.
588        vecs.push(wrap(vec.clone(), cluster_label));
589        vec
590    }
591
592    /// Deletes the series of `cluster_id`.
593    pub(crate) fn remove_cluster(&self, cluster_id: ClusterId) {
594        let cluster = cluster_id.to_string();
595        for vec in &self.delete_on_cluster_drop {
596            vec.delete_cluster(&cluster);
597        }
598    }
599
600    /// Statements without a cluster or strategy record under the "none" label value.
601    pub(crate) fn time_to_first_row_seconds(
602        &self,
603        cluster_id: Option<ClusterId>,
604        isolation_level: IsolationLevel,
605        strategy: Option<StatementExecutionStrategy>,
606        application_name: ApplicationNameHint,
607    ) -> Histogram {
608        let instance = match cluster_id {
609            Some(id) => Cow::Owned(id.to_string()),
610            None => Cow::Borrowed("none"),
611        };
612        self.time_to_first_row_seconds.with_label_values(&[
613            instance.as_ref(),
614            isolation_level.as_variant_str(),
615            strategy.map_or("none", |strategy| strategy.name()),
616            application_name.as_str(),
617        ])
618    }
619
620    pub(crate) fn determine_timestamp(
621        &self,
622        cluster_id: ClusterId,
623        respond_immediately: bool,
624        isolation_level: IsolationLevel,
625    ) -> GenericCounter<AtomicU64> {
626        self.determine_timestamp.with_label_values(&[
627            if respond_immediately { "true" } else { "false" },
628            isolation_level.as_variant_str(),
629            &cluster_id.to_string(),
630        ])
631    }
632
633    pub(crate) fn timestamp_difference_for_bounded_staleness_ms(
634        &self,
635        cluster_id: ClusterId,
636    ) -> Histogram {
637        self.timestamp_difference_for_bounded_staleness_ms
638            .with_label_values(&[&cluster_id.to_string()])
639    }
640}
641
642#[cfg(test)]
643mod tests {
644    use std::collections::BTreeSet;
645
646    use super::*;
647
648    const U1: ClusterId = ClusterId::User(1);
649    const U2: ClusterId = ClusterId::User(2);
650
651    fn families(registry: &MetricsRegistry) -> BTreeSet<String> {
652        registry
653            .gather()
654            .into_iter()
655            .map(|family| family.name().to_string())
656            .collect()
657    }
658
659    /// Names of the families with a series carrying `value` under any label.
660    fn families_with_label_value(registry: &MetricsRegistry, value: &str) -> BTreeSet<String> {
661        registry
662            .gather()
663            .into_iter()
664            .filter(|family| {
665                family
666                    .get_metric()
667                    .iter()
668                    .flat_map(|metric| metric.get_label())
669                    .any(|label| label.value() == value)
670            })
671            .map(|family| family.name().to_string())
672            .collect()
673    }
674
675    fn observe_first_row(
676        metrics: &ClusterLabeledMetrics,
677        cluster_id: Option<ClusterId>,
678        strategy: Option<StatementExecutionStrategy>,
679    ) {
680        metrics
681            .time_to_first_row_seconds(
682                cluster_id,
683                IsolationLevel::StrictSerializable,
684                strategy,
685                ApplicationNameHint::from_str("psql"),
686            )
687            .observe(0.1);
688    }
689
690    #[mz_ore::test]
691    fn dropping_a_cluster_removes_its_series() {
692        let registry = MetricsRegistry::new();
693        let metrics = ClusterLabeledMetrics::register_into(&registry);
694
695        observe_first_row(&metrics, Some(U1), None);
696        observe_first_row(
697            &metrics,
698            Some(U1),
699            Some(StatementExecutionStrategy::FastPath),
700        );
701        observe_first_row(&metrics, Some(U2), None);
702        metrics
703            .determine_timestamp(U1, true, IsolationLevel::Serializable)
704            .inc();
705        metrics
706            .timestamp_difference_for_bounded_staleness_ms(U1)
707            .observe(5.0);
708        assert_eq!(
709            families_with_label_value(&registry, "u1"),
710            families(&registry),
711            "every registered family needs a u1 series for the drop to be exercised"
712        );
713
714        metrics.remove_cluster(U1);
715
716        assert_eq!(
717            families_with_label_value(&registry, "u1"),
718            BTreeSet::new(),
719            "u1 series survived the drop"
720        );
721        assert_eq!(
722            families_with_label_value(&registry, "u2"),
723            BTreeSet::from(["mz_time_to_first_row_seconds".to_string()]),
724            "the other cluster's series must be untouched"
725        );
726    }
727
728    #[mz_ore::test]
729    fn statements_without_a_cluster_are_unaffected_by_drops() {
730        let registry = MetricsRegistry::new();
731        let metrics = ClusterLabeledMetrics::register_into(&registry);
732        observe_first_row(&metrics, None, Some(StatementExecutionStrategy::Constant));
733        observe_first_row(&metrics, Some(U1), None);
734        metrics.remove_cluster(U1);
735        assert_eq!(families_with_label_value(&registry, "u1"), BTreeSet::new());
736        assert_eq!(
737            families_with_label_value(&registry, "none"),
738            BTreeSet::from(["mz_time_to_first_row_seconds".to_string()])
739        );
740    }
741
742    #[mz_ore::test]
743    #[should_panic(expected = "is not a label of mz_test")]
744    fn registering_under_a_label_the_metric_lacks_panics() {
745        let registry = MetricsRegistry::new();
746        let _: IntCounterVec = ClusterLabeledMetrics::register(
747            &registry,
748            &mut Vec::new(),
749            ClusterLabeledVec::Counter,
750            "cluster_id",
751            metric!(
752                name: "mz_test",
753                help: "test",
754                var_labels: ["compute_instance"],
755            ),
756        );
757    }
758}
mz_adapter/metrics.rs

mz_adapter/
metrics.rs
