feat(storage): monitor avg_key_size and avg_epoch_count

grafana/risingwave-dashboard.dashboard.py

@@ -768,7 +768,7 @@ def section_compaction(outer_panels):
                     "Total bytes gotten from sstable_avg_key_size, for observing sstable_avg_key_size",
                     [
                         panels.target(
-                            f"sum by(le, job, instance)(rate({metric('compactor_sstable_avg_key_size_sum')}[$__rate_interval]))  / sum by(le, job, instance)(rate({metric('state_store_sstable_avg_key_size_count')}[$__rate_interval]))",
+                            f"sum by(le, job, instance)(rate({metric('compactor_sstable_avg_key_size_sum')}[$__rate_interval]))  / sum by(le, job, instance)(rate({metric('compactor_sstable_avg_key_size_count')}[$__rate_interval]))",
                             "avg_key_size - {{job}} @ {{instance}}",
                         ),
                         panels.target(
@@ -778,6 +778,17 @@ def section_compaction(outer_panels):
                     ],
                 ),
 
+                 panels.timeseries_count(
+                    "Hummock Sstable Stat",
+                    "Avg count gotten from sstable_distinct_epoch_count, for observing sstable_distinct_epoch_count",
+                    [
+                        panels.target(
+                            f"sum by(le, job, instance)(rate({metric('compactor_sstable_distinct_epoch_count_sum')}[$__rate_interval]))  / sum by(le, job, instance)(rate({metric('compactor_sstable_distinct_epoch_count_count')}[$__rate_interval]))",
+                            "avg_epoch_count - {{job}} @ {{instance}}",
+                        ),
+                    ],
+                ),
+
                 panels.timeseries_latency(
                     "Hummock Remote Read Duration",
                     "Total time of operations which read from remote storage when enable prefetch",

src/storage/src/hummock/sstable/builder.rs

@@ -83,6 +83,7 @@ pub struct SstableBuilderOutput<WO> {
     pub writer_output: WO,
     pub avg_key_size: usize,
     pub avg_value_size: usize,
+    pub epoch_count: usize,
 }
 
 pub struct SstableBuilder<W: SstableWriter, F: FilterBuilder> {
@@ -119,8 +120,7 @@ pub struct SstableBuilder<W: SstableWriter, F: FilterBuilder> {
 
     filter_builder: F,
 
-    min_epoch: u64,
-    max_epoch: u64,
+    epoch_set: BTreeSet<u64>,
 }
 
 impl<W: SstableWriter> SstableBuilder<W, XorFilterBuilder> {
@@ -168,8 +168,7 @@ impl<W: SstableWriter, F: FilterBuilder> SstableBuilder<W, F> {
             total_key_count: 0,
             table_stats: Default::default(),
             last_table_stats: Default::default(),
-            min_epoch: u64::MAX,
-            max_epoch: u64::MIN,
+            epoch_set: BTreeSet::default(),
         }
     }
 
@@ -238,6 +237,8 @@ impl<W: SstableWriter, F: FilterBuilder> SstableBuilder<W, F> {
         self.total_key_count += 1;
         self.last_table_stats.total_key_count += 1;
 
+        self.epoch_set.insert(full_key.epoch);
+
         if is_new_table && !self.block_builder.is_empty() {
             self.build_block().await?;
         }
@@ -263,9 +264,6 @@ impl<W: SstableWriter, F: FilterBuilder> SstableBuilder<W, F> {
         self.raw_key.clear();
         self.raw_value.clear();
 
-        self.min_epoch = cmp::min(self.min_epoch, full_key.epoch);
-        self.max_epoch = cmp::max(self.max_epoch, full_key.epoch);
-
         if self.block_builder.approximate_len() >= self.options.block_capacity {
             self.build_block().await?;
         }
@@ -363,6 +361,48 @@ impl<W: SstableWriter, F: FilterBuilder> SstableBuilder<W, F> {
             (tombstone_min_epoch, tombstone_max_epoch)
         };
 
+        let (avg_key_size, avg_value_size) = if self.table_stats.is_empty() {
+            (0, 0)
+        } else {
+            let total_key_count: usize = self
+                .table_stats
+                .values()
+                .map(|s| s.total_key_count as usize)
+                .sum();
+
+            if total_key_count == 0 {
+                (0, 0)
+            } else {
+                let total_key_size: usize = self
+                    .table_stats
+                    .values()
+                    .map(|s| s.total_key_size as usize)
+                    .sum();
+
+                let total_value_size: usize = self
+                    .table_stats
+                    .values()
+                    .map(|s| s.total_value_size as usize)
+                    .sum();
+
+                (
+                    total_key_size / total_key_count,
+                    total_value_size / total_key_count,
+                )
+            }
+        };
+
+        let (min_epoch, max_epoch) = {
+            if self.epoch_set.is_empty() {
+                (u64::MAX, u64::MIN)
+            } else {
+                (
+                    *self.epoch_set.first().unwrap(),
+                    *self.epoch_set.last().unwrap(),
+                )
+            }
+        };
+
         let sst_info = SstableInfo {
             id: self.sstable_id,
             key_range: Some(risingwave_pb::hummock::KeyRange {
@@ -377,43 +417,29 @@ impl<W: SstableWriter, F: FilterBuilder> SstableBuilder<W, F> {
             total_key_count: self.total_key_count,
             divide_version: 0,
             uncompressed_file_size: uncompressed_file_size + meta.encoded_size() as u64,
-            min_epoch: cmp::min(self.min_epoch, tombstone_min_epoch),
-            max_epoch: cmp::max(self.max_epoch, tombstone_max_epoch),
+            min_epoch: cmp::min(min_epoch, tombstone_min_epoch),
+            max_epoch: cmp::max(max_epoch, tombstone_max_epoch),
         };
         tracing::trace!(
-            "meta_size {} bloom_filter_size {}  add_key_counts {} stale_key_count {} min_epoch {} max_epoch {}",
+            "meta_size {} bloom_filter_size {}  add_key_counts {} stale_key_count {} min_epoch {} max_epoch {} epoch_count {}",
             meta.encoded_size(),
             meta.bloom_filter.len(),
             self.total_key_count,
             self.stale_key_count,
-            self.min_epoch,
-            self.max_epoch,
+            min_epoch,
+            max_epoch,
+            self.epoch_set.len()
         );
         let bloom_filter_size = meta.bloom_filter.len();
-        let (avg_key_size, avg_value_size) = if self.table_stats.is_empty() {
-            (0, 0)
-        } else {
-            let avg_key_size = self
-                .table_stats
-                .values()
-                .map(|s| s.total_key_size as usize)
-                .sum::<usize>()
-                / self.table_stats.len();
-            let avg_value_size = self
-                .table_stats
-                .values()
-                .map(|s| s.total_value_size as usize)
-                .sum::<usize>()
-                / self.table_stats.len();
-            (avg_key_size, avg_value_size)
-        };
+
         let writer_output = self.writer.finish(meta).await?;
         Ok(SstableBuilderOutput::<W::Output> {
             sst_info: LocalSstableInfo::with_stats(sst_info, self.table_stats),
             bloom_filter_size,
             writer_output,
             avg_key_size,
             avg_value_size,
+            epoch_count: self.epoch_set.len(),
         })
     }
 

src/storage/src/hummock/sstable/multi_builder.rs

@@ -198,6 +198,12 @@ where
                         .sstable_avg_value_size
                         .observe(builder_output.avg_value_size as _);
                 }
+
+                if builder_output.epoch_count != 0 {
+                    self.compactor_metrics
+                        .sstable_distinct_epoch_count
+                        .observe(builder_output.epoch_count as _);
+                }
             }
             self.sst_outputs.push(SplitTableOutput {
                 upload_join_handle: builder_output.writer_output,

src/storage/src/monitor/compactor_metrics.rs

@@ -42,6 +42,7 @@ pub struct CompactorMetrics {
     pub sstable_avg_value_size: Histogram,
     pub iter_scan_key_counts: GenericCounterVec<AtomicU64>,
     pub write_build_l0_bytes: GenericCounter<AtomicU64>,
+    pub sstable_distinct_epoch_count: Histogram,
 }
 
 impl CompactorMetrics {
@@ -171,7 +172,7 @@ impl CompactorMetrics {
         let opts = histogram_opts!(
             "compactor_sstable_avg_value_size",
             "Total bytes gotten from sstable_avg_value_size, for observing sstable_avg_value_size",
-            exponential_buckets(1.0, 2.0, 25).unwrap() // max 16MB
+            exponential_buckets(1.0, 2.0, 26).unwrap() // max 32MB
         );
 
         let sstable_avg_value_size = register_histogram_with_registry!(opts, registry).unwrap();
@@ -198,6 +199,15 @@ impl CompactorMetrics {
             registry
         ).unwrap();
 
+        let opts = histogram_opts!(
+            "compactor_sstable_distinct_epoch_count",
+            "Total number gotten from sstable_distinct_epoch_count, for observing sstable_distinct_epoch_count",
+            exponential_buckets(1.0, 2.0, 17).unwrap()
+        );
+
+        let sstable_distinct_epoch_count =
+            register_histogram_with_registry!(opts, registry).unwrap();
+
         Self {
             compaction_upload_sst_counts,
             compact_write_bytes,
@@ -219,6 +229,7 @@ impl CompactorMetrics {
             sstable_avg_value_size,
             iter_scan_key_counts,
             write_build_l0_bytes,
+            sstable_distinct_epoch_count,
         }
     }