From bfcc308e399f4046e285c49d5e87fbc8a1d4f3be Mon Sep 17 00:00:00 2001 From: "david.yaffe" Date: Fri, 2 Oct 2026 19:20:55 +0000 Subject: [PATCH 1/4] 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 2/4] 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 3/4] 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 4/4] 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)); - } }