risingwave_connector/source/iceberg/
metrics.rs

1// Copyright 2025 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::sync::{Arc, LazyLock};
16
17use prometheus::{Registry, exponential_buckets, histogram_opts};
18use risingwave_common::metrics::{
19    LabelGuardedHistogram, LabelGuardedHistogramVec, LabelGuardedIntCounter,
20    LabelGuardedIntCounterVec, LabelGuardedIntGaugeVec,
21};
22use risingwave_common::monitor::GLOBAL_METRICS_REGISTRY;
23use risingwave_common::{
24    register_guarded_histogram_vec_with_registry, register_guarded_int_counter_vec_with_registry,
25    register_guarded_int_gauge_vec_with_registry,
26};
27
28#[derive(Clone)]
29pub struct IcebergScanMetrics {
30    // -- Existing --
31    pub iceberg_read_bytes: LabelGuardedIntCounterVec,
32
33    // -- Snapshot & Discovery (List Executor) --
34    /// Time difference (seconds) between the latest available snapshot and
35    /// the last ingested snapshot.
36    pub iceberg_source_snapshot_lag_seconds: LabelGuardedIntGaugeVec,
37
38    /// Total number of snapshots discovered via incremental scan.
39    pub iceberg_source_snapshots_discovered_total: LabelGuardedIntCounterVec,
40
41    /// Time spent planning files from a snapshot (metadata operation).
42    pub iceberg_source_list_duration_seconds: LabelGuardedHistogramVec,
43
44    /// Files discovered per scan, labeled by `file_type` (data, `eq_delete`, `pos_delete`).
45    pub iceberg_source_files_discovered_total: LabelGuardedIntCounterVec,
46
47    // -- Data Reading (used in scan_task_to_chunk_with_deletes, labeled by table_name) --
48    /// Per-file read duration.
49    pub iceberg_source_file_read_duration_seconds: LabelGuardedHistogramVec,
50
51    /// Total rows read from Iceberg source.
52    pub iceberg_source_rows_read_total: LabelGuardedIntCounterVec,
53
54    /// Total files read from Iceberg source, labeled by `file_type`.
55    pub iceberg_source_files_read_total: LabelGuardedIntCounterVec,
56
57    // -- Delete Handling --
58    /// Rows removed by delete processing, labeled by `delete_type`.
59    pub iceberg_source_delete_rows_applied_total: LabelGuardedIntCounterVec,
60
61    /// Histogram of delete files attached per data file scan task.
62    pub iceberg_source_delete_files_per_data_file: LabelGuardedHistogramVec,
63
64    // -- Operational Health --
65    /// Number of files currently being fetched by the active reader.
66    pub iceberg_source_inflight_file_count: LabelGuardedIntGaugeVec,
67
68    /// Categorized scan error counter.
69    pub iceberg_source_scan_errors_total: LabelGuardedIntCounterVec,
70}
71
72#[derive(Clone)]
73pub struct IcebergFileScanMetrics {
74    read_bytes: LabelGuardedIntCounter,
75    file_read_duration_seconds: LabelGuardedHistogram,
76    rows_read_total: LabelGuardedIntCounter,
77    files_read_total: LabelGuardedIntCounter,
78    delete_rows_applied_total: LabelGuardedIntCounter,
79}
80
81impl IcebergFileScanMetrics {
82    pub fn new(metrics: &IcebergScanMetrics, table_name: &str) -> Self {
83        Self {
84            read_bytes: metrics
85                .iceberg_read_bytes
86                .with_guarded_label_values(&[table_name]),
87            file_read_duration_seconds: metrics
88                .iceberg_source_file_read_duration_seconds
89                .with_guarded_label_values(&[table_name]),
90            rows_read_total: metrics
91                .iceberg_source_rows_read_total
92                .with_guarded_label_values(&[table_name]),
93            files_read_total: metrics
94                .iceberg_source_files_read_total
95                .with_guarded_label_values(&[table_name, "data"]),
96            delete_rows_applied_total: metrics
97                .iceberg_source_delete_rows_applied_total
98                .with_guarded_label_values(&[table_name, "sdk_applied_approx"]),
99        }
100    }
101
102    pub fn record_read_bytes(&self, bytes: u64) {
103        self.read_bytes.inc_by(bytes);
104    }
105
106    pub fn record_file_read_duration(&self, seconds: f64) {
107        self.file_read_duration_seconds.observe(seconds);
108    }
109
110    pub fn record_rows_read(&self, rows: u64) {
111        self.rows_read_total.inc_by(rows);
112    }
113
114    pub fn record_file_read(&self) {
115        self.files_read_total.inc();
116    }
117
118    pub fn record_delete_rows_applied(&self, rows: u64) {
119        self.delete_rows_applied_total.inc_by(rows);
120    }
121}
122
123impl IcebergScanMetrics {
124    fn new(registry: &Registry) -> Self {
125        let iceberg_read_bytes = register_guarded_int_counter_vec_with_registry!(
126            "iceberg_read_bytes",
127            "Total size of iceberg read requests",
128            &["table_name"],
129            registry
130        )
131        .unwrap();
132
133        let iceberg_source_snapshot_lag_seconds = register_guarded_int_gauge_vec_with_registry!(
134            "iceberg_source_snapshot_lag_seconds",
135            "Lag between latest available snapshot and last ingested snapshot in seconds",
136            &["source_id", "source_name", "table_name"],
137            registry
138        )
139        .unwrap();
140
141        let iceberg_source_snapshots_discovered_total =
142            register_guarded_int_counter_vec_with_registry!(
143                "iceberg_source_snapshots_discovered_total",
144                "Total number of snapshots discovered via incremental scan",
145                &["source_id", "source_name", "table_name"],
146                registry
147            )
148            .unwrap();
149
150        let iceberg_source_list_duration_seconds = register_guarded_histogram_vec_with_registry!(
151            histogram_opts!(
152                "iceberg_source_list_duration_seconds",
153                "Time spent planning files from a snapshot",
154                exponential_buckets(0.01, 2.0, 15).unwrap() // 10ms to ~164s
155            ),
156            &["source_id", "source_name", "table_name"],
157            registry
158        )
159        .unwrap();
160
161        let iceberg_source_files_discovered_total =
162            register_guarded_int_counter_vec_with_registry!(
163                "iceberg_source_files_discovered_total",
164                "Total number of files discovered per scan",
165                &["source_id", "source_name", "table_name", "file_type"],
166                registry
167            )
168            .unwrap();
169
170        // Note: file-read metrics use ["table_name"] labels (matching iceberg_read_bytes)
171        // because the scan function doesn't have source-level context.
172        let iceberg_source_file_read_duration_seconds =
173            register_guarded_histogram_vec_with_registry!(
174                histogram_opts!(
175                    "iceberg_source_file_read_duration_seconds",
176                    "Per-file read duration",
177                    exponential_buckets(0.01, 2.0, 15).unwrap() // 10ms to ~164s
178                ),
179                &["table_name"],
180                registry
181            )
182            .unwrap();
183
184        let iceberg_source_rows_read_total = register_guarded_int_counter_vec_with_registry!(
185            "iceberg_source_rows_read_total",
186            "Total rows read from Iceberg source",
187            &["table_name"],
188            registry
189        )
190        .unwrap();
191
192        let iceberg_source_files_read_total = register_guarded_int_counter_vec_with_registry!(
193            "iceberg_source_files_read_total",
194            "Total files read from Iceberg source",
195            &["table_name", "file_type"],
196            registry
197        )
198        .unwrap();
199
200        let iceberg_source_delete_rows_applied_total =
201            register_guarded_int_counter_vec_with_registry!(
202                "iceberg_source_delete_rows_applied_total",
203                "Total rows removed by delete processing",
204                &["table_name", "delete_type"],
205                registry
206            )
207            .unwrap();
208
209        let iceberg_source_delete_files_per_data_file =
210            register_guarded_histogram_vec_with_registry!(
211                histogram_opts!(
212                    "iceberg_source_delete_files_per_data_file",
213                    "Number of delete files attached per data file scan task",
214                    // 1, 2, 4, 8, 16, 32, 64, 128
215                    exponential_buckets(1.0, 2.0, 8).unwrap()
216                ),
217                &["source_id", "source_name", "table_name"],
218                registry
219            )
220            .unwrap();
221
222        let iceberg_source_inflight_file_count = register_guarded_int_gauge_vec_with_registry!(
223            "iceberg_source_inflight_file_count",
224            "Number of files currently being fetched by the active reader",
225            &["source_id", "source_name", "table_name"],
226            registry
227        )
228        .unwrap();
229
230        let iceberg_source_scan_errors_total = register_guarded_int_counter_vec_with_registry!(
231            "iceberg_source_scan_errors_total",
232            "Total number of scan errors categorized by error type",
233            &["source_id", "source_name", "table_name", "error_type"],
234            registry
235        )
236        .unwrap();
237
238        Self {
239            iceberg_read_bytes,
240            iceberg_source_snapshot_lag_seconds,
241            iceberg_source_snapshots_discovered_total,
242            iceberg_source_list_duration_seconds,
243            iceberg_source_files_discovered_total,
244            iceberg_source_file_read_duration_seconds,
245            iceberg_source_rows_read_total,
246            iceberg_source_files_read_total,
247            iceberg_source_delete_rows_applied_total,
248            iceberg_source_delete_files_per_data_file,
249            iceberg_source_inflight_file_count,
250            iceberg_source_scan_errors_total,
251        }
252    }
253
254    pub fn for_test() -> Arc<Self> {
255        Arc::new(GLOBAL_ICEBERG_SCAN_METRICS.clone())
256    }
257}
258
259pub static GLOBAL_ICEBERG_SCAN_METRICS: LazyLock<IcebergScanMetrics> =
260    LazyLock::new(|| IcebergScanMetrics::new(&GLOBAL_METRICS_REGISTRY));
risingwave_connector/source/iceberg/metrics.rs

risingwave_connector/source/iceberg/
metrics.rs
