feat(storage): add metrics for compaction result check

grafana/risingwave-dev-dashboard.dashboard.py

@@ -214,6 +214,10 @@ def section_compaction(outer_panels):
                             f"sum({metric('storage_level_compact_frequency', successful_compaction_fiiler)}) by (compactor, group, task_type, result)",
                             "{{task_type}} - {{result}} - group-{{group}} @ {{compactor}}",
                         ),
+                        panels.target(
+                            f"sum({metric('storage_compaction_result_check_count')}) by (compactor, type)",
+                            "{{type}} @ {{compactor}}",
+                        ),
                     ],
                 ),
                 panels.timeseries_count(

src/storage/src/hummock/compactor/mod.rs

@@ -504,6 +504,7 @@ pub fn start_compactor(
                                             tracing::warn!(error = %e.as_report(), "Failed to report task {task_id:?}");
                                         }
                                         if enable_check_compaction_result && need_check_task {
+                                            context.compactor_metrics.compaction_result_check_count.with_label_values(&["compaction"]).inc();
                                             match check_compaction_result(&compact_task, context.clone()).await {
                                                 Err(e) => {
                                                     tracing::warn!(error = %e.as_report(), "Failed to check compaction task {}",compact_task.task_id);

src/storage/src/hummock/compactor/shared_buffer_compact.rs

@@ -233,6 +233,11 @@ async fn compact_shared_buffer(
             level0.extend(ssts);
         }
         if context.storage_opts.check_compaction_result {
+            context
+                .compactor_metrics
+                .compaction_result_check_count
+                .with_label_values(&["flush"])
+                .inc();
             let compaction_executor = context.compaction_executor.clone();
             let mut forward_iters = Vec::with_capacity(payload.len());
             let del_iter = ForwardMergeRangeIterator::new(HummockEpoch::MAX);

src/storage/src/monitor/compactor_metrics.rs

@@ -50,6 +50,7 @@ pub struct CompactorMetrics {
     pub sstable_distinct_epoch_count: Histogram,
     pub compaction_event_consumed_latency: Histogram,
     pub compaction_event_loop_iteration_latency: Histogram,
+    pub compaction_result_check_count: GenericCounterVec<AtomicU64>,
 }
 
 pub static GLOBAL_COMPACTOR_METRICS: LazyLock<CompactorMetrics> =
@@ -250,7 +251,13 @@ impl CompactorMetrics {
         );
         let compaction_event_loop_iteration_latency =
             register_histogram_with_registry!(opts, registry).unwrap();
-
+        let compaction_result_check_count = register_int_counter_vec_with_registry!(
+            "storage_compaction_result_check_count",
+            "KBs read from current level during history compactions to next level",
+            &["type"],
+            registry
+        )
+        .unwrap();
         Self {
             compaction_upload_sst_counts,
             compact_fast_runner_bytes,
@@ -277,6 +284,7 @@ impl CompactorMetrics {
             sstable_distinct_epoch_count,
             compaction_event_consumed_latency,
             compaction_event_loop_iteration_latency,
+            compaction_result_check_count,
         }
     }