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

mz_adapter/
metrics.rs
