mz_storage/metrics/
upsert.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 upsert sources.
11
12use std::collections::BTreeMap;
13use std::sync::{Arc, Mutex, Weak};
14
15use mz_ore::metric;
16use mz_ore::metrics::{
17    DeleteOnDropHistogram, DeregisterOnDropCounter, DeregisterOnDropGauge, DeregisterOnDropVec,
18    HistogramVec, MetricsRegistry, raw,
19};
20use mz_ore::stats::histogram_seconds_buckets;
21use mz_repr::GlobalId;
22use mz_rocksdb::RocksDBInstanceMetrics;
23use mz_storage_operators::metrics::BackpressureOperatorMetrics;
24use prometheus::core::{AtomicF64, AtomicU64};
25
26/// Metric definitions for the `upsert` operator.
27#[derive(Clone, Debug)]
28pub(crate) struct UpsertMetricDefs {
29    pub(crate) rehydration_latency: DeregisterOnDropVec<raw::GaugeVec>,
30    pub(crate) rehydration_total: DeregisterOnDropVec<raw::UIntGaugeVec>,
31    pub(crate) rehydration_updates: DeregisterOnDropVec<raw::UIntGaugeVec>,
32
33    // These are used by `shared`.
34    pub(crate) merge_snapshot_latency: HistogramVec,
35    pub(crate) merge_snapshot_updates: DeregisterOnDropVec<raw::IntCounterVec>,
36    pub(crate) merge_snapshot_inserts: DeregisterOnDropVec<raw::IntCounterVec>,
37    pub(crate) merge_snapshot_deletes: DeregisterOnDropVec<raw::IntCounterVec>,
38    pub(crate) upsert_inserts: DeregisterOnDropVec<raw::IntCounterVec>,
39    pub(crate) upsert_updates: DeregisterOnDropVec<raw::IntCounterVec>,
40    pub(crate) upsert_deletes: DeregisterOnDropVec<raw::IntCounterVec>,
41    pub(crate) multi_get_latency: HistogramVec,
42    pub(crate) multi_get_size: DeregisterOnDropVec<raw::IntCounterVec>,
43    pub(crate) multi_get_result_count: DeregisterOnDropVec<raw::IntCounterVec>,
44    pub(crate) multi_get_result_bytes: DeregisterOnDropVec<raw::IntCounterVec>,
45    pub(crate) multi_put_latency: HistogramVec,
46    pub(crate) multi_put_size: DeregisterOnDropVec<raw::IntCounterVec>,
47    pub(crate) restashed: DeregisterOnDropVec<raw::IntCounterVec>,
48
49    // These are used by `rocksdb`.
50    pub(crate) rocksdb_multi_get_latency: HistogramVec,
51    pub(crate) rocksdb_multi_get_size: DeregisterOnDropVec<raw::IntCounterVec>,
52    pub(crate) rocksdb_multi_get_result_count: DeregisterOnDropVec<raw::IntCounterVec>,
53    pub(crate) rocksdb_multi_get_result_bytes: DeregisterOnDropVec<raw::IntCounterVec>,
54    pub(crate) rocksdb_multi_get_count: DeregisterOnDropVec<raw::IntCounterVec>,
55    pub(crate) rocksdb_multi_put_count: DeregisterOnDropVec<raw::IntCounterVec>,
56    pub(crate) rocksdb_multi_put_latency: HistogramVec,
57    pub(crate) rocksdb_multi_put_size: DeregisterOnDropVec<raw::IntCounterVec>,
58    // These are maps so that multiple timely workers can interact with the same
59    // `DeleteOnDropHistogram`, which is only dropped once ALL workers drop it.
60    // The map may contain arbitrary, old `Weak`s for deleted sources, which are
61    // only cleaned if those sources are recreated.
62    //
63    // We don't parameterize these by `worker_id` like the `rehydration_*` ones
64    // to save on time-series cardinality.
65    pub(crate) shared: Arc<Mutex<BTreeMap<GlobalId, Weak<UpsertSharedMetrics>>>>,
66    pub(crate) rocksdb_shared:
67        Arc<Mutex<BTreeMap<GlobalId, Weak<mz_rocksdb::RocksDBSharedMetrics>>>>,
68}
69
70impl UpsertMetricDefs {
71    pub(crate) fn register_with(registry: &MetricsRegistry) -> Self {
72        let shared = Arc::new(Mutex::new(BTreeMap::new()));
73        let rocksdb_shared = Arc::new(Mutex::new(BTreeMap::new()));
74        Self {
75            rehydration_latency: registry.register(metric!(
76                name: "mz_storage_upsert_state_rehydration_latency",
77                help: "The latency, per-worker, in fractional seconds, \
78                    of rehydrating the upsert state for this source",
79                var_labels: ["source_id", "worker_id"],
80            )),
81            rehydration_total: registry.register(metric!(
82                name: "mz_storage_upsert_state_rehydration_total",
83                help: "The number of values \
84                    per-worker, rehydrated into the upsert state for \
85                    this source",
86                var_labels: ["source_id", "worker_id"],
87            )),
88            rehydration_updates: registry.register(metric!(
89                name: "mz_storage_upsert_state_rehydration_updates",
90                help: "The number of updates (both negative and positive), \
91                    per-worker, rehydrated into the upsert state for \
92                    this source",
93                var_labels: ["source_id", "worker_id"],
94            )),
95            // Choose a relatively low number of representative buckets.
96            merge_snapshot_latency: registry.register(metric!(
97                name: "mz_storage_upsert_merge_snapshot_latency",
98                help: "The latencies, in fractional seconds, \
99                    of merging snapshot updates into upsert state for this source. \
100                    Specific implementations of upsert state may have more detailed \
101                    metrics about sub-batches.",
102                var_labels: ["source_id"],
103                buckets: histogram_seconds_buckets(0.000_500, 32.0),
104            )),
105            merge_snapshot_updates: registry.register(metric!(
106                name: "mz_storage_upsert_merge_snapshot_updates_total",
107                help: "The batch size, \
108                    of merging snapshot updates into upsert state for this source. \
109                    Specific implementations of upsert state may have more detailed \
110                    metrics about sub-batches.",
111                var_labels: ["source_id", "worker_id"],
112            )),
113            merge_snapshot_inserts: registry.register(metric!(
114                name: "mz_storage_upsert_merge_snapshot_inserts_total",
115                help: "The number of inserts in a batch for merging snapshot updates \
116                    for this source.",
117                var_labels: ["source_id", "worker_id"],
118            )),
119            merge_snapshot_deletes: registry.register(metric!(
120                name: "mz_storage_upsert_merge_snapshot_deletes_total",
121                help: "The number of deletes in a batch for merging snapshot updates \
122                    for this source.",
123                var_labels: ["source_id", "worker_id"],
124            )),
125            upsert_inserts: registry.register(metric!(
126                name: "mz_storage_upsert_inserts_total",
127                help: "The number of inserts done by the upsert operator",
128                var_labels: ["source_id", "worker_id"],
129            )),
130            upsert_updates: registry.register(metric!(
131                name: "mz_storage_upsert_updates_total",
132                help: "The number of updates done by the upsert operator",
133                var_labels: ["source_id", "worker_id"],
134            )),
135            upsert_deletes: registry.register(metric!(
136                name: "mz_storage_upsert_deletes_total",
137                help: "The number of deletes done by the upsert operator.",
138                var_labels: ["source_id", "worker_id"],
139            )),
140            multi_get_latency: registry.register(metric!(
141                name: "mz_storage_upsert_multi_get_latency",
142                help: "The latencies, in fractional seconds, \
143                    of getting values from the upsert state for this source. \
144                    Specific implementations of upsert state may have more detailed \
145                    metrics about sub-batches.",
146                var_labels: ["source_id"],
147                buckets: histogram_seconds_buckets(0.000_500, 32.0),
148            )),
149            multi_get_size: registry.register(metric!(
150                name: "mz_storage_upsert_multi_get_size_total",
151                help: "The batch size, \
152                    of getting values from the upsert state for this source. \
153                    Specific implementations of upsert state may have more detailed \
154                    metrics about sub-batches.",
155                var_labels: ["source_id", "worker_id"],
156            )),
157            multi_get_result_count: registry.register(metric!(
158                name: "mz_storage_upsert_multi_get_result_count_total",
159                help: "The number of non-empty records returned in a multi_get batch. \
160                    Specific implementations of upsert state may have more detailed \
161                    metrics about sub-batches.",
162                var_labels: ["source_id", "worker_id"],
163            )),
164            multi_get_result_bytes: registry.register(metric!(
165                name: "mz_storage_upsert_multi_get_result_bytes_total",
166                help: "The total size of records returned in a multi_get batch. \
167                    Specific implementations of upsert state may have more detailed \
168                    metrics about sub-batches.",
169                var_labels: ["source_id", "worker_id"],
170            )),
171            multi_put_latency: registry.register(metric!(
172                name: "mz_storage_upsert_multi_put_latency",
173                help: "The latencies, in fractional seconds, \
174                    of getting values into the upsert state for this source. \
175                    Specific implementations of upsert state may have more detailed \
176                    metrics about sub-batches.",
177                var_labels: ["source_id"],
178                buckets: histogram_seconds_buckets(0.000_500, 32.0),
179            )),
180            multi_put_size: registry.register(metric!(
181                name: "mz_storage_upsert_multi_put_size_total",
182                help: "The batch size, \
183                    of getting values into the upsert state for this source. \
184                    Specific implementations of upsert state may have more detailed \
185                    metrics about sub-batches.",
186                var_labels: ["source_id", "worker_id"],
187            )),
188            restashed: registry.register(metric!(
189                name: "mz_storage_upsert_restashed_total",
190                help: "The number of updates the upsert v2 operator extracted from its \
191                    stash before the persist feedback frontier reached their time, and \
192                    pushed back.",
193                var_labels: ["source_id", "worker_id"],
194            )),
195            shared,
196            rocksdb_multi_get_latency: registry.register(metric!(
197                name: "mz_storage_rocksdb_multi_get_latency",
198                help: "The latencies, in fractional seconds, \
199                    of getting batches of values from RocksDB for this source.",
200                var_labels: ["source_id"],
201                buckets: histogram_seconds_buckets(0.000_500, 32.0),
202            )),
203            rocksdb_multi_get_size: registry.register(metric!(
204                name: "mz_storage_rocksdb_multi_get_size_total",
205                help: "The batch size, \
206                    of getting batches of values from RocksDB for this source.",
207                var_labels: ["source_id", "worker_id"],
208            )),
209            rocksdb_multi_get_result_count: registry.register(metric!(
210                name: "mz_storage_rocksdb_multi_get_result_count_total",
211                help: "The number of non-empty records returned, \
212                    when getting batches of values from RocksDB for this source.",
213                var_labels: ["source_id", "worker_id"],
214            )),
215            rocksdb_multi_get_result_bytes: registry.register(metric!(
216                name: "mz_storage_rocksdb_multi_get_result_bytes_total",
217                help: "The total size of records returned, \
218                    when getting batches of values from RocksDB for this source.",
219                var_labels: ["source_id", "worker_id"],
220            )),
221            rocksdb_multi_get_count: registry.register(metric!(
222                name: "mz_storage_rocksdb_multi_get_count_total",
223                help: "The number of calls to rocksdb multi_get.",
224                var_labels: ["source_id", "worker_id"],
225            )),
226            rocksdb_multi_put_count: registry.register(metric!(
227                name: "mz_storage_rocksdb_multi_put_count_total",
228                help: "The number of calls to rocksdb multi_put.",
229                var_labels: ["source_id", "worker_id"],
230            )),
231            rocksdb_multi_put_latency: registry.register(metric!(
232                name: "mz_storage_rocksdb_multi_put_latency",
233                help: "The latencies, in fractional seconds, \
234                    of putting batches of values into RocksDB for this source.",
235                var_labels: ["source_id"],
236                buckets: histogram_seconds_buckets(0.000_500, 32.0),
237            )),
238            rocksdb_multi_put_size: registry.register(metric!(
239                name: "mz_storage_rocksdb_multi_put_size_total",
240                help: "The batch size, \
241                    of putting batches of values into RocksDB for this source.",
242                var_labels: ["source_id", "worker_id"],
243            )),
244            rocksdb_shared,
245        }
246    }
247
248    /// Get a shared-across-workers instance of an `UpsertSharedMetrics`.
249    pub(crate) fn shared(&self, source_id: &GlobalId) -> Arc<UpsertSharedMetrics> {
250        let mut shared = self.shared.lock().expect("mutex poisoned");
251        if let Some(shared_metrics) = shared.get(source_id) {
252            if let Some(shared_metrics) = shared_metrics.upgrade() {
253                return Arc::clone(&shared_metrics);
254            } else {
255                assert!(shared.remove(source_id).is_some());
256            }
257        }
258        let shared_metrics = Arc::new(UpsertSharedMetrics::new(self, *source_id));
259        assert!(
260            shared
261                .insert(source_id.clone(), Arc::downgrade(&shared_metrics))
262                .is_none()
263        );
264        shared_metrics
265    }
266
267    /// Get a shared-across-workers instance of an `RocksDBSharedMetrics`
268    pub(crate) fn rocksdb_shared(
269        &self,
270        source_id: &GlobalId,
271    ) -> Arc<mz_rocksdb::RocksDBSharedMetrics> {
272        let mut rocksdb = self.rocksdb_shared.lock().expect("mutex poisoned");
273        if let Some(shared_metrics) = rocksdb.get(source_id) {
274            if let Some(shared_metrics) = shared_metrics.upgrade() {
275                return Arc::clone(&shared_metrics);
276            } else {
277                assert!(rocksdb.remove(source_id).is_some());
278            }
279        }
280
281        let rocksdb_metrics = {
282            let source_id = source_id.to_string();
283            mz_rocksdb::RocksDBSharedMetrics {
284                multi_get_latency: self
285                    .rocksdb_multi_get_latency
286                    .get_delete_on_drop_metric(vec![source_id.clone()]),
287                multi_put_latency: self
288                    .rocksdb_multi_put_latency
289                    .get_delete_on_drop_metric(vec![source_id.clone()]),
290            }
291        };
292
293        let rocksdb_metrics = Arc::new(rocksdb_metrics);
294        assert!(
295            rocksdb
296                .insert(source_id.clone(), Arc::downgrade(&rocksdb_metrics))
297                .is_none()
298        );
299        rocksdb_metrics
300    }
301}
302
303/// Metrics for upsert source shared across workers.
304#[derive(Debug)]
305pub(crate) struct UpsertSharedMetrics {
306    pub(crate) merge_snapshot_latency: DeleteOnDropHistogram<Vec<String>>,
307    pub(crate) multi_get_latency: DeleteOnDropHistogram<Vec<String>>,
308    pub(crate) multi_put_latency: DeleteOnDropHistogram<Vec<String>>,
309}
310
311impl UpsertSharedMetrics {
312    /// Create an `UpsertSharedMetrics` from the `UpsertMetricDefs`
313    fn new(metrics: &UpsertMetricDefs, source_id: GlobalId) -> Self {
314        let source_id = source_id.to_string();
315        UpsertSharedMetrics {
316            merge_snapshot_latency: metrics
317                .merge_snapshot_latency
318                .get_delete_on_drop_metric(vec![source_id.clone()]),
319            multi_get_latency: metrics
320                .multi_get_latency
321                .get_delete_on_drop_metric(vec![source_id.clone()]),
322            multi_put_latency: metrics
323                .multi_put_latency
324                .get_delete_on_drop_metric(vec![source_id.clone()]),
325        }
326    }
327}
328
329/// Metrics related to backpressure in `UPSERT` dataflows.
330#[derive(Clone, Debug)]
331pub(crate) struct UpsertBackpressureMetricDefs {
332    pub(crate) emitted_bytes: DeregisterOnDropVec<raw::IntCounterVec>,
333    pub(crate) last_backpressured_bytes: DeregisterOnDropVec<raw::UIntGaugeVec>,
334    pub(crate) retired_bytes: DeregisterOnDropVec<raw::IntCounterVec>,
335}
336
337impl UpsertBackpressureMetricDefs {
338    pub(crate) fn register_with(registry: &MetricsRegistry) -> Self {
339        Self {
340            emitted_bytes: registry.register(metric!(
341                name: "mz_storage_upsert_backpressure_emitted_bytes",
342                help: "A counter with the number of emitted bytes.",
343                var_labels: ["source_id", "worker_id"],
344            )),
345            last_backpressured_bytes: registry.register(metric!(
346                name: "mz_storage_upsert_backpressure_last_backpressured_bytes",
347                help: "The last count of bytes we are waiting to be retired in \
348                    the operator. This cannot be directly compared to \
349                    `retired_bytes`, but CAN indicate that backpressure is happening.",
350                var_labels: ["source_id", "worker_id"],
351            )),
352            retired_bytes: registry.register(metric!(
353                name: "mz_storage_upsert_backpressure_retired_bytes",
354                help:"A counter with the number of bytes retired by downstream processing.",
355                var_labels: ["source_id", "worker_id"],
356            )),
357        }
358    }
359}
360
361/// Metrics for the `upsert` operator.
362pub struct UpsertMetrics {
363    pub(crate) rehydration_latency: DeregisterOnDropGauge<AtomicF64>,
364    pub(crate) rehydration_total: DeregisterOnDropGauge<AtomicU64>,
365    pub(crate) rehydration_updates: DeregisterOnDropGauge<AtomicU64>,
366
367    pub(crate) merge_snapshot_updates: DeregisterOnDropCounter<AtomicU64>,
368    pub(crate) merge_snapshot_inserts: DeregisterOnDropCounter<AtomicU64>,
369    pub(crate) merge_snapshot_deletes: DeregisterOnDropCounter<AtomicU64>,
370    pub(crate) upsert_inserts: DeregisterOnDropCounter<AtomicU64>,
371    pub(crate) upsert_updates: DeregisterOnDropCounter<AtomicU64>,
372    pub(crate) upsert_deletes: DeregisterOnDropCounter<AtomicU64>,
373    pub(crate) multi_get_size: DeregisterOnDropCounter<AtomicU64>,
374    pub(crate) multi_get_result_bytes: DeregisterOnDropCounter<AtomicU64>,
375    pub(crate) multi_get_result_count: DeregisterOnDropCounter<AtomicU64>,
376    pub(crate) multi_put_size: DeregisterOnDropCounter<AtomicU64>,
377    pub(crate) restashed: DeregisterOnDropCounter<AtomicU64>,
378
379    pub(crate) shared: Arc<UpsertSharedMetrics>,
380    pub(crate) rocksdb_shared: Arc<mz_rocksdb::RocksDBSharedMetrics>,
381    pub(crate) rocksdb_instance_metrics: Arc<mz_rocksdb::RocksDBInstanceMetrics>,
382    // Owning the backpressure series here keeps them registered for the life
383    // of the upsert operator, past the `persist_source` operator that reports
384    // into them.
385    _backpressure_metrics: Option<UpsertBackpressureMetrics>,
386}
387
388/// The per-source, per-worker backpressure series for one upsert operator.
389///
390/// Owning this keeps the series registered. [Self::operator_metrics] hands the
391/// `backpressure` operator plain handles to them.
392#[derive(Debug)]
393pub(crate) struct UpsertBackpressureMetrics {
394    emitted_bytes: DeregisterOnDropCounter<AtomicU64>,
395    last_backpressured_bytes: DeregisterOnDropGauge<AtomicU64>,
396    retired_bytes: DeregisterOnDropCounter<AtomicU64>,
397}
398
399impl UpsertBackpressureMetrics {
400    pub(crate) fn new(defs: &UpsertBackpressureMetricDefs, id: GlobalId, worker_id: usize) -> Self {
401        let labels = vec![id.to_string(), worker_id.to_string()];
402        UpsertBackpressureMetrics {
403            emitted_bytes: defs
404                .emitted_bytes
405                .get_deregister_on_drop_metric(labels.clone()),
406            last_backpressured_bytes: defs
407                .last_backpressured_bytes
408                .get_deregister_on_drop_metric(labels.clone()),
409            retired_bytes: defs.retired_bytes.get_deregister_on_drop_metric(labels),
410        }
411    }
412
413    pub(crate) fn operator_metrics(&self) -> BackpressureOperatorMetrics {
414        BackpressureOperatorMetrics::new(
415            (*self.emitted_bytes).clone(),
416            (*self.last_backpressured_bytes).clone(),
417            (*self.retired_bytes).clone(),
418        )
419    }
420}
421
422impl UpsertMetrics {
423    /// Create an `UpsertMetrics` from the `UpsertMetricDefs`.
424    pub(crate) fn new(
425        defs: &UpsertMetricDefs,
426        source_id: GlobalId,
427        worker_id: usize,
428        backpressure_metrics: Option<UpsertBackpressureMetrics>,
429    ) -> Self {
430        let source_id_s = source_id.to_string();
431        let worker_id = worker_id.to_string();
432        Self {
433            rehydration_latency: defs
434                .rehydration_latency
435                .get_deregister_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
436            rehydration_total: defs
437                .rehydration_total
438                .get_deregister_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
439            rehydration_updates: defs
440                .rehydration_updates
441                .get_deregister_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
442            merge_snapshot_updates: defs
443                .merge_snapshot_updates
444                .get_deregister_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
445            merge_snapshot_inserts: defs
446                .merge_snapshot_inserts
447                .get_deregister_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
448            merge_snapshot_deletes: defs
449                .merge_snapshot_deletes
450                .get_deregister_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
451            upsert_inserts: defs
452                .upsert_inserts
453                .get_deregister_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
454            upsert_updates: defs
455                .upsert_updates
456                .get_deregister_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
457            upsert_deletes: defs
458                .upsert_deletes
459                .get_deregister_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
460            multi_get_size: defs
461                .multi_get_size
462                .get_deregister_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
463            multi_get_result_count: defs
464                .multi_get_result_count
465                .get_deregister_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
466            multi_get_result_bytes: defs
467                .multi_get_result_bytes
468                .get_deregister_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
469            multi_put_size: defs
470                .multi_put_size
471                .get_deregister_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
472            restashed: defs
473                .restashed
474                .get_deregister_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
475
476            shared: defs.shared(&source_id),
477            rocksdb_shared: defs.rocksdb_shared(&source_id),
478            rocksdb_instance_metrics: Arc::new(RocksDBInstanceMetrics {
479                multi_get_size: defs
480                    .rocksdb_multi_get_size
481                    .get_deregister_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
482                multi_get_result_count: defs
483                    .rocksdb_multi_get_result_count
484                    .get_deregister_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
485                multi_get_result_bytes: defs
486                    .rocksdb_multi_get_result_bytes
487                    .get_deregister_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
488                multi_get_count: defs
489                    .rocksdb_multi_get_count
490                    .get_deregister_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
491                multi_put_count: defs
492                    .rocksdb_multi_put_count
493                    .get_deregister_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
494                multi_put_size: defs
495                    .rocksdb_multi_put_size
496                    .get_deregister_on_drop_metric(vec![source_id_s, worker_id]),
497            }),
498            _backpressure_metrics: backpressure_metrics,
499        }
500    }
501}
mz_storage/metrics/upsert.rs

mz_storage/metrics/
upsert.rs
