feat(storage): add scan metrics (#662)

risingwavelabs · Mar 3, 2022 · b875e31 · b875e31
1 parent 6979c0e
commit b875e31
Show file tree

Hide file tree

Showing 2 changed files with 43 additions and 2 deletions.
diff --git a/rust/storage/src/monitor/monitored_store.rs b/rust/storage/src/monitor/monitored_store.rs
@@ -85,7 +85,15 @@ where
     {
         self.stats.range_scan_counts.inc();
 
-        self.inner.scan(key_range, limit, epoch).await
+        let timer = self.stats.range_scan_latency.start_timer();
+        let result = self.inner.scan(key_range, limit, epoch).await?;
+        timer.observe_duration();
+
+        self.stats
+            .range_scan_size
+            .observe(result.iter().map(|(k, v)| k.len() + v.len()).sum::<usize>() as _);
+
+        Ok(result)
     }
 
     async fn reverse_scan<R, B>(
@@ -100,7 +108,15 @@ where
     {
         self.stats.reverse_range_scan_counts.inc();
 
-        self.inner.scan(key_range, limit, epoch).await
+        let timer = self.stats.range_scan_latency.start_timer();
+        let result = self.inner.scan(key_range, limit, epoch).await?;
+        timer.observe_duration();
+
+        self.stats
+            .range_scan_size
+            .observe(result.iter().map(|(k, v)| k.len() + v.len()).sum::<usize>() as _);
+
+        Ok(result)
     }
 
     async fn ingest_batch(&self, kv_pairs: Vec<(Bytes, Option<Bytes>)>, epoch: u64) -> Result<()> {

diff --git a/rust/storage/src/monitor/state_store_stats.rs b/rust/storage/src/monitor/state_store_stats.rs
@@ -20,6 +20,9 @@ pub const BATCH_WRITE_LATENCY_SCALE: f64 = 0.1;
 pub const BATCH_WRITE_BUILD_TABLE_LATENCY_SCALE: f64 = 0.0001;
 pub const BATCH_WRITE_ADD_L0_LATENCT_SCALE: f64 = 0.00001;
 
+pub const RANGE_SCAN_SIZE_SCALE: f64 = 10000.0;
+pub const RANGE_SCAN_LATENCY_SCALE: f64 = 0.1;
+
 pub const ITER_NEXT_LATENCY_SCALE: f64 = 0.0001;
 pub const ITER_SEEK_LATENCY_SCALE: f64 = 0.0001;
 pub const ITER_NEXT_SIZE_SCALE: f64 = 400.0;
@@ -46,6 +49,8 @@ macro_rules! for_all_metrics {
 
             range_scan_counts: GenericCounter<AtomicU64>,
             reverse_range_scan_counts: GenericCounter<AtomicU64>,
+            range_scan_size: Histogram,
+            range_scan_latency: Histogram,
 
             batched_write_counts: GenericCounter<AtomicU64>,
             batch_write_tuple_counts: GenericCounter<AtomicU64>,
@@ -163,6 +168,24 @@ impl StateStoreStats {
         )
         .unwrap();
 
+        let buckets = DEFAULT_BUCKETS.map(|x| x * RANGE_SCAN_SIZE_SCALE).to_vec();
+        let opts = histogram_opts!(
+            "state_store_range_scan_size",
+            "Total bytes gotten from state store scan(), for calculating read throughput",
+            buckets
+        );
+        let range_scan_size = register_histogram_with_registry!(opts, registry).unwrap();
+
+        let buckets = DEFAULT_BUCKETS
+            .map(|x| x * RANGE_SCAN_LATENCY_SCALE)
+            .to_vec();
+        let opts = histogram_opts!(
+            "state_store_range_scan_latency",
+            "Total time of scan that have been issued to state store",
+            buckets
+        );
+        let range_scan_latency = register_histogram_with_registry!(opts, registry).unwrap();
+
         // ----- write_batch -----
         let batched_write_counts = register_int_counter_with_registry!(
             "state_store_batched_write_counts",
@@ -417,6 +440,8 @@ impl StateStoreStats {
 
             range_scan_counts,
             reverse_range_scan_counts,
+            range_scan_size,
+            range_scan_latency,
 
             batched_write_counts,
             batch_write_tuple_counts,