save master status to storage when masters change

Former-commit-id: 4989926a0028aed7d7700fd1d1f4ed27c20277cc

src/IStorage.h

@@ -12,6 +12,7 @@ public:
     virtual ~IStorageFactory() {}
     virtual class IStorage *create(int db, key_load_iterator itr, void *privdata) = 0;
     virtual class IStorage *createMetadataDb() = 0;
+    virtual std::string getMetadata() const = 0;
     virtual const char *name() const = 0;
     virtual size_t totalDiskspaceUsed() const = 0;
     virtual bool FSlow() const = 0;

src/replication.cpp

@@ -3783,6 +3783,45 @@ void disconnectMaster(redisMaster *mi)
     }
 }
 
+void saveMasterStatusToStorage()
+{
+    if (!g_pserver->m_pstorageFactory || !g_pserver->metadataDb) return;
+    if (listLength(g_pserver->masters) == 0) {
+        g_pserver->metadataDb->insert("repl_masters", 12, (void*)"", 0, true);
+        return;
+    }
+    sds val = sds(sdsempty());
+    listNode *ln;
+    listIter li;
+    redisMaster *mi;
+    listRewind(g_pserver->masters,&li);
+    while((ln = listNext(&li)) != NULL) {
+        mi = (redisMaster*)listNodeValue(ln);
+        if (!mi->master) {
+            // If master client is not available, use info from master struct - better than nothing
+            if (mi->master_replid[0] == 0) {
+                // if replid is null, there's no reason to save it
+                continue;
+            }
+            val = sdscatfmt(val, "%s:%I:%s:%i;", mi->master_replid,
+                mi->master_initial_offset,
+                mi->masterhost,
+                mi->masterport);
+        }
+        else {
+            if (mi->master->replid[0] == 0) {
+                // if replid is null, there's no reason to save it
+                continue;
+            }
+            val = sdscatfmt(val, "%s:%I:%s:%i;", mi->master->replid,
+                mi->master->reploff,
+                mi->masterhost,
+                mi->masterport);
+        }
+    }
+    g_pserver->metadataDb->insert("repl_masters", 12, (void*)val, sdslen(val), true);
+}
+
 /* Set replication to the specified master address and port. */
 struct redisMaster *replicationAddMaster(char *ip, int port) {
     // pre-reqs: We must not already have a replica in the list with the same tuple
@@ -3855,6 +3894,7 @@ struct redisMaster *replicationAddMaster(char *ip, int port) {
             mi->masterhost, mi->masterport);
         connectWithMaster(mi);
     }
+    saveMasterStatusToStorage();
     return mi;
 }
 
@@ -3938,6 +3978,8 @@ void replicationUnsetMaster(redisMaster *mi) {
     /* Restart the AOF subsystem in case we shut it down during a sync when
      * we were still a slave. */
     if (g_pserver->aof_enabled && g_pserver->aof_state == AOF_OFF) restartAOFAfterSYNC();
+
+    saveMasterStatusToStorage();
 }
 
 /* This function is called when the replica lose the connection with the
@@ -3969,6 +4011,8 @@ void replicationHandleMasterDisconnection(redisMaster *mi) {
                 mi->masterhost, mi->masterport);
             connectWithMaster(mi);
         }
+
+        saveMasterStatusToStorage();
     }
 }
 

src/server.cpp

@@ -3985,6 +3985,10 @@ void initServer(void) {
     latencyMonitorInit();
 
     if (g_pserver->m_pstorageFactory) {
+        std::string repl_masters = g_pserver->m_pstorageFactory->getMetadata();
+        if (!repl_masters.empty()) {
+            serverLog(LL_NOTICE, "Loaded repl-masters from storage provider: %s", repl_masters);
+        }
         g_pserver->metadataDb = g_pserver->m_pstorageFactory->createMetadataDb();
     }
 

src/storage/teststorageprovider.cpp

@@ -13,6 +13,11 @@ IStorage *TestStorageFactory::createMetadataDb()
     return metadataDb;
 }
 
+std::string TestStorageFactory::getMetadata() const
+{
+    return "";
+}
+
 const char *TestStorageFactory::name() const
 {
     return "TEST Storage Provider";

src/storage/teststorageprovider.h

@@ -6,6 +6,7 @@ class TestStorageFactory : public IStorageFactory
 {
     virtual class IStorage *create(int db, key_load_iterator itr, void *privdata) override;
     virtual class IStorage *createMetadataDb() override;
+    virtual std::string getMetadata() const override;
     virtual const char *name() const override;
     virtual size_t totalDiskspaceUsed() const override { return 0; }
     virtual bool FSlow() const { return false; }