Merge branch 'main' into docker-compose-update

.github/workflows/windsock_benches.yaml

@@ -26,7 +26,7 @@ jobs:
  run: |
  cargo windsock --bench-length-seconds 5 --operations-per-second 100
  cargo windsock --bench-length-seconds 5 --operations-per-second 100 --profilers flamegraph --name cassandra,compression=none,driver=scylla,operation=read_i64,protocol=v4,shotover=standard,topology=single
- cargo windsock --bench-length-seconds 5 --operations-per-second 100 --profilers flamegraph --name kafka,shotover=standard,size=1B,topology=single
+ cargo windsock --bench-length-seconds 5 --operations-per-second 100 --profilers sys_monitor --name kafka,shotover=standard,size=1B,topology=single
 
  # windsock/examples/cassandra.rs - this can stay here until windsock is moved to its own repo
  cargo run --release --example cassandra -- --bench-length-seconds 5 --operations-per-second 100

Cargo.toml

@@ -52,4 +52,4 @@ rand_distr = "0.4.1"
 clap = { version = "4.0.4", features = ["cargo", "derive"] }
 async-trait = "0.1.30"
 typetag = "0.2.5"
-aws-throwaway = "0.1.0"
+aws-throwaway = "0.1.1"

changelog.md

@@ -3,6 +3,17 @@
 Any breaking changes to the `topology.yaml` or `shotover` rust API should be documented here.
 This assists us in knowing when to make the next release a breaking release and assists users with making upgrades to new breaking releases.
 
+## 0.1.11
+
+### topology.yaml
+
+* No recorded changes
+
+### shotover rust api
+
+* `shotover::message_value` is now `shotover::frame::value`
+* `shotover::message_value::MessageValue` is now `shotover::frame::value::GenericValue`
+
 ## 0.1.10
 
 ### topology.yaml

ec2-cargo/Cargo.toml

@@ -15,3 +15,4 @@ aws-throwaway.workspace = true
 tracing-appender.workspace = true
 shellfish = { version = "0.8.0", features = ["async"] }
 rustyline = "11.0.0"
+cargo_metadata = "0.15.4"

ec2-cargo/src/main.rs

@@ -1,4 +1,5 @@
 use aws_throwaway::{ec2_instance::Ec2Instance, Aws, InstanceType};
+use cargo_metadata::{Metadata, MetadataCommand};
 use clap::Parser;
 use rustyline::DefaultEditor;
 use shellfish::{async_fn, handler::DefaultAsyncHandler, Command, Shell};
@@ -7,7 +8,7 @@ use tracing_subscriber::EnvFilter;
 
 /// Spins up an EC2 instance and then presents a shell from which you can run `cargo test` on the ec2 instance.
 ///
-/// TODO: Every time the shell runs a cargo command, any local changes to the repo are reuploaded to the ec2 instance.
+/// Every time the shell runs a cargo command, any local changes to the repo are reuploaded to the ec2 instance.
 ///
 /// When the shell is exited all created EC2 instances are destroyed.
 #[derive(Parser, Clone)]
