feat(cdc): add cdc backfill grafana metrics (#13777)

risingwavelabs · Dec 4, 2023 · 41f965e · 41f965e
1 parent 1bb99bb
commit 41f965e
Show file tree

Hide file tree

Showing 7 changed files with 50 additions and 6 deletions.
diff --git a/docker/dashboards/risingwave-dev-dashboard.json b/docker/dashboards/risingwave-dev-dashboard.json
diff --git a/docker/dashboards/risingwave-user-dashboard.json b/docker/dashboards/risingwave-user-dashboard.json
diff --git a/grafana/risingwave-dev-dashboard.dashboard.py b/grafana/risingwave-dev-dashboard.dashboard.py
@@ -827,6 +827,28 @@ def section_streaming(outer_panels):
                         ),
                     ],
                 ),
+                panels.timeseries_rowsps(
+                    "CDC Backfill Snapshot Read Throughput(rows)",
+                    "Total number of rows that have been read from the cdc backfill snapshot",
+                    [
+                        panels.target(
+                            f"rate({table_metric('stream_cdc_backfill_snapshot_read_row_count')}[$__rate_interval])",
+                            "table_id={{table_id}} actor={{actor_id}} @ {{%s}}"
+                            % NODE_LABEL,
+                            ),
+                    ],
+                ),
+                panels.timeseries_rowsps(
+                    "CDC Backfill Upstream Throughput(rows)",
+                    "Total number of rows that have been output from the cdc backfill upstream",
+                    [
+                        panels.target(
+                            f"rate({table_metric('stream_cdc_backfill_upstream_output_row_count')}[$__rate_interval])",
+                            "table_id={{table_id}} actor={{actor_id}} @ {{%s}}"
+                            % NODE_LABEL,
+                            ),
+                    ],
+                ),
                 panels.timeseries_count(
                     "Barrier Number",
                     "The number of barriers that have been ingested but not completely processed. This metric reflects the "

diff --git a/grafana/risingwave-dev-dashboard.json b/grafana/risingwave-dev-dashboard.json
diff --git a/grafana/risingwave-user-dashboard.json b/grafana/risingwave-user-dashboard.json
diff --git a/src/stream/src/executor/backfill/cdc/cdc_backfill.rs b/src/stream/src/executor/backfill/cdc/cdc_backfill.rs
@@ -337,15 +337,15 @@ impl<S: StateStore> CdcBackfillExecutor<S> {
                                     }
 
                                     self.metrics
-                                        .backfill_snapshot_read_row_count
+                                        .cdc_backfill_snapshot_read_row_count
                                         .with_label_values(&[
                                             upstream_table_id.to_string().as_str(),
                                             self.actor_ctx.id.to_string().as_str(),
                                         ])
                                         .inc_by(cur_barrier_snapshot_processed_rows);
 
                                     self.metrics
-                                        .backfill_upstream_output_row_count
+                                        .cdc_backfill_upstream_output_row_count
                                         .with_label_values(&[
                                             upstream_table_id.to_string().as_str(),
                                             self.actor_ctx.id.to_string().as_str(),

diff --git a/src/stream/src/executor/monitor/streaming_stats.rs b/src/stream/src/executor/monitor/streaming_stats.rs
@@ -122,6 +122,10 @@ pub struct StreamingMetrics {
     pub arrangement_backfill_snapshot_read_row_count: GenericCounterVec<AtomicU64>,
     pub arrangement_backfill_upstream_output_row_count: GenericCounterVec<AtomicU64>,
 
+    // CDC Backfill
+    pub cdc_backfill_snapshot_read_row_count: GenericCounterVec<AtomicU64>,
+    pub cdc_backfill_upstream_output_row_count: GenericCounterVec<AtomicU64>,
+
     // Over Window
     pub over_window_cached_entry_count: GenericGaugeVec<AtomicI64>,
     pub over_window_cache_lookup_count: GenericCounterVec<AtomicU64>,
@@ -684,6 +688,22 @@ impl StreamingMetrics {
             )
             .unwrap();
 
+        let cdc_backfill_snapshot_read_row_count = register_int_counter_vec_with_registry!(
+            "stream_cdc_backfill_snapshot_read_row_count",
+            "Total number of rows that have been read from the cdc_backfill snapshot",
+            &["table_id", "actor_id"],
+            registry
+        )
+        .unwrap();
+
+        let cdc_backfill_upstream_output_row_count = register_int_counter_vec_with_registry!(
+            "stream_cdc_backfill_upstream_output_row_count",
+            "Total number of rows that have been output from the cdc_backfill upstream",
+            &["table_id", "actor_id"],
+            registry
+        )
+        .unwrap();
+
         let over_window_cached_entry_count = register_int_gauge_vec_with_registry!(
             "stream_over_window_cached_entry_count",
             "Total entry (partition) count in over window executor cache",
@@ -980,6 +1000,8 @@ impl StreamingMetrics {
             backfill_upstream_output_row_count,
             arrangement_backfill_snapshot_read_row_count,
             arrangement_backfill_upstream_output_row_count,
+            cdc_backfill_snapshot_read_row_count,
+            cdc_backfill_upstream_output_row_count,
             over_window_cached_entry_count,
             over_window_cache_lookup_count,
             over_window_cache_miss_count,