Add the mapping relationship between LogIndex and RocksDB SequnceNumber

Author: wuxianrong

CMakeLists.txt

@@ -559,9 +559,9 @@ ExternalProject_Add(protobuf
   DEPENDS
   zlib
   URL
-  https://github.com/protocolbuffers/protobuf/releases/download/v3.17.3/protobuf-cpp-3.17.3.tar.gz
+  https://github.com/protocolbuffers/protobuf/releases/download/v21.12/protobuf-cpp-3.21.12.tar.gz
   URL_HASH
-  MD5=3fe4c2647e0991c014a386a896d0a116
+  MD5=6b4fd9cee2fa63834f29c7d433679855
   DOWNLOAD_NO_PROGRESS
   1
   UPDATE_COMMAND
@@ -572,6 +572,8 @@ ExternalProject_Add(protobuf
   1
   LOG_INSTALL
   1
+  LOG_OUTPUT_ON_FAILURE
+  ON
   SOURCE_SUBDIR
   cmake
   CMAKE_ARGS
@@ -655,9 +657,9 @@ ExternalProject_Add(brpc
   snappy
   zlib
   URL
-  https://github.com/apache/brpc/archive/refs/tags/1.6.0.tar.gz
+  https://github.com/apache/brpc/archive/refs/tags/1.9.0.tar.gz
   URL_HASH
-  MD5=0d37cea25bd006e89806f461ef7e39ba
+  MD5=a2b626d96a5b017f2a6701ffa594530c
   DOWNLOAD_NO_PROGRESS
   1
   UPDATE_COMMAND
@@ -668,6 +670,8 @@ ExternalProject_Add(brpc
   1
   LOG_INSTALL
   1
+  LOG_OUTPUT_ON_FAILURE
+  ON
   CMAKE_ARGS
   -DCMAKE_POLICY_VERSION_MINIMUM=3.5
   -DCMAKE_INSTALL_PREFIX=${STAGED_INSTALL_PREFIX}
@@ -713,6 +717,8 @@ ExternalProject_Add(braft
   1
   LOG_INSTALL
   1
+  LOG_OUTPUT_ON_FAILURE
+  ON
   CMAKE_ARGS
   -DCMAKE_POLICY_VERSION_MINIMUM=3.5
   -DCMAKE_INSTALL_PREFIX=${STAGED_INSTALL_PREFIX}

src/praft/include/praft/praft.h

@@ -97,8 +97,7 @@ class PikaStateMachine : public braft::StateMachine {
   void on_stop_following(const ::braft::LeaderChangeContext& ctx) override;
 
  private:
-  std::atomic<int64_t> applied_index_;
-  std::atomic<int64_t> leader_term_;
+
 };
 
 // Raft node wrapper
@@ -185,8 +184,8 @@ class RaftManager {
   std::shared_ptr<PikaRaftNode> GetRaftNode(const std::string& db_name);
 
   // Apply binlog entry to storage (public for PikaStateMachine to call)
-  rocksdb::Status ApplyBinlogEntry(const ::pikiwidb::Binlog& binlog);
-
+  rocksdb::Status ApplyBinlogEntry(const ::pikiwidb::Binlog& binlog, uint64_t log_index = 0);
+  
  private:
   std::atomic<bool> initialized_;
   std::atomic<bool> running_;
@@ -208,4 +207,3 @@ class RaftManager {
 }  // namespace pika_raft
 
 #endif  // PRAFT_PRAFT_H_
-

src/praft/src/praft.cc

@@ -29,9 +29,7 @@ extern std::unique_ptr<PikaServer> g_pika_server;
 namespace pika_raft {
 
 // PikaStateMachine implementation
-PikaStateMachine::PikaStateMachine() 
-    : applied_index_(0), leader_term_(-1) {
-}
+PikaStateMachine::PikaStateMachine() {} 
 
 void PikaStateMachine::on_apply(braft::Iterator& iter) {
   for (; iter.valid(); iter.next()) {
@@ -41,7 +39,6 @@ void PikaStateMachine::on_apply(braft::Iterator& iter) {
     int64_t index = iter.index();
 
     if (!g_pika_server || !g_pika_server->GetRaftManager()) {
-      applied_index_.store(index, std::memory_order_relaxed);
       // Run closure asynchronously in bthread to avoid blocking on_apply
       if (done) {
         braft::run_closure_in_bthread(done_guard.release());
@@ -55,14 +52,14 @@ void PikaStateMachine::on_apply(braft::Iterator& iter) {
       if (done) {
         done->status().set_error(EINVAL, "Failed to parse binlog");
       }
-      applied_index_.store(index, std::memory_order_relaxed);
       if (done) {
         braft::run_closure_in_bthread(done_guard.release());
       }
       continue;
     }
 
-    rocksdb::Status apply_status = g_pika_server->GetRaftManager()->ApplyBinlogEntry(binlog);
+    // Apply binlog with log index for tracking
+    rocksdb::Status apply_status = g_pika_server->GetRaftManager()->ApplyBinlogEntry(binlog, index);
 
     if (done) {
       if (apply_status.ok()) {
@@ -73,7 +70,6 @@ void PikaStateMachine::on_apply(braft::Iterator& iter) {
       }
     }
 
-    applied_index_.store(index, std::memory_order_relaxed);
 
     // Run closure asynchronously in bthread to avoid blocking on_apply
     if (done) {
@@ -94,19 +90,16 @@ int PikaStateMachine::on_snapshot_load(braft::SnapshotReader* reader) {
     in >> index;
     in.close();
 
-    applied_index_.store(index, std::memory_order_relaxed);
     return 0;
   }
 
   return -1;
 }
 
 void PikaStateMachine::on_leader_start(int64_t term) {
-  leader_term_.store(term, std::memory_order_relaxed);
 }
 
 void PikaStateMachine::on_leader_stop(const butil::Status& status) {
-  leader_term_.store(-1, std::memory_order_relaxed);
 }
 
 void PikaStateMachine::on_error(const ::braft::Error& e) {
@@ -465,6 +458,7 @@ pstd::Status RaftManager::RemoveNode(const std::string& db_name,
   }
 
   braft::PeerId peer_id = ParsePeerId(peer_addr);
+    
   if (peer_id.is_empty()) {
     return pstd::Status::Corruption("Invalid peer address: " + peer_addr);
   }
@@ -475,6 +469,7 @@ pstd::Status RaftManager::RemoveNode(const std::string& db_name,
 pstd::Status RaftManager::GetClusterInfo(const std::string& db_name, 
                                          std::string* info) {
   auto node = GetRaftNode(db_name);
+
   if (!node) {
     return pstd::Status::Corruption("Raft node not found for DB: " + db_name);
   }
@@ -501,14 +496,39 @@ pstd::Status RaftManager::CreateRaftNode(const std::string& db_name,
     return pstd::Status::Corruption("Raft node already exists for DB: " + db_name);
   }
 
-  // Create peer ID for this node
-  // Use localhost since Pika doesn't expose a host() method
-  std::string addr = "127.0.0.1:" + std::to_string(g_pika_conf->port() + 3000);
-  braft::PeerId peer_id = ParsePeerId(addr);
-  if (peer_id.is_empty()) {
-    return pstd::Status::Corruption("Failed to create peer ID");
+  // Determine the Raft port for this node
+  // Raft uses Pika port + 3000
+  int raft_port = g_pika_conf->port() + 3000;
+  
+  // Find the peer address from the peers list that matches our Raft port
+  // This allows the user to specify the exact address in RAFT.CLUSTER INIT command
+  braft::PeerId peer_id;
+  bool found = false;
+  
+  for (const auto& peer : peers) {
+    if (peer.addr.port == raft_port) {
+      peer_id = peer;
+      found = true;
+      LOG(INFO) << "Found matching peer address in cluster config: " << peer.to_string();
+      break;
+    }
   }
 
+  // If no matching peer found, return error
+  if (!found) {
+    std::string error_msg = "No matching peer address found in cluster config for Raft port " + 
+                           std::to_string(raft_port) + 
+                           ". Please include this node's address (with port " + 
+                           std::to_string(raft_port) + 
+                           ") in the RAFT.CLUSTER INIT command. " +
+                           "Example: RAFT.CLUSTER INIT <ip1>:" + std::to_string(raft_port) + 
+                           ",<ip2>:<port2>,...";
+    LOG(ERROR) << error_msg;
+    return pstd::Status::Corruption(error_msg);
+  }
+  
+  LOG(INFO) << "Creating Raft node for DB: " << db_name << " with address: " << peer_id.to_string();
+  
   // Create Raft node
   auto node = std::make_shared<PikaRaftNode>(group_id_ + "_" + db_name, peer_id);
 
@@ -625,7 +645,7 @@ void RaftManager::AppendLog(const std::string& db_name,
   node->GetRaftNode()->apply(task);
 }
 
-rocksdb::Status RaftManager::ApplyBinlogEntry(const ::pikiwidb::Binlog& binlog) {
+rocksdb::Status RaftManager::ApplyBinlogEntry(const ::pikiwidb::Binlog& binlog, uint64_t log_index) {
   std::string db_name = "db0";
 
   auto db = g_pika_server->GetDB(db_name);
@@ -640,14 +660,15 @@ rocksdb::Status RaftManager::ApplyBinlogEntry(const ::pikiwidb::Binlog& binlog)
     return rocksdb::Status::InvalidArgument("Storage is null");
   }
 
-  auto status = storage->OnBinlogWrite(binlog, 0);
+  // Pass log_index to storage layer for tracking
+  auto status = storage->OnBinlogWrite(binlog, log_index);
 
   if (!status.ok()) {
-    LOG(ERROR) << "Failed to apply binlog to " << db_name << ": " << status.ToString();
+    LOG(ERROR) << "Failed to apply binlog to " << db_name << " at log_index " << log_index 
+               << ": " << status.ToString();
   }
 
   return status;
 }
 
 }  // namespace pika_raft
-

src/storage/include/storage/storage.h

@@ -99,6 +99,8 @@ struct StorageOptions {
   std::function<void(const ::pikiwidb::Binlog&, std::promise<rocksdb::Status>&&,
                      CommitCallback)> append_log_function;
 
+  std::function<void(int64_t, bool)> do_snapshot_function;
+  
   Status ResetOptions(const OptionType& option_type, const std::unordered_map<std::string, std::string>& options_map);
 };
 
@@ -1159,6 +1161,8 @@ class Storage {
 
   rocksdb::Status OnBinlogWrite(const ::pikiwidb::Binlog& binlog, uint64_t log_index);
 
+  uint64_t GetSmallestFlushedLogIndex();
+  
  private:
   std::unique_ptr<RedisStrings> strings_db_;
   std::unique_ptr<RedisHashes> hashes_db_;

src/storage/include/storage/storage_define.h

@@ -0,0 +1,17 @@
+//  Copyright (c) 2024-present, Qihoo, Inc.  All rights reserved.
+//  This source code is licensed under the BSD-style license found in the
+//  LICENSE file in the root directory of this source tree. An additional grant
+//  of patent rights can be found in the PATENTS file in the same directory.
+
+#ifndef STORAGE_STORAGE_DEFINE_H_
+#define STORAGE_STORAGE_DEFINE_H_
+
+#include <cstdint>
+
+namespace storage {
+
+using LogIndex = int64_t;
+
+}  // namespace storage
+
+#endif  // STORAGE_STORAGE_DEFINE_H_