feat: add strategy config to fluvio benchmark (#4338)

Author: fraidev

crates/fluvio-benchmark/src/benchmark_driver.rs

@@ -1,20 +1,44 @@
+use std::{fs::File, time::Duration};
+
 use anyhow::Result;
+use fluvio_future::timer::sleep;
 
-use crate::{cli::BenchmarkCmd, producer_benchmark::ProducerBenchmark};
+use crate::{cli::BenchmarkMode, config::config_matrix::Matrix, producer_benchmark::ProducerBenchmark};
 
 pub struct BenchmarkDriver {}
 
 impl BenchmarkDriver {
-    pub async fn run_benchmark(cmd: BenchmarkCmd) -> Result<()> {
-        match cmd {
-            BenchmarkCmd::Producer(config) => {
+    pub async fn run_benchmark(mode: BenchmarkMode) -> Result<()> {
+        match mode {
+            BenchmarkMode::Producer(config) => {
                 ProducerBenchmark::run_benchmark(config).await?;
             }
-            BenchmarkCmd::Consumer(_) => {
+            BenchmarkMode::Consumer(_) => {
                 println!("consume not implemented");
             }
-        }
+            BenchmarkMode::Matrix { config } => {
+                let matrix_config = if let Some(path) = config {
+                    let file = File::open(&path).expect("file not found");
+                    serde_yaml::from_reader::<_, Matrix>(file).expect("deserialization failed")
+                } else {
+                    crate::config::config_matrix::default_config()
+                };
+                let benchmarks_configs = matrix_config.generate_configs();
+                for benchmark_config in benchmarks_configs {
+                    println!("Running benchmark: {:#?}", benchmark_config);
+                    match benchmark_config {
+                        crate::config::BenchmarkConfig::Producer(producer) => {
+                            ProducerBenchmark::run_benchmark(producer).await?;
+                        }
+                        crate::config::BenchmarkConfig::Consumer(_) => {
+                            println!("consume not implemented");
+                        }
+                    }
 
+                    sleep(Duration::from_secs(1)).await;
+                }
+            }
+        }
         Ok(())
     }
 }

crates/fluvio-benchmark/src/cli.rs

@@ -2,14 +2,14 @@ use clap::Parser;
 use anyhow::Result;
 
 use crate::{
-    config::{ConsumerConfig, ProducerConfig},
     benchmark_driver::BenchmarkDriver,
+    config::{ConsumerConfig, ProducerConfig},
 };
 
 #[derive(Debug, Parser)]
 pub struct BenchmarkOpt {
     #[clap(subcommand)]
-    benchmark: BenchmarkCmd,
+    benchmark: Option<BenchmarkMode>,
 }
 impl BenchmarkOpt {
     pub async fn process(self) -> Result<()> {
@@ -19,16 +19,28 @@ impl BenchmarkOpt {
 }
 
 #[derive(Debug, Parser)]
-pub enum BenchmarkCmd {
+pub enum BenchmarkMode {
+    /// Use a matrix configuration file
+    Matrix {
+        /// Path to the configuration file
+        #[arg(short, long)]
+        config: Option<String>,
+    },
     /// Run a producer benchmark
     Producer(ProducerConfig),
     /// Run a consumer benchmark
     Consumer(ConsumerConfig),
 }
 
-pub async fn run_benchmarks(args: BenchmarkOpt) -> Result<()> {
+pub async fn run_benchmarks(opt: BenchmarkOpt) -> Result<()> {
     println!("# Fluvio Benchmark Results");
-    BenchmarkDriver::run_benchmark(args.benchmark).await?;
+
+    if let Some(mode) = opt.benchmark {
+        BenchmarkDriver::run_benchmark(mode).await?;
+    } else {
+        BenchmarkDriver::run_benchmark(BenchmarkMode::Matrix { config: None }).await?;
+    }
+
     println!();
     Ok(())
 }

crates/fluvio-benchmark/src/config/config_matrix.rs

@@ -0,0 +1,182 @@
+use std::time::Duration;
+
+use fluvio::Compression;
+use serde::{Deserialize, Serialize};
+use bytesize::ByteSize;
+
+use crate::config::{BenchmarkConfig, RecordKeyAllocationStrategy};
+
+use super::{cross::CrossIterate, default_topic_name, ProducerConfigBuilder};
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct Matrix {
+    pub producer_config: Option<ProducerMatrixConfig>,
+    pub consumer_config: Option<ConsumerMatrixConfig>,
+    pub shared_config: SharedMatrixConfig,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ProducerMatrixConfig {
+    pub batch_size: Vec<ByteSize>,
+    pub queue_size: Vec<u64>,
+    pub max_request_size: Vec<ByteSize>,
+    pub linger: Vec<Duration>,
+    pub server_timeout: Vec<Duration>,
+    pub compression: Vec<Compression>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ConsumerMatrixConfig {}
+
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct SharedMatrixConfig {
+    pub num_samples: Vec<usize>,
+    pub time_between_samples: Vec<Duration>,
+    pub worker_timeout: Vec<Duration>,
+    pub topic_config: FluvioTopicMatrixConfig,
+    pub load_config: BenchmarkLoadMatrixConfig,
+}
+
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct BenchmarkLoadMatrixConfig {
+    pub record_key_allocation_strategy: Vec<RecordKeyAllocationStrategy>,
+    pub num_producers: Vec<u64>,
+    pub num_records: Vec<u64>,
+    pub record_size: Vec<ByteSize>,
+}
+
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct FluvioTopicMatrixConfig {
+    pub partitions: Vec<u32>,
+    pub replicas: Vec<u32>,
+    pub topic_name: Vec<String>,
+    pub delete_topic: Vec<bool>,
+    pub ignore_rack: Vec<bool>,
+}
+
+impl Matrix {
+    pub fn generate_configs(&self) -> Vec<BenchmarkConfig> {
+        let builder: Vec<ProducerConfigBuilder> = vec![ProducerConfigBuilder::default()];
+
+        if let Some(producer_config) = &self.producer_config {
+            let producer_config = builder
+                .cross_iterate(&producer_config.batch_size, |v, b| {
+                    b.batch_size(v);
+                })
+                .cross_iterate(&producer_config.queue_size, |v, b| {
+                    b.queue_size(v);
+                })
+                .cross_iterate(&producer_config.max_request_size, |v, b| {
+                    b.max_request_size(v);
+                })
+                .cross_iterate(&producer_config.linger, |v, b| {
+                    b.linger(v);
+                })
+                .cross_iterate(&producer_config.server_timeout, |v, b| {
+                    b.server_timeout(v);
+                })
+                .cross_iterate(&producer_config.compression, |v, b| {
+                    b.compression(v);
+                })
+                .cross_iterate(&self.shared_config.num_samples, |v, b| {
+                    b.num_samples(v);
+                })
+                .cross_iterate(&self.shared_config.time_between_samples, |v, b| {
+                    b.time_between_samples(v);
+                })
+                .cross_iterate(&self.shared_config.worker_timeout, |v, b| {
+                    b.worker_timeout(v);
+                })
+                .cross_iterate(&self.shared_config.topic_config.partitions, |v, b| {
+                    b.partitions(v);
+                })
+                .cross_iterate(&self.shared_config.topic_config.replicas, |v, b| {
+                    b.replicas(v);
+                })
+                .cross_iterate(&self.shared_config.topic_config.topic_name, |v, b| {
+                    b.topic_name(v);
+                })
+                .cross_iterate(&self.shared_config.topic_config.delete_topic, |v, b| {
+                    b.delete_topic(v);
+                })
+                .cross_iterate(&self.shared_config.topic_config.ignore_rack, |v, b| {
+                    b.ignore_rack(v);
+                })
+                .cross_iterate(
+                    &self
+                        .shared_config
+                        .load_config
+                        .record_key_allocation_strategy,
+                    |v, b| {
+                        b.record_key_allocation_strategy(v);
+                    },
+                )
+                .cross_iterate(&self.shared_config.load_config.num_producers, |v, b| {
+                    b.num_producers(v);
+                })
+                .cross_iterate(&self.shared_config.load_config.num_records, |v, b| {
+                    b.num_records(v);
+                })
+                .cross_iterate(&self.shared_config.load_config.record_size, |v, b| {
+                    b.record_size(v);
+                })
+                .build();
+
+            return producer_config
+                .into_iter()
+                .map(BenchmarkConfig::Producer)
+                .collect();
+        }
+
+        if let Some(_consumer) = &self.consumer_config {
+            todo!("Consumer config not implemented");
+        }
+
+        panic!("No producer or consumer config provided");
+    }
+}
+
+pub fn default_config() -> Matrix {
+    Matrix {
+        producer_config: Some(ProducerMatrixConfig {
+            batch_size: vec![ByteSize::kib(16), ByteSize::mib(1)],
+            queue_size: vec![100],
+            max_request_size: vec![ByteSize::mb(32)],
+            linger: vec![Duration::from_millis(0)],
+            server_timeout: vec![Duration::from_secs(600)],
+            compression: vec![Compression::None],
+        }),
+        consumer_config: None,
+        shared_config: SharedMatrixConfig {
+            num_samples: vec![2],
+            time_between_samples: vec![Duration::from_secs(1)],
+            worker_timeout: vec![Duration::from_secs(60)],
+            topic_config: FluvioTopicMatrixConfig {
+                partitions: vec![1],
+                replicas: vec![1],
+                topic_name: vec![default_topic_name()],
+                delete_topic: vec![true],
+                ignore_rack: vec![true],
+            },
+            load_config: BenchmarkLoadMatrixConfig {
+                record_key_allocation_strategy: vec![RecordKeyAllocationStrategy::NoKey],
+                num_producers: vec![1],
+                num_records: vec![100, 10_000, 100_000],
+                record_size: vec![ByteSize::kib(5)],
+            },
+        },
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_default_config() {
+        let matrix = default_config();
+        let configs = matrix.generate_configs();
+
+        assert_eq!(configs.len(), 6);
+    }
+}

crates/fluvio-benchmark/src/config/cross.rs

@@ -0,0 +1,50 @@
+use super::{ConsumerConfig, ConsumerConfigBuilder, ProducerConfig, ProducerConfigBuilder};
+
+pub trait CrossIterate<C, B> {
+    fn cross_iterate<T: Clone, F: Fn(T, &mut B) + Copy>(self, values: &[T], f: F) -> Self;
+    fn build(self) -> Vec<C>;
+}
+
+impl CrossIterate<ProducerConfig, ProducerConfigBuilder> for Vec<ProducerConfigBuilder> {
+    fn cross_iterate<T: Clone, F: Fn(T, &mut ProducerConfigBuilder) + Copy>(
+        self,
+        values: &[T],
+        f: F,
+    ) -> Self {
+        self.into_iter()
+            .flat_map(|builder| {
+                values.iter().map(move |value| {
+                    let mut clone = builder.clone();
+                    f(value.clone(), &mut clone);
+                    clone
+                })
+            })
+            .collect()
+    }
+
+    fn build(self) -> Vec<ProducerConfig> {
+        self.into_iter().map(|x| x.build().unwrap()).collect()
+    }
+}
+
+impl CrossIterate<ConsumerConfig, ConsumerConfigBuilder> for Vec<ConsumerConfigBuilder> {
+    fn cross_iterate<T: Clone, F: Fn(T, &mut ConsumerConfigBuilder) + Copy>(
+        self,
+        values: &[T],
+        f: F,
+    ) -> Self {
+        self.into_iter()
+            .flat_map(|builder| {
+                values.iter().map(move |value| {
+                    let mut clone = builder.clone();
+                    f(value.clone(), &mut clone);
+                    clone
+                })
+            })
+            .collect()
+    }
+
+    fn build(self) -> Vec<ConsumerConfig> {
+        self.into_iter().map(|x| x.build().unwrap()).collect()
+    }
+}