binlog compatibility enables pika 2.3.5 to migrate smoothly to pika 3…

….0 (#300)

conf/pika.conf

@@ -56,6 +56,11 @@ db-sync-path : ./dbsync/
 db-sync-speed : -1
 # The slave priority
 slave-priority : 100
+# When it becomes slave, the type of binlog it receives from the master
+# if this option is set to 'new', that means I will be a slave to Pika who's version 3.0
+# if this opsion is set to 'old', that means I will be a slave to Pika who's version 2.3.3 ~ 2.3.5
+# identify-binlog-type  [new | old]
+identify-binlog-type : new
 # network interface
 #network-interface : eth1
 # replication

include/pika_binlog_receiver_thread.h

@@ -16,6 +16,9 @@
 #include "include/pika_master_conn.h"
 #include "include/pika_new_master_conn.h"
 #include "include/pika_command.h"
+#include "include/pika_conf.h"
+
+extern PikaConf* g_pika_conf;
 
 class PikaBinlogReceiverThread {
  public:
@@ -46,7 +49,13 @@ class PikaBinlogReceiverThread {
         const std::string &ip_port,
         pink::ServerThread *thread,
         void* worker_specific_data) const override {
-      return new PikaNewMasterConn(connfd, ip_port, binlog_receiver_);
+        if (g_pika_conf->identify_binlog_type() == "new") {
+          LOG(INFO) << "Master conn factory create pika new master conn";
+          return new PikaNewMasterConn(connfd, ip_port, binlog_receiver_);
+        } else {
+          LOG(INFO) << "Master conn factory create pika master conn";
+          return new PikaMasterConn(connfd, ip_port, binlog_receiver_);
+        }
     }
 
    private:

include/pika_conf.h

@@ -30,6 +30,7 @@ class PikaConf : public slash::BaseConf {
   std::string double_master_sid()        { RWLock l(&rwlock_, false); return double_master_sid_; }
   std::string slaveof() {RWLock l(&rwlock_, false); return slaveof_;}
   int slave_priority() {RWLock l(&rwlock_, false); return slave_priority_;}
+  std::string identify_binlog_type() {RWLock l(&rwlock_, false); return identify_binlog_type_;}
   int thread_num()        { RWLock l(&rwlock_, false); return thread_num_; }
   int sync_thread_num()        { RWLock l(&rwlock_, false); return sync_thread_num_; }
   int sync_buffer_size()        { RWLock l(&rwlock_, false); return sync_buffer_size_; }
@@ -88,6 +89,10 @@ class PikaConf : public slash::BaseConf {
     slaveof_ = value;
   }
   void SetSlavePriority(const int value)        { RWLock l(&rwlock_, true); slave_priority_ = value; }
+  void SetIdentifyBinlogType(const std::string& value) {
+    RWLock l(&rwlock_, true);
+    identify_binlog_type_ = value;
+  }
   void SetBgsavePath(const std::string &value) {
     RWLock l(&rwlock_, true);
     bgsave_path_ = value;
@@ -195,6 +200,7 @@ class PikaConf : public slash::BaseConf {
   std::string bgsave_path_;
   std::string bgsave_prefix_;
   std::string pidfile_;
+  std::string identify_binlog_type_;
 
   //char pidfile_[PIKA_WORD_SIZE];
   std::string compression_;

src/pika_admin.cc

@@ -1080,8 +1080,12 @@ void ConfigCmd::ConfigGet(std::string &ret) {
     ret = "*2\r\n";
     EncodeString(&ret, "slave-priority");
     EncodeInt32(&ret, g_pika_conf->slave_priority());
+  } else if (get_item == "identify-binlog-type") {
+    ret = "*2\r\n";
+    EncodeString(&ret, "identify-binlog-type");
+    EncodeString(&ret, g_pika_conf->identify_binlog_type());
   } else if (get_item == "*") {
-    ret = "*86\r\n";
+    ret = "*88\r\n";
     EncodeString(&ret, "port");
     EncodeInt32(&ret, g_pika_conf->port());
     EncodeString(&ret, "double-master-ip");
@@ -1168,6 +1172,8 @@ void ConfigCmd::ConfigGet(std::string &ret) {
     EncodeString(&ret, g_pika_conf->slaveof());
     EncodeString(&ret, "slave-priority");
     EncodeInt32(&ret, g_pika_conf->slave_priority());
+    EncodeString(&ret, "identify-binlog-type");
+    EncodeString(&ret, g_pika_conf->identify_binlog_type());
   } else {
     ret = "*0\r\n";
   }
@@ -1176,7 +1182,7 @@ void ConfigCmd::ConfigGet(std::string &ret) {
 void ConfigCmd::ConfigSet(std::string& ret) {
   std::string set_item = config_args_v_[1];
   if (set_item == "*") {
-    ret = "*19\r\n";
+    ret = "*20\r\n";
     EncodeString(&ret, "loglevel");
     EncodeString(&ret, "timeout");
     EncodeString(&ret, "requirepass");
@@ -1196,6 +1202,7 @@ void ConfigCmd::ConfigSet(std::string& ret) {
     EncodeString(&ret, "compact-cron");
     EncodeString(&ret, "compact-interval");
     EncodeString(&ret, "slave-priority");
+    EncodeString(&ret, "identify-binlog-type");
     return;
   }
   std::string value = config_args_v_[2];
@@ -1359,6 +1366,18 @@ void ConfigCmd::ConfigSet(std::string& ret) {
       g_pika_conf->SetCompactInterval(value);
       ret = "+OK\r\n";
     }
+  } else if (set_item == "identify-binlog-type") {
+    int role = g_pika_server->role();
+    if (role & PIKA_ROLE_SLAVE || role & PIKA_ROLE_DOUBLE_MASTER) {
+      ret = "-ERR need to close master-slave or double-master mode first\r\n";
+      return;
+    } else if (value != "old" && value != "new") {
+      ret = "-ERR invalid identify-binlog-type\r\n";
+      return;
+    } else {
+      g_pika_conf->SetIdentifyBinlogType(value);
+      ret = "+OK\r\n";
+    }
   } else {
     ret = "-ERR No such configure item\r\n";
   }

src/pika_conf.cc

@@ -81,6 +81,7 @@ int PikaConf::Load()
   GetConfStr("compression", &compression_);
   GetConfBool("slave-read-only", &readonly_);
   GetConfInt("slave-priority", &slave_priority_);
+  GetConfStr("identify-binlog-type", &identify_binlog_type_);
 
   //
   // Immutable Sections
@@ -265,6 +266,7 @@ int PikaConf::ConfigRewrite() {
   SetConfStr("network-interface", network_interface_);
   SetConfStr("slaveof", slaveof_);
   SetConfInt("slave-priority", slave_priority_);
+  SetConfStr("identify-binlog-type", identify_binlog_type_);
 
   SetConfInt("binlog-file-size", binlog_file_size_);
   SetConfStr("compression", compression_);