feat(storage): add per-bucket S3 configuration via storage.s3.buckets

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
+- `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
@@ -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.
@@ -67,7 +68,8 @@ Hardcoding credentials into configuration files is not secure and strongly disco
 
 | Env variable | Description |
 | --- | --- |
-| `QW_S3_ENDPOINT` | Custom S3 endpoint. |
+| `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 object storages, including S3 and GCS |
 
 #### Storage flavors
@@ -104,6 +106,48 @@ storage:
     endpoint: https://storage.googleapis.com
 ```
 
+#### 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, e.g. `my-bucket`, not `s3://my-bucket` or `my-bucket/prefix`; such keys are rejected at startup. When an `s3://<bucket>/...` 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.).
+
+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.
+
+```yaml
+storage:
+  s3:
+    # Primary backend — used for any bucket not listed under `buckets`.
+    endpoint: https://s3.us-east-1.amazonaws.com
+    region: us-east-1
+    buckets:
+      # 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}
+      # 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 simply points at the bucket by its canonical URI; the matching bucket entry is applied automatically:
+
+```yaml
+index_id: logs-eu
+index_uri: s3://logs-bucket-eu/logs-eu
+```
+
 ### Azure storage configuration
 
 | Property | Description | Default value |

quickwit/quickwit-config/src/storage_config.rs

@@ -12,13 +12,13 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::collections::BTreeMap;
 use std::ops::Deref;
-use std::sync::OnceLock;
 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};
 
@@ -121,6 +121,20 @@ 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.is_empty(), "S3 bucket config key must not be empty");
+                ensure!(
+                    !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(),
+                    "S3 bucket config `{bucket}` cannot define nested `buckets`",
+                );
+            }
+        }
         Ok(())
     }
 
@@ -399,6 +413,19 @@ pub struct S3StorageConfig {
     pub disable_stalled_stream_protection_upload: bool,
     #[serde(default)]
     pub disable_stalled_stream_protection_download: bool,
+    /// Per-bucket S3-compatible backend overrides, keyed by bucket name. When an
+    /// `s3://<bucket>/...` 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. Bucket configs
+    /// cannot themselves define `buckets`.
+    #[serde(default)]
+    #[serde(skip_serializing_if = "BTreeMap::is_empty")]
+    pub buckets: BTreeMap<String, S3StorageConfig>,
+    /// 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_bucket_config: bool,
 }
 
 impl S3StorageConfig {
@@ -428,30 +455,63 @@ 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 bucket_config in self.buckets.values_mut() {
+            bucket_config.redact();
+        }
+    }
+
+    pub fn bucket_configs(&self) -> impl Iterator<Item = (&str, S3StorageConfig)> + '_ {
+        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<String> {
-        env::var("QW_S3_ENDPOINT")
-            .ok()
-            .or_else(|| self.endpoint.clone())
+        self.endpoint_with_env_override(|| get_from_env_opt("QW_S3_ENDPOINT", false))
     }
 
     pub fn force_path_style_access(&self) -> Option<bool> {
-        static FORCE_PATH_STYLE: OnceLock<Option<bool>> = 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)
+        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<String>,
+    ) -> Option<String> {
+        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<bool>,
+    ) -> Option<bool> {
+        if self.is_bucket_config {
+            return Some(self.force_path_style_access);
+        }
+        Some(env_force_path_style_access().unwrap_or(self.force_path_style_access))
+    }
 }
 
 impl fmt::Debug for S3StorageConfig {
@@ -479,6 +539,7 @@ impl fmt::Debug for S3StorageConfig {
                 "disable_stalled_stream_protection_download",
                 &self.disable_stalled_stream_protection_download,
             )
+            .field("buckets", &self.buckets)
             .finish()
     }
 }
@@ -830,4 +891,176 @@ mod tests {
             assert_eq!(s3_storage_config.flavor, Some(StorageBackendFlavor::MinIO));
         }
     }
+
+    #[test]
+    fn test_storage_s3_buckets_serde() {
+        let s3_storage_config_yaml = r#"
+            endpoint: https://primary.example.com
+            region: us-east-1
+            buckets:
+              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
+              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.buckets.len(), 2);
+
+        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);
+
+        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_buckets_reject_nested_buckets() {
+        let s3_storage_config_yaml = r#"
+            buckets:
+              logs-bucket:
+                buckets:
+                  nested-bucket:
+                    endpoint: https://nested.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();
+        let error_message = error.to_string();
+        assert!(error_message.contains("logs-bucket"));
+        assert!(
+            error_message.contains("nested"),
+            "unexpected error: {error_message}"
+        );
+    }
+
+    #[test]
+    fn test_storage_s3_buckets_reject_unmatchable_bucket_keys() {
+        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()
+            };
+            let storage_configs = StorageConfigs::new(vec![s3_storage_config.into()]);
+            let error = storage_configs.validate().unwrap_err();
+            assert!(
+                error.to_string().contains(expected_error),
+                "unexpected error for key `{bucket_key}`: {error}"
+            );
+        }
+    }
+
+    #[test]
+    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#"
+            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 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_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 {
+            endpoint: Some("https://primary.example.com".to_string()),
+            buckets: BTreeMap::from([(
+                "logs-bucket".to_string(),
+                S3StorageConfig {
+                    endpoint: Some("https://bucket.example.com".to_string()),
+                    ..Default::default()
+                },
+            )]),
+            ..Default::default()
+        };
+        let (_, bucket_config) = s3_storage_config.bucket_configs().next().unwrap();
+        let env_endpoint = || Some("https://env.example.com".to_string());
+        let env_force_path_style_access = || Some(true);
+
+        assert_eq!(
+            s3_storage_config.endpoint_with_env_override(env_endpoint),
+            Some("https://env.example.com".to_string())
+        );
+        assert_eq!(
+            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_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())
+        );
+    }
 }