risingwavelabs · Sunt-ing · Mar 9, 2022 · Mar 1, 2022 · Mar 1, 2022 · Mar 1, 2022
diff --git a/docs/developer/benchmark_tool/state_store.md b/docs/developer/benchmark_tool/state_store.md
@@ -9,9 +9,9 @@
  --writes 10000 \
  --reads 500 \
  --scans 200 \
- --deletes 2000
+ --deletes 2000 \
  --concurrency-num 4 \
- --seed 233
+ --seed 233 \
  --statistics
 ```
 
@@ -138,7 +138,7 @@ Example: `--benchmarks "writebatch,prefixscanrandom,getrandom"`
   - Size (bytes) of each prefix.
   - Default: 5
 
-- `--keys_per_prefix`
+- `--keys-per-prefix`
 
   - Control **average** number of keys generated per prefix.
   - Default: 10
@@ -158,6 +158,9 @@ Example: `--benchmarks "writebatch,prefixscanrandom,getrandom"`
 - `--statistics`
   - Detailed statistics of storage backend
 
+- `--calibrate-metric`
+  - Print performance by both self-measured metric and the state store metric system. This can be used to calibrate metric parameters, especially bucket specification.
+
 # Metrics
 
 - Letancy (`min/mean/P50/P95/P99/max/std_dev`)

diff --git a/rust/bench/ss_bench/main.rs b/rust/bench/ss_bench/main.rs
@@ -10,7 +10,7 @@ use risingwave_rpc_client::MetaClient;
 use risingwave_storage::monitor::DEFAULT_STATE_STORE_STATS;
 use risingwave_storage::{dispatch_state_store, StateStoreImpl};
 
-use crate::utils::store_statistics::print_statistics;
+use crate::utils::display_stat::print_statistics;
 
 #[derive(Parser, Debug)]
 pub(crate) struct Opts {
@@ -76,6 +76,9 @@ pub(crate) struct Opts {
     // ----- flag -----
     #[clap(long)]
     statistics: bool,
+
+    #[clap(long)]
+    calibrate_metric: bool,
 }
 
 fn preprocess_options(opts: &mut Opts) {

diff --git a/rust/bench/ss_bench/operations/mod.rs b/rust/bench/ss_bench/operations/mod.rs
@@ -5,9 +5,9 @@ use rand::prelude::StdRng;
 use rand::SeedableRng;
 use risingwave_storage::StateStore;
 
+use crate::utils::display_stat::*;
 use crate::utils::latency_stat::LatencyStat;
 use crate::Opts;
-
 pub(crate) mod get;
 pub(crate) mod prefix_scan_random;
 pub(crate) mod write_batch;
@@ -31,13 +31,17 @@ pub(crate) struct PerfMetrics {
 impl Operations {
     /// Run operations in the `--benchmarks` option
     pub(crate) async fn run(store: impl StateStore, opts: &Opts) {
+        let mut stat_display = DisplayStat::default();
+
         let mut runner = Operations {
             keys: vec![],
             prefixes: vec![],
             rng: StdRng::seed_from_u64(opts.seed),
         };
 
         for operation in opts.benchmarks.split(',') {
+            // (Sun Ting) TODO: remove statistics print for each operation
+            // after new performance display is ready
             match operation {
                 "writebatch" => runner.write_batch(&store, opts).await,
                 "deleterandom" => runner.delete_random(&store, opts).await,
@@ -46,6 +50,21 @@ impl Operations {
                 "prefixscanrandom" => runner.prefix_scan_random(&store, opts).await,
                 other => unimplemented!("operation \"{}\" is not supported.", other),
             }
+
+            stat_display.update_stat();
+
+            // display metrics from state store metric system
+            if opts.calibrate_metric {
+                match operation {
+                    "writebatch" => stat_display.display_write_batch(),
+                    "deleterandom" => stat_display.display_delete_random(),
+                    // (Sun Ting) TODO: implement other performance displays
+                    "getrandom" => {}
+                    "getseq" => {}
+                    "prefixscanrandom" => {}
+                    other => unimplemented!("operation \"{}\" is not supported.", other),
+                }
+            }
         }
     }
 

diff --git a/rust/bench/ss_bench/utils/display_stat.rs b/rust/bench/ss_bench/utils/display_stat.rs
@@ -0,0 +1,119 @@
+use prometheus::core::{AtomicU64, Collector, GenericCounter, Metric};
+use prometheus::Histogram;
+use risingwave_storage::for_all_metrics;
+use risingwave_storage::monitor::DEFAULT_STATE_STORE_STATS;
+
+use super::my_metrics::MyStateStoreStats;
+use crate::utils::my_metrics::MyHistogram;
+
+#[derive(Default)]
+pub(crate) struct DisplayStat {
+    pub(crate) prev_stat: MyStateStoreStats,
+    pub(crate) cur_stat: MyStateStoreStats,
+}
+
+impl DisplayStat {
+    pub(crate) fn update_stat(&mut self) {
+        // (Ting Sun) TODO: eliminate this clone
+        self.prev_stat = self.cur_stat.clone();
+        self.cur_stat = MyStateStoreStats::from_prom_stats(&**DEFAULT_STATE_STORE_STATS);
+    }
+
+    pub(crate) fn display_write_batch(&mut self) {
+        let perf = self.display_batch_inner();
+
+        println!(
+            "
+    writebatch
+      {}
+      OPS: {}  {} bytes/sec",
+            perf.histogram, perf.qps, perf.bytes_pre_sec
+        );
+    }
+
+    pub(crate) fn display_delete_random(&mut self) {
+        let perf = self.display_batch_inner();
+
+        println!(
+            "
+    deleterandom
+      {}
+      OPS: {}  {} bytes/sec",
+            perf.histogram, perf.qps, perf.bytes_pre_sec
+        );
+    }
+
+    fn display_batch_inner(&mut self) -> PerfMetrics {
+        let prev_latency_hist = &self.prev_stat.batch_write_latency;
+        let cur_latency_hist = &self.cur_stat.batch_write_latency;
+
+        let time_consume = cur_latency_hist.total_sum - prev_latency_hist.total_sum;
+
+        let ops = {
+            let written_batch_num = cur_latency_hist.total_count - prev_latency_hist.total_count;
+            written_batch_num as f64 / time_consume
+        };
+
+        let bytes_pre_sec = {
+            let prev_histogram = &self.prev_stat.batch_write_size;
+            let cur_histogram = &self.cur_stat.batch_write_size;
+
+            let written_bytes = cur_histogram.total_sum - prev_histogram.total_sum;
+            written_bytes / time_consume
+        };
+
+        PerfMetrics {
+            histogram: MyHistogram::from_diff(prev_latency_hist, cur_latency_hist),
+            qps: ops,
+            bytes_pre_sec,
+        }
+    }
+}
+
+pub(crate) struct PerfMetrics {
+    histogram: MyHistogram,
+    qps: f64,
+    bytes_pre_sec: f64,
+}
+
+/// Define extension method `print` used in `print_statistics`.
+trait Print {
+    fn print(&self);
+}
+
+impl Print for GenericCounter<AtomicU64> {
+    fn print(&self) {
+        let desc = &self.desc()[0].fq_name;
+        let counter = self.metric().get_counter().get_value() as u64;
+        println!("{desc} COUNT : {counter}");
+    }
+}
+
+impl Print for Histogram {
+    fn print(&self) {
+        let desc = &self.desc()[0].fq_name;
+
+        let histogram = MyHistogram::from_prom_hist(self.metric().get_histogram());
+        let p50 = histogram.get_percentile(50.0);
+        let p95 = histogram.get_percentile(95.0);
+        let p99 = histogram.get_percentile(99.0);
+        let p100 = histogram.get_percentile(100.0);
+
+        let sample_count = self.get_sample_count();
+        let sample_sum = self.get_sample_sum();
+
+        println!("{desc} P50 : {p50} P95 : {p95} P99 : {p99} P100 : {p100} COUNT : {sample_count} SUM : {sample_sum}");
+    }
+}
+
+macro_rules! print_statistics {
+    ($( $name:ident: $type:ty ),* ,) => {
+        pub(crate) fn print_statistics(stats: &risingwave_storage::monitor::StateStoreStats) {
+            println!("STATISTICS:");
+            // print for all fields
+            $( stats.$name.print(); )*
+            println!();
+        }
+    }
+}
+for_all_metrics! { print_statistics }
diff --git a/rust/bench/ss_bench/utils/mod.rs b/rust/bench/ss_bench/utils/mod.rs
@@ -1,3 +1,4 @@
+pub(crate) mod display_stat;
 pub(crate) mod latency_stat;
-pub(crate) mod store_statistics;
+pub(crate) mod my_metrics;
 pub(crate) mod workload;