Merge branch 'main' into kafka_sink_single_remove_host_config

Cargo.toml

@@ -32,14 +32,14 @@ bytes = "1.0.0"
 tokio = { version = "1.25.0", features = ["full", "macros"] }
 tokio-util = { version = "0.7.7" }
 tokio-openssl = "0.6.2"
-itertools = "0.11.0"
+itertools = "0.12.0"
 openssl = { version = "0.10.36", features = ["vendored"] }
 anyhow = "1.0.42"
 serde = { version = "1.0.111", features = ["derive"] }
 serde_yaml = "0.9.17"
 uuid = { version = "1.0.0", features = ["serde", "v4"] }
 reqwest = "0.11.6"
-redis = { version = "0.23.0", features = ["tokio-comp", "cluster"] }
+redis = { version = "0.23.3", features = ["tokio-comp", "cluster"] }
 cdrs-tokio = "8.0"
 cassandra-protocol = "3.0"
 tracing = "0.1.15"
@@ -61,7 +61,8 @@ 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.3.0"
+aws-throwaway = "0.4.0"
 tokio-bin-process = "0.4.0"
 ordered-float = { version = "4.0.0", features = ["serde"] }
 hyper = { version = "0.14.14", features = ["server"] }
+shell-quote = "0.4.0"

ec2-cargo/src/main.rs

@@ -41,7 +41,7 @@ async fn main() {
         return;
     }
 
-    let aws = Aws::new(CleanupResources::AllResources).await;
+    let aws = Aws::builder(CleanupResources::AllResources).build().await;
     let instance_type = InstanceType::from(args.instance_type.as_str());
     let instance = aws
         .create_ec2_instance(Ec2InstanceDefinition::new(instance_type).volume_size_gigabytes(40))
