Merge pull request #7555 from drmingdrmer/8-grpc-req

refactor(meta-service): remove redundant ActionHandler; move logic into MetaServiceImpl

Author: drmingdrmer

src/meta/client/src/grpc_action.rs

@@ -59,6 +59,33 @@ pub enum MetaGrpcReadReq {
     ListKV(ListKVReq), // since 2022-05-23
 }
 
+#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, derive_more::From)]
+pub enum MetaGrpcReq {
+    UpsertKV(UpsertKVReq),
+
+    GetKV(GetKVReq),
+    MGetKV(MGetKVReq),
+    ListKV(ListKVReq),
+}
+
+impl From<MetaGrpcWriteReq> for MetaGrpcReq {
+    fn from(r: MetaGrpcWriteReq) -> Self {
+        match r {
+            MetaGrpcWriteReq::UpsertKV(x) => x.into(),
+        }
+    }
+}
+
+impl From<MetaGrpcReadReq> for MetaGrpcReq {
+    fn from(r: MetaGrpcReadReq) -> Self {
+        match r {
+            MetaGrpcReadReq::GetKV(x) => x.into(),
+            MetaGrpcReadReq::MGetKV(x) => x.into(),
+            MetaGrpcReadReq::ListKV(x) => x.into(),
+        }
+    }
+}
+
 /// Try convert tonic::Request<RaftRequest> to DoActionAction.
 impl TryInto<MetaGrpcWriteReq> for Request<RaftRequest> {
     type Error = tonic::Status;

src/meta/client/src/lib.rs

@@ -18,6 +18,7 @@ mod kv_api_impl;
 mod message;
 
 pub use grpc_action::MetaGrpcReadReq;
+pub use grpc_action::MetaGrpcReq;
 pub use grpc_action::MetaGrpcWriteReq;
 pub use grpc_action::RequestFor;
 pub use grpc_client::ClientHandle;

src/meta/service/src/api/grpc/grpc_service.rs

@@ -21,7 +21,9 @@ use common_arrow::arrow_format::flight::data::BasicAuth;
 use common_base::base::tokio::sync::mpsc;
 use common_grpc::GrpcClaim;
 use common_grpc::GrpcToken;
+use common_meta_api::KVApi;
 use common_meta_client::MetaGrpcReadReq;
+use common_meta_client::MetaGrpcReq;
 use common_meta_client::MetaGrpcWriteReq;
 use common_meta_types::protobuf::meta_service_server::MetaService;
 use common_meta_types::protobuf::ClientInfo;
@@ -48,11 +50,11 @@ use tonic::Status;
 use tonic::Streaming;
 use tracing::info;
 
-use crate::executor::ActionHandler;
 use crate::meta_service::meta_service_impl::GrpcStream;
 use crate::meta_service::MetaNode;
 use crate::metrics::add_meta_metrics_meta_request_inflights;
 use crate::metrics::incr_meta_metrics_meta_recv_bytes;
+use crate::metrics::incr_meta_metrics_meta_request_result;
 use crate::metrics::incr_meta_metrics_meta_sent_bytes;
 use crate::version::from_digit_ver;
 use crate::version::to_digit_ver;
@@ -61,14 +63,14 @@ use crate::version::MIN_METACLI_SEMVER;
 
 pub struct MetaServiceImpl {
     token: GrpcToken,
-    action_handler: ActionHandler,
+    pub(crate) meta_node: Arc<MetaNode>,
 }
 
 impl MetaServiceImpl {
     pub fn create(meta_node: Arc<MetaNode>) -> Self {
         Self {
             token: GrpcToken::create(),
-            action_handler: ActionHandler::create(meta_node),
+            meta_node,
         }
     }
 
@@ -84,6 +86,47 @@ impl MetaServiceImpl {
         })?;
         Ok(claim)
     }
