refactor(meta-service): add RaftWriteError and RaftChangeMembershipError

drmingdrmer · drmingdrmer · commit fadbe42ad185 · 2022-09-15T21:32:59.000+08:00
- Meta-service replace returning error `MetaError` with
  `RaftChangeMembershipError` and `RaftWriteError` for `join()`, `leave()`
  and `write()`.

  These three methods do not need to all the sub errors in MetaError.

- Also remove unnecessary error checking before changing membership.
  The error will be returned by raft.

- Use RAII to report metrics of pending proposals.
diff --git a/src/meta/service/src/meta_service/meta_leader.rs b/src/meta/service/src/meta_service/meta_leader.rs
@@ -12,30 +12,26 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::collections::BTreeSet;
-
 use common_meta_api::KVApi;
-use common_meta_sled_store::openraft::error::ChangeMembershipError;
-use common_meta_sled_store::openraft::error::ClientWriteError;
-use common_meta_sled_store::openraft::error::InProgress;
 use common_meta_types::AppliedState;
 use common_meta_types::Cmd;
 use common_meta_types::ForwardRequest;
 use common_meta_types::ForwardResponse;
-use common_meta_types::ForwardToLeader;
 use common_meta_types::LogEntry;
 use common_meta_types::MetaError;
-use common_meta_types::MetaRaftError;
 use common_meta_types::Node;
-use common_meta_types::NodeId;
+use common_meta_types::RaftChangeMembershipError;
+use common_meta_types::RaftWriteError;
 use common_meta_types::SeqV;
+use common_metrics::counter::WithCount;
 use tracing::debug;
 use tracing::info;
 
 use crate::meta_service::ForwardRequestBody;
 use crate::meta_service::JoinRequest;
 use crate::meta_service::LeaveRequest;
 use crate::meta_service::MetaNode;
+use crate::metrics::ProposalPending;
 
 /// The container of APIs of a metasrv leader in a metasrv cluster.
 ///
