sui_indexer_alt_framework/
metrics.rs

1// Copyright (c) Mysten Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4use std::sync::Arc;
5use std::sync::atomic::AtomicU64;
6use std::time::Duration;
7
8use prometheus::Histogram;
9use prometheus::HistogramVec;
10use prometheus::IntCounter;
11use prometheus::IntCounterVec;
12use prometheus::IntGauge;
13use prometheus::IntGaugeVec;
14use prometheus::Registry;
15use prometheus::register_histogram_vec_with_registry;
16use prometheus::register_histogram_with_registry;
17use prometheus::register_int_counter_vec_with_registry;
18use prometheus::register_int_counter_with_registry;
19use prometheus::register_int_gauge_vec_with_registry;
20use tracing::warn;
21
22use crate::ingestion::error::Error;
23use crate::pipeline::Processor;
24
25/// Prefix used for metric names when no explicit prefix is provided.
26pub(crate) const DEFAULT_METRICS_PREFIX: &str = "indexer";
27
28/// Histogram buckets for the distribution of checkpoint fetching latencies.
29const INGESTION_LATENCY_SEC_BUCKETS: &[f64] = &[
30    0.001, 0.002, 0.005, 0.01, 0.02, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0, 20.0, 50.0, 100.0,
31];
32
33/// Histogram buckets for the distribution of checkpoint lag (difference between the system time and
34/// the timestamp in the checkpoint).
35const LAG_SEC_BUCKETS: &[f64] = &[
36    0.1, 0.15, 0.2, 0.25, 0.3, 0.35, 0.4, 0.45, 0.5, 0.55, 0.6, 0.65, 0.7, 0.75, 0.8, 0.85, 0.9,
37    0.95, 1.0, 2.0, 3.0, 4.0, 5.0, 10.0, 20.0, 50.0, 100.0, 1000.0,
38];
39
40/// Histogram buckets for the distribution of latencies for processing a checkpoint in the indexer
41/// (without having to call out to other services).
42const PROCESSING_LATENCY_SEC_BUCKETS: &[f64] = &[
43    0.001, 0.002, 0.005, 0.01, 0.02, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0, 20.0, 50.0, 100.0,
44];
45
46/// Histogram buckets for the distribution of latencies for writing to the database.
47const DB_UPDATE_LATENCY_SEC_BUCKETS: &[f64] = &[
48    0.01, 0.02, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0, 20.0, 50.0, 100.0, 200.0, 500.0, 1000.0,
49    2000.0, 5000.0, 10000.0,
50];
51
52/// Histogram buckets for the distribution of batch sizes (number of rows) written to the database.
53const BATCH_SIZE_BUCKETS: &[f64] = &[
54    1.0, 2.0, 5.0, 10.0, 20.0, 50.0, 100.0, 200.0, 500.0, 1000.0, 2000.0, 5000.0, 10000.0, 20000.0,
55];
56
57/// Metrics specific to the ingestion service.
58///
59/// Almost every metric is labeled by `cohort`: the framework mints one ingestion service per cohort,
60/// all sharing this handle, so the label keeps their series from colliding. The two exceptions are
61/// `total_ingested_bytes` (counted inside the checkpoint source, which is shared across cohorts) and
62/// `ingested_latest_checkpoint_latency` (recorded only by the factory's one-time tip probe, before
63/// any cohort exists).
64#[derive(Clone)]
65pub struct IngestionMetrics {
66    // Statistics related to fetching data from the remote store.
67    pub total_ingested_checkpoints: IntCounterVec,
68    pub total_ingested_transactions: IntCounterVec,
69    pub total_ingested_events: IntCounterVec,
70    pub total_ingested_objects: IntCounterVec,
71    pub total_ingested_bytes: IntCounter,
72    pub total_ingested_transient_retries: IntCounterVec,
73    pub total_ingested_not_found_retries: IntCounterVec,
74    pub total_streamed_checkpoints: IntCounterVec,
75    pub total_skipped_streamed_checkpoints: IntCounterVec,
76    pub total_out_of_order_streamed_checkpoints: IntCounterVec,
77    pub total_stream_disconnections: IntCounterVec,
78    pub total_streaming_connection_failures: IntCounterVec,
79
80    // Checkpoint lag metrics for the ingestion pipeline.
81    pub latest_ingested_checkpoint: IntGaugeVec,
82    pub latest_streamed_checkpoint: IntGaugeVec,
83    pub latest_skipped_streamed_checkpoint: IntGaugeVec,
84    pub latest_ingested_checkpoint_timestamp_lag_ms: IntGaugeVec,
85    pub ingested_checkpoint_timestamp_lag: HistogramVec,
86
87    pub ingested_checkpoint_latency: HistogramVec,
88    pub ingested_chain_id_latency: HistogramVec,
89    pub ingested_latest_checkpoint_latency: Histogram,
90
91    pub ingestion_concurrency_limit: IntGaugeVec,
92    pub ingestion_concurrency_inflight: IntGaugeVec,
93}
94
95/// A cohort's pre-bound view over [`IngestionMetrics`]. The framework mints one ingestion service
96/// per cohort; each binds its metrics' `cohort` label once, here, so emission sites stay plain
97/// `.inc()`/`.observe()`/`.set()` calls rather than repeating `.with_label_values(&[cohort])`.
98pub(crate) struct CohortMetrics {
99    pub(crate) total_ingested_checkpoints: IntCounter,
100    pub(crate) total_ingested_transactions: IntCounter,
101    pub(crate) total_ingested_events: IntCounter,
102    pub(crate) total_ingested_objects: IntCounter,
103    pub(crate) total_ingested_not_found_retries: IntCounter,
104    pub(crate) total_streamed_checkpoints: IntCounter,
105    pub(crate) total_skipped_streamed_checkpoints: IntCounter,
106    pub(crate) total_out_of_order_streamed_checkpoints: IntCounter,
107    pub(crate) total_stream_disconnections: IntCounter,
108    pub(crate) total_streaming_connection_failures: IntCounter,
109
110    pub(crate) ingested_checkpoint_latency: Histogram,
111    pub(crate) ingested_chain_id_latency: Histogram,
112
113    pub(crate) latest_streamed_checkpoint: IntGauge,
114    pub(crate) latest_skipped_streamed_checkpoint: IntGauge,
115    pub(crate) ingestion_concurrency_limit: IntGauge,
116    pub(crate) ingestion_concurrency_inflight: IntGauge,
117
118    /// Reports this cohort's checkpoint-lag gauges/histogram (with its own running-max state).
119    pub(crate) checkpoint_lag: Arc<CheckpointLagMetricReporter>,
120
121    /// Retries carry a `reason` label in addition to `cohort`, so this stays an unbound vector and
122    /// [`Self::inc_retry`] binds both labels.
123    total_ingested_transient_retries: IntCounterVec,
124    /// This client's `cohort` label value, for the two-label retry counter.
125    label: String,
126}
127
128#[derive(Clone)]
129pub struct IndexerMetrics {
130    // Statistics related to individual ingestion pipelines' handlers.
131    pub total_handler_checkpoints_received: IntCounterVec,
132    pub total_handler_processor_retries: IntCounterVec,
133    pub total_handler_checkpoints_processed: IntCounterVec,
134    pub total_handler_rows_created: IntCounterVec,
135
136    pub latest_processed_checkpoint: IntGaugeVec,
137    pub latest_processed_checkpoint_timestamp_lag_ms: IntGaugeVec,
138    pub processed_checkpoint_timestamp_lag: HistogramVec,
139
140    pub handler_checkpoint_latency: HistogramVec,
141
142    pub processor_concurrency_limit: IntGaugeVec,
143    pub processor_concurrency_inflight: IntGaugeVec,
144
145    // Statistics related to individual ingestion pipelines.
146    pub total_collector_checkpoints_received: IntCounterVec,
147    pub total_collector_rows_received: IntCounterVec,
148    pub total_collector_batches_created: IntCounterVec,
149    pub total_committer_batches_attempted: IntCounterVec,
150    pub total_committer_batches_succeeded: IntCounterVec,
151    pub total_committer_batches_failed: IntCounterVec,
152    pub total_committer_rows_committed: IntCounterVec,
153    pub total_committer_rows_affected: IntCounterVec,
154    pub total_watermarks_out_of_order: IntCounterVec,
155    pub total_pruner_chunks_attempted: IntCounterVec,
156    pub total_pruner_chunks_deleted: IntCounterVec,
157    pub total_pruner_rows_deleted: IntCounterVec,
158
159    // Checkpoint lag metrics for the collector.
160    pub latest_collected_checkpoint: IntGaugeVec,
161    pub latest_collected_checkpoint_timestamp_lag_ms: IntGaugeVec,
162    pub collected_checkpoint_timestamp_lag: HistogramVec,
163
164    // Checkpoint lag metrics for the committer.
165    // We can only report partially committed checkpoints, since the concurrent committer isn't aware of
166    // when a checkpoint is fully committed. So we report whenever we see a checkpoint. Since data from
167    // the same checkpoint is batched continuously, this is a good proxy for the last committed checkpoint.
168    pub latest_partially_committed_checkpoint: IntGaugeVec,
169    pub latest_partially_committed_checkpoint_timestamp_lag_ms: IntGaugeVec,
170    pub partially_committed_checkpoint_timestamp_lag: HistogramVec,
171
172    // Checkpoint lag metrics for the watermarker.
173    // The latest watermarked checkpoint metric is already covered by watermark_checkpoint_in_db.
174    // While we already have watermark_timestamp_in_db_ms metric, reporting the lag explicitly
175    // for consistency.
176    pub latest_watermarked_checkpoint_timestamp_lag_ms: IntGaugeVec,
177    pub watermarked_checkpoint_timestamp_lag: HistogramVec,
178
179    pub collector_gather_latency: HistogramVec,
180    pub collector_batch_size: HistogramVec,
181    pub total_collector_skipped_checkpoints: IntCounterVec,
182    pub collector_reader_lo: IntGaugeVec,
183    pub committer_commit_latency: HistogramVec,
184    pub committer_tx_rows: HistogramVec,
185    pub watermark_gather_latency: HistogramVec,
186    pub watermark_commit_latency: HistogramVec,
187    pub watermark_pruner_read_latency: HistogramVec,
188    pub watermark_pruner_write_latency: HistogramVec,
189    pub pruner_delete_latency: HistogramVec,
190
191    pub watermark_epoch: IntGaugeVec,
192    pub watermark_checkpoint: IntGaugeVec,
193    pub watermark_transaction: IntGaugeVec,
194    pub watermark_timestamp_ms: IntGaugeVec,
195    pub watermark_reader_lo: IntGaugeVec,
196    pub watermark_pruner_hi: IntGaugeVec,
197
198    pub watermark_epoch_in_db: IntGaugeVec,
199    pub watermark_checkpoint_in_db: IntGaugeVec,
200    pub watermark_transaction_in_db: IntGaugeVec,
201    pub watermark_timestamp_in_db_ms: IntGaugeVec,
202    pub watermark_reader_lo_in_db: IntGaugeVec,
203    pub watermark_pruner_hi_in_db: IntGaugeVec,
204}
205
206/// A helper struct to report metrics regarding the checkpoint lag at various points in the indexer.
207pub(crate) struct CheckpointLagMetricReporter {
208    /// Metric to report the lag distribution of each checkpoint.
209    checkpoint_time_lag_histogram: Histogram,
210    /// Metric to report the lag of the checkpoint with the highest sequence number observed so far.
211    /// This is needed since concurrent pipelines observe checkpoints out of order.
212    latest_checkpoint_time_lag_gauge: IntGauge,
213    /// Metric to report the sequence number of the checkpoint with the highest sequence number observed so far.
214    latest_checkpoint_sequence_number_gauge: IntGauge,
215    // Internal state to keep track of the highest checkpoint sequence number reported so far.
216    latest_reported_checkpoint: AtomicU64,
217}
218
219impl IngestionMetrics {
220    pub fn new(prefix: Option<&str>, registry: &Registry) -> Arc<Self> {
221        let prefix = prefix.unwrap_or(DEFAULT_METRICS_PREFIX);
222        let name = |n| format!("{prefix}_{n}");
223        Arc::new(Self {
224            total_ingested_checkpoints: register_int_counter_vec_with_registry!(
225                name("total_ingested_checkpoints"),
226                "Total number of checkpoints fetched from the remote store",
227                &["cohort"],
228                registry,
229            )
230            .unwrap(),
231            total_ingested_transactions: register_int_counter_vec_with_registry!(
232                name("total_ingested_transactions"),
233                "Total number of transactions fetched from the remote store",
234                &["cohort"],
235                registry,
236            )
237            .unwrap(),
238            total_ingested_events: register_int_counter_vec_with_registry!(
239                name("total_ingested_events"),
240                "Total number of events fetched from the remote store",
241                &["cohort"],
242                registry,
243            )
244            .unwrap(),
245            total_ingested_objects: register_int_counter_vec_with_registry!(
246                name("total_ingested_objects"),
247                "Total number of objects in checkpoints fetched from the remote store",
248                &["cohort"],
249                registry,
250            )
251            .unwrap(),
252            total_ingested_bytes: register_int_counter_with_registry!(
253                name("total_ingested_bytes"),
254                "Total number of bytes fetched from the remote store",
255                registry,
256            )
257            .unwrap(),
258            total_ingested_transient_retries: register_int_counter_vec_with_registry!(
259                name("total_ingested_retries"),
260                "Total number of retries due to transient errors while fetching data from the \
261                 remote store",
262                &["reason", "cohort"],
263                registry,
264            )
265            .unwrap(),
266            total_ingested_not_found_retries: register_int_counter_vec_with_registry!(
267                name("total_ingested_not_found_retries"),
268                "Total number of retries due to the not found errors while fetching data from the \
269                 remote store",
270                &["cohort"],
271                registry,
272            )
273            .unwrap(),
274            total_streamed_checkpoints: register_int_counter_vec_with_registry!(
275                name("total_streamed_checkpoints"),
276                "Total number of checkpoints received from gRPC streaming",
277                &["cohort"],
278                registry,
279            )
280            .unwrap(),
281            total_skipped_streamed_checkpoints: register_int_counter_vec_with_registry!(
282                name("total_skipped_streamed_checkpoints"),
283                "Total number of streamed checkpoints skipped because they were already processed",
284                &["cohort"],
285                registry,
286            )
287            .unwrap(),
288            total_out_of_order_streamed_checkpoints: register_int_counter_vec_with_registry!(
289                name("total_out_of_order_streamed_checkpoints"),
290                "Total number of streamed checkpoints received out of order",
291                &["cohort"],
292                registry,
293            )
294            .unwrap(),
295            total_stream_disconnections: register_int_counter_vec_with_registry!(
296                name("total_stream_disconnections"),
297                "Total number of times the gRPC stream was disconnected",
298                &["cohort"],
299                registry,
300            )
301            .unwrap(),
302            total_streaming_connection_failures: register_int_counter_vec_with_registry!(
303                name("total_streaming_connection_failures"),
304                "Total number of failures due to streaming service connection or peek failures",
305                &["cohort"],
306                registry,
307            )
308            .unwrap(),
309            latest_ingested_checkpoint: register_int_gauge_vec_with_registry!(
310                name("latest_ingested_checkpoint"),
311                "Latest checkpoint sequence number fetched from the remote store",
312                &["cohort"],
313                registry,
314            )
315            .unwrap(),
316            latest_streamed_checkpoint: register_int_gauge_vec_with_registry!(
317                name("latest_streamed_checkpoint"),
318                "Latest checkpoint sequence number received from gRPC streaming",
319                &["cohort"],
320                registry,
321            )
322            .unwrap(),
323            latest_skipped_streamed_checkpoint: register_int_gauge_vec_with_registry!(
324                name("latest_skipped_streamed_checkpoint"),
325                "Latest streamed checkpoint sequence number skipped because it was already processed",
326                &["cohort"],
327                registry,
328            )
329            .unwrap(),
330            latest_ingested_checkpoint_timestamp_lag_ms: register_int_gauge_vec_with_registry!(
331                name("latest_ingested_checkpoint_timestamp_lag_ms"),
332                "Difference between the system timestamp when the latest checkpoint was fetched and the \
333                 timestamp in the checkpoint, in milliseconds",
334                &["cohort"],
335                registry,
336            )
337            .unwrap(),
338            ingested_checkpoint_timestamp_lag: register_histogram_vec_with_registry!(
339                name("ingested_checkpoint_timestamp_lag"),
340                "Difference between the system timestamp when a checkpoint was fetched and the \
341                 timestamp in each checkpoint, in seconds",
342                &["cohort"],
343                LAG_SEC_BUCKETS.to_vec(),
344                registry,
345            )
346            .unwrap(),
347            ingested_checkpoint_latency: register_histogram_vec_with_registry!(
348                name("ingested_checkpoint_latency"),
349                "Time taken to fetch a checkpoint from the remote store, including retries",
350                &["cohort"],
351                INGESTION_LATENCY_SEC_BUCKETS.to_vec(),
352                registry,
353            )
354            .unwrap(),
355            ingested_chain_id_latency: register_histogram_vec_with_registry!(
356                name("ingested_chain_id_latency"),
357                "Time taken to fetch the chain identifier, including retries",
358                &["cohort"],
359                INGESTION_LATENCY_SEC_BUCKETS.to_vec(),
360                registry,
361            )
362            .unwrap(),
363            ingested_latest_checkpoint_latency: register_histogram_with_registry!(
364                name("ingested_latest_checkpoint_latency"),
365                "Time taken to fetch the latest checkpoint number, including retries",
366                INGESTION_LATENCY_SEC_BUCKETS.to_vec(),
367                registry,
368            )
369            .unwrap(),
370            ingestion_concurrency_limit: register_int_gauge_vec_with_registry!(
371                name("ingestion_concurrency_limit"),
372                "Current adaptive concurrency limit for checkpoint ingestion",
373                &["cohort"],
374                registry,
375            )
376            .unwrap(),
377            ingestion_concurrency_inflight: register_int_gauge_vec_with_registry!(
378                name("ingestion_concurrency_inflight"),
379                "Current number of in-flight checkpoint ingestion tasks",
380                &["cohort"],
381                registry,
382            )
383            .unwrap(),
384        })
385    }
386}
387
388impl CohortMetrics {
389    /// Bind every cohort-labeled ingestion metric to `label` once, so services minted for
390    /// different cohorts report under their own label without re-binding at each emission site.
391    pub(crate) fn new(metrics: &IngestionMetrics, label: &str) -> Arc<Self> {
392        let c = label;
393        let checkpoint_lag = CheckpointLagMetricReporter::with_label(
394            &metrics.ingested_checkpoint_timestamp_lag,
395            &metrics.latest_ingested_checkpoint_timestamp_lag_ms,
396            &metrics.latest_ingested_checkpoint,
397            c,
398        );
399        Arc::new(Self {
400            total_ingested_checkpoints: metrics.total_ingested_checkpoints.with_label_values(&[c]),
401            total_ingested_transactions: metrics
402                .total_ingested_transactions
403                .with_label_values(&[c]),
404            total_ingested_events: metrics.total_ingested_events.with_label_values(&[c]),
405            total_ingested_objects: metrics.total_ingested_objects.with_label_values(&[c]),
406            total_ingested_not_found_retries: metrics
407                .total_ingested_not_found_retries
408                .with_label_values(&[c]),
409            total_streamed_checkpoints: metrics.total_streamed_checkpoints.with_label_values(&[c]),
410            total_skipped_streamed_checkpoints: metrics
411                .total_skipped_streamed_checkpoints
412                .with_label_values(&[c]),
413            total_out_of_order_streamed_checkpoints: metrics
414                .total_out_of_order_streamed_checkpoints
415                .with_label_values(&[c]),
416            total_stream_disconnections: metrics
417                .total_stream_disconnections
418                .with_label_values(&[c]),
419            total_streaming_connection_failures: metrics
420                .total_streaming_connection_failures
421                .with_label_values(&[c]),
422            ingested_checkpoint_latency: metrics
423                .ingested_checkpoint_latency
424                .with_label_values(&[c]),
425            ingested_chain_id_latency: metrics.ingested_chain_id_latency.with_label_values(&[c]),
426            latest_streamed_checkpoint: metrics.latest_streamed_checkpoint.with_label_values(&[c]),
427            latest_skipped_streamed_checkpoint: metrics
428                .latest_skipped_streamed_checkpoint
429                .with_label_values(&[c]),
430            ingestion_concurrency_limit: metrics
431                .ingestion_concurrency_limit
432                .with_label_values(&[c]),
433            ingestion_concurrency_inflight: metrics
434                .ingestion_concurrency_inflight
435                .with_label_values(&[c]),
436            checkpoint_lag,
437            total_ingested_transient_retries: metrics.total_ingested_transient_retries.clone(),
438            label: label.to_string(),
439        })
440    }
441
442    /// Register that we're retrying a checkpoint fetch due to a transient error, logging the
443    /// reason and error.
444    pub(crate) fn inc_retry(
445        &self,
446        checkpoint: u64,
447        reason: &str,
448        error: Error,
449    ) -> backoff::Error<Error> {
450        warn!(
451            checkpoint,
452            reason,
453            "Retrying due to error: {}",
454            error_with_sources(&error)
455        );
456
457        self.total_ingested_transient_retries
458            .with_label_values(&[reason, self.label.as_str()])
459            .inc();
460
461        backoff::Error::transient(error)
462    }
463}
464
465impl IndexerMetrics {
466    pub fn new(prefix: Option<&str>, registry: &Registry) -> Arc<Self> {
467        let prefix = prefix.unwrap_or(DEFAULT_METRICS_PREFIX);
468        let name = |n| format!("{prefix}_{n}");
469        Arc::new(Self {
470            total_handler_checkpoints_received: register_int_counter_vec_with_registry!(
471                name("total_handler_checkpoints_received"),
472                "Total number of checkpoints received by this handler",
473                &["pipeline"],
474                registry,
475            )
476            .unwrap(),
477            total_handler_processor_retries: register_int_counter_vec_with_registry!(
478                name("total_handler_processor_retries"),
479                "Total number of handler retries after transient processing failures",
480                &["pipeline"],
481                registry,
482            )
483            .unwrap(),
484            total_handler_checkpoints_processed: register_int_counter_vec_with_registry!(
485                name("total_handler_checkpoints_processed"),
486                "Total number of checkpoints processed (converted into rows) by this handler",
487                &["pipeline"],
488                registry,
489            )
490            .unwrap(),
491            total_handler_rows_created: register_int_counter_vec_with_registry!(
492                name("total_handler_rows_created"),
493                "Total number of rows created by this handler",
494                &["pipeline"],
495                registry,
496            )
497            .unwrap(),
498            latest_processed_checkpoint: register_int_gauge_vec_with_registry!(
499                name("latest_processed_checkpoint"),
500                "Latest checkpoint sequence number processed by this handler",
501                &["pipeline"],
502                registry,
503            )
504            .unwrap(),
505            latest_processed_checkpoint_timestamp_lag_ms: register_int_gauge_vec_with_registry!(
506                name("latest_processed_checkpoint_timestamp_lag_ms"),
507                "Difference between the system timestamp when the latest checkpoint was processed and the \
508                 timestamp in the checkpoint, in milliseconds",
509                &["pipeline"],
510                registry,
511            )
512            .unwrap(),
513            processed_checkpoint_timestamp_lag: register_histogram_vec_with_registry!(
514                name("processed_checkpoint_timestamp_lag"),
515                "Difference between the system timestamp when a checkpoint was processed and the \
516                 timestamp in each checkpoint, in seconds",
517                &["pipeline"],
518                LAG_SEC_BUCKETS.to_vec(),
519                registry,
520            )
521            .unwrap(),
522            handler_checkpoint_latency: register_histogram_vec_with_registry!(
523                name("handler_checkpoint_latency"),
524                "Time taken to process a checkpoint by this handler",
525                &["pipeline"],
526                PROCESSING_LATENCY_SEC_BUCKETS.to_vec(),
527                registry,
528            )
529            .unwrap(),
530            processor_concurrency_limit: register_int_gauge_vec_with_registry!(
531                name("processor_concurrency_limit"),
532                "Current adaptive concurrency limit for this processor",
533                &["pipeline"],
534                registry,
535            )
536            .unwrap(),
537            processor_concurrency_inflight: register_int_gauge_vec_with_registry!(
538                name("processor_concurrency_inflight"),
539                "Current number of in-flight processor tasks for this pipeline",
540                &["pipeline"],
541                registry,
542            )
543            .unwrap(),
544            total_collector_checkpoints_received: register_int_counter_vec_with_registry!(
545                name("total_collector_checkpoints_received"),
546                "Total number of checkpoints received by this collector",
547                &["pipeline"],
548                registry,
549            )
550            .unwrap(),
551            total_collector_rows_received: register_int_counter_vec_with_registry!(
552                name("total_collector_rows_received"),
553                "Total number of rows received by this collector",
554                &["pipeline"],
555                registry,
556            )
557            .unwrap(),
558            total_collector_batches_created: register_int_counter_vec_with_registry!(
559                name("total_collector_batches_created"),
560                "Total number of batches created by this collector",
561                &["pipeline"],
562                registry,
563            )
564            .unwrap(),
565            total_committer_batches_attempted: register_int_counter_vec_with_registry!(
566                name("total_committer_batches_attempted"),
567                "Total number of batches writes attempted by this committer",
568                &["pipeline"],
569                registry,
570            )
571            .unwrap(),
572            total_committer_batches_succeeded: register_int_counter_vec_with_registry!(
573                name("total_committer_batches_succeeded"),
574                "Total number of successful batches writes by this committer",
575                &["pipeline"],
576                registry,
577            )
578            .unwrap(),
579            total_committer_batches_failed: register_int_counter_vec_with_registry!(
580                name("total_committer_batches_failed"),
581                "Total number of failed batches writes by this committer",
582                &["pipeline"],
583                registry,
584            )
585            .unwrap(),
586            total_committer_rows_committed: register_int_counter_vec_with_registry!(
587                name("total_committer_rows_committed"),
588                "Total number of rows sent to the database by this committer",
589                &["pipeline"],
590                registry,
591            )
592            .unwrap(),
593            total_committer_rows_affected: register_int_counter_vec_with_registry!(
594                name("total_committer_rows_affected"),
595                "Total number of rows actually written to the database by this committer",
596                &["pipeline"],
597                registry,
598            )
599            .unwrap(),
600            total_watermarks_out_of_order: register_int_counter_vec_with_registry!(
601                name("watermark_out_of_order"),
602                "Number of times this committer encountered a batch for a checkpoint before its watermark",
603                &["pipeline"],
604                registry,
605            )
606            .unwrap(),
607            total_pruner_chunks_attempted: register_int_counter_vec_with_registry!(
608                name("pruner_chunks_attempted"),
609                "Number of chunks this pruner attempted to delete",
610                &["pipeline"],
611                registry,
612            )
613            .unwrap(),
614            total_pruner_chunks_deleted: register_int_counter_vec_with_registry!(
615                name("pruner_chunks_deleted"),
616                "Number of chunks this pruner successfully deleted",
617                &["pipeline"],
618                registry,
619            )
620            .unwrap(),
621            total_pruner_rows_deleted: register_int_counter_vec_with_registry!(
622                name("pruner_rows_deleted"),
623                "Number of rows this pruner successfully deleted",
624                &["pipeline"],
625                registry,
626            )
627            .unwrap(),
628            latest_collected_checkpoint: register_int_gauge_vec_with_registry!(
629                name("latest_collected_checkpoint"),
630                "Latest checkpoint sequence number collected by this collector",
631                &["pipeline"],
632                registry,
633            )
634            .unwrap(),
635            latest_collected_checkpoint_timestamp_lag_ms: register_int_gauge_vec_with_registry!(
636                name("latest_collected_checkpoint_timestamp_lag_ms"),
637                "Difference between the system timestamp when the latest checkpoint was collected and the \
638                 timestamp in the checkpoint, in milliseconds",
639                &["pipeline"],
640                registry,
641            )
642            .unwrap(),
643            collected_checkpoint_timestamp_lag: register_histogram_vec_with_registry!(
644                name("collected_checkpoint_timestamp_lag"),
645                "Difference between the system timestamp when a checkpoint was collected and the \
646                 timestamp in each checkpoint, in seconds",
647                &["pipeline"],
648                LAG_SEC_BUCKETS.to_vec(),
649                registry,
650            )
651            .unwrap(),
652            latest_partially_committed_checkpoint: register_int_gauge_vec_with_registry!(
653                name("latest_partially_committed_checkpoint"),
654                "Latest checkpoint sequence number partially committed by this collector",
655                &["pipeline"],
656                registry,
657            )
658            .unwrap(),
659            latest_partially_committed_checkpoint_timestamp_lag_ms: register_int_gauge_vec_with_registry!(
660                name("latest_partially_committed_checkpoint_timestamp_lag_ms"),
661                "Difference between the system timestamp when the latest checkpoint was partially committed and the \
662                 timestamp in the checkpoint, in milliseconds",
663                &["pipeline"],
664                registry,
665            )
666            .unwrap(),
667            partially_committed_checkpoint_timestamp_lag: register_histogram_vec_with_registry!(
668                name("partially_committed_checkpoint_timestamp_lag"),
669                "Difference between the system timestamp when a checkpoint was partially committed and the \
670                 timestamp in each checkpoint, in seconds",
671                &["pipeline"],
672                LAG_SEC_BUCKETS.to_vec(),
673                registry,
674            )
675            .unwrap(),
676            latest_watermarked_checkpoint_timestamp_lag_ms: register_int_gauge_vec_with_registry!(
677                name("latest_watermarked_checkpoint_timestamp_lag_ms"),
678                "Difference between the system timestamp when the latest checkpoint was watermarked and the \
679                 timestamp in the checkpoint, in milliseconds",
680                &["pipeline"],
681                registry,
682            )
683            .unwrap(),
684            watermarked_checkpoint_timestamp_lag: register_histogram_vec_with_registry!(
685                name("watermarked_checkpoint_timestamp_lag"),
686                "Difference between the system timestamp when a checkpoint was watermarked and the \
687                 timestamp in each checkpoint, in seconds",
688                &["pipeline"],
689                LAG_SEC_BUCKETS.to_vec(),
690                registry,
691            )
692            .unwrap(),
693            collector_gather_latency: register_histogram_vec_with_registry!(
694                name("collector_gather_latency"),
695                "Time taken to gather rows into a batch by this collector",
696                &["pipeline"],
697                PROCESSING_LATENCY_SEC_BUCKETS.to_vec(),
698                registry,
699            )
700            .unwrap(),
701            collector_batch_size: register_histogram_vec_with_registry!(
702                name("collector_batch_size"),
703                "Number of rows in a batch written to the database by this collector",
704                &["pipeline"],
705                BATCH_SIZE_BUCKETS.to_vec(),
706                registry,
707            )
708            .unwrap(),
709            total_collector_skipped_checkpoints: register_int_counter_vec_with_registry!(
710                name("total_collector_skipped_checkpoints"),
711                "Number of checkpoints skipped by the tasked pipeline's collector due to being below the main reader lo watermark",
712                &["pipeline"],
713                registry,
714            ).unwrap(),
715            collector_reader_lo: register_int_gauge_vec_with_registry!(
716                name("collector_reader_lo"),
717                "Reader low watermark as observed by the collector",
718                &["pipeline"],
719                registry,
720            )
721            .unwrap(),
722            committer_commit_latency: register_histogram_vec_with_registry!(
723                name("committer_commit_latency"),
724                "Time taken to write a batch of rows to the database by this committer",
725                &["pipeline"],
726                DB_UPDATE_LATENCY_SEC_BUCKETS.to_vec(),
727                registry,
728            )
729            .unwrap(),
730            committer_tx_rows: register_histogram_vec_with_registry!(
731                name("committer_tx_rows"),
732                "Number of rows written to the database in a single database transaction by this committer",
733                &["pipeline"],
734                BATCH_SIZE_BUCKETS.to_vec(),
735                registry,
736            )
737            .unwrap(),
738            watermark_gather_latency: register_histogram_vec_with_registry!(
739                name("watermark_gather_latency"),
740                "Time taken to calculate the new high watermark after a write by this committer",
741                &["pipeline"],
742                PROCESSING_LATENCY_SEC_BUCKETS.to_vec(),
743                registry,
744            )
745            .unwrap(),
746            watermark_commit_latency: register_histogram_vec_with_registry!(
747                name("watermark_commit_latency"),
748                "Time taken to write the new high watermark to the database by this committer",
749                &["pipeline"],
750                DB_UPDATE_LATENCY_SEC_BUCKETS.to_vec(),
751                registry,
752            )
753            .unwrap(),
754            watermark_pruner_read_latency: register_histogram_vec_with_registry!(
755                name("watermark_pruner_read_latency"),
756                "Time taken to read pruner's next upper and lowerbounds from the database by this pruner",
757                &["pipeline"],
758                DB_UPDATE_LATENCY_SEC_BUCKETS.to_vec(),
759                registry,
760            )
761            .unwrap(),
762            watermark_pruner_write_latency: register_histogram_vec_with_registry!(
763                name("watermark_pruner_write_latency"),
764                "Time taken to write the pruner's new upperbound to the database by this pruner",
765                &["pipeline"],
766                DB_UPDATE_LATENCY_SEC_BUCKETS.to_vec(),
767                registry,
768            )
769            .unwrap(),
770            pruner_delete_latency: register_histogram_vec_with_registry!(
771                name("pruner_delete_latency"),
772                "Time taken to delete a chunk of data from the database by this pruner",
773                &["pipeline"],
774                DB_UPDATE_LATENCY_SEC_BUCKETS.to_vec(),
775                registry,
776            )
777            .unwrap(),
778            watermark_epoch: register_int_gauge_vec_with_registry!(
779                name("watermark_epoch"),
780                "Current epoch high watermark for this committer",
781                &["pipeline"],
782                registry,
783            )
784            .unwrap(),
785            watermark_checkpoint: register_int_gauge_vec_with_registry!(
786                name("watermark_checkpoint"),
787                "Current checkpoint high watermark for this committer",
788                &["pipeline"],
789                registry,
790            )
791            .unwrap(),
792            watermark_transaction: register_int_gauge_vec_with_registry!(
793                name("watermark_transaction"),
794                "Current transaction high watermark for this committer",
795                &["pipeline"],
796                registry,
797            )
798            .unwrap(),
799            watermark_timestamp_ms: register_int_gauge_vec_with_registry!(
800                name("watermark_timestamp_ms"),
801                "Current timestamp high watermark for this committer, in milliseconds",
802                &["pipeline"],
803                registry,
804            )
805            .unwrap(),
806            watermark_reader_lo: register_int_gauge_vec_with_registry!(
807                name("watermark_reader_lo"),
808                "Current reader low watermark for this pruner",
809                &["pipeline"],
810                registry,
811            )
812            .unwrap(),
813            watermark_pruner_hi: register_int_gauge_vec_with_registry!(
814                name("watermark_pruner_hi"),
815                "Current pruner high watermark for this pruner",
816                &["pipeline"],
817                registry,
818            )
819            .unwrap(),
820            watermark_epoch_in_db: register_int_gauge_vec_with_registry!(
821                name("watermark_epoch_in_db"),
822                "Last epoch high watermark this committer wrote to the DB",
823                &["pipeline"],
824                registry,
825            )
826            .unwrap(),
827            watermark_checkpoint_in_db: register_int_gauge_vec_with_registry!(
828                name("watermark_checkpoint_in_db"),
829                "Last checkpoint high watermark this committer wrote to the DB",
830                &["pipeline"],
831                registry,
832            )
833            .unwrap(),
834            watermark_transaction_in_db: register_int_gauge_vec_with_registry!(
835                name("watermark_transaction_in_db"),
836                "Last transaction high watermark this committer wrote to the DB",
837                &["pipeline"],
838                registry,
839            )
840            .unwrap(),
841            watermark_timestamp_in_db_ms: register_int_gauge_vec_with_registry!(
842                name("watermark_timestamp_ms_in_db"),
843                "Last timestamp high watermark this committer wrote to the DB, in milliseconds",
844                &["pipeline"],
845                registry,
846            )
847            .unwrap(),
848            watermark_reader_lo_in_db: register_int_gauge_vec_with_registry!(
849                name("watermark_reader_lo_in_db"),
850                "Last reader low watermark this pruner wrote to the DB",
851                &["pipeline"],
852                registry,
853            )
854            .unwrap(),
855            watermark_pruner_hi_in_db: register_int_gauge_vec_with_registry!(
856                name("watermark_pruner_hi_in_db"),
857                "Last pruner high watermark this pruner wrote to the DB",
858                &["pipeline"],
859                registry,
860            )
861            .unwrap(),
862        })
863    }
864
865    pub(crate) fn inc_processor_retry<P: Processor>(
866        &self,
867        checkpoint: u64,
868        error: &anyhow::Error,
869        delay: Duration,
870    ) {
871        warn!(
872            pipeline = P::NAME,
873            checkpoint,
874            retry_delay_ms = delay.as_millis(),
875            "Retrying processor after error: {error:?}",
876        );
877
878        self.total_handler_processor_retries
879            .with_label_values(&[P::NAME])
880            .inc();
881    }
882}
883
884impl CheckpointLagMetricReporter {
885    pub fn new(
886        checkpoint_time_lag_histogram: Histogram,
887        latest_checkpoint_time_lag_gauge: IntGauge,
888        latest_checkpoint_sequence_number_gauge: IntGauge,
889    ) -> Arc<Self> {
890        Arc::new(Self {
891            checkpoint_time_lag_histogram,
892            latest_checkpoint_time_lag_gauge,
893            latest_checkpoint_sequence_number_gauge,
894            latest_reported_checkpoint: AtomicU64::new(0),
895        })
896    }
897
898    pub fn new_for_pipeline<P: Processor>(
899        checkpoint_time_lag_histogram: &HistogramVec,
900        latest_checkpoint_time_lag_gauge: &IntGaugeVec,
901        latest_checkpoint_sequence_number_gauge: &IntGaugeVec,
902    ) -> Arc<Self> {
903        Self::with_label(
904            checkpoint_time_lag_histogram,
905            latest_checkpoint_time_lag_gauge,
906            latest_checkpoint_sequence_number_gauge,
907            P::NAME,
908        )
909    }
910
911    /// Bind a set of checkpoint-lag vecs to a single `label` value, so callers reporting under
912    /// different labels (e.g. one per cohort) don't overwrite each other's checkpoint-lag gauges.
913    pub fn with_label(
914        checkpoint_time_lag_histogram: &HistogramVec,
915        latest_checkpoint_time_lag_gauge: &IntGaugeVec,
916        latest_checkpoint_sequence_number_gauge: &IntGaugeVec,
917        label: &str,
918    ) -> Arc<Self> {
919        Self::new(
920            checkpoint_time_lag_histogram.with_label_values(&[label]),
921            latest_checkpoint_time_lag_gauge.with_label_values(&[label]),
922            latest_checkpoint_sequence_number_gauge.with_label_values(&[label]),
923        )
924    }
925
926    pub fn report_lag(&self, cp_sequence_number: u64, checkpoint_timestamp_ms: u64) {
927        let lag = chrono::Utc::now().timestamp_millis() - checkpoint_timestamp_ms as i64;
928        self.checkpoint_time_lag_histogram
929            .observe((lag as f64) / 1000.0);
930
931        let prev = self
932            .latest_reported_checkpoint
933            .fetch_max(cp_sequence_number, std::sync::atomic::Ordering::Relaxed);
934        if cp_sequence_number > prev {
935            self.latest_checkpoint_sequence_number_gauge
936                .set(cp_sequence_number as i64);
937            self.latest_checkpoint_time_lag_gauge.set(lag);
938        }
939    }
940}
941
942/// Render an error together with its [`std::error::Error::source`] chain. Several wrappers
943/// (object_store, reqwest, anyhow) already inline their source's message in their own `Display`,
944/// so we skip any source whose message is already contained in its parent's message to avoid
945/// repetition, and append only genuinely-new sources. This surfaces the underlying `io::Error`
946/// ("Connection refused" / "operation timed out") that is otherwise hidden behind reqwest's terse
947/// "error sending request" leaf.
948fn error_with_sources(mut err: &dyn std::error::Error) -> String {
949    let mut out = String::new();
950    let mut next = err.to_string();
951
952    let mut prefix = "";
953    while let Some(src) = err.source() {
954        err = src;
955        let msg = err.to_string();
956        if !next.contains(&msg) {
957            out.push_str(prefix);
958            out.push_str(&next);
959            prefix = ": ";
960            next = msg;
961        }
962    }
963
964    out.push_str(prefix);
965    out.push_str(&next);
966    out
967}
968
969#[cfg(test)]
970pub(crate) mod tests {
971    use std::sync::Arc;
972
973    use prometheus::Registry;
974
975    use super::*;
976
977    /// Construct IndexerMetrics for test purposes.
978    pub fn test_metrics() -> Arc<IndexerMetrics> {
979        IndexerMetrics::new(None, &Registry::new())
980    }
981
982    /// Construct IngestionMetrics for test purposes.
983    pub fn test_ingestion_metrics() -> Arc<IngestionMetrics> {
984        IngestionMetrics::new(None, &Registry::new())
985    }
986
987    #[test]
988    fn test_error_with_sources_dedupes_inlined_and_appends_new() {
989        /// Leaf error with no source — mimics the underlying `io::Error`.
990        #[derive(thiserror::Error, Debug)]
991        #[error("operation timed out")]
992        struct Leaf;
993
994        /// Wrapper whose `Display` does NOT inline its source — mimics reqwest's
995        /// terse "error sending request" that hides the real source.
996        #[derive(thiserror::Error, Debug)]
997        #[error("error sending request")]
998        struct Terse(#[source] Leaf);
999
1000        /// Wrapper whose `Display` DOES inline its source — mimics object_store.
1001        #[derive(thiserror::Error, Debug)]
1002        #[error("Error performing GET: {0}")]
1003        struct Inlining(#[source] Terse);
1004
1005        // The already-inlined "error sending request" fragment is not repeated, but the
1006        // genuinely-new "operation timed out" source (hidden behind the terse leaf) is appended.
1007        let err = Inlining(Terse(Leaf));
1008        assert_eq!(
1009            error_with_sources(&err),
1010            "Error performing GET: error sending request: operation timed out"
1011        );
1012    }
1013
1014    #[test]
1015    fn test_error_with_sources_dedupes_inlined_mid_message() {
1016        /// Leaf error with no source.
1017        #[derive(thiserror::Error, Debug)]
1018        #[error("operation timed out")]
1019        struct Leaf;
1020
1021        /// Wrapper that inlines its source in the *middle* of its `Display`, not as a suffix.
1022        #[derive(thiserror::Error, Debug)]
1023        #[error("failed [{0}] while fetching")]
1024        struct MidInlining(#[source] Leaf);
1025
1026        // The inlined source is contained in its parent's message, so it is not appended again
1027        // even though the parent's message does not end with it.
1028        let err = MidInlining(Leaf);
1029        assert_eq!(
1030            error_with_sources(&err),
1031            "failed [operation timed out] while fetching"
1032        );
1033    }
1034}
sui_indexer_alt_framework/metrics.rs

sui_indexer_alt_framework/
metrics.rs
