Add benchmarks for kafka

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:?}"
+ );
+}