Merge pull request #18 from socutes:openraft

cluster is developed based on openraft

Author: socutes

Cargo.lock

@@ -18,7 +18,7 @@ members = [
     "src/placement-center",
     "src/cmd",
     "src/protocol",
-    "src/clients"
+    "src/clients",
 ]
 
 resolver = "2"
@@ -46,10 +46,18 @@ rocksdb = "0.22.0"
 bincode = "1.3.3"
 tokio-util = { version = "0.7.9", features = ["codec"] }
 mobc = "0.8.4"
+openraft = { git = "https://github.com/databendlabs/openraft.git", features = [
+    "serde",
+    "type-alias",
+] }
+byteorder = "1.5.0"
+tracing = "0.1.40"
+tracing-subscriber = { version = "0.3.0", features = ["env-filter"] }
+
 
 ## workspaces members
 placement-center = { path = "src/placement-center" }
 cmd = { path = "src/cmd" }
 protocol = { path = "src/protocol" }
 clients = { path = "src/clients" }
-common-base = { path = "src/common/base" }
+common-base = { path = "src/common/base" }

Cargo.toml

@@ -16,12 +16,14 @@ use crate::{poll::ClientPool, retry_sleep_time, retry_times};
 use common_base::errors::RobustMQError;
 use kv::kv_interface_call;
 use log::error;
+use openraft::openraft_interface_call;
 use std::{sync::Arc, time::Duration};
 use tokio::time::sleep;
 
 #[derive(Clone, Debug)]
 pub enum PlacementCenterService {
     Kv,
+    OpenRaft,
 }
 
 #[derive(Clone, Debug, PartialEq, Eq, Hash)]
@@ -31,9 +33,15 @@ pub enum PlacementCenterInterface {
     Get,
     Delete,
     Exists,
+
+    // Open Raft
+    Vote,
+    Append,
+    Snapshot,
 }
 
 pub mod kv;
+pub mod openraft;
 
 async fn retry_call(
     service: PlacementCenterService,
@@ -56,6 +64,16 @@ async fn retry_call(
                 )
                 .await
             }
