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

mz_adapter/
metrics.rs
