The data backup and recovery functions have been added

Author: wuxianrong

.github/workflows/pika.yml

@@ -34,11 +34,7 @@ jobs:
       - name: Install Deps
         run: |
           sudo apt-get update
-          sudo apt-get install -y gcc-11 g++-11 libssl-dev autoconf libprotobuf-dev protobuf-compiler clang-tidy 
-          sudo update-alternatives --install /usr/bin/gcc gcc /usr/bin/gcc-11 100
-          sudo update-alternatives --install /usr/bin/g++ g++ /usr/bin/g++-11 100
-          gcc --version
-          g++ --version
+          sudo apt-get install -y autoconf libprotobuf-dev protobuf-compiler clang-tidy
 
       - name: Configure CMake
         # Configure CMake in a 'build' subdirectory. `CMAKE_BUILD_TYPE` is only required if you are using a single-configuration generator such as make.
@@ -163,7 +159,7 @@ jobs:
       - name: Install deps
         run: |
           dnf update -y
-          dnf install -y bash cmake wget git autoconf gcc perl-Digest-SHA tcl which tar g++ tar epel-release gcc-c++ libstdc++-devel openssl-devel gcc-toolset-11
+          dnf install -y bash cmake wget git autoconf gcc perl-Digest-SHA tcl which tar g++ tar epel-release gcc-c++ libstdc++-devel gcc-toolset-13
 
       - name: Set up Go
         uses: actions/setup-go@v5
@@ -177,7 +173,7 @@ jobs:
 
       - name: Configure CMake
         run: |
-          source /opt/rh/gcc-toolset-11/enable
+          source /opt/rh/gcc-toolset-13/enable
           cmake -B build -DCMAKE_BUILD_TYPE=${{ env.BUILD_TYPE }} -DUSE_PIKA_TOOLS=ON -DCMAKE_CXX_FLAGS_DEBUG=-fsanitize=address .
 
       - uses: actions/cache@v3

include/pika_db.h

