mz_storage/metrics/sink/
iceberg.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 iceberg sinks.
11
12use mz_ore::{
13    metric,
14    metrics::{
15        DeregisterOnDropCounter, DeregisterOnDropGauge, DeregisterOnDropHistogram,
16        DeregisterOnDropVec, MetricTag, MetricVisibility, raw,
17    },
18    stats::histogram_seconds_buckets,
19};
20use mz_repr::GlobalId;
21use prometheus::core::AtomicU64;
22
23#[derive(Debug, Clone)]
24pub(crate) struct IcebergSinkMetricDefs {
25    /// Number of data files written by the iceberg sink.
26    pub data_files_written: DeregisterOnDropVec<raw::IntCounterVec>,
27    /// Number of delete files written by the iceberg sink.
28    pub delete_files_written: DeregisterOnDropVec<raw::IntCounterVec>,
29    /// Number of stashed rows in the iceberg sink.
30    pub stashed_rows: DeregisterOnDropVec<raw::UIntGaugeVec>,
31    /// Number of snapshots committed by the iceberg sink.
32    pub snapshots_committed: DeregisterOnDropVec<raw::IntCounterVec>,
33    /// Commit failures in the iceberg sink.
34    pub commit_failures: DeregisterOnDropVec<raw::IntCounterVec>,
35    /// Commit conflicts in the iceberg sink.
36    pub commit_conflicts: DeregisterOnDropVec<raw::IntCounterVec>,
37    /// Time spent committing batches to Iceberg.
38    pub commit_duration_seconds: DeregisterOnDropVec<raw::HistogramVec>,
39    /// Time spent closing Iceberg DeltaWriters.
40    pub writer_close_duration_seconds: DeregisterOnDropVec<raw::HistogramVec>,
41}
42
43impl IcebergSinkMetricDefs {
44    pub(crate) fn register_with(registry: &mz_ore::metrics::MetricsRegistry) -> Self {
45        Self {
46            data_files_written: registry.register(metric!(
47                name: "mz_sink_iceberg_data_files_written",
48                help: "Number of data files written by the iceberg sink",
49                var_labels: ["sink_id", "worker_id"],
50                visibility: MetricVisibility::Public,
51                tags: [MetricTag::Sink],
52            )),
53            delete_files_written: registry.register(metric!(
54                name: "mz_sink_iceberg_delete_files_written",
55                help: "Number of delete files written by the iceberg sink",
56                var_labels: ["sink_id", "worker_id"],
57                visibility: MetricVisibility::Public,
58                tags: [MetricTag::Sink],
59            )),
60            stashed_rows: registry.register(metric!(
61                name: "mz_sink_iceberg_stashed_rows",
62                help: "Number of stashed rows in the iceberg sink",
63                var_labels: ["sink_id", "worker_id"]
64            )),
65            snapshots_committed: registry.register(metric!(
66                name: "mz_sink_iceberg_snapshots_committed",
67                help: "Number of snapshots committed by the iceberg sink",
68                var_labels: ["sink_id", "worker_id"],
69                visibility: MetricVisibility::Public,
70                tags: [MetricTag::Sink],
71            )),
72            commit_failures: registry.register(metric!(
73                name: "mz_sink_iceberg_commit_failures",
74                help: "Number of commit failures in the iceberg sink",
75                var_labels: ["sink_id", "worker_id"],
76                visibility: MetricVisibility::Public,
77                tags: [MetricTag::Sink],
78            )),
79            commit_conflicts: registry.register(metric!(
80                name: "mz_sink_iceberg_commit_conflicts",
81                help: "Number of commit conflicts in the iceberg sink",
82                var_labels: ["sink_id", "worker_id"],
83                visibility: MetricVisibility::Public,
84                tags: [MetricTag::Sink],
85            )),
86            commit_duration_seconds: registry.register(metric!(
87                name: "mz_sink_iceberg_commit_duration_seconds",
88                help: "Time spent committing batches to Iceberg in seconds",
89                var_labels: ["sink_id", "worker_id"],
90                buckets: histogram_seconds_buckets(0.001, 32.0),
91                visibility: MetricVisibility::Public,
92                tags: [MetricTag::Sink],
93            )),
94            writer_close_duration_seconds: registry.register(metric!(
95                name: "mz_sink_iceberg_writer_close_duration_seconds",
96                help: "Time spent closing Iceberg DeltaWriters in seconds",
97                var_labels: ["sink_id", "worker_id"],
98                buckets: histogram_seconds_buckets(0.001, 32.0),
99            )),
100        }
101    }
102}
103
104pub(crate) struct IcebergSinkMetrics {
105    /// Number of data files written by the iceberg sink.
106    pub data_files_written: DeregisterOnDropCounter<AtomicU64>,
107    /// Number of delete files written by the iceberg sink.
108    pub delete_files_written: DeregisterOnDropCounter<AtomicU64>,
109    /// Number of stashed rows in the iceberg sink.
110    pub stashed_rows: DeregisterOnDropGauge<AtomicU64>,
111    /// Number of snapshots committed by the iceberg sink.
112    pub snapshots_committed: DeregisterOnDropCounter<AtomicU64>,
113    /// Number of commit failures in the iceberg sink.
114    pub commit_failures: DeregisterOnDropCounter<AtomicU64>,
115    /// Number of commit conflicts in the iceberg sink.
116    pub commit_conflicts: DeregisterOnDropCounter<AtomicU64>,
117    /// Time spent committing batches to Iceberg.
118    pub commit_duration_seconds: DeregisterOnDropHistogram,
119    /// Time spent closing Iceberg DeltaWriters.
120    pub writer_close_duration_seconds: DeregisterOnDropHistogram,
121}
122
123impl IcebergSinkMetrics {
124    /// Create a `IcebergSinkMetrics` from the `IcebergSinkMetricDefs`.
125    pub(crate) fn new(defs: &IcebergSinkMetricDefs, sink_id: GlobalId, worker_id: usize) -> Self {
126        let labels = vec![sink_id.to_string(), worker_id.to_string()];
127        Self {
128            data_files_written: defs
129                .data_files_written
130                .get_deregister_on_drop_metric(labels.clone()),
131            delete_files_written: defs
132                .delete_files_written
133                .get_deregister_on_drop_metric(labels.clone()),
134            stashed_rows: defs
135                .stashed_rows
136                .get_deregister_on_drop_metric(labels.clone()),
137            snapshots_committed: defs
138                .snapshots_committed
139                .get_deregister_on_drop_metric(labels.clone()),
140            commit_failures: defs
141                .commit_failures
142                .get_deregister_on_drop_metric(labels.clone()),
143            commit_conflicts: defs
144                .commit_conflicts
145                .get_deregister_on_drop_metric(labels.clone()),
146            commit_duration_seconds: defs
147                .commit_duration_seconds
148                .get_deregister_on_drop_metric(labels.clone()),
149            writer_close_duration_seconds: defs
150                .writer_close_duration_seconds
151                .get_deregister_on_drop_metric(labels),
152        }
153    }
154}
mz_storage/metrics/sink/iceberg.rs

mz_storage/metrics/sink/
iceberg.rs