@@ -196,7 +196,7 @@ cargo windsock {} 2>&1
 async fn rsync_push_shotover(state: &State) {
     let instance = &state.instance;
     let project_root_dir = &state.cargo_meta.workspace_root;
-    let address = instance.public_ip();
+    let address = instance.public_ip().unwrap().to_string();
 
     rsync(
         state,
@@ -213,7 +213,7 @@ async fn rsync_push_shotover(state: &State) {
 async fn rsync_fetch_windsock_results(state: &State) {
     let instance = &state.instance;
     let windsock_dir = &state.cargo_meta.target_directory.join("windsock_data");
-    let address = instance.public_ip();
+    let address = instance.public_ip().unwrap().to_string();
 
     rsync(
         state,

shotover-proxy/Cargo.toml

@@ -53,8 +53,8 @@ opensearch = "2.1.0"
 serde_json = "1.0.103"
 time = { version = "0.3.25" }
 inferno = { version = "0.11.15", default-features = false, features = ["multithreaded", "nameattr"] }
-shell-quote = "0.3.0"
 hyper.workspace = true
+shell-quote.workspace = true
 
 [features]
 # Include WIP alpha transforms in the public API

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

@@ -36,7 +36,9 @@ impl WindsockAws {
                 shotover_instance: RwLock::new(None),
                 bencher_instance: RwLock::new(None),
                 docker_instances: RwLock::new(vec![]),
-                aws: Aws::new(CleanupResources::WithAppTag(AWS_THROWAWAY_TAG.to_owned())).await,
+                aws: Aws::builder(CleanupResources::WithAppTag(AWS_THROWAWAY_TAG.to_owned()))
+                    .build()
+                    .await,
             }
         })
         .await
@@ -198,7 +200,7 @@ sudo docker system prune -af"#,
         let mut env_args = String::new();
         for (key, value) in envs {
             let key_value =
-                String::from_utf8(shell_quote::bash::escape(format!("{key}={value}"))).unwrap();
+                String::from_utf8(shell_quote::Bash::quote(&format!("{key}={value}"))).unwrap();
             env_args.push_str(&format!(" -e {key_value}"))
         }
         let output = self

shotover-proxy/tests/transforms/log_to_file.rs

@@ -15,15 +15,20 @@ async fn log_to_file() {
 
     assert_ok(redis::cmd("SET").arg("foo").arg(42), &mut connection).await;
     let request = std::fs::read("message-log/1/requests/message1.bin").unwrap();
-    assert_eq!(
-        request,
-        "*3\r\n$3\r\nSET\r\n$3\r\nfoo\r\n$2\r\n42\r\n".as_bytes()
-    );
+    assert_eq_string(&request, "*3\r\n$3\r\nSET\r\n$3\r\nfoo\r\n$2\r\n42\r\n");
 
     let response = std::fs::read("message-log/1/responses/message1.bin").unwrap();
-    assert_eq!(response, "+OK\r\n".as_bytes());
+    assert_eq_string(&response, "+OK\r\n");
 
     shotover.shutdown_and_then_consume_events(&[]).await;
 
     std::fs::remove_dir_all("message-log").unwrap();
 }
+
+/// Gives useful error message when both expected and actual data are valid utf8 strings
+fn assert_eq_string(actual_bytes: &[u8], expected_str: &str) {
+    match std::str::from_utf8(actual_bytes) {
+        Ok(actual) => assert_eq!(actual, expected_str),
+        Err(_) => assert_eq!(actual_bytes, expected_str.as_bytes()),
+    }
+}

shotover/Cargo.toml

@@ -15,7 +15,7 @@ alpha-transforms = []
 
 [dependencies]
 atomic_enum = "0.2.0"
-pretty-hex = "0.3.0"
+pretty-hex = "0.4.0"
 tokio-stream = "0.1.2"
 bytes-utils = "0.1.1"
 derivative = "2.1.1"
@@ -75,8 +75,8 @@ crc16 = "0.4.0"
 ordered-float.workspace = true
 
 #Crypto
-aws-config = "0.56.0"
-aws-sdk-kms = "0.33"
+aws-config = "1.0.0"
+aws-sdk-kms = "1.1.0"
 strum_macros = "0.25"
 chacha20poly1305 = { version = "0.10.0", features = ["std"] }
 generic-array = { version = "0.14", features = ["serde"] }

shotover/src/transforms/protect/key_management.rs

@@ -1,8 +1,8 @@
 use crate::transforms::protect::aws_kms::AWSKeyManagement;
 use crate::transforms::protect::local_kek::LocalKeyManagement;
 use anyhow::{anyhow, Result};
-use aws_config::meta::region::RegionProviderChain;
 use aws_config::SdkConfig;
+use aws_config::{meta::region::RegionProviderChain, BehaviorVersion};
 use aws_sdk_kms::config::Region;
 use aws_sdk_kms::Client as KmsClient;
 use base64::{engine::general_purpose, Engine as _};
@@ -37,16 +37,11 @@ pub enum KeyManagerConfig {
 }
 
 async fn config(region: String, endpoint: Option<String>) -> SdkConfig {
-    let region_provider = RegionProviderChain::first_try(Region::new(region));
+    let builder = aws_config::defaults(BehaviorVersion::v2023_11_09())
+        .region(RegionProviderChain::first_try(Region::new(region)));
     match endpoint {
-        Some(endpoint) => {
-            aws_config::from_env()
-                .region(region_provider)
-                .endpoint_url(endpoint)
-                .load()
-                .await
-        }
-        None => aws_config::from_env().region(region_provider).load().await,
+        Some(endpoint) => builder.endpoint_url(endpoint).load().await,
+        None => builder.load().await,
     }
 }
 

test-helpers/Cargo.toml

@@ -33,4 +33,4 @@ serde_yaml.workspace = true
 anyhow.workspace = true
 rcgen.workspace = true
 docker-compose-runner = "0.2.0"
-rdkafka = { version = "0.34", features = ["cmake-build"], optional = true }
+rdkafka = { version = "0.36", features = ["cmake-build"], optional = true }

windsock-cloud-docker/Cargo.toml

@@ -7,6 +7,6 @@ license = "Apache-2.0"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-shell-quote = "0.3.0"
+shell-quote.workspace = true
 tokio.workspace = true
 subprocess.workspace = true

windsock-cloud-docker/src/container.rs

@@ -65,7 +65,7 @@ impl Container {
                 if x.is_empty() {
                     String::from("''")
                 } else {
-                    String::from_utf8(shell_quote::bash::escape(x)).unwrap()
+                    String::from_utf8(shell_quote::Bash::quote(&x)).unwrap()
                 }
             })
             .collect();