mz_persist/
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
10//! Implementation-specific metrics for persist blobs and consensus
11
12use std::time::Instant;
13
14use mz_ore::metric;
15use mz_ore::metrics::{Counter, IntCounter, MetricsRegistry};
16use mz_ore::stats::histogram_seconds_buckets;
17use prometheus::{Gauge, Histogram, IntCounterVec, IntGauge};
18
19/// Metrics for [crate::hedge::HedgedBlob].
20#[derive(Debug, Clone)]
21pub struct BlobHedgeMetrics {
22    pub(crate) fired: IntCounter,
23    pub(crate) won: IntCounter,
24    pub(crate) won_seconds: Histogram,
25    pub(crate) skipped_budget: IntCounter,
26    pub(crate) skipped_concurrency: IntCounter,
27    pub(crate) skipped_unavailable: IntCounter,
28    pub(crate) errors: IntCounter,
29    pub(crate) warm_errors: IntCounter,
30    pub(crate) armed: IntGauge,
31    pub(crate) rtt_latency: Gauge,
32}
33
34impl BlobHedgeMetrics {
35    /// Returns a new [BlobHedgeMetrics] instance connected to the given
36    /// registry.
37    pub fn new(registry: &MetricsRegistry) -> Self {
38        let skipped: IntCounterVec = registry.register(metric!(
39            name: "mz_persist_blob_hedges_skipped",
40            help: "hedge requests not fired for a get that exceeded the hedge delay, by reason",
41            var_labels: ["reason"],
42        ));
43        BlobHedgeMetrics {
44            fired: registry.register(metric!(
45                name: "mz_persist_blob_hedges_fired",
46                help: "blob gets that fired a hedge request",
47            )),
48            won: registry.register(metric!(
49                name: "mz_persist_blob_hedges_won",
50                help: "blob gets where the hedge request won the race",
51            )),
52            won_seconds: registry.register(metric!(
53                name: "mz_persist_blob_hedge_won_seconds",
54                help: "end-to-end latency of blob gets won by the hedge request",
55                buckets: histogram_seconds_buckets(0.000_500, 32.0),
56            )),
57            skipped_budget: skipped.with_label_values(&["budget"]),
58            skipped_concurrency: skipped.with_label_values(&["concurrency"]),
59            skipped_unavailable: skipped.with_label_values(&["unavailable"]),
60            errors: registry.register(metric!(
61                name: "mz_persist_blob_hedge_errors",
62                help: "hedge requests (the hedge leg only, not the primary) that completed with an error",
63            )),
64            warm_errors: registry.register(metric!(
65                name: "mz_persist_blob_hedge_warm_errors",
66                help: "warm-path liveness gets on the hedge sibling that failed or timed out",
67            )),
68            armed: registry.register(metric!(
69                name: "mz_persist_blob_hedge_armed",
70                help: "1 if this process opened a hedge sibling and can hedge when enabled",
71            )),
72            rtt_latency: registry.register(metric!(
73                name: "mz_persist_blob_hedge_rtt_latency",
74                help: "roundtrip-time of the most recent successful warm-path liveness gets on the hedge sibling",
75            )),
76        }
77    }
78}
79
80/// Metrics specific to S3Blob's internal workings.
81#[derive(Debug, Clone)]
82pub struct S3BlobMetrics {
83    pub(crate) operation_timeouts: IntCounter,
84    pub(crate) operation_attempt_timeouts: IntCounter,
85    pub(crate) connect_timeouts: IntCounter,
86    pub(crate) read_timeouts: IntCounter,
87    pub(crate) get_part: IntCounter,
88    pub(crate) get_invalid_resp: IntCounter,
89    pub(crate) set_single: IntCounter,
90    pub(crate) set_multi_create: IntCounter,
91    pub(crate) set_multi_part: IntCounter,
92    pub(crate) set_multi_complete: IntCounter,
93    pub(crate) delete_head: IntCounter,
94    pub(crate) delete_object: IntCounter,
95    pub(crate) list_objects: IntCounter,
96    pub(crate) error_counts: IntCounterVec,
97}
98
99impl S3BlobMetrics {
100    /// Returns a new [S3BlobMetrics] instance connected to the given registry.
101    pub fn new(registry: &MetricsRegistry) -> Self {
102        let operations: IntCounterVec = registry.register(metric!(
103            name: "mz_persist_s3_operations",
104            help: "number of raw s3 calls on behalf of Blob interface methods",
105            var_labels: ["op"],
106        ));
107        let errors: IntCounterVec = registry.register(metric!(
108            name: "mz_persist_s3_errors",
109            help: "errors",
110            var_labels: ["op", "code"],
111        ));
112        Self {
113            operation_timeouts: registry.register(metric!(
114                name: "mz_persist_s3_operation_timeouts",
115                help: "number of operation timeouts (including retries)",
116            )),
117            operation_attempt_timeouts: registry.register(metric!(
118                name: "mz_persist_s3_operation_attempt_timeouts",
119                help: "number of operation attempt timeouts (within a single retry)",
120            )),
121            connect_timeouts: registry.register(metric!(
122                name: "mz_persist_s3_connect_timeouts",
123                help: "number of timeouts establishing a connection to S3",
124            )),
125            read_timeouts: registry.register(metric!(
126                name: "mz_persist_s3_read_timeouts",
127                help: "number of timeouts waiting on first response byte from S3",
128            )),
129            get_part: operations.with_label_values(&["get_part"]),
130            get_invalid_resp: operations.with_label_values(&["get_invalid_resp"]),
131            set_single: operations.with_label_values(&["set_single"]),
132            set_multi_create: operations.with_label_values(&["set_multi_create"]),
133            set_multi_part: operations.with_label_values(&["set_multi_part"]),
134            set_multi_complete: operations.with_label_values(&["set_multi_complete"]),
135            delete_head: operations.with_label_values(&["delete_head"]),
136            delete_object: operations.with_label_values(&["delete_object"]),
137            list_objects: operations.with_label_values(&["list_objects"]),
138            error_counts: errors,
139        }
140    }
141}
142
143/// Metrics specific to our usage of Arrow and Parquet.
144#[derive(Debug, Clone)]
145pub struct ArrowMetrics {
146    pub(crate) key: ArrowColumnMetrics,
147    pub(crate) val: ArrowColumnMetrics,
148    pub(crate) part_build_seconds: Counter,
149    pub(crate) part_build_count: IntCounter,
150    pub(crate) concat_bytes: IntCounter,
151}
152
153impl ArrowMetrics {
154    /// Returns a new [ArrowMetrics] instance connected to the given registry.
155    pub fn new(registry: &MetricsRegistry) -> Self {
156        let op_count: IntCounterVec = registry.register(metric!(
157            name: "mz_persist_columnar_op_count",
158            help: "number of rows we've run the specified op on in our structured columnar format",
159            var_labels: ["op", "column", "result"],
160        ));
161
162        let part_build_seconds: Counter = registry.register(metric!(
163            name: "mz_persist_columnar_part_build_seconds",
164            help: "number of seconds we've spent encoding our structured columnar format",
165        ));
166        let part_build_count: IntCounter = registry.register(metric!(
167            name: "mz_persist_columnar_part_build_count",
168            help: "number of times we've encoded our structured columnar format",
169        ));
170        let concat_bytes: IntCounter = registry.register(metric!(
171            name: "mz_persist_columnar_part_concat_bytes",
172            help: "number of bytes we've copied when concatenating updates",
173        ));
174
175        ArrowMetrics {
176            key: ArrowColumnMetrics::new(&op_count, "key"),
177            val: ArrowColumnMetrics::new(&op_count, "val"),
178            part_build_seconds,
179            part_build_count,
180            concat_bytes,
181        }
182    }
183
184    /// Metrics for the top-level 'k_s' column.
185    pub fn key(&self) -> &ArrowColumnMetrics {
186        &self.key
187    }
188
189    /// Metrics for the top-level 'v_s' column.
190    pub fn val(&self) -> &ArrowColumnMetrics {
191        &self.val
192    }
193
194    /// Measure and report how long building a Part takes.
195    pub fn measure_part_build<R, F: FnOnce() -> R>(&self, f: F) -> R {
196        let start = Instant::now();
197        let r = f();
198        let duration = start.elapsed();
199
200        self.part_build_count.inc();
201        self.part_build_seconds.inc_by(duration.as_secs_f64());
202
203        r
204    }
205}
206
207/// Metrics for a top-level [`arrow`] column in our structured representation.
208#[derive(Debug, Clone)]
209pub struct ArrowColumnMetrics {
210    correct_count: IntCounter,
211    invalid_count: IntCounter,
212}
213
214impl ArrowColumnMetrics {
215    fn new(count: &IntCounterVec, col: &'static str) -> Self {
216        ArrowColumnMetrics {
217            correct_count: count.with_label_values(&["validation", col, "correct"]),
218            invalid_count: count.with_label_values(&["validation", col, "invalid"]),
219        }
220    }
221
222    /// Measure and report statistics for validation.
223    pub fn report_valid<F: FnOnce() -> bool>(&self, f: F) -> bool {
224        let is_valid = f();
225        if is_valid {
226            self.correct_count.inc();
227        } else {
228            self.invalid_count.inc();
229        }
230        is_valid
231    }
232}
233
234/// Metrics for a Parquet file that we write to S3.
235#[derive(Debug, Clone)]
236pub struct ParquetMetrics {
237    pub(crate) encoded_size: IntCounterVec,
238    pub(crate) num_row_groups: IntCounterVec,
239    pub(crate) k_metrics: ParquetColumnMetrics,
240    pub(crate) v_metrics: ParquetColumnMetrics,
241    pub(crate) t_metrics: ParquetColumnMetrics,
242    pub(crate) d_metrics: ParquetColumnMetrics,
243    pub(crate) k_s_metrics: ParquetColumnMetrics,
244    pub(crate) v_s_metrics: ParquetColumnMetrics,
245    pub(crate) elided_null_buffers: IntCounter,
246}
247
248impl ParquetMetrics {
249    pub(crate) fn new(registry: &MetricsRegistry) -> Self {
250        let encoded_size: IntCounterVec = registry.register(metric!(
251            name: "mz_persist_parquet_encoded_size",
252            help: "encoded size of a parquet file that we write to S3",
253            var_labels: ["format"],
254        ));
255        let num_row_groups: IntCounterVec = registry.register(metric!(
256            name: "mz_persist_parquet_row_group_count",
257            help: "count of row groups in a parquet file",
258            var_labels: ["format"],
259        ));
260
261        let column_size: IntCounterVec = registry.register(metric!(
262            name: "mz_persist_parquet_column_size",
263            help: "size in bytes of a column within a parquet file",
264            var_labels: ["col", "compressed"],
265        ));
266
267        ParquetMetrics {
268            encoded_size,
269            num_row_groups,
270            k_metrics: ParquetColumnMetrics::new("k", &column_size),
271            v_metrics: ParquetColumnMetrics::new("v", &column_size),
272            t_metrics: ParquetColumnMetrics::new("t", &column_size),
273            d_metrics: ParquetColumnMetrics::new("d", &column_size),
274            k_s_metrics: ParquetColumnMetrics::new("k_s", &column_size),
275            v_s_metrics: ParquetColumnMetrics::new("v_s", &column_size),
276            elided_null_buffers: registry.register(metric!(
277                name: "mz_persist_parquet_elided_null_buffer_count",
278                help: "times we dropped an unnecessary null buffer returned by parquet decoding",
279            )),
280        }
281    }
282}
283
284/// Metrics for a column within a Parquet file that we write to S3.
285#[derive(Debug, Clone)]
286pub struct ParquetColumnMetrics {
287    pub(crate) uncompressed_size: IntCounter,
288    pub(crate) compressed_size: IntCounter,
289}
290
291impl ParquetColumnMetrics {
292    pub(crate) fn new(col: &'static str, size: &IntCounterVec) -> Self {
293        ParquetColumnMetrics {
294            uncompressed_size: size.with_label_values(&[col, "uncompressed"]),
295            compressed_size: size.with_label_values(&[col, "compressed"]),
296        }
297    }
298
299    pub(crate) fn report_sizes(&self, uncompressed: u64, compressed: u64) {
300        self.uncompressed_size.inc_by(uncompressed);
301        self.compressed_size.inc_by(compressed);
302    }
303}
304
305/// Metrics for `ColumnarRecords`.
306#[derive(Debug)]
307pub struct ColumnarMetrics {
308    pub(crate) parquet: ParquetMetrics,
309    pub(crate) arrow: ArrowMetrics,
310}
311
312impl ColumnarMetrics {
313    /// Returns a new [ColumnarMetrics].
314    pub fn new(registry: &MetricsRegistry) -> Self {
315        ColumnarMetrics {
316            parquet: ParquetMetrics::new(registry),
317            arrow: ArrowMetrics::new(registry),
318        }
319    }
320
321    /// Returns a reference to the [`arrow`] metrics for our structured data representation.
322    pub fn arrow(&self) -> &ArrowMetrics {
323        &self.arrow
324    }
325
326    /// Returns a reference to the [`parquet`] metrics for our structured data representation.
327    pub fn parquet(&self) -> &ParquetMetrics {
328        &self.parquet
329    }
330
331    /// Returns a [ColumnarMetrics] disconnected from any metrics registry.
332    ///
333    /// Exposed for testing.
334    pub fn disconnected() -> Self {
335        Self::new(&MetricsRegistry::new())
336    }
337}
mz_persist/metrics.rs

mz_persist/
metrics.rs
