diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 2c2cd13f33..78fe26255f 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -f667fcebdb5a2d1c22c1928d6a316ba57bb43d72 \ No newline at end of file +150c3176f2c192366f2e8298ef6ece43780b529e \ No newline at end of file diff --git a/backend/windmill-common/src/s3_helpers.rs b/backend/windmill-common/src/s3_helpers.rs index 198283db2d..4547c29d51 100644 --- a/backend/windmill-common/src/s3_helpers.rs +++ b/backend/windmill-common/src/s3_helpers.rs @@ -117,10 +117,10 @@ pub struct S3Object { #[cfg(feature = "parquet")] pub async fn get_etag_or_empty( - object_store_resource: &ObjectStoreResource, + object_store_resource: &mut ObjectStoreResource, s3_object: S3Object, ) -> Option { - let object_store_client = build_object_store_client(object_store_resource); + let object_store_client = build_object_store_client(object_store_resource).await; if object_store_client.is_err() { return None; } @@ -166,11 +166,11 @@ pub fn render_endpoint( } #[cfg(feature = "parquet")] -pub fn build_object_store_client( +pub async fn build_object_store_client( resource_ref: &ObjectStoreResource, ) -> error::Result> { match resource_ref { - ObjectStoreResource::S3(s3_resource_ref) => build_s3_client(&s3_resource_ref, None), + ObjectStoreResource::S3(s3_resource_ref) => build_s3_client(&s3_resource_ref).await, ObjectStoreResource::Azure(azure_blob_resource_ref) => { build_azure_blob_client(&azure_blob_resource_ref) } @@ -225,10 +225,21 @@ use aws_config::{default_provider::credentials::DefaultCredentialsChain, Region} use object_store::CredentialProvider; #[cfg(feature = "parquet")] -pub fn build_s3_client( - s3_resource_ref: &S3Resource, - credential_providers: Option, -) -> error::Result> { +pub async fn build_s3_client(s3_resource_ref: &S3Resource) -> error::Result> { + let static_creds = s3_resource_ref.access_key.as_ref().is_some_and(|x| x != "") + || s3_resource_ref.secret_key.as_ref().is_some_and(|x| x != ""); + + let credentials_provider = if !static_creds { + Some( + DefaultCredentialsChain::builder() + .region(Region::new(s3_resource_ref.region.clone())) + .build() + .await, + ) + } else { + None + }; + let s3_resource = s3_resource_ref.clone(); let endpoint = render_endpoint( s3_resource.endpoint, @@ -244,7 +255,7 @@ pub fn build_s3_client( .with_bucket_name(s3_resource.bucket) .with_endpoint(endpoint); - if let Some(credentials_provider) = credential_providers { + if let Some(credentials_provider) = credentials_provider { store_builder = store_builder.with_credentials(Arc::new(AwsCredentialAdapter { inner: credentials_provider, })); @@ -399,18 +410,7 @@ pub async fn build_s3_client_from_settings( ) -> error::Result> { let region = none_if_empty(settings.region) .unwrap_or_else(|| std::env::var("AWS_REGION").unwrap_or_else(|_| "us-east-1".to_string())); - let access_key = none_if_empty(settings.access_key); - let secret_key = none_if_empty(settings.secret_key); - let credentials_provider = if access_key.is_none() && secret_key.is_none() { - Some( - DefaultCredentialsChain::builder() - .region(Region::new(region.clone())) - .build() - .await, - ) - } else { - None - }; + let s3_resource = S3Resource { endpoint: none_if_empty(settings.endpoint).unwrap_or_else(|| { std::env::var("S3_ENDPOINT").unwrap_or_else(|_| format!("s3.{region}.amazonaws.com")) @@ -419,15 +419,15 @@ pub async fn build_s3_client_from_settings( std::env::var("S3_CACHE_BUCKET").unwrap_or_else(|_| "missingbucket".to_string()) }), region, - access_key, - secret_key, + access_key: settings.access_key, + secret_key: settings.secret_key, use_ssl: !settings.allow_http.unwrap_or(true), path_style: settings.path_style, port: settings.port, token: None, }; - build_s3_client(&s3_resource, credentials_provider) + build_s3_client(&s3_resource).await } #[cfg(feature = "parquet")] diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index 406f128917..e22f9be54c 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -736,8 +736,8 @@ async fn arg_value_hash_additions( .await; storage = s3_object.storage.clone(); - if let Some(s3_resource) = s3_resource_opt.ok().flatten() { - let etag = get_etag_or_empty(&s3_resource, s3_object.clone()).await; + if let Some(mut s3_resource) = s3_resource_opt.ok().flatten() { + let etag = get_etag_or_empty(&mut s3_resource, s3_object.clone()).await; tracing::warn!("Enriching s3 arg value with etag: {:?}", etag); result.insert(s3_object.s3.clone(), etag.unwrap_or_default()); // TODO: maybe inject a random value to invalidate the cache? } @@ -793,9 +793,9 @@ pub async fn get_cached_resource_value_if_valid( return None; } for (s3_file_key, s3_file_etag) in s3_etags { - if let Some(object_store_resource) = object_store_resource_opt.clone() { + if let Some(mut object_store_resource) = object_store_resource_opt.clone() { let etag = get_etag_or_empty( - &object_store_resource, + &mut object_store_resource, S3Object { s3: s3_file_key.clone(), storage: cached_resource.storage.clone(),