fix(s3): support instance-policy credentials in object storage tests (#10238)

* fix(s3): cache ambient aws credentials and surface credential chain errors Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SCczeVBDhDnLXWqxPWw6od * fix(frontend): clarify that object storage connection tests run on the server Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SCczeVBDhDnLXWqxPWw6od * fix: render test connection tooltip in popup and harden cache test Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SCczeVBDhDnLXWqxPWw6od * better doc * fix(frontend): correct tooltip wording, tests run from the executing worker Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SCczeVBDhDnLXWqxPWw6od * fix(s3): single-flight ambient credential provider creation * fix(frontend): clarify object storage tests run on the server process * fix(frontend): make instance object storage test tooltip provider-neutral --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com> Co-authored-by: Ruben Fiszel <ruben@windmill.dev>
2026-09-09 08:03:50 +00:00 · 2026-07-21 20:02:34 +02:00
parent 2ce21c9ef8
commit ec6324409d
3 changed files with 165 additions and 33 deletions
@@ -283,15 +283,15 @@ pub async fn test_s3_bucket(
        let mut list = client.list(Some(
            &windmill_object_store::object_store_reexports::Path::from("".to_string()),
        ));
-        let first_file = list.next().await;
-        if first_file.is_some() {
-            if let Err(e) = first_file.as_ref().unwrap() {
+        match list.next().await {
+            Some(Err(e)) => {
                tracing::error!("error listing bucket: {e:#}");
-                error::Error::internal_err(format!("Failed to list files in blob storage: {e:#}"));
+                return Err(error::Error::internal_err(format!(
+                    "Failed to list files in blob storage: {e:#}"
+                )));
            }
-            tracing::info!("Listed files: {:?}", first_file.unwrap());
-        } else {
-            tracing::info!("No files in blob storage");
+            Some(Ok(first_file)) => tracing::info!("Listed files: {:?}", first_file),
+            None => tracing::info!("No files in blob storage"),
        }

        let path = windmill_object_store::object_store_reexports::Path::from(format!(
@@ -361,12 +361,7 @@ pub async fn build_s3_client(s3_resource_ref: &S3Resource) -> error::Result<Arc<
        || 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,
-        )
+        Some(ambient_aws_credentials_provider(&s3_resource_ref.region).await)
    } else {
        None
    };
@@ -764,10 +759,92 @@ pub async fn build_s3_client_from_settings(
    build_s3_client(&s3_resource).await
 }

+// Resolving the default chain goes over the network (ECS/IMDS) on instances relying on an
+// instance role, and object_store asks its CredentialProvider on every request — so resolved
+// credentials must be cached and only re-fetched when close to expiring.
+#[cfg(feature = "parquet")]
+#[derive(Debug)]
+struct AmbientAwsCredentials {
+    chain: DefaultCredentialsChain,
+    cached: RwLock<Option<(aws_sdk_sts::config::Credentials, std::time::Instant)>>,
+}
+
+#[cfg(feature = "parquet")]
+impl AmbientAwsCredentials {
+    // Credentials without an expiry (env vars, static profile) are still re-resolved
+    // periodically so runtime changes to the environment are eventually picked up.
+    const NO_EXPIRY_TTL: std::time::Duration = std::time::Duration::from_secs(300);
+    const EXPIRY_MARGIN: std::time::Duration = std::time::Duration::from_secs(120);
+
+    fn still_valid(creds: &aws_sdk_sts::config::Credentials, age: std::time::Duration) -> bool {
+        match creds.expiry() {
+            Some(expiry) => std::time::SystemTime::now() + Self::EXPIRY_MARGIN < expiry,
+            None => age < Self::NO_EXPIRY_TTL,
+        }
+    }
+
+    async fn get(&self) -> anyhow::Result<aws_sdk_sts::config::Credentials> {
+        if let Some((creds, fetched_at)) = self.cached.read().await.as_ref() {
+            if Self::still_valid(creds, fetched_at.elapsed()) {
+                return Ok(creds.clone());
+            }
+        }
+        // The write lock is held across the chain resolution so concurrent requests don't all
+        // hit the metadata service at once.
+        let mut guard = self.cached.write().await;
+        if let Some((creds, fetched_at)) = guard.as_ref() {
+            if Self::still_valid(creds, fetched_at.elapsed()) {
+                return Ok(creds.clone());
+            }
+        }
+        let creds = self.chain.provide_credentials().await.map_err(|e| {
+            anyhow::anyhow!(
+                "no S3 access key/secret key is configured and no ambient AWS credentials could \
+                 be loaded through the AWS SDK default chain (env vars, profile, ECS/EC2 instance \
+                 role): {cause}. If an EC2/ECS instance role is expected to be used, the instance \
+                 metadata service must be reachable from the process running Windmill — on EC2 the \
+                 AWS Rust SDK only supports IMDSv2, so when Windmill runs in a Docker container \
+                 the instance metadata hop limit (HttpPutResponseHopLimit) must be at least 2",
+                cause = format!("{:#}", anyhow::Error::new(e))
+            )
+        })?;
+        *guard = Some((creds.clone(), std::time::Instant::now()));
+        Ok(creds)
+    }
+}
+
+#[cfg(feature = "parquet")]
+lazy_static::lazy_static! {
+    static ref AMBIENT_AWS_CREDS_PROVIDERS: Cache<String, Arc<AmbientAwsCredentials>> =
+        Cache::new(20);
+}
+
+#[cfg(feature = "parquet")]
+async fn ambient_aws_credentials_provider(region: &str) -> Arc<AmbientAwsCredentials> {
+    // Single-flight: concurrent cold misses for the same region must share one provider,
+    // otherwise each gets its own instance and their per-instance refresh locks can't serialize
+    // the initial credential resolution — every caller would hit the metadata service.
+    match AMBIENT_AWS_CREDS_PROVIDERS
+        .get_value_or_guard_async(region)
+        .await
+    {
+        Ok(provider) => provider,
+        Err(guard) => {
+            let chain = DefaultCredentialsChain::builder()
+                .region(Region::new(region.to_string()))
+                .build()
+                .await;
+            let provider = Arc::new(AmbientAwsCredentials { chain, cached: RwLock::new(None) });
+            let _ = guard.insert(provider.clone());
+            provider
+        }
+    }
+}
+
 #[cfg(feature = "parquet")]
 #[derive(Debug)]
 struct AwsCredentialAdapter {
-    pub inner: DefaultCredentialsChain,
+    pub inner: Arc<AmbientAwsCredentials>,
 }

 #[cfg(feature = "parquet")]
@@ -775,9 +852,9 @@ struct AwsCredentialAdapter {
 impl CredentialProvider for AwsCredentialAdapter {
    type Credential = AwsCredential;
    async fn get_credential(&self) -> object_store::Result<Arc<Self::Credential>> {
-        let creds = self.inner.provide_credentials().await.map_err(|e| {
-            tracing::error!("Error getting credentials: {:?}", e);
-            object_store::Error::Generic { store: "AWS", source: Box::new(e) }
+        let creds = self.inner.get().await.map_err(|e| {
+            tracing::error!("Error getting AWS credentials: {e:#}");
+            object_store::Error::Generic { store: "AWS", source: e.into() }
        })?;
        Ok(Arc::new(Self::Credential {
            key_id: creds.access_key_id().to_string(),
@@ -1481,6 +1558,45 @@ pub async fn get_logs_from_store(
 mod tests {
    use super::*;

+    // --- ambient credentials cache tests ---
+
+    #[cfg(feature = "parquet")]
+    #[test]
+    fn test_ambient_credentials_still_valid() {
+        use std::time::{Duration, SystemTime};
+
+        fn creds(expiry: Option<SystemTime>) -> aws_sdk_sts::config::Credentials {
+            let mut builder = aws_sdk_sts::config::Credentials::builder()
+                .access_key_id("AK")
+                .secret_access_key("SK")
+                .provider_name("test");
+            if let Some(expiry) = expiry {
+                builder = builder.expiry(expiry);
+            }
+            builder.build()
+        }
+
+        // Expiry far in the future: valid regardless of fetch time
+        assert!(AmbientAwsCredentials::still_valid(
+            &creds(Some(SystemTime::now() + Duration::from_secs(3600))),
+            Duration::ZERO
+        ));
+        // Expiry within the refresh margin: must be re-fetched
+        assert!(!AmbientAwsCredentials::still_valid(
+            &creds(Some(SystemTime::now() + Duration::from_secs(30))),
+            Duration::ZERO
+        ));
+        // No expiry: valid while fresh, re-fetched after the TTL
+        assert!(AmbientAwsCredentials::still_valid(
+            &creds(None),
+            Duration::ZERO
+        ));
+        assert!(!AmbientAwsCredentials::still_valid(
+            &creds(None),
+            AmbientAwsCredentials::NO_EXPIRY_TTL + Duration::from_secs(1)
+        ));
+    }
+
    // --- render_endpoint tests ---

    #[test]
@@ -3,15 +3,16 @@

 	import { Database, Loader2 } from 'lucide-svelte'
 	import Button from './common/button/Button.svelte'
+	import Tooltip from './meltComponents/Tooltip.svelte'
 	import { sendUserToast } from '$lib/toast'
 	import { workspaceStore } from '$lib/stores'
 	import { tryEvery } from '$lib/utils'

 	interface Props {
-		workspaceOverride?: string | undefined;
-		resourceType: string | undefined;
-		args?: Record<string, any> | any;
-		buttonTextOverride?: string | undefined;
+		workspaceOverride?: string | undefined
+		resourceType: string | undefined
+		args?: Record<string, any> | any
+		buttonTextOverride?: string | undefined
 	}

 	let {
@@ -19,13 +20,15 @@
 		resourceType,
 		args = {},
 		buttonTextOverride = undefined
-	}: Props = $props();
+	}: Props = $props()

 	const scripts: {
 		[key: string]: {
 			code: string
 			lang: string
 			argName: string
+			// Shown as an info tooltip next to the button, e.g. to clarify where the test executes
+			tooltip?: string
 			additionalCheck?: (testResult: CompletedJob) => CompletedJob
 		}
 	} = {
@@ -97,7 +100,9 @@ export async function main(s3: S3) {
 }
 `,
 			lang: 'bun',
-			argName: 's3'
+			argName: 's3',
+			tooltip:
+				'The storage operations of this test run on the Windmill server (the API process), not on the worker. If no access key/secret key is set, the ambient AWS credentials of the server (environment variables, instance role) are used — scripts using this resource directly through an S3 SDK resolve credentials on the worker instead, so results may differ.'
 		},
 		azure_blob: {
 			code: `
@@ -125,7 +130,9 @@ export async function main(s3: S3) {
 }
 `,
 			lang: 'bun',
-			argName: 's3'
+			argName: 's3',
+			tooltip:
+				'The storage operations of this test run on the Windmill server (the API process), not on the worker.'
 		},
 		graphql: {
 			code: '{ __typename }',
@@ -171,7 +178,9 @@ export async function main(bucket: any) {
 }
 `,
 			lang: 'bun',
-			argName: 'bucket'
+			argName: 'bucket',
+			tooltip:
+				"The storage operations of this test run on the Windmill server (the API process). If no credentials are configured, the server's ambient credentials for the configured provider (environment variables, instance role) are used."
 		}
 	}

@@ -236,13 +245,20 @@ export async function main(bucket: any) {
 	}
 </script>

-{#if Object.keys(scripts).includes(resourceType || '')}
-	<Button spacingSize="sm" size="xs" unifiedSize="md" variant="default" on:click={testConnection}>
-		{#if loading}
-			<Loader2 class="animate-spin mr-2 !h-4 !w-4" />
-		{:else}
-			<Database class="mr-2 !h-4 !w-4" />
+{#if resourceType && Object.keys(scripts).includes(resourceType)}
+	<div class="flex items-center gap-1">
+		<Button spacingSize="sm" size="xs" unifiedSize="md" variant="default" on:click={testConnection}>
+			{#if loading}
+				<Loader2 class="animate-spin mr-2 !h-4 !w-4" />
+			{:else}
+				<Database class="mr-2 !h-4 !w-4" />
+			{/if}
+			{buttonTextOverride ?? 'Test connection'}
+		</Button>
+		{#if scripts[resourceType].tooltip}
+			<Tooltip>
+				{#snippet text()}{scripts[resourceType].tooltip}{/snippet}
+			</Tooltip>
 		{/if}
-		{buttonTextOverride ?? 'Test connection'}
-	</Button>
+	</div>
 {/if}