S3 Proxy better errors + DuckDB S3 fix (#6740)

* Support for MinIO and other S3 impls in S3 Proxy

* Nice S3 Proxy error messages in DuckDB executor

* nit

* useless code

* super nit

* ee repo ref

backend/ee-repo-ref.txt

@@ -1 +1 @@
-0835279921261f03720513c0f7ffabc4df44db6f
+965976ae4bcf37692b7ce0df6be60e9453c261e3

backend/windmill-common/src/s3_helpers.rs

@@ -14,6 +14,7 @@ use object_store::gcp::GoogleCloudStorageBuilder;
 use object_store::ObjectStore;
 #[cfg(feature = "parquet")]
 use object_store::{aws::AmazonS3Builder, ClientOptions};
+use quick_cache::sync::Cache;
 #[cfg(feature = "parquet")]
 use reqwest::header::HeaderMap;
 use serde::de::Visitor;
@@ -414,18 +415,6 @@ pub struct AzureBlobResource {
     pub federated_token_file: Option<String>,
 }
 
-impl AzureBlobResource {
-    pub fn get_endpoint_url(&self) -> error::Result<String> {
-        Ok(render_endpoint(
-            self.endpoint.clone().unwrap_or_else(|| "".to_string()),
-            self.use_ssl.unwrap_or(false),
-            None,
-            None,
-            "".to_string(),
-        ))
-    }
-}
-
 fn as_string<'de, D>(deserializer: D) -> Result<String, D::Error>
 where
     D: serde::de::Deserializer<'de>,
@@ -526,7 +515,6 @@ pub async fn build_object_store_client(
     }
 }
 