+
+    pub async fn execute_kv_req(&self, req: MetaGrpcReq) -> RaftReply {
+        // To keep the code IDE-friendly, we manually expand the enum variants and dispatch them one by one
+
+        match req {
+            MetaGrpcReq::UpsertKV(a) => {
+                let res = self.meta_node.upsert_kv(a).await;
+                incr_meta_metrics_meta_request_result(res.is_ok());
+                RaftReply::from(res)
+            }
+            MetaGrpcReq::GetKV(a) => {
+                let res = self.meta_node.get_kv(&a.key).await;
+                incr_meta_metrics_meta_request_result(res.is_ok());
+                RaftReply::from(res)
+            }
+            MetaGrpcReq::MGetKV(a) => {
+                let res = self.meta_node.mget_kv(&a.keys).await;
+                incr_meta_metrics_meta_request_result(res.is_ok());
+                RaftReply::from(res)
+            }
+            MetaGrpcReq::ListKV(a) => {
+                let res = self.meta_node.prefix_list_kv(&a.prefix).await;
+                incr_meta_metrics_meta_request_result(res.is_ok());
+                RaftReply::from(res)
+            }
+        }
+    }
+
+    pub async fn execute_txn(&self, req: TxnRequest) -> TxnReply {
+        let ret = self.meta_node.transaction(req).await;
+        incr_meta_metrics_meta_request_result(ret.is_ok());
+
+        match ret {
+            Ok(resp) => resp,
+            Err(err) => TxnReply {
+                success: false,
+                error: serde_json::to_string(&err).expect("fail to serialize"),
+                responses: vec![],
+            },
+        }
+    }
 }
 
 #[async_trait::async_trait]
@@ -154,13 +197,14 @@ impl MetaService for MetaServiceImpl {
 
         incr_meta_metrics_meta_recv_bytes(request.get_ref().encoded_len() as u64);
 
-        let action: MetaGrpcWriteReq = request.try_into()?;
+        let req: MetaGrpcWriteReq = request.try_into()?;
+        let req: MetaGrpcReq = req.into();
 
         add_meta_metrics_meta_request_inflights(1);
 
-        info!("Receive write_action: {:?}", action);
+        info!("Receive write_action: {:?}", req);
 
-        let body = self.action_handler.execute_write(action).await;
+        let body = self.execute_kv_req(req).await;
 
         add_meta_metrics_meta_request_inflights(-1);
 
@@ -175,13 +219,14 @@ impl MetaService for MetaServiceImpl {
 
         incr_meta_metrics_meta_recv_bytes(request.get_ref().encoded_len() as u64);
 
-        let action: MetaGrpcReadReq = request.try_into()?;
+        let req: MetaGrpcReadReq = request.try_into()?;
+        let req: MetaGrpcReq = req.into();
 
         add_meta_metrics_meta_request_inflights(1);
 
-        info!("Receive read_action: {:?}", action);
+        info!("Receive read_action: {:?}", req);
 
-        let res = self.action_handler.execute_read(action).await;
+        let res = self.execute_kv_req(req).await;
 
         add_meta_metrics_meta_request_inflights(-1);
 
@@ -201,7 +246,7 @@ impl MetaService for MetaServiceImpl {
         &self,
         _request: Request<common_meta_types::protobuf::Empty>,
     ) -> Result<Response<Self::ExportStream>, Status> {
-        let meta_node = &self.action_handler.meta_node;
+        let meta_node = &self.meta_node;
 
         let res = meta_node.sto.export().await?;
 
@@ -222,7 +267,7 @@ impl MetaService for MetaServiceImpl {
     ) -> Result<Response<Self::WatchStream>, Status> {
         let (tx, rx) = mpsc::channel(4);
 
-        let meta_node = &self.action_handler.meta_node;
+        let meta_node = &self.meta_node;
         meta_node.create_watcher_stream(request.into_inner(), tx);
 
         let output_stream = tokio_stream::wrappers::ReceiverStream::new(rx);
@@ -243,7 +288,7 @@ impl MetaService for MetaServiceImpl {
 
         info!("Receive txn_request: {:?}", request);
 
-        let body = self.action_handler.execute_txn(request).await;
+        let body = self.execute_txn(request).await;
         add_meta_metrics_meta_request_inflights(-1);
 
         incr_meta_metrics_meta_sent_bytes(body.encoded_len() as u64);
@@ -256,7 +301,7 @@ impl MetaService for MetaServiceImpl {
         request: Request<MemberListRequest>,
     ) -> Result<Response<MemberListReply>, Status> {
         self.check_token(request.metadata())?;
-        let meta_node = &self.action_handler.meta_node;
+        let meta_node = &self.meta_node;
         let members = meta_node.get_meta_addrs().await.map_err(|e| {
             Status::internal(format!("Cannot get metasrv member list, error: {:?}", e))
         })?;

src/meta/service/src/executor/action_handler.rs

@@ -16,7 +16,6 @@
 
 pub mod api;
 pub mod configs;
-pub mod executor;
 pub mod export;
 pub mod meta_service;
 pub mod metrics;