feat(bench): enhance bench suite with rate limiting and CPU identific…

…ation

This commit introduces several improvements to the benchmark suite:
- Added rate limiting functionality to the consumer and producer benchmarks.
- Updated the benchmark scripts to include new test scenarios with rate limits.
- Enhanced the output directory naming by appending the CPU name in lowercase.
- Updated dependencies in `Cargo.lock` and `Cargo.toml` to support new features.
- Modified chart creation functions to allow optional title and subtext stripping.
- Refactored scripts to replace large batch tests with normal batch tests.

.github/workflows/performance.yml

@@ -35,7 +35,7 @@ jobs:
 
       - name: Run predefined benchmarks
         timeout-minutes: 60
-        run: ./scripts/performance/run-standard-performance-suite.sh
+        run: ./scripts/performance/run-standard-performance-suite.sh --identifier iggy-ci
 
       - name: Upload benchmark results
         uses: actions/upload-artifact@v4

bench/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "bench"
-version = "0.2.0"
+version = "0.2.1"
 edition = "2021"
 license = "Apache-2.0"
 # Due to dependency to integration, which has a dependency to server, setting
@@ -22,6 +22,7 @@ iggy-bench-report = { path = "report" }
 integration = { path = "../integration" }
 nonzero_lit = "0.1.2"
 serde = { version = "1.0.217", features = ["derive"] }
+sysinfo = "0.33.1"
 tokio = { version = "1.43.0", features = ["full"] }
 toml = "0.8.20"
 tracing = { version = "0.1.41" }

bench/report/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "iggy-bench-report"
-version = "0.2.0"
+version = "0.2.1"
 edition = "2021"
 description = "Benchmark report and chart generation library for iggy-bench binary and iggy-benchmarks-dashboard web app"
 license = "Apache-2.0"

bench/report/src/lib.rs

@@ -13,10 +13,14 @@ use plotting::chart_kind::ChartKind;
 
 pub use types::*;
 
