feat: Observability of object store queries

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<Ve
     Ok(serialized)
 }
 
+/// Returns the number of retried splits that succeeded (rescued) and that failed again.
+fn split_retry_outcomes(
+    num_retried_splits: usize,
+    retry_result: &crate::Result<LeafSearchResponse>,
+) -> (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

quickwit/quickwit-search/src/error.rs

@@ -170,7 +170,8 @@ impl From<serde_json::Error> for SearchError {
 
 impl From<anyhow::Error> 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:?}"
+        );
+    }
 }

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,15 @@ 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 once (on another \
+                 searcher when one is available), 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 \

quickwit/quickwit-storage/src/error.rs

@@ -81,8 +81,8 @@ impl From<StorageError> 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"
+        );
+    }
+}

quickwit/quickwit-storage/src/metrics.rs

@@ -37,8 +37,9 @@ 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_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,
@@ -79,22 +80,33 @@ 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_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 \