From f3173b221c5b4e9c2d8766d88b8af7392f391721 Mon Sep 17 00:00:00 2001 From: Darkheir Date: Fri, 23 Oct 2026 10:54:21 +0200 Subject: [PATCH 1/3] feat: Observability of object store queries Signed-off-by: Darkheir --- .../quickwit-search/src/cluster_client.rs | 48 +++++++++++++++++++ quickwit/quickwit-search/src/error.rs | 18 ++++++- quickwit/quickwit-search/src/metrics.rs | 9 ++++ quickwit/quickwit-storage/src/error.rs | 22 ++++++++- quickwit/quickwit-storage/src/metrics.rs | 16 +++---- .../quickwit-storage/src/metrics_wrappers.rs | 48 +++++++++++++------ .../src/object_storage/azure_blob_storage.rs | 42 ++++++++++++---- .../object_storage/s3_compatible_storage.rs | 20 ++++---- .../src/opendal_storage/base.rs | 15 ++++-- 9 files changed, 189 insertions(+), 49 deletions(-) diff --git a/quickwit/quickwit-search/src/cluster_client.rs b/quickwit/quickwit-search/src/cluster_client.rs index 461ffae7521..a9ed822f5a7 100644 --- a/quickwit/quickwit-search/src/cluster_client.rs +++ b/quickwit/quickwit-search/src/cluster_client.rs @@ -24,6 +24,7 @@ use quickwit_proto::search::{ use tantivy::aggregation::intermediate_agg_result::IntermediateAggregationResults; use tracing::{debug, error, info, warn}; +use crate::metrics::SEARCH_METRICS; use crate::retry::search::LeafSearchRetryPolicy; use crate::retry::{DefaultRetryPolicy, RetryPolicy, retry_client}; use crate::{ @@ -114,7 +115,21 @@ impl ClusterClient { "Leaf search response error: `{:?}`. Retry once to execute {:?} with {:?}", response_res, retry_request, client ); + let num_retried_splits: usize = retry_request + .leaf_requests + .iter() + .map(|leaf_req| leaf_req.split_offsets.len()) + .sum(); let retry_result = client.leaf_search(retry_request).await; + let (num_rescued_splits, num_failed_splits) = + split_retry_outcomes(num_retried_splits, &retry_result); + let split_retries_total = &SEARCH_METRICS.leaf_search_split_retries_total; + split_retries_total + .with_label_values(["rescued"]) + .inc_by(num_rescued_splits as u64); + split_retries_total + .with_label_values(["failed"]) + .inc_by(num_failed_splits as u64); response_res = merge_original_with_retry_leaf_search_results(response_res, retry_result); response_res } @@ -234,6 +249,18 @@ fn merge_intermediate_aggregation(left: &[u8], right: &[u8]) -> crate::Result, +) -> (usize, usize) { + let num_failed_splits = match retry_result { + Ok(retry_response) => retry_response.failed_splits.len().min(num_retried_splits), + Err(_) => num_retried_splits, + }; + (num_retried_splits - num_failed_splits, num_failed_splits) +} + /// Merge two leaf search response. /// /// # Quirk @@ -599,6 +626,27 @@ mod tests { Ok(()) } + #[test] + fn test_split_retry_outcomes() { + let retry_response = LeafSearchResponse { + failed_splits: vec![SplitSearchError { + error: "error".to_string(), + split_id: "split_2".to_string(), + retryable_error: true, + }], + ..Default::default() + }; + assert_eq!(split_retry_outcomes(3, &Ok(retry_response)), (2, 1)); + assert_eq!( + split_retry_outcomes(3, &Ok(LeafSearchResponse::default())), + (3, 0) + ); + assert_eq!( + split_retry_outcomes(3, &Err(SearchError::Internal("error".to_string()))), + (0, 3) + ); + } + #[tokio::test] async fn test_put_kv_happy_path() { // 3 servers 1, 2, 3 diff --git a/quickwit/quickwit-search/src/error.rs b/quickwit/quickwit-search/src/error.rs index b8573025e4d..e4cd42f17c7 100644 --- a/quickwit/quickwit-search/src/error.rs +++ b/quickwit/quickwit-search/src/error.rs @@ -170,7 +170,8 @@ impl From for SearchError { impl From for SearchError { fn from(any_error: anyhow::Error) -> Self { - SearchError::Internal(any_error.to_string()) + // The alternate format keeps the whole chain of causes, not only the outermost context. + SearchError::Internal(format!("{any_error:#}")) } } @@ -233,4 +234,19 @@ mod tests { "unexpected error: {search_error:?}" ); } + + #[test] + fn test_search_error_from_anyhow_keeps_the_causes() { + let any_error = anyhow::anyhow!("internal timeout on get_slice") + .context("failed to fetch hotcache and footer for split `split-1`"); + let search_error = SearchError::from(any_error); + assert!( + matches!( + &search_error, + SearchError::Internal(msg) if msg == "failed to fetch hotcache and footer for split \ + `split-1`: internal timeout on get_slice" + ), + "unexpected error: {search_error:?}" + ); + } } diff --git a/quickwit/quickwit-search/src/metrics.rs b/quickwit/quickwit-search/src/metrics.rs index 18be753eef9..fc3cfaa0ad8 100644 --- a/quickwit/quickwit-search/src/metrics.rs +++ b/quickwit/quickwit-search/src/metrics.rs @@ -151,6 +151,7 @@ pub struct SearchMetrics { pub leaf_search_split_duration_secs: Histogram, pub leaf_search_permit_wait_duration_secs: Histogram, pub job_assigned_total: IntCounterVec<1>, + pub leaf_search_split_retries_total: IntCounterVec<1>, pub leaf_search_single_split_tasks_pending: IntGauge, pub leaf_search_single_split_tasks_ongoing: IntGauge, pub leaf_search_single_split_warmup_num_bytes: Histogram, @@ -284,6 +285,14 @@ impl Default for SearchMetrics { &[], ["affinity"], ), + leaf_search_split_retries_total: new_counter_vec( + "leaf_search_split_retries_total", + "Number of splits whose leaf search failed and was retried on another searcher, \ + by outcome of the retry: `rescued` or `failed` again.", + "search", + &[], + ["outcome"], + ), searcher_local_kv_store_size_bytes: new_gauge( "searcher_local_kv_store_size_bytes", "Size of the searcher kv store in bytes. This store is used to cache scroll \ diff --git a/quickwit/quickwit-storage/src/error.rs b/quickwit/quickwit-storage/src/error.rs index e3873d91bd8..2578c8eba7b 100644 --- a/quickwit/quickwit-storage/src/error.rs +++ b/quickwit/quickwit-storage/src/error.rs @@ -81,8 +81,8 @@ impl From for io::Error { StorageErrorKind::NotFound => io::ErrorKind::NotFound, _ => io::ErrorKind::Other, }; - // TODO: This is swallowing the context of the source error. - io::Error::new(io_error_kind, storage_err.source.to_string()) + // The alternate format keeps the whole chain of causes, not only the outermost context. + io::Error::new(io_error_kind, format!("{:#}", storage_err.source)) } } @@ -196,3 +196,21 @@ impl fmt::Display for BulkDeleteError { Ok(()) } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_io_error_from_storage_error_keeps_the_causes() { + let storage_error = StorageErrorKind::Timeout.with_error( + anyhow::anyhow!("internal timeout on get_slice").context("get_slice failed"), + ); + let io_error = io::Error::from(storage_error); + assert_eq!(io_error.kind(), io::ErrorKind::Other); + assert_eq!( + io_error.to_string(), + "get_slice failed: internal timeout on get_slice" + ); + } +} diff --git a/quickwit/quickwit-storage/src/metrics.rs b/quickwit/quickwit-storage/src/metrics.rs index d8124d243b3..5e5106e45f7 100644 --- a/quickwit/quickwit-storage/src/metrics.rs +++ b/quickwit/quickwit-storage/src/metrics.rs @@ -37,8 +37,8 @@ pub struct StorageMetrics { pub searcher_split_cache: ComponentCacheMetrics, pub get_slice_timeout_successes: [IntCounter; 3], pub get_slice_timeout_all_timeouts: IntCounter, - pub object_storage_requests_total: IntCounterVec<2>, - pub object_storage_request_duration: HistogramVec<2>, + pub object_storage_requests_total: IntCounterVec<3>, + pub object_storage_request_duration: HistogramVec<3>, pub object_storage_get_slice_in_flight_count: IntGauge, pub object_storage_get_slice_in_flight_num_bytes: IntGauge, pub object_storage_get_object_connections_in_use: IntGauge, @@ -79,20 +79,20 @@ impl Default for StorageMetrics { get_slice_timeout_all_timeouts, object_storage_requests_total: new_counter_vec( "object_storage_requests_total", - "Number of requests to the object store, by action and status. Requests are \ - recorded when the response headers are returned, download failures will not \ + "Number of requests to the object store, by action, status and bucket. Requests \ + are recorded when the response headers are returned, download failures will not \ appear as errors.", "storage", &[], - ["action", "status"], + ["action", "status", "bucket"], ), object_storage_request_duration: new_histogram_vec( "object_storage_request_duration", - "Time to first byte (TTFB) for object store requests, by action and status. For \ - download requests, excludes the time to download the response body.", + "Time to first byte (TTFB) for object store requests, by action, status and \ + bucket. For download requests, excludes the time to download the response body.", "storage", &[], - ["action", "status"], + ["action", "status", "bucket"], vec![0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0], ), object_storage_get_slice_in_flight_count: new_gauge( diff --git a/quickwit/quickwit-storage/src/metrics_wrappers.rs b/quickwit/quickwit-storage/src/metrics_wrappers.rs index adbe63aa673..bd4587b8d49 100644 --- a/quickwit/quickwit-storage/src/metrics_wrappers.rs +++ b/quickwit/quickwit-storage/src/metrics_wrappers.rs @@ -84,7 +84,7 @@ pub trait AsRequestStatus { /// This also records cancellations that occur when a timeout is applied to the /// request future. #[pin_project(PinnedDrop)] -pub struct RequestMetricsWrapper +pub struct RequestMetricsWrapper<'a, F, Marker> where F: Future, F::Output: AsRequestStatus, @@ -92,6 +92,7 @@ where #[pin] tracked: F, action: ActionLabel, + bucket: &'a str, start: Option, uploaded_bytes: Option, status: RequestStatus, @@ -99,7 +100,7 @@ where } #[pinned_drop] -impl PinnedDrop for RequestMetricsWrapper +impl PinnedDrop for RequestMetricsWrapper<'_, F, Marker> where F: Future, F::Output: AsRequestStatus, @@ -110,7 +111,7 @@ where RequestStatus::Done => return, RequestStatus::Ready(s) => s.as_str(), }; - let label_values = [self.action.as_str(), status]; + let label_values = [self.action.as_str(), status, self.bucket]; STORAGE_METRICS .object_storage_requests_total .with_label_values(label_values) @@ -130,7 +131,7 @@ where } } -impl Future for RequestMetricsWrapper +impl Future for RequestMetricsWrapper<'_, F, Marker> where F: Future, F::Output: AsRequestStatus, @@ -146,25 +147,33 @@ where } } +/// `bucket` is the bucket (or container) the request targets: request metrics are labelled with +/// it, as object stores usually rate limit requests per bucket. pub trait RequestMetricsWrapperExt where F: Future, F::Output: AsRequestStatus, { - fn with_count_metric(self, action: ActionLabel) -> RequestMetricsWrapper; + fn with_count_metric( + self, + action: ActionLabel, + bucket: &str, + ) -> RequestMetricsWrapper<'_, F, Marker>; /// For download requests, this should wrap the initial request future to /// track time to first byte (TTFB). fn with_count_and_duration_metrics( self, action: ActionLabel, - ) -> RequestMetricsWrapper; + bucket: &str, + ) -> RequestMetricsWrapper<'_, F, Marker>; fn with_count_and_upload_metrics( self, action: ActionLabel, + bucket: &str, bytes: u64, - ) -> RequestMetricsWrapper; + ) -> RequestMetricsWrapper<'_, F, Marker>; } impl RequestMetricsWrapperExt for F @@ -172,10 +181,15 @@ where F: Future, F::Output: AsRequestStatus, { - fn with_count_metric(self, action: ActionLabel) -> RequestMetricsWrapper { + fn with_count_metric( + self, + action: ActionLabel, + bucket: &str, + ) -> RequestMetricsWrapper<'_, F, Marker> { RequestMetricsWrapper { tracked: self, action, + bucket, status: RequestStatus::Pending, start: None, uploaded_bytes: None, @@ -186,10 +200,12 @@ where fn with_count_and_duration_metrics( self, action: ActionLabel, - ) -> RequestMetricsWrapper { + bucket: &str, + ) -> RequestMetricsWrapper<'_, F, Marker> { RequestMetricsWrapper { tracked: self, action, + bucket, status: RequestStatus::Pending, start: Some(Instant::now()), uploaded_bytes: None, @@ -200,11 +216,13 @@ where fn with_count_and_upload_metrics( self, action: ActionLabel, + bucket: &str, bytes: u64, - ) -> RequestMetricsWrapper { + ) -> RequestMetricsWrapper<'_, F, Marker> { RequestMetricsWrapper { tracked: self, action, + bucket, status: RequestStatus::Pending, start: None, uploaded_bytes: Some(bytes), @@ -617,10 +635,10 @@ pub mod opendal_helpers { use super::*; /// Records a request occurrence for this action with unknown status. - pub fn record_request(action: ActionLabel) { + pub fn record_request(action: ActionLabel, bucket: &str) { STORAGE_METRICS .object_storage_requests_total - .with_label_values([action.as_str(), "unknown"]) + .with_label_values([action.as_str(), "unknown", bucket]) .inc(); } @@ -641,11 +659,11 @@ pub mod opendal_helpers { } /// Records a request occurrence for this action with unknown status. - pub fn record_request_with_timer(action: ActionLabel) -> HistogramTimer { - record_request(action); + pub fn record_request_with_timer(action: ActionLabel, bucket: &str) -> HistogramTimer { + record_request(action, bucket); STORAGE_METRICS .object_storage_request_duration - .with_label_values([action.as_str(), "unknown"]) + .with_label_values([action.as_str(), "unknown", bucket]) .start_timer() } } diff --git a/quickwit/quickwit-storage/src/object_storage/azure_blob_storage.rs b/quickwit/quickwit-storage/src/object_storage/azure_blob_storage.rs index 30a6b8e54c7..4f468e98b59 100644 --- a/quickwit/quickwit-storage/src/object_storage/azure_blob_storage.rs +++ b/quickwit/quickwit-storage/src/object_storage/azure_blob_storage.rs @@ -237,7 +237,12 @@ impl AzureBlobStorage { let stream = self.container_client.blob_client(&name).get().into_stream(); (stream, None) }; - let bytes = download_all(&mut response_stream, download_kind).await?; + let bytes = download_all( + &mut response_stream, + download_kind, + self.container_client.container_name(), + ) + .await?; Result::<_, AzureErrorWrapper>::Ok(bytes) }) .await @@ -258,7 +263,11 @@ impl AzureBlobStorage { .put_block_blob(data) .hash(hash) .into_future() - .with_count_and_upload_metrics(ActionLabel::PutObject, payload.len()) + .with_count_and_upload_metrics( + ActionLabel::PutObject, + self.container_client.container_name(), + payload.len(), + ) .await?; Result::<(), AzureErrorWrapper>::Ok(()) }) @@ -297,6 +306,7 @@ impl AzureBlobStorage { .into_future() .with_count_and_upload_metrics( ActionLabel::UploadPart, + self.container_client.container_name(), range.end - range.start, ) .await?; @@ -328,7 +338,10 @@ impl AzureBlobStorage { blob_client .put_block_list(block_list) .into_future() - .with_count_metric(ActionLabel::CompleteMultipartUpload) + .with_count_metric( + ActionLabel::CompleteMultipartUpload, + self.container_client.container_name(), + ) .await .map_err(AzureErrorWrapper::from)?; @@ -345,7 +358,10 @@ impl Storage for AzureBlobStorage { .max_results(NonZeroU32::new(1u32).expect("1 is always non-zero.")) .into_stream() .next() - .with_count_metric(ActionLabel::ListObjects) + .with_count_metric( + ActionLabel::ListObjects, + self.container_client.container_name(), + ) .await { let _ = first_blob_result?; @@ -377,7 +393,10 @@ impl Storage for AzureBlobStorage { while let Some(chunk_result) = output_stream .next() - .with_count_metric(ActionLabel::GetObject) + .with_count_metric( + ActionLabel::GetObject, + self.container_client.container_name(), + ) .await { let chunk_response = chunk_result.map_err(AzureErrorWrapper::from)?; @@ -401,7 +420,10 @@ impl Storage for AzureBlobStorage { .blob_client(blob_name) .delete() .into_future() - .with_count_metric(ActionLabel::DeleteObject) + .with_count_metric( + ActionLabel::DeleteObject, + self.container_client.container_name(), + ) .await .map_err(|err| AzureErrorWrapper::from(err).into()); ignore_error_kind!(StorageErrorKind::NotFound, delete_res)?; @@ -524,7 +546,10 @@ impl Storage for AzureBlobStorage { .blob_client(name) .get_properties() .into_future() - .with_count_metric(ActionLabel::HeadObject) + .with_count_metric( + ActionLabel::HeadObject, + self.container_client.container_name(), + ) .await; match properties_result { Ok(response) => Ok(response.blob.properties.content_length), @@ -585,6 +610,7 @@ const AZURE_STREAM_ERROR: &str = "stream"; async fn download_all( chunk_stream: &mut Pageable, download_kind: DownloadKind, + container_name: &str, ) -> Result { // The guard only comes into existence once a response has arrived: a // failure or a cancellation before that is a request failure, recorded as @@ -595,7 +621,7 @@ async fn download_all( let mut total_num_bytes: usize = 0; while let Some(chunk_result) = chunk_stream .next() - .with_count_metric(ActionLabel::GetObject) + .with_count_metric(ActionLabel::GetObject, container_name) .await { let chunk_response = chunk_result.map_err(|error| { 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 69a14a15aed..2c338b4e41e 100644 --- a/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage.rs +++ b/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage.rs @@ -403,7 +403,7 @@ impl S3CompatibleObjectStorage { } req_builder .send() - .with_count_and_upload_metrics(ActionLabel::PutObject, len) + .with_count_and_upload_metrics(ActionLabel::PutObject, bucket, len) .await .map_err(|sdk_error| { if sdk_error.is_retryable() { @@ -456,7 +456,7 @@ impl S3CompatibleObjectStorage { } req_builder .send() - .with_count_metric(ActionLabel::CreateMultipartUpload) + .with_count_metric(ActionLabel::CreateMultipartUpload, &self.bucket) .await }) .await? @@ -574,7 +574,7 @@ impl S3CompatibleObjectStorage { } let upload_part_output = req_builder .send() - .with_count_and_upload_metrics(ActionLabel::UploadPart, part.len()) + .with_count_and_upload_metrics(ActionLabel::UploadPart, &self.bucket, part.len()) .await .map_err(|s3_err| { if s3_err.is_retryable() { @@ -654,7 +654,7 @@ impl S3CompatibleObjectStorage { .multipart_upload(completed_upload.clone()) .upload_id(upload_id) .send() - .with_count_metric(ActionLabel::CompleteMultipartUpload) + .with_count_metric(ActionLabel::CompleteMultipartUpload, &self.bucket) .await }) .await?; @@ -669,7 +669,7 @@ impl S3CompatibleObjectStorage { .key(key) .upload_id(upload_id) .send() - .with_count_metric(ActionLabel::AbortMultipartUpload) + .with_count_metric(ActionLabel::AbortMultipartUpload, &self.bucket) .await }) .await?; @@ -708,7 +708,7 @@ impl S3CompatibleObjectStorage { let connection_permit = acquire_get_object_connection_permit().await; let get_object_output = req_builder .send() - .with_count_and_duration_metrics(ActionLabel::GetObject) + .with_count_and_duration_metrics(ActionLabel::GetObject, &self.bucket) .await?; Ok((get_object_output, connection_permit)) } @@ -822,7 +822,7 @@ impl S3CompatibleObjectStorage { .bucket(self.bucket.clone()) .delete(delete.clone()) .send() - .with_count_and_duration_metrics(ActionLabel::DeleteObjects) + .with_count_and_duration_metrics(ActionLabel::DeleteObjects, &self.bucket) .await }) .await @@ -893,7 +893,7 @@ impl Storage for S3CompatibleObjectStorage { .bucket(self.bucket.clone()) .max_keys(1) .send() - .with_count_metric(ActionLabel::ListObjects) + .with_count_metric(ActionLabel::ListObjects, &self.bucket) .await?; Ok(()) } @@ -936,7 +936,7 @@ impl Storage for S3CompatibleObjectStorage { .bucket(&bucket) .key(&key) .send() - .with_count_and_duration_metrics(ActionLabel::DeleteObject) + .with_count_and_duration_metrics(ActionLabel::DeleteObject, &bucket) .await }) .await; @@ -1024,7 +1024,7 @@ impl Storage for S3CompatibleObjectStorage { } req_builder .send() - .with_count_metric(ActionLabel::HeadObject) + .with_count_metric(ActionLabel::HeadObject, &bucket) .await }) .await?; diff --git a/quickwit/quickwit-storage/src/opendal_storage/base.rs b/quickwit/quickwit-storage/src/opendal_storage/base.rs index 819e4e1f713..e101d7e76f9 100644 --- a/quickwit/quickwit-storage/src/opendal_storage/base.rs +++ b/quickwit/quickwit-storage/src/opendal_storage/base.rs @@ -82,7 +82,7 @@ impl Storage for OpendalStorage { async fn put(&self, path: &Path, payload: Box) -> StorageResult<()> { // we record one put even though it may involve multiple parts - opendal_helpers::record_request(ActionLabel::PutObject); + opendal_helpers::record_request(ActionLabel::PutObject, &self.op.info().name()); let path = path.as_os_str().to_string_lossy(); let mut payload_reader = payload.byte_stream().await?.into_async_read(); @@ -121,7 +121,7 @@ impl Storage for OpendalStorage { // Unlike other object store implementations, in flight requests are // recorded before issuing the query to the object store. let _inflight_guards = object_storage_get_slice_in_flight_guards(size); - opendal_helpers::record_request(ActionLabel::GetObject); + opendal_helpers::record_request(ActionLabel::GetObject, &self.op.info().name()); // `Buffer::to_bytes` is zero-copy when the underlying buffer is contiguous, and coalesces // into a single `Bytes` otherwise — avoiding the extra `Vec` round-trip `to_vec` would // perform. @@ -154,7 +154,10 @@ impl Storage for OpendalStorage { async fn delete(&self, path: &Path) -> StorageResult<()> { let path = path.as_os_str().to_string_lossy(); - let _timer = opendal_helpers::record_request_with_timer(ActionLabel::DeleteObject); + let _timer = opendal_helpers::record_request_with_timer( + ActionLabel::DeleteObject, + &self.op.info().name(), + ); self.op.delete(&path).await?; Ok(()) } @@ -168,8 +171,10 @@ impl Storage for OpendalStorage { if storage_info.name().starts_with("sample-bucket") && storage_info.scheme() == "gcs" { let mut bulk_error = BulkDeleteError::default(); for (index, path) in paths.iter().enumerate() { - let _timer = - opendal_helpers::record_request_with_timer(ActionLabel::DeleteObjects); + let _timer = opendal_helpers::record_request_with_timer( + ActionLabel::DeleteObjects, + &storage_info.name(), + ); let result = self.op.delete(&path.as_os_str().to_string_lossy()).await; if let Err(err) = result { let storage_error_kind = err.kind(); From c05b2028b0f64c7f7fb895e99210cdda1bd76f95 Mon Sep 17 00:00:00 2001 From: Darkheir Date: Fri, 23 Oct 2026 11:46:26 +0200 Subject: [PATCH 2/3] Clarify split retry metric help text A retry goes back to the same searcher when it is the only one available. Co-Authored-By: Claude Opus 5.5 --- quickwit/quickwit-search/src/metrics.rs | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/quickwit/quickwit-search/src/metrics.rs b/quickwit/quickwit-search/src/metrics.rs index fc3cfaa0ad8..ba375203d1a 100644 --- a/quickwit/quickwit-search/src/metrics.rs +++ b/quickwit/quickwit-search/src/metrics.rs @@ -287,8 +287,9 @@ impl Default for SearchMetrics { ), leaf_search_split_retries_total: new_counter_vec( "leaf_search_split_retries_total", - "Number of splits whose leaf search failed and was retried on another searcher, \ - by outcome of the retry: `rescued` or `failed` again.", + "Number of splits whose leaf search failed and was retried once (on another \ + searcher when one is available), by outcome of the retry: `rescued` or `failed` \ + again.", "search", &[], ["outcome"], From 29e299969cb14efa6983a9edb30ba93cdb7e68e7 Mon Sep 17 00:00:00 2001 From: Darkheir Date: Fri, 23 Oct 2026 14:27:05 +0200 Subject: [PATCH 3/3] feat: AWS retries metric Signed-off-by: Darkheir --- quickwit/quickwit-storage/src/metrics.rs | 12 + .../quickwit-storage/src/metrics_wrappers.rs | 226 ++++++++++++++++ .../object_storage/s3_compatible_storage.rs | 243 ++++++++++-------- 3 files changed, 369 insertions(+), 112 deletions(-) diff --git a/quickwit/quickwit-storage/src/metrics.rs b/quickwit/quickwit-storage/src/metrics.rs index 5e5106e45f7..4bab6d8d516 100644 --- a/quickwit/quickwit-storage/src/metrics.rs +++ b/quickwit/quickwit-storage/src/metrics.rs @@ -39,6 +39,7 @@ pub struct StorageMetrics { pub get_slice_timeout_all_timeouts: IntCounter, pub object_storage_requests_total: IntCounterVec<3>, pub object_storage_request_duration: HistogramVec<3>, + pub object_storage_operation_attempts: HistogramVec<2>, pub object_storage_get_slice_in_flight_count: IntGauge, pub object_storage_get_slice_in_flight_num_bytes: IntGauge, pub object_storage_get_object_connections_in_use: IntGauge, @@ -95,6 +96,17 @@ impl Default for StorageMetrics { ["action", "status", "bucket"], vec![0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0], ), + object_storage_operation_attempts: new_histogram_vec( + "object_storage_operation_attempts", + "Number of attempts per object store operation (a request and its retries), by \ + action and outcome: `success`, `exhausted` (every attempt failed with a \ + retryable error), `permanent_error`, `cancelled_during_attempt` or \ + `cancelled_during_backoff`.", + "storage", + &[], + ["action", "outcome"], + vec![1.0, 2.0, 3.0, 4.0, 5.0], + ), object_storage_get_slice_in_flight_count: new_gauge( "object_storage_get_slice_in_flight_count", "Number of get_object for which the memory was allocated but the download is \ diff --git a/quickwit/quickwit-storage/src/metrics_wrappers.rs b/quickwit/quickwit-storage/src/metrics_wrappers.rs index bd4587b8d49..0cdf02ee3d0 100644 --- a/quickwit/quickwit-storage/src/metrics_wrappers.rs +++ b/quickwit/quickwit-storage/src/metrics_wrappers.rs @@ -12,9 +12,11 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::fmt::Debug; use std::io; use std::marker::PhantomData; use std::pin::Pin; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::task::{Context, Poll, ready}; use std::time::Instant; @@ -22,6 +24,8 @@ use aws_smithy_types::byte_stream::ByteStream; use bytes::{Bytes, BytesMut}; use once_cell::sync::Lazy; use pin_project::{pin_project, pinned_drop}; +use quickwit_aws::retry::{AwsRetryable, aws_retry}; +use quickwit_common::retry::RetryParams; use quickwit_common::thread_pool::with_priority::{Priority, ThreadPoolWithPriority}; use tokio::io::{AsyncBufRead, AsyncWrite}; @@ -231,6 +235,89 @@ where } } +/// Runs an object store operation with [`aws_retry`] and records the number of attempts it took, +/// by outcome: `success`, `exhausted` (every attempt failed with a retryable error), +/// `permanent_error`, `cancelled_during_attempt` or `cancelled_during_backoff`. +/// +/// The cancellation outcomes record operations dropped by the caller, for instance on a timeout. +/// A cancellation while waiting between two attempts is not visible in the request metrics. +pub(crate) async fn aws_retry_with_metrics( + retry_params: &RetryParams, + action: ActionLabel, + f: impl Fn() -> Fut, +) -> Result +where + Fut: Future>, + E: AwsRetryable + Debug + 'static, +{ + let mut attempts_tracker = AttemptsTracker::new(action); + let result = aws_retry(retry_params, || { + let attempts_tracker = &attempts_tracker; + let attempt = f(); + async move { + attempts_tracker.start_attempt(); + let result = attempt.await; + attempts_tracker.end_attempt(); + result + } + }) + .await; + attempts_tracker.outcome_opt = Some(match &result { + Ok(_) => "success", + Err(error) if error.is_retryable() => "exhausted", + Err(_) => "permanent_error", + }); + result +} + +/// Counts the attempts of an operation, and records them when dropped. +struct AttemptsTracker { + action: ActionLabel, + num_attempts: AtomicUsize, + attempt_in_progress: AtomicBool, + // `None` until the operation completes: dropping the tracker before that means the operation + // was cancelled. + outcome_opt: Option<&'static str>, +} + +impl AttemptsTracker { + fn new(action: ActionLabel) -> Self { + Self { + action, + num_attempts: AtomicUsize::new(0), + attempt_in_progress: AtomicBool::new(false), + outcome_opt: None, + } + } + + fn start_attempt(&self) { + self.num_attempts.fetch_add(1, Ordering::Relaxed); + self.attempt_in_progress.store(true, Ordering::Relaxed); + } + + fn end_attempt(&self) { + self.attempt_in_progress.store(false, Ordering::Relaxed); + } +} + +impl Drop for AttemptsTracker { + fn drop(&mut self) { + let num_attempts = *self.num_attempts.get_mut(); + if num_attempts == 0 { + return; + } + let outcome = match self.outcome_opt { + Some(outcome) => outcome, + None if *self.attempt_in_progress.get_mut() => "cancelled_during_attempt", + None => "cancelled_during_backoff", + }; + STORAGE_METRICS + .object_storage_operation_attempts + .with_label_values([self.action.as_str(), outcome]) + .observe(num_attempts as f64); + } +} + mod s3_impls { use aws_sdk_s3::error::{ProvideErrorMetadata, SdkError}; @@ -670,8 +757,147 @@ pub mod opendal_helpers { #[cfg(test)] mod tests { + use std::sync::Arc; + use std::time::Duration; + + use quickwit_common::metrics::Histogram; + use super::*; + #[derive(Debug)] + enum TestError { + Retryable, + Permanent, + } + + impl AwsRetryable for TestError { + fn is_retryable(&self) -> bool { + matches!(self, TestError::Retryable) + } + } + + // No other test runs an operation labelled `list_objects`, and each test below checks a + // distinct outcome, so they don't interfere with each other. + fn list_objects_attempts(outcome: &str) -> Histogram { + STORAGE_METRICS + .object_storage_operation_attempts + .with_label_values(["list_objects", outcome]) + } + + /// Runs `operation` and returns the number of operations and attempts recorded for `outcome`. + async fn record_operation(outcome: &str, operation: impl Future) -> (u64, f64) { + let histogram = list_objects_attempts(outcome); + let (count_before, sum_before) = (histogram.get_sample_count(), histogram.get_sample_sum()); + operation.await; + ( + histogram.get_sample_count() - count_before, + histogram.get_sample_sum() - sum_before, + ) + } + + /// Fails with `error` on the first `num_failures` attempts, then succeeds. + fn fail_then_succeed( + num_failures: usize, + error: fn() -> TestError, + ) -> impl Fn() -> futures::future::Ready> { + let num_calls = Arc::new(AtomicUsize::new(0)); + move || { + let num_previous_calls = num_calls.fetch_add(1, Ordering::Relaxed); + futures::future::ready(if num_previous_calls < num_failures { + Err(error()) + } else { + Ok(()) + }) + } + } + + #[tokio::test] + async fn test_aws_retry_with_metrics_success_after_retries() { + let retry_params = RetryParams::for_test(); + let recorded = record_operation("success", async { + aws_retry_with_metrics( + &retry_params, + ActionLabel::ListObjects, + fail_then_succeed(2, || TestError::Retryable), + ) + .await + .unwrap(); + }) + .await; + assert_eq!(recorded, (1, 3.0)); + } + + #[tokio::test] + async fn test_aws_retry_with_metrics_exhausted() { + let retry_params = RetryParams::for_test(); + let recorded = record_operation("exhausted", async { + aws_retry_with_metrics( + &retry_params, + ActionLabel::ListObjects, + fail_then_succeed(usize::MAX, || TestError::Retryable), + ) + .await + .unwrap_err(); + }) + .await; + assert_eq!(recorded, (1, retry_params.max_attempts as f64)); + } + + #[tokio::test] + async fn test_aws_retry_with_metrics_permanent_error() { + let retry_params = RetryParams::for_test(); + let recorded = record_operation("permanent_error", async { + aws_retry_with_metrics( + &retry_params, + ActionLabel::ListObjects, + fail_then_succeed(usize::MAX, || TestError::Permanent), + ) + .await + .unwrap_err(); + }) + .await; + assert_eq!(recorded, (1, 1.0)); + } + + #[tokio::test] + async fn test_aws_retry_with_metrics_cancelled_during_backoff() { + tokio::time::pause(); + let retry_params = RetryParams { + base_delay: Duration::from_secs(10), + max_delay: Duration::from_secs(10), + max_attempts: 3, + }; + let recorded = record_operation("cancelled_during_backoff", async { + let operation = aws_retry_with_metrics( + &retry_params, + ActionLabel::ListObjects, + fail_then_succeed(usize::MAX, || TestError::Retryable), + ); + // The first attempt fails at once, and the timeout elapses during the backoff. + tokio::time::timeout(Duration::from_secs(1), operation) + .await + .unwrap_err(); + }) + .await; + assert_eq!(recorded, (1, 1.0)); + } + + #[tokio::test] + async fn test_aws_retry_with_metrics_cancelled_during_attempt() { + tokio::time::pause(); + let retry_params = RetryParams::for_test(); + let recorded = record_operation("cancelled_during_attempt", async { + let operation = aws_retry_with_metrics(&retry_params, ActionLabel::ListObjects, || { + futures::future::pending::>() + }); + tokio::time::timeout(Duration::from_secs(1), operation) + .await + .unwrap_err(); + }) + .await; + assert_eq!(recorded, (1, 1.0)); + } + #[tokio::test] async fn test_coalesce_segments_does_not_copy_a_single_segment() { let segment = Bytes::from_static(b"warmup payload"); 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 2c338b4e41e..e582396fc45 100644 --- a/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage.rs +++ b/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage.rs @@ -36,7 +36,7 @@ use bytes::Bytes; use futures::{StreamExt, stream}; use once_cell::sync::{Lazy, OnceCell}; use quickwit_aws::http_client::s3_http_client; -use quickwit_aws::retry::{AwsRetryable, aws_retry}; +use quickwit_aws::retry::AwsRetryable; use quickwit_aws::{aws_behavior_version, get_aws_config}; use quickwit_common::metrics::GaugeGuard; use quickwit_common::retry::{Retry, RetryParams}; @@ -51,8 +51,8 @@ use tracing::{info, instrument, warn}; use crate::metrics::{STORAGE_METRICS, object_storage_get_slice_in_flight_guards}; use crate::metrics_wrappers::{ - ActionLabel, DownloadKind, RequestMetricsWrapperExt, collect_with_download_metrics, - copy_with_download_metrics, + ActionLabel, DownloadKind, RequestMetricsWrapperExt, aws_retry_with_metrics, + collect_with_download_metrics, copy_with_download_metrics, }; use crate::object_storage::MultiPartPolicy; use crate::stable_deref_bytes::into_owned_bytes; @@ -422,7 +422,7 @@ impl S3CompatibleObjectStorage { len: u64, ) -> StorageResult<()> { let bucket = &self.bucket; - aws_retry(&self.retry_params, || async { + aws_retry_with_metrics(&self.retry_params, ActionLabel::PutObject, || async { self.put_single_part_single_try(bucket, key, payload.clone(), len) .await }) @@ -432,33 +432,37 @@ impl S3CompatibleObjectStorage { } async fn create_multipart_upload(&self, key: &str) -> StorageResult { - let upload_id = aws_retry(&self.retry_params, || async { - let mut req_builder = self - .s3_client - .create_multipart_upload() - .bucket(self.bucket.clone()) - .key(key); - match &self.encryption { - Some(S3EncryptionConfig::SseC { - key, - key_md5, - read_only: false, - }) => { - req_builder = req_builder - .set_sse_customer_algorithm(Some("AES256".to_string())) - .set_sse_customer_key(Some(key.clone())) - .set_sse_customer_key_md5(Some(key_md5.clone())); + let upload_id = aws_retry_with_metrics( + &self.retry_params, + ActionLabel::CreateMultipartUpload, + || async { + let mut req_builder = self + .s3_client + .create_multipart_upload() + .bucket(self.bucket.clone()) + .key(key); + match &self.encryption { + Some(S3EncryptionConfig::SseC { + key, + key_md5, + read_only: false, + }) => { + req_builder = req_builder + .set_sse_customer_algorithm(Some("AES256".to_string())) + .set_sse_customer_key(Some(key.clone())) + .set_sse_customer_key_md5(Some(key_md5.clone())); + } + Some(S3EncryptionConfig::SseC { + read_only: true, .. + }) => {} + None => {} } - Some(S3EncryptionConfig::SseC { - read_only: true, .. - }) => {} - None => {} - } - req_builder - .send() - .with_count_metric(ActionLabel::CreateMultipartUpload, &self.bucket) - .await - }) + req_builder + .send() + .with_count_metric(ActionLabel::CreateMultipartUpload, &self.bucket) + .await + }, + ) .await? .upload_id .ok_or_else(|| { @@ -607,7 +611,7 @@ impl S3CompatibleObjectStorage { stream::iter(parts.into_iter().map(|part| { let payload = payload.clone(); let upload_id = upload_id.clone(); - aws_retry(&self.retry_params, move || { + aws_retry_with_metrics(&self.retry_params, ActionLabel::UploadPart, move || { self.upload_part(upload_id.clone(), key, part.clone(), payload.clone()) }) })) @@ -646,32 +650,40 @@ impl S3CompatibleObjectStorage { let completed_upload = CompletedMultipartUpload::builder() .set_parts(Some(completed_parts)) .build(); - aws_retry(&self.retry_params, || async { - self.s3_client - .complete_multipart_upload() - .bucket(self.bucket.clone()) - .key(key) - .multipart_upload(completed_upload.clone()) - .upload_id(upload_id) - .send() - .with_count_metric(ActionLabel::CompleteMultipartUpload, &self.bucket) - .await - }) + aws_retry_with_metrics( + &self.retry_params, + ActionLabel::CompleteMultipartUpload, + || async { + self.s3_client + .complete_multipart_upload() + .bucket(self.bucket.clone()) + .key(key) + .multipart_upload(completed_upload.clone()) + .upload_id(upload_id) + .send() + .with_count_metric(ActionLabel::CompleteMultipartUpload, &self.bucket) + .await + }, + ) .await?; Ok(()) } async fn abort_multipart_upload(&self, key: &str, upload_id: &str) -> StorageResult<()> { - aws_retry(&self.retry_params, || async { - self.s3_client - .abort_multipart_upload() - .bucket(self.bucket.clone()) - .key(key) - .upload_id(upload_id) - .send() - .with_count_metric(ActionLabel::AbortMultipartUpload, &self.bucket) - .await - }) + aws_retry_with_metrics( + &self.retry_params, + ActionLabel::AbortMultipartUpload, + || async { + self.s3_client + .abort_multipart_upload() + .bucket(self.bucket.clone()) + .key(key) + .upload_id(upload_id) + .send() + .with_count_metric(ActionLabel::AbortMultipartUpload, &self.bucket) + .await + }, + ) .await?; Ok(()) } @@ -725,32 +737,33 @@ impl S3CompatibleObjectStorage { }; // 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(&self.retry_params, || async { - let (get_object_output, _connection_permit) = self - .get_object(path, range_opt.clone()) + 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(|sdk_error| { + if sdk_error.is_retryable() { + Retry::Transient(StorageError::from(sdk_error)) + } else { + Retry::Permanent(StorageError::from(sdk_error)) + } + })?; + // 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 - .map_err(|sdk_error| { - if sdk_error.is_retryable() { - Retry::Transient(StorageError::from(sdk_error)) - } else { - Retry::Permanent(StorageError::from(sdk_error)) - } - })?; - // 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(), - ) + // GET requests are idempotent, and a body error is a transport failure. + .map_err(|io_error| Retry::Transient(StorageError::from(io_error))) + }) .await - // GET requests are idempotent, and a body error is a transport failure. - .map_err(|io_error| Retry::Transient(StorageError::from(io_error))) - }) - .await - .map_err(Retry::into_inner)?; + .map_err(Retry::into_inner)?; Ok(payload) } @@ -816,7 +829,7 @@ impl S3CompatibleObjectStorage { for (path_chunk, delete) in &mut delete_requests_it { let delete_objects_res: StorageResult = - aws_retry(&self.retry_params, || async { + aws_retry_with_metrics(&self.retry_params, ActionLabel::DeleteObjects, || async { self.s3_client .delete_objects() .bucket(self.bucket.clone()) @@ -917,7 +930,10 @@ impl Storage for S3CompatibleObjectStorage { async fn copy_to(&self, path: &Path, output: &mut dyn SendableAsync) -> StorageResult<()> { let (get_object_output, connection_permit) = - aws_retry(&self.retry_params, || self.get_object(path, None)).await?; + aws_retry_with_metrics(&self.retry_params, ActionLabel::GetObject, || { + self.get_object(path, None) + }) + .await?; let mut body_read = BufReader::new(get_object_output.body.into_async_read()); copy_with_download_metrics(&mut body_read, output, DownloadKind::Object).await?; // The body has been fully read: release the connection before flushing the output. @@ -930,16 +946,17 @@ impl Storage for S3CompatibleObjectStorage { async fn delete(&self, path: &Path) -> StorageResult<()> { let bucket = self.bucket.clone(); let key = self.key(path); - let delete_res = aws_retry(&self.retry_params, || async { - self.s3_client - .delete_object() - .bucket(&bucket) - .key(&key) - .send() - .with_count_and_duration_metrics(ActionLabel::DeleteObject, &bucket) - .await - }) - .await; + let delete_res = + aws_retry_with_metrics(&self.retry_params, ActionLabel::DeleteObject, || async { + self.s3_client + .delete_object() + .bucket(&bucket) + .key(&key) + .send() + .with_count_and_duration_metrics(ActionLabel::DeleteObject, &bucket) + .await + }) + .await; match delete_res { Ok(_) => Ok(()), @@ -977,10 +994,11 @@ impl Storage for S3CompatibleObjectStorage { path: &Path, range: Range, ) -> crate::StorageResult> { - let (get_object_output, connection_permit) = aws_retry(&self.retry_params, || { - self.get_object(path, Some(range.clone())) - }) - .await?; + let (get_object_output, connection_permit) = + aws_retry_with_metrics(&self.retry_params, ActionLabel::GetObject, || { + self.get_object(path, Some(range.clone())) + }) + .await?; Ok(Box::new(S3AsyncRead { read: get_object_output.body.into_async_read(), _connection_permit: connection_permit, @@ -1007,27 +1025,28 @@ impl Storage for S3CompatibleObjectStorage { async fn file_num_bytes(&self, path: &Path) -> StorageResult { let bucket = self.bucket.clone(); let key = self.key(path); - let head_object_output = aws_retry(&self.retry_params, || async { - let mut req_builder = self.s3_client.head_object().bucket(&bucket).key(&key); - match &self.encryption { - Some(S3EncryptionConfig::SseC { - key, - key_md5, - read_only: _, - }) => { - req_builder = req_builder - .set_sse_customer_algorithm(Some("AES256".to_string())) - .set_sse_customer_key(Some(key.clone())) - .set_sse_customer_key_md5(Some(key_md5.clone())); + let head_object_output = + aws_retry_with_metrics(&self.retry_params, ActionLabel::HeadObject, || async { + let mut req_builder = self.s3_client.head_object().bucket(&bucket).key(&key); + match &self.encryption { + Some(S3EncryptionConfig::SseC { + key, + key_md5, + read_only: _, + }) => { + req_builder = req_builder + .set_sse_customer_algorithm(Some("AES256".to_string())) + .set_sse_customer_key(Some(key.clone())) + .set_sse_customer_key_md5(Some(key_md5.clone())); + } + None => {} } - None => {} - } - req_builder - .send() - .with_count_metric(ActionLabel::HeadObject, &bucket) - .await - }) - .await?; + req_builder + .send() + .with_count_metric(ActionLabel::HeadObject, &bucket) + .await + }) + .await?; Ok(head_object_output.content_length().unwrap_or(0) as u64) }