Add benchmarks for kafka (#1085)

shotover-proxy/examples/kafka_bench.rs

@@ -0,0 +1,27 @@
+use test_helpers::docker_compose::DockerCompose;
+use test_helpers::kafka_producer_perf_test::run_producer_bench;
+use test_helpers::shotover_process::ShotoverProcessBuilder;
+
+#[tokio::main]
+async fn main() {
+    test_helpers::bench::init();
+
+    let config_dir = "tests/test-configs/kafka/passthrough";
+    {
+        let _compose = DockerCompose::new(&format!("{}/docker-compose.yaml", config_dir));
+        let shotover =
+            ShotoverProcessBuilder::new_with_topology(&format!("{}/topology.yaml", config_dir))
+                .start()
+                .await;
+
+        println!("Benching Shotover ...");
+        run_producer_bench("[localhost]:9192");
+
+        shotover.shutdown_and_then_consume_events(&[]).await;
+    }
+
+    // restart the docker container to avoid running out of disk space
+    let _compose = DockerCompose::new(&format!("{}/docker-compose.yaml", config_dir));
+    println!("\nBenching Direct Kafka ...");
+    run_producer_bench("[localhost]:9092");
+}

shotover-proxy/examples/kafka_flamegraph.rs

@@ -0,0 +1,32 @@
+use test_helpers::docker_compose::DockerCompose;
+use test_helpers::flamegraph::Perf;
+use test_helpers::kafka_producer_perf_test::run_producer_bench;
+use test_helpers::shotover_process::ShotoverProcessBuilder;
+
+// To get useful results you will need to modify the Cargo.toml like:
+// [profile.release]
+// #lto = "fat"
+// codegen-units = 1
+// debug = true
+
+#[tokio::main]
+async fn main() {
+    test_helpers::bench::init();
+    let config_dir = "tests/test-configs/kafka/passthrough";
+    {
+        let _compose = DockerCompose::new(&format!("{}/docker-compose.yaml", config_dir));
+
+        let shotover =
+            ShotoverProcessBuilder::new_with_topology(&format!("{}/topology.yaml", config_dir))
+                .start()
+                .await;
+
+        let perf = Perf::new(shotover.child.as_ref().unwrap().id().unwrap());
+
+        println!("Benching Shotover ...");
+        run_producer_bench("[localhost]:9192");
+
+        shotover.shutdown_and_then_consume_events(&[]).await;
+        perf.flamegraph();
+    }
+}

test-helpers/src/kafka_producer_perf_test.rs

@@ -0,0 +1,19 @@
+use crate::run_command_to_stdout;
+
+pub fn run_producer_bench(address_bench: &str) {
+    run_command_to_stdout(
+        "kafka-producer-perf-test.sh",
+        &[
+            "--producer-props",
+            &format!("bootstrap.servers={address_bench}"),
+            "--record-size",
+            "1000",
+            "--throughput",
+            "-1",
+            "--num-records",
+            "5000000",
+            "--topic",
+            "foo",
+        ],
+    );
+}

test-helpers/src/latte.rs

@@ -1,3 +1,5 @@
+use crate::run_command_to_stdout;
+
 // TODO: Shelling out directly like this is just for experimenting.
 // Either:
 // * get access to latte as a crate
@@ -85,15 +87,3 @@ impl Latte {
         run_command_to_stdout("latte", &["show", file_b, "-b", file_a]);
     }
 }
-
-/// unlike crate::docker_compose::run_command stdout of the command is sent to the stdout of the application
-fn run_command_to_stdout(command: &str, args: &[&str]) {
-    assert!(
-        std::process::Command::new(command)
-            .args(args)
-            .status()
-            .unwrap()
-            .success(),
-        "Failed to run: {command} {args:?}"
-    );
-}

test-helpers/src/lib.rs

@@ -3,6 +3,7 @@ pub mod cert;
 pub mod connection;
 pub mod docker_compose;
 pub mod flamegraph;
+pub mod kafka_producer_perf_test;
 pub mod latte;
 pub mod lazy;
 pub mod metrics;
@@ -26,3 +27,14 @@ pub fn try_wait_for_socket_to_open(address: &str, port: u16) -> Result<()> {
     }
     Ok(())
 }
+
+fn run_command_to_stdout(command: &str, args: &[&str]) {
+    assert!(
+        std::process::Command::new(command)
+            .args(args)
+            .status()
+            .unwrap()
+            .success(),
+        "Failed to run: {command} {args:?}"
+    );
+}