F #4809: Refresh zone list on server add/remove to keep an updated list

of servers

include/FedReplicaManager.h

@@ -99,6 +99,26 @@ public:
      */
     int start();
 
+    /**
+     *  Start the replication threads, and updates the server list of the zone
+     */
+    void start_replica_threads()
+    {
+        std::vector<int> zids;
+
+        update_zones(zids);
+
+        ReplicaManager::start_replica_threads(zids);
+    }
+
+    /**
+     *  Updates the list of zones and servers in the zone. This function is
+     *  invoked when a server becomes leader, or whenever a server is +
+     *  added/removed to the zone
+     *    @param zids ids of zones
+     */
+    void update_zones(std::vector<int>& zids);
+
     /**
      *  Adds a new zone to the replication list and starts the associated
      *  replica thread
@@ -185,6 +205,7 @@ private:
 
     std::string log_retention;
 
+
     // -------------------------------------------------------------------------
     // Action Listener interface
     // -------------------------------------------------------------------------

src/raft/FedReplicaManager.cc

@@ -38,53 +38,20 @@ FedReplicaManager::FedReplicaManager(time_t _t, time_t _p, SqlDB * d,
     const std::string& l, bool s): ReplicaManager(), timer_period(_t),
     purge_period(_p), logdb(d), log_retention(l)
 {
-    Nebula& nd       = Nebula::instance();
-    ZonePool * zpool = nd.get_zonepool();
+    Nebula& nd = Nebula::instance();
 
-    std::vector<int> zone_ids;
-    vector<int>::iterator it;
-
-    std::map<int, std::string> zone_servers;
-
-    int zone_id = nd.get_zone_id();
+    pthread_mutex_init(&mutex, 0);
 
     am.addListener(this);
 
-    if ( zpool->list_zones(zone_ids) != 0 )
-    {
-        return;
-    }
-
-    for ( it = zone_ids.begin() ; it != zone_ids.end(); )
-    {
-        if ( *it == zone_id )
-        {
-            it = zone_ids.erase(it);
-        }
-        else
-        {
-            zpool->get_zone_servers(*it, zone_servers);
-
-            ZoneServers * zs = new ZoneServers(*it, zone_servers);
-
-            zones.insert(make_pair(*it, zs));
-
-            zone_servers.clear();
-
-            ++it;
-        }
-    }
-
     get_last_index(last_index);
 
     // Replica threads are started on master in solo mode.
     // HA start/stop the replica threads on leader/follower states will
     if ( nd.is_federation_master() && s )
     {
-        start_replica_threads(zone_ids);
+        start_replica_threads();
     }
-
-    pthread_mutex_init(&mutex, 0);
 };
 
 /* -------------------------------------------------------------------------- */
@@ -209,7 +176,53 @@ int FedReplicaManager::start()
 
 void FedReplicaManager::finalize_action(const ActionRequest& ar)
 {
-    NebulaLog::log("FRM", Log::INFO, "Raft Consensus Manager...");
+    NebulaLog::log("FRM", Log::INFO, "Federation Replica Manager...");
+}
+
+/* -------------------------------------------------------------------------- */
+/* -------------------------------------------------------------------------- */
+
+void FedReplicaManager::update_zones(std::vector<int>& zone_ids)
+{
+    Nebula& nd       = Nebula::instance();
+    ZonePool * zpool = nd.get_zonepool();
+
+    vector<int>::iterator it;
+
+    std::map<int, std::string> zone_servers;
+
+    int zone_id = nd.get_zone_id();
+
+    if ( zpool->list_zones(zone_ids) != 0 )
+    {
+        return;
+    }
+
+    pthread_mutex_lock(&mutex);
+
+    zones.clear();
+
+    for (it = zone_ids.begin() ; it != zone_ids.end(); )
+    {
+        if ( *it == zone_id )
+        {
+            it = zone_ids.erase(it);
+        }
+        else
+        {
+            zpool->get_zone_servers(*it, zone_servers);
+
+            ZoneServers * zs = new ZoneServers(*it, zone_servers);
+
+            zones.insert(make_pair(*it, zs));
+
+            zone_servers.clear();
+
+            ++it;
+        }
+    }
+
+    pthread_mutex_unlock(&mutex);
 }
 
 /* -------------------------------------------------------------------------- */

src/raft/RaftManager.cc

@@ -17,6 +17,7 @@
 #include "Nebula.h"
 
 #include "RaftManager.h"
+#include "FedReplicaManager.h"
 #include "ZoneServer.h"
 #include "Client.h"
 
@@ -336,6 +337,8 @@ void RaftManager::leader()
     LogDB * logdb     = nd.get_logdb();
     AclManager * aclm = nd.get_aclm();
 
+    FedReplicaManager * frm = nd.get_frm();
+
     std::map<int, std::string>::iterator it;
     std::vector<int> _follower_ids;
 
@@ -404,6 +407,11 @@ void RaftManager::leader()
 
     aclm->reload_rules();
 
+    if ( nd.is_federation_master() )
+    {
+        frm->start_replica_threads();
+    }
+
     logdb->insert_raft_state(raft_state_xml);
 
     NebulaLog::log("RCM", Log::INFO, "oned is now the leader of zone");
@@ -419,6 +427,8 @@ void RaftManager::follower(unsigned int _term)
     Nebula& nd    = Nebula::instance();
     LogDB * logdb = nd.get_logdb();
 
+    FedReplicaManager * frm = nd.get_frm();
+
     std::string raft_state_xml;
 
     logdb->setup_index(lapplied, lindex);
@@ -466,6 +476,11 @@ void RaftManager::follower(unsigned int _term)
 
     pthread_mutex_unlock(&mutex);
 
+    if ( nd.is_federation_master() )
+    {
+        frm->stop_replica_threads();
+    }
+
     logdb->insert_raft_state(raft_state_xml);
 }
 

src/rm/RequestManagerZone.cc

@@ -67,7 +67,10 @@ void ZoneAddServer::request_execute(xmlrpc_c::paramList const& paramList,
 
     zone->unlock();
 
+    std::vector<int> zids;
+
 	Nebula::instance().get_raftm()->add_server(zs_id);
+	Nebula::instance().get_frm()->update_zones(zids);
 
     success_response(id, att);
 }
@@ -110,7 +113,10 @@ void ZoneDeleteServer::request_execute(xmlrpc_c::paramList const& paramList,
 
     zone->unlock();
 
+    std::vector<int> zids;
+
 	Nebula::instance().get_raftm()->delete_server(zs_id);
+	Nebula::instance().get_frm()->update_zones(zids);
 
     success_response(id, att);
 }