feature #4809: Start the federation replica manager with OpenNebula

include/FedReplicaManager.h

@@ -37,8 +37,10 @@ public:
      *  @param _p purge timeout for log
      *  @param d pointer to underlying DB (LogDB)
      *  @param l log_retention length (num records)
+     *  @param s true if operating in solo mode
      */
-    FedReplicaManager(time_t _t, time_t _p, SqlDB * d, long l);
+    FedReplicaManager(time_t _t, time_t _p, SqlDB * d, const std::string& l,
+            bool s);
 
     virtual ~FedReplicaManager();
 
@@ -86,6 +88,16 @@ public:
      */
     void delete_zone(int zone_id);
 
+    /**
+     *  Bootstrap federated log
+     */
+    static int bootstrap(SqlDB *_db);
+
+    pthread_t get_thread_id() const
+    {
+        return frm_thread;
+    };
+
 private:
     friend void * frm_loop(void *arg);
 
@@ -143,7 +155,7 @@ private:
 
     SqlDB * logdb;
 
-    long log_retention;
+    std::string log_retention;
 
     // -------------------------------------------------------------------------
     // Action Listener interface

include/Nebula.h

@@ -52,6 +52,7 @@
 #include "MarketPlaceManager.h"
 #include "IPAMManager.h"
 #include "RaftManager.h"
+#include "FedReplicaManager.h"
 
 #include "DefaultQuotas.h"
 
@@ -230,6 +231,12 @@ public:
     {
         return raftm;
     };
+
+    FedReplicaManager * get_frm()
+    {
+        return frm;
+    };
+
     // --------------------------------------------------------------
     // Environment & Configuration
     // --------------------------------------------------------------
@@ -679,7 +686,7 @@ private:
         dspool(0), clpool(0), docpool(0), zonepool(0), secgrouppool(0),
         vdcpool(0), vrouterpool(0), marketpool(0), apppool(0), vmgrouppool(0),
         lcm(0), vmm(0), im(0), tm(0), dm(0), rm(0), hm(0), authm(0), aclm(0),
-        imagem(0), marketm(0), ipamm(0), raftm(0)
+        imagem(0), marketm(0), ipamm(0), raftm(0), frm(0)
     {
         const char * nl = getenv("ONE_LOCATION");
 
@@ -744,6 +751,7 @@ private:
         delete marketm;
         delete ipamm;
         delete raftm;
+        delete frm;
         delete nebula_configuration;
         delete logdb;
         delete system_db;
@@ -836,6 +844,7 @@ private:
     MarketPlaceManager *    marketm;
     IPAMManager *           ipamm;
     RaftManager *           raftm;
+    FedReplicaManager *     frm;
 
     // ---------------------------------------------------------------
     // Implementation functions

src/nebula/Nebula.cc

@@ -366,6 +366,7 @@ void Nebula::start(bool bootstrap_only)
             rc += SecurityGroupPool::bootstrap(logdb);
             rc += VirtualRouterPool::bootstrap(logdb);
             rc += VMGroupPool::bootstrap(logdb);
+            rc += FedReplicaManager::bootstrap(logdb);
 
             // Create the system tables only if bootstrap went well
             if (rc == 0)
@@ -693,6 +694,24 @@ void Nebula::start(bool bootstrap_only)
        throw runtime_error("Could not start the Raft Consensus Manager");
     }
 
+    // ---- FedReplica Manager ----
+    try
+    {
+        frm = new FedReplicaManager(timer_period, log_purge, logdb,
+                log_retention, solo);
+    }
+    catch (bad_alloc&)
+    {
+        throw;
+    }
+
+    rc = frm->start();
+
+    if ( rc != 0 )
+    {
+       throw runtime_error("Could not start the Federation Replica Manager");
+    }
+
     // ---- Virtual Machine Manager ----
     try
     {
@@ -1076,6 +1095,7 @@ void Nebula::start(bool bootstrap_only)
 	ipamm->finalize();
     aclm->finalize();
     raftm->finalize();
+    frm->finalize();
 
     //sleep to wait drivers???
 
@@ -1091,6 +1111,7 @@ void Nebula::start(bool bootstrap_only)
     pthread_join(marketm->get_thread_id(),0);
     pthread_join(ipamm->get_thread_id(),0);
     pthread_join(raftm->get_thread_id(),0);
+    pthread_join(frm->get_thread_id(),0);
 
     if(is_federation_slave())
     {

src/raft/FedReplicaManager.cc

@@ -33,9 +33,9 @@ const time_t FedReplicaManager::xmlrpc_timeout_ms = 2000;
 /* -------------------------------------------------------------------------- */
 /* -------------------------------------------------------------------------- */
 
-FedReplicaManager::FedReplicaManager(time_t _t, time_t _p, SqlDB * d, long l):
-    ReplicaManager(), timer_period(_t), purge_period(_t), logdb(d),
-    log_retention(l)
+FedReplicaManager::FedReplicaManager(time_t _t, time_t _p, SqlDB * d,
+    const std::string& l, bool s): ReplicaManager(), timer_period(_t),
+    purge_period(_t), logdb(d), log_retention(l)
 {
     Nebula& nd       = Nebula::instance();
     ZonePool * zpool = nd.get_zonepool();
@@ -65,7 +65,10 @@ FedReplicaManager::FedReplicaManager(time_t _t, time_t _p, SqlDB * d, long l):
 
     get_last_index(last_index);
 
-    start_replica_threads(zone_ids);
+    if ( nd.is_federation_master() && s )
+    {
+        start_replica_threads(zone_ids);
+    }
 
     pthread_mutex_init(&mutex, 0);
 };
@@ -74,6 +77,8 @@ FedReplicaManager::FedReplicaManager(time_t _t, time_t _p, SqlDB * d, long l):
 
 FedReplicaManager::~FedReplicaManager()
 {
+    Nebula& nd = Nebula::instance();
+
     std::map<int, ZoneServers *>::iterator it;
 
     for ( it = zones.begin() ; it != zones.end() ; ++it )
@@ -82,6 +87,11 @@ FedReplicaManager::~FedReplicaManager()
     }
 
     zones.clear();
+
+    if ( nd.is_federation_master() )
+    {
+        stop_replica_threads();
+    }
 };
 
 /* -------------------------------------------------------------------------- */
@@ -181,6 +191,13 @@ int FedReplicaManager::start()
     return rc;
 }
 
+/* -------------------------------------------------------------------------- */
+
+void FedReplicaManager::finalize_action(const ActionRequest& ar)
+{
+    NebulaLog::log("RCM", Log::INFO, "Raft Consensus Manager...");
+}
+
 /* -------------------------------------------------------------------------- */
 /* -------------------------------------------------------------------------- */
 
@@ -335,6 +352,15 @@ int FedReplicaManager::get_last_index(unsigned int& index)
     return rc;
 }
 
+/* -------------------------------------------------------------------------- */
+
+int FedReplicaManager::bootstrap(SqlDB *_db)
+{
+    std::ostringstream oss(db_bootstrap);
+
+    return _db->exec_local_wr(oss);
+}
+
 /* -------------------------------------------------------------------------- */
 /* -------------------------------------------------------------------------- */