@@ -94,6 +94,8 @@ class DB : public std::enable_shared_from_this<DB>, public pstd::noncopyable {
   std::shared_ptr<storage::Storage> storage() const;
   void GetBgSaveMetaData(std::vector<std::string>* fileNames, std::string* snapshot_uuid);
   void BgSaveDB();
+  pstd::Status CreateCheckpoint(const std::string& checkpoint_dir);
+  pstd::Status LoadDBFromCheckpoint(const std::string& checkpoint_dir);
   void SetBinlogIoError();
   void SetBinlogIoErrorrelieve();
   bool IsBinlogIoError();

include/pika_server.h

@@ -71,6 +71,8 @@ enum TaskType {
   kCompactRangeSets,
   kCompactRangeZSets,
   kCompactRangeList,
+  kLoadDBFromCheckpoint,
+  kCreateCheckpoint,
 };
 
 struct TaskArg {

src/pika_db.cc

@@ -67,6 +67,75 @@ void DB::BgSaveDB() {
   g_pika_server->BGSaveTaskSchedule(&DoBgSave, static_cast<void*>(bg_task_arg));
 }
 
+pstd::Status DB::CreateCheckpoint(const std::string& checkpoint_dir) {
+  std::string checkpoint_sub_path = checkpoint_dir;
+  if (!checkpoint_sub_path.empty() && checkpoint_sub_path.back() != '/') {
+    checkpoint_sub_path.push_back('/');
+  }
+  checkpoint_sub_path += db_name_;
+
+  if (!pstd::FileExists(checkpoint_sub_path)) {
+    if (pstd::CreatePath(checkpoint_sub_path, 0755) != 0) {
+      return Status::IOError("Failed to create checkpoint path", checkpoint_sub_path);
+    }
+  }
+
+  std::shared_lock guard(dbs_rw_);
+  auto tasks = storage_->CreateCheckpoint(checkpoint_sub_path);
+  for (auto& task : tasks) {
+    auto status = task.get();
+    if (!status.ok()) {
+      return Status::Corruption("Create checkpoint failed: " + status.ToString());
+    }
+  }
+  return Status::OK();
+}
+
+pstd::Status DB::LoadDBFromCheckpoint(const std::string& checkpoint_dir) {
+  std::string checkpoint_sub_path = checkpoint_dir;
+  if (!checkpoint_sub_path.empty() && checkpoint_sub_path.back() != '/') {
+    checkpoint_sub_path.push_back('/');
+  }
+  checkpoint_sub_path += db_name_;
+
+  if (!pstd::FileExists(checkpoint_sub_path)) {
+    return Status::NotFound("Checkpoint dir does not exist: " + checkpoint_sub_path);
+  }
+
+  std::lock_guard<std::shared_mutex> guard(dbs_rw_);
+  opened_ = false;
+
+  auto old_storage = storage_;
+  storage_.reset();
+  if (old_storage) {
+    old_storage->Close();
+  }
+
+  storage_ = std::make_shared<storage::Storage>();
+  auto checkpoint_tasks = storage_->LoadCheckpoint(checkpoint_sub_path, db_path_);
+  for (auto& task : checkpoint_tasks) {
+    auto status = task.get();
+    if (!status.ok()) {
+      storage_.reset();
+      return Status::Corruption("Load checkpoint failed: " + status.ToString());
+    }
+  }
+
+  storage::StorageOptions storage_options = g_pika_server->storage_options();
+  auto open_status = storage_->Open(storage_options, db_path_);
+  if (!open_status.ok()) {
+    storage_.reset();
+    return Status::Corruption("Storage open failed: " + open_status.ToString());
+  }
+
+  if (!g_pika_conf->raft_enabled()) {
+    storage_->DisableWal(false);
+  }
+
+  opened_ = true;
+  return Status::OK();
+}
+
 void DB::SetBinlogIoError() { return binlog_io_error_.store(true); }
 void DB::SetBinlogIoErrorrelieve() { return binlog_io_error_.store(false); }
 bool DB::IsBinlogIoError() { return binlog_io_error_.load(); }

src/pika_server.cc

@@ -497,6 +497,37 @@ Status PikaServer::DoSameThingSpecificDB(const std::set<std::string>& dbs, const
       case TaskType::kCompactRangeList:
         db_item.second->CompactRange(storage::DataType::kLists, arg.argv[0], arg.argv[1]);
         break;
+      case TaskType::kLoadDBFromCheckpoint: {
+        // arg.argv[0] should contain checkpoint_path
+        if (arg.argv.empty()) {
+          LOG(ERROR) << "LoadDBFromCheckpoint requires checkpoint_path argument";
+          return Status::InvalidArgument("Missing checkpoint_path");
+        }
+        std::string checkpoint_path = arg.argv[0];
+        auto s = db_item.second->LoadDBFromCheckpoint(checkpoint_path);
+        if (!s.ok()) {
+          LOG(ERROR) << "Failed to load DB from checkpoint: " << s.ToString();
+          return s;
+        }
+        LOG(INFO) << "Successfully loaded DB " << db_item.first << " from checkpoint: " << checkpoint_path;
+        break;
+      }
+      case TaskType::kCreateCheckpoint: {
+        // arg.argv[0] should contain checkpoint_path
+        if (arg.argv.empty()) {
+          LOG(ERROR) << "CreateCheckpoint requires checkpoint_path argument";
+          return Status::InvalidArgument("Missing checkpoint_path");
+        }
+        std::string checkpoint_path = arg.argv[0];
+        auto s = db_item.second->CreateCheckpoint(checkpoint_path);
+        if (!s.ok()) {
+          LOG(ERROR) << "Failed to create checkpoint: " << s.ToString();
+          return s;
+        }
+        LOG(INFO) << "Successfully created checkpoint for DB " << db_item.first << " at: " << checkpoint_path;
+        break;
+      }
+      
       default:
         break;
     }

src/praft/include/praft/praft.h

@@ -18,6 +18,7 @@
 #include "braft/raft.h"
 #include "braft/storage.h"
 #include "braft/util.h"
+#include "braft/file_system_adaptor.h"
 #include "pstd/include/pstd_mutex.h"
 #include "pstd/include/pstd_status.h"
 #include "rocksdb/status.h"
@@ -31,6 +32,7 @@ namespace storage {
 class Storage;
 }
 
+// Forward declarations
 namespace pikiwidb {
 class Binlog;
 }
@@ -74,6 +76,8 @@ class PikaStateMachine : public braft::StateMachine {
   PikaStateMachine();
   ~PikaStateMachine() override = default;
 
+  void SetLeaderTerm(std::atomic<int64_t>* leader_term);
+
   // Apply committed log entry
   void on_apply(braft::Iterator& iter) override;
 
@@ -97,7 +101,8 @@ class PikaStateMachine : public braft::StateMachine {
   void on_stop_following(const ::braft::LeaderChangeContext& ctx) override;
 
  private:
-
+  std::atomic<bool> is_node_first_start_up_{true};  // 标记节点是否首次启动
+  std::atomic<int64_t>* leader_term_{nullptr};
 };
 
 // Raft node wrapper
@@ -130,6 +135,11 @@ class PikaRaftNode {
   // Get cluster status information
   void GetStatus(std::string* status_str);
 
+  void GetLeaderLeaseStatus(braft::LeaderLeaseStatus* status) const;
+  
+  // Trigger a snapshot creation
+  pstd::Status DoSnapshot(int64_t self_snapshot_index = 0, bool is_sync = true);
+
   braft::Node* GetRaftNode() { return node_.get(); }
 
  private:
@@ -144,6 +154,10 @@ class PikaRaftNode {
   std::string raft_log_uri_;
   std::string raft_meta_uri_;
   std::string raft_snapshot_uri_;
+  
+  // Snapshot adaptor
+  scoped_refptr<braft::FileSystemAdaptor> snapshot_adaptor_;
+  std::atomic<int64_t> leader_term_{-1};
 };
 
 // Raft cluster manager

src/praft/include/praft/psnapshot.h

@@ -0,0 +1,34 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <string>
+
+#include "braft/file_system_adaptor.h"
+#include "braft/macros.h"
+#include "braft/snapshot.h"
+
+#define PRAFT_SNAPSHOT_META_FILE "__raft_snapshot_meta"
+#define PRAFT_SNAPSHOT_PATH "snapshot/snapshot_"
+#define IS_RDONLY 0x01
+
+// 自定义文件系统适配器，用于Braft快照生成
+class PPosixFileSystemAdaptor : public braft::PosixFileSystemAdaptor {
+public:
+  PPosixFileSystemAdaptor() {}
+  ~PPosixFileSystemAdaptor() {}
+
+  braft::FileAdaptor* open(const std::string& path, int oflag, const ::google::protobuf::Message* file_meta,
+                           butil::File::Error* e) override;
+  
+  void AddAllFiles(const std::string& dir, braft::LocalSnapshotMetaTable* snapshot_meta_memtable,
+                   const std::string& base_path);
+
+private:
+  braft::raft_mutex_t mutex_;
+};