Add retry logic for router (#4027)

* Add retry logic for router * Fix various issues
quickwit-oss · Oct 27, 2023 · 742347b · 742347b
1 parent 572d6d5
commit 742347b
Show file tree

Hide file tree

Showing 29 changed files with 1,177 additions and 418 deletions.
diff --git a/quickwit/Cargo.lock b/quickwit/Cargo.lock
diff --git a/quickwit/quickwit-control-plane/src/control_plane.rs b/quickwit/quickwit-control-plane/src/control_plane.rs
@@ -180,11 +180,12 @@ fn convert_metastore_error<T>(
         | MetastoreError::JsonDeserializeError { .. }
         | MetastoreError::JsonSerializeError { .. }
         | MetastoreError::NotFound(_) => true,
-        MetastoreError::Unavailable(_)
+        MetastoreError::Connection { .. }
+        | MetastoreError::Db { .. }
+        | MetastoreError::InconsistentControlPlaneState { .. }
         | MetastoreError::Internal { .. }
         | MetastoreError::Io { .. }
-        | MetastoreError::Connection { .. }
-        | MetastoreError::Db { .. } => false,
+        | MetastoreError::Unavailable(_) => false,
     };
     if is_transaction_certainly_aborted {
         // If the metastore transaction is certain to have been aborted,
@@ -363,10 +364,22 @@ impl Handler<GetOrCreateOpenShardsRequest> for ControlPlane {
         request: GetOrCreateOpenShardsRequest,
         ctx: &ActorContext<Self>,
     ) -> Result<Self::Reply, ActorExitStatus> {
-        Ok(self
+        let response = match self
             .ingest_controller
             .get_or_create_open_shards(request, &mut self.model, ctx.progress())
-            .await)
+            .await
+        {
+            Ok(response) => response,
+            Err(ControlPlaneError::Metastore(metastore_error)) => {
+                return convert_metastore_error(metastore_error);
+            }
+            Err(control_plane_error) => {
+                return Ok(Err(control_plane_error));
+            }
+        };
+        // TODO: Why do we return an error if the indexing scheduler fails?
+        self.indexing_scheduler.on_index_change(&self.model).await?;
+        Ok(Ok(response))
     }
 }
 
@@ -379,7 +392,7 @@ mod tests {
         ListIndexesMetadataResponseExt,
     };
     use quickwit_proto::control_plane::GetOrCreateOpenShardsSubrequest;
-    use quickwit_proto::ingest::Shard;
+    use quickwit_proto::ingest::{Shard, ShardState};
     use quickwit_proto::metastore::{
         EntityKind, ListIndexesMetadataRequest, ListIndexesMetadataResponse, ListShardsRequest,
         ListShardsResponse, ListShardsSubresponse, MetastoreError, SourceType,
@@ -641,7 +654,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn test_control_plane_get_open_shards() {
+    async fn test_control_plane_get_or_create_open_shards() {
         let universe = Universe::with_accelerated_time();
 
         let cluster_id = "test-cluster".to_string();
@@ -675,6 +688,7 @@ mod tests {
                     index_uid: "test-index:0".to_string(),
                     source_id: INGEST_SOURCE_ID.to_string(),
                     shard_id: 1,
+                    shard_state: ShardState::Open as i32,
                     ..Default::default()
                 }],
                 next_shard_id: 2,
@@ -716,11 +730,6 @@ mod tests {
         universe.assert_quit().await;
     }
 
-    #[tokio::test]
-    async fn test_control_plane_close_shards() {
-        // TODO: Write test when the RPC is actually called by ingesters.
-    }
-
     #[tokio::test]
     async fn test_control_plane_supervision_reload_from_metastore() {
         let universe = Universe::default();