feat: support query opened shards info

cluster/src/cluster_impl.rs

@@ -298,6 +298,16 @@ impl Inner {
                 msg: format!("close non-existent shard, shard_id:{shard_id}"),
             })
     }
+
+    fn list_shards(&self) -> Vec<ShardInfo> {
+        let shards = self.shard_set.all_shards();
+        let mut shard_infos = Vec::with_capacity(shards.len());
+        for shard in shards {
+            let shard_info = shard.shard_info();
+            shard_infos.push(shard_info);
+        }
+        shard_infos
+    }
 }
 
 #[async_trait]
@@ -345,6 +355,10 @@ impl Cluster for ClusterImpl {
         self.inner.close_shard(shard_id)
     }
 
+    fn list_shards(&self) -> Vec<ShardInfo> {
+        self.inner.list_shards()
+    }
+
     async fn route_tables(&self, req: &RouteTablesRequest) -> Result<RouteTablesResponse> {
         self.inner.route_tables(req).await
     }

cluster/src/lib.rs

@@ -170,6 +170,9 @@ pub trait Cluster {
     /// Return error if the shard is not found.
     async fn close_shard(&self, shard_id: ShardId) -> Result<ShardRef>;
 
+    /// list shards
+    fn list_shards(&self) -> Vec<ShardInfo>;
+
     async fn route_tables(&self, req: &RouteTablesRequest) -> Result<RouteTablesResponse>;
     async fn fetch_nodes(&self) -> Result<ClusterNodesResp>;
     fn shard_lock_manager(&self) -> ShardLockManagerRef;

meta_client/src/types.rs

@@ -9,7 +9,7 @@ use common_types::{
     table::{TableId, TableName},
 };
 use common_util::{config::ReadableDuration, error::BoxError};
-use serde::Deserialize;
+use serde::{Deserialize, Serialize};
 use snafu::{OptionExt, ResultExt};
 use table_engine::partition::PartitionInfo;
 
@@ -161,7 +161,7 @@ pub struct NodeInfo {
     pub shard_infos: Vec<ShardInfo>,
 }
 
-#[derive(Debug, Default, Clone, PartialEq, Eq)]
+#[derive(Debug, Default, Clone, PartialEq, Eq, Serialize)]
 pub struct ShardInfo {
     pub id: ShardId,
     pub role: ShardRole,
@@ -175,7 +175,7 @@ impl ShardInfo {
     }
 }
 
-#[derive(Debug, Default, Copy, Clone, Eq, PartialEq)]
+#[derive(Debug, Default, Copy, Clone, Eq, PartialEq, Serialize)]
 pub enum ShardRole {
     #[default]
     Leader,

router/src/cluster_based.rs

@@ -198,6 +198,10 @@ mod tests {
             unimplemented!();
         }
 
+        fn list_shards(&self) -> Vec<ShardInfo> {
+            unimplemented!();
+        }
+
         async fn route_tables(
             &self,
             req: &RouteTablesRequest,

server/src/http.rs

@@ -8,6 +8,7 @@ use std::{
 };
 
 use analytic_engine::setup::OpenedWals;
+use cluster::ClusterRef;
 use common_types::bytes::Bytes;
 use common_util::{
     error::{BoxError, GenericError},
@@ -118,6 +119,9 @@ pub enum Error {
 
     #[snafu(display("{msg}"))]
     QueryMaybeExceedTTL { msg: String },
+
+    #[snafu(display("query shards only supported in cluster mode"))]
+    NoCluster {},
 }
 
 define_result!(Error);
@@ -129,6 +133,7 @@ impl reject::Reject for Error {}
 /// Endpoints beginning with /debug are for internal use, and may subject to
 /// breaking changes.
 pub struct Service<Q> {
+    cluster: Option<ClusterRef>,
     proxy: Arc<Proxy<Q>>,
     engine_runtimes: Arc<EngineRuntimes>,
     log_runtime: Arc<RuntimeLevel>,
@@ -198,6 +203,7 @@ impl<Q: QueryExecutor + 'static> Service<Q> {
             .or(self.profile_cpu())
             .or(self.profile_heap())
             .or(self.server_config())
+            .or(self.shards())
             .or(self.stats())
             .with(warp::log("http_requests"))
             .with(warp::log::custom(|info| {
@@ -492,6 +498,23 @@ impl<Q: QueryExecutor + 'static> Service<Q> {
             .map(move || server_config_content.clone())
     }
 
+    // GET /debug/shards
+    fn shards(
+        &self,
+    ) -> impl Filter<Extract = (impl warp::Reply,), Error = warp::Rejection> + Clone {
+        warp::path!("debug" / "shards")
+            .and(warp::get())
+            .and(self.with_cluster())
+            .and_then(|cluster: Option<ClusterRef>| async move {
+                let cluster = match cluster {
+                    Some(cluster) => cluster,
+                    None => return Err(reject::custom(Error::NoCluster {})),
+                };
+                let shard_infos = cluster.list_shards();
+                Ok(reply::json(&shard_infos))
+            })
+    }
+
     // GET /debug/stats
     fn stats(&self) -> impl Filter<Extract = (impl warp::Reply,), Error = warp::Rejection> + Clone {
         let opened_wals = self.opened_wals.clone();
@@ -605,6 +628,13 @@ impl<Q: QueryExecutor + 'static> Service<Q> {
         warp::any().map(move || proxy.clone())
     }
 
+    fn with_cluster(
+        &self,
+    ) -> impl Filter<Extract = (Option<ClusterRef>,), Error = Infallible> + Clone {
+        let cluster = self.cluster.clone();
+        warp::any().map(move || cluster.clone())
+    }
+
     fn with_runtime(&self) -> impl Filter<Extract = (Arc<Runtime>,), Error = Infallible> + Clone {
         let runtime = self.engine_runtimes.default_runtime.clone();
         warp::any().map(move || runtime.clone())
@@ -631,6 +661,7 @@ pub struct Builder<Q> {
     engine_runtimes: Option<Arc<EngineRuntimes>>,
     log_runtime: Option<Arc<RuntimeLevel>>,
     config_content: Option<String>,
+    cluster: Option<ClusterRef>,
     proxy: Option<Arc<Proxy<Q>>>,
     opened_wals: Option<OpenedWals>,
 }
@@ -642,6 +673,7 @@ impl<Q> Builder<Q> {
             engine_runtimes: None,
             log_runtime: None,
             config_content: None,
+            cluster: None,
             proxy: None,
             opened_wals: None,
         }
@@ -662,6 +694,11 @@ impl<Q> Builder<Q> {
         self
     }
 
+    pub fn cluster(mut self, cluster: Option<ClusterRef>) -> Self {
+        self.cluster = cluster;
+        self
+    }
+
     pub fn proxy(mut self, proxy: Arc<Proxy<Q>>) -> Self {
         self.proxy = Some(proxy);
         self
@@ -680,11 +717,13 @@ impl<Q: QueryExecutor + 'static> Builder<Q> {
         let log_runtime = self.log_runtime.context(MissingLogRuntime)?;
         let config_content = self.config_content.context(MissingInstance)?;
         let proxy = self.proxy.context(MissingProxy)?;
+        let cluster = self.cluster;
         let opened_wals = self.opened_wals.context(MissingWal)?;
 
         let (tx, rx) = oneshot::channel();
 
         let service = Service {
+            cluster,
             proxy,
             engine_runtimes,
             log_runtime,
@@ -732,6 +771,7 @@ fn error_to_status_code(err: &Error) -> StatusCode {
         | Error::AlreadyStarted { .. }
         | Error::MissingRouter { .. }
         | Error::MissingWal { .. }
+        | Error::NoCluster { .. }
         | Error::HandleUpdateLogLevel { .. } => StatusCode::INTERNAL_SERVER_ERROR,
         Error::QueryMaybeExceedTTL { .. } => StatusCode::OK,
     }

server/src/server.rs

@@ -371,6 +371,7 @@ impl<Q: QueryExecutor + 'static> Builder<Q> {
             .engine_runtimes(engine_runtimes.clone())
             .log_runtime(log_runtime)
             .config_content(config_content)
+            .cluster(self.cluster.clone())
             .proxy(proxy.clone())
             .opened_wals(opened_wals.clone())
             .build()