Implement and plug ingest source v2 (#3812)

quickwit-oss · Sep 18, 2023 · 08ab948 · 08ab948
1 parent 2de3fb9
commit 08ab948
Show file tree

Hide file tree

Showing 65 changed files with 2,606 additions and 857 deletions.
diff --git a/quickwit/Cargo.lock b/quickwit/Cargo.lock
diff --git a/quickwit/quickwit-actors/src/actor_handle.rs b/quickwit/quickwit-actors/src/actor_handle.rs
@@ -188,7 +188,7 @@ impl<A: Actor> ActorHandle<A> {
                 return self.wait_for_observable_state_callback(oneshot_rx).await;
             } else {
                 error!(
-                    actor_id = self.actor_context.actor_instance_id(),
+                    actor_id=%self.actor_context.actor_instance_id(),
                     "Failed to send observe message"
                 );
             }

diff --git a/quickwit/quickwit-cli/Cargo.toml b/quickwit/quickwit-cli/Cargo.toml
@@ -58,10 +58,11 @@ quickwit-actors = { workspace = true }
 quickwit-cluster = { workspace = true }
 quickwit-common = { workspace = true }
 quickwit-config = { workspace = true }
-quickwit-index-management = { workspace = true }
 quickwit-directories = { workspace = true }
 quickwit-doc-mapper = { workspace = true }
+quickwit-index-management = { workspace = true }
 quickwit-indexing = { workspace = true }
+quickwit-ingest = { workspace = true }
 quickwit-metastore = { workspace = true }
 quickwit-proto = { workspace = true }
 quickwit-rest-client = { workspace = true }

diff --git a/quickwit/quickwit-cli/src/service.rs b/quickwit/quickwit-cli/src/service.rs
@@ -75,31 +75,31 @@ impl RunCliCommand {
 
     pub async fn execute(&self) -> anyhow::Result<()> {
         debug!(args = ?self, "run-service");
-        let mut config = load_node_config(&self.config_uri).await?;
+        let mut node_config = load_node_config(&self.config_uri).await?;
         let (storage_resolver, metastore_resolver) =
-            get_resolvers(&config.storage_configs, &config.metastore_configs);
+            get_resolvers(&node_config.storage_configs, &node_config.metastore_configs);
         crate::busy_detector::set_enabled(true);
 
         if let Some(services) = &self.services {
             tracing::info!(services = %services.iter().join(", "), "Setting services from override.");
-            config.enabled_services = services.clone();
+            node_config.enabled_services = services.clone();
         }
         let telemetry_handle_opt =
-            quickwit_telemetry::start_telemetry_loop(quickwit_telemetry_info(&config));
+            quickwit_telemetry::start_telemetry_loop(quickwit_telemetry_info(&node_config));
         quickwit_telemetry::send_telemetry_event(TelemetryEvent::RunCommand).await;
         // TODO move in serve quickwit?
         let runtimes_config = RuntimesConfig::default();
-        start_actor_runtimes(runtimes_config, &config.enabled_services)?;
+        start_actor_runtimes(runtimes_config, &node_config.enabled_services)?;
         let shutdown_signal = Box::pin(async move {
             signal::ctrl_c()
                 .await
                 .expect("Registering a signal handler for SIGINT should not fail.");
         });
         let serve_result = serve_quickwit(
-            config,
+            node_config,
             runtimes_config,
-            storage_resolver,
             metastore_resolver,
+            storage_resolver,
             shutdown_signal,
         )
         .await;

diff --git a/quickwit/quickwit-cli/src/tool.rs b/quickwit/quickwit-cli/src/tool.rs
@@ -47,6 +47,7 @@ use quickwit_indexing::models::{
     DetachIndexingPipeline, DetachMergePipeline, IndexingStatistics, SpawnPipeline,
 };
 use quickwit_indexing::IndexingPipeline;
+use quickwit_ingest::IngesterPool;
 use quickwit_metastore::Metastore;
 use quickwit_proto::search::SearchResponse;
 use quickwit_search::{single_node_search, SearchResponseRest};
@@ -457,6 +458,7 @@ pub async fn local_ingest_docs_cli(args: LocalIngestDocsArgs) -> anyhow::Result<
         cluster,
         metastore,
         None,
+        IngesterPool::default(),
         storage_resolver,
     )
     .await?;
@@ -586,6 +588,7 @@ pub async fn merge_cli(args: MergeArgs) -> anyhow::Result<()> {
         cluster,
         metastore,
         None,
+        IngesterPool::default(),
         storage_resolver,
     )
     .await?;

diff --git a/quickwit/quickwit-config/src/source_config/mod.rs b/quickwit/quickwit-config/src/source_config/mod.rs
@@ -45,6 +45,9 @@ pub const INGEST_API_SOURCE_ID: &str = "_ingest-api-source";
 /// Reserved source ID used for native Quickwit ingest.
 pub const INGEST_SOURCE_ID: &str = "_ingest-source";
 
+pub const RESERVED_SOURCE_IDS: &[&str] =
+    &[CLI_INGEST_SOURCE_ID, INGEST_API_SOURCE_ID, INGEST_SOURCE_ID];
+
 #[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
 #[serde(into = "VersionedSourceConfig")]
 #[serde(try_from = "VersionedSourceConfig")]
@@ -122,8 +125,8 @@ impl SourceConfig {
     pub fn ingest_default() -> Self {
         Self {
             source_id: INGEST_SOURCE_ID.to_string(),
-            max_num_pipelines_per_indexer: NonZeroUsize::new(1).unwrap(),
-            desired_num_pipelines: NonZeroUsize::new(1).unwrap(),
+            max_num_pipelines_per_indexer: NonZeroUsize::new(1).expect("1 should be non-zero"),
+            desired_num_pipelines: NonZeroUsize::new(1).expect("1 should be non-zero"),
             enabled: false,
             source_params: SourceParams::Ingest,
             transform_config: None,
@@ -135,8 +138,8 @@ impl SourceConfig {
     pub fn ingest_api_default() -> Self {
         Self {
             source_id: INGEST_API_SOURCE_ID.to_string(),
-            max_num_pipelines_per_indexer: NonZeroUsize::new(1).unwrap(),
-            desired_num_pipelines: NonZeroUsize::new(1).unwrap(),
+            max_num_pipelines_per_indexer: NonZeroUsize::new(1).expect("1 should be non-zero"),
+            desired_num_pipelines: NonZeroUsize::new(1).expect("1 should be non-zero"),
             enabled: true,
             source_params: SourceParams::IngestApi,
             transform_config: None,
@@ -148,8 +151,8 @@ impl SourceConfig {
     pub fn cli_ingest_source() -> Self {
         Self {
             source_id: CLI_INGEST_SOURCE_ID.to_string(),
-            max_num_pipelines_per_indexer: NonZeroUsize::new(1).unwrap(),
-            desired_num_pipelines: NonZeroUsize::new(1).unwrap(),
+            max_num_pipelines_per_indexer: NonZeroUsize::new(1).expect("1 should be non-zero"),
+            desired_num_pipelines: NonZeroUsize::new(1).expect("1 should be non-zero"),
             enabled: true,
             source_params: SourceParams::IngestCli,
             transform_config: None,
@@ -161,8 +164,8 @@ impl SourceConfig {
     pub fn for_test(source_id: &str, source_params: SourceParams) -> Self {
         Self {
             source_id: source_id.to_string(),
-            max_num_pipelines_per_indexer: NonZeroUsize::new(1).unwrap(),
-            desired_num_pipelines: NonZeroUsize::new(1).unwrap(),
+            max_num_pipelines_per_indexer: NonZeroUsize::new(1).expect("1 should be non-zero"),
+            desired_num_pipelines: NonZeroUsize::new(1).expect("1 should be non-zero"),
             enabled: true,
             source_params,
             transform_config: None,
@@ -658,8 +661,8 @@ mod tests {
             load_source_config_from_user_config(config_format, file_content.as_bytes()).unwrap();
         let expected_source_config = SourceConfig {
             source_id: "hdfs-logs-kinesis-source".to_string(),
-            max_num_pipelines_per_indexer: NonZeroUsize::new(1).unwrap(),
-            desired_num_pipelines: NonZeroUsize::new(1).unwrap(),
+            max_num_pipelines_per_indexer: NonZeroUsize::new(1).expect("1 should be non-zero"),
+            desired_num_pipelines: NonZeroUsize::new(1).expect("1 should be non-zero"),
             enabled: true,
             source_params: SourceParams::Kinesis(KinesisSourceParams {
                 stream_name: "emr-cluster-logs".to_string(),
@@ -1066,8 +1069,8 @@ mod tests {
         let source_config: SourceConfig = ConfigFormat::Json.parse(&file_content).unwrap();
         let expected_source_config = SourceConfig {
             source_id: INGEST_API_SOURCE_ID.to_string(),
-            max_num_pipelines_per_indexer: NonZeroUsize::new(1).unwrap(),
-            desired_num_pipelines: NonZeroUsize::new(1).unwrap(),
+            max_num_pipelines_per_indexer: NonZeroUsize::new(1).expect("1 should be non-zero"),
+            desired_num_pipelines: NonZeroUsize::new(1).expect("1 should be non-zero"),
             enabled: true,
             source_params: SourceParams::IngestApi,
             transform_config: Some(TransformConfig {

diff --git a/quickwit/quickwit-config/src/source_config/serialize.rs b/quickwit/quickwit-config/src/source_config/serialize.rs
@@ -22,11 +22,8 @@ use std::num::NonZeroUsize;
 use anyhow::bail;
 use serde::{Deserialize, Serialize};
 
-use super::TransformConfig;
-use crate::{
-    validate_identifier, ConfigFormat, SourceConfig, SourceInputFormat, SourceParams,
-    CLI_INGEST_SOURCE_ID, INGEST_API_SOURCE_ID,
-};
+use super::{TransformConfig, RESERVED_SOURCE_IDS};
+use crate::{validate_identifier, ConfigFormat, SourceConfig, SourceInputFormat, SourceParams};
 
 type SourceConfigForSerialization = SourceConfigV0_6;
 
@@ -72,7 +69,7 @@ impl SourceConfigForSerialization {
     ///
     /// TODO refactor #1065
     fn validate_and_build(self) -> anyhow::Result<SourceConfig> {
-        if self.source_id != CLI_INGEST_SOURCE_ID && self.source_id != INGEST_API_SOURCE_ID {
+        if !RESERVED_SOURCE_IDS.contains(&self.source_id.as_str()) {
             validate_identifier("Source ID", &self.source_id)?;
         }
         let desired_num_pipelines = NonZeroUsize::new(self.desired_num_pipelines)