-pub fn create_throughput_chart(report: &BenchmarkReport, dark: bool) -> Chart {
+pub fn create_throughput_chart(
+    report: &BenchmarkReport,
+    dark: bool,
+    strip_title_and_subtext: bool,
+) -> Chart {
     let title = report.title(ChartKind::Throughput);
 
-    let mut chart = IggyChart::new(&title, &report.subtext(), dark)
+    let mut chart = IggyChart::new(&title, &report.subtext(), dark, strip_title_and_subtext)
         .with_time_x_axis()
         .with_dual_y_axis("Throughput [MB/s]", "Throughput [msg/s]");
 
@@ -74,10 +78,14 @@ pub fn create_throughput_chart(report: &BenchmarkReport, dark: bool) -> Chart {
     chart.inner
 }
 
-pub fn create_latency_chart(report: &BenchmarkReport, dark: bool) -> Chart {
+pub fn create_latency_chart(
+    report: &BenchmarkReport,
+    dark: bool,
+    strip_title_and_subtext: bool,
+) -> Chart {
     let title = report.title(ChartKind::Latency);
 
-    let mut chart = IggyChart::new(&title, &report.subtext(), dark)
+    let mut chart = IggyChart::new(&title, &report.subtext(), dark, strip_title_and_subtext)
         .with_time_x_axis()
         .with_y_axis("Latency [ms]");
 

bench/report/src/plotting/chart.rs

@@ -19,9 +19,10 @@ const AXIS_TEXT_SIZE: u32 = 16;
 
 impl IggyChart {
     /// Create a new `IggyChart` with default tooltip, legend, grid, and toolbox.
-    pub fn new(title: &str, subtext: &str, dark: bool) -> Self {
-        let mut chart = Chart::new()
-            .title(
+    pub fn new(title: &str, subtext: &str, dark: bool, strip_title_and_subtext: bool) -> Self {
+        let chart = Chart::new();
+        let chart = if !strip_title_and_subtext {
+            chart.title(
                 Title::new()
                     .text(title)
                     .text_align(TextAlign::Center)
@@ -31,6 +32,16 @@ impl IggyChart {
                     .left("50%")
                     .top("1%"),
             )
+        } else {
+            chart
+        };
+        let grid_top = if !strip_title_and_subtext {
+            "16%"
+        } else {
+            "4%"
+        };
+
+        let chart = chart
             .tooltip(Tooltip::new().axis_pointer(AxisPointer::new().type_(AxisPointerType::Cross)))
             .legend(
                 Legend::new()
@@ -46,7 +57,13 @@ impl IggyChart {
                     .item_height(14)
                     .type_(LegendType::Scroll),
             )
-            .grid(Grid::new().left("5%").right("19%").top("16%").bottom("8%"))
+            .grid(
+                Grid::new()
+                    .left("5%")
+                    .right("20%")
+                    .top(grid_top)
+                    .bottom("8%"),
+            )
             .data_zoom(
                 DataZoom::new()
                     .show(true)
@@ -65,9 +82,11 @@ impl IggyChart {
                 ),
             );
 
-        if dark {
-            chart = chart.background_color("#242424");
-        }
+        let chart = if dark {
+            chart.background_color("#242424")
+        } else {
+            chart
+        };
 
         Self { inner: chart }
     }

bench/src/actors/consumer.rs

@@ -1,6 +1,7 @@
 use super::utils::calculate_latency_from_first_message;
 use crate::analytics::metrics::individual::from_records;
 use crate::analytics::record::BenchmarkRecord;
+use crate::rate_limiter::RateLimiter;
 use human_repr::HumanCount;
 use iggy::client::{ConsumerGroupClient, MessageClient};
 use iggy::clients::client::IggyClient;
@@ -34,6 +35,7 @@ pub struct Consumer {
     moving_average_window: u32,
     polling_kind: PollingKind,
     calculate_latency_from_message_payload: bool,
+    rate_limiter: Option<RateLimiter>,
 }
 
 impl Consumer {
@@ -52,6 +54,7 @@ impl Consumer {
         moving_average_window: u32,
         polling_kind: PollingKind,
         calculate_latency_from_message_payload: bool,
+        rate_limiter: Option<RateLimiter>,
     ) -> Self {
         Self {
             client_factory,
@@ -67,6 +70,7 @@ impl Consumer {
             moving_average_window,
             polling_kind,
             calculate_latency_from_message_payload,
+            rate_limiter,
         }
     }
 
@@ -184,6 +188,9 @@ impl Consumer {
         let mut records = Vec::with_capacity(message_batches as usize);
         let start_timestamp = Instant::now();
         while self.batches_left_to_receive.load(Ordering::Acquire) > 0 {
+            if let Some(limiter) = &self.rate_limiter {
+                limiter.throttle(batch_size_total_bytes).await;
+            }
             let offset = current_iteration * messages_per_batch as u64;
 
             let (strategy, auto_commit) = match self.polling_kind {

bench/src/benchmarks/consumer_benchmark.rs

@@ -1,6 +1,7 @@
 use crate::actors::consumer::Consumer;
 use crate::args::common::IggyBenchArgs;
 use crate::benchmarks::benchmark::{BenchmarkFutures, Benchmarkable};
+use crate::rate_limiter::RateLimiter;
 use async_trait::async_trait;
 use iggy::messages::poll_messages::PollingKind;
 use iggy_bench_report::benchmark_kind::BenchmarkKind;
@@ -72,6 +73,8 @@ impl Benchmarkable for ConsumerBenchmark {
                 args.moving_average_window(),
                 polling_kind,
                 false, // TODO: Calculate latency from timestamp in first message, it should be an argument to iggy-bench
+                args.rate_limit()
+                    .map(|rl| RateLimiter::new(rl.as_bytes_u64())),
             );
 
             let future = Box::pin(async move { consumer.run().await });

bench/src/benchmarks/consumer_group_benchmark.rs

@@ -3,6 +3,7 @@ use crate::{
     actors::consumer::Consumer,
     args::common::IggyBenchArgs,
     benchmarks::{CONSUMER_GROUP_BASE_ID, CONSUMER_GROUP_NAME_PREFIX},
+    rate_limiter::RateLimiter,
 };
 use async_trait::async_trait;
 use iggy::{
@@ -105,6 +106,9 @@ impl Benchmarkable for ConsumerGroupBenchmark {
                 self.args.moving_average_window(),
                 polling_kind,
                 false, // TODO: Calculate latency from timestamp in first message, it should be an argument to iggy-bench
+                self.args
+                    .rate_limit()
+                    .map(|rl| RateLimiter::new(rl.as_bytes_u64())),
             );
             let future = Box::pin(async move { consumer.run().await });
             futures.as_mut().unwrap().push(future);

bench/src/benchmarks/producer_and_consumer_benchmark.rs

@@ -81,6 +81,9 @@ impl Benchmarkable for ProducerAndConsumerBenchmark {
                 self.args.moving_average_window(),
                 polling_kind,
                 false, // TODO: Calculate latency from timestamp in first message, it should be an argument to iggy-bench
+                self.args
+                    .rate_limit()
+                    .map(|rl| RateLimiter::new(rl.as_bytes_u64())),
             );
             let future = Box::pin(async move { consumer.run().await });
             futures.as_mut().unwrap().push(future);

bench/src/benchmarks/producer_and_consumer_group_benchmark.rs

@@ -146,6 +146,9 @@ impl Benchmarkable for ProducerAndConsumerGroupBenchmark {
                 self.args.moving_average_window(),
                 polling_kind,
                 false, // TODO: Calculate latency from timestamp in first message, it should be an argument to iggy-bench
+                self.args
+                    .rate_limit()
+                    .map(|rl| RateLimiter::new(rl.as_bytes_u64())),
             );
             let future = Box::pin(async move { consumer.run().await });
             futures.as_mut().unwrap().push(future);

bench/src/main.rs

@@ -15,6 +15,7 @@ use std::fs;
 use std::path::Path;
 use tracing::{error, info};
 use tracing_subscriber::{fmt, layer::SubscriberExt, util::SubscriberInitExt, EnvFilter};
+use utils::cpu_name::append_cpu_name_lowercase;
 
 #[tokio::main]
 async fn main() -> Result<(), IggyError> {
@@ -32,7 +33,9 @@ async fn main() -> Result<(), IggyError> {
         if !dir_path.exists() {
             fs::create_dir_all(dir_path).unwrap();
         }
-        dir_path.join(args.generate_dir_name())
+        let mut dir_name = args.generate_dir_name();
+        append_cpu_name_lowercase(&mut dir_name);
+        dir_path.join(dir_name)
     });
 
     // Configure logging

bench/src/plot.rs

@@ -20,7 +20,7 @@ impl ChartType {
         }
     }
 
-    fn create_chart(&self) -> fn(&BenchmarkReport, bool) -> Chart {
+    fn create_chart(&self) -> fn(&BenchmarkReport, bool, bool) -> Chart {
         match self {
             ChartType::Throughput => iggy_bench_report::create_throughput_chart,
             ChartType::Latency => iggy_bench_report::create_latency_chart,
@@ -71,7 +71,7 @@ pub fn plot_chart(
     should_open_in_browser: bool,
 ) -> std::io::Result<()> {
     let data_processing_start = Instant::now();
-    let chart = (chart_type.create_chart())(report, true); // Use dark theme by default
+    let chart = (chart_type.create_chart())(report, true, false); // Use dark theme by default
     let data_processing_time = data_processing_start.elapsed();
 
     let chart_render_start = Instant::now();

bench/src/runner.rs

@@ -3,6 +3,7 @@ use crate::args::common::IggyBenchArgs;
 use crate::benchmarks::benchmark::Benchmarkable;
 use crate::plot::{plot_chart, ChartType};
 use crate::utils::collect_server_logs_and_save_to_file;
+use crate::utils::cpu_name::append_cpu_name_lowercase;
 use crate::utils::server_starter::start_server_if_needed;
 use futures::future::select_all;
 use iggy::error::IggyError;
@@ -74,9 +75,10 @@ impl BenchmarkRunner {
 
         if let Some(output_dir) = benchmark.args().output_dir() {
             // Generate the full output path using the directory name generator
-            let dir_name = benchmark.args().generate_dir_name();
+            let mut dir_name = benchmark.args().generate_dir_name();
+            append_cpu_name_lowercase(&mut dir_name);
             let full_output_path = Path::new(&output_dir)
-                .join(dir_name)
+                .join(dir_name.clone())
                 .to_string_lossy()
                 .to_string();
 

bench/src/utils/cpu_name.rs

@@ -0,0 +1,18 @@
+use sysinfo::System;
+
+pub fn append_cpu_name_lowercase(to: &mut String) {
+    let mut sys = System::new();
+    sys.refresh_all();
+
+    let cpu = sys
+        .cpus()
+        .first()
+        .map(|cpu| (cpu.brand().to_string(), cpu.frequency()))
+        .unwrap_or_else(|| (String::from("unknown"), 0))
+        .0
+        .to_lowercase()
+        .replace(' ', "_");
+
+    to.push('_');
+    to.push_str(&cpu);
+}

bench/src/utils/mod.rs

@@ -11,6 +11,7 @@ use std::{fs, path::Path};
 use tracing::error;
 
 pub mod client_factory;
+pub mod cpu_name;
 pub mod server_starter;
 
 pub async fn get_server_stats(