fix: add replication-id to implement master/slave Replication (OpenAtomFoundation#1951)

* Add replication-id to implement master/slave Replication

---------

Co-authored-by: wuxianrong <wuxianrong@360.cn>

Author: Mixficsol

conf/pika.conf

@@ -413,4 +413,8 @@ default-slot-num : 1024
 
 # Rsync Rate limiting configuration
 throttle-bytes-per-second : 307200000
-max-rsync-parallel-num : 4
+max-rsync-parallel-num : 4
+
+# The synchronization mode of Pika primary/secondary replication is determined by ReplicationID. ReplicationID in one replication_cluster are the same
+# replication-id :
+

docs/ops/APIDifference.md

@@ -125,4 +125,7 @@ slaveof命令允许通过指定write2file(binlog)的文件名称及同步位置
 类似于pkhscanrange, 逆序
 
 ### diskrecovery 
-Pika 原创命令，功能为当磁盘意外写满后，RocksDB 会进入写保护状态，当我们将空间调整为充足空间时，这个命令可以将 RocksDB 的写保护状态解除，变为可以继续写的状态, 避免了 Pika 因为磁盘写满后需要重启才能恢复写的情况，执行成功时返回 OK，如果当前磁盘空间依然不足，执行这个命令返回`"The available disk capacity is insufficient`，该命令执行时不需要额外参数，只需要执行 diskrecovery 即可。
+Pika 原创命令，功能为当磁盘意外写满后，RocksDB 会进入写保护状态，当我们将空间调整为充足空间时，这个命令可以将 RocksDB 的写保护状态解除，变为可以继续写的状态, 避免了 Pika 因为磁盘写满后需要重启才能恢复写的情况，执行成功时返回 OK，如果当前磁盘空间依然不足，执行这个命令返回`"The available disk capacity is insufficient`，该命令执行时不需要额外参数，只需要执行 diskrecovery 即可。
+
+### clearreplicationid
+Pika 原创命令，功能是清空 Pika 实例的 replicationid 的值并持久化到配置文件中

include/pika_admin.h

@@ -288,6 +288,7 @@ class ConfigCmd : public Cmd {
   void ConfigSet(std::string& ret);
   void ConfigRewrite(std::string& ret);
   void ConfigResetstat(std::string& ret);
+  void ConfigRewriteReplicationID(std::string& ret);
 };
 
 class MonitorCmd : public Cmd {
@@ -462,6 +463,18 @@ class DiskRecoveryCmd : public Cmd {
   std::map<std::string, uint64_t> background_errors_;
 };
 
+class ClearReplicationIDCmd : public Cmd {
+ public:
+  ClearReplicationIDCmd(const std::string& name, int arity, uint16_t flag) : Cmd(name, arity, flag) {}
+  void Do(std::shared_ptr<Slot> slot = nullptr) override;
+  void Split(std::shared_ptr<Slot> slot, const HintKeys& hint_keys) override{};
+  void Merge() override{};
+  Cmd* Clone() override { return new ClearReplicationIDCmd(*this); }
+
+ private:
+  void DoInitial() override;
+};
+
 #ifdef WITH_COMMAND_DOCS
 class CommandCmd : public Cmd {
  public:

include/pika_command.h

@@ -51,6 +51,7 @@ const std::string kCmdNameQuit = "quit";
 const std::string kCmdNameHello = "hello";
 const std::string kCmdNameCommand = "command";
 const std::string kCmdNameDiskRecovery = "diskrecovery";
+const std::string kCmdNameClearReplicationID = "clearreplicationid";
 
 // Migrate slot
 const std::string kCmdNameSlotsMgrtSlot = "slotsmgrtslot";

include/pika_conf.h

@@ -22,6 +22,7 @@
 #define kBinlogReadWinDefaultSize 9000
 #define kBinlogReadWinMaxSize 90000
 const uint32_t configRunIDSize = 40;
+const uint32_t configReplicationIDSize = 50;
 
 // global class, class members well initialized
 class PikaConf : public pstd::BaseConf {
@@ -150,6 +151,10 @@ class PikaConf : public pstd::BaseConf {
     std::shared_lock l(rwlock_);
     return master_run_id_;
   }
+  std::string replication_id() {
+    std::shared_lock l(rwlock_);
+    return replication_id_;
+  }
   std::string requirepass() {
     std::shared_lock l(rwlock_);
     return requirepass_;
@@ -374,6 +379,11 @@ class PikaConf : public pstd::BaseConf {
     TryPushDiffCommands("master-run-id", value);
     master_run_id_ = value;
   }
+  void SetReplicationID(const std::string& value) {
+    std::lock_guard l(rwlock_);
+    TryPushDiffCommands("replication-id", value);
+    replication_id_ = value;
+  }
   void SetSlavePriority(const int value) {
     std::lock_guard l(rwlock_);
     TryPushDiffCommands("slave-priority", std::to_string(value));
@@ -577,6 +587,7 @@ class PikaConf : public pstd::BaseConf {
 
   int Load();
   int ConfigRewrite();
+  int ConfigRewriteReplicationID();
 
  private:
   pstd::Status InternalGetTargetDB(const std::string& db_name, uint32_t* target);
@@ -610,6 +621,7 @@ class PikaConf : public pstd::BaseConf {
   std::string server_id_;
   std::string run_id_;
   std::string master_run_id_;
+  std::string replication_id_;
   std::string requirepass_;
   std::string masterauth_;
   std::string userpass_;

src/pika_admin.cc

@@ -151,6 +151,10 @@ void SlaveofCmd::Do(std::shared_ptr<Slot> slot) {
     return;
   }
 
+  /* The return value of the slaveof command OK does not really represent whether
+   * the data synchronization was successful, but only changes the status of the
+   * slaveof executor to slave */
+
   bool sm_ret = g_pika_server->SetMaster(master_ip_, static_cast<int32_t>(master_port_));
 
   if (sm_ret) {
@@ -1038,7 +1042,7 @@ void InfoCmd::InfoReplication(std::string& info) {
       info.append("ERR: server role is error\r\n");
       return;
   }
-
+  tmp_stream << "ReplicationID: " << g_pika_conf->replication_id() << "\r\n";
   std::string slaves_list_str;
   switch (host_role) {
     case PIKA_ROLE_SLAVE:
@@ -1324,6 +1328,8 @@ void ConfigCmd::Do(std::shared_ptr<Slot> slot) {
     ConfigRewrite(config_ret);
   } else if (strcasecmp(config_args_v_[0].data(), "resetstat") == 0) {
     ConfigResetstat(config_ret);
+  } else if (strcasecmp(config_args_v_[0].data(), "rewritereplicationid") == 0) {
+    ConfigRewriteReplicationID(config_ret);
   }
   res_.AppendStringRaw(config_ret);
 }
@@ -1849,6 +1855,12 @@ void ConfigCmd::ConfigGet(std::string& ret) {
     EncodeNumber(&config_body, g_pika_conf->max_rsync_parallel_num());
   }
 
+  if (pstd::stringmatch(pattern.data(), "replication-id", 1) != 0) {
+    elements += 2;
+    EncodeString(&config_body, "replication-id");
+    EncodeString(&config_body, g_pika_conf->replication_id());
+  }
+
   std::stringstream resp;
   resp << "*" << std::to_string(elements) << "\r\n" << config_body;
   ret = resp.str();
@@ -2183,12 +2195,12 @@ void ConfigCmd::ConfigSet(std::string& ret) {
     g_pika_conf->SetThrottleBytesPerSecond(static_cast<int>(ival));
     ret = "+OK\r\n";
   } else if (set_item == "max-rsync-parallel-num") {
-     if ((pstd::string2int(value.data(), value.size(), &ival) == 0) || ival > kMaxRsyncParallelNum) {
-       ret = "-ERR Invalid argument \'" + value + "\' for CONFIG SET 'max-rsync-parallel-num'\r\n";
-       return;
-     }
-     g_pika_conf->SetMaxRsyncParallelNum(static_cast<int>(ival));
-     ret = "+OK\r\n";
+    if ((pstd::string2int(value.data(), value.size(), &ival) == 0) || ival > kMaxRsyncParallelNum) {
+      ret = "-ERR Invalid argument \'" + value + "\' for CONFIG SET 'max-rsync-parallel-num'\r\n";
+      return;
+    }
+    g_pika_conf->SetMaxRsyncParallelNum(static_cast<int>(ival));
+    ret = "+OK\r\n";
   } else {
     ret = "-ERR Unsupported CONFIG parameter: " + set_item + "\r\n";
   }
@@ -2202,6 +2214,14 @@ void ConfigCmd::ConfigRewrite(std::string& ret) {
   }
 }
 
+void ConfigCmd::ConfigRewriteReplicationID(std::string &ret) {
+  if (g_pika_conf->ConfigRewriteReplicationID() != 0) {
+    ret = "+OK\r\n";
+  } else {
+    ret = "-ERR Rewire ReplicationID CONFIG fail\r\n";
+  }
+}
+
 void ConfigCmd::ConfigResetstat(std::string& ret) {
   g_pika_server->ResetStat();
   ret = "+OK\r\n";
@@ -2627,6 +2647,19 @@ void DiskRecoveryCmd::Do(std::shared_ptr<Slot> slot) {
   res_.SetRes(CmdRes::kOk, "The disk error has been recovered");
 }
 
+void ClearReplicationIDCmd::DoInitial() {
+  if (!CheckArg(argv_.size())) {
+    res_.SetRes(CmdRes::kWrongNum, kCmdNameClearReplicationID);
+    return;
+  }
+}
+
+void ClearReplicationIDCmd::Do(std::shared_ptr<Slot> slot) {
+  g_pika_conf->SetReplicationID("");
+  g_pika_conf->ConfigRewriteReplicationID();
+  res_.SetRes(CmdRes::kOk, "ReplicationID is cleared");
+}
+
 #ifdef WITH_COMMAND_DOCS
 
 bool CommandCmd::CommandFieldCompare::operator()(const std::string& a, const std::string& b) const {

src/pika_command.cc

@@ -94,6 +94,8 @@ void InitCmdTable(CmdTable* cmd_table) {
   cmd_table->insert(std::pair<std::string, std::unique_ptr<Cmd>>(kCmdNameQuit, std::move(quitptr)));
   std::unique_ptr<Cmd> diskrecoveryptr = std::make_unique<DiskRecoveryCmd>(kCmdNameDiskRecovery, 1, kCmdFlagsRead | kCmdFlagsAdmin);
   cmd_table->insert(std::pair<std::string, std::unique_ptr<Cmd>>(kCmdNameDiskRecovery, std::move(diskrecoveryptr)));
+  std::unique_ptr<Cmd> clearreplicationidptr = std::make_unique<ClearReplicationIDCmd>(kCmdNameClearReplicationID, 1, kCmdFlagsWrite | kCmdFlagsAdmin);
+  cmd_table->insert(std::pair<std::string, std::unique_ptr<Cmd>>(kCmdNameClearReplicationID, std::move(clearreplicationidptr)));
 
 #ifdef WITH_COMMAND_DOCS
   std::unique_ptr<Cmd> commandptr = std::make_unique<CommandCmd>(kCmdNameCommand, -1, kCmdFlagsRead | kCmdFlagsAdmin);

src/pika_conf.cc

@@ -158,6 +158,7 @@ int PikaConf::Load() {
   } else if (run_id_.length() != configRunIDSize) {
     LOG(FATAL) << "run-id " << run_id_ << " is invalid, its string length should be " << configRunIDSize;
   }
+  GetConfStr("replication-id", &replication_id_);
   GetConfStr("requirepass", &requirepass_);
   GetConfStr("masterauth", &masterauth_);
   GetConfStr("userpass", &userpass_);
@@ -642,6 +643,7 @@ int PikaConf::ConfigRewrite() {
   SetConfStr("write-binlog", write_binlog_ ? "yes" : "no");
   SetConfStr("run-id", run_id_);
   SetConfStr("master-run-id", master_run_id_);
+  SetConfStr("replication-id", replication_id_);
   SetConfInt("max-cache-statistic-keys", max_cache_statistic_keys_);
   SetConfInt("small-compaction-threshold", small_compaction_threshold_);
   SetConfInt("max-client-response-size", static_cast<int32_t>(max_client_response_size_));
@@ -687,6 +689,29 @@ int PikaConf::ConfigRewrite() {
   return static_cast<int>(WriteBack());
 }
 
+int PikaConf::ConfigRewriteReplicationID() {
+  std::lock_guard l(rwlock_);
+  SetConfStr("replication-id", replication_id_);
+  if (!diff_commands_.empty()) {
+    std::vector<pstd::BaseConf::Rep::ConfItem> filtered_items;
+    for (const auto& diff_command : diff_commands_) {
+      if (!diff_command.second.empty()) {
+        pstd::BaseConf::Rep::ConfItem item(pstd::BaseConf::Rep::kConf, diff_command.first, diff_command.second);
+        filtered_items.push_back(item);
+      }
+    }
+    if (!filtered_items.empty()) {
+      pstd::BaseConf::Rep::ConfItem comment_item(pstd::BaseConf::Rep::kComment, "# Generated by ReplicationID CONFIG REWRITE\n");
+      PushConfItem(comment_item);
+      for (const auto& item : filtered_items) {
+        PushConfItem(item);
+      }
+    }
+    diff_commands_.clear();
+  }
+  return static_cast<int>(WriteBack());
+}
+
 rocksdb::CompressionType PikaConf::GetCompression(const std::string& value) {
   if (value == "snappy") {
     return rocksdb::CompressionType::kSnappyCompression;

src/pika_inner_message.proto

@@ -117,6 +117,7 @@ message InnerResponse {
     required bool      classic_mode = 1;
     repeated DBInfo    dbs_info  = 2;
     required string    run_id = 3;
+    optional string    replication_id = 4;
   }
 
   // master to slave

src/pika_repl_client_conn.cc

@@ -122,12 +122,27 @@ void PikaReplClientConn::HandleMetaSyncResponse(void* arg) {
     return;
   }
 
-  if (meta_sync.run_id() == "" || g_pika_server->master_run_id() != meta_sync.run_id()) {
-    LOG(INFO) << "Run id is not equal, need to do full sync, remote master run id: " << meta_sync.run_id()
-              << ", local run id: " << g_pika_server->master_run_id();
+  // The relicationid obtained from the server is null
+  if (meta_sync.replication_id() == "") {
+    LOG(WARNING) << "Meta Sync Failed: the relicationid obtained from the server is null, keep sending MetaSync msg";
+    return;
+  }
+
+  // The Replicationids of both the primary and secondary Replicationid are not empty and are not equal
+  if (g_pika_conf->replication_id() != meta_sync.replication_id() && g_pika_conf->replication_id() != "") {
+    LOG(WARNING) << "Meta Sync Failed: replicationid on both sides of the connection are inconsistent";
+    g_pika_server->SyncError();
+    conn->NotifyClose();
+    return;
+  }
+
+  // First synchronization between the master and slave
+  if (g_pika_conf->replication_id() != meta_sync.replication_id()) {
+    LOG(INFO) << "New node is added to the cluster and requires full replication, remote replication id: " << meta_sync.replication_id()
+              << ", local replication id: " << g_pika_conf->replication_id();
     g_pika_server->force_full_sync_ = true;
-    g_pika_server->set_master_run_id(meta_sync.run_id());
-    g_pika_conf->SetMasterRunID(meta_sync.run_id());
+    g_pika_conf->SetReplicationID(meta_sync.replication_id());
+    g_pika_conf->ConfigRewriteReplicationID();
   }
 
   g_pika_conf->SetWriteBinlog("yes");