Cleanup integration tests (#5429)

* Uncomment some tests

* Remove ingest retries

* Fix lint and comment flaky test

* Remove ingest from basic test

* Improve wording

* Ignore test known to be flaky

* Reference issue instead of TODO

* Add back directory cleanup test with link to issue

* Assert shutdown timeout

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

@@ -146,6 +146,7 @@ impl ClusterSandboxBuilder {
         }
     }
 
+    /// Builds the cluster config, starts the nodes and waits for them to be ready
     pub async fn build_and_start(self) -> ClusterSandbox {
         self.build_config().await.start().await
     }
@@ -265,33 +266,15 @@ macro_rules! ingest_json {
     };
 }
 
-pub(crate) async fn ingest_with_retry(
+pub(crate) async fn ingest(
     client: &QuickwitClient,
     index_id: &str,
     ingest_source: IngestSource,
     commit_type: CommitType,
 ) -> anyhow::Result<()> {
-    wait_until_predicate(
-        || {
-            let commit_type_clone = commit_type;
-            let ingest_source_clone = ingest_source.clone();
-            async move {
-                // Index one record.
-                if let Err(err) = client
-                    .ingest(index_id, ingest_source_clone, None, None, commit_type_clone)
-                    .await
-                {
-                    debug!(index=%index_id, err=%err, "failed to ingest");
-                    false
-                } else {
-                    true
-                }
-            }
-        },
-        Duration::from_secs(10),
-        Duration::from_millis(100),
-    )
-    .await?;
+    client
+        .ingest(index_id, ingest_source, None, None, commit_type)
+        .await?;
     Ok(())
 }
 
@@ -495,16 +478,17 @@ impl ClusterSandbox {
     /// Shutdown nodes that only provide the specified services
     pub async fn shutdown_services(
         &mut self,
-        shutdown_services: &HashSet<QuickwitService>,
+        shutdown_services: impl IntoIterator<Item = QuickwitService>,
     ) -> Result<Vec<HashMap<String, ActorExitStatus>>, anyhow::Error> {
         // We need to drop rest clients first because reqwest can hold connections open
         // preventing rest server's graceful shutdown.
         let mut shutdown_futures = Vec::new();
         let mut shutdown_nodes = HashMap::new();
         let mut i = 0;
+        let shutdown_services_map = HashSet::from_iter(shutdown_services);
         while i < self.node_shutdown_handles.len() {
             let handler_services = &self.node_shutdown_handles[i].node_services;
-            if handler_services.is_subset(shutdown_services) {
+            if handler_services.is_subset(&shutdown_services_map) {
                 let handler_to_shutdown = self.node_shutdown_handles.remove(i);
                 shutdown_nodes.insert(
                     handler_to_shutdown.node_id.clone(),
@@ -527,7 +511,7 @@ impl ClusterSandbox {
     pub async fn shutdown(
         mut self,
     ) -> Result<Vec<HashMap<String, ActorExitStatus>>, anyhow::Error> {
-        self.shutdown_services(&QuickwitService::supported_services())
+        self.shutdown_services(QuickwitService::supported_services())
             .await
     }
 }

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

@@ -20,4 +20,4 @@
 mod cluster_sandbox;
 mod shutdown;
 
-pub(crate) use cluster_sandbox::{ingest_with_retry, ClusterSandbox, ClusterSandboxBuilder};
+pub(crate) use cluster_sandbox::{ingest, ClusterSandbox, ClusterSandboxBuilder};

quickwit/quickwit-integration-tests/src/tests/basic_tests.rs

@@ -17,25 +17,13 @@
 // You should have received a copy of the GNU Affero General Public License
 // along with this program. If not, see <http://www.gnu.org/licenses/>.
 
-use std::path::PathBuf;
-use std::str::FromStr;
 use std::time::Duration;
 
 use hyper::{Body, Method, Request, StatusCode};
 use quickwit_config::service::QuickwitService;
-use quickwit_rest_client::models::IngestSource;
-use quickwit_rest_client::rest_client::CommitType;
 use quickwit_serve::SearchRequestQueryString;
 
-use crate::test_utils::{ingest_with_retry, ClusterSandboxBuilder};
-
-fn get_ndjson_filepath(ndjson_dataset_filename: &str) -> String {
-    format!(
-        "{}/resources/tests/{}",
-        env!("CARGO_MANIFEST_DIR"),
-        ndjson_dataset_filename
-    )
-}
+use crate::test_utils::ClusterSandboxBuilder;
 
 #[tokio::test]
 async fn test_ui_redirect_on_get() {
@@ -130,20 +118,6 @@ async fn test_multi_nodes_cluster() {
         .build_and_start()
         .await;
 
-    {
-        // Wait for indexer to fully start.
-        // The starting time is a bit long for a cluster.
-        tokio::time::sleep(Duration::from_secs(3)).await;
-        let indexing_service_counters = sandbox
-            .indexer_rest_client
-            .node_stats()
-            .indexing()
-            .await
-            .unwrap();
-        assert_eq!(indexing_service_counters.num_running_pipelines, 0);
-    }
-
-    // Create index
     sandbox
         .indexer_rest_client
         .indexes()
@@ -163,17 +137,18 @@ async fn test_multi_nodes_cluster() {
         )
         .await
         .unwrap();
+
     assert!(sandbox
         .indexer_rest_client
         .node_health()
         .is_live()
         .await
         .unwrap());
 
-    // Wait until indexing pipelines are started.
+    // Assert that at least 1 indexing pipelines is successfully started
     sandbox.wait_for_indexing_pipelines(1).await.unwrap();
 
-    // Check search is working.
+    // Check that search is working
     let search_response_empty = sandbox
         .searcher_rest_client
         .search(
@@ -187,30 +162,5 @@ async fn test_multi_nodes_cluster() {
         .unwrap();
     assert_eq!(search_response_empty.num_hits, 0);
 
-    // Check that ingest request send to searcher is forwarded to indexer and thus indexed.
-    let ndjson_filepath = get_ndjson_filepath("documents_to_ingest.json");
-    let ingest_source = IngestSource::File(PathBuf::from_str(&ndjson_filepath).unwrap());
-    ingest_with_retry(
-        &sandbox.searcher_rest_client,
-        "my-new-multi-node-index",
-        ingest_source,
-        CommitType::Auto,
-    )
-    .await
-    .unwrap();
-    // Wait until split is committed and search.
-    tokio::time::sleep(Duration::from_secs(4)).await;
-    let search_response_one_hit = sandbox
-        .searcher_rest_client
-        .search(
-            "my-new-multi-node-index",
-            SearchRequestQueryString {
-                query: "body:bar".to_string(),
-                ..Default::default()
-            },
-        )
-        .await
-        .unwrap();
-    assert_eq!(search_response_one_hit.num_hits, 1);
     sandbox.shutdown().await.unwrap();
 }