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    DeleteOnDropCounter, DeleteOnDropGauge, DeleteOnDropHistogram, GaugeVec, HistogramVec,
18    IntCounterVec, MetricsRegistry, UIntGaugeVec,
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: GaugeVec,
30    pub(crate) rehydration_total: UIntGaugeVec,
31    pub(crate) rehydration_updates: UIntGaugeVec,
32
33    // These are used by `shared`.
34    pub(crate) merge_snapshot_latency: HistogramVec,
35    pub(crate) merge_snapshot_updates: IntCounterVec,
36    pub(crate) merge_snapshot_inserts: IntCounterVec,
37    pub(crate) merge_snapshot_deletes: IntCounterVec,
38    pub(crate) upsert_inserts: IntCounterVec,
39    pub(crate) upsert_updates: IntCounterVec,
40    pub(crate) upsert_deletes: IntCounterVec,
41    pub(crate) multi_get_latency: HistogramVec,
42    pub(crate) multi_get_size: IntCounterVec,
43    pub(crate) multi_get_result_count: IntCounterVec,
44    pub(crate) multi_get_result_bytes: IntCounterVec,
45    pub(crate) multi_put_latency: HistogramVec,
46    pub(crate) multi_put_size: IntCounterVec,
47    pub(crate) restashed: IntCounterVec,
48
49    // These are used by `rocksdb`.
50    pub(crate) rocksdb_multi_get_latency: HistogramVec,
51    pub(crate) rocksdb_multi_get_size: IntCounterVec,
52    pub(crate) rocksdb_multi_get_result_count: IntCounterVec,
53    pub(crate) rocksdb_multi_get_result_bytes: IntCounterVec,
54    pub(crate) rocksdb_multi_get_count: IntCounterVec,
55    pub(crate) rocksdb_multi_put_count: IntCounterVec,
56    pub(crate) rocksdb_multi_put_latency: HistogramVec,
57    pub(crate) rocksdb_multi_put_size: 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: IntCounterVec,
333    pub(crate) last_backpressured_bytes: UIntGaugeVec,
334    pub(crate) retired_bytes: IntCounterVec,
335}
336
337impl UpsertBackpressureMetricDefs {
338    pub(crate) fn register_with(registry: &MetricsRegistry) -> Self {
339        // We add a `worker_id` label here, even though only 1 worker is ever
340        // active, as this is the simplest way to avoid the non-active
341        // workers from un-registering metrics. This is consistent with how
342        // `persist_sink` metrics work.
343        Self {
344            emitted_bytes: registry.register(metric!(
345                name: "mz_storage_upsert_backpressure_emitted_bytes",
346                help: "A counter with the number of emitted bytes.",
347                var_labels: ["source_id", "worker_id"],
348            )),
349            last_backpressured_bytes: registry.register(metric!(
350                name: "mz_storage_upsert_backpressure_last_backpressured_bytes",
351                help: "The last count of bytes we are waiting to be retired in \
352                    the operator. This cannot be directly compared to \
353                    `retired_bytes`, but CAN indicate that backpressure is happening.",
354                var_labels: ["source_id", "worker_id"],
355            )),
356            retired_bytes: registry.register(metric!(
357                name: "mz_storage_upsert_backpressure_retired_bytes",
358                help:"A counter with the number of bytes retired by downstream processing.",
359                var_labels: ["source_id", "worker_id"],
360            )),
361        }
362    }
363}
364
365/// Metrics for the `upsert` operator.
366pub struct UpsertMetrics {
367    pub(crate) rehydration_latency: DeleteOnDropGauge<AtomicF64, Vec<String>>,
368    pub(crate) rehydration_total: DeleteOnDropGauge<AtomicU64, Vec<String>>,
369    pub(crate) rehydration_updates: DeleteOnDropGauge<AtomicU64, Vec<String>>,
370
371    pub(crate) merge_snapshot_updates: DeleteOnDropCounter<AtomicU64, Vec<String>>,
372    pub(crate) merge_snapshot_inserts: DeleteOnDropCounter<AtomicU64, Vec<String>>,
373    pub(crate) merge_snapshot_deletes: DeleteOnDropCounter<AtomicU64, Vec<String>>,
374    pub(crate) upsert_inserts: DeleteOnDropCounter<AtomicU64, Vec<String>>,
375    pub(crate) upsert_updates: DeleteOnDropCounter<AtomicU64, Vec<String>>,
376    pub(crate) upsert_deletes: DeleteOnDropCounter<AtomicU64, Vec<String>>,
377    pub(crate) multi_get_size: DeleteOnDropCounter<AtomicU64, Vec<String>>,
378    pub(crate) multi_get_result_bytes: DeleteOnDropCounter<AtomicU64, Vec<String>>,
379    pub(crate) multi_get_result_count: DeleteOnDropCounter<AtomicU64, Vec<String>>,
380    pub(crate) multi_put_size: DeleteOnDropCounter<AtomicU64, Vec<String>>,
381    pub(crate) restashed: DeleteOnDropCounter<AtomicU64, Vec<String>>,
382
383    pub(crate) shared: Arc<UpsertSharedMetrics>,
384    pub(crate) rocksdb_shared: Arc<mz_rocksdb::RocksDBSharedMetrics>,
385    pub(crate) rocksdb_instance_metrics: Arc<mz_rocksdb::RocksDBInstanceMetrics>,
386    // Owning the backpressure series here keeps them registered for the life
387    // of the upsert operator, past the `persist_source` operator that reports
388    // into them.
389    _backpressure_metrics: Option<UpsertBackpressureMetrics>,
390}
391
392/// The per-source, per-worker backpressure series for one upsert operator.
393///
394/// Owning this keeps the series registered. [Self::operator_metrics] hands the
395/// `backpressure` operator plain handles to them.
396#[derive(Debug)]
397pub(crate) struct UpsertBackpressureMetrics {
398    emitted_bytes: DeleteOnDropCounter<AtomicU64, Vec<String>>,
399    last_backpressured_bytes: DeleteOnDropGauge<AtomicU64, Vec<String>>,
400    retired_bytes: DeleteOnDropCounter<AtomicU64, Vec<String>>,
401}
402
403impl UpsertBackpressureMetrics {
404    pub(crate) fn new(defs: &UpsertBackpressureMetricDefs, id: GlobalId, worker_id: usize) -> Self {
405        let labels = vec![id.to_string(), worker_id.to_string()];
406        UpsertBackpressureMetrics {
407            emitted_bytes: defs.emitted_bytes.get_delete_on_drop_metric(labels.clone()),
408            last_backpressured_bytes: defs
409                .last_backpressured_bytes
410                .get_delete_on_drop_metric(labels.clone()),
411            retired_bytes: defs.retired_bytes.get_delete_on_drop_metric(labels),
412        }
413    }
414
415    pub(crate) fn operator_metrics(&self) -> BackpressureOperatorMetrics {
416        BackpressureOperatorMetrics::new(
417            (*self.emitted_bytes).clone(),
418            (*self.last_backpressured_bytes).clone(),
419            (*self.retired_bytes).clone(),
420        )
421    }
422}
423
424impl UpsertMetrics {
425    /// Create an `UpsertMetrics` from the `UpsertMetricDefs`.
426    pub(crate) fn new(
427        defs: &UpsertMetricDefs,
428        source_id: GlobalId,
429        worker_id: usize,
430        backpressure_metrics: Option<UpsertBackpressureMetrics>,
431    ) -> Self {
432        let source_id_s = source_id.to_string();
433        let worker_id = worker_id.to_string();
434        Self {
435            rehydration_latency: defs
436                .rehydration_latency
437                .get_delete_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
438            rehydration_total: defs
439                .rehydration_total
440                .get_delete_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
441            rehydration_updates: defs
442                .rehydration_updates
443                .get_delete_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
444            merge_snapshot_updates: defs
445                .merge_snapshot_updates
446                .get_delete_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
447            merge_snapshot_inserts: defs
448                .merge_snapshot_inserts
449                .get_delete_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
450            merge_snapshot_deletes: defs
451                .merge_snapshot_deletes
452                .get_delete_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
453            upsert_inserts: defs
454                .upsert_inserts
455                .get_delete_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
456            upsert_updates: defs
457                .upsert_updates
458                .get_delete_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
459            upsert_deletes: defs
460                .upsert_deletes
461                .get_delete_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
462            multi_get_size: defs
463                .multi_get_size
464                .get_delete_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
465            multi_get_result_count: defs
466                .multi_get_result_count
467                .get_delete_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
468            multi_get_result_bytes: defs
469                .multi_get_result_bytes
470                .get_delete_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
471            multi_put_size: defs
472                .multi_put_size
473                .get_delete_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
474            restashed: defs
475                .restashed
476                .get_delete_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
477
478            shared: defs.shared(&source_id),
479            rocksdb_shared: defs.rocksdb_shared(&source_id),
480            rocksdb_instance_metrics: Arc::new(RocksDBInstanceMetrics {
481                multi_get_size: defs
482                    .rocksdb_multi_get_size
483                    .get_delete_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
484                multi_get_result_count: defs
485                    .rocksdb_multi_get_result_count
486                    .get_delete_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
487                multi_get_result_bytes: defs
488                    .rocksdb_multi_get_result_bytes
489                    .get_delete_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
490                multi_get_count: defs
491                    .rocksdb_multi_get_count
492                    .get_delete_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
493                multi_put_count: defs
494                    .rocksdb_multi_put_count
495                    .get_delete_on_drop_metric(vec![source_id_s.clone(), worker_id.clone()]),
496                multi_put_size: defs
497                    .rocksdb_multi_put_size
498                    .get_delete_on_drop_metric(vec![source_id_s, worker_id]),
499            }),
500            _backpressure_metrics: backpressure_metrics,
501        }
502    }
503}
mz_storage/metrics/upsert.rs

mz_storage/metrics/
upsert.rs
