mz_storage/metrics/source/
mysql.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
10//! Metrics for MySQL.
11
12use std::sync::Arc;
13use std::sync::Mutex;
14
15use mz_ore::metric;
16use mz_ore::metrics::{
17    DeregisterOnDropCounter, DeregisterOnDropGauge, DeregisterOnDropVec, MetricsRegistry, raw,
18};
19use mz_repr::GlobalId;
20use prometheus::core::{AtomicF64, AtomicU64};
21
22#[derive(Clone, Debug)]
23pub(crate) struct MySqlSourceMetricDefs {
24    pub(crate) total_messages: DeregisterOnDropVec<raw::IntCounterVec>,
25    pub(crate) ignored_messages: DeregisterOnDropVec<raw::IntCounterVec>,
26    pub(crate) insert_rows: DeregisterOnDropVec<raw::IntCounterVec>,
27    pub(crate) update_rows: DeregisterOnDropVec<raw::IntCounterVec>,
28    pub(crate) delete_rows: DeregisterOnDropVec<raw::IntCounterVec>,
29    pub(crate) tables: DeregisterOnDropVec<raw::UIntGaugeVec>,
30    pub(crate) gtid_txids: DeregisterOnDropVec<raw::UIntGaugeVec>,
31
32    pub(crate) snapshot_defs: MySqlSnapshotMetricDefs,
33}
34
35impl MySqlSourceMetricDefs {
36    pub(crate) fn register_with(registry: &MetricsRegistry) -> Self {
37        Self {
38            total_messages: registry.register(metric!(
39                name: "mz_mysql_per_source_messages_total",
40                help: "The total number of replication messages for this source, not expected to be the sum of the other values.",
41                var_labels: ["source_id"],
42            )),
43            ignored_messages: registry.register(metric!(
44                name: "mz_mysql_per_source_ignored_messages",
45                help: "The number of messages ignored because of an irrelevant type or relation_id",
46                var_labels: ["source_id"],
47            )),
48            insert_rows: registry.register(metric!(
49                name: "mz_mysql_per_source_inserts",
50                help: "The number of inserts for all tables in this source",
51                var_labels: ["source_id"],
52            )),
53            update_rows: registry.register(metric!(
54                name: "mz_mysql_per_source_updates",
55                help: "The number of updates for all tables in this source",
56                var_labels: ["source_id"],
57            )),
58            delete_rows: registry.register(metric!(
59                name: "mz_mysql_per_source_deletes",
60                help: "The number of deletes for all tables in this source",
61                var_labels: ["source_id"],
62            )),
63            tables: registry.register(metric!(
64                name: "mz_mysql_per_source_tables_count",
65                help: "The number of upstream tables for this source",
66                var_labels: ["source_id"],
67            )),
68            gtid_txids: registry.register(metric!(
69                name: "mz_mysql_sum_gtid_txns",
70                help: "The sum of all transaction-ids committed for each GTID Source-ID UUID seen for this source",
71                var_labels: ["source_id"],
72            )),
73            snapshot_defs: MySqlSnapshotMetricDefs::register_with(registry),
74        }
75    }
76}
77
78/// Metrics for MySql sources.
79pub(crate) struct MySqlSourceMetrics {
80    pub(crate) inserts: DeregisterOnDropCounter<AtomicU64>,
81    pub(crate) updates: DeregisterOnDropCounter<AtomicU64>,
82    pub(crate) deletes: DeregisterOnDropCounter<AtomicU64>,
83    pub(crate) ignored: DeregisterOnDropCounter<AtomicU64>,
84    pub(crate) total: DeregisterOnDropCounter<AtomicU64>,
85    pub(crate) tables: DeregisterOnDropGauge<AtomicU64>,
86    pub(crate) gtid_txids: DeregisterOnDropGauge<AtomicU64>,
87    pub(crate) snapshot_metrics: MySqlSnapshotMetrics,
88}
89
90impl MySqlSourceMetrics {
91    /// Create a `MySqlSourceMetrics` from the `MySqlSourceMetricDefs`.
92    pub(crate) fn new(defs: &MySqlSourceMetricDefs, source_id: GlobalId) -> Self {
93        let labels = &[source_id.to_string()];
94        Self {
95            inserts: defs
96                .insert_rows
97                .get_deregister_on_drop_metric(labels.to_vec()),
98            updates: defs
99                .update_rows
100                .get_deregister_on_drop_metric(labels.to_vec()),
101            deletes: defs
102                .delete_rows
103                .get_deregister_on_drop_metric(labels.to_vec()),
104            ignored: defs
105                .ignored_messages
106                .get_deregister_on_drop_metric(labels.to_vec()),
107            total: defs
108                .total_messages
109                .get_deregister_on_drop_metric(labels.to_vec()),
110            tables: defs.tables.get_deregister_on_drop_metric(labels.to_vec()),
111            gtid_txids: defs
112                .gtid_txids
113                .get_deregister_on_drop_metric(labels.to_vec()),
114            snapshot_metrics: MySqlSnapshotMetrics {
115                source_id,
116                gauges: Default::default(),
117                defs: defs.snapshot_defs.clone(),
118            },
119        }
120    }
121}
122
123#[derive(Clone, Debug)]
124pub(crate) struct MySqlSnapshotMetricDefs {
125    pub(crate) table_count_latency: DeregisterOnDropVec<raw::GaugeVec>,
126}
127
128impl MySqlSnapshotMetricDefs {
129    pub(crate) fn register_with(registry: &MetricsRegistry) -> Self {
130        Self {
131            table_count_latency: registry.register(metric!(
132                name: "mz_mysql_snapshot_count_latency",
133                help: "The wall time used to obtain snapshot sizes.",
134                var_labels: ["source_id", "table_name", "schema"],
135            )),
136        }
137    }
138}
139
140#[derive(Clone)]
141pub(crate) struct MySqlSnapshotMetrics {
142    source_id: GlobalId,
143    // This has to be shared between tokio tasks and the replication operator, as the collection
144    // of these metrics happens once in those tasks, which do not live long enough to keep them
145    // alive.
146    gauges: Arc<Mutex<Vec<DeregisterOnDropGauge<AtomicF64>>>>,
147    defs: MySqlSnapshotMetricDefs,
148}
149
150impl MySqlSnapshotMetrics {
151    pub(crate) fn record_table_count_latency(
152        &self,
153        table_name: String,
154        schema: String,
155        latency: f64,
156    ) {
157        let latency_gauge = self
158            .defs
159            .table_count_latency
160            .get_deregister_on_drop_metric(vec![self.source_id.to_string(), table_name, schema]);
161        latency_gauge.set(latency);
162        self.gauges.lock().expect("poisoned").push(latency_gauge)
163    }
164}
mz_storage/metrics/source/mysql.rs

mz_storage/metrics/source/
mysql.rs