-
 #[derive(PartialEq)]
 pub enum BundleFormat {
     Esm,
@@ -534,7 +522,7 @@ pub enum BundleFormat {
 }
 
 impl BundleFormat {
-   pub fn from_string(s: &str) -> Option<Self> {
+    pub fn from_string(s: &str) -> Option<Self> {
         match s {
             "esm" => Some(Self::Esm),
             "cjs" => Some(Self::Cjs),
@@ -543,49 +531,51 @@ impl BundleFormat {
     }
 }
 
-pub async fn upload_artifact_to_store(path: &str, data: bytes::Bytes, standalone_dir: &str) -> error::Result<()> {
+pub async fn upload_artifact_to_store(
+    path: &str,
+    data: bytes::Bytes,
+    standalone_dir: &str,
+) -> error::Result<()> {
     #[cfg(all(feature = "enterprise", feature = "parquet"))]
     let object_store = crate::s3_helpers::get_object_store().await;
     #[cfg(not(all(feature = "enterprise", feature = "parquet")))]
     let object_store: Option<()> = None;
-    Ok(if &crate::utils::MODE_AND_ADDONS.mode
-        == &crate::utils::Mode::Standalone
-        && object_store.is_none()
-    {
-        let path = format!("{}/{}", standalone_dir, path);
-        tracing::info!("Writing file to path {path}");
-
-        let split_path = path.split("/").collect::<Vec<&str>>();
-        std::fs::create_dir_all(
-            split_path[..split_path.len() - 1].join("/"),
-        )?;
-
-        crate::worker::write_file_bytes(
-            &path,
-            &data,
-        )?;
-    } else {
-        #[cfg(not(all(feature = "enterprise", feature = "parquet")))]
+    Ok(
+        if &crate::utils::MODE_AND_ADDONS.mode == &crate::utils::Mode::Standalone
+            && object_store.is_none()
         {
-            return Err(error::Error::ExecutionErr("codebase is an EE feature".to_string()));
-        }
+            let path = format!("{}/{}", standalone_dir, path);
+            tracing::info!("Writing file to path {path}");
 
-        #[cfg(all(feature = "enterprise", feature = "parquet"))]
-        if let Some(os) = object_store {
+            let split_path = path.split("/").collect::<Vec<&str>>();
+            std::fs::create_dir_all(split_path[..split_path.len() - 1].join("/"))?;
 
-            if let Err(e) = os
-                .put(&object_store::path::Path::from(path), data.into())
-                .await
-            {
-                tracing::info!("Failed to put snapshot to s3 at {path}: {:?}", e);
-                return Err(error::Error::ExecutionErr(format!("Failed to put {path} to s3")));
-            }
+            crate::worker::write_file_bytes(&path, &data)?;
         } else {
-            return Err(error::Error::BadConfig("Object store is required for snapshot script and is not configured for servers".to_string()));
-        }
-    })
-}
+            #[cfg(not(all(feature = "enterprise", feature = "parquet")))]
+            {
+                return Err(error::Error::ExecutionErr(
+                    "codebase is an EE feature".to_string(),
+                ));
+            }
 
+            #[cfg(all(feature = "enterprise", feature = "parquet"))]
+            if let Some(os) = object_store {
+                if let Err(e) = os
+                    .put(&object_store::path::Path::from(path), data.into())
+                    .await
+                {
+                    tracing::info!("Failed to put snapshot to s3 at {path}: {:?}", e);
+                    return Err(error::Error::ExecutionErr(format!(
+                        "Failed to put {path} to s3"
+                    )));
+                }
+            } else {
+                return Err(error::Error::BadConfig("Object store is required for snapshot script and is not configured for servers".to_string()));
+            }
+        },
+    )
+}
 
 #[cfg(feature = "parquet")]
 pub async fn attempt_fetch_bytes(
@@ -1246,21 +1236,11 @@ pub fn duckdb_connection_settings_internal(
     return Ok(response);
 }
 
-impl ObjectStoreResource {
-    pub fn get_endpoint_url(&self) -> error::Result<String> {
-        match self {
-            ObjectStoreResource::S3(s3_resource) => Ok(render_endpoint(
-                s3_resource.endpoint.clone(),
-                s3_resource.use_ssl,
-                s3_resource.port,
-                s3_resource.path_style,
-                s3_resource.bucket.clone(),
-            )),
-            ObjectStoreResource::Gcs(gcs_resource) => Ok(format!(
-                "https://storage.googleapis.com/{}",
-                gcs_resource.bucket
-            )),
-            ObjectStoreResource::Azure(az_resource) => az_resource.get_endpoint_url(),
-        }
-    }
+// DuckDB does not parse anything in case of S3 errors and just returns a generic error message.
+// To display better error messages, we cache the errors in a Map<Token, ErrorMessage>
+//
+// We leverage the fact that workers have an internal server to insert the error message
+// from the S3 Proxy, and read it directly in memory from the worker.
+lazy_static::lazy_static! {
+    pub static ref S3_PROXY_LAST_ERRORS_CACHE: Cache<String, String> = Cache::new(4);
 }

backend/windmill-worker/src/duckdb_executor.rs

@@ -10,7 +10,7 @@ use serde_json::value::RawValue;
 use serde_json::{json, Value};
 use uuid::Uuid;
 use windmill_common::error::{to_anyhow, Error, Result};
-use windmill_common::s3_helpers::S3Object;
+use windmill_common::s3_helpers::{S3Object, S3_PROXY_LAST_ERRORS_CACHE};
 use windmill_common::utils::sanitize_string_from_password;
 use windmill_common::worker::Connection;
 use windmill_common::workspaces::{get_ducklake_from_db_unchecked, DucklakeCatalogResourceType};
@@ -128,7 +128,7 @@ pub async fn do_duckdb(
         let base_internal_url = client.base_internal_url.clone();
         let w_id = job.workspace_id.clone();
 
-        let (result, column_order) = tokio::task::spawn_blocking(move || {
+        let result = tokio::task::spawn_blocking(move || {
             run_duckdb_ffi_safe(
                 query_block_list.iter().map(String::as_str),
                 query_block_list.len(),
@@ -139,7 +139,21 @@ pub async fn do_duckdb(
             )
         })
         .await
-        .map_err(to_anyhow)??;
+        .map_err(|e| Error::from(to_anyhow(e)))
+        .and_then(|r| r);
+        let (result, column_order) = match result {
+            Ok(r) => r,
+            Err(e) => {
+                if let Some(s3_proxy_err) = S3_PROXY_LAST_ERRORS_CACHE.get(&client.token) {
+                    return Err(Error::ExecutionErr(format!(
+                        "{}\n\nS3 Related Error: {}",
+                        e.to_string(),
+                        s3_proxy_err,
+                    )));
+                }
+                return Err(e);
+            }
+        };
 
         drop(bigquery_credentials);