Fixing features, more renaming and reorganising

Signed-off-by: itowlson <[email protected]>

Author: itowlson

crates/blobstore-azure/Cargo.toml

@@ -19,7 +19,7 @@ spin-core = { path = "../core" }
 spin-factor-blobstore = { path = "../factor-blobstore" }
 tokio = { workspace = true }
 tokio-stream = "0.1.16"
-tokio-util = "0.7.12"
+tokio-util = { version = "0.7.12", features = ["compat"] }
 uuid = { version = "1.0", features = ["v4"] }
 wasmtime-wasi = { workspace = true }
 

crates/blobstore-fs/Cargo.toml

@@ -16,7 +16,7 @@ spin-core = { path = "../core" }
 spin-factor-blobstore = { path = "../factor-blobstore" }
 tokio = { workspace = true }
 tokio-stream = "0.1.16"
-tokio-util = "0.7.12"
+tokio-util = { version = "0.7.12", features = ["codec", "compat"] }
 walkdir = "2.5"
 wasmtime-wasi = { workspace = true }
 

crates/blobstore-s3/Cargo.toml

@@ -25,7 +25,7 @@ spin-core = { path = "../core" }
 spin-factor-blobstore = { path = "../factor-blobstore" }
 tokio = { workspace = true }
 tokio-stream = "0.1.16"
-tokio-util = "0.7.12"
+tokio-util = { version = "0.7.12", features = ["compat"] }
 uuid = { version = "1.0", features = ["v4"] }
 wasmtime-wasi = { workspace = true }
 

crates/blobstore-s3/src/lib.rs

@@ -2,7 +2,7 @@ mod store;
 
 use serde::Deserialize;
 use spin_factor_blobstore::runtime_config::spin::MakeBlobStore;
-use store::BlobStoreS3;
+use store::S3ContainerManager;
 
 /// A blob store that uses a S3-compatible service as the backend.
 /// This currently supports only AWS S3
@@ -40,29 +40,28 @@ impl MakeBlobStore for S3BlobStore {
 
     type RuntimeConfig = S3BlobStoreRuntimeConfig;
 
-    type ContainerManager = BlobStoreS3;
+    type ContainerManager = S3ContainerManager;
 
     fn make_store(
         &self,
         runtime_config: Self::RuntimeConfig,
     ) -> anyhow::Result<Self::ContainerManager> {
         let auth = match (&runtime_config.access_key, &runtime_config.secret_key) {
             (Some(access_key), Some(secret_key)) => {
-                store::BlobStoreS3AuthOptions::RuntimeConfigValues(
-                    store::BlobStoreS3RuntimeConfigOptions::new(
-                        access_key.clone(),
-                        secret_key.clone(),
-                        runtime_config.token.clone(),
-                    ),
-                )
+                store::S3AuthOptions::AccessKey(store::S3KeyAuth::new(
+                    access_key.clone(),
+                    secret_key.clone(),
+                    runtime_config.token.clone(),
+                ))
             }
-            (None, None) => store::BlobStoreS3AuthOptions::Environmental,
+            (None, None) => store::S3AuthOptions::Environmental,
             _ => anyhow::bail!(
                 "either both of access_key and secret_key must be provided, or neither"
             ),
         };
 
-        let blob_store = BlobStoreS3::new(runtime_config.region, auth, runtime_config.bucket)?;
+        let blob_store =
+            S3ContainerManager::new(runtime_config.region, auth, runtime_config.bucket)?;
         Ok(blob_store)
     }
 }

crates/blobstore-s3/src/store.rs

@@ -1,16 +1,18 @@
-use aws_sdk_s3::config::http::HttpResponse as AwsHttpResponse;
-use aws_sdk_s3::error::SdkError;
-use aws_sdk_s3::operation::get_object;
-use aws_sdk_s3::operation::list_objects_v2;
-use aws_smithy_async::future::pagination_stream::PaginationStream;
 use std::sync::Arc;
-use tokio::sync::Mutex;
 
 use anyhow::Result;
 use spin_core::async_trait;
 use spin_factor_blobstore::{Container, ContainerManager, Error};
 
