mz_storage/metrics/source/
sql_server.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 Postgres.
11
12use std::collections::BTreeMap;
13use std::rc::Rc;
14use std::sync::Mutex;
15
16use mz_ore::metric;
17use mz_ore::metrics::{
18    DeregisterOnDropCounter, DeregisterOnDropGauge, DeregisterOnDropVec, MetricsRegistry, raw,
19};
20use mz_repr::GlobalId;
21use prometheus::core::{AtomicF64, AtomicI64, AtomicU64};
22
23/// Definitions for Postgres source metrics.
24#[derive(Clone, Debug)]
25pub(crate) struct SqlServerSourceMetricDefs {
26    pub(crate) ignored_messages: DeregisterOnDropVec<raw::IntCounterVec>,
27    pub(crate) insert_messages: DeregisterOnDropVec<raw::IntCounterVec>,
28    pub(crate) update_messages: DeregisterOnDropVec<raw::IntCounterVec>,
29    pub(crate) delete_messages: DeregisterOnDropVec<raw::IntCounterVec>,
30    pub(crate) snapshot_table_count: DeregisterOnDropVec<raw::UIntGaugeVec>,
31    pub(crate) snapshot_table_size_latency: DeregisterOnDropVec<raw::GaugeVec>,
32    pub(crate) snapshot_table_lock: DeregisterOnDropVec<raw::IntGaugeVec>,
33}
34
35impl SqlServerSourceMetricDefs {
36    pub(crate) fn register_with(registry: &MetricsRegistry) -> Self {
37        Self {
38            ignored_messages: registry.register(metric!(
39                name: "mz_sql_server_per_source_ignored_messages",
40                help: "The number of messages ignored because of an irrelevant type or relation_id",
41                var_labels: ["source_id", "worker_id"],
42            )),
43            insert_messages: registry.register(metric!(
44                name: "mz_sql_server_per_source_inserts",
45                help: "The number of inserts for all tables in this source",
46                var_labels: ["source_id", "worker_id"],
47            )),
48            update_messages: registry.register(metric!(
49                name: "mz_sql_server_per_source_updates",
50                help: "The number of updates for all tables in this source",
51                var_labels: ["source_id", "worker_id"],
52            )),
53            delete_messages: registry.register(metric!(
54                name: "mz_sql_server_per_source_deletes",
55                help: "The number of deletes for all tables in this source",
56                var_labels: ["source_id", "worker_id"],
57            )),
58            snapshot_table_count: registry.register(metric!(
59                name: "mz_sql_server_snapshot_table_count",
60                help: "The number of tables that SQL Server still needs to snapshot",
61                var_labels: ["source_id", "worker_id"],
62            )),
63            snapshot_table_size_latency: registry.register(metric!(
64                name: "mz_sql_server_snapshot_count_latency",
65                help: "The wall time used to obtain snapshot sizes.",
66                var_labels: ["source_id", "worker_id", "table_name"],
67            )),
68            snapshot_table_lock: registry.register(metric!(
69                name: "mz_sql_server_snapshot_table_lock",
70                help: "The upstream tables locked for snapshot.",
71                var_labels: ["source_id", "worker_id", "table_name"],
72            )),
73        }
74    }
75}
76#[derive(Clone)]
77/// Metrics for Postgres sources.
78pub(crate) struct SqlServerSourceMetrics {
79    // stored as String to avoid having to convert them repeatedly.
80    source_id: String,
81    worker_id: String,
82    defs: SqlServerSourceMetricDefs,
83    // Currently, this structure is not accessed across threads.
84    snapshot_table_size_latency: Rc<Mutex<BTreeMap<String, DeregisterOnDropGauge<AtomicF64>>>>,
85    snapshot_table_lock_count: Rc<Mutex<BTreeMap<String, DeregisterOnDropGauge<AtomicI64>>>>,
86
87    pub(crate) inserts: DeregisterOnDropCounter<AtomicU64>,
88    pub(crate) updates: DeregisterOnDropCounter<AtomicU64>,
89    pub(crate) deletes: DeregisterOnDropCounter<AtomicU64>,
90    pub(crate) ignored: DeregisterOnDropCounter<AtomicU64>,
91    pub(crate) snapshot_table_count: DeregisterOnDropGauge<AtomicU64>,
92}
93
94impl SqlServerSourceMetrics {
95    /// Create a `SqlServerSourceMetrics` from the `SqlServerSourceMetricDefs`.
96    pub(crate) fn new(
97        defs: &SqlServerSourceMetricDefs,
98        source_id: GlobalId,
99        worker_id: usize,
100    ) -> Self {
101        let source_id_labels = &[source_id.to_string(), worker_id.to_string()];
102        Self {
103            source_id: source_id.to_string(),
104            worker_id: worker_id.to_string(),
105            defs: defs.clone(),
106            inserts: defs
107                .insert_messages
108                .get_deregister_on_drop_metric(source_id_labels.to_vec()),
109            updates: defs
110                .update_messages
111                .get_deregister_on_drop_metric(source_id_labels.to_vec()),
112            deletes: defs
113                .delete_messages
114                .get_deregister_on_drop_metric(source_id_labels.to_vec()),
115            ignored: defs
116                .ignored_messages
117                .get_deregister_on_drop_metric(source_id_labels.to_vec()),
118            snapshot_table_count: defs
119                .snapshot_table_count
120                .get_deregister_on_drop_metric(source_id_labels.to_vec()),
121            snapshot_table_size_latency: Default::default(),
122            snapshot_table_lock_count: Default::default(),
123        }
124    }
125
126    pub fn set_snapshot_table_size_latency(&self, table_name: &str, latency: f64) {
127        let mut snapshot_table_size_latency =
128            self.snapshot_table_size_latency.lock().expect("poisoned");
129        match snapshot_table_size_latency.entry(table_name.to_string()) {
130            std::collections::btree_map::Entry::Vacant(vacant_entry) => {
131                let labels = vec![
132                    self.source_id.clone(),
133                    self.worker_id.clone(),
134                    table_name.to_string(),
135                ];
136                let metric = self
137                    .defs
138                    .snapshot_table_size_latency
139                    .get_deregister_on_drop_metric(labels);
140                vacant_entry.insert(metric).set(latency);
141            }
142            std::collections::btree_map::Entry::Occupied(occupied_entry) => {
143                occupied_entry.get().set(latency)
144            }
145        }
146    }
147
148    pub fn update_snapshot_table_lock_count(&self, table_name: &str, delta: i64) {
149        let mut snapshot_table_lock_count =
150            self.snapshot_table_lock_count.lock().expect("poisoned");
151        match snapshot_table_lock_count.entry(table_name.to_string()) {
152            std::collections::btree_map::Entry::Vacant(vacant_entry) => {
153                let labels = vec![
154                    self.source_id.clone(),
155                    self.worker_id.clone(),
156                    table_name.to_string(),
157                ];
158                let metric = self
159                    .defs
160                    .snapshot_table_lock
161                    .get_deregister_on_drop_metric(labels);
162                vacant_entry.insert(metric).add(delta);
163            }
164            std::collections::btree_map::Entry::Occupied(occupied_entry) => {
165                occupied_entry.get().add(delta);
166            }
167        }
168    }
169}
mz_storage/metrics/source/sql_server.rs

mz_storage/metrics/source/
sql_server.rs
