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

mz_persist/
metrics.rs
