feat: Implementing Redis-Raft commands to build a cluster.

CMakeLists.txt

@@ -148,6 +148,7 @@ ENABLE_TESTING()
 
 ADD_SUBDIRECTORY(src/pstd)
 ADD_SUBDIRECTORY(src/net)
+ADD_SUBDIRECTORY(src/praft)
 ADD_SUBDIRECTORY(src/storage)
 ADD_SUBDIRECTORY(src)
 
@@ -200,4 +201,3 @@ ADD_CUSTOM_TARGET(cpplint echo '${LINT_FILES}' | xargs -n12 -P8
   --linelength=120
   --filter=-legal/copyright,-build/header_guard,-runtime/references
 )
-

cmake/gflags.cmake

@@ -7,9 +7,6 @@ INCLUDE_GUARD()
 
 INCLUDE(cmake/utils.cmake)
 
-SET(MY_BUILD_TYPE ${CMAKE_BUILD_TYPE})
-SET(CMAKE_BUILD_TYPE ${THIRD_PARTY_BUILD_TYPE})
-
 FetchContent_DeclareGitHubWithMirror(gflags
   gflags/gflags v2.2.2
   SHA256=19713a36c9f32b33df59d1c79b4958434cb005b5b47dc5400a7a4b078111d9b5
@@ -22,6 +19,7 @@ FetchContent_MakeAvailableWithArgs(gflags
   BUILD_gflags_LIB=ON
   BUILD_gflags_nothreads_LIB=OFF
   BUILD_TESTING=OFF
+  CMAKE_BUILD_TYPE=${THIRD_PARTY_BUILD_TYPE}
 )
 
 FIND_PACKAGE(Threads REQUIRED)
@@ -30,6 +28,4 @@ TARGET_LINK_LIBRARIES(gflags_static Threads::Threads)
 
 SET(GFLAGS_INCLUDE_PATH ${CMAKE_CURRENT_BINARY_DIR}/_deps/gflags-build/include)
 SET(GFLAGS_LIBRARY ${CMAKE_CURRENT_BINARY_DIR}/_deps/gflags-build/libgflags.a)
-SET(GFLAGS_LIB ${CMAKE_CURRENT_BINARY_DIR}/_deps/gflags-build/libgflags.a)
-
-SET(CMAKE_BUILD_TYPE ${MY_BUILD_TYPE})
+SET(GFLAGS_LIB ${CMAKE_CURRENT_BINARY_DIR}/_deps/gflags-build/libgflags.a)

cmake/glog.cmake

@@ -26,4 +26,4 @@ IF (CMAKE_BUILD_TYPE STREQUAL "Release")
   SET(GLOG_LIB ${CMAKE_CURRENT_BINARY_DIR}/_deps/glog-build/libglog.a CACHE BOOL "" FORCE)
 ELSEIF (CMAKE_BUILD_TYPE STREQUAL "Debug")
   SET(GLOG_LIB ${CMAKE_CURRENT_BINARY_DIR}/_deps/glog-build/libglogd.a CACHE BOOL "" FORCE)
-ENDIF()
+ENDIF()

cmake/leveldb.cmake

@@ -6,9 +6,9 @@
 INCLUDE_GUARD()
 
 FETCHCONTENT_DECLARE(
-        leveldb
-        GIT_REPOSITORY https://github.com/google/leveldb.git
-        GIT_TAG main
+    leveldb
+    GIT_REPOSITORY https://github.com/google/leveldb.git
+    GIT_TAG main
 )
 SET(LEVELDB_BUILD_TESTS OFF CACHE BOOL "" FORCE)
 SET(LEVELDB_BUILD_BENCHMARKS OFF CACHE BOOL "" FORCE)

cmake/modules/glog/Findgflags.cmake

@@ -8,4 +8,4 @@ IF (gflags_SOURCE_DIR)
 
   # add_library(gflags_static::gflags_static ALIAS gflags_static)
   INSTALL(TARGETS gflags_static EXPORT glog-targets)
-ENDIF ()
+ENDIF ()

cmake/protobuf.cmake

@@ -164,4 +164,4 @@ IF (NOT PROTOBUF_FOUND)
     SET(PROTOBUF_PROTOC_EXECUTABLE ${extern_protobuf_PROTOC_EXECUTABLE}
             CACHE FILEPATH "protobuf executable." FORCE)
     PROMPT_PROTOBUF_LIB(extern_protobuf zlib)
-ENDIF (NOT PROTOBUF_FOUND)
+ENDIF (NOT PROTOBUF_FOUND)

cmake/utils.cmake

@@ -44,4 +44,4 @@ FUNCTION(FetchContent_DeclareGitHubWithMirror dep repo tag hash)
     https://github.com/${repo}/archive/${tag}.zip
     ${hash}
   )
