Replaces indexing service with codegen version (#3675)

Replaces hand-crafted indexing service with auto generated version.

See #3108

Co-authored-by: Adrien Guillo <[email protected]>

Author: imotov

quickwit/Cargo.lock

@@ -24,7 +24,6 @@ serde_json = { workspace = true }
 sync_wrapper = { workspace = true }
 
 quickwit-common = { workspace = true }
-quickwit-proto = { workspace = true }
 
 [features]
 testsuite = []

quickwit/quickwit-actors/Cargo.toml

@@ -31,7 +31,6 @@ use std::fmt;
 use std::num::NonZeroU64;
 
 use once_cell::sync::Lazy;
-use quickwit_proto::{ServiceError, ServiceErrorCode};
 use tokio::time::Duration;
 mod actor;
 mod actor_context;
@@ -129,13 +128,3 @@ pub enum AskError<E: fmt::Debug> {
     #[error("The handler returned an error: `{0:?}`.")]
     ErrorReply(#[from] E),
 }
-
-impl<E: fmt::Debug + ServiceError> ServiceError for AskError<E> {
-    fn status_code(&self) -> ServiceErrorCode {
-        match self {
-            AskError::MessageNotDelivered => ServiceErrorCode::Internal,
-            AskError::ProcessMessageError => ServiceErrorCode::Internal,
-            AskError::ErrorReply(err) => err.status_code(),
-        }
-    }
-}

quickwit/quickwit-actors/src/lib.rs

@@ -313,8 +313,8 @@ pub(crate) fn build_indexing_plan(
 #[cfg(test)]
 mod tests {
     use std::collections::HashMap;
-    use std::net::SocketAddr;
     use std::num::NonZeroUsize;
+    use std::time::Duration;
 
     use itertools::Itertools;
     use proptest::prelude::*;
@@ -324,11 +324,12 @@ mod tests {
         FileSourceParams, KafkaSourceParams, SourceConfig, SourceInputFormat, SourceParams,
         CLI_INGEST_SOURCE_ID, INGEST_API_SOURCE_ID,
     };
-    use quickwit_indexing::indexing_client::create_indexing_service_client;
     use quickwit_proto::indexing::IndexingTask;
-    use quickwit_proto::IndexUid;
+    use quickwit_proto::{IndexUid, IndexingServiceClient};
     use rand::seq::SliceRandom;
     use serde_json::json;
+    use tonic::transport::Endpoint;
+    use tower::timeout::Timeout;
 
     use super::{build_physical_indexing_plan, IndexSourceId};
     use crate::indexing_plan::build_indexing_plan;
@@ -351,8 +352,11 @@ mod tests {
     ) -> Vec<(String, IndexerNodeInfo)> {
         let mut members = Vec::new();
         for idx in 0..num_members {
-            let addr: SocketAddr = ([127, 0, 0, 1], 10).into();
-            let client = create_indexing_service_client(addr).await.unwrap();
+            let channel = Timeout::new(
+                Endpoint::from_static("http://127.0.0.1:10").connect_lazy(),
+                Duration::from_secs(1),
+            );
+            let client = IndexingServiceClient::from_channel(channel);
             members.push((
                 (1 + idx).to_string(),
                 IndexerNodeInfo {

quickwit/quickwit-control-plane/src/indexing_plan.rs

@@ -31,9 +31,8 @@ use quickwit_actors::{AskError, Mailbox, Universe};
 use quickwit_common::pubsub::EventSubscriber;
 use quickwit_common::tower::Pool;
 use quickwit_config::SourceParams;
-use quickwit_indexing::indexing_client::IndexingServiceClient;
 use quickwit_metastore::{Metastore, MetastoreEvent};
-use quickwit_proto::indexing::IndexingTask;
+use quickwit_proto::indexing::{IndexingServiceClient, IndexingTask};
 use scheduler::IndexingScheduler;
 use tracing::error;
 

quickwit/quickwit-control-plane/src/lib.rs

@@ -30,6 +30,7 @@ use quickwit_actors::{Actor, ActorContext, ActorExitStatus, Handler};
 use quickwit_config::SourceConfig;
 use quickwit_metastore::Metastore;
 use quickwit_proto::indexing::{ApplyIndexingPlanRequest, IndexingTask};
+use quickwit_proto::IndexingService;
 use serde::Serialize;
 use tracing::{debug, error, info, warn};
 
@@ -268,20 +269,26 @@ impl IndexingScheduler {
     ) {
         debug!("Apply physical indexing plan: {:?}", new_physical_plan);
         for (node_id, indexing_tasks) in new_physical_plan.indexing_tasks_per_node() {
-            let indexer = indexers
-                .iter_mut()
-                .find(|indexer| &indexer.0 == node_id)
-                .expect("This should never happen as the plan was built from these indexers.");
-            if let Err(error) = indexer
-                .1
-                .client
-                .apply_indexing_plan(ApplyIndexingPlanRequest {
-                    indexing_tasks: indexing_tasks.clone(),
-                })
-                .await
-            {
-                error!(indexer_node_id=%indexer.0, err=?error, "Error occurred when appling indexing plan to indexer.");
-            }
+            // We don't want to block on a slow indexer so we apply this change asynchronously
+            tokio::spawn({
+                let indexer = indexers
+                    .iter()
+                    .find(|indexer| &indexer.0 == node_id)
+                    .expect("This should never happen as the plan was built from these indexers.")
+                    .clone();
+                let indexing_tasks = indexing_tasks.clone();
+                async move {
+                    if let Err(error) = indexer
+                        .1
+                        .client
+                        .clone()
+                        .apply_indexing_plan(ApplyIndexingPlanRequest { indexing_tasks })
+                        .await
+                    {
+                        error!(indexer_node_id=%indexer.0, err=?error, "Error occurred when applying indexing plan to indexer.");
+                    }
+                }
+            });
         }
         self.state.num_applied_physical_indexing_plan += 1;
         self.state.last_applied_plan_timestamp = Some(Instant::now());
@@ -522,10 +529,10 @@ mod tests {
     use quickwit_common::tower::{Change, Pool};
     use quickwit_config::service::QuickwitService;
     use quickwit_config::{KafkaSourceParams, SourceConfig, SourceInputFormat, SourceParams};
-    use quickwit_indexing::indexing_client::IndexingServiceClient;
     use quickwit_indexing::IndexingService;
     use quickwit_metastore::{IndexMetadata, MockMetastore};
     use quickwit_proto::indexing::{ApplyIndexingPlanRequest, IndexingTask};
+    use quickwit_proto::IndexingServiceClient;
     use serde_json::json;
 
     use super::{IndexingScheduler, CONTROL_PLAN_LOOP_INTERVAL};
@@ -576,10 +583,9 @@ mod tests {
                         if node.enabled_services().contains(&QuickwitService::Indexer) =>
                     {
                         let node_id = node.node_id().to_string();
-                        let grpc_addr = node.grpc_advertise_addr();
                         let indexing_tasks = node.indexing_tasks().to_vec();
                         let client_mailbox = indexing_clients.get(&node_id).unwrap().clone();
-                        let client = IndexingServiceClient::from_service(client_mailbox, grpc_addr);
+                        let client = IndexingServiceClient::from_mailbox(client_mailbox);
                         Some(Change::Insert(
                             node_id,
                             IndexerNodeInfo {

quickwit/quickwit-control-plane/src/scheduler.rs

@@ -47,9 +47,10 @@ use quickwit_common::split_file;
 use quickwit_common::temp_dir::TempDirectory;
 use quickwit_indexing::actors::MergeExecutor;
 use quickwit_indexing::merge_policy::MergeOperation;
-use quickwit_indexing::models::{IndexingPipelineId, MergeScratch};
+use quickwit_indexing::models::MergeScratch;
 use quickwit_indexing::{get_tantivy_directory_from_split_bundle, TestSandbox};
 use quickwit_metastore::{ListSplitsQuery, Split, SplitMetadata, SplitState};
+use quickwit_proto::indexing_api::IndexingPipelineId;
 use quickwit_proto::IndexUid;
 use serde_json::Value as JsonValue;
 use tantivy::{Directory, Inventory};

quickwit/quickwit-indexing/failpoints/mod.rs

@@ -39,6 +39,7 @@ use quickwit_config::IndexingSettings;
 use quickwit_doc_mapper::DocMapper;
 use quickwit_metastore::checkpoint::{IndexCheckpointDelta, SourceCheckpointDelta};
 use quickwit_metastore::Metastore;
+use quickwit_proto::indexing_api::IndexingPipelineId;
 use quickwit_query::get_quickwit_fastfield_normalizer_manager;
 use serde::Serialize;
 use tantivy::schema::Schema;
@@ -52,8 +53,8 @@ use ulid::Ulid;
 
 use crate::actors::IndexSerializer;
 use crate::models::{
-    CommitTrigger, EmptySplit, IndexedSplitBatchBuilder, IndexedSplitBuilder, IndexingPipelineId,
-    NewPublishLock, ProcessedDoc, ProcessedDocBatch, PublishLock,
+    CommitTrigger, EmptySplit, IndexedSplitBatchBuilder, IndexedSplitBuilder, NewPublishLock,
+    ProcessedDoc, ProcessedDocBatch, PublishLock,
 };
 
 // Random partition id used to gather partitions exceeding the maximum number of partitions.

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

@@ -31,6 +31,7 @@ use quickwit_common::KillSwitch;
 use quickwit_config::{IndexingSettings, SourceConfig};
 use quickwit_doc_mapper::DocMapper;
 use quickwit_metastore::{Metastore, MetastoreError};
+use quickwit_proto::indexing_api::IndexingPipelineId;
 use quickwit_storage::Storage;
 use tokio::join;
 use tokio::sync::Semaphore;
@@ -44,7 +45,7 @@ use crate::actors::sequencer::Sequencer;
 use crate::actors::uploader::UploaderType;
 use crate::actors::{Indexer, Packager, Publisher, Uploader};
 use crate::merge_policy::MergePolicy;
-use crate::models::{IndexingPipelineId, IndexingStatistics, Observe};
+use crate::models::{IndexingStatistics, Observe};
 use crate::source::{quickwit_supported_sources, SourceActor, SourceExecutionContext};
 use crate::split_store::IndexingSplitStore;
 use crate::SplitsUpdateMailbox;

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

@@ -18,6 +18,7 @@
 // along with this program. If not, see <http://www.gnu.org/licenses/>.
 
 use std::collections::{HashMap, HashSet};
+use std::fmt::Debug;
 use std::path::PathBuf;
 use std::sync::Arc;
 
@@ -36,66 +37,26 @@ use quickwit_config::{
     build_doc_mapper, IndexConfig, IndexerConfig, SourceConfig, INGEST_API_SOURCE_ID,
 };
 use quickwit_ingest::{DropQueueRequest, IngestApiService, ListQueuesRequest, QUEUES_DIR_NAME};
-use quickwit_metastore::{IndexMetadata, Metastore, MetastoreError};
-use quickwit_proto::indexing::{ApplyIndexingPlanRequest, IndexingTask};
-use quickwit_proto::{IndexUid, ServiceError, ServiceErrorCode};
-use quickwit_storage::{StorageError, StorageResolver, StorageResolverError};
+use quickwit_metastore::{IndexMetadata, Metastore};
+use quickwit_proto::indexing::{ApplyIndexingPlanRequest, ApplyIndexingPlanResponse, IndexingTask};
+use quickwit_proto::indexing_api::{IndexingPipelineId, IndexingServiceError};
+use quickwit_proto::IndexUid;
+use quickwit_storage::StorageResolver;
 use serde::{Deserialize, Serialize};
-use thiserror::Error;
 use tokio::sync::Semaphore;
 use tracing::{debug, error, info, warn};
 
 use super::merge_pipeline::{MergePipeline, MergePipelineParams};
 use super::MergePlanner;
 use crate::models::{
-    DetachIndexingPipeline, DetachMergePipeline, IndexingPipelineId, Observe, ObservePipeline,
-    SpawnPipeline,
+    DetachIndexingPipeline, DetachMergePipeline, Observe, ObservePipeline, SpawnPipeline,
 };
 use crate::split_store::{LocalSplitStore, SplitStoreQuota};
 use crate::{IndexingPipeline, IndexingPipelineParams, IndexingSplitStore, IndexingStatistics};
 
 /// Name of the indexing directory, usually located at `<data_dir_path>/indexing`.
 pub const INDEXING_DIR_NAME: &str = "indexing";
 
-#[derive(Error, Debug)]
-pub enum IndexingServiceError {
-    #[error("Indexing pipeline `{index_id}` for source `{source_id}` does not exist.")]
-    MissingPipeline { index_id: String, source_id: String },
-    #[error(
-        "Pipeline #{pipeline_ord} for index `{index_id}` and source `{source_id}` already exists."
-    )]
-    PipelineAlreadyExists {
-        index_id: String,
-        source_id: String,
-        pipeline_ord: usize,
-    },
-    #[error("Failed to resolve the storage `{0}`.")]
-    StorageResolverError(#[from] StorageResolverError),
-    #[error("Storage error `{0}`.")]
-    StorageError(#[from] StorageError),
-    #[error("Metastore error `{0}`.")]
-    MetastoreError(#[from] MetastoreError),
-    #[error("Invalid params `{0}`.")]
-    InvalidParams(anyhow::Error),
-    #[error("Spanw pipelines errors `{pipeline_ids:?}`.")]
-    SpawnPipelinesError {
-        pipeline_ids: Vec<IndexingPipelineId>,
-    },
-}
-
-impl ServiceError for IndexingServiceError {
-    fn status_code(&self) -> ServiceErrorCode {
-        match self {
-            Self::MissingPipeline { .. } => ServiceErrorCode::NotFound,
-            Self::PipelineAlreadyExists { .. } => ServiceErrorCode::BadRequest,
-            Self::StorageResolverError(_) | Self::StorageError(_) => ServiceErrorCode::Internal,
-            Self::MetastoreError(_) => ServiceErrorCode::Internal,
-            Self::InvalidParams(_) => ServiceErrorCode::BadRequest,
-            Self::SpawnPipelinesError { .. } => ServiceErrorCode::Internal,
-        }
-    }
-}
-
 #[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
 pub struct IndexingServiceCounters {
     pub num_running_pipelines: usize,
@@ -142,6 +103,17 @@ pub struct IndexingService {
     cooperative_indexing_permits: Option<Arc<Semaphore>>,
 }
 
+impl Debug for IndexingService {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        formatter
+            .debug_struct("IndexingService")
+            .field("cluster_id", &self.cluster.cluster_id())
+            .field("self_node_id", &self.node_id)
+            .field("indexing_root_directory", &self.indexing_root_directory)
+            .finish()
+    }
+}
+
 impl IndexingService {
     #[allow(clippy::too_many_arguments)]
     pub async fn new(
@@ -271,11 +243,12 @@ impl IndexingService {
             .join(&pipeline_id.source_id)
             .join(&pipeline_id.pipeline_ord.to_string())
             .tempdir_in(&self.indexing_root_directory)
-            .map_err(|error| IndexingServiceError::StorageError(error.into()))?;
+            .map_err(IndexingServiceError::Io)?;
         let storage = self
             .storage_resolver
             .resolve(&index_config.index_uri)
-            .await?;
+            .await
+            .map_err(|err| IndexingServiceError::StorageResolverError(err.to_string()))?;
         let merge_policy =
             crate::merge_policy::merge_policy_from_settings(&index_config.indexing_settings);
         let split_store = IndexingSplitStore::new(storage.clone(), self.local_split_store.clone());
@@ -336,7 +309,11 @@ impl IndexingService {
         index_id: &str,
     ) -> Result<IndexMetadata, IndexingServiceError> {
         let _protect_guard = ctx.protect_zone();
-        let index_metadata = self.metastore.index_metadata(index_id).await?;
+        let index_metadata = self
+            .metastore
+            .index_metadata(index_id)
+            .await
+            .map_err(|err| IndexingServiceError::MetastoreError(err.to_string()))?;
         Ok(index_metadata)
     }
 
@@ -440,7 +417,7 @@ impl IndexingService {
         &mut self,
         ctx: &ActorContext<Self>,
         physical_indexing_plan_request: ApplyIndexingPlanRequest,
-    ) -> Result<(), IndexingServiceError> {
+    ) -> Result<ApplyIndexingPlanResponse, IndexingServiceError> {
         let mut updated_pipeline_ids: HashSet<IndexingPipelineId> = HashSet::new();
         let mut pipeline_ordinals: HashMap<&IndexingTask, usize> = HashMap::new();
         for indexing_task in physical_indexing_plan_request.indexing_tasks.iter() {
@@ -484,7 +461,7 @@ impl IndexingService {
             });
         }
 
-        Ok(())
+        Ok(ApplyIndexingPlanResponse {})
     }
 
     /// Spawns the pipelines with supplied ids and returns a list of failed pipelines.
@@ -757,7 +734,7 @@ impl Handler<Observe> for IndexingService {
 
 #[async_trait]
 impl Handler<ApplyIndexingPlanRequest> for IndexingService {
-    type Reply = Result<(), IndexingServiceError>;
+    type Reply = Result<ApplyIndexingPlanResponse, IndexingServiceError>;
 
     async fn handle(
         &mut self,