Merge pull request databendlabs#2751 from drmingdrmer/refact-meta-client

[query] refactor: explicit impl of building a MetaFlightClientConf

query/src/catalogs/impls/catalog/metastore_catalog.rs

@@ -92,7 +92,8 @@ impl MetaStoreCatalog {
         } else {
             tracing::info!("use remote meta");
 
-            let meta_client_provider = Arc::new(MetaClientProvider::new(&conf));
+            let meta_client_provider =
+                Arc::new(MetaClientProvider::new(conf.meta.to_flight_client_config()));
             let meta_remote = MetaRemote::create(meta_client_provider);
             Arc::new(meta_remote)
         };

query/src/clusters/cluster.rs

@@ -57,7 +57,7 @@ pub struct ClusterDiscovery {
 
 impl ClusterDiscovery {
     async fn create_meta_client(cfg: &Config) -> Result<Arc<dyn KVApi>> {
-        let meta_api_provider = MetaClientProvider::new(cfg);
+        let meta_api_provider = MetaClientProvider::new(cfg.meta.to_flight_client_config());
         match meta_api_provider.try_get_kv_client().await {
             Ok(client) => Ok(client),
             Err(cause) => Err(cause.add_message_back("(while create cluster api).")),

query/src/common/meta/meta_client.rs

@@ -17,7 +17,6 @@ use std::sync::Arc;
 
 use common_exception::Result;
 use common_meta_api::KVApi;
-use common_meta_api::MetaApi;
 use common_meta_flight::MetaFlightClient;
 use common_meta_flight::MetaFlightClientConf;
 
@@ -28,18 +27,16 @@ use common_meta_flight::MetaFlightClientConf;
 
 #[derive(Clone)]
 pub struct MetaClientProvider {
-    // do not depend on query::configs::Config in case of moving back to sdk
-    // also @see config_converter.rs
     conf: MetaFlightClientConf,
 }
 
 impl MetaClientProvider {
-    pub fn new(conf: impl Into<MetaFlightClientConf>) -> Self {
-        MetaClientProvider { conf: conf.into() }
+    pub fn new(conf: MetaFlightClientConf) -> Self {
+        MetaClientProvider { conf }
     }
 
     /// Get meta async client, trait is defined in MetaApi.
-    pub async fn try_get_meta_client(&self) -> Result<Arc<dyn MetaApi>> {
+    pub async fn try_get_meta_client(&self) -> Result<Arc<MetaFlightClient>> {
         let client = MetaFlightClient::try_new(&self.conf).await?;
         Ok(Arc::new(client))
     }

query/src/common/meta/mod.rs

@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-mod config_converter;
 mod meta_client;
 
 pub use meta_client::MetaClientProvider;

query/src/configs/config_meta.rs

@@ -14,6 +14,9 @@
 
 use std::fmt;
 
+use common_flight_rpc::FlightClientConf;
+use common_flight_rpc::FlightClientTlsConfig;
+use common_meta_flight::MetaFlightClientConf;
 use structopt::StructOpt;
 use structopt_toml::StructOptToml;
 
@@ -106,6 +109,39 @@ impl MetaConfig {
             META_RPC_TLS_SERVICE_DOMAIN_NAME
         );
     }
+
+    pub fn is_tls_enabled(&self) -> bool {
+        !self.rpc_tls_meta_server_root_ca_cert.is_empty()
+            && !self.rpc_tls_meta_service_domain_name.is_empty()
+    }
+
+    pub fn to_flight_tls_config(&self) -> Option<FlightClientTlsConfig> {
+        if !self.is_tls_enabled() {
+            return None;
+        }
+
+        Some(FlightClientTlsConfig {
+            rpc_tls_server_root_ca_cert: self.rpc_tls_meta_server_root_ca_cert.clone(),
+            domain_name: self.rpc_tls_meta_service_domain_name.clone(),
+        })
+    }
+
+    pub fn to_flight_client_config(&self) -> MetaFlightClientConf {
+        let meta_config = FlightClientConf {
+            address: self.meta_address.clone(),
+            username: self.meta_username.clone(),
+            password: self.meta_password.clone(),
+            tls_conf: self.to_flight_tls_config(),
+        };
+
+        MetaFlightClientConf {
+            // kv service is configured by conf.meta
+            kv_service_config: meta_config.clone(),
+            // copy meta config from query config
+            meta_service_config: meta_config,
+            client_timeout_in_second: self.meta_client_timeout_in_second,
+        }
+    }
 }
 
 impl fmt::Debug for MetaConfig {

query/src/users/user_mgr.rs

@@ -34,7 +34,10 @@ pub struct UserManager {
 
 impl UserManager {
     async fn create_kv_client(cfg: &Config) -> Result<Arc<dyn KVApi>> {
-        match MetaClientProvider::new(cfg).try_get_kv_client().await {
+        match MetaClientProvider::new(cfg.meta.to_flight_client_config())
+            .try_get_kv_client()
+            .await
+        {
             Ok(client) => Ok(client),
             Err(cause) => Err(cause.add_message_back("(while create user api).")),
         }