+
+            PlacementCenterService::OpenRaft => {
+                openraft_interface_call(
+                    interface.clone(),
+                    client_poll.clone(),
+                    addr.clone(),
+                    request.clone(),
+                )
+                .await
+            }
         };
 
         match result {

src/clients/src/placement/mod.rs

@@ -0,0 +1,100 @@
+// Copyright 2023 RobustMQ Team
+//
+// 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 super::PlacementCenterInterface;
+use crate::{
+    placement::{retry_call, PlacementCenterService},
+    poll::ClientPool,
+};
+use common_base::errors::RobustMQError;
+use prost::Message as _;
+use protocol::openraft::{
+    AppendReply, AppendRequest, SnapshotReply, SnapshotRequest, VoteReply, VoteRequest,
+};
+use std::sync::Arc;
+
+pub async fn placement_openraft_vote(
+    client_poll: Arc<ClientPool>,
+    addrs: Vec<String>,
+    request: VoteRequest,
+) -> Result<VoteReply, RobustMQError> {
+    let request_data = VoteRequest::encode_to_vec(&request);
+    match retry_call(
+        PlacementCenterService::OpenRaft,
+        PlacementCenterInterface::Vote,
+        client_poll,
+        addrs,
+        request_data,
+    )
+    .await
+    {
+        Ok(data) => match VoteReply::decode(data.as_ref()) {
+            Ok(da) => return Ok(da),
+            Err(e) => return Err(RobustMQError::CommmonError(e.to_string())),
+        },
+        Err(e) => {
+            return Err(e);
+        }
+    }
+}
+
+pub async fn placement_openraft_append(
+    client_poll: Arc<ClientPool>,
+    addrs: Vec<String>,
+    request: AppendRequest,
+) -> Result<AppendReply, RobustMQError> {
+    let request_data = AppendRequest::encode_to_vec(&request);
+    match retry_call(
+        PlacementCenterService::OpenRaft,
+        PlacementCenterInterface::Append,
+        client_poll,
+        addrs,
+        request_data,
+    )
+    .await
+    {
+        Ok(data) => match AppendReply::decode(data.as_ref()) {
+            Ok(da) => return Ok(da),
+            Err(e) => return Err(RobustMQError::CommmonError(e.to_string())),
+        },
+        Err(e) => {
+            return Err(e);
+        }
+    }
+}
+
+pub async fn placement_openraft_snapshot(
+    client_poll: Arc<ClientPool>,
+    addrs: Vec<String>,
+    request: SnapshotRequest,
+) -> Result<SnapshotReply, RobustMQError> {
+    let request_data = SnapshotRequest::encode_to_vec(&request);
+    match retry_call(
+        PlacementCenterService::OpenRaft,
+        PlacementCenterInterface::Snapshot,
+        client_poll,
+        addrs,
+        request_data,
+    )
+    .await
+    {
+        Ok(data) => match SnapshotReply::decode(data.as_ref()) {
+            Ok(da) => return Ok(da),
+            Err(e) => return Err(RobustMQError::CommmonError(e.to_string())),
+        },
+        Err(e) => {
+            return Err(e);
+        }
+    }
+}

src/clients/src/placement/openraft/call.rs

@@ -0,0 +1,72 @@
+// Copyright 2023 RobustMQ Team
+//
+// 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 super::OpenRaftServiceManager;
+use common_base::errors::RobustMQError;
+use mobc::Connection;
+use prost::Message;
+use protocol::openraft::{
+    AppendReply, AppendRequest, SnapshotReply, SnapshotRequest, VoteReply, VoteRequest,
+};
+
+pub(crate) async fn inner_vote(
+    mut client: Connection<OpenRaftServiceManager>,
+    request: Vec<u8>,
+) -> Result<Vec<u8>, RobustMQError> {
+    match VoteRequest::decode(request.as_ref()) {
+        Ok(request) => match client.vote(request).await {
+            Ok(result) => {
+                return Ok(VoteReply::encode_to_vec(&result.into_inner()));
+            }
+            Err(e) => return Err(RobustMQError::GrpcServerStatus(e)),
+        },
+        Err(e) => {
+            return Err(RobustMQError::CommmonError(e.to_string()));
+        }
+    }
+}
+
+pub(crate) async fn inner_append(
+    mut client: Connection<OpenRaftServiceManager>,
+    request: Vec<u8>,
+) -> Result<Vec<u8>, RobustMQError> {
+    match AppendRequest::decode(request.as_ref()) {
+        Ok(request) => match client.append(request).await {
+            Ok(result) => {
+                return Ok(AppendReply::encode_to_vec(&result.into_inner()));
+            }
+            Err(e) => return Err(RobustMQError::GrpcServerStatus(e)),
+        },
+        Err(e) => {
+            return Err(RobustMQError::CommmonError(e.to_string()));
+        }
+    }
+}
+
+pub(crate) async fn inner_snapshot(
+    mut client: Connection<OpenRaftServiceManager>,
+    request: Vec<u8>,
+) -> Result<Vec<u8>, RobustMQError> {
+    match SnapshotRequest::decode(request.as_ref()) {
+        Ok(request) => match client.snapshot(request).await {
+            Ok(result) => {
+                return Ok(SnapshotReply::encode_to_vec(&result.into_inner()));
+            }
+            Err(e) => return Err(RobustMQError::GrpcServerStatus(e)),
+        },
+        Err(e) => {
+            return Err(RobustMQError::CommmonError(e.to_string()));
+        }
+    }
+}

src/clients/src/placement/openraft/inner.rs

@@ -0,0 +1,110 @@
+// Copyright 2023 RobustMQ Team
+//
+// 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 super::PlacementCenterInterface;
+use crate::poll::ClientPool;
+use common_base::errors::RobustMQError;
+use inner::{inner_append, inner_snapshot, inner_vote};
+use mobc::{Connection, Manager};
+use protocol::openraft::open_raft_service_client::OpenRaftServiceClient;
+use std::sync::Arc;
+use tonic::transport::Channel;
+
+pub mod call;
+mod inner;
+
+pub(crate) async fn openraft_interface_call(
+    interface: PlacementCenterInterface,
+    client_poll: Arc<ClientPool>,
+    addr: String,
+    request: Vec<u8>,
+) -> Result<Vec<u8>, RobustMQError> {
+    match openraft_client(client_poll.clone(), addr.clone()).await {
+        Ok(client) => {
+            let result = match interface {
+                PlacementCenterInterface::Vote => inner_vote(client, request.clone()).await,
+                PlacementCenterInterface::Append => inner_append(client, request.clone()).await,
+                PlacementCenterInterface::Snapshot => inner_snapshot(client, request.clone()).await,
+                _ => {
+                    return Err(RobustMQError::CommmonError(format!(
+                        "openraft service does not support service interfaces [{:?}]",
+                        interface
+                    )))
+                }
+            };
+            match result {
+                Ok(data) => return Ok(data),
+                Err(e) => {
+                    return Err(e);
+                }
+            }
+        }
+        Err(e) => {
+            return Err(e);
+        }
+    }
+}
+
+async fn openraft_client(
+    client_poll: Arc<ClientPool>,
+    addr: String,
+) -> Result<Connection<OpenRaftServiceManager>, RobustMQError> {
+    match client_poll
+        .placement_center_openraft_services_client(addr)
+        .await
+    {
+        Ok(client) => {
+            return Ok(client);
+        }
+        Err(e) => {
+            return Err(e);
+        }
+    }
+}
+
+#[derive(Clone)]
+pub struct OpenRaftServiceManager {
+    pub addr: String,
+}
+
+impl OpenRaftServiceManager {
+    pub fn new(addr: String) -> Self {
+        Self { addr }
+    }
+}
+
+#[tonic::async_trait]
+impl Manager for OpenRaftServiceManager {
+    type Connection = OpenRaftServiceClient<Channel>;
+    type Error = RobustMQError;
+
+    async fn connect(&self) -> Result<Self::Connection, Self::Error> {
+        match OpenRaftServiceClient::connect(format!("http://{}", self.addr.clone())).await {
+            Ok(client) => {
+                return Ok(client);
+            }
+            Err(err) => {
+                return Err(RobustMQError::CommmonError(format!(
+                    "{},{}",
+                    err.to_string(),
+                    self.addr.clone()
+                )))
+            }
+        };
+    }
+
+    async fn check(&self, conn: Self::Connection) -> Result<Self::Connection, Self::Error> {
+        Ok(conn)
+    }
+}