feat: distinguish between normal node startup and snapshot loading

cmake/braft.cmake

@@ -16,8 +16,11 @@ ExternalProject_Add(
         extern_braft
         ${EXTERNAL_PROJECT_LOG_ARGS}
         DEPENDS brpc
-        GIT_REPOSITORY "https://github.com/pikiwidb/braft.git"
-        GIT_TAG master
+        # The pr on braft is not merged, so I am using my own warehouse to run the test for the time being
+        GIT_REPOSITORY https://github.com/panlei-coder/braft.git
+        GIT_TAG merge-master-playback
+        # GIT_REPOSITORY "https://github.com/pikiwidb/braft.git"
+        # GIT_TAG master
         GIT_SHALLOW true
         PREFIX ${BRAFT_SOURCES_DIR}
         UPDATE_COMMAND ""

cmake/openssl.cmake

@@ -4,7 +4,7 @@
 # of patent rights can be found in the PATENTS file in the same directory.
 
 INCLUDE(ExternalProject)
-
+    
 SET(OPENSSL_SOURCE_DIR ${THIRD_PARTY_PATH}/openssl)
 SET(OPENSSL_INSTALL_DIR ${THIRD_PARTY_PATH}/install/openssl)
 SET(OPENSSL_INCLUDE_DIR ${OPENSSL_INSTALL_DIR}/include)
@@ -49,4 +49,4 @@ ADD_DEPENDENCIES(crypto OpenSSL)
 SET(OPENSSL_CRYPTO_LIBRARY ${OPENSSL_INSTALL_DIR}/${OPENSSL_LIB}/libcrypto.a)
 
 SET(OPENSSL_INCLUDE_DIR ${THIRD_PARTY_PATH}/install/openssl/include)
-SET(OPENSSL_ROOT_DIR ${THIRD_PARTY_PATH}/install/openssl/bin/openssl)
+SET(OPENSSL_ROOT_DIR ${THIRD_PARTY_PATH}/install/openssl/bin/openssl)

pikiwidb.conf

@@ -319,7 +319,7 @@ slowlog-max-len 128
 # to the same DB is distributed among several RocksDB instances.
 
 # RocksDB instances number per DB
-db-instance-num 3
+db-instance-num 1
 # default is 86400 * 7
 small-compaction-threshold 604800
 # default is 86400 * 3

save_load.sh

@@ -1,4 +1,5 @@
 #!/bin/bash
+
 killall -9 pikiwidb
 mkdir leader follower1
 
@@ -10,9 +11,8 @@ redis-cli -p 7777 raft.cluster init
 
 redis-benchmark -p 7777 -c 5 -n 10000 -r 10000 -d 1024 -t hset
 redis-cli -p 7777 raft.node dosnapshot
+redis-benchmark -p 7777 -c 5 -n 10000 -r 10000 -d 1024 -t hset
 redis-cli -p 7777 raft.node dosnapshot
-
-sleep 10
-
+redis-benchmark -p 7777 -c 5 -n 10000 -r 10000 -d 1024 -t hset
 
 redis-cli -p 8888 raft.cluster join 127.0.0.1:7777

src/CMakeLists.txt

@@ -31,7 +31,6 @@ TARGET_INCLUDE_DIRECTORIES(pikiwidb
         PRIVATE ${BRPC_INCLUDE_DIR}
         )
 
-
-TARGET_LINK_LIBRARIES(pikiwidb net; dl; fmt; storage; rocksdb; pstd braft brpc ssl crypto zlib protobuf leveldb gflags  z praft praft_pb "${LIB}")
+TARGET_LINK_LIBRARIES(pikiwidb net; dl; fmt; storage; rocksdb; pstd braft brpc ssl crypto zlib protobuf leveldb gflags z praft praft_pb "${LIB}")
 
 SET_TARGET_PROPERTIES(pikiwidb PROPERTIES LINKER_LANGUAGE CXX)

src/cmd_raft.cc

@@ -122,7 +122,9 @@ void RaftNodeCmd::DoCmdRemove(PClient* client) {
 }
 
 void RaftNodeCmd::DoCmdSnapshot(PClient* client) {
-  auto s = PRAFT.DoSnapshot();
+  // @todo need to get self_snapshot_index
+  // auto self_snapshot_index = PSTORE.GetBackend(client->GetCurrentDB())->GetStorage()->GetSmallestFlushedLogIndex();
+  auto s = PRAFT.DoSnapshot();  // self_snapshot_index
   if (s.ok()) {
     client->SetRes(CmdRes::kOK);
   }

src/db.cc

@@ -37,8 +37,10 @@ rocksdb::Status DB::Open() {
     storage_options.append_log_function = [&r = PRAFT](const Binlog& log, std::promise<rocksdb::Status>&& promise) {
       r.AppendLog(log, std::move(promise));
     };
-    storage_options.do_snapshot_function =
-        std::bind(&pikiwidb::PRaft::DoSnapshot, &pikiwidb::PRAFT, std::placeholders::_1, std::placeholders::_2);
+    storage_options.do_snapshot_function = [raft = &pikiwidb::PRAFT](auto&& self_snapshot_index, auto&& is_sync) {
+      raft->DoSnapshot(std::forward<decltype(self_snapshot_index)>(self_snapshot_index),
+                       std::forward<decltype(is_sync)>(is_sync));
+    };
   }
 
   storage_options.db_instance_num = g_config.db_instance_num.load();

src/praft/praft.cc

@@ -9,6 +9,7 @@
 
 #include <cassert>
 
+#include "braft/raft.h"
 #include "braft/snapshot.h"
 #include "braft/util.h"
 #include "brpc/server.h"
@@ -150,7 +151,7 @@ butil::Status PRaft::Init(std::string& group_id, bool initial_conf_is_null) {
   node_options_.fsm = this;
   node_options_.node_owns_fsm = false;
   node_options_.snapshot_interval_s = 0;
-  std::string prefix = "local://" + g_config.db_path.ToString() + "_praft";
+  std::string prefix = "local://" + g_config.db_path.ToString() + std::to_string(db_id_) + "/_praft";
   node_options_.log_uri = prefix + "/log";
   node_options_.raft_meta_uri = prefix + "/raft_meta";
   node_options_.snapshot_uri = prefix + "/snapshot";
@@ -241,6 +242,24 @@ butil::Status PRaft::GetListPeers(std::vector<braft::PeerId>* peers) {
   return node_->list_peers(peers);
 }
 
+uint64_t PRaft::GetTerm(uint64_t log_index) {
+  if (!node_) {
+    ERROR("Node is not initialized");
+    return 0;
+  } else {
+    return node_->get_term(log_index);
+  }
+}
+
+uint64_t PRaft::GetLastLogIndex(bool is_flush) {
+  if (!node_) {
+    ERROR("Node is not initialized");
+    return 0;
+  } else {
+    return node_->get_last_log_index(is_flush);
+  }
+}
+
 void PRaft::SendNodeRequest(PClient* client) {
   assert(client);
 
@@ -521,10 +540,17 @@ butil::Status PRaft::DoSnapshot(int64_t self_snapshot_index, bool is_sync) {
   if (!node_) {
     return ERROR_LOG_AND_STATUS("Node is not initialized");
   }
-  braft::SynchronizedClosure done;
-  node_->snapshot(&done, self_snapshot_index);
-  done.wait();
-  return done.status();
+
+  if (is_sync) {
+    braft::SynchronizedClosure done;
+    node_->snapshot(&done, self_snapshot_index);
+    done.wait();
+    return done.status();
+  } else {
+    node_->snapshot(nullptr, self_snapshot_index);
+    butil::Status status;
+    return status;
+  }
 }
 
 void PRaft::OnClusterCmdConnectionFailed([[maybe_unused]] EventLoop* loop, const char* peer_ip, int port) {
@@ -629,10 +655,38 @@ void PRaft::on_snapshot_save(braft::SnapshotWriter* writer, braft::Closure* done
 int PRaft::on_snapshot_load(braft::SnapshotReader* reader) {
   CHECK(!IsLeader()) << "Leader is not supposed to load snapshot";
   assert(reader);
+
+  if (is_node_first_start_up_) {
+    // get replay point of one db's
+    /*
+    1. When a node starts normally, because all memory data is flushed to disks and
+       snapshots are truncated to the latest, the flush-index and apply-index are the
+       same when the node starts, so the maximum log index should be obtained.
+    2. When a node is improperly shut down and restarted, the minimum flush-index should
+       be obtained as the starting point for fault recovery.
+    */
+    // @todo GetSmallestFlushedLogIndex
+    uint64_t replay_point = PSTORE.GetBackend(db_id_)->GetStorage()->GetSmallestFlushedLogIndex();
+    node_->set_self_playback_point(replay_point);
+    is_node_first_start_up_ = false;
+    INFO("set replay_point: {}", replay_point);
+
+    /*
+    If a node has just joined the cluster and does not have any data,
+    it does not load the local snapshot at startup. Therefore,
+    LoadDBFromCheckPoint is required after loading the snapshot from the leader.
+    */
+    if (GetLastLogIndex() != 0) {
+      return 0;
+    }
+  }
+
+  // 3. When a snapshot is installed on a node, you do not need to set a playback point.
   auto reader_path = reader->get_path();                             // xx/snapshot_0000001
   auto path = g_config.db_path.ToString() + std::to_string(db_id_);  // db/db_id
   TasksVector tasks(1, {TaskType::kLoadDBFromCheckpoint, db_id_, {{TaskArg::kCheckpointPath, reader_path}}, true});
   PSTORE.HandleTaskSpecificDB(tasks);
+  INFO("load snapshot success!");
   return 0;
 }
 

src/praft/praft.h

@@ -137,6 +137,8 @@ class PRaft : public braft::StateMachine {
   std::string GetGroupID() const;
   braft::NodeStatus GetNodeStatus() const;
   butil::Status GetListPeers(std::vector<braft::PeerId>* peers);
+  uint64_t GetTerm(uint64_t log_index);
+  uint64_t GetLastLogIndex(bool is_flush = false);
 
   bool IsInitialized() const { return node_ != nullptr && server_ != nullptr; }
 
@@ -164,6 +166,8 @@ class PRaft : public braft::StateMachine {
   ClusterCmdContext cluster_cmd_ctx_;  // context for cluster join/remove command
   std::string group_id_;               // group id
   int db_id_ = 0;                      // db_id
+
+  bool is_node_first_start_up_ = true;
 };
 
 }  // namespace pikiwidb

src/praft/psnapshot.cc

@@ -11,33 +11,44 @@
 #include "psnapshot.h"
 
 #include "braft/local_file_meta.pb.h"
+#include "braft/snapshot.h"
 #include "butil/files/file_path.h"
 
 #include "pstd/log.h"
+#include "pstd/pstd_string.h"
 
 #include "config.h"
+#include "praft.h"
 #include "store.h"
 
 namespace pikiwidb {
 
-extern PConfig g_config;
-
 braft::FileAdaptor* PPosixFileSystemAdaptor::open(const std::string& path, int oflag,
                                                   const ::google::protobuf::Message* file_meta, butil::File::Error* e) {
   if ((oflag & IS_RDONLY) == 0) {  // This is a read operation
     bool snapshots_exists = false;
     std::string snapshot_path;
+    int db_id = -1;
 
     // parse snapshot path
     butil::FilePath parse_snapshot_path(path);
     std::vector<std::string> components;
+    bool is_find_db = false;
     parse_snapshot_path.GetComponents(&components);
-    for (auto component : components) {
+    for (const auto& component : components) {
       snapshot_path += component + "/";
+
+      if (is_find_db && pstd::String2int(component, &db_id)) {
+        is_find_db = false;
+      }
+
       if (component.find("snapshot_") != std::string::npos) {
         break;
+      } else if (component == "db") {
+        is_find_db = true;
       }
     }
+
     // check whether snapshots have been created
     std::lock_guard<braft::raft_mutex_t> guard(mutex_);
     if (!snapshot_path.empty()) {
@@ -55,6 +66,8 @@ braft::FileAdaptor* PPosixFileSystemAdaptor::open(const std::string& path, int o
 
     // Snapshot generation
     if (!snapshots_exists) {
+      assert(db_id >= 0);
+
       braft::LocalSnapshotMetaTable snapshot_meta_memtable;
       std::string meta_path = snapshot_path + "/" PRAFT_SNAPSHOT_META_FILE;
       INFO("start to generate snapshot in path {}", snapshot_path);
@@ -66,6 +79,14 @@ braft::FileAdaptor* PPosixFileSystemAdaptor::open(const std::string& path, int o
       PSTORE.HandleTaskSpecificDB(tasks);
       AddAllFiles(snapshot_path, &snapshot_meta_memtable, snapshot_path);
 
+      // update snapshot last log index and last_log_term
+      auto& new_meta = const_cast<braft::SnapshotMeta&>(snapshot_meta_memtable.meta());
+      auto last_log_index = 30000;  // @todo PSTORE.GetBackend(db_id)->GetStorage()->GetSmallestFlushedLogIndex();
+      new_meta.set_last_included_index(last_log_index);
+      auto last_log_term = PRAFT.GetTerm(last_log_index);
+      new_meta.set_last_included_term(last_log_term);
+      INFO("Succeed to fix snapshot meta: {}, {}", last_log_index, last_log_term);
+
       auto rc = snapshot_meta_memtable.save_to_file(fs, meta_path);
       if (rc == 0) {
         INFO("Succeed to save snapshot in path {}", snapshot_path);

src/storage/include/storage/storage.h

@@ -1123,6 +1123,8 @@ class Storage {
   void GetRocksDBInfo(std::string& info);
   Status OnBinlogWrite(const pikiwidb::Binlog& log, LogIndex log_idx);
 
+  LogIndex GetSmallestFlushedLogIndex() const;
+
  private:
   std::vector<std::unique_ptr<Redis>> insts_;
   std::unique_ptr<SlotIndexer> slot_indexer_;

src/storage/src/log_index.cc

@@ -7,6 +7,8 @@
 
 #include "log_index.h"
 
+#include "pstd/log.h"
+
 #include <algorithm>
 #include <cinttypes>
 #include <set>
@@ -167,6 +169,7 @@ void LogIndexAndSequenceCollectorPurger::OnFlushCompleted(rocksdb::DB *db,
   auto count = count_.fetch_add(1);
 
   if (count % 10 == 0) {
+    INFO("do snapshot after flush: {}", smallest_flushed_log_index);
     callback_(smallest_flushed_log_index, false);
   }
 

src/storage/src/redis.h

@@ -110,6 +110,9 @@ class Redis {
   void UpdateAppliedLogIndexOfColumnFamily(size_t cf_idx, LogIndex logidx, SequenceNumber seqno) {
     log_index_of_all_cfs_.Update(cf_idx, logidx, seqno);
   }
+  LogIndex GetSmallestFlushedLogIndex() const {
+    return log_index_of_all_cfs_.GetSmallestLogIndex(-1).smallest_flushed_log_index;
+  }
   bool IsRestarting() const { return is_starting_; }
   void StartingPhaseEnd() { is_starting_ = false; }
 

src/storage/src/storage.cc

@@ -2517,4 +2517,13 @@ Status Storage::OnBinlogWrite(const pikiwidb::Binlog& log, LogIndex log_idx) {
   return s;
 }
 
+LogIndex Storage::GetSmallestFlushedLogIndex() const {
+  LogIndex smallest_flushed_log_index = INT64_MAX;
+  for (auto& inst : insts_) {
+    smallest_flushed_log_index = std::min(smallest_flushed_log_index, inst->GetSmallestFlushedLogIndex());
+  }
+
+  return smallest_flushed_log_index;
+}
+
 }  //  namespace storage