From ac65655b1c9c1d9513ab6353e8165ce00a766a64 Mon Sep 17 00:00:00 2001 From: Darkheir Date: Fri, 23 Oct 2026 16:13:38 +0200 Subject: [PATCH 1/4] refactor: Group node-level S3 storage settings in S3NodeOptions The S3 assembly thread pool is a setting of the node building the storage resolver, not of the storage configs. Move it into an `S3NodeOptions` struct handed to `StorageResolver::configured_with_s3_node_options`, so other node-level S3 settings can be added without a new resolver constructor. No behavior change. Co-Authored-By: Claude Opus 5.5 --- quickwit/quickwit-cli/src/lib.rs | 38 ++++++++-------- quickwit/quickwit-cli/src/service.rs | 5 +-- quickwit/quickwit-cli/src/tool.rs | 44 +++++++++++++------ quickwit/quickwit-storage/src/lib.rs | 2 +- .../src/object_storage/mod.rs | 2 +- .../s3_compatible_storage_resolver.rs | 29 ++++++++---- .../quickwit-storage/src/storage_resolver.rs | 32 +++++--------- 7 files changed, 86 insertions(+), 66 deletions(-) diff --git a/quickwit/quickwit-cli/src/lib.rs b/quickwit/quickwit-cli/src/lib.rs index a1a975cc75b..fb0eb1e57b0 100644 --- a/quickwit/quickwit-cli/src/lib.rs +++ b/quickwit/quickwit-cli/src/lib.rs @@ -35,7 +35,7 @@ use quickwit_metastore::{IndexMetadataResponseExt, MetastoreResolver}; use quickwit_proto::metastore::{IndexMetadataRequest, MetastoreService, MetastoreServiceClient}; use quickwit_rest_client::models::Timeout; use quickwit_rest_client::rest_client::{DEFAULT_BASE_URL, QuickwitClient, QuickwitClientBuilder}; -use quickwit_storage::{StorageResolver, load_file}; +use quickwit_storage::{S3NodeOptions, StorageResolver, load_file}; use reqwest::Url; use tabled::settings::object::Rows; use tabled::settings::panel::Header; @@ -247,38 +247,35 @@ async fn load_node_config(config_uri: &Uri) -> anyhow::Result { Ok(config) } -/// Builds the storage and metastore resolvers. -/// -/// When `s3_assembly_thread_pool_opt` is set, S3 storages assemble large downloads on that pool -/// instead of the tokio runtime. Only searcher-only nodes set it (see -/// [`s3_assembly_thread_pool_for_services`]). +/// Builds the storage and metastore resolvers. `s3_node_options` are the S3 storage settings that +/// depend on the node rather than on the storage configs (see [`s3_node_options`]). fn get_resolvers( storage_configs: &StorageConfigs, metastore_configs: &MetastoreConfigs, - s3_assembly_thread_pool_opt: Option, + s3_node_options: S3NodeOptions, ) -> (StorageResolver, MetastoreResolver) { // The CLI tests rely on the unconfigured singleton resolvers, so it's better to return them if // the storage and metastore configs are not set. - if storage_configs.is_empty() - && metastore_configs.is_empty() - && s3_assembly_thread_pool_opt.is_none() - { + if storage_configs.is_empty() && metastore_configs.is_empty() && s3_node_options.is_empty() { return ( StorageResolver::unconfigured(), MetastoreResolver::unconfigured(), ); } - let storage_resolver = match s3_assembly_thread_pool_opt { - Some(thread_pool) => { - StorageResolver::configured_with_s3_assembly_thread_pool(storage_configs, thread_pool) - } - None => StorageResolver::configured(storage_configs), - }; + let storage_resolver = + StorageResolver::configured_with_s3_node_options(storage_configs, s3_node_options); let metastore_resolver = MetastoreResolver::configured(storage_resolver.clone(), metastore_configs); (storage_resolver, metastore_resolver) } +/// Returns the S3 storage settings of a node running `node_config`. +fn s3_node_options(node_config: &NodeConfig) -> S3NodeOptions { + S3NodeOptions { + assembly_thread_pool: s3_assembly_thread_pool_for_services(&node_config.enabled_services), + } +} + /// Returns the thread pool on which S3 storages should assemble large downloads, if any. /// /// Only nodes that run the searcher service alone get one: the search thread pool. On such nodes @@ -553,7 +550,10 @@ mod tests { }; let storage_configs = StorageConfigs::new(vec![s3_storage_config.into()]); let metastore_configs = MetastoreConfigs::default(); - let (_storage_resolver, _metastore_resolver) = - get_resolvers(&storage_configs, &metastore_configs, None); + let (_storage_resolver, _metastore_resolver) = get_resolvers( + &storage_configs, + &metastore_configs, + S3NodeOptions::default(), + ); } } diff --git a/quickwit/quickwit-cli/src/service.rs b/quickwit/quickwit-cli/src/service.rs index e02c5c6de58..c0a9f19e019 100644 --- a/quickwit/quickwit-cli/src/service.rs +++ b/quickwit/quickwit-cli/src/service.rs @@ -30,8 +30,7 @@ use tracing::{debug, info}; use crate::checklist::{BLUE_COLOR, RED_COLOR}; use crate::{ - config_cli_arg, get_resolvers, load_node_config, s3_assembly_thread_pool_for_services, - start_actor_runtimes, + config_cli_arg, get_resolvers, load_node_config, s3_node_options, start_actor_runtimes, }; pub fn build_run_command() -> Command { @@ -118,7 +117,7 @@ impl RunCliCommand { let (storage_resolver, metastore_resolver) = get_resolvers( &node_config.storage_configs, &node_config.metastore_configs, - s3_assembly_thread_pool_for_services(&node_config.enabled_services), + s3_node_options(&node_config), ); crate::busy_detector::set_enabled(true); // TODO move in serve quickwit? diff --git a/quickwit/quickwit-cli/src/tool.rs b/quickwit/quickwit-cli/src/tool.rs index 57115dbde68..e9921aef50d 100644 --- a/quickwit/quickwit-cli/src/tool.rs +++ b/quickwit/quickwit-cli/src/tool.rs @@ -55,7 +55,7 @@ use quickwit_search::{SearchResponseRest, single_node_search}; use quickwit_serve::{ BodyFormat, SearchRequestQueryString, SortBy, search_request_from_api_request, }; -use quickwit_storage::{BundleStorage, Storage}; +use quickwit_storage::{BundleStorage, S3NodeOptions, Storage}; use tantivy::Index; use tantivy::directory::FileSlice; use tantivy::schema::FieldType; @@ -585,8 +585,11 @@ pub async fn local_ingest_docs_cli(args: LocalIngestDocsArgs) -> anyhow::Result< println!("❯ Ingesting documents locally..."); let config = load_node_config(&args.config_uri).await?; - let (storage_resolver, metastore_resolver) = - get_resolvers(&config.storage_configs, &config.metastore_configs, None); + let (storage_resolver, metastore_resolver) = get_resolvers( + &config.storage_configs, + &config.metastore_configs, + S3NodeOptions::default(), + ); let mut metastore = metastore_resolver.resolve(&config.metastore_uri).await?; let source_params = if let Some(uri) = args.input_path_opt.as_ref() { @@ -715,8 +718,11 @@ pub async fn local_search_cli(args: LocalSearchArgs) -> anyhow::Result<()> { debug!(args=?args, "local-search"); println!("❯ Searching directly on the index storage (without calling REST API)..."); let config = load_node_config(&args.config_uri).await?; - let (storage_resolver, metastore_resolver) = - get_resolvers(&config.storage_configs, &config.metastore_configs, None); + let (storage_resolver, metastore_resolver) = get_resolvers( + &config.storage_configs, + &config.metastore_configs, + S3NodeOptions::default(), + ); let metastore: MetastoreServiceClient = metastore_resolver.resolve(&config.metastore_uri).await?; let aggs = args @@ -754,8 +760,11 @@ pub async fn merge_cli(args: MergeArgs) -> anyhow::Result<()> { debug!(args=?args, "run-merge-operations"); println!("❯ Merging splits locally..."); let config = load_node_config(&args.config_uri).await?; - let (storage_resolver, metastore_resolver) = - get_resolvers(&config.storage_configs, &config.metastore_configs, None); + let (storage_resolver, metastore_resolver) = get_resolvers( + &config.storage_configs, + &config.metastore_configs, + S3NodeOptions::default(), + ); let mut metastore = metastore_resolver.resolve(&config.metastore_uri).await?; run_index_checklist(&mut metastore, &storage_resolver, &args.index_id, None).await?; // The indexing service needs to update its cluster chitchat state so that the control plane is @@ -842,8 +851,11 @@ pub async fn merge_mature_cli(args: MatureMergeArgs) -> anyhow::Result<()> { info!(merge_config=?args.merge_config, "merge-mature configuration"); println!("❯ Scanning all indexes for mature merge opportunities..."); let config = load_node_config(&args.config_uri).await?; - let (storage_resolver, metastore_resolver) = - get_resolvers(&config.storage_configs, &config.metastore_configs, None); + let (storage_resolver, metastore_resolver) = get_resolvers( + &config.storage_configs, + &config.metastore_configs, + S3NodeOptions::default(), + ); let metastore = metastore_resolver.resolve(&config.metastore_uri).await?; let runtimes_config = RuntimesConfig::default(); @@ -874,8 +886,11 @@ pub async fn garbage_collect_index_cli(args: GarbageCollectIndexArgs) -> anyhow: println!("❯ Garbage collecting index..."); let config = load_node_config(&args.config_uri).await?; - let (storage_resolver, metastore_resolver) = - get_resolvers(&config.storage_configs, &config.metastore_configs, None); + let (storage_resolver, metastore_resolver) = get_resolvers( + &config.storage_configs, + &config.metastore_configs, + S3NodeOptions::default(), + ); let metastore = metastore_resolver.resolve(&config.metastore_uri).await?; let mut index_service = IndexService::new(metastore, storage_resolver); let removal_info = index_service @@ -939,8 +954,11 @@ async fn extract_split_cli(args: ExtractSplitArgs) -> anyhow::Result<()> { println!("❯ Extracting split..."); let config = load_node_config(&args.config_uri).await?; - let (storage_resolver, metastore_resolver) = - get_resolvers(&config.storage_configs, &config.metastore_configs, None); + let (storage_resolver, metastore_resolver) = get_resolvers( + &config.storage_configs, + &config.metastore_configs, + S3NodeOptions::default(), + ); let metastore = metastore_resolver.resolve(&config.metastore_uri).await?; let index_metadata = metastore .index_metadata(IndexMetadataRequest::for_index_id(args.index_id)) diff --git a/quickwit/quickwit-storage/src/lib.rs b/quickwit/quickwit-storage/src/lib.rs index 524374917e7..3423ff2d10a 100644 --- a/quickwit/quickwit-storage/src/lib.rs +++ b/quickwit/quickwit-storage/src/lib.rs @@ -72,7 +72,7 @@ pub use self::local_file_storage::{LocalFileStorage, LocalFileStorageFactory}; #[cfg(feature = "azure")] pub use self::object_storage::{AzureBlobStorage, AzureBlobStorageFactory}; pub use self::object_storage::{ - MultiPartPolicy, S3CompatibleObjectStorage, S3CompatibleObjectStorageFactory, + MultiPartPolicy, S3CompatibleObjectStorage, S3CompatibleObjectStorageFactory, S3NodeOptions, }; #[cfg(feature = "gcs")] pub use self::opendal_storage::GoogleCloudStorageFactory; diff --git a/quickwit/quickwit-storage/src/object_storage/mod.rs b/quickwit/quickwit-storage/src/object_storage/mod.rs index e914c107291..bf712520f15 100644 --- a/quickwit/quickwit-storage/src/object_storage/mod.rs +++ b/quickwit/quickwit-storage/src/object_storage/mod.rs @@ -16,7 +16,7 @@ mod error; mod s3_compatible_storage; pub use self::s3_compatible_storage::S3CompatibleObjectStorage; -pub use self::s3_compatible_storage_resolver::S3CompatibleObjectStorageFactory; +pub use self::s3_compatible_storage_resolver::{S3CompatibleObjectStorageFactory, S3NodeOptions}; mod policy; pub use crate::object_storage::policy::MultiPartPolicy; 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 465904322e0..ebb20a31698 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 @@ -26,6 +26,21 @@ use crate::{ DebouncedStorage, S3CompatibleObjectStorage, Storage, StorageFactory, StorageResolverError, }; +/// Settings of the S3 storages that belong to the node building them rather than to the S3 +/// storage config. +#[derive(Clone, Default)] +pub struct S3NodeOptions { + /// See [`S3CompatibleObjectStorage::with_assembly_thread_pool`]. + pub assembly_thread_pool: Option, +} + +impl S3NodeOptions { + /// Whether no option is set. + pub fn is_empty(&self) -> bool { + self.assembly_thread_pool.is_none() + } +} + /// S3 compatible object storage resolver. pub struct S3CompatibleObjectStorageFactory { storage_config: S3StorageConfig, @@ -35,8 +50,7 @@ pub struct S3CompatibleObjectStorageFactory { // end up being used, or if something like azure, gcs, or even local files, will be used // instead. s3_client: OnceCell, - /// See [`S3CompatibleObjectStorage::with_assembly_thread_pool`]. - assembly_thread_pool: Option, + node_options: S3NodeOptions, } impl S3CompatibleObjectStorageFactory { @@ -45,15 +59,14 @@ impl S3CompatibleObjectStorageFactory { Self { storage_config, s3_client: OnceCell::new(), - assembly_thread_pool: None, + node_options: S3NodeOptions::default(), } } - /// Every storage resolved by this factory assembles large multi-segment response bodies on - /// `thread_pool`. See [`S3CompatibleObjectStorage::with_assembly_thread_pool`]. - pub fn with_assembly_thread_pool(self, thread_pool: ThreadPoolWithPriority) -> Self { + /// Applies `node_options` to every storage resolved by this factory. + pub fn with_node_options(self, node_options: S3NodeOptions) -> Self { Self { - assembly_thread_pool: Some(thread_pool), + node_options, ..self } } @@ -74,7 +87,7 @@ impl StorageFactory for S3CompatibleObjectStorageFactory { let mut storage = S3CompatibleObjectStorage::from_uri_and_client(&self.storage_config, uri, s3_client) .await?; - if let Some(thread_pool) = &self.assembly_thread_pool { + if let Some(thread_pool) = &self.node_options.assembly_thread_pool { storage = storage.with_assembly_thread_pool(thread_pool.clone()); } Ok(Arc::new(DebouncedStorage::new(storage))) diff --git a/quickwit/quickwit-storage/src/storage_resolver.rs b/quickwit/quickwit-storage/src/storage_resolver.rs index f654a86bd72..d21412e08dc 100644 --- a/quickwit/quickwit-storage/src/storage_resolver.rs +++ b/quickwit/quickwit-storage/src/storage_resolver.rs @@ -17,7 +17,6 @@ use std::fmt; use std::sync::Arc; use once_cell::sync::Lazy; -use quickwit_common::thread_pool::with_priority::ThreadPoolWithPriority; use quickwit_common::uri::{Protocol, Uri}; use quickwit_config::{StorageBackend, StorageConfigs}; @@ -27,7 +26,9 @@ use crate::AzureBlobStorageFactory; use crate::GoogleCloudStorageFactory; use crate::local_file_storage::LocalFileStorageFactory; use crate::ram_storage::RamStorageFactory; -use crate::{S3CompatibleObjectStorageFactory, Storage, StorageFactory, StorageResolverError}; +use crate::{ + S3CompatibleObjectStorageFactory, S3NodeOptions, Storage, StorageFactory, StorageResolverError, +}; /// Returns the [`Storage`] instance associated with the protocol of a URI. The actual creation of /// storage objects is delegated to pre-registered [`StorageFactory`]. The resolver is only @@ -87,30 +88,19 @@ impl StorageResolver { /// Creates and returns a [`StorageResolver`]. pub fn configured(storage_configs: &StorageConfigs) -> Self { - Self::configured_with_options(storage_configs, None) + Self::configured_with_s3_node_options(storage_configs, S3NodeOptions::default()) } - /// Same as [`StorageResolver::configured`], but S3 storages assemble large multi-segment - /// downloads on `s3_assembly_thread_pool` instead of the tokio runtime. This is a property - /// of the node building the resolver, not of the storage configs: searcher-only nodes hand - /// over their search thread pool. - pub fn configured_with_s3_assembly_thread_pool( + /// Same as [`StorageResolver::configured`], with S3 storage settings that are properties of + /// the node building the resolver rather than of the storage configs. + pub fn configured_with_s3_node_options( storage_configs: &StorageConfigs, - s3_assembly_thread_pool: ThreadPoolWithPriority, + s3_node_options: S3NodeOptions, ) -> Self { - Self::configured_with_options(storage_configs, Some(s3_assembly_thread_pool)) - } - - fn configured_with_options( - storage_configs: &StorageConfigs, - s3_assembly_thread_pool_opt: Option, - ) -> Self { - let mut s3_storage_factory = S3CompatibleObjectStorageFactory::new( + let s3_storage_factory = S3CompatibleObjectStorageFactory::new( storage_configs.find_s3().cloned().unwrap_or_default(), - ); - if let Some(thread_pool) = s3_assembly_thread_pool_opt { - s3_storage_factory = s3_storage_factory.with_assembly_thread_pool(thread_pool); - } + ) + .with_node_options(s3_node_options); let mut builder = StorageResolver::builder() .register(LocalFileStorageFactory) .register(RamStorageFactory::default()) From f7dd2787f1ca49d3aabb79abf1eecf63e0a5e64f Mon Sep 17 00:00:00 2001 From: Darkheir Date: Fri, 23 Oct 2026 17:11:11 +0200 Subject: [PATCH 2/4] refactor: Classify S3 SDK errors for retries in a single helper The same transient/permanent classification of SDK errors was repeated in the put, upload part and get paths. Move it to `into_retry`. No behavior change. Co-Authored-By: Claude Opus 5.5 --- .../object_storage/s3_compatible_storage.rs | 37 ++++++++----------- 1 file changed, 16 insertions(+), 21 deletions(-) diff --git a/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage.rs b/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage.rs index e582396fc45..7c9b11f3677 100644 --- a/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage.rs +++ b/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage.rs @@ -146,6 +146,19 @@ impl AsyncRead for S3AsyncRead { } } +/// Classifies an SDK error for the retry loop: transient or permanent. +fn into_retry(sdk_error: SdkError) -> Retry +where + SdkError: AwsRetryable, + StorageError: From>, +{ + if sdk_error.is_retryable() { + Retry::Transient(StorageError::from(sdk_error)) + } else { + Retry::Permanent(StorageError::from(sdk_error)) + } +} + /// S3-compatible object storage implementation. pub struct S3CompatibleObjectStorage { s3_client: S3Client, @@ -405,13 +418,7 @@ impl S3CompatibleObjectStorage { .send() .with_count_and_upload_metrics(ActionLabel::PutObject, bucket, len) .await - .map_err(|sdk_error| { - if sdk_error.is_retryable() { - Retry::Transient(StorageError::from(sdk_error)) - } else { - Retry::Permanent(StorageError::from(sdk_error)) - } - })?; + .map_err(into_retry)?; Ok(()) } @@ -580,13 +587,7 @@ impl S3CompatibleObjectStorage { .send() .with_count_and_upload_metrics(ActionLabel::UploadPart, &self.bucket, part.len()) .await - .map_err(|s3_err| { - if s3_err.is_retryable() { - Retry::Transient(StorageError::from(s3_err)) - } else { - Retry::Permanent(StorageError::from(s3_err)) - } - })?; + .map_err(into_retry)?; let completed_part = CompletedPart::builder() .set_e_tag(upload_part_output.e_tag) @@ -742,13 +743,7 @@ impl S3CompatibleObjectStorage { let (get_object_output, _connection_permit) = self .get_object(path, range_opt.clone()) .await - .map_err(|sdk_error| { - if sdk_error.is_retryable() { - Retry::Transient(StorageError::from(sdk_error)) - } else { - Retry::Permanent(StorageError::from(sdk_error)) - } - })?; + .map_err(into_retry)?; // only record ranged get request as being in flight let _in_flight_guards = range_opt .clone() From 742128598d80962c7d2a4db421ab393f7e2f1296 Mon Sep 17 00:00:00 2001 From: Darkheir Date: Fri, 23 Oct 2026 17:11:44 +0200 Subject: [PATCH 3/4] refactor: Extract a single S3 GetObject attempt from get_to_bytes Move the body of the retried closure of `get_to_bytes`, the GetObject request and the read of its body, to `get_to_bytes_attempt`. No behavior change. Co-Authored-By: Claude Opus 5.5 --- .../object_storage/s3_compatible_storage.rs | 55 +++++++++++-------- 1 file changed, 31 insertions(+), 24 deletions(-) diff --git a/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage.rs b/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage.rs index 7c9b11f3677..0ea4643183d 100644 --- a/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage.rs +++ b/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage.rs @@ -726,39 +726,46 @@ impl S3CompatibleObjectStorage { Ok((get_object_output, connection_permit)) } - async fn get_to_bytes( + /// A single attempt of [`Self::get_to_bytes`]. The body is read within the attempt: a + /// connection dropped or stalled while streaming the body fails it, and the whole GET is sent + /// again. + async fn get_to_bytes_attempt( &self, path: &Path, range_opt: Option>, - ) -> StorageResult { + ) -> Result> { let download_kind = if range_opt.is_some() { DownloadKind::Slice } else { DownloadKind::Object }; - // The body is read within the retried closure: a connection dropped or stalled while - // streaming the body fails the attempt, and the whole GET is sent again. - let payload = - aws_retry_with_metrics(&self.retry_params, ActionLabel::GetObject, || async { - let (get_object_output, _connection_permit) = self - .get_object(path, range_opt.clone()) - .await - .map_err(into_retry)?; - // only record ranged get request as being in flight - let _in_flight_guards = range_opt - .clone() - .map(|range| object_storage_get_slice_in_flight_guards(range.len())); - collect_with_download_metrics( - get_object_output.body, - download_kind, - self.assembly_thread_pool.as_ref(), - ) - .await - // GET requests are idempotent, and a body error is a transport failure. - .map_err(|io_error| Retry::Transient(StorageError::from(io_error))) - }) + let (get_object_output, _connection_permit) = self + .get_object(path, range_opt.clone()) .await - .map_err(Retry::into_inner)?; + .map_err(into_retry)?; + // only record ranged get request as being in flight + let _in_flight_guards = + range_opt.map(|range| object_storage_get_slice_in_flight_guards(range.len())); + collect_with_download_metrics( + get_object_output.body, + download_kind, + self.assembly_thread_pool.as_ref(), + ) + .await + // GET requests are idempotent, and a body error is a transport failure. + .map_err(|io_error| Retry::Transient(StorageError::from(io_error))) + } + + async fn get_to_bytes( + &self, + path: &Path, + range_opt: Option>, + ) -> StorageResult { + let payload = aws_retry_with_metrics(&self.retry_params, ActionLabel::GetObject, || { + self.get_to_bytes_attempt(path, range_opt.clone()) + }) + .await + .map_err(Retry::into_inner)?; Ok(payload) } From 9dcd303ea672f337b01ac0fc152effeb618ecaf5 Mon Sep 17 00:00:00 2001 From: Darkheir Date: Fri, 23 Oct 2026 17:15:27 +0200 Subject: [PATCH 4/4] feat: Time out S3 attempts individually and retry throttled requests longer Signed-off-by: Darkheir --- quickwit/quickwit-aws/src/error.rs | 8 +- quickwit/quickwit-aws/src/retry.rs | 17 +- quickwit/quickwit-cli/src/lib.rs | 9 + quickwit/quickwit-cli/src/tool.rs | 4 +- quickwit/quickwit-common/src/retry.rs | 68 +++++- .../quickwit-config/src/node_config/mod.rs | 15 +- quickwit/quickwit-search/src/leaf.rs | 53 ++++- quickwit/quickwit-storage/src/debouncer.rs | 5 + .../object_storage/s3_compatible_storage.rs | 207 +++++++++++++++++- .../s3_compatible_storage_resolver.rs | 9 +- quickwit/quickwit-storage/src/storage.rs | 6 + 11 files changed, 369 insertions(+), 32 deletions(-) diff --git a/quickwit/quickwit-aws/src/error.rs b/quickwit/quickwit-aws/src/error.rs index e97a9fa5ac8..7107482d3a0 100644 --- a/quickwit/quickwit-aws/src/error.rs +++ b/quickwit/quickwit-aws/src/error.rs @@ -16,7 +16,7 @@ use aws_config::retry::ErrorKind; use aws_runtime::retries::classifiers::{THROTTLING_ERRORS, TRANSIENT_ERRORS}; -use aws_sdk_s3::error::SdkError; +use aws_sdk_s3::error::{ProvideErrorMetadata, SdkError}; use aws_sdk_s3::operation::abort_multipart_upload::AbortMultipartUploadError; use aws_sdk_s3::operation::complete_multipart_upload::CompleteMultipartUploadError; use aws_sdk_s3::operation::create_multipart_upload::CreateMultipartUploadError; @@ -30,7 +30,7 @@ use aws_sdk_s3::operation::upload_part::UploadPartError; use crate::retry::AwsRetryable; impl AwsRetryable for SdkError -where E: AwsRetryable +where E: AwsRetryable + ProvideErrorMetadata { fn is_retryable(&self) -> bool { match self { @@ -55,6 +55,10 @@ where E: AwsRetryable _ => false, } } + + fn is_throttling(&self) -> bool { + matches!(self.code(), Some(code) if THROTTLING_ERRORS.contains(&code)) + } } fn is_retryable(meta: &aws_sdk_s3::error::ErrorMetadata) -> bool { diff --git a/quickwit/quickwit-aws/src/retry.rs b/quickwit/quickwit-aws/src/retry.rs index b52ce35bdc8..7a7e012dd83 100644 --- a/quickwit/quickwit-aws/src/retry.rs +++ b/quickwit/quickwit-aws/src/retry.rs @@ -23,14 +23,19 @@ pub trait AwsRetryable { fn is_retryable(&self) -> bool { false } + + fn is_throttling(&self) -> bool { + false + } } impl AwsRetryable for Retry { fn is_retryable(&self) -> bool { - match self { - Retry::Transient(_) => true, - Retry::Permanent(_) => false, - } + Retryable::is_retryable(self) + } + + fn is_throttling(&self) -> bool { + Retryable::is_throttling(self) } } @@ -43,6 +48,10 @@ where E: AwsRetryable fn is_retryable(&self) -> bool { self.0.is_retryable() } + + fn is_throttling(&self) -> bool { + self.0.is_throttling() + } } pub async fn aws_retry(retry_params: &RetryParams, f: impl Fn() -> Fut) -> Result diff --git a/quickwit/quickwit-cli/src/lib.rs b/quickwit/quickwit-cli/src/lib.rs index fb0eb1e57b0..b1e65fd902e 100644 --- a/quickwit/quickwit-cli/src/lib.rs +++ b/quickwit/quickwit-cli/src/lib.rs @@ -270,9 +270,18 @@ fn get_resolvers( } /// Returns the S3 storage settings of a node running `node_config`. +/// +/// The searcher's storage timeout policy is applied by the S3 storages themselves, to each attempt +/// of a request, so it is handed over to them whenever the node runs the searcher. fn s3_node_options(node_config: &NodeConfig) -> S3NodeOptions { + let timeout_policy = if node_config.is_service_enabled(QuickwitService::Searcher) { + node_config.searcher_config.storage_timeout_policy.clone() + } else { + None + }; S3NodeOptions { assembly_thread_pool: s3_assembly_thread_pool_for_services(&node_config.enabled_services), + timeout_policy, } } diff --git a/quickwit/quickwit-cli/src/tool.rs b/quickwit/quickwit-cli/src/tool.rs index e9921aef50d..0020463d948 100644 --- a/quickwit/quickwit-cli/src/tool.rs +++ b/quickwit/quickwit-cli/src/tool.rs @@ -65,7 +65,7 @@ use tracing::{debug, info}; use crate::checklist::{GREEN_COLOR, RED_COLOR}; use crate::{ THROUGHPUT_WINDOW_SIZE, config_cli_arg, get_resolvers, load_node_config, run_index_checklist, - start_actor_runtimes, + s3_node_options, start_actor_runtimes, }; pub fn build_tool_command() -> Command { @@ -721,7 +721,7 @@ pub async fn local_search_cli(args: LocalSearchArgs) -> anyhow::Result<()> { let (storage_resolver, metastore_resolver) = get_resolvers( &config.storage_configs, &config.metastore_configs, - S3NodeOptions::default(), + s3_node_options(&config), ); let metastore: MetastoreServiceClient = metastore_resolver.resolve(&config.metastore_uri).await?; diff --git a/quickwit/quickwit-common/src/retry.rs b/quickwit/quickwit-common/src/retry.rs index 93e1c3d099b..37fbe725196 100644 --- a/quickwit/quickwit-common/src/retry.rs +++ b/quickwit/quickwit-common/src/retry.rs @@ -24,12 +24,17 @@ pub trait Retryable { fn is_retryable(&self) -> bool { false } + + fn is_throttling(&self) -> bool { + false + } } #[derive(Debug, Eq, PartialEq)] pub enum Retry { Permanent(E), Transient(E), + Throttled(E), } impl Retry { @@ -37,6 +42,7 @@ impl Retry { match self { Self::Transient(error) => error, Self::Permanent(error) => error, + Self::Throttled(error) => error, } } } @@ -45,11 +51,20 @@ impl Retryable for Retry { fn is_retryable(&self) -> bool { match self { Retry::Permanent(_) => false, - Retry::Transient(_) => true, + Retry::Transient(_) | Retry::Throttled(_) => true, } } + + fn is_throttling(&self) -> bool { + matches!(self, Retry::Throttled(_)) + } } +/// Number of attempts granted to an operation that keeps being throttled, on top of +/// [`RetryParams::max_attempts`]. Throttling can last longer than the other transient errors, and +/// the exponential backoff spaces these last attempts out, so they add little load. +pub const THROTTLING_EXTRA_ATTEMPTS: usize = 2; + #[derive(Debug, Clone, Copy)] pub struct RetryParams { pub base_delay: Duration, @@ -159,7 +174,13 @@ where } num_attempts += 1; - if num_attempts >= retry_params.max_attempts { + // Throttled operations get extra attempts, unless retries are disabled altogether. + let max_attempts = if error.is_throttling() && retry_params.max_attempts > 1 { + retry_params.max_attempts + THROTTLING_EXTRA_ATTEMPTS + } else { + retry_params.max_attempts + }; + if num_attempts >= max_attempts { warn!( num_attempts=%num_attempts, "request failed" @@ -188,11 +209,14 @@ where #[cfg(test)] mod tests { use std::sync::RwLock; + use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::Duration; use futures::future::ready; - use super::{MockableSleep, RetryParams, Retryable, retry_with_mockable_sleep}; + use super::{ + MockableSleep, RetryParams, Retryable, THROTTLING_EXTRA_ATTEMPTS, retry_with_mockable_sleep, + }; #[derive(Debug, Eq, PartialEq)] pub enum Retry { @@ -266,6 +290,44 @@ mod tests { ); } + /// Returns the number of attempts of an operation that always fails with `error`. + async fn count_attempts(retry_params: RetryParams, error: fn() -> super::Retry<()>) -> usize { + let num_attempts = AtomicUsize::new(0); + retry_with_mockable_sleep( + &retry_params, + || { + num_attempts.fetch_add(1, Ordering::Relaxed); + ready(Err::<(), _>(error())) + }, + NoopSleep, + ) + .await + .unwrap_err(); + num_attempts.into_inner() + } + + #[tokio::test] + async fn test_retry_gives_throttled_operations_extra_attempts() { + let retry_params = RetryParams { + base_delay: Duration::from_millis(1), + max_delay: Duration::from_millis(2), + max_attempts: 3, + }; + assert_eq!( + count_attempts(retry_params, || super::Retry::Transient(())).await, + 3 + ); + assert_eq!( + count_attempts(retry_params, || super::Retry::Throttled(())).await, + 3 + THROTTLING_EXTRA_ATTEMPTS + ); + // Disabled retries stay disabled for throttled operations. + assert_eq!( + count_attempts(RetryParams::no_retries(), || super::Retry::Throttled(())).await, + 1 + ); + } + #[tokio::test] async fn test_retry_retries_up_to_max_attempts_times() { let retry_sequence: Vec<_> = (0..29) diff --git a/quickwit/quickwit-config/src/node_config/mod.rs b/quickwit/quickwit-config/src/node_config/mod.rs index 9809282278c..67a3ecfda39 100644 --- a/quickwit/quickwit-config/src/node_config/mod.rs +++ b/quickwit/quickwit-config/src/node_config/mod.rs @@ -413,6 +413,10 @@ impl std::fmt::Display for CachePolicy { /// This policy is inspired by this guidance. It does not track instanteneous throughput, but /// computes an overall timeout using the following formula: /// `timeout_offset + num_bytes_get_request / min_throughtput` +/// +/// S3-compatible storages apply this timeout to each attempt of a request, response body +/// included, inside their own retry loop: the delays between attempts, for instance after a +/// throttling error, are not counted. Other storages apply it to the whole `get_slice` call. #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] pub struct StorageTimeoutPolicy { pub min_throughtput_bytes_per_secs: u64, @@ -423,15 +427,18 @@ pub struct StorageTimeoutPolicy { } impl StorageTimeoutPolicy { - pub fn compute_timeout(&self, num_bytes: usize) -> impl Iterator { + /// Returns the timeout of a single attempt to download `num_bytes`. + pub fn timeout(&self, num_bytes: usize) -> Duration { let min_download_time_secs: f64 = if self.min_throughtput_bytes_per_secs == 0 { 0.0f64 } else { num_bytes as f64 / self.min_throughtput_bytes_per_secs as f64 }; - let timeout = Duration::from_millis(self.timeout_millis) - + Duration::from_secs_f64(min_download_time_secs); - std::iter::repeat_n(timeout, self.max_num_retries + 1) + Duration::from_millis(self.timeout_millis) + Duration::from_secs_f64(min_download_time_secs) + } + + pub fn compute_timeout(&self, num_bytes: usize) -> impl Iterator { + std::iter::repeat_n(self.timeout(num_bytes), self.max_num_retries + 1) } } diff --git a/quickwit/quickwit-search/src/leaf.rs b/quickwit/quickwit-search/src/leaf.rs index e5c0c44dd26..974586fb6cd 100644 --- a/quickwit/quickwit-search/src/leaf.rs +++ b/quickwit/quickwit-search/src/leaf.rs @@ -147,18 +147,24 @@ pub(crate) async fn open_split_bundle( /// if configured in the searcher config. /// /// The goal here is too ensure a low latency. +/// +/// Storages that already apply this policy themselves, to each attempt of a request, are left as +/// is (see [`Storage::timeout_policy`]). fn configure_storage_retries( searcher_context: &SearcherContext, index_storage: Arc, ) -> Arc { - if let Some(storage_timeout_policy) = &searcher_context.searcher_config.storage_timeout_policy { - Arc::new(TimeoutAndRetryStorage::new( - index_storage, - storage_timeout_policy.clone(), - )) - } else { - index_storage + let Some(storage_timeout_policy) = &searcher_context.searcher_config.storage_timeout_policy + else { + return index_storage; + }; + if index_storage.timeout_policy().as_ref() == Some(storage_timeout_policy) { + return index_storage; } + Arc::new(TimeoutAndRetryStorage::new( + index_storage, + storage_timeout_policy.clone(), + )) } /// Opens a `tantivy::Index` for the given split with several cache layers: @@ -1732,6 +1738,7 @@ mod tests { use std::ops::Bound; use bytes::BufMut; + use quickwit_config::{SearcherConfig, StorageTimeoutPolicy}; use quickwit_directories::write_hotcache; use rand::Rng; use tantivy::TantivyDocument; @@ -1742,6 +1749,38 @@ mod tests { use super::*; + #[tokio::test] + async fn test_configure_storage_retries_skips_storages_applying_the_policy() { + let timeout_policy = StorageTimeoutPolicy { + min_throughtput_bytes_per_secs: 100_000, + timeout_millis: 2_000, + max_num_retries: 1, + }; + let mut searcher_config = SearcherConfig::default(); + searcher_config.storage_timeout_policy = Some(timeout_policy.clone()); + let searcher_context = SearcherContext::new(searcher_config, None, None); + let storage_applying = |policy_opt: Option| -> Arc { + let mut storage = quickwit_storage::MockStorage::new(); + storage + .expect_timeout_policy() + .returning(move || policy_opt.clone()); + Arc::new(storage) + }; + let is_wrapped = |storage: Arc| { + let configured_storage = configure_storage_retries(&searcher_context, storage.clone()); + !std::ptr::addr_eq(Arc::as_ptr(&configured_storage), Arc::as_ptr(&storage)) + }; + // The storage applies the searcher's policy itself. + assert!(!is_wrapped(storage_applying(Some(timeout_policy.clone())))); + // The storage applies no policy, or another one. + assert!(is_wrapped(storage_applying(None))); + let other_policy = StorageTimeoutPolicy { + max_num_retries: 0, + ..timeout_policy + }; + assert!(is_wrapped(storage_applying(Some(other_policy)))); + } + fn bool_filter(ast: impl Into) -> QueryAst { BoolQuery { must: vec![QueryAst::MatchAll], diff --git a/quickwit/quickwit-storage/src/debouncer.rs b/quickwit/quickwit-storage/src/debouncer.rs index 60e9a3cf299..3198a8de42c 100644 --- a/quickwit/quickwit-storage/src/debouncer.rs +++ b/quickwit/quickwit-storage/src/debouncer.rs @@ -23,6 +23,7 @@ use fnv::FnvHashMap; use futures::future::{BoxFuture, WeakShared}; use futures::{Future, FutureExt}; use quickwit_common::uri::Uri; +use quickwit_config::StorageTimeoutPolicy; use tantivy::directory::OwnedBytes; use tokio::io::AsyncRead; @@ -191,6 +192,10 @@ impl Storage for DebouncedStorage { self.underlying.uri() } + fn timeout_policy(&self) -> Option { + self.underlying.timeout_policy() + } + async fn file_num_bytes(&self, path: &Path) -> StorageResult { self.underlying.file_num_bytes(path).await } diff --git a/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage.rs b/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage.rs index 0ea4643183d..84bc0b6d353 100644 --- a/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage.rs +++ b/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage.rs @@ -16,6 +16,7 @@ use std::collections::HashMap; use std::ops::Range; use std::path::{Path, PathBuf}; use std::pin::Pin; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::task::{Context, Poll}; use std::time::Duration; use std::{fmt, io}; @@ -43,7 +44,7 @@ use quickwit_common::retry::{Retry, RetryParams}; use quickwit_common::thread_pool::with_priority::ThreadPoolWithPriority; use quickwit_common::uri::Uri; use quickwit_common::{chunk_range, into_u64_range}; -use quickwit_config::{S3EncryptionConfig, S3StorageConfig}; +use quickwit_config::{S3EncryptionConfig, S3StorageConfig, StorageTimeoutPolicy}; use regex::Regex; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt, BufReader, ReadBuf}; use tokio::sync::Semaphore; @@ -146,19 +147,32 @@ impl AsyncRead for S3AsyncRead { } } -/// Classifies an SDK error for the retry loop: transient or permanent. +/// Classifies an SDK error for the retry loop: throttled, transient or permanent. fn into_retry(sdk_error: SdkError) -> Retry where SdkError: AwsRetryable, StorageError: From>, { - if sdk_error.is_retryable() { + if sdk_error.is_throttling() { + Retry::Throttled(StorageError::from(sdk_error)) + } else if sdk_error.is_retryable() { Retry::Transient(StorageError::from(sdk_error)) } else { Retry::Permanent(StorageError::from(sdk_error)) } } +/// Records the outcome of a `get_slice` subject to a timeout policy, in the same metric as +/// [`crate::TimeoutAndRetryStorage`]: either it gave up because its last attempt timed out, or it +/// completed, successfully or not, after `num_timeouts` timed-out attempts. +fn record_get_slice_timeout_outcome(num_timeouts: usize, gave_up_on_timeout: bool) { + if gave_up_on_timeout { + STORAGE_METRICS.get_slice_timeout_all_timeouts.inc(); + } else { + STORAGE_METRICS.get_slice_timeout_successes[num_timeouts.min(2)].inc(); + } +} + /// S3-compatible object storage implementation. pub struct S3CompatibleObjectStorage { s3_client: S3Client, @@ -175,6 +189,9 @@ pub struct S3CompatibleObjectStorage { /// config: it is set by the node that builds the storage (see /// [`Self::with_assembly_thread_pool`]). assembly_thread_pool: Option, + /// Timeout of each attempt to download a slice. Like the assembly thread pool, it is set by + /// the node that builds the storage (see [`Self::with_timeout_policy`]). + timeout_policy: Option, } impl fmt::Debug for S3CompatibleObjectStorage { @@ -285,6 +302,7 @@ impl S3CompatibleObjectStorage { disable_multipart_upload, encryption: s3_storage_config.encryption.clone(), assembly_thread_pool: None, + timeout_policy: None, }) } @@ -306,6 +324,16 @@ impl S3CompatibleObjectStorage { } } + /// Applies `timeout_policy` to each attempt to download a slice, response body included. A + /// timed-out attempt is retried, `max_num_retries` times at most, like any transient error: + /// the delays between attempts are not counted in the timeout. + pub fn with_timeout_policy(self, timeout_policy: StorageTimeoutPolicy) -> Self { + Self { + timeout_policy: Some(timeout_policy), + ..self + } + } + /// Sets the multipart policy. /// /// See `MultiPartPolicy`. @@ -761,11 +789,37 @@ impl S3CompatibleObjectStorage { path: &Path, range_opt: Option>, ) -> StorageResult { - let payload = aws_retry_with_metrics(&self.retry_params, ActionLabel::GetObject, || { - self.get_to_bytes_attempt(path, range_opt.clone()) - }) - .await - .map_err(Retry::into_inner)?; + // The timeout policy only applies to slices, whose size is known. + let slice_timeout_policy_opt = range_opt.as_ref().zip(self.timeout_policy.as_ref()); + let num_timeouts = AtomicUsize::new(0); + let last_attempt_timed_out = AtomicBool::new(false); + let payload_res = + aws_retry_with_metrics(&self.retry_params, ActionLabel::GetObject, || async { + let attempt = self.get_to_bytes_attempt(path, range_opt.clone()); + let Some((range, timeout_policy)) = slice_timeout_policy_opt else { + return attempt.await; + }; + let attempt_res = + tokio::time::timeout(timeout_policy.timeout(range.len()), attempt).await; + last_attempt_timed_out.store(attempt_res.is_err(), Ordering::Relaxed); + attempt_res.unwrap_or_else(|_elapsed| { + let error = StorageErrorKind::Timeout + .with_error(anyhow!("internal timeout on get_slice")); + // Timed-out attempts are retried `max_num_retries` times at most. + if num_timeouts.fetch_add(1, Ordering::Relaxed) < timeout_policy.max_num_retries + { + Err(Retry::Transient(error)) + } else { + Err(Retry::Permanent(error)) + } + }) + }) + .await; + if slice_timeout_policy_opt.is_some() { + let gave_up_on_timeout = payload_res.is_err() && last_attempt_timed_out.into_inner(); + record_get_slice_timeout_outcome(num_timeouts.into_inner(), gave_up_on_timeout); + } + let payload = payload_res.map_err(Retry::into_inner)?; Ok(payload) } @@ -1056,6 +1110,10 @@ impl Storage for S3CompatibleObjectStorage { fn uri(&self) -> &Uri { &self.uri } + + fn timeout_policy(&self) -> Option { + self.timeout_policy.clone() + } } #[cfg(test)] @@ -1069,6 +1127,7 @@ mod tests { use hyper::http; use quickwit_aws::aws_behavior_version; use quickwit_common::chunk_range; + use quickwit_common::retry::THROTTLING_EXTRA_ATTEMPTS; use quickwit_common::uri::Uri; use super::*; @@ -1166,6 +1225,7 @@ mod tests { disable_multipart_upload: false, encryption: None, assembly_thread_pool: None, + timeout_policy: None, }; assert_eq!( s3_storage.relative_path("indexes/foo"), @@ -1215,6 +1275,7 @@ mod tests { disable_multipart_upload: false, encryption: None, assembly_thread_pool: None, + timeout_policy: None, }; let _ = s3_storage .bulk_delete(&[Path::new("foo"), Path::new("bar")]) @@ -1254,6 +1315,7 @@ mod tests { disable_multipart_upload: false, encryption: None, assembly_thread_pool: None, + timeout_policy: None, }; let _ = s3_storage .bulk_delete(&[Path::new("foo"), Path::new("bar")]) @@ -1338,6 +1400,7 @@ mod tests { disable_multipart_upload: false, encryption: None, assembly_thread_pool: None, + timeout_policy: None, }; let bulk_delete_error = s3_storage .bulk_delete(&[ @@ -1431,6 +1494,7 @@ mod tests { disable_multipart_upload: false, encryption: None, assembly_thread_pool: None, + timeout_policy: None, }; s3_storage .put(Path::new("my-path"), Box::new(vec![1, 2, 3])) @@ -1475,6 +1539,7 @@ mod tests { disable_multipart_upload: false, encryption: None, assembly_thread_pool: None, + timeout_policy: None, } } @@ -1518,6 +1583,130 @@ mod tests { assert_eq!(client.actual_requests().count(), retry_params.max_attempts); } + fn get_object_event(status: u16, body: SdkBody) -> ReplayEvent { + ReplayEvent::new( + http::Request::builder().body(SdkBody::empty()).unwrap(), + http::Response::builder().status(status).body(body).unwrap(), + ) + } + + fn get_object_error_event(status: u16, code: &str) -> ReplayEvent { + let body = format!("{code}message"); + get_object_event(status, SdkBody::from(body)) + } + + /// A response whose body never arrives: only a timeout can end the attempt. + fn get_object_event_with_pending_body() -> ReplayEvent { + let pending_body = http_body_util::StreamBody::new(futures::stream::pending::< + Result, io::Error>, + >()); + get_object_event(206, SdkBody::from_body_1_x(pending_body)) + } + + fn timeout_policy(timeout: Duration, max_num_retries: usize) -> StorageTimeoutPolicy { + StorageTimeoutPolicy { + min_throughtput_bytes_per_secs: 0, + timeout_millis: timeout.as_millis() as u64, + max_num_retries, + } + } + + #[tokio::test] + async fn test_s3_compatible_storage_get_slice_retries_timed_out_attempt() { + tokio::time::pause(); + let client = StaticReplayClient::new(vec![ + get_object_event_with_pending_body(), + get_object_event(206, SdkBody::from("abcd")), + ]); + let s3_storage = s3_storage_for_get_test(client.clone()) + .with_timeout_policy(timeout_policy(Duration::from_secs(2), 1)); + + let slice = s3_storage + .get_slice(Path::new("my-path"), 0..4) + .await + .unwrap(); + assert_eq!(slice.as_slice(), b"abcd"); + assert_eq!(client.actual_requests().count(), 2); + } + + #[tokio::test] + async fn test_s3_compatible_storage_get_slice_timeout_without_retries() { + tokio::time::pause(); + let client = StaticReplayClient::new(vec![ + get_object_event_with_pending_body(), + get_object_event(206, SdkBody::from("abcd")), + ]); + let s3_storage = s3_storage_for_get_test(client.clone()) + .with_timeout_policy(timeout_policy(Duration::from_secs(2), 0)); + + let error = s3_storage + .get_slice(Path::new("my-path"), 0..4) + .await + .unwrap_err(); + assert_eq!(error.kind(), StorageErrorKind::Timeout); + assert_eq!(client.actual_requests().count(), 1); + } + + #[tokio::test] + async fn test_s3_compatible_storage_get_slice_timeout_excludes_backoff() { + tokio::time::pause(); + let client = StaticReplayClient::new(vec![ + get_object_error_event(503, "SlowDown"), + get_object_event(206, SdkBody::from("abcd")), + ]); + // The backoff after the throttling error, 1 to 2s, exceeds the 1s timeout: only the + // attempts are timed. + let s3_storage = S3CompatibleObjectStorage { + retry_params: RetryParams { + base_delay: Duration::from_secs(2), + max_delay: Duration::from_secs(10), + max_attempts: 3, + }, + ..s3_storage_for_get_test(client.clone()) + } + .with_timeout_policy(timeout_policy(Duration::from_secs(1), 0)); + + let slice = s3_storage + .get_slice(Path::new("my-path"), 0..4) + .await + .unwrap(); + assert_eq!(slice.as_slice(), b"abcd"); + assert_eq!(client.actual_requests().count(), 2); + } + + #[tokio::test] + async fn test_s3_compatible_storage_get_slice_retries_throttling_longer() { + let max_attempts = RetryParams::for_test().max_attempts; + // A throttled slice gets extra attempts... + let client = StaticReplayClient::new( + (0..max_attempts + THROTTLING_EXTRA_ATTEMPTS - 1) + .map(|_| get_object_error_event(503, "SlowDown")) + .chain([get_object_event(206, SdkBody::from("abcd"))]) + .collect(), + ); + let slice = s3_storage_for_get_test(client.clone()) + .get_slice(Path::new("my-path"), 0..4) + .await + .unwrap(); + assert_eq!(slice.as_slice(), b"abcd"); + assert_eq!( + client.actual_requests().count(), + max_attempts + THROTTLING_EXTRA_ATTEMPTS + ); + + // ... that other transient errors don't get. + let client = StaticReplayClient::new( + (0..max_attempts) + .map(|_| get_object_error_event(500, "InternalError")) + .collect(), + ); + s3_storage_for_get_test(client.clone()) + .get_slice(Path::new("my-path"), 0..4) + .await + .unwrap_err(); + assert_eq!(client.actual_requests().count(), max_attempts); + } + #[tokio::test] async fn test_sse_c_headers_in_regular_put() { for read_only in [false, true] { @@ -1555,6 +1744,7 @@ mod tests { read_only, }), assembly_thread_pool: None, + timeout_policy: None, }; let small_payload = vec![1u8; 100]; @@ -1657,6 +1847,7 @@ mod tests { read_only, }), assembly_thread_pool: None, + timeout_policy: None, }; // Test multipart upload with large payload that triggers multipart (15MB > 10MB 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 ebb20a31698..031a4103f3f 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 @@ -18,7 +18,7 @@ use async_trait::async_trait; use aws_sdk_s3::Client as S3Client; use quickwit_common::thread_pool::with_priority::ThreadPoolWithPriority; use quickwit_common::uri::Uri; -use quickwit_config::{S3StorageConfig, StorageBackend}; +use quickwit_config::{S3StorageConfig, StorageBackend, StorageTimeoutPolicy}; use tokio::sync::OnceCell; use super::s3_compatible_storage::create_s3_client; @@ -32,12 +32,14 @@ use crate::{ pub struct S3NodeOptions { /// See [`S3CompatibleObjectStorage::with_assembly_thread_pool`]. pub assembly_thread_pool: Option, + /// See [`S3CompatibleObjectStorage::with_timeout_policy`]. + pub timeout_policy: Option, } impl S3NodeOptions { /// Whether no option is set. pub fn is_empty(&self) -> bool { - self.assembly_thread_pool.is_none() + self.assembly_thread_pool.is_none() && self.timeout_policy.is_none() } } @@ -90,6 +92,9 @@ impl StorageFactory for S3CompatibleObjectStorageFactory { if let Some(thread_pool) = &self.node_options.assembly_thread_pool { storage = storage.with_assembly_thread_pool(thread_pool.clone()); } + if let Some(timeout_policy) = &self.node_options.timeout_policy { + storage = storage.with_timeout_policy(timeout_policy.clone()); + } Ok(Arc::new(DebouncedStorage::new(storage))) } } diff --git a/quickwit/quickwit-storage/src/storage.rs b/quickwit/quickwit-storage/src/storage.rs index da5dcb4ff15..c0ee877bc60 100644 --- a/quickwit/quickwit-storage/src/storage.rs +++ b/quickwit/quickwit-storage/src/storage.rs @@ -19,6 +19,7 @@ use std::path::{Path, PathBuf}; use async_trait::async_trait; use quickwit_common::uri::Uri; +use quickwit_config::StorageTimeoutPolicy; use tempfile::TempPath; use tokio::fs::File; use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt, BufWriter}; @@ -137,6 +138,11 @@ pub trait Storage: fmt::Debug + Send + Sync + 'static { /// Returns an URI identifying the storage fn uri(&self) -> &Uri; + + /// Returns the timeout policy that the storage applies to its requests itself, if any. + fn timeout_policy(&self) -> Option { + None + } } async fn default_copy_to_file(