bottomless: prefix AWS env vars in use with LIBSQL_BOTTOMLESS_ (#649)

* bottomless: prefix AWS env vars in use with LIBSQL_BOTTOMLESS_

* changed env vars in workflow defs

* post rebase fixes

Author: Horusiath

.github/workflows/rust.yml

@@ -91,9 +91,9 @@ jobs:
         command: test
         args: --verbose
       env:
-        AWS_ACCESS_KEY_ID: minioadmin
-        AWS_SECRET_ACCESS_KEY: minioadmin
-        AWS_DEFAULT_REGION: eu-central-2
+        LIBSQL_BOTTOMLESS_AWS_ACCESS_KEY_ID: minioadmin
+        LIBSQL_BOTTOMLESS_AWS_SECRET_ACCESS_KEY: minioadmin
+        LIBSQL_BOTTOMLESS_AWS_DEFAULT_REGION: eu-central-2
         LIBSQL_BOTTOMLESS_BUCKET: bottomless
         LIBSQL_BOTTOMLESS_ENDPOINT: http://localhost:9000
 

README.md

@@ -83,9 +83,9 @@ environment variables can be used to configure the replication:
 ```bash
 LIBSQL_BOTTOMLESS_BUCKET=my-bucket                 # Default bucket name: bottomless
 LIBSQL_BOTTOMLESS_ENDPOINT='http://localhost:9000' # address can be overridden for local testing, e.g. with Minio
-AWS_SECRET_ACCESS_KEY=                             # regular AWS variables are used
-AWS_ACCESS_KEY_ID=                                 # ... to set up auth, regions, etc.
-AWS_REGION=                                        # .
+LIBSQL_BOTTOMLESS_AWS_SECRET_ACCESS_KEY=           # regular AWS variables are used
+LIBSQL_BOTTOMLESS_AWS_ACCESS_KEY_ID=               # ... to set up auth, regions, etc.
+LIBSQL_BOTTOMLESS_AWS_REGION=                      # .
 ```
 
 ### bottomless-cli

bottomless/src/replicator.rs

@@ -6,6 +6,7 @@ use crate::wal::WalFileReader;
 use anyhow::anyhow;
 use arc_swap::ArcSwap;
 use async_compression::tokio::write::GzipEncoder;
+use aws_sdk_s3::config::{Credentials, Region};
 use aws_sdk_s3::error::SdkError;
 use aws_sdk_s3::operation::get_object::builders::GetObjectFluentBuilder;
 use aws_sdk_s3::operation::get_object::GetObjectError;
@@ -85,6 +86,9 @@ pub struct Options {
     /// Kind of compression algorithm used on the WAL frames to be sent to S3.
     pub use_compression: CompressionKind,
     pub aws_endpoint: Option<String>,
+    pub access_key_id: Option<String>,
+    pub secret_access_key: Option<String>,
+    pub region: Option<String>,
     pub db_id: Option<String>,
     /// Bucket directory name where all S3 objects are backed up. General schema is:
     /// - `{db-name}-{uuid-v7}` subdirectories:
@@ -109,14 +113,34 @@ pub struct Options {
 }
 
 impl Options {
-    pub async fn client_config(&self) -> Config {
+    pub async fn client_config(&self) -> Result<Config> {
         let mut loader = aws_config::from_env();
         if let Some(endpoint) = self.aws_endpoint.as_deref() {
             loader = loader.endpoint_url(endpoint);
         }
-        aws_sdk_s3::config::Builder::from(&loader.load().await)
+        let region = self
+            .region
+            .clone()
+            .ok_or(anyhow!("LIBSQL_BOTTOMLESS_AWS_DEFAULT_REGION was not set"))?;
+        let access_key_id = self
+            .access_key_id
+            .clone()
+            .ok_or(anyhow!("LIBSQL_BOTTOMLESS_AWS_ACCESS_KEY_ID was not set"))?;
+        let secret_access_key = self.secret_access_key.clone().ok_or(anyhow!(
+            "LIBSQL_BOTTOMLESS_AWS_SECRET_ACCESS_KEY was not set"
+        ))?;
+        let conf = aws_sdk_s3::config::Builder::from(&loader.load().await)
             .force_path_style(true)
-            .build()
+            .region(Region::new(region))
+            .credentials_provider(Credentials::new(
+                access_key_id,
+                secret_access_key,
+                None,
+                None,
+                "Static",
+            ))
+            .build();
+        Ok(conf)
     }
 
     pub fn from_env() -> Result<Self> {
@@ -132,6 +156,15 @@ impl Options {
                 options.max_batch_interval = Duration::from_secs(seconds);
             }
         }
+        if let Ok(access_key_id) = std::env::var("LIBSQL_BOTTOMLESS_AWS_ACCESS_KEY_ID") {
+            options.access_key_id = Some(access_key_id);
+        }
+        if let Ok(secret_access_key) = std::env::var("LIBSQL_BOTTOMLESS_AWS_SECRET_ACCESS_KEY") {
+            options.secret_access_key = Some(secret_access_key);
+        }
+        if let Ok(region) = std::env::var("LIBSQL_BOTTOMLESS_AWS_DEFAULT_REGION") {
+            options.region = Some(region);
+        }
         if let Ok(count) = std::env::var("LIBSQL_BOTTOMLESS_BATCH_MAX_FRAMES") {
             match count.parse::<usize>() {
                 Ok(count) => options.max_frames_per_batch = count,
@@ -208,6 +241,9 @@ impl Default for Options {
             restore_transaction_page_swap_after: 1000,
             db_id,
             aws_endpoint: None,
+            access_key_id: None,
+            secret_access_key: None,
+            region: None,
             restore_transaction_cache_fpath: ".bottomless.restore".to_string(),
             bucket_name: "bottomless".to_string(),
         }
@@ -222,7 +258,7 @@ impl Replicator {
     }
 
     pub async fn with_options<S: Into<String>>(db_path: S, options: Options) -> Result<Self> {
-        let config = options.client_config().await;
+        let config = options.client_config().await?;
         let client = Client::from_conf(config);
         let bucket = options.bucket_name.clone();
         let generation = Arc::new(ArcSwap::new(Arc::new(Self::generate_generation())));

docker-compose/docker-compose-with-bottomless.yml

@@ -6,9 +6,9 @@ services:
       - SQLD_NODE=primary
       - SQLD_ENABLE_BOTTOMLESS_REPLICATION=true
       - LIBSQL_BOTTOMLESS_ENDPOINT=http://s3:9000
-      - AWS_ACCESS_KEY_ID=minioadmin
-      - AWS_SECRET_ACCESS_KEY=minioadmin
-      - AWS_DEFAULT_REGION=eu-central-2
+      - LIBSQL_BOTTOMLESS_AWS_ACCESS_KEY_ID=minioadmin
+      - LIBSQL_BOTTOMLESS_AWS_SECRET_ACCESS_KEY=minioadmin
+      - LIBSQL_BOTTOMLESS_AWS_DEFAULT_REGION=eu-central-2
       - RUST_LOG=info,bottomless=trace
     ports:
       - "6000:5000"

sqld/src/test/bottomless.rs

@@ -1,5 +1,6 @@
 use crate::{run_server, Config};
 use anyhow::Result;
+use aws_sdk_s3::config::{Credentials, Region};
 use libsql_client::{Connection, QueryResult, Statement, Value};
 use std::net::ToSocketAddrs;
 use std::path::PathBuf;
@@ -322,16 +323,29 @@ where
     db.batch(stmts).await
 }
 
+async fn s3_config() -> aws_sdk_s3::config::Config {
+    let loader = aws_config::from_env().endpoint_url(S3_URL);
+    aws_sdk_s3::config::Builder::from(&loader.load().await)
+        .force_path_style(true)
+        .region(Region::new(
+            std::env::var("LIBSQL_BOTTOMLESS_AWS_DEFAULT_REGION").unwrap(),
+        ))
+        .credentials_provider(Credentials::new(
+            std::env::var("LIBSQL_BOTTOMLESS_AWS_ACCESS_KEY_ID").unwrap(),
+            std::env::var("LIBSQL_BOTTOMLESS_AWS_SECRET_ACCESS_KEY").unwrap(),
+            None,
+            None,
+            "Static",
+        ))
+        .build()
+}
+
 /// Checks if the corresponding bucket is empty (has any elements) or not.
 /// If bucket was not found, it's equivalent of an empty one.
 async fn assert_bucket_occupancy(bucket: &str, expect_empty: bool) {
     use aws_sdk_s3::Client;
 
-    let loader = aws_config::from_env().endpoint_url(S3_URL);
-    let conf = aws_sdk_s3::config::Builder::from(&loader.load().await)
-        .force_path_style(true)
-        .build();
-    let client = Client::from_conf(conf);
+    let client = Client::from_conf(s3_config().await);
     if let Ok(out) = client.list_objects().bucket(bucket).send().await {
         let contents = out.contents().unwrap_or_default();
         if expect_empty {
@@ -388,11 +402,7 @@ impl S3BucketCleaner {
         use aws_sdk_s3::types::{Delete, ObjectIdentifier};
         use aws_sdk_s3::Client;
 
-        let loader = aws_config::from_env().endpoint_url(S3_URL);
-        let conf = aws_sdk_s3::config::Builder::from(&loader.load().await)
-            .force_path_style(true)
-            .build();
-        let client = Client::from_conf(conf);
+        let client = Client::from_conf(s3_config().await);
         let objects = client.list_objects().bucket(bucket).send().await?;
         let mut delete_keys = Vec::new();
         for o in objects.contents().unwrap_or_default() {