From f35c9ffe63cd8a79bbdfe498dbb08e53efd9e5a8 Mon Sep 17 00:00:00 2001 From: Diego Imbert <70353967+diegoimbert@users.noreply.github.com> Date: Fri, 3 Oct 2025 13:06:46 +0200 Subject: [PATCH] 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 | 2 +- backend/windmill-common/src/s3_helpers.rs | 110 +++++++----------- .../windmill-worker/src/duckdb_executor.rs | 20 +++- 3 files changed, 63 insertions(+), 69 deletions(-) diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 537f161c12..1a0dfefd79 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -0835279921261f03720513c0f7ffabc4df44db6f \ No newline at end of file +965976ae4bcf37692b7ce0df6be60e9453c261e3 \ 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 3935593de6..899e3a2fd5 100644 --- a/backend/windmill-common/src/s3_helpers.rs +++ b/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, } -impl AzureBlobResource { - pub fn get_endpoint_url(&self) -> error::Result { - 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 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 { + pub fn from_string(s: &str) -> Option { 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::>(); - 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::>(); + 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 { - 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 +// +// 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 = Cache::new(4); } diff --git a/backend/windmill-worker/src/duckdb_executor.rs b/backend/windmill-worker/src/duckdb_executor.rs index ceb0ccb96e..697fe7ff4d 100644 --- a/backend/windmill-worker/src/duckdb_executor.rs +++ b/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);