@@ -30,6 +31,7 @@ async fn main() {
  .with_writer(non_blocking)
  .init();
 
+ let cargo_meta = MetadataCommand::new().exec().unwrap();
  let args = Args::parse();
  if args.cleanup {
  Aws::cleanup_resources_static().await;
@@ -74,8 +76,16 @@ docker compose "$@"
 ' | sudo dd of=/bin/docker-compose
 sudo chmod +x /bin/docker-compose
 
-git clone https://github.com/shotover/shotover-proxy
 echo "export RUST_BACKTRACE=1" >> .profile
+echo "export CARGO_TERM_COLOR=always" >> .profile
+echo 'source "$HOME/.cargo/env"' >> .profile
+
+source .profile
+if [ "$(uname -m)" = "aarch64" ]; then
+ curl -LsSf https://get.nexte.st/latest/linux-arm | tar zxf - -C ${CARGO_HOME:-~/.cargo}/bin
+else
+ curl -LsSf https://get.nexte.st/latest/linux | tar zxf - -C ${CARGO_HOME:-~/.cargo}/bin
+fi
 "#).await;
  while let Some(line) = receiver.recv().await {
  println!("{}", line)
@@ -84,7 +94,10 @@ echo "export RUST_BACKTRACE=1" >> .profile
  println!("Finished creating instance.");
 
  let mut shell = Shell::new_with_async_handler(
- State { instance },
+ State {
+ cargo_meta,
+ instance,
+ },
  "ec2-cargo$ ",
  DefaultAsyncHandler::default(),
  DefaultEditor::new().unwrap(),
@@ -111,15 +124,17 @@ echo "export RUST_BACKTRACE=1" >> .profile
 }
 
 async fn test(state: &mut State, mut args: Vec<String>) -> Result<(), Box<dyn Error>> {
+ rsync_shotover(state).await;
  args.remove(0);
  let args = args.join(" ");
  let mut receiver = state
  .instance
  .ssh()
  .shell_stdout_lines(&format!(
  r#"
-cd shotover-proxy
-RUST_BACKTRACE=1 ~/.cargo/bin/cargo test --color always {} 2>&1
+source .profile
+cd shotover
+cargo nextest run {} 2>&1
 "#,
  args
  ))
@@ -131,6 +146,57 @@ RUST_BACKTRACE=1 ~/.cargo/bin/cargo test --color always {} 2>&1
  Ok(())
 }
 
+async fn rsync_shotover(state: &State) {
+ let instance = &state.instance;
+ let target_dir = &state.cargo_meta.target_directory;
+ let project_root_dir = &state.cargo_meta.workspace_root;
+
+ let key_path = target_dir.join("ec2-cargo-privatekey");
+ tokio::fs::remove_file(&key_path).await.ok();
+ tokio::fs::write(&key_path, instance.client_private_key())
+ .await
+ .unwrap();
+ let output = tokio::process::Command::new("chmod")
+ .args(&["400".to_owned(), format!("{}", key_path)])
+ .output()
+ .await
+ .unwrap();
+ if !output.status.success() {
+ let stdout = String::from_utf8(output.stdout).unwrap();
+ let stderr = String::from_utf8(output.stderr).unwrap();
+ panic!("chmod failed:\nstdout:\n{stdout}\nstderr:\n{stderr}")
+ }
+
+ let known_hosts_path = target_dir.join("ec2-cargo-known_hosts");
+ tokio::fs::write(&known_hosts_path, instance.openssh_known_hosts_line())
+ .await
+ .unwrap();
+
+ let address = instance.public_ip();
+ let output = tokio::process::Command::new("rsync")
+ .args(&[
+ "--delete".to_owned(),
+ "--exclude".to_owned(),
+ "target".to_owned(),
+ "-e".to_owned(),
+ format!(
+ "ssh -i {} -o 'UserKnownHostsFile {}'",
+ key_path, known_hosts_path
+ ),
+ "-ra".to_owned(),
+ format!("{}/", project_root_dir), // trailing slash means copy the contents of the directory instead of the directory itself
+ format!("ubuntu@{address}:/home/ubuntu/shotover"),
+ ])
+ .output()
+ .await
+ .unwrap();
+ if !output.status.success() {
+ let stdout = String::from_utf8(output.stdout).unwrap();
+ let stderr = String::from_utf8(output.stderr).unwrap();
+ panic!("rsync failed:\nstdout:\n{stdout}\nstderr:\n{stderr}")
+ }
+}
+
 fn ssh_instructions(state: &mut State, mut _args: Vec<String>) -> Result<(), Box<dyn Error>> {
  println!(
  "Run the following to ssh into the EC2 instance:\n{}",
@@ -141,5 +207,6 @@ fn ssh_instructions(state: &mut State, mut _args: Vec<String>) -> Result<(), Box
 }
 
 struct State {
+ cargo_meta: Metadata,
  instance: Ec2Instance,
 }

shotover-proxy/Cargo.toml

@@ -16,7 +16,6 @@ anyhow.workspace = true
 tokio.workspace = true
 tracing.workspace = true
 clap.workspace = true
-criterion = { version = "0.5.0", features = ["async_tokio"] }
 rstest = "0.18.0"
 cassandra-cpp = { version = "2.0.0" }
 test-helpers = { path = "../test-helpers" }
@@ -53,10 +52,7 @@ regex = "1.7.0"
 cql-ws = { git = "https://github.com/conorbros/cql-ws" }
 opensearch = "2.1.0"
 serde_json = "1.0.103"
-
-[[bench]]
-name = "benches"
-harness = false
+time = { version = "0.3.25" }
 
 [features]
 # Include WIP alpha transforms in the public API

shotover-proxy/examples/windsock/aws/mod.rs

@@ -52,6 +52,15 @@ impl WindsockAws {
  .create_ec2_instance(InstanceType::M6aLarge, 8)
  .await,
  });
+ instance
+ .instance
+ .ssh()
+ .shell(
+ r#"
+sudo apt-get update
+sudo apt-get install -y sysstat"#,
+ )
+ .await;
  instance
  .instance
  .ssh()
@@ -85,6 +94,7 @@ until sudo apt-get update -qq
 do
  sleep 1
 done
+sudo apt-get install -y sysstat
 curl -sSL https://get.docker.com/ | sudo sh"#,
  )
  .await;
@@ -108,6 +118,15 @@ curl -sSL https://get.docker.com/ | sudo sh"#,
  .create_ec2_instance(InstanceType::M6aLarge, 8)
  .await,
  });
