From e7bfdbb6beb3608f95f43c538b5705c5bba4f228 Mon Sep 17 00:00:00 2001 From: Harry Papadopoulos Date: Mon, 8 Jun 2026 13:45:20 +0100 Subject: [PATCH 1/9] feat(storage): add named S3-compatible backends via s3+:// URIs - storage.s3.named. config map for arbitrary S3-compatible endpoints (separate creds, region, endpoint, checksum, flags per entry). - URIs of the form s3+://bucket/path route to the named backend; plain s3:// continues to use the primary endpoint. - Preserve the s3+ qualifier through Uri::parse_str normalization so the resolver can recover the backend name on deserialization (previously the qualifier was stripped and named URIs silently resolved to the primary endpoint). - Tests: URI parse/serde roundtrip, NamedS3StorageConfig serde + redact + as_s3_config projection, parse_named_key. - Docs: list s3+:// in the URI protocols and add a Named S3 backends section with a YAML example. Co-Authored-By: Claude Opus 4.7 --- docs/configuration/storage-config.md | 35 +++++ quickwit/quickwit-common/src/uri.rs | 41 +++++- .../quickwit-config/src/storage_config.rs | 133 ++++++++++++++++++ .../s3_compatible_storage_resolver.rs | 58 +++++++- 4 files changed, 262 insertions(+), 5 deletions(-) diff --git a/docs/configuration/storage-config.md b/docs/configuration/storage-config.md index 5c5d24af416..72b1f878da0 100644 --- a/docs/configuration/storage-config.md +++ b/docs/configuration/storage-config.md @@ -15,6 +15,7 @@ Quickwit currently supports four types of storage providers: Storage URIs refer to different storage providers identified by a URI "protocol" or "scheme". Quickwit supports the following storage URI protocols: - `s3://` for Amazon S3 and S3-compatible +- `s3+://` for additional S3-compatible backends configured under `storage.s3.named.` (see [Named S3 backends](#named-s3-backends)) - `azure://` for Azure Blob Storage - `file://` for local file systems - `gs://` for Google Cloud Storage @@ -104,6 +105,40 @@ storage: endpoint: https://storage.googleapis.com ``` +#### Named S3 backends + +In addition to the primary `s3:` block, you can declare any number of additional S3-compatible backends under `storage.s3.named.`. Each entry is an independent endpoint with its own credentials, region, and flags. Indexes route to a named backend via the URI scheme `s3+://bucket/path` (plain `s3://` continues to use the primary endpoint). + +Each named entry accepts the same fields as the primary `s3:` block, *except* `named` itself (no recursion). If `access_key_id` / `secret_access_key` are omitted on a named entry, the global AWS SDK credential chain is used (env vars, instance metadata, etc.). + +```yaml +storage: + s3: + # Primary backend — addressed by plain `s3://...` URIs. + endpoint: https://s3.us-east-1.amazonaws.com + region: us-east-1 + named: + # Addressed by `s3+secondary://bucket/path` URIs. + secondary: + endpoint: https://s3.eu-west-3.amazonaws.com + region: eu-west-3 + access_key_id: ${SECONDARY_S3_ACCESS_KEY_ID} + secret_access_key: ${SECONDARY_S3_SECRET_ACCESS_KEY} + # Addressed by `s3+seaweed://bucket/path` URIs. Falls back to the + # global AWS SDK credentials when keys are omitted. + seaweed: + endpoint: http://seaweedfs-s3:8333 + region: us-east-1 + force_path_style_access: true +``` + +An index pointed at a named backend declares its URI accordingly: + +```yaml +index_id: logs-eu +index_uri: s3+secondary://logs-bucket/logs-eu +``` + ### Azure storage configuration | Property | Description | Default value | diff --git a/quickwit/quickwit-common/src/uri.rs b/quickwit/quickwit-common/src/uri.rs index ff191fb04cb..23d30658614 100644 --- a/quickwit/quickwit-common/src/uri.rs +++ b/quickwit/quickwit-common/src/uri.rs @@ -89,6 +89,9 @@ impl FromStr for Protocol { "ram" => Ok(Protocol::Ram), "s3" => Ok(Protocol::S3), "gs" => Ok(Protocol::Google), + // `s3+://...` for a named S3-compatible backend configured under + // `storage.s3.named.`. Routes to the same factory as `s3://`. + s if s.starts_with("s3+") && s.len() > 3 => Ok(Protocol::S3), _ => bail!("unknown URI protocol `{protocol}`"), } } @@ -262,9 +265,13 @@ impl Uri { if uri_str.is_empty() { bail!("failed to parse empty URI"); } - let (protocol, mut path) = match uri_str.split_once(PROTOCOL_SEPARATOR) { - None => (Protocol::File, uri_str.to_string()), - Some((protocol, path)) => (Protocol::from_str(protocol)?, path.to_string()), + let (scheme_opt, protocol, mut path) = match uri_str.split_once(PROTOCOL_SEPARATOR) { + None => (None, Protocol::File, uri_str.to_string()), + Some((scheme, path)) => ( + Some(scheme.to_string()), + Protocol::from_str(scheme)?, + path.to_string(), + ), }; if protocol == Protocol::File { if path.starts_with('~') { @@ -292,8 +299,14 @@ impl Uri { .to_string_lossy() .to_string(); } + // Preserve `s3+` qualifier so the storage resolver can route to + // the named backend; other schemes normalize to canonical form. + let display_scheme: &str = match scheme_opt.as_deref() { + Some(s) if s.starts_with("s3+") => s, + _ => protocol.as_str(), + }; Ok(Self { - uri: format!("{protocol}{PROTOCOL_SEPARATOR}{path}"), + uri: format!("{display_scheme}{PROTOCOL_SEPARATOR}{path}"), protocol, }) } @@ -812,4 +825,24 @@ mod tests { serde_json::Value::String("s3://bucket/key".to_string()) ); } + + #[test] + fn test_uri_s3_named_preserved() { + // The `s3+` qualifier is the routing token for named S3-compatible + // backends (`storage.s3.named.`). It must survive parse + serialize + // so the storage resolver can recover the backend name on deserialization; + // before this guarantee, the qualifier was stripped by URI normalization + // and every `s3+://` URI silently resolved to the primary endpoint. + let uri = Uri::from_str("s3+alt://bucket/key").unwrap(); + assert_eq!(uri.protocol(), Protocol::S3); + assert_eq!(uri.as_str(), "s3+alt://bucket/key"); + let json = serde_json::to_value(&uri).unwrap(); + assert_eq!( + json, + serde_json::Value::String("s3+alt://bucket/key".to_string()) + ); + let round_trip: Uri = serde_json::from_value(json).unwrap(); + assert_eq!(round_trip.as_str(), "s3+alt://bucket/key"); + assert_eq!(round_trip.protocol(), Protocol::S3); + } } diff --git a/quickwit/quickwit-config/src/storage_config.rs b/quickwit/quickwit-config/src/storage_config.rs index d04cd93aaa0..98dadf734f7 100644 --- a/quickwit/quickwit-config/src/storage_config.rs +++ b/quickwit/quickwit-config/src/storage_config.rs @@ -362,6 +362,87 @@ pub struct S3StorageConfig { pub disable_stalled_stream_protection_upload: bool, #[serde(default)] pub disable_stalled_stream_protection_download: bool, + /// Additional named S3-compatible backends, addressed via `s3+://bucket/path` + /// URIs. Each entry is an independent endpoint with its own credentials, region, + /// etc. The map key (``) is the routing token used in the URI scheme. + #[serde(default)] + #[serde(skip_serializing_if = "std::collections::BTreeMap::is_empty")] + pub named: std::collections::BTreeMap, +} + +/// Configuration for a named S3-compatible backend nested under +/// `storage.s3.named.`. Mirrors `S3StorageConfig` but cannot itself +/// have a `named` field (no recursion). +#[derive(Clone, Default, Eq, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct NamedS3StorageConfig { + #[serde(default)] + #[serde(skip_serializing_if = "Option::is_none")] + pub flavor: Option, + #[serde(default)] + pub access_key_id: Option, + #[serde(default)] + pub secret_access_key: Option, + #[serde(default)] + pub region: Option, + #[serde(default)] + pub endpoint: Option, + #[serde(default)] + pub force_path_style_access: bool, + #[serde(default)] + pub disable_multi_object_delete: bool, + #[serde(default)] + pub disable_multipart_upload: bool, + #[serde(default)] + pub checksum_algorithm: ChecksumAlgorithm, +} + +impl NamedS3StorageConfig { + /// Project this named config back into a full `S3StorageConfig` + /// (with an empty `named` map) so it can flow through the existing + /// S3 client construction code unchanged. + pub fn as_s3_config(&self) -> S3StorageConfig { + S3StorageConfig { + flavor: self.flavor, + access_key_id: self.access_key_id.clone(), + secret_access_key: self.secret_access_key.clone(), + region: self.region.clone(), + endpoint: self.endpoint.clone(), + force_path_style_access: self.force_path_style_access, + disable_multi_object_delete: self.disable_multi_object_delete, + disable_multipart_upload: self.disable_multipart_upload, + checksum_algorithm: self.checksum_algorithm, + disable_checksums: false, + disable_stalled_stream_protection_upload: false, + disable_stalled_stream_protection_download: false, + named: Default::default(), + } + } + + pub fn redact(&mut self) { + if let Some(secret_access_key) = self.secret_access_key.as_mut() { + *secret_access_key = "***redacted***".to_string(); + } + } +} + +impl fmt::Debug for NamedS3StorageConfig { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("NamedS3StorageConfig") + .field("flavor", &self.flavor) + .field("access_key_id", &self.access_key_id) + .field( + "secret_access_key", + &self.secret_access_key.as_ref().map(|_| "***redacted***"), + ) + .field("region", &self.region) + .field("endpoint", &self.endpoint) + .field("force_path_style_access", &self.force_path_style_access) + .field("disable_multi_object_delete", &self.disable_multi_object_delete) + .field("disable_multipart_upload", &self.disable_multipart_upload) + .field("checksum_algorithm", &self.checksum_algorithm) + .finish() + } } impl S3StorageConfig { @@ -397,6 +478,9 @@ impl S3StorageConfig { if let Some(secret_access_key) = self.secret_access_key.as_mut() { *secret_access_key = "***redacted***".to_string(); } + for named_config in self.named.values_mut() { + named_config.redact(); + } } pub fn endpoint(&self) -> Option { @@ -442,6 +526,7 @@ impl fmt::Debug for S3StorageConfig { "disable_stalled_stream_protection_download", &self.disable_stalled_stream_protection_download, ) + .field("named", &self.named) .finish() } } @@ -740,4 +825,52 @@ mod tests { assert_eq!(s3_storage_config.flavor, Some(StorageBackendFlavor::MinIO)); } } + + #[test] + fn test_storage_s3_named_backends_serde() { + let s3_storage_config_yaml = r#" + endpoint: https://primary.example.com + region: us-east-1 + named: + alt: + endpoint: https://alt.example.com + region: eu-west-3 + force_path_style_access: true + access_key_id: alt-key + secret_access_key: alt-secret + secondary: + endpoint: http://secondary.example.com:8333 + region: us-east-1 + force_path_style_access: true + "#; + let s3_storage_config: S3StorageConfig = + serde_yaml::from_str(s3_storage_config_yaml).unwrap(); + assert_eq!(s3_storage_config.named.len(), 2); + + let alt = s3_storage_config.named.get("alt").unwrap(); + assert_eq!(alt.region.as_deref(), Some("eu-west-3")); + assert_eq!(alt.access_key_id.as_deref(), Some("alt-key")); + assert!(alt.force_path_style_access); + + // `as_s3_config` projects a named entry back into a full S3StorageConfig + // (with an empty `named` map) so it can drive the S3 client builder + // unchanged. + let projected = alt.as_s3_config(); + assert_eq!(projected.region.as_deref(), Some("eu-west-3")); + assert_eq!(projected.access_key_id.as_deref(), Some("alt-key")); + assert!(projected.force_path_style_access); + assert!(projected.named.is_empty()); + } + + #[test] + fn test_storage_s3_named_backends_redact() { + let mut named = NamedS3StorageConfig { + access_key_id: Some("public-key".to_string()), + secret_access_key: Some("super-secret".to_string()), + ..Default::default() + }; + named.redact(); + assert_eq!(named.access_key_id.as_deref(), Some("public-key")); + assert_eq!(named.secret_access_key.as_deref(), Some("***redacted***")); + } } diff --git a/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage_resolver.rs b/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage_resolver.rs index 92b4406486b..70e4d5159fc 100644 --- a/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage_resolver.rs +++ b/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage_resolver.rs @@ -12,19 +12,28 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::collections::HashMap; use std::sync::Arc; use async_trait::async_trait; use aws_sdk_s3::Client as S3Client; use quickwit_common::uri::Uri; use quickwit_config::{S3StorageConfig, StorageBackend}; -use tokio::sync::OnceCell; +use tokio::sync::{Mutex, OnceCell}; use super::s3_compatible_storage::create_s3_client; use crate::{ DebouncedStorage, S3CompatibleObjectStorage, Storage, StorageFactory, StorageResolverError, }; +/// Extracts the named-backend key out of an `s3+://...` URI, if any. +/// Returns `None` for plain `s3://...`. +fn parse_named_key(uri: &Uri) -> Option<&str> { + let scheme_end = uri.as_str().find("://")?; + let scheme = &uri.as_str()[..scheme_end]; + scheme.strip_prefix("s3+") +} + /// S3 compatible object storage resolver. pub struct S3CompatibleObjectStorageFactory { storage_config: S3StorageConfig, @@ -34,6 +43,8 @@ pub struct S3CompatibleObjectStorageFactory { // end up being used, or if something like azure, gcs, or even local files, will be used // instead. s3_client: OnceCell, + // One cached S3Client per named backend. Built lazily on first use. + named_s3_clients: Mutex>, } impl S3CompatibleObjectStorageFactory { @@ -42,6 +53,7 @@ impl S3CompatibleObjectStorageFactory { Self { storage_config, s3_client: OnceCell::new(), + named_s3_clients: Mutex::new(HashMap::new()), } } } @@ -53,6 +65,30 @@ impl StorageFactory for S3CompatibleObjectStorageFactory { } async fn resolve(&self, uri: &Uri) -> Result, StorageResolverError> { + if let Some(name) = parse_named_key(uri) { + let named_config = self + .storage_config + .named + .get(name) + .ok_or_else(|| { + StorageResolverError::InvalidUri(format!( + "no `storage.s3.named.{name}` entry configured for URI `{uri}`" + )) + })? + .as_s3_config(); + let mut clients = self.named_s3_clients.lock().await; + let client = if let Some(client) = clients.get(name) { + client.clone() + } else { + let client = create_s3_client(&named_config).await; + clients.insert(name.to_string(), client.clone()); + client + }; + drop(clients); + let storage = + S3CompatibleObjectStorage::from_uri_and_client(&named_config, uri, client).await?; + return Ok(Arc::new(DebouncedStorage::new(storage))); + } let s3_client = self .s3_client .get_or_init(|| create_s3_client(&self.storage_config)) @@ -64,3 +100,23 @@ impl StorageFactory for S3CompatibleObjectStorageFactory { Ok(Arc::new(DebouncedStorage::new(storage))) } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_parse_named_key() { + // Plain s3:// URIs route through the primary backend. + assert_eq!(parse_named_key(&Uri::for_test("s3://bucket/key")), None); + // `s3+` URIs return the named-backend key. + assert_eq!( + parse_named_key(&Uri::for_test("s3+alt://bucket/key")), + Some("alt") + ); + assert_eq!( + parse_named_key(&Uri::for_test("s3+with-dash://bucket/key")), + Some("with-dash") + ); + } +} From 886ef31f02d6215907389aa9c0c94a2d3663cd32 Mon Sep 17 00:00:00 2001 From: Harry Papadopoulos Date: Wed, 17 Jun 2026 09:52:12 +0100 Subject: [PATCH 2/9] fix(storage): address review findings for named S3 backends Six issues from the Codex review of the s3+ named-backend feature: - uri: slice path at the real `://` separator and preserve the `s3+` scheme in parent()/file_name() (was corrupting paths to `t://bucket`). - config: env overrides (QW_S3_ENDPOINT, QW_S3_FORCE_PATH_STYLE_ACCESS) now apply to the primary backend only; named backends are self-contained. Removed the process-wide OnceLock that cached the first client's path-style. - config: as_s3_config() applies flavor expansion for named backends. - config: accept the disable_multi_object_delete_requests alias on named. - config: accept and pass through disable_checksums and the stalled-stream protection toggles on named backends (were denied and hard-coded false). Adds unit tests for URI scheme preservation, field parity, flavor expansion, and named-backend endpoint isolation. Updates storage-config docs. --- docs/configuration/storage-config.md | 2 + quickwit/quickwit-common/src/uri.rs | 44 +++++- .../quickwit-config/src/storage_config.rs | 146 +++++++++++++++--- 3 files changed, 172 insertions(+), 20 deletions(-) diff --git a/docs/configuration/storage-config.md b/docs/configuration/storage-config.md index 72b1f878da0..2366462d88a 100644 --- a/docs/configuration/storage-config.md +++ b/docs/configuration/storage-config.md @@ -111,6 +111,8 @@ In addition to the primary `s3:` block, you can declare any number of additional Each named entry accepts the same fields as the primary `s3:` block, *except* `named` itself (no recursion). If `access_key_id` / `secret_access_key` are omitted on a named entry, the global AWS SDK credential chain is used (env vars, instance metadata, etc.). +Named backends are self-contained: the process-wide `QW_S3_ENDPOINT` and `QW_S3_FORCE_PATH_STYLE_ACCESS` overrides apply to the primary `s3:` backend only. A named backend always uses its own `endpoint` and `force_path_style_access` values. + ```yaml storage: s3: diff --git a/quickwit/quickwit-common/src/uri.rs b/quickwit/quickwit-common/src/uri.rs index 23d30658614..3200ba41cbe 100644 --- a/quickwit/quickwit-common/src/uri.rs +++ b/quickwit/quickwit-common/src/uri.rs @@ -189,13 +189,33 @@ impl Uri { let parent_path = path.parent()?; Some(Self { - uri: format!("{protocol}{PROTOCOL_SEPARATOR}{}", parent_path.display()), + // Preserve the scheme verbatim so an `s3+` qualifier survives. + uri: format!( + "{}{PROTOCOL_SEPARATOR}{}", + self.scheme(), + parent_path.display() + ), protocol, }) } + /// Returns the URI scheme, preserving any `s3+` qualifier, which may + /// differ from the canonical protocol string (e.g. `s3+alt` vs `s3`). + fn scheme(&self) -> &str { + match self.uri.split_once(PROTOCOL_SEPARATOR) { + Some((scheme, _path)) => scheme, + None => self.protocol.as_str(), + } + } + fn path(&self) -> &Path { - Path::new(&self.uri[self.protocol.as_str().len() + PROTOCOL_SEPARATOR.len()..]) + // Slice at the actual `://` separator rather than assuming the scheme + // equals the canonical protocol — `s3+` schemes are longer. + let path = match self.uri.split_once(PROTOCOL_SEPARATOR) { + Some((_scheme, path)) => path, + None => &self.uri, + }; + Path::new(path) } /// Returns the last component of the URI. @@ -676,6 +696,26 @@ mod tests { ); } + #[test] + fn test_uri_named_s3_scheme() { + // `s3+` schemes are preserved end-to-end: `path` strips the real + // scheme (not the canonical `s3`), and `parent`/`file_name` keep the + // qualifier intact. + let uri = Uri::for_test("s3+alt://bucket/foo/bar"); + assert_eq!(uri.as_str(), "s3+alt://bucket/foo/bar"); + assert_eq!(uri.protocol(), Protocol::S3); + assert_eq!(uri.parent().unwrap(), "s3+alt://bucket/foo"); + assert_eq!(uri.file_name().unwrap(), Path::new("bar")); + + let uri = Uri::for_test("s3+with-dash://bucket/key"); + assert_eq!(uri.parent().unwrap(), "s3+with-dash://bucket"); + assert_eq!(uri.file_name().unwrap(), Path::new("key")); + + // Mirrors the plain-`s3` guard: a bucket-only URI has no parent. + assert!(Uri::for_test("s3+alt://bucket").parent().is_none()); + assert!(Uri::for_test("s3+alt://bucket/").parent().is_none()); + } + #[test] fn test_uri_file_name() { assert!(Uri::for_test("file:///").file_name().is_none()); diff --git a/quickwit/quickwit-config/src/storage_config.rs b/quickwit/quickwit-config/src/storage_config.rs index 98dadf734f7..c6e3e0ab16a 100644 --- a/quickwit/quickwit-config/src/storage_config.rs +++ b/quickwit/quickwit-config/src/storage_config.rs @@ -13,7 +13,6 @@ // limitations under the License. use std::ops::Deref; -use std::sync::OnceLock; use std::{env, fmt}; use anyhow::ensure; @@ -368,6 +367,12 @@ pub struct S3StorageConfig { #[serde(default)] #[serde(skip_serializing_if = "std::collections::BTreeMap::is_empty")] pub named: std::collections::BTreeMap, + /// Set when this config is the projection of a named backend. Named + /// backends are self-contained, so the process-wide `QW_S3_ENDPOINT` / + /// `QW_S3_FORCE_PATH_STYLE_ACCESS` overrides apply to the primary backend + /// only. Not serialized; defaults to `false` (the primary backend). + #[serde(skip)] + pub is_named_backend: bool, } /// Configuration for a named S3-compatible backend nested under @@ -389,12 +394,20 @@ pub struct NamedS3StorageConfig { pub endpoint: Option, #[serde(default)] pub force_path_style_access: bool, + #[serde(alias = "disable_multi_object_delete_requests")] #[serde(default)] pub disable_multi_object_delete: bool, #[serde(default)] pub disable_multipart_upload: bool, #[serde(default)] pub checksum_algorithm: ChecksumAlgorithm, + /// Deprecated: applies into `checksum_algorithm: disabled`. + #[serde(default, skip_serializing)] + pub disable_checksums: bool, + #[serde(default)] + pub disable_stalled_stream_protection_upload: bool, + #[serde(default)] + pub disable_stalled_stream_protection_download: bool, } impl NamedS3StorageConfig { @@ -402,7 +415,7 @@ impl NamedS3StorageConfig { /// (with an empty `named` map) so it can flow through the existing /// S3 client construction code unchanged. pub fn as_s3_config(&self) -> S3StorageConfig { - S3StorageConfig { + let mut s3_config = S3StorageConfig { flavor: self.flavor, access_key_id: self.access_key_id.clone(), secret_access_key: self.secret_access_key.clone(), @@ -412,11 +425,17 @@ impl NamedS3StorageConfig { disable_multi_object_delete: self.disable_multi_object_delete, disable_multipart_upload: self.disable_multipart_upload, checksum_algorithm: self.checksum_algorithm, - disable_checksums: false, - disable_stalled_stream_protection_upload: false, - disable_stalled_stream_protection_download: false, + disable_checksums: self.disable_checksums, + disable_stalled_stream_protection_upload: self.disable_stalled_stream_protection_upload, + disable_stalled_stream_protection_download: self + .disable_stalled_stream_protection_download, named: Default::default(), - } + is_named_backend: true, + }; + // Expand `flavor` shortcuts (region/path-style/checksum defaults) the + // same way the primary backend does at config load time. + s3_config.apply_flavor(); + s3_config } pub fn redact(&mut self) { @@ -438,9 +457,20 @@ impl fmt::Debug for NamedS3StorageConfig { .field("region", &self.region) .field("endpoint", &self.endpoint) .field("force_path_style_access", &self.force_path_style_access) - .field("disable_multi_object_delete", &self.disable_multi_object_delete) + .field( + "disable_multi_object_delete", + &self.disable_multi_object_delete, + ) .field("disable_multipart_upload", &self.disable_multipart_upload) .field("checksum_algorithm", &self.checksum_algorithm) + .field( + "disable_stalled_stream_protection_upload", + &self.disable_stalled_stream_protection_upload, + ) + .field( + "disable_stalled_stream_protection_download", + &self.disable_stalled_stream_protection_download, + ) .finish() } } @@ -484,20 +514,26 @@ impl S3StorageConfig { } pub fn endpoint(&self) -> Option { - env::var("QW_S3_ENDPOINT") - .ok() - .or_else(|| self.endpoint.clone()) + // `QW_S3_ENDPOINT` overrides the primary backend only; named backends + // are self-contained and use their own configured endpoint. + if !self.is_named_backend + && let Ok(endpoint) = env::var("QW_S3_ENDPOINT") + { + return Some(endpoint); + } + self.endpoint.clone() } pub fn force_path_style_access(&self) -> Option { - static FORCE_PATH_STYLE: OnceLock> = OnceLock::new(); - *FORCE_PATH_STYLE.get_or_init(|| { - let force_path_style_access = get_bool_from_env( - "QW_S3_FORCE_PATH_STYLE_ACCESS", - self.force_path_style_access, - ); - Some(force_path_style_access) - }) + // `QW_S3_FORCE_PATH_STYLE_ACCESS` overrides the primary backend only. + // No process-wide cache: each backend must honor its own setting. + if self.is_named_backend { + return Some(self.force_path_style_access); + } + Some(get_bool_from_env( + "QW_S3_FORCE_PATH_STYLE_ACCESS", + self.force_path_style_access, + )) } } @@ -873,4 +909,78 @@ mod tests { assert_eq!(named.access_key_id.as_deref(), Some("public-key")); assert_eq!(named.secret_access_key.as_deref(), Some("***redacted***")); } + + #[test] + fn test_storage_s3_named_backends_field_parity() { + // Named backends accept the same fields as the primary S3 block, + // including the legacy `disable_multi_object_delete_requests` alias and + // the stalled-stream toggles, and project them through `as_s3_config`. + let s3_storage_config_yaml = r#" + named: + alt: + endpoint: https://alt.example.com + disable_multi_object_delete_requests: true + disable_stalled_stream_protection_upload: true + disable_stalled_stream_protection_download: true + checksum_algorithm: disabled + "#; + let s3_storage_config: S3StorageConfig = + serde_yaml::from_str(s3_storage_config_yaml).unwrap(); + let alt = s3_storage_config.named.get("alt").unwrap(); + assert!(alt.disable_multi_object_delete); + assert!(alt.disable_stalled_stream_protection_upload); + assert!(alt.disable_stalled_stream_protection_download); + + let projected = alt.as_s3_config(); + assert!(projected.is_named_backend); + assert!(projected.disable_multi_object_delete); + assert!(projected.disable_stalled_stream_protection_upload); + assert!(projected.disable_stalled_stream_protection_download); + assert_eq!(projected.checksum_algorithm, ChecksumAlgorithm::Disabled); + + // A genuinely unknown field is still rejected. + let invalid_yaml = r#" + named: + alt: + bogus_field: true + "#; + assert!(serde_yaml::from_str::(invalid_yaml).is_err()); + } + + #[test] + fn test_storage_s3_named_backend_applies_flavor() { + // `flavor` shortcuts expand for named backends just like the primary. + let s3_storage_config_yaml = r#" + named: + minio-backend: + flavor: minio + endpoint: http://minio.example.com:9000 + "#; + let s3_storage_config: S3StorageConfig = + serde_yaml::from_str(s3_storage_config_yaml).unwrap(); + let projected = s3_storage_config + .named + .get("minio-backend") + .unwrap() + .as_s3_config(); + assert_eq!(projected.region.as_deref(), Some("minio")); + assert!(projected.force_path_style_access); + } + + #[test] + fn test_storage_s3_named_backend_uses_own_endpoint() { + // A named backend is self-contained: `endpoint()` returns its configured + // endpoint regardless of the process-wide `QW_S3_ENDPOINT` override, + // which applies to the primary backend only. + let named = NamedS3StorageConfig { + endpoint: Some("https://named.example.com".to_string()), + ..Default::default() + }; + let projected = named.as_s3_config(); + assert!(projected.is_named_backend); + assert_eq!( + projected.endpoint(), + Some("https://named.example.com".to_string()) + ); + } } From 2d499ce4dfe5b657be318379b81baf5120541fc9 Mon Sep 17 00:00:00 2001 From: Harry Papadopoulos Date: Fri, 19 Jun 2026 14:35:48 +0100 Subject: [PATCH 3/9] fix(storage): validate named S3 backend names and avoid lock-across-await - config: reject URL-incompatible named-backend names at config load; the name is embedded in the s3+:// scheme which the metrics/DataFusion path parses with url::Url. Restricted to lowercase ASCII alnum and `-`. - storage: cache each named backend behind its own OnceCell; the mutex is held only synchronously to fetch/insert the cell, never across the client build, so independent backends initialize concurrently. --- .../quickwit-config/src/storage_config.rs | 44 +++++++++++++++++++ .../s3_compatible_storage_resolver.rs | 26 ++++++----- 2 files changed, 58 insertions(+), 12 deletions(-) diff --git a/quickwit/quickwit-config/src/storage_config.rs b/quickwit/quickwit-config/src/storage_config.rs index c6e3e0ab16a..707aaad11a9 100644 --- a/quickwit/quickwit-config/src/storage_config.rs +++ b/quickwit/quickwit-config/src/storage_config.rs @@ -91,6 +91,19 @@ pub enum StorageBackendFlavor { #[derive(Debug, Clone, Default, Eq, PartialEq, Serialize, Deserialize)] pub struct StorageConfigs(#[serde_as(as = "EnumMap")] Vec); +/// Named backend names must be valid lowercase URL-scheme tails (`s3+://`). +fn validate_named_s3_backend_name(name: &str) -> anyhow::Result<()> { + ensure!(!name.is_empty(), "named S3 backend name must not be empty"); + for character in name.chars() { + ensure!( + character.is_ascii_lowercase() || character.is_ascii_digit() || character == '-', + "invalid named S3 backend name `{name}`: only lowercase ASCII letters, digits, and \ + `-` are allowed (the name is used in the `s3+{name}://` URI scheme)" + ); + } + Ok(()) +} + impl StorageConfigs { pub fn new(storage_configs: Vec) -> Self { Self(storage_configs) @@ -124,6 +137,13 @@ impl StorageConfigs { "{left:?} storage config is defined multiple times", ); } + for storage_config in self.0.iter() { + if let StorageConfig::S3(s3_storage_config) = storage_config { + for name in s3_storage_config.named.keys() { + validate_named_s3_backend_name(name)?; + } + } + } Ok(()) } @@ -983,4 +1003,28 @@ mod tests { Some("https://named.example.com".to_string()) ); } + + #[test] + fn test_validate_named_s3_backend_name() { + for valid in ["alt", "seaweedfs", "ovh-morocco", "s3alt", "minio-backend"] { + validate_named_s3_backend_name(valid).unwrap(); + } + for invalid in ["", "prod_logs", "Prod", "a.b", "a/b", "a b"] { + validate_named_s3_backend_name(invalid).unwrap_err(); + } + } + + #[test] + fn test_storage_configs_reject_url_incompatible_named_backend() { + let s3_storage_config_yaml = r#" + named: + prod_logs: + endpoint: https://logs.example.com + "#; + let s3_storage_config: S3StorageConfig = + serde_yaml::from_str(s3_storage_config_yaml).unwrap(); + let storage_configs = StorageConfigs::new(vec![s3_storage_config.into()]); + let error = storage_configs.validate().unwrap_err().to_string(); + assert!(error.contains("prod_logs"), "unexpected error: {error}"); + } } diff --git a/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage_resolver.rs b/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage_resolver.rs index 70e4d5159fc..1ef8357671a 100644 --- a/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage_resolver.rs +++ b/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage_resolver.rs @@ -13,13 +13,13 @@ // limitations under the License. use std::collections::HashMap; -use std::sync::Arc; +use std::sync::{Arc, Mutex}; use async_trait::async_trait; use aws_sdk_s3::Client as S3Client; use quickwit_common::uri::Uri; use quickwit_config::{S3StorageConfig, StorageBackend}; -use tokio::sync::{Mutex, OnceCell}; +use tokio::sync::OnceCell; use super::s3_compatible_storage::create_s3_client; use crate::{ @@ -43,8 +43,8 @@ pub struct S3CompatibleObjectStorageFactory { // end up being used, or if something like azure, gcs, or even local files, will be used // instead. s3_client: OnceCell, - // One cached S3Client per named backend. Built lazily on first use. - named_s3_clients: Mutex>, + // Per-named-backend client cell; the mutex is held only to fetch/insert the cell, never across the build. + named_s3_clients: Mutex>>>, } impl S3CompatibleObjectStorageFactory { @@ -76,15 +76,17 @@ impl StorageFactory for S3CompatibleObjectStorageFactory { )) })? .as_s3_config(); - let mut clients = self.named_s3_clients.lock().await; - let client = if let Some(client) = clients.get(name) { - client.clone() - } else { - let client = create_s3_client(&named_config).await; - clients.insert(name.to_string(), client.clone()); - client + let client_cell = { + let mut clients = self + .named_s3_clients + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + Arc::clone(clients.entry(name.to_string()).or_default()) }; - drop(clients); + let client = client_cell + .get_or_init(|| create_s3_client(&named_config)) + .await + .clone(); let storage = S3CompatibleObjectStorage::from_uri_and_client(&named_config, uri, client).await?; return Ok(Arc::new(DebouncedStorage::new(storage))); From 2fb35cf61862921e9914120e398414726e9f7d75 Mon Sep 17 00:00:00 2001 From: Harry Papadopoulos Date: Tue, 30 Jun 2026 09:09:51 +0100 Subject: [PATCH 4/9] refactor(storage): route S3 by per-bucket profile instead of s3+ scheme Replaces the named-backend design (which embedded a config label in the `s3+://` URI scheme, persisting it into the metastore) with per-bucket profiles keyed by bucket name under `storage.s3.profiles.`. URIs stay canonical `s3://bucket/path`, so nothing synthetic is persisted: routing is by bucket name, resolved at runtime from node-local config. - quickwit-common/uri.rs: revert all s3+ scheme handling to upstream; the Uri type no longer needs to carry a non-canonical scheme. - quickwit-config: `named` -> `profiles` (keyed by bucket), `NamedS3StorageConfig` -> `S3ProfileConfig`, `is_named_backend` -> `is_profile`. Drop the URL-scheme name validation (a bucket name is not a scheme). - quickwit-storage: resolver extracts the bucket via parse_s3_uri and looks it up in `profiles`; match -> that profile's config + per-bucket client cache, no match -> primary backend. - docs: rewrite as "Per-bucket S3 profiles", including the bucket-name uniqueness caveat (a bucket name maps to exactly one backend). Validated end-to-end through the REST stack: ingest -> split -> upload -> publish -> search against in-cluster SeaweedFS and external OVH backends. Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/configuration/storage-config.md | 31 +-- quickwit/quickwit-common/src/uri.rs | 85 +------- .../quickwit-config/src/storage_config.rs | 203 ++++++++---------- .../s3_compatible_storage_resolver.rs | 63 ++---- 4 files changed, 126 insertions(+), 256 deletions(-) diff --git a/docs/configuration/storage-config.md b/docs/configuration/storage-config.md index 2366462d88a..4a9c163fd28 100644 --- a/docs/configuration/storage-config.md +++ b/docs/configuration/storage-config.md @@ -14,8 +14,7 @@ Quickwit currently supports four types of storage providers: ## Storage URIs Storage URIs refer to different storage providers identified by a URI "protocol" or "scheme". Quickwit supports the following storage URI protocols: -- `s3://` for Amazon S3 and S3-compatible -- `s3+://` for additional S3-compatible backends configured under `storage.s3.named.` (see [Named S3 backends](#named-s3-backends)) +- `s3://` for Amazon S3 and S3-compatible (per-bucket overrides via `storage.s3.profiles`, see [Per-bucket S3 profiles](#per-bucket-s3-profiles)) - `azure://` for Azure Blob Storage - `file://` for local file systems - `gs://` for Google Cloud Storage @@ -105,40 +104,42 @@ storage: endpoint: https://storage.googleapis.com ``` -#### Named S3 backends +#### Per-bucket S3 profiles -In addition to the primary `s3:` block, you can declare any number of additional S3-compatible backends under `storage.s3.named.`. Each entry is an independent endpoint with its own credentials, region, and flags. Indexes route to a named backend via the URI scheme `s3+://bucket/path` (plain `s3://` continues to use the primary endpoint). +In addition to the primary `s3:` block, you can declare per-bucket overrides under `storage.s3.profiles`. The map key is the bucket name; when an `s3:///...` URI is resolved, an exact match supplies that bucket's own endpoint, credentials, region, and flags. Any bucket not listed falls back to the fields on the primary `s3:` block. URIs stay canonical `s3://` — routing is by bucket name, so nothing extra is persisted in the index metadata. -Each named entry accepts the same fields as the primary `s3:` block, *except* `named` itself (no recursion). If `access_key_id` / `secret_access_key` are omitted on a named entry, the global AWS SDK credential chain is used (env vars, instance metadata, etc.). +Each profile accepts the same fields as the primary `s3:` block, *except* `profiles` itself (no recursion). If `access_key_id` / `secret_access_key` are omitted on a profile, the global AWS SDK credential chain is used (env vars, instance metadata, etc.). -Named backends are self-contained: the process-wide `QW_S3_ENDPOINT` and `QW_S3_FORCE_PATH_STYLE_ACCESS` overrides apply to the primary `s3:` backend only. A named backend always uses its own `endpoint` and `force_path_style_access` values. +Profiles are self-contained: the process-wide `QW_S3_ENDPOINT` and `QW_S3_FORCE_PATH_STYLE_ACCESS` overrides apply to the primary `s3:` backend only. A profile always uses its own `endpoint` and `force_path_style_access` values. + +> Because routing keys on bucket name, a given bucket name maps to exactly one backend. If you need the *same* bucket name on two different endpoints, give the buckets distinct names. ```yaml storage: s3: - # Primary backend — addressed by plain `s3://...` URIs. + # Primary backend — used for any bucket not listed under `profiles`. endpoint: https://s3.us-east-1.amazonaws.com region: us-east-1 - named: - # Addressed by `s3+secondary://bucket/path` URIs. - secondary: + profiles: + # Buckets named `logs-bucket-eu` resolve to this endpoint. + logs-bucket-eu: endpoint: https://s3.eu-west-3.amazonaws.com region: eu-west-3 access_key_id: ${SECONDARY_S3_ACCESS_KEY_ID} secret_access_key: ${SECONDARY_S3_SECRET_ACCESS_KEY} - # Addressed by `s3+seaweed://bucket/path` URIs. Falls back to the - # global AWS SDK credentials when keys are omitted. - seaweed: + # Buckets named `seaweed-logs` resolve here. Falls back to the global AWS + # SDK credentials when keys are omitted. + seaweed-logs: endpoint: http://seaweedfs-s3:8333 region: us-east-1 force_path_style_access: true ``` -An index pointed at a named backend declares its URI accordingly: +An index simply points at the bucket by its canonical URI; the matching profile is applied automatically: ```yaml index_id: logs-eu -index_uri: s3+secondary://logs-bucket/logs-eu +index_uri: s3://logs-bucket-eu/logs-eu ``` ### Azure storage configuration diff --git a/quickwit/quickwit-common/src/uri.rs b/quickwit/quickwit-common/src/uri.rs index 3200ba41cbe..ff191fb04cb 100644 --- a/quickwit/quickwit-common/src/uri.rs +++ b/quickwit/quickwit-common/src/uri.rs @@ -89,9 +89,6 @@ impl FromStr for Protocol { "ram" => Ok(Protocol::Ram), "s3" => Ok(Protocol::S3), "gs" => Ok(Protocol::Google), - // `s3+://...` for a named S3-compatible backend configured under - // `storage.s3.named.`. Routes to the same factory as `s3://`. - s if s.starts_with("s3+") && s.len() > 3 => Ok(Protocol::S3), _ => bail!("unknown URI protocol `{protocol}`"), } } @@ -189,33 +186,13 @@ impl Uri { let parent_path = path.parent()?; Some(Self { - // Preserve the scheme verbatim so an `s3+` qualifier survives. - uri: format!( - "{}{PROTOCOL_SEPARATOR}{}", - self.scheme(), - parent_path.display() - ), + uri: format!("{protocol}{PROTOCOL_SEPARATOR}{}", parent_path.display()), protocol, }) } - /// Returns the URI scheme, preserving any `s3+` qualifier, which may - /// differ from the canonical protocol string (e.g. `s3+alt` vs `s3`). - fn scheme(&self) -> &str { - match self.uri.split_once(PROTOCOL_SEPARATOR) { - Some((scheme, _path)) => scheme, - None => self.protocol.as_str(), - } - } - fn path(&self) -> &Path { - // Slice at the actual `://` separator rather than assuming the scheme - // equals the canonical protocol — `s3+` schemes are longer. - let path = match self.uri.split_once(PROTOCOL_SEPARATOR) { - Some((_scheme, path)) => path, - None => &self.uri, - }; - Path::new(path) + Path::new(&self.uri[self.protocol.as_str().len() + PROTOCOL_SEPARATOR.len()..]) } /// Returns the last component of the URI. @@ -285,13 +262,9 @@ impl Uri { if uri_str.is_empty() { bail!("failed to parse empty URI"); } - let (scheme_opt, protocol, mut path) = match uri_str.split_once(PROTOCOL_SEPARATOR) { - None => (None, Protocol::File, uri_str.to_string()), - Some((scheme, path)) => ( - Some(scheme.to_string()), - Protocol::from_str(scheme)?, - path.to_string(), - ), + let (protocol, mut path) = match uri_str.split_once(PROTOCOL_SEPARATOR) { + None => (Protocol::File, uri_str.to_string()), + Some((protocol, path)) => (Protocol::from_str(protocol)?, path.to_string()), }; if protocol == Protocol::File { if path.starts_with('~') { @@ -319,14 +292,8 @@ impl Uri { .to_string_lossy() .to_string(); } - // Preserve `s3+` qualifier so the storage resolver can route to - // the named backend; other schemes normalize to canonical form. - let display_scheme: &str = match scheme_opt.as_deref() { - Some(s) if s.starts_with("s3+") => s, - _ => protocol.as_str(), - }; Ok(Self { - uri: format!("{display_scheme}{PROTOCOL_SEPARATOR}{path}"), + uri: format!("{protocol}{PROTOCOL_SEPARATOR}{path}"), protocol, }) } @@ -696,26 +663,6 @@ mod tests { ); } - #[test] - fn test_uri_named_s3_scheme() { - // `s3+` schemes are preserved end-to-end: `path` strips the real - // scheme (not the canonical `s3`), and `parent`/`file_name` keep the - // qualifier intact. - let uri = Uri::for_test("s3+alt://bucket/foo/bar"); - assert_eq!(uri.as_str(), "s3+alt://bucket/foo/bar"); - assert_eq!(uri.protocol(), Protocol::S3); - assert_eq!(uri.parent().unwrap(), "s3+alt://bucket/foo"); - assert_eq!(uri.file_name().unwrap(), Path::new("bar")); - - let uri = Uri::for_test("s3+with-dash://bucket/key"); - assert_eq!(uri.parent().unwrap(), "s3+with-dash://bucket"); - assert_eq!(uri.file_name().unwrap(), Path::new("key")); - - // Mirrors the plain-`s3` guard: a bucket-only URI has no parent. - assert!(Uri::for_test("s3+alt://bucket").parent().is_none()); - assert!(Uri::for_test("s3+alt://bucket/").parent().is_none()); - } - #[test] fn test_uri_file_name() { assert!(Uri::for_test("file:///").file_name().is_none()); @@ -865,24 +812,4 @@ mod tests { serde_json::Value::String("s3://bucket/key".to_string()) ); } - - #[test] - fn test_uri_s3_named_preserved() { - // The `s3+` qualifier is the routing token for named S3-compatible - // backends (`storage.s3.named.`). It must survive parse + serialize - // so the storage resolver can recover the backend name on deserialization; - // before this guarantee, the qualifier was stripped by URI normalization - // and every `s3+://` URI silently resolved to the primary endpoint. - let uri = Uri::from_str("s3+alt://bucket/key").unwrap(); - assert_eq!(uri.protocol(), Protocol::S3); - assert_eq!(uri.as_str(), "s3+alt://bucket/key"); - let json = serde_json::to_value(&uri).unwrap(); - assert_eq!( - json, - serde_json::Value::String("s3+alt://bucket/key".to_string()) - ); - let round_trip: Uri = serde_json::from_value(json).unwrap(); - assert_eq!(round_trip.as_str(), "s3+alt://bucket/key"); - assert_eq!(round_trip.protocol(), Protocol::S3); - } } diff --git a/quickwit/quickwit-config/src/storage_config.rs b/quickwit/quickwit-config/src/storage_config.rs index 707aaad11a9..b4526b0c0d5 100644 --- a/quickwit/quickwit-config/src/storage_config.rs +++ b/quickwit/quickwit-config/src/storage_config.rs @@ -91,19 +91,6 @@ pub enum StorageBackendFlavor { #[derive(Debug, Clone, Default, Eq, PartialEq, Serialize, Deserialize)] pub struct StorageConfigs(#[serde_as(as = "EnumMap")] Vec); -/// Named backend names must be valid lowercase URL-scheme tails (`s3+://`). -fn validate_named_s3_backend_name(name: &str) -> anyhow::Result<()> { - ensure!(!name.is_empty(), "named S3 backend name must not be empty"); - for character in name.chars() { - ensure!( - character.is_ascii_lowercase() || character.is_ascii_digit() || character == '-', - "invalid named S3 backend name `{name}`: only lowercase ASCII letters, digits, and \ - `-` are allowed (the name is used in the `s3+{name}://` URI scheme)" - ); - } - Ok(()) -} - impl StorageConfigs { pub fn new(storage_configs: Vec) -> Self { Self(storage_configs) @@ -137,13 +124,6 @@ impl StorageConfigs { "{left:?} storage config is defined multiple times", ); } - for storage_config in self.0.iter() { - if let StorageConfig::S3(s3_storage_config) = storage_config { - for name in s3_storage_config.named.keys() { - validate_named_s3_backend_name(name)?; - } - } - } Ok(()) } @@ -381,26 +361,27 @@ pub struct S3StorageConfig { pub disable_stalled_stream_protection_upload: bool, #[serde(default)] pub disable_stalled_stream_protection_download: bool, - /// Additional named S3-compatible backends, addressed via `s3+://bucket/path` - /// URIs. Each entry is an independent endpoint with its own credentials, region, - /// etc. The map key (``) is the routing token used in the URI scheme. + /// Per-bucket S3-compatible backend overrides, keyed by bucket name. When an + /// `s3:///...` URI is resolved, an exact match here supplies that + /// bucket's own endpoint, credentials, region, and flags; any bucket not + /// listed falls back to the fields on this (primary) backend. #[serde(default)] #[serde(skip_serializing_if = "std::collections::BTreeMap::is_empty")] - pub named: std::collections::BTreeMap, - /// Set when this config is the projection of a named backend. Named - /// backends are self-contained, so the process-wide `QW_S3_ENDPOINT` / + pub profiles: std::collections::BTreeMap, + /// Set when this config is the projection of a per-bucket profile. Profiles + /// are self-contained, so the process-wide `QW_S3_ENDPOINT` / /// `QW_S3_FORCE_PATH_STYLE_ACCESS` overrides apply to the primary backend /// only. Not serialized; defaults to `false` (the primary backend). #[serde(skip)] - pub is_named_backend: bool, + pub is_profile: bool, } -/// Configuration for a named S3-compatible backend nested under -/// `storage.s3.named.`. Mirrors `S3StorageConfig` but cannot itself -/// have a `named` field (no recursion). +/// Per-bucket S3-compatible backend override nested under +/// `storage.s3.profiles.`. Mirrors `S3StorageConfig` but cannot itself +/// nest further profiles (no recursion). #[derive(Clone, Default, Eq, PartialEq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] -pub struct NamedS3StorageConfig { +pub struct S3ProfileConfig { #[serde(default)] #[serde(skip_serializing_if = "Option::is_none")] pub flavor: Option, @@ -430,9 +411,9 @@ pub struct NamedS3StorageConfig { pub disable_stalled_stream_protection_download: bool, } -impl NamedS3StorageConfig { - /// Project this named config back into a full `S3StorageConfig` - /// (with an empty `named` map) so it can flow through the existing +impl S3ProfileConfig { + /// Project this profile back into a full `S3StorageConfig` + /// (with an empty `profiles` map) so it can flow through the existing /// S3 client construction code unchanged. pub fn as_s3_config(&self) -> S3StorageConfig { let mut s3_config = S3StorageConfig { @@ -449,8 +430,8 @@ impl NamedS3StorageConfig { disable_stalled_stream_protection_upload: self.disable_stalled_stream_protection_upload, disable_stalled_stream_protection_download: self .disable_stalled_stream_protection_download, - named: Default::default(), - is_named_backend: true, + profiles: Default::default(), + is_profile: true, }; // Expand `flavor` shortcuts (region/path-style/checksum defaults) the // same way the primary backend does at config load time. @@ -465,9 +446,9 @@ impl NamedS3StorageConfig { } } -impl fmt::Debug for NamedS3StorageConfig { +impl fmt::Debug for S3ProfileConfig { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("NamedS3StorageConfig") + f.debug_struct("S3ProfileConfig") .field("flavor", &self.flavor) .field("access_key_id", &self.access_key_id) .field( @@ -528,15 +509,15 @@ impl S3StorageConfig { if let Some(secret_access_key) = self.secret_access_key.as_mut() { *secret_access_key = "***redacted***".to_string(); } - for named_config in self.named.values_mut() { - named_config.redact(); + for profile in self.profiles.values_mut() { + profile.redact(); } } pub fn endpoint(&self) -> Option { - // `QW_S3_ENDPOINT` overrides the primary backend only; named backends - // are self-contained and use their own configured endpoint. - if !self.is_named_backend + // `QW_S3_ENDPOINT` overrides the primary backend only; per-bucket + // profiles are self-contained and use their own configured endpoint. + if !self.is_profile && let Ok(endpoint) = env::var("QW_S3_ENDPOINT") { return Some(endpoint); @@ -547,7 +528,7 @@ impl S3StorageConfig { pub fn force_path_style_access(&self) -> Option { // `QW_S3_FORCE_PATH_STYLE_ACCESS` overrides the primary backend only. // No process-wide cache: each backend must honor its own setting. - if self.is_named_backend { + if self.is_profile { return Some(self.force_path_style_access); } Some(get_bool_from_env( @@ -582,7 +563,7 @@ impl fmt::Debug for S3StorageConfig { "disable_stalled_stream_protection_download", &self.disable_stalled_stream_protection_download, ) - .field("named", &self.named) + .field("profiles", &self.profiles) .finish() } } @@ -883,61 +864,77 @@ mod tests { } #[test] - fn test_storage_s3_named_backends_serde() { + fn test_storage_s3_profiles_serde() { let s3_storage_config_yaml = r#" endpoint: https://primary.example.com region: us-east-1 - named: - alt: + profiles: + logs-bucket-eu: endpoint: https://alt.example.com region: eu-west-3 force_path_style_access: true access_key_id: alt-key secret_access_key: alt-secret - secondary: - endpoint: http://secondary.example.com:8333 + seaweed-logs: + endpoint: http://seaweedfs-s3:8333 region: us-east-1 force_path_style_access: true "#; let s3_storage_config: S3StorageConfig = serde_yaml::from_str(s3_storage_config_yaml).unwrap(); - assert_eq!(s3_storage_config.named.len(), 2); + assert_eq!(s3_storage_config.profiles.len(), 2); - let alt = s3_storage_config.named.get("alt").unwrap(); - assert_eq!(alt.region.as_deref(), Some("eu-west-3")); - assert_eq!(alt.access_key_id.as_deref(), Some("alt-key")); - assert!(alt.force_path_style_access); + let eu = s3_storage_config.profiles.get("logs-bucket-eu").unwrap(); + assert_eq!(eu.region.as_deref(), Some("eu-west-3")); + assert_eq!(eu.access_key_id.as_deref(), Some("alt-key")); + assert!(eu.force_path_style_access); - // `as_s3_config` projects a named entry back into a full S3StorageConfig - // (with an empty `named` map) so it can drive the S3 client builder + // `as_s3_config` projects a profile back into a full S3StorageConfig + // (with an empty `profiles` map) so it can drive the S3 client builder // unchanged. - let projected = alt.as_s3_config(); + let projected = eu.as_s3_config(); assert_eq!(projected.region.as_deref(), Some("eu-west-3")); assert_eq!(projected.access_key_id.as_deref(), Some("alt-key")); assert!(projected.force_path_style_access); - assert!(projected.named.is_empty()); + assert!(projected.profiles.is_empty()); + } + + #[test] + fn test_storage_s3_profiles_bucket_name_keys() { + // The map key is a bucket name, so dotted bucket names are accepted as-is + // — no URI-scheme syntax is imposed on the key. + let s3_storage_config_yaml = r#" + profiles: + my.dotted.bucket: + endpoint: https://logs.example.com + "#; + let s3_storage_config: S3StorageConfig = + serde_yaml::from_str(s3_storage_config_yaml).unwrap(); + assert!(s3_storage_config.profiles.contains_key("my.dotted.bucket")); + let storage_configs = StorageConfigs::new(vec![s3_storage_config.into()]); + storage_configs.validate().unwrap(); } #[test] - fn test_storage_s3_named_backends_redact() { - let mut named = NamedS3StorageConfig { + fn test_storage_s3_profiles_redact() { + let mut profile = S3ProfileConfig { access_key_id: Some("public-key".to_string()), secret_access_key: Some("super-secret".to_string()), ..Default::default() }; - named.redact(); - assert_eq!(named.access_key_id.as_deref(), Some("public-key")); - assert_eq!(named.secret_access_key.as_deref(), Some("***redacted***")); + profile.redact(); + assert_eq!(profile.access_key_id.as_deref(), Some("public-key")); + assert_eq!(profile.secret_access_key.as_deref(), Some("***redacted***")); } #[test] - fn test_storage_s3_named_backends_field_parity() { - // Named backends accept the same fields as the primary S3 block, - // including the legacy `disable_multi_object_delete_requests` alias and - // the stalled-stream toggles, and project them through `as_s3_config`. + fn test_storage_s3_profiles_field_parity() { + // Profiles accept the same fields as the primary S3 block, including the + // legacy `disable_multi_object_delete_requests` alias and the + // stalled-stream toggles, and project them through `as_s3_config`. let s3_storage_config_yaml = r#" - named: - alt: + profiles: + logs-bucket: endpoint: https://alt.example.com disable_multi_object_delete_requests: true disable_stalled_stream_protection_upload: true @@ -946,13 +943,13 @@ mod tests { "#; let s3_storage_config: S3StorageConfig = serde_yaml::from_str(s3_storage_config_yaml).unwrap(); - let alt = s3_storage_config.named.get("alt").unwrap(); - assert!(alt.disable_multi_object_delete); - assert!(alt.disable_stalled_stream_protection_upload); - assert!(alt.disable_stalled_stream_protection_download); + let profile = s3_storage_config.profiles.get("logs-bucket").unwrap(); + assert!(profile.disable_multi_object_delete); + assert!(profile.disable_stalled_stream_protection_upload); + assert!(profile.disable_stalled_stream_protection_download); - let projected = alt.as_s3_config(); - assert!(projected.is_named_backend); + let projected = profile.as_s3_config(); + assert!(projected.is_profile); assert!(projected.disable_multi_object_delete); assert!(projected.disable_stalled_stream_protection_upload); assert!(projected.disable_stalled_stream_protection_download); @@ -960,27 +957,27 @@ mod tests { // A genuinely unknown field is still rejected. let invalid_yaml = r#" - named: - alt: + profiles: + logs-bucket: bogus_field: true "#; assert!(serde_yaml::from_str::(invalid_yaml).is_err()); } #[test] - fn test_storage_s3_named_backend_applies_flavor() { - // `flavor` shortcuts expand for named backends just like the primary. + fn test_storage_s3_profile_applies_flavor() { + // `flavor` shortcuts expand for profiles just like the primary backend. let s3_storage_config_yaml = r#" - named: - minio-backend: + profiles: + minio-bucket: flavor: minio endpoint: http://minio.example.com:9000 "#; let s3_storage_config: S3StorageConfig = serde_yaml::from_str(s3_storage_config_yaml).unwrap(); let projected = s3_storage_config - .named - .get("minio-backend") + .profiles + .get("minio-bucket") .unwrap() .as_s3_config(); assert_eq!(projected.region.as_deref(), Some("minio")); @@ -988,43 +985,19 @@ mod tests { } #[test] - fn test_storage_s3_named_backend_uses_own_endpoint() { - // A named backend is self-contained: `endpoint()` returns its configured + fn test_storage_s3_profile_uses_own_endpoint() { + // A profile is self-contained: `endpoint()` returns its configured // endpoint regardless of the process-wide `QW_S3_ENDPOINT` override, // which applies to the primary backend only. - let named = NamedS3StorageConfig { - endpoint: Some("https://named.example.com".to_string()), + let profile = S3ProfileConfig { + endpoint: Some("https://profile.example.com".to_string()), ..Default::default() }; - let projected = named.as_s3_config(); - assert!(projected.is_named_backend); + let projected = profile.as_s3_config(); + assert!(projected.is_profile); assert_eq!( projected.endpoint(), - Some("https://named.example.com".to_string()) + Some("https://profile.example.com".to_string()) ); } - - #[test] - fn test_validate_named_s3_backend_name() { - for valid in ["alt", "seaweedfs", "ovh-morocco", "s3alt", "minio-backend"] { - validate_named_s3_backend_name(valid).unwrap(); - } - for invalid in ["", "prod_logs", "Prod", "a.b", "a/b", "a b"] { - validate_named_s3_backend_name(invalid).unwrap_err(); - } - } - - #[test] - fn test_storage_configs_reject_url_incompatible_named_backend() { - let s3_storage_config_yaml = r#" - named: - prod_logs: - endpoint: https://logs.example.com - "#; - let s3_storage_config: S3StorageConfig = - serde_yaml::from_str(s3_storage_config_yaml).unwrap(); - let storage_configs = StorageConfigs::new(vec![s3_storage_config.into()]); - let error = storage_configs.validate().unwrap_err().to_string(); - assert!(error.contains("prod_logs"), "unexpected error: {error}"); - } } diff --git a/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage_resolver.rs b/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage_resolver.rs index 1ef8357671a..50b70ac03d4 100644 --- a/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage_resolver.rs +++ b/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage_resolver.rs @@ -21,19 +21,11 @@ use quickwit_common::uri::Uri; use quickwit_config::{S3StorageConfig, StorageBackend}; use tokio::sync::OnceCell; -use super::s3_compatible_storage::create_s3_client; +use super::s3_compatible_storage::{create_s3_client, parse_s3_uri}; use crate::{ DebouncedStorage, S3CompatibleObjectStorage, Storage, StorageFactory, StorageResolverError, }; -/// Extracts the named-backend key out of an `s3+://...` URI, if any. -/// Returns `None` for plain `s3://...`. -fn parse_named_key(uri: &Uri) -> Option<&str> { - let scheme_end = uri.as_str().find("://")?; - let scheme = &uri.as_str()[..scheme_end]; - scheme.strip_prefix("s3+") -} - /// S3 compatible object storage resolver. pub struct S3CompatibleObjectStorageFactory { storage_config: S3StorageConfig, @@ -43,8 +35,9 @@ pub struct S3CompatibleObjectStorageFactory { // end up being used, or if something like azure, gcs, or even local files, will be used // instead. s3_client: OnceCell, - // Per-named-backend client cell; the mutex is held only to fetch/insert the cell, never across the build. - named_s3_clients: Mutex>>>, + // Per-bucket client cell for buckets matched by a `storage.s3.profiles.` entry. + // The mutex is held only to fetch/insert the cell, never across the client build. + profile_s3_clients: Mutex>>>, } impl S3CompatibleObjectStorageFactory { @@ -53,7 +46,7 @@ impl S3CompatibleObjectStorageFactory { Self { storage_config, s3_client: OnceCell::new(), - named_s3_clients: Mutex::new(HashMap::new()), + profile_s3_clients: Mutex::new(HashMap::new()), } } } @@ -65,30 +58,26 @@ impl StorageFactory for S3CompatibleObjectStorageFactory { } async fn resolve(&self, uri: &Uri) -> Result, StorageResolverError> { - if let Some(name) = parse_named_key(uri) { - let named_config = self - .storage_config - .named - .get(name) - .ok_or_else(|| { - StorageResolverError::InvalidUri(format!( - "no `storage.s3.named.{name}` entry configured for URI `{uri}`" - )) - })? - .as_s3_config(); + // A `storage.s3.profiles.` entry, if present, supplies that bucket's + // own endpoint/credentials/region; any unlisted bucket uses the primary backend. + if let Some((bucket, _prefix)) = parse_s3_uri(uri) + && let Some(profile) = self.storage_config.profiles.get(&bucket) + { + let profile_config = profile.as_s3_config(); let client_cell = { let mut clients = self - .named_s3_clients + .profile_s3_clients .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); - Arc::clone(clients.entry(name.to_string()).or_default()) + Arc::clone(clients.entry(bucket).or_default()) }; let client = client_cell - .get_or_init(|| create_s3_client(&named_config)) + .get_or_init(|| create_s3_client(&profile_config)) .await .clone(); let storage = - S3CompatibleObjectStorage::from_uri_and_client(&named_config, uri, client).await?; + S3CompatibleObjectStorage::from_uri_and_client(&profile_config, uri, client) + .await?; return Ok(Arc::new(DebouncedStorage::new(storage))); } let s3_client = self @@ -102,23 +91,3 @@ impl StorageFactory for S3CompatibleObjectStorageFactory { Ok(Arc::new(DebouncedStorage::new(storage))) } } - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_parse_named_key() { - // Plain s3:// URIs route through the primary backend. - assert_eq!(parse_named_key(&Uri::for_test("s3://bucket/key")), None); - // `s3+` URIs return the named-backend key. - assert_eq!( - parse_named_key(&Uri::for_test("s3+alt://bucket/key")), - Some("alt") - ); - assert_eq!( - parse_named_key(&Uri::for_test("s3+with-dash://bucket/key")), - Some("with-dash") - ); - } -} From bfcc308e399f4046e285c49d5e87fbc8a1d4f3be Mon Sep 17 00:00:00 2001 From: "david.yaffe" Date: Fri, 2 Oct 2026 19:20:55 +0000 Subject: [PATCH 5/9] Rename storage.s3.profiles to storage.s3.buckets and dedupe config "Profiles" collides with AWS credential profiles; the map keys are bucket names, which also matches Hadoop's per-bucket S3A config. - Drop S3ProfileConfig: `buckets` is a map of S3StorageConfig, so serde, flavors, redaction, and Debug are shared with the primary backend. Nested `buckets` are rejected in StorageConfigs::validate. - apply_flavor recurses into bucket configs at load time. - The resolver builds one backend (config + lazy client) per bucket at construction, so resolve() has a single code path and needs no mutex. --- docs/configuration/storage-config.md | 16 +- .../quickwit-config/src/storage_config.rs | 312 +++++++----------- .../s3_compatible_storage_resolver.rs | 62 ++-- 3 files changed, 157 insertions(+), 233 deletions(-) diff --git a/docs/configuration/storage-config.md b/docs/configuration/storage-config.md index 7bc995543e6..7732ddc7e42 100644 --- a/docs/configuration/storage-config.md +++ b/docs/configuration/storage-config.md @@ -14,7 +14,7 @@ Quickwit currently supports four types of storage providers: ## Storage URIs Storage URIs refer to different storage providers identified by a URI "protocol" or "scheme". Quickwit supports the following storage URI protocols: -- `s3://` for Amazon S3 and S3-compatible (per-bucket overrides via `storage.s3.profiles`, see [Per-bucket S3 profiles](#per-bucket-s3-profiles)) +- `s3://` for Amazon S3 and S3-compatible (per-bucket overrides via `storage.s3.buckets`, see [Per-bucket S3 configuration](#per-bucket-s3-configuration)) - `azure://` for Azure Blob Storage - `file://` for local file systems - `gs://` for Google Cloud Storage @@ -104,23 +104,23 @@ storage: endpoint: https://storage.googleapis.com ``` -#### Per-bucket S3 profiles +#### Per-bucket S3 configuration -In addition to the primary `s3:` block, you can declare per-bucket overrides under `storage.s3.profiles`. The map key is the bucket name; when an `s3:///...` URI is resolved, an exact match supplies that bucket's own endpoint, credentials, region, and flags. Any bucket not listed falls back to the fields on the primary `s3:` block. URIs stay canonical `s3://` — routing is by bucket name, so nothing extra is persisted in the index metadata. +In addition to the primary `s3:` block, you can declare per-bucket overrides under `storage.s3.buckets`. The map key is the bucket name; when an `s3:///...` URI is resolved, an exact match supplies that bucket's own endpoint, credentials, region, and flags. Any bucket not listed falls back to the fields on the primary `s3:` block. URIs stay canonical `s3://` — routing is by bucket name, so nothing extra is persisted in the index metadata. -Each profile accepts the same fields as the primary `s3:` block, *except* `profiles` itself (no recursion). If `access_key_id` / `secret_access_key` are omitted on a profile, the global AWS SDK credential chain is used (env vars, instance metadata, etc.). +Each bucket entry accepts the same fields as the primary `s3:` block, *except* `buckets` itself (no nesting). Bucket entries do not inherit fields from the primary block. If `access_key_id` / `secret_access_key` are omitted on a bucket entry, the global AWS SDK credential chain is used (env vars, instance metadata, etc.). -Profiles are self-contained: the process-wide `QW_S3_ENDPOINT` and `QW_S3_FORCE_PATH_STYLE_ACCESS` overrides apply to the primary `s3:` backend only. A profile always uses its own `endpoint` and `force_path_style_access` values. +Bucket entries are self-contained: the process-wide `QW_S3_ENDPOINT` and `QW_S3_FORCE_PATH_STYLE_ACCESS` overrides apply to the primary `s3:` backend only. A bucket entry always uses its own `endpoint` and `force_path_style_access` values. > Because routing keys on bucket name, a given bucket name maps to exactly one backend. If you need the *same* bucket name on two different endpoints, give the buckets distinct names. ```yaml storage: s3: - # Primary backend — used for any bucket not listed under `profiles`. + # Primary backend — used for any bucket not listed under `buckets`. endpoint: https://s3.us-east-1.amazonaws.com region: us-east-1 - profiles: + buckets: # Buckets named `logs-bucket-eu` resolve to this endpoint. logs-bucket-eu: endpoint: https://s3.eu-west-3.amazonaws.com @@ -135,7 +135,7 @@ storage: force_path_style_access: true ``` -An index simply points at the bucket by its canonical URI; the matching profile is applied automatically: +An index simply points at the bucket by its canonical URI; the matching bucket entry is applied automatically: ```yaml index_id: logs-eu diff --git a/quickwit/quickwit-config/src/storage_config.rs b/quickwit/quickwit-config/src/storage_config.rs index 994bed18770..d5784378cb7 100644 --- a/quickwit/quickwit-config/src/storage_config.rs +++ b/quickwit/quickwit-config/src/storage_config.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::collections::BTreeMap; use std::ops::Deref; use std::{env, fmt}; @@ -120,6 +121,14 @@ impl StorageConfigs { "{left:?} storage config is defined multiple times", ); } + if let Some(s3_storage_config) = self.find_s3() { + for (bucket, bucket_config) in &s3_storage_config.buckets { + ensure!( + bucket_config.buckets.is_empty(), + "S3 bucket config `{bucket}` cannot define nested `buckets`", + ); + } + } Ok(()) } @@ -401,116 +410,16 @@ pub struct S3StorageConfig { /// Per-bucket S3-compatible backend overrides, keyed by bucket name. When an /// `s3:///...` URI is resolved, an exact match here supplies that /// bucket's own endpoint, credentials, region, and flags; any bucket not - /// listed falls back to the fields on this (primary) backend. + /// listed falls back to the fields on this (primary) backend. Bucket configs + /// cannot themselves define `buckets`. #[serde(default)] - #[serde(skip_serializing_if = "std::collections::BTreeMap::is_empty")] - pub profiles: std::collections::BTreeMap, - /// Set when this config is the projection of a per-bucket profile. Profiles - /// are self-contained, so the process-wide `QW_S3_ENDPOINT` / - /// `QW_S3_FORCE_PATH_STYLE_ACCESS` overrides apply to the primary backend - /// only. Not serialized; defaults to `false` (the primary backend). + #[serde(skip_serializing_if = "BTreeMap::is_empty")] + pub buckets: BTreeMap, + /// Set on the configs returned by [`S3StorageConfig::bucket_configs`]. Bucket + /// configs are self-contained, so the process-wide `QW_S3_ENDPOINT` and + /// `QW_S3_FORCE_PATH_STYLE_ACCESS` overrides apply to the primary backend only. #[serde(skip)] - pub is_profile: bool, -} - -/// Per-bucket S3-compatible backend override nested under -/// `storage.s3.profiles.`. Mirrors `S3StorageConfig` but cannot itself -/// nest further profiles (no recursion). -#[derive(Clone, Default, Eq, PartialEq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct S3ProfileConfig { - #[serde(default)] - #[serde(skip_serializing_if = "Option::is_none")] - pub flavor: Option, - #[serde(default)] - pub access_key_id: Option, - #[serde(default)] - pub secret_access_key: Option, - #[serde(default)] - pub region: Option, - #[serde(default)] - pub endpoint: Option, - #[serde(default)] - pub force_path_style_access: bool, - #[serde(alias = "disable_multi_object_delete_requests")] - #[serde(default)] - pub disable_multi_object_delete: bool, - #[serde(default)] - pub disable_multipart_upload: bool, - #[serde(default)] - pub checksum_algorithm: ChecksumAlgorithm, - /// Deprecated: applies into `checksum_algorithm: disabled`. - #[serde(default, skip_serializing)] - pub disable_checksums: bool, - #[serde(default)] - pub disable_stalled_stream_protection_upload: bool, - #[serde(default)] - pub disable_stalled_stream_protection_download: bool, -} - -impl S3ProfileConfig { - /// Project this profile back into a full `S3StorageConfig` - /// (with an empty `profiles` map) so it can flow through the existing - /// S3 client construction code unchanged. - pub fn as_s3_config(&self) -> S3StorageConfig { - let mut s3_config = S3StorageConfig { - flavor: self.flavor, - access_key_id: self.access_key_id.clone(), - secret_access_key: self.secret_access_key.clone(), - region: self.region.clone(), - endpoint: self.endpoint.clone(), - force_path_style_access: self.force_path_style_access, - disable_multi_object_delete: self.disable_multi_object_delete, - disable_multipart_upload: self.disable_multipart_upload, - checksum_algorithm: self.checksum_algorithm, - disable_checksums: self.disable_checksums, - disable_stalled_stream_protection_upload: self.disable_stalled_stream_protection_upload, - disable_stalled_stream_protection_download: self - .disable_stalled_stream_protection_download, - profiles: Default::default(), - is_profile: true, - }; - // Expand `flavor` shortcuts (region/path-style/checksum defaults) the - // same way the primary backend does at config load time. - s3_config.apply_flavor(); - s3_config - } - - pub fn redact(&mut self) { - if let Some(secret_access_key) = self.secret_access_key.as_mut() { - *secret_access_key = "***redacted***".to_string(); - } - } -} - -impl fmt::Debug for S3ProfileConfig { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("S3ProfileConfig") - .field("flavor", &self.flavor) - .field("access_key_id", &self.access_key_id) - .field( - "secret_access_key", - &self.secret_access_key.as_ref().map(|_| "***redacted***"), - ) - .field("region", &self.region) - .field("endpoint", &self.endpoint) - .field("force_path_style_access", &self.force_path_style_access) - .field( - "disable_multi_object_delete", - &self.disable_multi_object_delete, - ) - .field("disable_multipart_upload", &self.disable_multipart_upload) - .field("checksum_algorithm", &self.checksum_algorithm) - .field( - "disable_stalled_stream_protection_upload", - &self.disable_stalled_stream_protection_upload, - ) - .field( - "disable_stalled_stream_protection_download", - &self.disable_stalled_stream_protection_download, - ) - .finish() - } + pub is_bucket_config: bool, } impl S3StorageConfig { @@ -540,21 +449,34 @@ impl S3StorageConfig { if self.disable_checksums { self.checksum_algorithm = ChecksumAlgorithm::Disabled; } + for bucket_config in self.buckets.values_mut() { + bucket_config.apply_flavor(); + } } pub fn redact(&mut self) { if let Some(secret_access_key) = self.secret_access_key.as_mut() { *secret_access_key = "***redacted***".to_string(); } - for profile in self.profiles.values_mut() { - profile.redact(); + for bucket_config in self.buckets.values_mut() { + bucket_config.redact(); } } + pub fn bucket_configs(&self) -> impl Iterator + '_ { + self.buckets.iter().map(|(bucket, bucket_config)| { + let bucket_config = S3StorageConfig { + is_bucket_config: true, + ..bucket_config.clone() + }; + (bucket.as_str(), bucket_config) + }) + } + pub fn endpoint(&self) -> Option { - // `QW_S3_ENDPOINT` overrides the primary backend only; per-bucket - // profiles are self-contained and use their own configured endpoint. - if !self.is_profile + // `QW_S3_ENDPOINT` overrides the primary backend only; bucket configs + // are self-contained and use their own configured endpoint. + if !self.is_bucket_config && let Ok(endpoint) = env::var("QW_S3_ENDPOINT") { return Some(endpoint); @@ -565,7 +487,7 @@ impl S3StorageConfig { pub fn force_path_style_access(&self) -> Option { // `QW_S3_FORCE_PATH_STYLE_ACCESS` overrides the primary backend only. // No process-wide cache: each backend must honor its own setting. - if self.is_profile { + if self.is_bucket_config { return Some(self.force_path_style_access); } Some(get_bool_from_env( @@ -600,7 +522,7 @@ impl fmt::Debug for S3StorageConfig { "disable_stalled_stream_protection_download", &self.disable_stalled_stream_protection_download, ) - .field("profiles", &self.profiles) + .field("buckets", &self.buckets) .finish() } } @@ -954,11 +876,11 @@ mod tests { } #[test] - fn test_storage_s3_profiles_serde() { + fn test_storage_s3_buckets_serde() { let s3_storage_config_yaml = r#" endpoint: https://primary.example.com region: us-east-1 - profiles: + buckets: logs-bucket-eu: endpoint: https://alt.example.com region: eu-west-3 @@ -972,122 +894,126 @@ mod tests { "#; let s3_storage_config: S3StorageConfig = serde_yaml::from_str(s3_storage_config_yaml).unwrap(); - assert_eq!(s3_storage_config.profiles.len(), 2); + assert_eq!(s3_storage_config.buckets.len(), 2); - let eu = s3_storage_config.profiles.get("logs-bucket-eu").unwrap(); + let eu = s3_storage_config.buckets.get("logs-bucket-eu").unwrap(); assert_eq!(eu.region.as_deref(), Some("eu-west-3")); assert_eq!(eu.access_key_id.as_deref(), Some("alt-key")); assert!(eu.force_path_style_access); + assert!(!eu.is_bucket_config); - // `as_s3_config` projects a profile back into a full S3StorageConfig - // (with an empty `profiles` map) so it can drive the S3 client builder - // unchanged. - let projected = eu.as_s3_config(); - assert_eq!(projected.region.as_deref(), Some("eu-west-3")); - assert_eq!(projected.access_key_id.as_deref(), Some("alt-key")); - assert!(projected.force_path_style_access); - assert!(projected.profiles.is_empty()); + let bucket_configs: Vec<(&str, S3StorageConfig)> = + s3_storage_config.bucket_configs().collect(); + assert_eq!(bucket_configs.len(), 2); + let (bucket, bucket_config) = &bucket_configs[0]; + assert_eq!(*bucket, "logs-bucket-eu"); + assert_eq!(bucket_config.region.as_deref(), Some("eu-west-3")); + assert!(bucket_config.is_bucket_config); } #[test] - fn test_storage_s3_profiles_bucket_name_keys() { - // The map key is a bucket name, so dotted bucket names are accepted as-is - // — no URI-scheme syntax is imposed on the key. + fn test_storage_s3_buckets_dotted_bucket_name() { + // The map key is a bucket name, so dotted bucket names are accepted as-is. let s3_storage_config_yaml = r#" - profiles: + buckets: my.dotted.bucket: endpoint: https://logs.example.com "#; let s3_storage_config: S3StorageConfig = serde_yaml::from_str(s3_storage_config_yaml).unwrap(); - assert!(s3_storage_config.profiles.contains_key("my.dotted.bucket")); + assert!(s3_storage_config.buckets.contains_key("my.dotted.bucket")); let storage_configs = StorageConfigs::new(vec![s3_storage_config.into()]); storage_configs.validate().unwrap(); } #[test] - fn test_storage_s3_profiles_redact() { - let mut profile = S3ProfileConfig { - access_key_id: Some("public-key".to_string()), - secret_access_key: Some("super-secret".to_string()), - ..Default::default() - }; - profile.redact(); - assert_eq!(profile.access_key_id.as_deref(), Some("public-key")); - assert_eq!(profile.secret_access_key.as_deref(), Some("***redacted***")); - } - - #[test] - fn test_storage_s3_profiles_field_parity() { - // Profiles accept the same fields as the primary S3 block, including the - // legacy `disable_multi_object_delete_requests` alias and the - // stalled-stream toggles, and project them through `as_s3_config`. + fn test_storage_s3_buckets_reject_nested_buckets() { let s3_storage_config_yaml = r#" - profiles: + buckets: logs-bucket: - endpoint: https://alt.example.com - disable_multi_object_delete_requests: true - disable_stalled_stream_protection_upload: true - disable_stalled_stream_protection_download: true - checksum_algorithm: disabled + buckets: + nested-bucket: + endpoint: https://nested.example.com "#; let s3_storage_config: S3StorageConfig = serde_yaml::from_str(s3_storage_config_yaml).unwrap(); - let profile = s3_storage_config.profiles.get("logs-bucket").unwrap(); - assert!(profile.disable_multi_object_delete); - assert!(profile.disable_stalled_stream_protection_upload); - assert!(profile.disable_stalled_stream_protection_download); - - let projected = profile.as_s3_config(); - assert!(projected.is_profile); - assert!(projected.disable_multi_object_delete); - assert!(projected.disable_stalled_stream_protection_upload); - assert!(projected.disable_stalled_stream_protection_download); - assert_eq!(projected.checksum_algorithm, ChecksumAlgorithm::Disabled); - - // A genuinely unknown field is still rejected. - let invalid_yaml = r#" - profiles: - logs-bucket: - bogus_field: true - "#; - assert!(serde_yaml::from_str::(invalid_yaml).is_err()); + let storage_configs = StorageConfigs::new(vec![s3_storage_config.into()]); + let error = storage_configs.validate().unwrap_err(); + assert!(error.to_string().contains("logs-bucket")); } #[test] - fn test_storage_s3_profile_applies_flavor() { - // `flavor` shortcuts expand for profiles just like the primary backend. + fn test_storage_s3_buckets_redact() { + let mut s3_storage_config = S3StorageConfig { + buckets: BTreeMap::from([( + "logs-bucket".to_string(), + S3StorageConfig { + access_key_id: Some("public-key".to_string()), + secret_access_key: Some("super-secret".to_string()), + ..Default::default() + }, + )]), + ..Default::default() + }; + assert!(!format!("{s3_storage_config:?}").contains("super-secret")); + + s3_storage_config.redact(); + let bucket_config = &s3_storage_config.buckets["logs-bucket"]; + assert_eq!(bucket_config.access_key_id.as_deref(), Some("public-key")); + assert_eq!( + bucket_config.secret_access_key.as_deref(), + Some("***redacted***") + ); + } + + #[test] + fn test_storage_s3_buckets_apply_flavor() { + // `flavor` shortcuts expand for bucket configs just like the primary backend. let s3_storage_config_yaml = r#" - profiles: + buckets: minio-bucket: flavor: minio endpoint: http://minio.example.com:9000 + legacy-bucket: + disable_checksums: true "#; let s3_storage_config: S3StorageConfig = serde_yaml::from_str(s3_storage_config_yaml).unwrap(); - let projected = s3_storage_config - .profiles - .get("minio-bucket") - .unwrap() - .as_s3_config(); - assert_eq!(projected.region.as_deref(), Some("minio")); - assert!(projected.force_path_style_access); + let mut storage_configs = StorageConfigs::new(vec![s3_storage_config.into()]); + storage_configs.apply_flavors(); + let s3_storage_config = storage_configs.find_s3().unwrap(); + + let minio_bucket_config = &s3_storage_config.buckets["minio-bucket"]; + assert_eq!(minio_bucket_config.region.as_deref(), Some("minio")); + assert!(minio_bucket_config.force_path_style_access); + + let legacy_bucket_config = &s3_storage_config.buckets["legacy-bucket"]; + assert_eq!( + legacy_bucket_config.checksum_algorithm, + ChecksumAlgorithm::Disabled + ); } #[test] - fn test_storage_s3_profile_uses_own_endpoint() { - // A profile is self-contained: `endpoint()` returns its configured - // endpoint regardless of the process-wide `QW_S3_ENDPOINT` override, - // which applies to the primary backend only. - let profile = S3ProfileConfig { - endpoint: Some("https://profile.example.com".to_string()), + fn test_storage_s3_bucket_config_uses_own_endpoint() { + // A bucket config is self-contained: `endpoint()` and + // `force_path_style_access()` skip the process-wide env overrides, which + // apply to the primary backend only. + let s3_storage_config = S3StorageConfig { + buckets: BTreeMap::from([( + "logs-bucket".to_string(), + S3StorageConfig { + endpoint: Some("https://bucket.example.com".to_string()), + ..Default::default() + }, + )]), ..Default::default() }; - let projected = profile.as_s3_config(); - assert!(projected.is_profile); + let (_, bucket_config) = s3_storage_config.bucket_configs().next().unwrap(); assert_eq!( - projected.endpoint(), - Some("https://profile.example.com".to_string()) + bucket_config.endpoint(), + Some("https://bucket.example.com".to_string()) ); + assert_eq!(bucket_config.force_path_style_access(), Some(false)); } } diff --git a/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage_resolver.rs b/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage_resolver.rs index 50b70ac03d4..29128052fc3 100644 --- a/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage_resolver.rs +++ b/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage_resolver.rs @@ -13,7 +13,7 @@ // limitations under the License. use std::collections::HashMap; -use std::sync::{Arc, Mutex}; +use std::sync::Arc; use async_trait::async_trait; use aws_sdk_s3::Client as S3Client; @@ -28,6 +28,12 @@ use crate::{ /// S3 compatible object storage resolver. pub struct S3CompatibleObjectStorageFactory { + default_backend: S3Backend, + // Backends for the buckets listed under `storage.s3.buckets`, keyed by bucket name. + bucket_backends: HashMap, +} + +struct S3Backend { storage_config: S3StorageConfig, // we cache the S3Client so we don't rebuild one every time we build a new Storage (for // every search query). @@ -35,18 +41,27 @@ pub struct S3CompatibleObjectStorageFactory { // end up being used, or if something like azure, gcs, or even local files, will be used // instead. s3_client: OnceCell, - // Per-bucket client cell for buckets matched by a `storage.s3.profiles.` entry. - // The mutex is held only to fetch/insert the cell, never across the client build. - profile_s3_clients: Mutex>>>, +} + +impl S3Backend { + fn new(storage_config: S3StorageConfig) -> Self { + Self { + storage_config, + s3_client: OnceCell::new(), + } + } } impl S3CompatibleObjectStorageFactory { /// Creates a new S3-compatible storage factory. pub fn new(storage_config: S3StorageConfig) -> Self { + let bucket_backends = storage_config + .bucket_configs() + .map(|(bucket, bucket_config)| (bucket.to_string(), S3Backend::new(bucket_config))) + .collect(); Self { - storage_config, - s3_client: OnceCell::new(), - profile_s3_clients: Mutex::new(HashMap::new()), + default_backend: S3Backend::new(storage_config), + bucket_backends, } } } @@ -58,35 +73,18 @@ impl StorageFactory for S3CompatibleObjectStorageFactory { } async fn resolve(&self, uri: &Uri) -> Result, StorageResolverError> { - // A `storage.s3.profiles.` entry, if present, supplies that bucket's - // own endpoint/credentials/region; any unlisted bucket uses the primary backend. - if let Some((bucket, _prefix)) = parse_s3_uri(uri) - && let Some(profile) = self.storage_config.profiles.get(&bucket) - { - let profile_config = profile.as_s3_config(); - let client_cell = { - let mut clients = self - .profile_s3_clients - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - Arc::clone(clients.entry(bucket).or_default()) - }; - let client = client_cell - .get_or_init(|| create_s3_client(&profile_config)) - .await - .clone(); - let storage = - S3CompatibleObjectStorage::from_uri_and_client(&profile_config, uri, client) - .await?; - return Ok(Arc::new(DebouncedStorage::new(storage))); - } - let s3_client = self + // A `storage.s3.buckets.` entry, if present, supplies that bucket's own + // endpoint/credentials/region; any unlisted bucket uses the primary backend. + let backend = parse_s3_uri(uri) + .and_then(|(bucket, _prefix)| self.bucket_backends.get(&bucket)) + .unwrap_or(&self.default_backend); + let s3_client = backend .s3_client - .get_or_init(|| create_s3_client(&self.storage_config)) + .get_or_init(|| create_s3_client(&backend.storage_config)) .await .clone(); let storage = - S3CompatibleObjectStorage::from_uri_and_client(&self.storage_config, uri, s3_client) + S3CompatibleObjectStorage::from_uri_and_client(&backend.storage_config, uri, s3_client) .await?; Ok(Arc::new(DebouncedStorage::new(storage))) } From c29079b46b2f6427f52a14778e7b4706766bedc9 Mon Sep 17 00:00:00 2001 From: "david.yaffe" Date: Tue, 6 Oct 2026 16:46:56 +0000 Subject: [PATCH 6/9] Validate S3 bucket keys and test per-bucket routing - Reject `storage.s3.buckets` keys that are empty or contain `/`. Such keys can never match a parsed bucket name, so the bucket would silently fall back to the primary backend and its credentials. - Test bucket routing in the S3 factory: exact match, dotted names, bucket-only URIs, no prefix matching, and fallback to the primary. - Make the env-override test set QW_S3_ENDPOINT and QW_S3_FORCE_PATH_STYLE_ACCESS, and restore the test that unknown fields in a bucket entry are rejected. - Skip URI parsing when no buckets are configured, and drop the primary backend's duplicate copy of the bucket configs. - Docs: add the `buckets` property and the QW_S3_FORCE_PATH_STYLE_ACCESS env variable to the reference tables. --- docs/configuration/storage-config.md | 4 +- .../quickwit-config/src/storage_config.rs | 68 +++++++++++++- .../s3_compatible_storage_resolver.rs | 90 +++++++++++++++++-- 3 files changed, 153 insertions(+), 9 deletions(-) diff --git a/docs/configuration/storage-config.md b/docs/configuration/storage-config.md index 7732ddc7e42..9d1658a88b4 100644 --- a/docs/configuration/storage-config.md +++ b/docs/configuration/storage-config.md @@ -58,6 +58,7 @@ This section contains one configuration subsection per storage provider. If a st | `disable_multipart_upload` | Disables [multipart upload](https://docs.aws.amazon.com/AmazonS3/latest/userguide/mpuoverview.html) of objects. Required by some S3-compatible providers (GCS). | `false` | | `checksum_algorithm` | Upload [checksum](https://docs.aws.amazon.com/AmazonS3/latest/userguide/checking-object-integrity.html) algorithm. Allowed values: `crc32c` (computed and validated by the AWS SDK), `md5` (sent client-side via `Content-MD5`; useful for S3-compatible providers that predate `x-amz-checksum-*`), or `disabled`. | `crc32c` | | `disable_checksums` | **Deprecated.** Previously a boolean that disabled all request/response checksums. Equivalent to setting `checksum_algorithm: disabled`. | `false` | +| `buckets` | Per-bucket S3 configurations keyed by bucket name. See [Per-bucket S3 configuration](#per-bucket-s3-configuration). | | :::warning Hardcoding credentials into configuration files is not secure and strongly discouraged. Prefer the alternative authentication methods that your storage backend may provide. @@ -68,6 +69,7 @@ Hardcoding credentials into configuration files is not secure and strongly disco | Env variable | Description | | --- | --- | | `QW_S3_ENDPOINT` | Custom S3 endpoint. | +| `QW_S3_FORCE_PATH_STYLE_ACCESS` | Overrides `force_path_style_access`. | | `QW_S3_MAX_CONCURRENCY` | Limit the number of concurrent requests to S3 | #### Storage flavors @@ -106,7 +108,7 @@ storage: #### Per-bucket S3 configuration -In addition to the primary `s3:` block, you can declare per-bucket overrides under `storage.s3.buckets`. The map key is the bucket name; when an `s3:///...` URI is resolved, an exact match supplies that bucket's own endpoint, credentials, region, and flags. Any bucket not listed falls back to the fields on the primary `s3:` block. URIs stay canonical `s3://` — routing is by bucket name, so nothing extra is persisted in the index metadata. +In addition to the primary `s3:` block, you can declare per-bucket overrides under `storage.s3.buckets`. The map key is the bare bucket name (no `s3://` prefix or path; such keys are rejected at startup); when an `s3:///...` URI is resolved, an exact match supplies that bucket's own endpoint, credentials, region, and flags. Any bucket not listed falls back to the fields on the primary `s3:` block. URIs stay canonical `s3://` — routing is by bucket name, so nothing extra is persisted in the index metadata. Each bucket entry accepts the same fields as the primary `s3:` block, *except* `buckets` itself (no nesting). Bucket entries do not inherit fields from the primary block. If `access_key_id` / `secret_access_key` are omitted on a bucket entry, the global AWS SDK credential chain is used (env vars, instance metadata, etc.). diff --git a/quickwit/quickwit-config/src/storage_config.rs b/quickwit/quickwit-config/src/storage_config.rs index d5784378cb7..42505d7c73f 100644 --- a/quickwit/quickwit-config/src/storage_config.rs +++ b/quickwit/quickwit-config/src/storage_config.rs @@ -123,6 +123,11 @@ impl StorageConfigs { } if let Some(s3_storage_config) = self.find_s3() { for (bucket, bucket_config) in &s3_storage_config.buckets { + ensure!( + !bucket.is_empty() && !bucket.contains('/'), + "S3 bucket config key `{bucket}` must be a bare bucket name, without `s3://` \ + or `/`", + ); ensure!( bucket_config.buckets.is_empty(), "S3 bucket config `{bucket}` cannot define nested `buckets`", @@ -942,6 +947,32 @@ mod tests { assert!(error.to_string().contains("logs-bucket")); } + #[test] + fn test_storage_s3_buckets_reject_unknown_fields() { + let s3_storage_config_yaml = r#" + buckets: + logs-bucket: + endpiont: https://typo.example.com + "#; + serde_yaml::from_str::(s3_storage_config_yaml).unwrap_err(); + } + + #[test] + fn test_storage_s3_buckets_reject_unmatchable_bucket_keys() { + for bucket_key in ["", "s3://logs-bucket", "logs-bucket/prefix"] { + let s3_storage_config = S3StorageConfig { + buckets: BTreeMap::from([(bucket_key.to_string(), S3StorageConfig::default())]), + ..Default::default() + }; + let storage_configs = StorageConfigs::new(vec![s3_storage_config.into()]); + let error = storage_configs.validate().unwrap_err(); + assert!( + error.to_string().contains("bare bucket name"), + "unexpected error for key `{bucket_key}`: {error}" + ); + } + } + #[test] fn test_storage_s3_buckets_redact() { let mut s3_storage_config = S3StorageConfig { @@ -1000,6 +1031,7 @@ mod tests { // `force_path_style_access()` skip the process-wide env overrides, which // apply to the primary backend only. let s3_storage_config = S3StorageConfig { + endpoint: Some("https://primary.example.com".to_string()), buckets: BTreeMap::from([( "logs-bucket".to_string(), S3StorageConfig { @@ -1010,10 +1042,42 @@ mod tests { ..Default::default() }; let (_, bucket_config) = s3_storage_config.bucket_configs().next().unwrap(); + + // SAFETY: this test may not be entirely sound if not run with nextest or --test-threads=1 + // as this is only a test, and it would be extremely inconvenient to run it in a different + // way, we are keeping it that way + let previous_endpoint = env::var("QW_S3_ENDPOINT").ok(); + let previous_force_path_style = env::var("QW_S3_FORCE_PATH_STYLE_ACCESS").ok(); + unsafe { + env::set_var("QW_S3_ENDPOINT", "https://env.example.com"); + env::set_var("QW_S3_FORCE_PATH_STYLE_ACCESS", "true"); + } + let primary_endpoint = s3_storage_config.endpoint(); + let primary_force_path_style = s3_storage_config.force_path_style_access(); + let bucket_endpoint = bucket_config.endpoint(); + let bucket_force_path_style = bucket_config.force_path_style_access(); + unsafe { + match previous_endpoint { + Some(endpoint) => env::set_var("QW_S3_ENDPOINT", endpoint), + None => env::remove_var("QW_S3_ENDPOINT"), + } + match previous_force_path_style { + Some(force_path_style) => { + env::set_var("QW_S3_FORCE_PATH_STYLE_ACCESS", force_path_style) + } + None => env::remove_var("QW_S3_FORCE_PATH_STYLE_ACCESS"), + } + } + + assert_eq!( + primary_endpoint, + Some("https://env.example.com".to_string()) + ); + assert_eq!(primary_force_path_style, Some(true)); assert_eq!( - bucket_config.endpoint(), + bucket_endpoint, Some("https://bucket.example.com".to_string()) ); - assert_eq!(bucket_config.force_path_style_access(), Some(false)); + assert_eq!(bucket_force_path_style, Some(false)); } } diff --git a/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage_resolver.rs b/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage_resolver.rs index 29128052fc3..a71195c07b6 100644 --- a/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage_resolver.rs +++ b/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage_resolver.rs @@ -54,16 +54,28 @@ impl S3Backend { impl S3CompatibleObjectStorageFactory { /// Creates a new S3-compatible storage factory. - pub fn new(storage_config: S3StorageConfig) -> Self { + pub fn new(mut storage_config: S3StorageConfig) -> Self { let bucket_backends = storage_config .bucket_configs() .map(|(bucket, bucket_config)| (bucket.to_string(), S3Backend::new(bucket_config))) .collect(); + storage_config.buckets.clear(); Self { default_backend: S3Backend::new(storage_config), bucket_backends, } } + + /// Returns the backend serving `uri`: the `storage.s3.buckets.` entry whose key + /// exactly matches the URI's bucket, or the primary backend otherwise. + fn backend_for_uri(&self, uri: &Uri) -> &S3Backend { + if self.bucket_backends.is_empty() { + return &self.default_backend; + } + parse_s3_uri(uri) + .and_then(|(bucket, _prefix)| self.bucket_backends.get(&bucket)) + .unwrap_or(&self.default_backend) + } } #[async_trait] @@ -73,11 +85,7 @@ impl StorageFactory for S3CompatibleObjectStorageFactory { } async fn resolve(&self, uri: &Uri) -> Result, StorageResolverError> { - // A `storage.s3.buckets.` entry, if present, supplies that bucket's own - // endpoint/credentials/region; any unlisted bucket uses the primary backend. - let backend = parse_s3_uri(uri) - .and_then(|(bucket, _prefix)| self.bucket_backends.get(&bucket)) - .unwrap_or(&self.default_backend); + let backend = self.backend_for_uri(uri); let s3_client = backend .s3_client .get_or_init(|| create_s3_client(&backend.storage_config)) @@ -89,3 +97,73 @@ impl StorageFactory for S3CompatibleObjectStorageFactory { Ok(Arc::new(DebouncedStorage::new(storage))) } } + +#[cfg(test)] +mod tests { + use std::collections::BTreeMap; + + use super::*; + + fn bucket_config(endpoint: &str) -> S3StorageConfig { + S3StorageConfig { + endpoint: Some(endpoint.to_string()), + ..Default::default() + } + } + + #[test] + fn test_s3_factory_routes_by_exact_bucket_name() { + let storage_config = S3StorageConfig { + endpoint: Some("https://primary.example.com".to_string()), + buckets: BTreeMap::from([ + ( + "logs-bucket-eu".to_string(), + bucket_config("https://eu.example.com"), + ), + ( + "my.dotted.bucket".to_string(), + bucket_config("https://dotted.example.com"), + ), + ]), + ..Default::default() + }; + let factory = S3CompatibleObjectStorageFactory::new(storage_config); + assert!(factory.default_backend.storage_config.buckets.is_empty()); + + let endpoint_for = |uri: &'static str| -> (Option, bool) { + let backend = factory.backend_for_uri(&Uri::for_test(uri)); + ( + backend.storage_config.endpoint.clone(), + backend.storage_config.is_bucket_config, + ) + }; + assert_eq!( + endpoint_for("s3://logs-bucket-eu/indexes/logs"), + (Some("https://eu.example.com".to_string()), true) + ); + assert_eq!( + endpoint_for("s3://logs-bucket-eu"), + (Some("https://eu.example.com".to_string()), true) + ); + assert_eq!( + endpoint_for("s3://my.dotted.bucket/indexes"), + (Some("https://dotted.example.com".to_string()), true) + ); + assert_eq!( + endpoint_for("s3://logs-bucket-eu-2/indexes"), + (Some("https://primary.example.com".to_string()), false) + ); + assert_eq!( + endpoint_for("s3://other-bucket/indexes"), + (Some("https://primary.example.com".to_string()), false) + ); + } + + #[test] + fn test_s3_factory_without_buckets_uses_primary_backend() { + let factory = + S3CompatibleObjectStorageFactory::new(bucket_config("https://primary.example.com")); + let backend = factory.backend_for_uri(&Uri::for_test("s3://any-bucket/indexes")); + assert!(std::ptr::eq(backend, &factory.default_backend)); + } +} From 1faeb193de5a46b53e9cd5c3eda0c91d4cb8dff5 Mon Sep 17 00:00:00 2001 From: "david.yaffe" Date: Tue, 6 Oct 2026 17:13:33 +0000 Subject: [PATCH 7/9] Test S3 buckets through node config loading and clarify docs - Load a full node config with `storage.s3.buckets` to cover validation and flavor expansion on the real load path, including a quoted numeric bucket name and a rejected `s3://` key. - Check that the nested-buckets error comes from the nested check. - Docs: QW_S3_ENDPOINT and QW_S3_FORCE_PATH_STYLE_ACCESS apply to the primary backend only; every node must declare the same `buckets` entries; quote bucket names YAML would parse as non-strings. --- docs/configuration/storage-config.md | 10 ++-- .../src/node_config/serialize.rs | 54 +++++++++++++++++++ .../quickwit-config/src/storage_config.rs | 9 +++- 3 files changed, 68 insertions(+), 5 deletions(-) diff --git a/docs/configuration/storage-config.md b/docs/configuration/storage-config.md index 9d1658a88b4..9e59ee4cfff 100644 --- a/docs/configuration/storage-config.md +++ b/docs/configuration/storage-config.md @@ -68,8 +68,8 @@ Hardcoding credentials into configuration files is not secure and strongly disco | Env variable | Description | | --- | --- | -| `QW_S3_ENDPOINT` | Custom S3 endpoint. | -| `QW_S3_FORCE_PATH_STYLE_ACCESS` | Overrides `force_path_style_access`. | +| `QW_S3_ENDPOINT` | Custom S3 endpoint. Applies to the primary `s3:` backend only, not to `buckets` entries. | +| `QW_S3_FORCE_PATH_STYLE_ACCESS` | Overrides `force_path_style_access`. Applies to the primary `s3:` backend only, not to `buckets` entries. | | `QW_S3_MAX_CONCURRENCY` | Limit the number of concurrent requests to S3 | #### Storage flavors @@ -112,7 +112,11 @@ In addition to the primary `s3:` block, you can declare per-bucket overrides und Each bucket entry accepts the same fields as the primary `s3:` block, *except* `buckets` itself (no nesting). Bucket entries do not inherit fields from the primary block. If `access_key_id` / `secret_access_key` are omitted on a bucket entry, the global AWS SDK credential chain is used (env vars, instance metadata, etc.). -Bucket entries are self-contained: the process-wide `QW_S3_ENDPOINT` and `QW_S3_FORCE_PATH_STYLE_ACCESS` overrides apply to the primary `s3:` backend only. A bucket entry always uses its own `endpoint` and `force_path_style_access` values. +Bucket entries are self-contained: the process-wide `QW_S3_ENDPOINT` and `QW_S3_FORCE_PATH_STYLE_ACCESS` overrides apply to the primary `s3:` backend only. A bucket entry always uses its own `endpoint` and `force_path_style_access` values. A bucket entry without an `endpoint` uses the default AWS endpoint for its region, even when `QW_S3_ENDPOINT` is set. + +Routing is decided from each node's own configuration. Every node that reads or writes a bucket (indexers, searchers, janitor, metastore) must declare the same `storage.s3.buckets` entry; a node without it uses the primary backend for that bucket. + +In YAML, quote bucket names that YAML would read as a number, boolean, or null (for example `"123"`, `"true"`, or `"null"`). Unquoted, they fail to parse. > Because routing keys on bucket name, a given bucket name maps to exactly one backend. If you need the *same* bucket name on two different endpoints, give the buckets distinct names. diff --git a/quickwit/quickwit-config/src/node_config/serialize.rs b/quickwit/quickwit-config/src/node_config/serialize.rs index b69c354ed58..aad9320629e 100644 --- a/quickwit/quickwit-config/src/node_config/serialize.rs +++ b/quickwit/quickwit-config/src/node_config/serialize.rs @@ -1169,6 +1169,60 @@ mod tests { ); } + #[tokio::test] + async fn test_node_config_storage_s3_buckets() { + let config_yaml = r#" + version: 0.8 + node_id: "node-1" + metastore_uri: postgres://username:password@host:port/db + storage: + s3: + endpoint: https://primary.example.com + buckets: + minio-bucket: + flavor: minio + endpoint: http://minio.example.com:9000 + "123": + endpoint: https://numeric.example.com + "#; + let config = load_node_config_with_env( + ConfigFormat::Yaml, + config_yaml.as_bytes(), + &Default::default(), + None, + ) + .await + .unwrap(); + let s3_storage_config = config.storage_configs.find_s3().unwrap(); + let minio_bucket_config = &s3_storage_config.buckets["minio-bucket"]; + assert_eq!(minio_bucket_config.region.as_deref(), Some("minio")); + assert!(minio_bucket_config.force_path_style_access); + assert!(s3_storage_config.buckets.contains_key("123")); + + let invalid_config_yaml = r#" + version: 0.8 + node_id: "node-1" + metastore_uri: postgres://username:password@host:port/db + storage: + s3: + buckets: + s3://logs-bucket: + endpoint: https://logs.example.com + "#; + let error = load_node_config_with_env( + ConfigFormat::Yaml, + invalid_config_yaml.as_bytes(), + &Default::default(), + None, + ) + .await + .unwrap_err(); + assert!( + error.to_string().contains("bare bucket name"), + "unexpected error: {error}" + ); + } + #[tokio::test] async fn test_node_config_config_default_values_default_indexer_searcher_config() { let config_yaml = r#" diff --git a/quickwit/quickwit-config/src/storage_config.rs b/quickwit/quickwit-config/src/storage_config.rs index 42505d7c73f..68d574891ff 100644 --- a/quickwit/quickwit-config/src/storage_config.rs +++ b/quickwit/quickwit-config/src/storage_config.rs @@ -944,7 +944,12 @@ mod tests { serde_yaml::from_str(s3_storage_config_yaml).unwrap(); let storage_configs = StorageConfigs::new(vec![s3_storage_config.into()]); let error = storage_configs.validate().unwrap_err(); - assert!(error.to_string().contains("logs-bucket")); + let error_message = error.to_string(); + assert!(error_message.contains("logs-bucket")); + assert!( + error_message.contains("nested"), + "unexpected error: {error_message}" + ); } #[test] @@ -952,7 +957,7 @@ mod tests { let s3_storage_config_yaml = r#" buckets: logs-bucket: - endpiont: https://typo.example.com + endpoint_url: https://typo.example.com "#; serde_yaml::from_str::(s3_storage_config_yaml).unwrap_err(); } From 8c1e34d54a944db1b9634bab9e9a2b602311fc7d Mon Sep 17 00:00:00 2001 From: "david.yaffe" Date: Tue, 6 Oct 2026 18:01:53 +0000 Subject: [PATCH 8/9] Drop redundant S3 bucket tests Remove tests already covered elsewhere: the full node-config load test (validation and flavor expansion are unit tested), the dotted-name and no-buckets factory tests (covered by the routing test), and the bucket unknown-field test (bucket entries share S3StorageConfig's deny_unknown_fields). --- .../src/node_config/serialize.rs | 54 ------------------- .../quickwit-config/src/storage_config.rs | 25 --------- .../s3_compatible_storage_resolver.rs | 8 --- 3 files changed, 87 deletions(-) diff --git a/quickwit/quickwit-config/src/node_config/serialize.rs b/quickwit/quickwit-config/src/node_config/serialize.rs index aad9320629e..b69c354ed58 100644 --- a/quickwit/quickwit-config/src/node_config/serialize.rs +++ b/quickwit/quickwit-config/src/node_config/serialize.rs @@ -1169,60 +1169,6 @@ mod tests { ); } - #[tokio::test] - async fn test_node_config_storage_s3_buckets() { - let config_yaml = r#" - version: 0.8 - node_id: "node-1" - metastore_uri: postgres://username:password@host:port/db - storage: - s3: - endpoint: https://primary.example.com - buckets: - minio-bucket: - flavor: minio - endpoint: http://minio.example.com:9000 - "123": - endpoint: https://numeric.example.com - "#; - let config = load_node_config_with_env( - ConfigFormat::Yaml, - config_yaml.as_bytes(), - &Default::default(), - None, - ) - .await - .unwrap(); - let s3_storage_config = config.storage_configs.find_s3().unwrap(); - let minio_bucket_config = &s3_storage_config.buckets["minio-bucket"]; - assert_eq!(minio_bucket_config.region.as_deref(), Some("minio")); - assert!(minio_bucket_config.force_path_style_access); - assert!(s3_storage_config.buckets.contains_key("123")); - - let invalid_config_yaml = r#" - version: 0.8 - node_id: "node-1" - metastore_uri: postgres://username:password@host:port/db - storage: - s3: - buckets: - s3://logs-bucket: - endpoint: https://logs.example.com - "#; - let error = load_node_config_with_env( - ConfigFormat::Yaml, - invalid_config_yaml.as_bytes(), - &Default::default(), - None, - ) - .await - .unwrap_err(); - assert!( - error.to_string().contains("bare bucket name"), - "unexpected error: {error}" - ); - } - #[tokio::test] async fn test_node_config_config_default_values_default_indexer_searcher_config() { let config_yaml = r#" diff --git a/quickwit/quickwit-config/src/storage_config.rs b/quickwit/quickwit-config/src/storage_config.rs index 68d574891ff..b4b8bd2277e 100644 --- a/quickwit/quickwit-config/src/storage_config.rs +++ b/quickwit/quickwit-config/src/storage_config.rs @@ -916,21 +916,6 @@ mod tests { assert!(bucket_config.is_bucket_config); } - #[test] - fn test_storage_s3_buckets_dotted_bucket_name() { - // The map key is a bucket name, so dotted bucket names are accepted as-is. - let s3_storage_config_yaml = r#" - buckets: - my.dotted.bucket: - endpoint: https://logs.example.com - "#; - let s3_storage_config: S3StorageConfig = - serde_yaml::from_str(s3_storage_config_yaml).unwrap(); - assert!(s3_storage_config.buckets.contains_key("my.dotted.bucket")); - let storage_configs = StorageConfigs::new(vec![s3_storage_config.into()]); - storage_configs.validate().unwrap(); - } - #[test] fn test_storage_s3_buckets_reject_nested_buckets() { let s3_storage_config_yaml = r#" @@ -952,16 +937,6 @@ mod tests { ); } - #[test] - fn test_storage_s3_buckets_reject_unknown_fields() { - let s3_storage_config_yaml = r#" - buckets: - logs-bucket: - endpoint_url: https://typo.example.com - "#; - serde_yaml::from_str::(s3_storage_config_yaml).unwrap_err(); - } - #[test] fn test_storage_s3_buckets_reject_unmatchable_bucket_keys() { for bucket_key in ["", "s3://logs-bucket", "logs-bucket/prefix"] { diff --git a/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage_resolver.rs b/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage_resolver.rs index a71195c07b6..c08a70b1434 100644 --- a/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage_resolver.rs +++ b/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage_resolver.rs @@ -158,12 +158,4 @@ mod tests { (Some("https://primary.example.com".to_string()), false) ); } - - #[test] - fn test_s3_factory_without_buckets_uses_primary_backend() { - let factory = - S3CompatibleObjectStorageFactory::new(bucket_config("https://primary.example.com")); - let backend = factory.backend_for_uri(&Uri::for_test("s3://any-bucket/indexes")); - assert!(std::ptr::eq(backend, &factory.default_backend)); - } } From ee11eb628b40056be1c58358aa4d2cf5677dd3d5 Mon Sep 17 00:00:00 2001 From: Harry Papadopoulos Date: Fri, 23 Oct 2026 07:35:56 +0100 Subject: [PATCH 9/9] refactor(storage): address review nits on per-bucket S3 config - Read QW_S3_ENDPOINT via get_from_env_opt. - Test env overrides through injectable helpers instead of mutating process env. - Rename S3Backend to LazyS3Client to avoid confusion with StorageBackend. - Split bucket key validation into empty-key and URI/path checks. - Reword bucket key docs. Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/configuration/storage-config.md | 2 +- .../quickwit-config/src/storage_config.rs | 103 +++++++++--------- .../s3_compatible_storage_resolver.rs | 14 +-- 3 files changed, 61 insertions(+), 58 deletions(-) diff --git a/docs/configuration/storage-config.md b/docs/configuration/storage-config.md index 96098a73804..60983091b73 100644 --- a/docs/configuration/storage-config.md +++ b/docs/configuration/storage-config.md @@ -108,7 +108,7 @@ storage: #### Per-bucket S3 configuration -In addition to the primary `s3:` block, you can declare per-bucket overrides under `storage.s3.buckets`. The map key is the bare bucket name (no `s3://` prefix or path; such keys are rejected at startup); when an `s3:///...` URI is resolved, an exact match supplies that bucket's own endpoint, credentials, region, and flags. Any bucket not listed falls back to the fields on the primary `s3:` block. URIs stay canonical `s3://` — routing is by bucket name, so nothing extra is persisted in the index metadata. +In addition to the primary `s3:` block, you can declare per-bucket overrides under `storage.s3.buckets`. The map key is the bucket name, e.g. `my-bucket`, not `s3://my-bucket` or `my-bucket/prefix`; such keys are rejected at startup. When an `s3:///...` URI is resolved, an exact match supplies that bucket's own endpoint, credentials, region, and flags. Any bucket not listed falls back to the fields on the primary `s3:` block. URIs stay canonical `s3://` — routing is by bucket name, so nothing extra is persisted in the index metadata. Each bucket entry accepts the same fields as the primary `s3:` block, *except* `buckets` itself (no nesting). Bucket entries do not inherit fields from the primary block. If `access_key_id` / `secret_access_key` are omitted on a bucket entry, the global AWS SDK credential chain is used (env vars, instance metadata, etc.). diff --git a/quickwit/quickwit-config/src/storage_config.rs b/quickwit/quickwit-config/src/storage_config.rs index b4b8bd2277e..c618cfb7c3f 100644 --- a/quickwit/quickwit-config/src/storage_config.rs +++ b/quickwit/quickwit-config/src/storage_config.rs @@ -18,7 +18,7 @@ use std::{env, fmt}; use anyhow::ensure; use itertools::Itertools; -use quickwit_common::{get_bool_from_env, get_from_env_opt}; +use quickwit_common::{get_bool_from_env_opt, get_from_env_opt}; use serde::{Deserialize, Serialize}; use serde_with::{EnumMap, serde_as}; @@ -123,10 +123,11 @@ impl StorageConfigs { } if let Some(s3_storage_config) = self.find_s3() { for (bucket, bucket_config) in &s3_storage_config.buckets { + ensure!(!bucket.is_empty(), "S3 bucket config key must not be empty"); ensure!( - !bucket.is_empty() && !bucket.contains('/'), - "S3 bucket config key `{bucket}` must be a bare bucket name, without `s3://` \ - or `/`", + !bucket.contains('/'), + "invalid S3 bucket config key `{bucket}`: expected a bucket name (e.g. \ + `my-bucket`), not a URI or path", ); ensure!( bucket_config.buckets.is_empty(), @@ -479,26 +480,37 @@ impl S3StorageConfig { } pub fn endpoint(&self) -> Option { - // `QW_S3_ENDPOINT` overrides the primary backend only; bucket configs - // are self-contained and use their own configured endpoint. - if !self.is_bucket_config - && let Ok(endpoint) = env::var("QW_S3_ENDPOINT") - { - return Some(endpoint); - } - self.endpoint.clone() + self.endpoint_with_env_override(|| get_from_env_opt("QW_S3_ENDPOINT", false)) } pub fn force_path_style_access(&self) -> Option { - // `QW_S3_FORCE_PATH_STYLE_ACCESS` overrides the primary backend only. - // No process-wide cache: each backend must honor its own setting. + self.force_path_style_access_with_env_override(|| { + get_bool_from_env_opt("QW_S3_FORCE_PATH_STYLE_ACCESS") + }) + } + + // `QW_S3_ENDPOINT` overrides the primary backend only; bucket configs + // are self-contained and use their own configured endpoint. + fn endpoint_with_env_override( + &self, + env_endpoint: impl FnOnce() -> Option, + ) -> Option { + if self.is_bucket_config { + return self.endpoint.clone(); + } + env_endpoint().or_else(|| self.endpoint.clone()) + } + + // `QW_S3_FORCE_PATH_STYLE_ACCESS` overrides the primary backend only. + // No process-wide cache: each backend must honor its own setting. + fn force_path_style_access_with_env_override( + &self, + env_force_path_style_access: impl FnOnce() -> Option, + ) -> Option { if self.is_bucket_config { return Some(self.force_path_style_access); } - Some(get_bool_from_env( - "QW_S3_FORCE_PATH_STYLE_ACCESS", - self.force_path_style_access, - )) + Some(env_force_path_style_access().unwrap_or(self.force_path_style_access)) } } @@ -939,7 +951,11 @@ mod tests { #[test] fn test_storage_s3_buckets_reject_unmatchable_bucket_keys() { - for bucket_key in ["", "s3://logs-bucket", "logs-bucket/prefix"] { + for (bucket_key, expected_error) in [ + ("", "must not be empty"), + ("s3://logs-bucket", "not a URI or path"), + ("logs-bucket/prefix", "not a URI or path"), + ] { let s3_storage_config = S3StorageConfig { buckets: BTreeMap::from([(bucket_key.to_string(), S3StorageConfig::default())]), ..Default::default() @@ -947,7 +963,7 @@ mod tests { let storage_configs = StorageConfigs::new(vec![s3_storage_config.into()]); let error = storage_configs.validate().unwrap_err(); assert!( - error.to_string().contains("bare bucket name"), + error.to_string().contains(expected_error), "unexpected error for key `{bucket_key}`: {error}" ); } @@ -1022,42 +1038,29 @@ mod tests { ..Default::default() }; let (_, bucket_config) = s3_storage_config.bucket_configs().next().unwrap(); - - // SAFETY: this test may not be entirely sound if not run with nextest or --test-threads=1 - // as this is only a test, and it would be extremely inconvenient to run it in a different - // way, we are keeping it that way - let previous_endpoint = env::var("QW_S3_ENDPOINT").ok(); - let previous_force_path_style = env::var("QW_S3_FORCE_PATH_STYLE_ACCESS").ok(); - unsafe { - env::set_var("QW_S3_ENDPOINT", "https://env.example.com"); - env::set_var("QW_S3_FORCE_PATH_STYLE_ACCESS", "true"); - } - let primary_endpoint = s3_storage_config.endpoint(); - let primary_force_path_style = s3_storage_config.force_path_style_access(); - let bucket_endpoint = bucket_config.endpoint(); - let bucket_force_path_style = bucket_config.force_path_style_access(); - unsafe { - match previous_endpoint { - Some(endpoint) => env::set_var("QW_S3_ENDPOINT", endpoint), - None => env::remove_var("QW_S3_ENDPOINT"), - } - match previous_force_path_style { - Some(force_path_style) => { - env::set_var("QW_S3_FORCE_PATH_STYLE_ACCESS", force_path_style) - } - None => env::remove_var("QW_S3_FORCE_PATH_STYLE_ACCESS"), - } - } + let env_endpoint = || Some("https://env.example.com".to_string()); + let env_force_path_style_access = || Some(true); assert_eq!( - primary_endpoint, + s3_storage_config.endpoint_with_env_override(env_endpoint), Some("https://env.example.com".to_string()) ); - assert_eq!(primary_force_path_style, Some(true)); assert_eq!( - bucket_endpoint, + s3_storage_config + .force_path_style_access_with_env_override(env_force_path_style_access), + Some(true) + ); + assert_eq!( + bucket_config.endpoint_with_env_override(env_endpoint), Some("https://bucket.example.com".to_string()) ); - assert_eq!(bucket_force_path_style, Some(false)); + assert_eq!( + bucket_config.force_path_style_access_with_env_override(env_force_path_style_access), + Some(false) + ); + assert_eq!( + s3_storage_config.endpoint_with_env_override(|| None), + Some("https://primary.example.com".to_string()) + ); } } diff --git a/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage_resolver.rs b/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage_resolver.rs index c08a70b1434..208b461ad1c 100644 --- a/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage_resolver.rs +++ b/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage_resolver.rs @@ -28,12 +28,12 @@ use crate::{ /// S3 compatible object storage resolver. pub struct S3CompatibleObjectStorageFactory { - default_backend: S3Backend, + default_backend: LazyS3Client, // Backends for the buckets listed under `storage.s3.buckets`, keyed by bucket name. - bucket_backends: HashMap, + bucket_backends: HashMap, } -struct S3Backend { +struct LazyS3Client { storage_config: S3StorageConfig, // we cache the S3Client so we don't rebuild one every time we build a new Storage (for // every search query). @@ -43,7 +43,7 @@ struct S3Backend { s3_client: OnceCell, } -impl S3Backend { +impl LazyS3Client { fn new(storage_config: S3StorageConfig) -> Self { Self { storage_config, @@ -57,18 +57,18 @@ impl S3CompatibleObjectStorageFactory { pub fn new(mut storage_config: S3StorageConfig) -> Self { let bucket_backends = storage_config .bucket_configs() - .map(|(bucket, bucket_config)| (bucket.to_string(), S3Backend::new(bucket_config))) + .map(|(bucket, bucket_config)| (bucket.to_string(), LazyS3Client::new(bucket_config))) .collect(); storage_config.buckets.clear(); Self { - default_backend: S3Backend::new(storage_config), + default_backend: LazyS3Client::new(storage_config), bucket_backends, } } /// Returns the backend serving `uri`: the `storage.s3.buckets.` entry whose key /// exactly matches the URI's bucket, or the primary backend otherwise. - fn backend_for_uri(&self, uri: &Uri) -> &S3Backend { + fn backend_for_uri(&self, uri: &Uri) -> &LazyS3Client { if self.bucket_backends.is_empty() { return &self.default_backend; }