@@ -98,7 +94,7 @@ impl<'a> MetaLeader<'a> {
     ///
     /// If the node is already in cluster membership, it still returns Ok.
     #[tracing::instrument(level = "debug", skip(self))]
-    pub async fn join(&self, req: JoinRequest) -> Result<(), MetaError> {
+    pub async fn join(&self, req: JoinRequest) -> Result<(), RaftChangeMembershipError> {
         let node_id = req.node_id;
         let endpoint = req.endpoint;
         let grpc_api_addr = req.grpc_api_addr;
@@ -109,35 +105,28 @@ impl<'a> MetaLeader<'a> {
             return Ok(());
         }
 
-        // deal with joint config，If the cluster is in joint config,
-        // we need to return an Inprogress error with membership log id.
-        match membership.get_ith_config(1) {
-            Some(_membership) => Err(MetaRaftError::ChangeMembershipError(
-                ChangeMembershipError::InProgress(InProgress {
-                    membership_log_id: metrics.membership_config.log_id,
-                }),
-            )
-            .into()),
-            None => {
-                // safe unwrap: if the first config is None, panic is the expected behavior here.
-                let mut membership = membership.get_ith_config(0).unwrap().clone();
-                membership.insert(node_id);
-                let ent = LogEntry {
-                    txid: None,
-                    time_ms: None,
-                    cmd: Cmd::AddNode {
-                        node_id,
-                        node: Node {
-                            name: node_id.to_string(),
-                            endpoint,
-                            grpc_api_addr: Some(grpc_api_addr),
-                        },
-                    },
-                };
-                self.write(ent).await?;
-                self.change_membership(membership).await
-            }
-        }
+        // safe unwrap: if the first config is None, panic is the expected behavior here.
+        let mut membership = membership.get_ith_config(0).unwrap().clone();
+        membership.insert(node_id);
+        let ent = LogEntry {
+            txid: None,
+            time_ms: None,
+            cmd: Cmd::AddNode {
+                node_id,
+                node: Node {
+                    name: node_id.to_string(),
+                    endpoint,
+                    grpc_api_addr: Some(grpc_api_addr),
+                },
+            },
+        };
+        self.write(ent).await?;
+
+        self.meta_node
+            .raft
+            .change_membership(membership, true)
+            .await?;
+        Ok(())
     }
 
     /// A node leave the cluster.
@@ -147,7 +136,7 @@ impl<'a> MetaLeader<'a> {
     ///
     /// If the node is not in cluster membership, it still returns Ok.
     #[tracing::instrument(level = "debug", skip(self))]
-    pub async fn leave(&self, req: LeaveRequest) -> Result<(), MetaError> {
+    pub async fn leave(&self, req: LeaveRequest) -> Result<(), RaftChangeMembershipError> {
         let node_id = req.node_id;
         let metrics = self.meta_node.raft.metrics().borrow().clone();
         let membership = metrics.membership_config.membership.clone();
@@ -156,93 +145,42 @@ impl<'a> MetaLeader<'a> {
             return Ok(());
         }
 
-        // deal with joint config，If the cluster is in joint config,
-        // we need to return an Inprogress error with membership log id.
-        match membership.get_ith_config(1) {
-            Some(_membership) => Err(MetaRaftError::ChangeMembershipError(
-                ChangeMembershipError::InProgress(InProgress {
-                    membership_log_id: metrics.membership_config.log_id,
-                }),
-            )
-            .into()),
-            None => {
-                // safe unwrap: if the first config is None, panic is the expected behavior here.
-                let mut membership = membership.get_ith_config(0).unwrap().clone();
-                membership.remove(&node_id);
-
-                self.change_membership(membership).await?;
-                let ent = LogEntry {
-                    txid: None,
-                    time_ms: None,
-                    cmd: Cmd::RemoveNode { node_id },
-                };
-                self.write(ent).await?;
-                Ok(())
-            }
-        }
-    }
+        // safe unwrap: if the first config is None, panic is the expected behavior here.
+        let mut membership = membership.get_ith_config(0).unwrap().clone();
+        membership.remove(&node_id);
 
-    #[tracing::instrument(level = "debug", skip(self))]
-    pub async fn change_membership(&self, membership: BTreeSet<NodeId>) -> Result<(), MetaError> {
-        let res = self
-            .meta_node
+        self.meta_node
             .raft
             .change_membership(membership, true)
-            .await;
+            .await?;
 
-        let err = match res {
-            Ok(_) => return Ok(()),
-            Err(e) => e,
+        let ent = LogEntry {
+            txid: None,
+            time_ms: None,
+            cmd: Cmd::RemoveNode { node_id },
         };
-
-        match err {
-            ClientWriteError::ChangeMembershipError(e) => {
-                Err(MetaRaftError::ChangeMembershipError(e).into())
-            }
-            // TODO(xp): enable MetaNode::RaftError when RaftError impl Serialized
-            ClientWriteError::Fatal(fatal) => Err(MetaRaftError::RaftFatal(fatal).into()),
-            ClientWriteError::ForwardToLeader(to_leader) => {
-                Err(MetaRaftError::ForwardToLeader(ForwardToLeader {
-                    leader_id: to_leader.leader_id,
-                })
-                .into())
-            }
-        }
+        self.write(ent).await?;
+        Ok(())
     }
 
     /// Write a log through local raft node and return the states before and after applying the log.
     ///
     /// If the raft node is not a leader, it returns MetaRaftError::ForwardToLeader.
     #[tracing::instrument(level = "debug", skip(self, entry))]
-    pub async fn write(&self, mut entry: LogEntry) -> Result<AppliedState, MetaError> {
+    pub async fn write(&self, mut entry: LogEntry) -> Result<AppliedState, RaftWriteError> {
         // Add consistent clock time to log entry.
         entry.time_ms = Some(SeqV::<()>::now_ms());
 
+        // report metrics
+        let _guard = WithCount::new((), ProposalPending);
+
         info!("write LogEntry: {:?}", entry);
         let write_rst = self.meta_node.raft.client_write(entry).await;
-
         info!("raft.client_write rst: {:?}", write_rst);
 
         match write_rst {
-            Ok(resp) => {
-                let data = resp.data;
-                Ok(data)
-            }
-
-            Err(cli_write_err) => match cli_write_err {
-                // fatal error
-                ClientWriteError::Fatal(fatal) => Err(MetaRaftError::RaftFatal(fatal).into()),
-                // retryable error
-                ClientWriteError::ForwardToLeader(to_leader) => {
-                    Err(MetaRaftError::ForwardToLeader(ForwardToLeader {
-                        leader_id: to_leader.leader_id,
-                    })
-                    .into())
-                }
-                ClientWriteError::ChangeMembershipError(_) => {
-                    unreachable!("there should not be a ChangeMembershipError for client_write")
-                }
-            },
+            Ok(resp) => Ok(resp.data),
+            Err(cli_write_err) => Err(RaftWriteError::from_raft_err(cli_write_err)),
         }
     }
 }
diff --git a/src/meta/service/src/meta_service/meta_service_impl.rs b/src/meta/service/src/meta_service/meta_service_impl.rs
@@ -33,7 +33,6 @@ use tonic::codegen::futures_core::Stream;
 use crate::meta_service::ForwardRequestBody;
 use crate::meta_service::MetaNode;
 use crate::metrics::incr_meta_metrics_proposals_failed;
-use crate::metrics::incr_meta_metrics_proposals_pending;
 use crate::metrics::incr_meta_metrics_recv_bytes_from_peer;
 use crate::metrics::incr_meta_metrics_snapshot_recv_failure_from_peer;
 use crate::metrics::incr_meta_metrics_snapshot_recv_inflights_from_peer;
@@ -72,12 +71,9 @@ impl RaftService for RaftServiceImpl {
     ) -> Result<tonic::Response<RaftReply>, tonic::Status> {
         common_tracing::extract_remote_span_as_parent(&request);
 
-        let mes = request.into_inner();
-        let ent: LogEntry = mes.try_into()?;
+        let raft_req = request.into_inner();
+        let ent: LogEntry = raft_req.try_into()?;
 
-        incr_meta_metrics_proposals_pending(1);
-
-        // TODO(xp): call meta_node.write()
         let res = self
             .meta_node
             .handle_forwardable_request(ForwardRequest {
@@ -86,8 +82,6 @@ impl RaftService for RaftServiceImpl {
             })
             .await;
 
-        incr_meta_metrics_proposals_pending(-1);
-
         let res = match res {
             Ok(r) => {
                 let a: Result<AppliedState, MetaError> = r.try_into().map_err(|e: &str| {
@@ -115,10 +109,10 @@ impl RaftService for RaftServiceImpl {
 
         let req = request.into_inner();
 
-        let admin_req: ForwardRequest = serde_json::from_str(&req.data)
+        let forward_req: ForwardRequest = serde_json::from_str(&req.data)
             .map_err(|x| tonic::Status::invalid_argument(x.to_string()))?;
 
-        let res = self.meta_node.handle_forwardable_request(admin_req).await;
+        let res = self.meta_node.handle_forwardable_request(forward_req).await;
 
         let raft_mes: RaftReply = res.into();
 
diff --git a/src/meta/service/src/metrics/meta_metrics.rs b/src/meta/service/src/metrics/meta_metrics.rs
@@ -714,6 +714,15 @@ impl counter::Count for RequestInFlight {
     }
 }
 
+/// RAII metrics counter of pending raft proposals
+pub(crate) struct ProposalPending;
+
+impl counter::Count for ProposalPending {
+    fn incr_count(&mut self, n: i64) {
+        PROPOSALS_PENDING.add(n);
+    }
+}
+
 /// Encode metrics as prometheus format string
 pub fn meta_metrics_to_prometheus_string() -> String {
     use prometheus::Encoder;
diff --git a/src/meta/service/src/metrics/mod.rs b/src/meta/service/src/metrics/mod.rs
@@ -47,4 +47,5 @@ pub use meta_metrics::set_meta_metrics_last_log_index;
 pub use meta_metrics::set_meta_metrics_last_seq;
 pub use meta_metrics::set_meta_metrics_node_is_health;
 pub use meta_metrics::set_meta_metrics_proposals_applied;
+pub(crate) use meta_metrics::ProposalPending;
 pub(crate) use meta_metrics::RequestInFlight;
diff --git a/src/meta/service/tests/it/meta_node/meta_node_request_forwarding.rs b/src/meta/service/tests/it/meta_node/meta_node_request_forwarding.rs
@@ -18,8 +18,7 @@ use common_base::base::tokio;
 use common_meta_types::Cmd;
 use common_meta_types::ForwardToLeader;
 use common_meta_types::LogEntry;
-use common_meta_types::MetaError;
-use common_meta_types::MetaRaftError;
+use common_meta_types::RaftWriteError;
 use common_meta_types::UpsertKV;
 use databend_meta::meta_service::meta_leader::MetaLeader;
 use databend_meta::meta_service::MetaNode;
@@ -59,9 +58,9 @@ async fn test_meta_node_forward_to_leader() -> anyhow::Result<()> {
             assert!(rst.is_err());
             let e = rst.unwrap_err();
             match e {
-                MetaError::MetaRaftError(MetaRaftError::ForwardToLeader(ForwardToLeader {
+                RaftWriteError::ForwardToLeader(ForwardToLeader {
                     leader_id: forward_leader_id,
-                })) => {
+                }) => {
                     assert_eq!(Some(leader_id), forward_leader_id);
                 }
                 _ => {
diff --git a/src/meta/types/src/lib.rs b/src/meta/types/src/lib.rs
@@ -131,6 +131,8 @@ pub use meta_raft_errors::Fatal;
 pub use meta_raft_errors::ForwardToLeader;
 pub use meta_raft_errors::InitializeError;
 pub use meta_raft_errors::MetaRaftError;
+pub use meta_raft_errors::RaftChangeMembershipError;
+pub use meta_raft_errors::RaftWriteError;
 pub use meta_raft_errors::RetryableError;
 pub use meta_storage_errors::MetaStorageError;
 pub use meta_storage_errors::MetaStorageResult;
diff --git a/src/meta/types/src/meta_raft_errors.rs b/src/meta/types/src/meta_raft_errors.rs