+ instance
+ .instance
+ .ssh()
+ .shell(
+ r#"
+sudo apt-get update
+sudo apt-get install -y sysstat"#,
+ )
+ .await;
 
  // PROFILE is set in build.rs from PROFILE listed in https://doc.rust-lang.org/cargo/reference/environment-variables.html#environment-variables-cargo-sets-for-build-scripts
  let profile = if env!("PROFILE") == "release" {

shotover-proxy/examples/windsock/cassandra.rs

@@ -1,10 +1,11 @@
 use crate::{
  aws::{Ec2InstanceWithDocker, Ec2InstanceWithShotover, RunningShotover},
  common::{rewritten_file, Shotover},
- profilers::ProfilerRunner,
+ profilers::{self, CloudProfilerRunner, ProfilerRunner},
 };
 use anyhow::Result;
 use async_trait::async_trait;
+use aws_throwaway::ec2_instance::Ec2Instance;
 use cdrs_tokio::{
  cluster::{
  session::{
@@ -419,7 +420,7 @@ impl Bench for CassandraBench {
  }
 
  fn supported_profilers(&self) -> Vec<String> {
- ProfilerRunner::supported_profilers(self.shotover)
+ profilers::supported_profilers(self.shotover)
  }
 
  fn cores_required(&self) -> usize {
@@ -429,7 +430,7 @@ impl Bench for CassandraBench {
  async fn orchestrate_cloud(
  &self,
  _running_in_release: bool,
- _profiling: Profiling,
+ profiling: Profiling,
  parameters: BenchParameters,
  ) -> Result<()> {
  let aws = crate::aws::WindsockAws::get().await;
@@ -451,6 +452,24 @@ impl Bench for CassandraBench {
  let cassandra_ip = cassandra_instance1.instance.private_ip().to_string();
  let shotover_ip = shotover_instance.instance.private_ip().to_string();
 
+ let mut profiler_instances: HashMap<String, &Ec2Instance> =
+ [("bencher".to_owned(), &bench_instance.instance)].into();
+ if let Shotover::ForcedMessageParsed | Shotover::Standard = self.shotover {
+ profiler_instances.insert("shotover".to_owned(), &shotover_instance.instance);
+ }
+ match self.topology {
+ Topology::Cluster3 => {
+ profiler_instances.insert("cassandra1".to_owned(), &cassandra_instance1.instance);
+ profiler_instances.insert("cassandra2".to_owned(), &cassandra_instance2.instance);
+ profiler_instances.insert("cassandra3".to_owned(), &cassandra_instance3.instance);
+ }
+ Topology::Single => {
+ profiler_instances.insert("cassandra".to_owned(), &cassandra_instance1.instance);
+ }
+ }
+ let mut profiler =
+ CloudProfilerRunner::new(self.name(), profiling, profiler_instances).await;
+
  let cassandra_nodes = vec![
  AwsNodeInfo {
  instance: cassandra_instance1.clone(),
@@ -487,6 +506,8 @@ impl Bench for CassandraBench {
  .run_bencher(&self.run_args(&destination, &parameters), &self.name())
  .await;
 
+ profiler.finish();
+
  if let Some(running_shotover) = running_shotover {
  running_shotover.shutdown().await;
  }
@@ -524,7 +545,7 @@ impl Bench for CassandraBench {
  panic!("Mocked cassandra database does not provide a clustered mode")
  }
  };
- let mut profiler = ProfilerRunner::new(profiling);
+ let mut profiler = ProfilerRunner::new(self.name(), profiling);
  let shotover = match self.shotover {
  Shotover::Standard => Some(
  ShotoverProcessBuilder::new_with_topology(&format!("{config_dir}/topology.yaml"))

shotover-proxy/examples/windsock/common.rs

@@ -1,4 +1,4 @@
-use anyhow::anyhow;
+use anyhow::Context;
 use std::path::Path;
 
 #[derive(Clone, Copy)]
@@ -24,7 +24,7 @@ impl Shotover {
 pub async fn rewritten_file(path: &Path, find_replace: &[(&str, &str)]) -> String {
  let mut text = tokio::fs::read_to_string(path)
  .await
- .map_err(|e| anyhow!(e).context(format!("Failed to read from {path:?}")))
+ .with_context(|| format!("Failed to read from {path:?}"))
  .unwrap();
  for (find, replace) in find_replace {
  text = text.replace(find, replace);