add api endpoint to get some control-plan internal info (#4339)

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

@@ -32,8 +32,9 @@ use quickwit_config::SourceConfig;
 use quickwit_ingest::{IngesterPool, LocalShardsUpdate};
 use quickwit_metastore::IndexMetadata;
 use quickwit_proto::control_plane::{
-    ControlPlaneError, ControlPlaneResult, GetOrCreateOpenShardsRequest,
-    GetOrCreateOpenShardsResponse,
+    ControlPlaneError, ControlPlaneResult, GetDebugStateRequest, GetDebugStateResponse,
+    GetOrCreateOpenShardsRequest, GetOrCreateOpenShardsResponse, PhysicalIndexingPlanEntry,
+    ShardTableEntry,
 };
 use quickwit_proto::indexing::ShardPositionsUpdate;
 use quickwit_proto::metastore::{
@@ -179,6 +180,37 @@ impl ControlPlane {
             .schedule_indexing_plan_if_needed(&self.model);
         Ok(())
     }
+
+    fn debug_state(&self) -> GetDebugStateResponse {
+        let shard_table = self
+            .model
+            .all_shards_with_source()
+            .map(|(source, shards)| ShardTableEntry {
+                source_id: source.to_string(),
+                shards: shards
+                    .map(|shard_entry| shard_entry.shard.clone())
+                    .collect(),
+            })
+            .collect();
+        let physical_index_plan = self
+            .indexing_scheduler
+            .observable_state()
+            .last_applied_physical_plan
+            .map(|plan| {
+                plan.indexing_tasks_per_indexer()
+                    .iter()
+                    .map(|(node_id, tasks)| PhysicalIndexingPlanEntry {
+                        node_id: node_id.clone(),
+                        tasks: tasks.clone(),
+                    })
+                    .collect()
+            })
+            .unwrap_or_default();
+        GetDebugStateResponse {
+            shard_table,
+            physical_index_plan,
+        }
+    }
 }
 
 #[async_trait]
@@ -528,6 +560,19 @@ impl Handler<LocalShardsUpdate> for ControlPlane {
     }
 }
 
+#[async_trait]
+impl Handler<GetDebugStateRequest> for ControlPlane {
+    type Reply = ControlPlaneResult<GetDebugStateResponse>;
+
+    async fn handle(
+        &mut self,
+        _: GetDebugStateRequest,
+        _ctx: &ActorContext<Self>,
+    ) -> Result<Self::Reply, ActorExitStatus> {
+        Ok(Ok(self.debug_state()))
+    }
+}
+
 #[derive(Clone)]
 pub struct ControlPlaneEventSubscriber(WeakMailbox<ControlPlane>);
 

quickwit/quickwit-control-plane/src/model/mod.rs

@@ -248,6 +248,12 @@ impl ControlPlaneModel {
         self.shard_table.all_shards()
     }
 
+    pub(crate) fn all_shards_with_source(
+        &self,
+    ) -> impl Iterator<Item = (&SourceUid, impl Iterator<Item = &ShardEntry>)> + '_ {
+        self.shard_table.all_shards_with_source()
+    }
+
     pub fn list_shards_for_node(
         &self,
         ingester: &NodeId,

quickwit/quickwit-control-plane/src/model/shard_table.rs

@@ -284,6 +284,14 @@ impl ShardTable {
             .flat_map(|table_entry| table_entry.shard_entries.values())
     }
 
+    pub(crate) fn all_shards_with_source(
+        &self,
+    ) -> impl Iterator<Item = (&SourceUid, impl Iterator<Item = &ShardEntry>)> + '_ {
+        self.table_entries
+            .iter()
+            .map(|(source, shard_table)| (source, shard_table.shard_entries.values()))
+    }
+
     pub(crate) fn all_shards_mut(&mut self) -> impl Iterator<Item = &mut ShardEntry> + '_ {
         self.table_entries
             .values_mut()

quickwit/quickwit-proto/protos/quickwit/control_plane.proto

@@ -21,6 +21,7 @@ syntax = "proto3";
 
 package quickwit.control_plane;
 
+import "quickwit/indexing.proto";
 import "quickwit/ingest.proto";
 import "quickwit/metastore.proto";
 
@@ -59,6 +60,9 @@ service ControlPlaneService {
   // Returns the list of open shards for one or several sources. If the control plane is not able to find any
   // for a source, it will pick a pair of leader-follower ingesters and will open a new shard.
   rpc GetOrCreateOpenShards(GetOrCreateOpenShardsRequest) returns (GetOrCreateOpenShardsResponse);
+
+  // Return some innerstate of the control plane meant to assist debugging.
+  rpc GetDebugState(GetDebugStateRequest) returns (GetDebugStateResponse);
 }
 
 // Shard API
@@ -103,3 +107,21 @@ message GetOrCreateOpenShardsFailure {
   string source_id = 3;
   GetOrCreateOpenShardsFailureReason reason = 4;
 }
+
+message GetDebugStateRequest {
+}
+
+message GetDebugStateResponse {
+  repeated ShardTableEntry shard_table = 1;
+  repeated PhysicalIndexingPlanEntry physical_index_plan = 2;
+}
+
+message ShardTableEntry {
+  string source_id = 1;
+  repeated quickwit.ingest.Shard shards = 2;
+}
+
+message PhysicalIndexingPlanEntry {
+  string node_id = 1;
+  repeated quickwit.indexing.IndexingTask tasks = 2;
+}