refactor(cluster): refactor cluster for dyn cluster [part 1] (#17023)

* refactor(cluster): refactor cluster for dyn cluster [part 1]

* refactor(cluster): refactor cluster for dyn cluster

* refactor(cluster): refactor cluster for dyn cluster

* refactor(cluster): refactor cluster for dyn cluster

* refactor(cluster): refactor cluster for dyn cluster

* refactor(cluster): refactor cluster for dyn cluster

src/meta/types/src/cluster.rs

@@ -82,6 +82,11 @@ pub struct NodeInfo {
     pub flight_address: String,
     pub discovery_address: String,
     pub binary_version: String,
+
+    #[serde(skip_serializing_if = "String::is_empty")]
+    pub cluster_id: String,
+    #[serde(skip_serializing_if = "String::is_empty")]
+    pub warehouse_id: String,
 }
 
 impl NodeInfo {
@@ -103,6 +108,8 @@ impl NodeInfo {
             flight_address,
             discovery_address,
             binary_version,
+            cluster_id: "".to_string(),
+            warehouse_id: "".to_string(),
         }
     }
 

src/meta/types/tests/it/cluster.rs

@@ -25,6 +25,8 @@ fn test_node_info_ip_port() -> anyhow::Result<()> {
         flight_address: "1.2.3.4:123".to_string(),
         discovery_address: "4.5.6.7:456".to_string(),
         binary_version: "v0.8-binary-version".to_string(),
+        cluster_id: "".to_string(),
+        warehouse_id: "".to_string(),
     };
 
     let (ip, port) = n.ip_port()?;
@@ -34,3 +36,32 @@ fn test_node_info_ip_port() -> anyhow::Result<()> {
 
     Ok(())
 }
+
+#[test]
+fn test_serde_node_info() {
+    let mut info = NodeInfo {
+        id: "test_id".to_string(),
+        secret: "test_secret".to_string(),
+        version: 1,
+        cpu_nums: 1,
+        http_address: "7.8.9.10:987".to_string(),
+        flight_address: "1.2.3.4:123".to_string(),
+        discovery_address: "4.5.6.7:456".to_string(),
+        binary_version: "v0.8-binary-version".to_string(),
+        cluster_id: String::new(),
+        warehouse_id: String::new(),
+    };
+
+    let json_str = serde_json::to_string(&info).unwrap();
+    assert_eq!(info, serde_json::from_str::<NodeInfo>(&json_str).unwrap());
+    assert!(!json_str.contains("cluster"));
+    assert!(!json_str.contains("warehouse"));
+
+    info.cluster_id = String::from("test-cluster-id");
+    info.warehouse_id = String::from("test-warehouse-id");
+
+    assert_eq!(
+        info,
+        serde_json::from_slice::<NodeInfo>(&serde_json::to_vec(&info).unwrap()).unwrap()
+    );
+}

src/query/management/src/lib.rs

@@ -14,7 +14,6 @@
 
 #![allow(clippy::uninlined_format_args)]
 
-mod cluster;
 mod connection;
 mod file_format;
 mod network_policy;
@@ -26,14 +25,13 @@ mod setting;
 mod stage;
 pub mod udf;
 mod user;
+mod warehouse;
 
 mod client_session;
 pub mod errors;
 mod procedure;
 
 pub use client_session::ClientSessionMgr;
-pub use cluster::ClusterApi;
-pub use cluster::ClusterMgr;
 pub use connection::ConnectionMgr;
 pub use file_format::FileFormatMgr;
 pub use network_policy::NetworkPolicyMgr;
@@ -52,3 +50,5 @@ pub use stage::StageApi;
 pub use stage::StageMgr;
 pub use user::UserApi;
 pub use user::UserMgr;
+pub use warehouse::ClusterApi;
+pub use warehouse::ClusterMgr;

src/query/management/src/warehouse/cluster_api.rs

@@ -0,0 +1,34 @@
+// Copyright 2021 Datafuse Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use databend_common_exception::Result;
+use databend_common_meta_types::NodeInfo;
+
+/// Databend-query cluster management API
+#[async_trait::async_trait]
+pub trait ClusterApi: Sync + Send {
+    /// Add a new node.
+    async fn add_node(&self, node: NodeInfo) -> Result<()>;
+
+    /// Keep the tenant's cluster node alive.
+    async fn heartbeat(&self, node: &NodeInfo) -> Result<()>;
+
+    /// Get the tenant's cluster all nodes.
+    async fn get_nodes(&self, warehouse: &str, cluster: &str) -> Result<Vec<NodeInfo>>;
+
+    /// Drop the tenant's cluster one node by node.id.
+    async fn drop_node(&self, node_id: String) -> Result<()>;
+
+    async fn get_local_addr(&self) -> Result<Option<String>>;
+}