-ENDFUNCTION()
+ENDFUNCTION()

pikiwidb.conf

@@ -7,6 +7,9 @@ daemonize no
 # port 0 is not permitted.
 port 9221
 
+# Braft relies on brpc to communicate via the default port number plus the port offset
+raft-port-offset 10
+
 # If you want you can bind a single interface, if the bind option is not
 # specified all the interfaces will listen for incoming connections.
 #

src/CMakeLists.txt

@@ -8,15 +8,17 @@ AUX_SOURCE_DIRECTORY(. PIKIWIDB_SRC)
 ADD_EXECUTABLE(pikiwidb ${PIKIWIDB_SRC})
 SET(EXECUTABLE_OUTPUT_PATH ${PROJECT_SOURCE_DIR}/bin)
 
-TARGET_INCLUDE_DIRECTORIES(pikiwidb PRIVATE
-        ${PROJECT_SOURCE_DIR}/src
-        ${PROJECT_SOURCE_DIR}/src/pstd
-        ${PROJECT_SOURCE_DIR}/src/net
-        ${PROJECT_SOURCE_DIR}/src/storage/include
-        ${rocksdb_SOURCE_DIR}/
-        ${rocksdb_SOURCE_DIR}/include
+TARGET_INCLUDE_DIRECTORIES(pikiwidb
+        PRIVATE ${PROJECT_SOURCE_DIR}/src
+        PRIVATE ${PROJECT_SOURCE_DIR}/src/pstd
+        PRIVATE ${PROJECT_SOURCE_DIR}/src/net
+        PRIVATE ${PROJECT_SOURCE_DIR}/src/storage/include
+        PRIVATE ${rocksdb_SOURCE_DIR}/
+        PRIVATE ${rocksdb_SOURCE_DIR}/include
         PRIVATE ${BRAFT_INCLUDE_DIR}
         PRIVATE ${BRPC_INCLUDE_DIR}
+        PRIVATE ${GFLAGS_INCLUDE_PATH}
+        PRIVATE ${PROJECT_SOURCE_DIR}/src/praft
         )
 
 IF (CMAKE_SYSTEM_NAME MATCHES "Darwin")
@@ -34,5 +36,6 @@ ELSE ()
     MESSAGE(FATAL_ERROR "only support linux or macOS")
 ENDIF ()
 
-TARGET_LINK_LIBRARIES(pikiwidb net; dl; fmt; storage; rocksdb; braft brpc ssl crypto zlib protobuf leveldb gflags "${MAC_LIBRARY}")
-SET_TARGET_PROPERTIES(pikiwidb PROPERTIES LINKER_LANGUAGE CXX)
+TARGET_LINK_LIBRARIES(pikiwidb net; dl; fmt; storage; rocksdb; pstd braft brpc ssl crypto zlib protobuf leveldb gflags rt crypto dl z praft "${MAC_LIBRARY}")
+
+SET_TARGET_PROPERTIES(pikiwidb PROPERTIES LINKER_LANGUAGE CXX)

src/base_cmd.h

@@ -26,6 +26,10 @@ const std::string kCmdNameDel = "del";
 const std::string kCmdNameExists = "exists";
 const std::string kCmdNamePExpire = "pexpire";
 
+// raft cmd
+const std::string kCmdNameRaftCluster = "raft.cluster";
+const std::string kCmdNameRaftNode = "raft.node";
+
 // string cmd
 const std::string kCmdNameSet = "set";
 const std::string kCmdNameGet = "get";
@@ -62,6 +66,7 @@ const std::string kCmdNameFlushdb = "flushdb";
 const std::string kCmdNameFlushall = "flushall";
 const std::string kCmdNameAuth = "auth";
 const std::string kCmdNameSelect = "select";
+const std::string kCmdNameInfo = "info";
 
 // hash cmd
 const std::string kCmdNameHSet = "hset";
@@ -119,6 +124,7 @@ enum CmdFlags {
   kCmdFlagsProtected = (1 << 12),        // Don't accept in scripts
   kCmdFlagsModuleNoCluster = (1 << 13),  // No cluster mode support
   kCmdFlagsNoMulti = (1 << 14),          // Cannot be pipelined
+  kCmdFlagsRaft = (1 << 15),             // raft
 };
 
 enum AclCategory {
@@ -142,7 +148,8 @@ enum AclCategory {
   kAclCategoryDangerous = (1 << 17),
   kAclCategoryConnection = (1 << 18),
   kAclCategoryTransaction = (1 << 19),
-  kAclCategoryScripting = (1 << 20)
+  kAclCategoryScripting = (1 << 20),
+  kAclCategoryRaft = (1 << 21),
 };
 
 /**

src/client.cc

@@ -15,6 +15,7 @@
 #include "pstd_string.h"
 #include "slow_log.h"
 #include "store.h"
+#include "praft.h"
 
 namespace pikiwidb {
 
@@ -130,6 +131,10 @@ void CmdRes::SetRes(CmdRes::CmdRet _ret, const std::string& content) {
     case kInvalidCursor:
       AppendStringRaw("-ERR invalid cursor");
       break;
+    case kWrongLeader:
+      AppendStringRaw("-ERR wrong leader");
+      AppendStringRaw(content);
+      AppendStringRaw(CRLF);
     default:
       break;
   }
@@ -268,6 +273,15 @@ int PClient::handlePacket(const char* start, int bytes) {
     }
   }
 
+  if (isJoinCmdTarget()) {
+    // Proccees the packet at one turn.
+    auto [len, is_disconnect] = PRAFT.ProcessClusterJoinCmdResponse(this, start, bytes);
+    if (is_disconnect) {
+      conn->ActiveClose();
+    }
+    return len;
+  }
+
   auto parseRet = parser_.ParseRequest(ptr, end);
   if (parseRet == PParseResult::kError) {
     if (!parser_.IsInitialState()) {
@@ -410,7 +424,7 @@ PClient::PClient(TcpConnection* obj)
 
 int PClient::HandlePackets(pikiwidb::TcpConnection* obj, const char* start, int size) {
   int total = 0;
-
+  LOG(INFO) << start;
   while (total < size) {
     auto processed = handlePacket(start + total, size - total);
     if (processed <= 0) {
@@ -437,6 +451,9 @@ void PClient::OnConnect() {
     if (g_config.masterauth.empty()) {
       SetAuth();
     }
+  } else if (isJoinCmdTarget()) {
+    SetName("ClusterJoinCmdConnection");
+    PRAFT.SendNodeInfoRequest(this);
   } else {
     if (g_config.password.empty()) {
       SetAuth();
@@ -509,6 +526,10 @@ bool PClient::isPeerMaster() const {
   return repl_addr.GetIP() == PeerIP() && repl_addr.GetPort() == PeerPort();
 }
 
+bool PClient::isJoinCmdTarget() const {
+  return PRAFT.GetJoinCtx().GetPeerIp() == PeerIP() && PRAFT.GetJoinCtx().GetPort() == PeerPort();
+}
+
 int PClient::uniqueID() const {
   if (auto c = getTcpConnection(); c) {
     return c->GetUniqueId();
@@ -673,6 +694,7 @@ void PClient::FeedMonitors(const std::vector<std::string>& params) {
     }
   }
 }
+
 void PClient::SetKey(std::vector<std::string>& names) {
   keys_ = std::move(names);  // use std::move clear copy expense
 }

src/client.h

@@ -48,6 +48,7 @@ class CmdRes {
     kErrOther,
     KIncrByOverFlow,
     kInvalidCursor,
+    kWrongLeader,
   };
 
   CmdRes() = default;
@@ -209,6 +210,8 @@ class PClient : public std::enable_shared_from_this<PClient>, public CmdRes {
   bool isPeerMaster() const;
   int uniqueID() const;
 
+  bool isJoinCmdTarget() const;
+
   // TcpConnection's life is undetermined, so use weak ptr for safety.
   std::weak_ptr<TcpConnection> tcp_connection_;
 

src/cmd_admin.cc

@@ -7,6 +7,8 @@
 
 #include "cmd_admin.h"
 #include "store.h"
+#include "braft/raft.h"
+#include "praft.h"
 
 namespace pikiwidb {
 
@@ -78,4 +80,77 @@ void SelectCmd::DoCmd(PClient* client) {
   client->SetRes(CmdRes::kOK);
 }
 
+InfoCmd::InfoCmd(const std::string& name, int16_t arity) 
+    : BaseCmd(name, arity, kCmdFlagsAdmin | kCmdFlagsReadonly, kAclCategoryAdmin) {}
+
+bool InfoCmd::DoInitial(PClient* client) { return true; }
+
+/*
+* INFO raft
+* Querying Node Information.
+* Reply:
+*   raft_node_id:595100767
+    raft_state:up
+    raft_role:follower
+    raft_is_voting:yes
+    raft_leader_id:1733428433
+    raft_current_term:1
+    raft_num_nodes:2
+    raft_num_voting_nodes:2
+    raft_node1:id=1733428433,state=connected,voting=yes,addr=localhost,port=5001,last_conn_secs=5,conn_errors=0,conn_oks=1
+*/
+// @todo The info raft command is only supported for the time being
+void InfoCmd::DoCmd(PClient* client) {
+  if (client->argv_.size() <= 1) {
+    return client->SetRes(CmdRes::kWrongNum, client->CmdName());
+  }
+
+  auto cmd = client->argv_[1];
+  if (!strcasecmp(cmd.c_str(), "RAFT")) {
+    if (client->argv_.size() != 2) {
+      return client->SetRes(CmdRes::kWrongNum, client->CmdName());
+    }
+
+    if (!PRAFT.IsInitialized()) {
+      return client->SetRes(CmdRes::kErrOther, "don't already cluster member");
+    }
+
+    auto node_status = PRAFT.GetNodeStatus();
+    if (node_status.state == braft::State::STATE_END) {
+      return client->SetRes(CmdRes::kErrOther, "Node is not initialized");
+    }
+
+    std::string message("");
+    message += "raft_group_id:" + PRAFT.GetGroupId() + "\r\n";
+    message += "raft_node_id:" + PRAFT.GetNodeId() + "\r\n";
+    if (braft::is_active_state(node_status.state)) {
+      message += "raft_state:up\r\n";
+    } else {
+      message += "raft_state:down\r\n";
+    }    
+    message += "raft_role:" + std::string(braft::state2str(node_status.state)) + "\r\n";
+    // message += "raft_is_voting:" + node_status.is_voting + "\r\n";
+    message += "raft_leader_id:" + node_status.leader_id.to_string() + "\r\n";
+    message += "raft_current_term:" + std::to_string(node_status.term) + "\r\n";
+    // message += "raft_num_nodes:" + std::to_string(node_status.num_nodes) + "\r\n";
+    // message += "raft_num_voting_nodes:" + std::to_string(node_status.num_voting_nodes) + "\r\n";
+
+    if (PRAFT.IsLeader()) {
+      std::vector<braft::PeerId> peers;
+      auto status = PRAFT.GetListPeers(&peers);
+      if (!status.ok()) {
+        return client->SetRes(CmdRes::kErrOther, status.error_str());
+      }
+
+      for (int i = 0; i < peers.size(); i++) {
+        message += "raft_node" + std::to_string(i) + ":addr=" + butil::ip2str(peers[i].addr.ip).c_str() + ",port=" + std::to_string(peers[i].addr.port) + "\r\n";
+      }
+    }
+
+    client->AppendString(message);
+  } else {
+    client->SetRes(CmdRes::kErrOther, "ERR the cmd is not supported");
+  }
+}
+
 }  // namespace pikiwidb

src/cmd_admin.h

@@ -84,4 +84,15 @@ class SelectCmd : public BaseCmd {
   void DoCmd(PClient* client) override;
 };
 
+class InfoCmd : public BaseCmd {
+  public:
+    InfoCmd(const std::string& name, int16_t arity);
+
+  protected:
+    bool DoInitial(PClient* client) override;
+
+  private:
+    void DoCmd(PClient* client) override;
+};
+
 }  // namespace pikiwidb