Add benchmarks for kafka

.github/workflows/build_and_test.yaml

@@ -76,6 +76,8 @@ jobs:
  cargo run --release --example cassandra_cluster_bench
  cargo run --release --example cassandra_cluster_flamegraph
  cargo run --release --example cassandra_bench_mocked
+ cargo run --release --example kafka_bench
+ cargo run --release --example kafka_flamegraph
  if: ${{ matrix.name == 'Ubuntu 20.04 - Release' }}
  - name: Ensure that tests did not create or modify any files that arent .gitignore'd
  run: |

shotover-proxy/build/install_ubuntu_packages.sh

@@ -10,6 +10,9 @@ sudo apt-get install -y gcc-aarch64-linux-gnu
 # Install dependencies of the cpp-driver even if they are already on CI so that we can run this locally
 sudo apt-get install -y libuv1 libuv1-dev cmake g++ libssl-dev zlib1g-dev
 
+# Install kafka to get access to kafka-producer-perf-test.sh and kafka-consumer-perf-test.sh
+sudo apt-get install -y kafka
+
 # set VERSION to one of the tags here: https://github.com/datastax/cpp-driver/tags
 VERSION=2.16.2
 

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