diff --git a/docs/configuration/storage-config.md b/docs/configuration/storage-config.md index dc0e2628c75..60983091b73 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 +- `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:///...` 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 | diff --git a/quickwit/quickwit-config/src/storage_config.rs b/quickwit/quickwit-config/src/storage_config.rs index c164e9419e5..c618cfb7c3f 100644 --- a/quickwit/quickwit-config/src/storage_config.rs +++ b/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:///...` 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, + /// 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 + '_ { + 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 { - 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 { - 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) + 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(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()) + ); + } } 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..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 @@ -12,6 +12,7 @@ // 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; @@ -20,13 +21,19 @@ 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, }; /// S3 compatible object storage resolver. pub struct S3CompatibleObjectStorageFactory { + default_backend: LazyS3Client, + // Backends for the buckets listed under `storage.s3.buckets`, keyed by bucket name. + bucket_backends: HashMap, +} + +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). @@ -36,9 +43,8 @@ pub struct S3CompatibleObjectStorageFactory { s3_client: OnceCell, } -impl S3CompatibleObjectStorageFactory { - /// Creates a new S3-compatible storage factory. - pub fn new(storage_config: S3StorageConfig) -> Self { +impl LazyS3Client { + fn new(storage_config: S3StorageConfig) -> Self { Self { storage_config, s3_client: OnceCell::new(), @@ -46,6 +52,32 @@ impl S3CompatibleObjectStorageFactory { } } +impl S3CompatibleObjectStorageFactory { + /// Creates a new S3-compatible storage factory. + pub fn new(mut storage_config: S3StorageConfig) -> Self { + let bucket_backends = storage_config + .bucket_configs() + .map(|(bucket, bucket_config)| (bucket.to_string(), LazyS3Client::new(bucket_config))) + .collect(); + storage_config.buckets.clear(); + Self { + 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) -> &LazyS3Client { + 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] impl StorageFactory for S3CompatibleObjectStorageFactory { fn backend(&self) -> StorageBackend { @@ -53,14 +85,77 @@ impl StorageFactory for S3CompatibleObjectStorageFactory { } async fn resolve(&self, uri: &Uri) -> Result, StorageResolverError> { - let s3_client = self + let backend = self.backend_for_uri(uri); + 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))) } } + +#[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) + ); + } +}