AWS crate replacement enhancements

Author: DJAndries

.env.example

@@ -1,5 +1,6 @@
 AWS_ACCESS_KEY_ID=test
 AWS_SECRET_ACCESS_KEY=test
+AWS_REGION=us-west-2
 S3_ENDPOINT=http://localhost:4566
 KAFKA_BROKERS=127.0.0.1:9092
 DATABASE_URL=postgres://star:password@localhost:5642

Cargo.lock

@@ -30,8 +30,8 @@ calendar-duration = "1.0"
 time = { version = "0.3", features = ["parsing", "serde"] }
 tokio-util = "0.7"
 reqwest = { version = "0.11", features = ["json"] }
-aws-config = "1.6"
-aws-sdk-s3 = { version = "1.83", features = ["behavior-version-latest"] }
+aws-config = { version = "1.6", features = ["behavior-version-latest"] }
+aws-sdk-s3 = { version = "1.85", features = ["behavior-version-latest"] }
 prometheus-client = "0.22"
 sentry = "0.36"
 jemallocator = "0.5"

Cargo.toml

@@ -5,7 +5,7 @@ networks:
 
 services:
   localstack:
-    image: localstack/localstack@sha256:41fe0bd366e68ad9c29d75b1ed2bc4a0888b0295fee5e4dff366ff475e57c4d8
+    image: localstack/localstack@sha256:b1b9f264d6c63fcacc7009a0924fd444f20d2149e9874500683d6c37e980dd60
     ports:
       - 4510-4559:4510-4559
       - 4566:4566
@@ -14,9 +14,8 @@ services:
       - DOCKER_HOST=unix:///var/run/docker.sock
       - LOCALSTACK_SERVICES=s3
     volumes:
-      - "/tmp/localstack:/tmp/localstack"
       - "/var/run/docker.sock:/var/run/docker.sock"
-      - "./misc/create_localstack_bucket.sh:/docker-entrypoint-initaws.d/create_bucket.sh"
+      - "./misc/create_localstack_bucket.sh:/etc/localstack/init/ready.d/create_bucket.sh"
     networks:
       - star-agg
   db:

docker-compose.yml

@@ -1,5 +1,5 @@
 use aws_sdk_s3::{
-  config::{http::HttpResponse, Region},
+  config::http::HttpResponse,
   error::SdkError,
   operation::put_object::PutObjectError,
   primitives::ByteStream,
@@ -27,19 +27,15 @@ pub struct DataLake {
 
 impl DataLake {
   pub async fn new() -> Self {
-    let region = Region::new("us-west-2");
-    let provider = aws_config::default_provider::credentials::DefaultCredentialsChain::builder()
-      .region(region.clone())
-      .build()
-      .await;
-    let endpoint = env::var(S3_ENDPOINT_ENV_VAR).unwrap_or_default();
-    let config = aws_sdk_s3::config::Builder::new()
-      .region(region)
-      .credentials_provider(provider)
-      .endpoint_url(endpoint)
-      .force_path_style(true)
-      .build();
-    let s3 = Client::from_conf(config);
+    let endpoint = env::var(S3_ENDPOINT_ENV_VAR).ok();
+    let aws_config = aws_config::from_env().load().await;
+    let mut s3_config_builder = aws_sdk_s3::config::Builder::from(&aws_config);
+    if let Some(endpoint) = endpoint {
+      s3_config_builder = s3_config_builder.endpoint_url(endpoint).force_path_style(true);
+    }
+    let s3_config = s3_config_builder.build();
+
+    let s3 = Client::from_conf(s3_config);
 
     Self {
       s3,
@@ -62,7 +58,7 @@ impl DataLake {
       .put_object()
       .acl(aws_sdk_s3::types::ObjectCannedAcl::BucketOwnerFullControl)
       .body(ByteStream::from(contents))
-      .bucket(self.bucket_name.clone())
+      .bucket(&self.bucket_name)
       .key(full_key)
       .send()
       .await?;

src/lake.rs

@@ -13,7 +13,7 @@ use tokio_util::sync::CancellationToken;
 
 const BATCH_SIZE_ENV_KEY: &str = "LAKE_SINK_BATCH_SIZE";
 const BATCH_SIZE_DEFAULT: &str = "1000";
-const BATCH_TIMEOUT_SECS: u64 = 45;
+const BATCH_TIMEOUT_SECS: u64 = 10;
 
 #[derive(Error, From, Display, Debug)]
 #[display(fmt = "Lake sink error: {}")]