Add a quickwit-integration-tests crate (#3200)

Fixes #2984

quickwit/Cargo.toml

@@ -17,6 +17,7 @@ members = [
   "quickwit-grpc-clients",
   "quickwit-indexing",
   "quickwit-ingest",
+  "quickwit-integration-tests",
   "quickwit-jaeger",
   "quickwit-janitor",
   "quickwit-macros",
@@ -210,6 +211,7 @@ quickwit-doc-mapper = { version = "0.5.0", path = "./quickwit-doc-mapper" }
 quickwit-grpc-clients = { version = "0.5.0", path = "./quickwit-grpc-clients" }
 quickwit-indexing = { version = "0.5.0", path = "./quickwit-indexing" }
 quickwit-ingest = { version = "0.5.0", path = "./quickwit-ingest" }
+quickwit-integration-tests = { version = "0.5.0", path = "./quickwit-integration-tests" }
 quickwit-jaeger = { version = "0.5.0", path = "./quickwit-jaeger" }
 quickwit-janitor = { version = "0.5.0", path = "./quickwit-janitor" }
 quickwit-macros = { version = "0.5.0", path = "./quickwit-macros" }

quickwit/quickwit-actors/src/registry.rs

@@ -181,16 +181,18 @@ impl ActorRegistry {
         }
     }
 
-    pub async fn quit(&self) -> Vec<ActorExitStatus> {
+    pub async fn quit(&self) -> HashMap<String, ActorExitStatus> {
         let mut obs_futures = Vec::new();
+        let mut actor_ids = Vec::new();
         for registry_for_type in self.actors.read().unwrap().values() {
             for obs in &registry_for_type.observables {
                 let obs_clone = obs.clone();
                 obs_futures.push(async move { obs_clone.quit().await });
+                actor_ids.push(obs.actor_instance_id().to_string());
             }
         }
         let res = future::join_all(obs_futures).await;
-        res.into_iter().collect()
+        actor_ids.into_iter().zip(res).collect()
     }
 
     pub fn is_empty(&self) -> bool {

quickwit/quickwit-actors/src/universe.rs

@@ -17,6 +17,7 @@
 // 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::collections::HashMap;
 use std::thread;
 use std::time::Duration;
 
@@ -120,7 +121,7 @@ impl Universe {
     }
 
     /// Gracefully quits all registered actors.
-    pub async fn quit(&self) -> Vec<ActorExitStatus> {
+    pub async fn quit(&self) -> HashMap<String, ActorExitStatus> {
         self.spawn_ctx.registry.quit().await
     }
 
@@ -132,7 +133,7 @@ impl Universe {
         assert!(!self
             .quit()
             .await
-            .into_iter()
+            .values()
             .any(|status| matches!(status, ActorExitStatus::Panicked)));
     }
 }
@@ -233,7 +234,10 @@ mod tests {
         universe.sleep(Duration::from_secs(200)).await;
         let res = universe.quit().await;
         assert_eq!(res.len(), 1);
-        assert!(matches!(res.first().unwrap(), ActorExitStatus::Quit));
+        assert!(matches!(
+            res.values().next().unwrap(),
+            ActorExitStatus::Quit
+        ));
         assert!(matches!(handler.quit().await, (ActorExitStatus::Quit, 4)));
     }
 
@@ -246,7 +250,7 @@ mod tests {
         assert!(!universe
             .quit()
             .await
-            .into_iter()
+            .values()
             .any(|status| matches!(status, ActorExitStatus::Panicked)));
     }
 
@@ -260,7 +264,7 @@ mod tests {
         assert!(universe
             .quit()
             .await
-            .into_iter()
+            .values()
             .any(|status| matches!(status, ActorExitStatus::Panicked)));
     }
 

quickwit/quickwit-cli/src/service.rs

@@ -26,6 +26,7 @@ use quickwit_common::uri::Uri;
 use quickwit_config::service::QuickwitService;
 use quickwit_serve::serve_quickwit;
 use quickwit_telemetry::payload::TelemetryEvent;
+use tokio::signal;
 use tracing::debug;
 
 use crate::{config_cli_arg, load_quickwit_config, start_actor_runtimes};
@@ -80,7 +81,12 @@ impl RunCliCommand {
         quickwit_telemetry::send_telemetry_event(telemetry_event).await;
         // TODO move in serve quickwit?
         start_actor_runtimes(&config.enabled_services)?;
-        serve_quickwit(config).await?;
+        let _ = serve_quickwit(config, async move {
+            signal::ctrl_c()
+                .await
+                .expect("Failure listening for CTRL+C signal")
+        })
+        .await?;
         Ok(())
     }
 }

quickwit/quickwit-integration-tests/Cargo.toml

@@ -0,0 +1,39 @@
+[package]
+name = "quickwit-integration-tests"
+version = "0.5.0"
+authors = ["Quickwit, Inc. <[email protected]>"]
+edition = "2021"
+license = "AGPL-3.0-or-later"                           # For a commercial, license, contact [email protected]
+description = "Quickwit's integration tests"
+repository = "https://github.com/quickwit-oss/quickwit"
+homepage = "https://quickwit.io/"
+documentation = "https://quickwit.io/docs/"
+
+[dependencies]
+
+[dev-dependencies]
+anyhow = { workspace = true }
+bytes = { workspace = true }
+chitchat = { workspace = true }
+futures-util = { workspace = true }
+hyper = { workspace = true }
+itertools = { workspace = true }
+rand = { workspace = true }
+reqwest = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+tempfile = { workspace = true }
+tokio = { workspace = true }
+tokio-stream = { workspace = true }
+tracing = { workspace = true }
+
+quickwit-actors = { workspace = true, features = ["testsuite"] }
+quickwit-cluster = { workspace = true, features = ["testsuite"] }
+quickwit-common = { workspace = true, features = ["testsuite"] }
+quickwit-config = { workspace = true, features = ["testsuite"] }
+quickwit-indexing = { workspace = true, features = ["testsuite"] }
+quickwit-metastore = { workspace = true, features = ["testsuite"] }
+quickwit-search = { workspace = true, features = ["testsuite"] }
+quickwit-rest-client = { workspace = true }
+quickwit-serve = { workspace = true }
+quickwit-proto = { workspace = true }

quickwit/quickwit-integration-tests/resources/tests/documents_to_ingest.json

@@ -0,0 +1,3 @@
+{"body":"foo"}
+{"body":"bar"}
+{"body":"baz"}

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

@@ -0,0 +1,23 @@
+// Copyright (C) 2023 Quickwit, Inc.
+//
+// Quickwit is offered under the AGPL v3.0 and as commercial software.
+// For commercial licensing, contact us at [email protected].
+//
+// AGPL:
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as
+// published by the Free Software Foundation, either version 3 of the
+// License, or (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// 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/>.
+
+#[cfg(test)]
+mod test_utils;
+#[cfg(test)]
+mod tests;