Skip cpu if cannot do better (#6001)

* Skipping running tantivy search if cannot do better.

Also,
- removes some noisy intermediary traces/span.
- logs the ast/agg in the root node.
- logs a summary about the split execution (cache, pruned, etc.) in the leaf node

Co-authored-by: fulmicoton <paul.masurel@datadoghq.com>

Author: fulmicoton

quickwit/quickwit-cli/src/lib.rs

@@ -16,6 +16,7 @@
 
 use std::collections::HashSet;
 use std::str::FromStr;
+use std::sync::OnceLock;
 
 use anyhow::Context;
 use clap::{Arg, ArgMatches, arg};
@@ -107,6 +108,17 @@ fn client_args() -> Vec<Arg> {
     ]
 }
 
+pub fn install_default_crypto_ring_provider() {
+    static CALL_ONLY_ONCE: OnceLock<Result<(), ()>> = OnceLock::new();
+    CALL_ONLY_ONCE
+        .get_or_init(|| {
+            rustls::crypto::ring::default_provider()
+                .install_default()
+                .map_err(|_| ())
+        })
+        .expect("rustls crypto ring default provider installation should not fail");
+}
+
 #[derive(Debug, Eq, PartialEq)]
 pub struct ClientArgs {
     pub cluster_endpoint: Url,

quickwit/quickwit-cli/src/main.rs

@@ -19,12 +19,12 @@ use std::collections::BTreeMap;
 use anyhow::Context;
 use colored::Colorize;
 use opentelemetry::global;
-use quickwit_cli::busy_detector;
 use quickwit_cli::checklist::RED_COLOR;
 use quickwit_cli::cli::{CliCommand, build_cli};
 #[cfg(feature = "jemalloc")]
 use quickwit_cli::jemalloc::start_jemalloc_metrics_loop;
 use quickwit_cli::logger::setup_logging_and_tracing;
+use quickwit_cli::{busy_detector, install_default_crypto_ring_provider};
 use quickwit_common::runtimes::scrape_tokio_runtime_metrics;
 use quickwit_serve::BuildInfo;
 use tracing::error;
@@ -93,9 +93,7 @@ async fn main_impl() -> anyhow::Result<()> {
         }
     };
 
-    rustls::crypto::ring::default_provider()
-        .install_default()
-        .expect("rustls crypto ring default provider installation should not fail");
+    install_default_crypto_ring_provider();
 
     #[cfg(feature = "jemalloc")]
     start_jemalloc_metrics_loop();

quickwit/quickwit-common/src/io.rs

@@ -377,7 +377,7 @@ mod tests {
         }
         controlled_write.flush().await.unwrap();
         let elapsed = start.elapsed();
-        assert!(elapsed <= Duration::from_millis(5));
+        assert!(elapsed <= Duration::from_millis(10));
         assert_eq!(io_controls.num_bytes(), 2_000_000u64);
     }
 

quickwit/quickwit-common/src/metrics.rs

@@ -40,6 +40,26 @@ pub struct IntCounterVec<const N: usize> {
 }
 
 impl<const N: usize> IntCounterVec<N> {
+    pub fn new(
+        name: &str,
+        help: &str,
+        subsystem: &str,
+        const_labels: &[(&str, &str)],
+        label_names: [&str; N],
+    ) -> IntCounterVec<N> {
+        let owned_const_labels: HashMap<String, String> = const_labels
+            .iter()
+            .map(|(label_name, label_value)| (label_name.to_string(), label_value.to_string()))
+            .collect();
+        let counter_opts = Opts::new(name, help)
+            .namespace("quickwit")
+            .subsystem(subsystem)
+            .const_labels(owned_const_labels);
+        let underlying = PrometheusIntCounterVec::new(counter_opts, &label_names)
+            .expect("failed to create counter vec");
+        IntCounterVec { underlying }
+    }
+
     pub fn with_label_values(&self, label_values: [&str; N]) -> IntCounter {
         self.underlying.with_label_values(&label_values)
     }
@@ -92,21 +112,10 @@ pub fn new_counter_vec<const N: usize>(
     const_labels: &[(&str, &str)],
     label_names: [&str; N],
 ) -> IntCounterVec<N> {
-    let owned_const_labels: HashMap<String, String> = const_labels
-        .iter()
-        .map(|(label_name, label_value)| (label_name.to_string(), label_value.to_string()))
-        .collect();
-    let counter_opts = Opts::new(name, help)
-        .namespace("quickwit")
-        .subsystem(subsystem)
-        .const_labels(owned_const_labels);
-    let underlying = PrometheusIntCounterVec::new(counter_opts, &label_names)
-        .expect("failed to create counter vec");
-
-    let collector = Box::new(underlying.clone());
+    let int_counter_vec = IntCounterVec::new(name, help, subsystem, const_labels, label_names);
+    let collector = Box::new(int_counter_vec.underlying.clone());
     prometheus::register(collector).expect("failed to register counter vec");
-
-    IntCounterVec { underlying }
+    int_counter_vec
 }
 
 pub fn new_float_gauge(

quickwit/quickwit-indexing/src/actors/indexing_pipeline.rs

@@ -733,8 +733,18 @@ mod tests {
         let pipeline = IndexingPipeline::new(pipeline_params);
         let (_pipeline_mailbox, pipeline_handle) = universe.spawn_builder().spawn(pipeline);
         let (pipeline_exit_status, pipeline_statistics) = pipeline_handle.join().await;
-        assert_eq!(pipeline_statistics.generation, 1);
-        assert_eq!(pipeline_statistics.num_spawn_attempts, 1 + num_fails);
+        assert_eq!(
+            pipeline_statistics.generation, 1,
+            "generation is {}, expected 1",
+            pipeline_statistics.generation
+        );
+        assert_eq!(
+            pipeline_statistics.num_spawn_attempts,
+            1 + num_fails,
+            "num spawn attempts is {}, expected 1 + {}",
+            pipeline_statistics.num_spawn_attempts,
+            1 + num_fails
+        );
         assert!(pipeline_exit_status.is_success());
         Ok(())
     }

quickwit/quickwit-integration-tests/src/test_utils/cluster_sandbox.rs

@@ -176,10 +176,7 @@ pub struct ResolvedClusterConfig {
 impl ResolvedClusterConfig {
     /// Start a cluster using this config and waits for the nodes to be ready
     pub async fn start(self) -> ClusterSandbox {
-        rustls::crypto::ring::default_provider()
-            .install_default()
-            .expect("rustls crypto ring default provider installation should not fail");
-
+        quickwit_cli::install_default_crypto_ring_provider();
         let mut node_shutdown_handles = Vec::new();
         let runtimes_config = RuntimesConfig::light_for_tests();
         let storage_resolver = StorageResolver::unconfigured();