diff --git a/docs/configuration/storage-config.md b/docs/configuration/storage-config.md index 7bc995543e6..9e59ee4cfff 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 @@ -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 S3 | #### Storage flavors @@ -104,23 +106,27 @@ 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 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 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. 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 `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 +141,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..b4b8bd2277e 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,19 @@ 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() && !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`", + ); + } + } Ok(()) } @@ -401,116 +415,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 +454,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 +492,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 +527,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 +881,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 +899,165 @@ 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); - - // `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()); + 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_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_reject_nested_buckets() { let s3_storage_config_yaml = r#" - profiles: - my.dotted.bucket: - endpoint: https://logs.example.com + 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(); - assert!(s3_storage_config.profiles.contains_key("my.dotted.bucket")); let storage_configs = StorageConfigs::new(vec![s3_storage_config.into()]); - storage_configs.validate().unwrap(); + 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_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***")); + 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_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#" - profiles: - 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 - "#; - 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()); + 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_profile_applies_flavor() { - // `flavor` shortcuts expand for profiles just like the primary backend. + 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 { + 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 projected = profile.as_s3_config(); - assert!(projected.is_profile); + 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!( - projected.endpoint(), - Some("https://profile.example.com".to_string()) + bucket_endpoint, + Some("https://bucket.example.com".to_string()) ); + 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 50b70ac03d4..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 @@ -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,22 +41,43 @@ 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 S3CompatibleObjectStorageFactory { - /// Creates a new S3-compatible storage factory. - pub fn new(storage_config: S3StorageConfig) -> Self { +impl S3Backend { + fn new(storage_config: S3StorageConfig) -> Self { Self { storage_config, s3_client: OnceCell::new(), - profile_s3_clients: Mutex::new(HashMap::new()), } } } +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(), 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] impl StorageFactory for S3CompatibleObjectStorageFactory { fn backend(&self) -> StorageBackend { @@ -58,36 +85,77 @@ 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 + 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) + ); + } +}