-pub struct BlobStoreS3 {
+mod auth;
+mod incoming_data;
+mod object_names;
+
+pub use auth::{S3AuthOptions, S3KeyAuth};
+use incoming_data::S3IncomingData;
+use object_names::S3ObjectNames;
+
+pub struct S3ContainerManager {
     builder: object_store::aws::AmazonS3Builder,
     client: async_once_cell::Lazy<
         aws_sdk_s3::Client,
@@ -19,85 +21,32 @@ pub struct BlobStoreS3 {
     bucket: Option<String>,
 }
 
-/// AWS S3 runtime config literal options for authentication
-#[derive(Clone, Debug)]
-pub struct BlobStoreS3RuntimeConfigOptions {
-    /// The access key for the AWS S3 account role.
-    access_key: String,
-    /// The secret key for authorization on the AWS S3 account.
-    secret_key: String,
-    /// The token for authorization on the AWS S3 account.
-    token: Option<String>,
-}
-
-impl BlobStoreS3RuntimeConfigOptions {
-    pub fn new(access_key: String, secret_key: String, token: Option<String>) -> Self {
-        Self {
-            access_key,
-            secret_key,
-            token,
-        }
-    }
-}
-
-impl aws_credential_types::provider::ProvideCredentials for BlobStoreS3RuntimeConfigOptions {
-    fn provide_credentials<'a>(
-        &'a self,
-    ) -> aws_credential_types::provider::future::ProvideCredentials<'a>
-    where
-        Self: 'a,
-    {
-        aws_credential_types::provider::future::ProvideCredentials::ready(Ok(
-            aws_credential_types::Credentials::new(
-                self.access_key.clone(),
-                self.secret_key.clone(),
-                self.token.clone(),
-                None, // Optional expiration time
-                "spin_custom_aws_provider",
-            ),
-        ))
-    }
-}
-
-/// AWS S3 authentication options
-#[derive(Clone, Debug)]
-pub enum BlobStoreS3AuthOptions {
-    /// Runtime Config values indicates the account and key have been specified directly
-    RuntimeConfigValues(BlobStoreS3RuntimeConfigOptions),
-    /// Use environment variables
-    Environmental,
-}
-
-impl BlobStoreS3 {
+impl S3ContainerManager {
     pub fn new(
         region: String,
-        auth_options: BlobStoreS3AuthOptions,
+        auth_options: S3AuthOptions,
         bucket: Option<String>,
     ) -> Result<Self> {
         let builder = match &auth_options {
-            BlobStoreS3AuthOptions::RuntimeConfigValues(config) => {
-                object_store::aws::AmazonS3Builder::new()
-                    .with_region(&region)
-                    .with_access_key_id(&config.access_key)
-                    .with_secret_access_key(&config.secret_key)
-                    .with_token(config.token.clone().unwrap_or_default())
-            }
-            BlobStoreS3AuthOptions::Environmental => object_store::aws::AmazonS3Builder::from_env(),
+            S3AuthOptions::AccessKey(config) => object_store::aws::AmazonS3Builder::new()
+                .with_region(&region)
+                .with_access_key_id(&config.access_key)
+                .with_secret_access_key(&config.secret_key)
+                .with_token(config.token.clone().unwrap_or_default()),
+            S3AuthOptions::Environmental => object_store::aws::AmazonS3Builder::from_env(),
         };
 
         let region_clone = region.clone();
         let client_fut = Box::pin(async move {
             let sdk_config = match auth_options {
-                BlobStoreS3AuthOptions::RuntimeConfigValues(config) => {
-                    aws_config::SdkConfig::builder()
-                        .credentials_provider(aws_sdk_s3::config::SharedCredentialsProvider::new(
-                            config,
-                        ))
-                        .region(aws_config::Region::new(region_clone))
-                        .behavior_version(aws_config::BehaviorVersion::latest())
-                        .build()
-                }
-                BlobStoreS3AuthOptions::Environmental => {
+                S3AuthOptions::AccessKey(config) => aws_config::SdkConfig::builder()
+                    .credentials_provider(aws_sdk_s3::config::SharedCredentialsProvider::new(
+                        config,
+                    ))
+                    .region(aws_config::Region::new(region_clone))
+                    .behavior_version(aws_config::BehaviorVersion::latest())
+                    .build(),
+                S3AuthOptions::Environmental => {
                     aws_config::load_defaults(aws_config::BehaviorVersion::latest()).await
                 }
             };
@@ -113,7 +62,7 @@ impl BlobStoreS3 {
 }
 
 #[async_trait]
-impl ContainerManager for BlobStoreS3 {
+impl ContainerManager for S3ContainerManager {
     async fn get(&self, name: &str) -> Result<Arc<dyn Container>, Error> {
         let name = self.bucket.clone().unwrap_or_else(|| name.to_owned());
 
@@ -280,7 +229,7 @@ impl Container for S3Container {
             .bucket(&self.name)
             .into_paginator()
             .send();
-        Ok(Box::new(S3BlobsList::new(stm)))
+        Ok(Box::new(S3ObjectNames::new(stm)))
     }
 }
 
@@ -311,154 +260,3 @@ impl S3Container {
         Ok(())
     }
 }
-
-struct S3IncomingData {
-    get_obj_resp: Option<get_object::GetObjectOutput>,
-}
-
-impl S3IncomingData {
-    fn new(get_obj_resp: get_object::GetObjectOutput) -> Self {
-        Self {
-            get_obj_resp: Some(get_obj_resp),
-        }
-    }
-
-    fn consume_async_impl(&mut self) -> wasmtime_wasi::pipe::AsyncReadStream {
-        use futures::TryStreamExt;
-        use tokio_util::compat::FuturesAsyncReadCompatExt;
-        let stm = self.consume_as_stream();
-        let ar = stm.into_async_read();
-        let arr = ar.compat();
-        wasmtime_wasi::pipe::AsyncReadStream::new(arr)
-    }
-
-    fn consume_as_stream(
-        &mut self,
-    ) -> impl futures::stream::Stream<Item = Result<Vec<u8>, std::io::Error>> {
-        use futures::StreamExt;
-        let rr = self
-            .get_obj_resp
-            .take()
-            .expect("get object resp was already consumed");
-        let ar = rr.body.into_async_read();
-        let s = tokio_util::io::ReaderStream::new(ar);
-        s.map(|by| by.map(|b| b.to_vec()))
-    }
-}
-
-struct S3BlobsList {
-    stm: Mutex<
-        PaginationStream<
-            Result<
-                list_objects_v2::ListObjectsV2Output,
-                SdkError<list_objects_v2::ListObjectsV2Error, AwsHttpResponse>,
-            >,
-        >,
-    >,
-    read_but_not_yet_returned: Vec<String>,
-    end_stm_after_read_but_not_yet_returned: bool,
-}
-
-impl S3BlobsList {
-    fn new(
-        stm: PaginationStream<
-            Result<
-                list_objects_v2::ListObjectsV2Output,
-                SdkError<list_objects_v2::ListObjectsV2Error, AwsHttpResponse>,
-            >,
-        >,
-    ) -> Self {
-        Self {
-            stm: Mutex::new(stm),
-            read_but_not_yet_returned: Default::default(),
-            end_stm_after_read_but_not_yet_returned: false,
-        }
-    }
-
-    async fn read_impl(&mut self, len: u64) -> anyhow::Result<(Vec<String>, bool)> {
-        let len: usize = len.try_into().unwrap();
-
-        // If we have names outstanding, send that first.  (We are allowed to send less than len,
-        // and so sending all pending stuff before paging, rather than trying to manage a mix of
-        // pending stuff with newly retrieved chunks, simplifies the code.)
-        if !self.read_but_not_yet_returned.is_empty() {
-            if self.read_but_not_yet_returned.len() <= len {
-                // We are allowed to send all pending names
-                let to_return = self.read_but_not_yet_returned.drain(..).collect();
-                return Ok((to_return, self.end_stm_after_read_but_not_yet_returned));
-            } else {
-                // Send as much as we can. The rest remains in the pending buffer to send,
-                // so this does not represent end of stream.
-                let to_return = self.read_but_not_yet_returned.drain(0..len).collect();
-                return Ok((to_return, false));
-            }
-        }
-
-        // Get one chunk and send as much as we can of it. Aagin, we don't need to try to
-        // pack the full length here - we can send chunk by chunk.
-
-        let Some(chunk) = self.stm.get_mut().next().await else {
-            return Ok((vec![], false));
-        };
-        let chunk = chunk.unwrap();
-
-        let at_end = chunk.continuation_token().is_none();
-        let mut names: Vec<_> = chunk
-            .contents
-            .unwrap_or_default()
-            .into_iter()
-            .flat_map(|blob| blob.key)
-            .collect();
-
-        if names.len() <= len {
-            // We can send them all!
-            Ok((names, at_end))
-        } else {
-            // We have more names than we can send in this response. Send what we can and
-            // stash the rest.
-            let to_return: Vec<_> = names.drain(0..len).collect();
-            self.read_but_not_yet_returned = names;
-            self.end_stm_after_read_but_not_yet_returned = at_end;
-            Ok((to_return, false))
-        }
-    }
-}
-
-#[async_trait]
-impl spin_factor_blobstore::IncomingData for S3IncomingData {
-    async fn consume_sync(&mut self) -> anyhow::Result<Vec<u8>> {
-        let Some(goo) = self.get_obj_resp.take() else {
-            anyhow::bail!("oh no");
-        };
-
-        Ok(goo.body.collect().await?.to_vec())
-    }
-
-    fn consume_async(&mut self) -> wasmtime_wasi::pipe::AsyncReadStream {
-        self.consume_async_impl()
-    }
-
-    async fn size(&mut self) -> anyhow::Result<u64> {
-        use anyhow::Context;
-        let goo = self.get_obj_resp.as_ref().context("resp has been taken")?;
-        Ok(goo
-            .content_length()
-            .context("content-length not returned")?
-            .try_into()?)
-    }
-}
-
-#[async_trait]
-impl spin_factor_blobstore::ObjectNames for S3BlobsList {
-    async fn read(&mut self, len: u64) -> anyhow::Result<(Vec<String>, bool)> {
-        self.read_impl(len).await // Separate function because rust-analyser gives better intellisense when async_trait isn't in the picture!
-    }
-
-    async fn skip(&mut self, num: u64) -> anyhow::Result<(u64, bool)> {
-        // TODO: there is a question (raised as an issue on the repo) about the required behaviour
-        // here. For now I assume that skipping fewer than `num` is allowed as long as we are
-        // honest about it. Because it is easier that is why.
-        let (skipped, at_end) = self.read_impl(num).await?;
-        Ok((skipped.len().try_into().unwrap(), at_end))
-    }
-}