restarting agents inform rest of their new endpoints

arangod/Agency/Agent.cpp

@@ -924,7 +924,6 @@ TimePoint const& Agent::leaderSince() const {
 
 // Notify inactive pool members of configuration change()
 void Agent::notifyInactive() const {
-  if (_config.poolSize() > _config.size()) {
 
   std::map<std::string, std::string> pool = _config.pool();
   std::string path = "/_api/agency_priv/inform";
@@ -942,6 +941,7 @@ void Agent::notifyInactive() const {
   for (auto const& p : pool) {
 
     if (p.first != id()) {
+
       auto headerFields =
         std::make_unique<std::unordered_map<std::string, std::string>>();
 
@@ -954,6 +954,36 @@ void Agent::notifyInactive() const {
   }
 
 }
+
+void Agent::updatePeerEndpoint(query_t const& message) {
+
+  VPackSlice slice = message->slice();
+
+  if (!slice.isObject() || slice.length() == 0) {
+    THROW_ARANGO_EXCEPTION_MESSAGE(
+      TRI_ERROR_AGENCY_INFORM_MUST_BE_OBJECT,
+      std::string("Inproper greeting: ") + slice.toJson());
+  }
+
+  std::string uuid, endpoint;
+  try {
+    uuid = slice.keyAt(0).copyString();
+  } catch (std::exception const& e) {
+    THROW_ARANGO_EXCEPTION_MESSAGE(
+      TRI_ERROR_AGENCY_INFORM_MUST_BE_OBJECT,
+      std::string("Cannot deal with UUID: ") + e.what());
+  }
+
+  try {
+    endpoint = slice.valueAt(0).copyString();
+  } catch (std::exception const& e) {
+    THROW_ARANGO_EXCEPTION_MESSAGE(
+      TRI_ERROR_AGENCY_INFORM_MUST_BE_OBJECT,
+      std::string("Cannot deal with UUID: ") + e.what());
+  }
+
+  _config.updateEndpoint(uuid, endpoint);
+
 }
 
 void Agent::notify(query_t const& message) {
@@ -989,6 +1019,7 @@ void Agent::notify(query_t const& message) {
   }
 
   _config.update(message);
+
   _state.persistActiveAgents(_config.activeToBuilder(),
                              _config.poolToBuilder());
 }

arangod/Agency/Agent.h

@@ -196,6 +196,8 @@ class Agent : public arangodb::Thread {
   /// @brief Get start time of leadership
   TimePoint const& leaderSince() const;
 
+  void updatePeerEndpoint(query_t const& message);
+
   /// @brief State reads persisted state and prepares the agent
   friend class State;
 

arangod/Agency/AgentConfiguration.cpp

@@ -341,12 +341,27 @@ query_t config_t::poolToBuilder() const {
   return ret;
 }
 
+
+void config_t::updateEndpoint(std::string const& id, std::string const& ep) {
+  WRITE_LOCKER(readLocker, _lock);
+  if (_pool[id] != ep) {
+    _pool[id] = ep;
+    ++_version;
+  }
+}
+
+
 void config_t::update(query_t const& message) {
   VPackSlice slice = message->slice();
   std::map<std::string, std::string> pool;
   bool changed = false;
   for (auto const& p : VPackObjectIterator(slice.get(poolStr))) {
-    pool[p.key.copyString()] = p.value.copyString();
+    auto const& id = p.key.copyString();
+    if (id != _id) {
+      pool[id] = p.value.copyString();
+    } else {
+      pool[id] = _endpoint;
+    }
   }
   std::vector<std::string> active;
   for (auto const& a : VPackArrayIterator(slice.get(activeStr))) {
@@ -522,7 +537,6 @@ bool config_t::merge(VPackSlice const& conf) {
   WRITE_LOCKER(writeLocker, _lock); // All must happen under the lock or else ...
 
   _id = conf.get(idStr).copyString();  // I get my id
-  _pool[_id] = _endpoint;              // Register my endpoint with it
   _startup = "persistence";
 
   std::stringstream ss;
@@ -562,7 +576,12 @@ bool config_t::merge(VPackSlice const& conf) {
   if (conf.hasKey(poolStr)) {  // Persistence only
     LOG_TOPIC(DEBUG, Logger::AGENCY) << "Found agent pool in persistence:";
     for (auto const& peer : VPackObjectIterator(conf.get(poolStr))) {
-      _pool[peer.key.copyString()] = peer.value.copyString();
+      auto const& id = peer.key.copyString();
+      if (id != _id) {
+        _pool[id] = peer.value.copyString();
+      } else {
+        _pool[id] = _endpoint;
+      }
     }
     ss << conf.get(poolStr).toJson() << " (persisted)";
   } else {

arangod/Agency/AgentConfiguration.h

@@ -194,6 +194,9 @@ struct config_t {
   /// @brief
   std::string startup() const;
 
+  /// @brief Update an indivdual uuid's endpoint
+  void updateEndpoint(std::string const&, std::string const&);
+  
 
 };
 }

arangod/Agency/Inception.cpp

@@ -167,8 +167,16 @@ bool Inception::restartingActiveAgent() {
   auto        pool      = myConfig.pool();
   auto        active    = myConfig.active();
   auto const& clientId  = myConfig.id();
+  auto const& clientEp  = myConfig.endpoint();
   auto const majority   = (myConfig.size()+1)/2;
 
+  Builder greeting;
+  {
+    VPackObjectBuilder b(&greeting);
+    greeting.add(clientId, VPackValue(clientEp));
+  }
+  auto const& greetstr = greeting.toJson();
+
   seconds const timeout(3600);
   
   CONDITION_LOCKER(guard, _cv);
@@ -195,7 +203,7 @@ bool Inception::restartingActiveAgent() {
       if (p.first != myConfig.id() && p.first != "") {
         
         auto comres = arangodb::ClusterComm::instance()->syncRequest(
-          clientId, 1, p.second, rest::RequestType::GET, path, std::string(),
+          clientId, 1, p.second, rest::RequestType::POST, path, greetstr,
           std::unordered_map<std::string, std::string>(), 2.0);
         
         if (comres->status == CL_COMM_SENT) {

arangod/Agency/RestAgencyHandler.cpp

@@ -419,6 +419,16 @@ inline RestStatus RestAgencyHandler::handleRead() {
 }
 
 RestStatus RestAgencyHandler::handleConfig() {
+  if (_request->requestType() == rest::RequestType::POST) {
+    try {
+      arangodb::velocypack::Options options;
+      _agent->updatePeerEndpoint(_request->toVelocyPackBuilderPtr(&options));
+    } catch (std::exception const& e) {
+      generateError(
+        rest::ResponseCode::SERVER_ERROR, TRI_ERROR_INTERNAL, e.what());
+      return RestStatus::DONE;
+    }
+  }
   Builder body;
   body.add(VPackValue(VPackValueType::Object));
   body.add("term", Value(_agent->term()));
@@ -466,7 +476,8 @@ RestStatus RestAgencyHandler::execute() {
       } else if (suffixes[0] == "transact") {
         return handleTransact();
       } else if (suffixes[0] == "config") {
-        if (_request->requestType() != rest::RequestType::GET) {
+        if (_request->requestType() != rest::RequestType::GET &&
+            _request->requestType() != rest::RequestType::POST) {
           return reportMethodNotAllowed();
         }
         return handleConfig();

arangod/Cluster/v8-cluster.cpp

@@ -383,6 +383,9 @@ static void JS_SetAgency(v8::FunctionCallbackInfo<v8::Value> const& args) {
   AgencyComm comm;
   AgencyCommResult result = comm.setValue(key, builder.slice(), ttl);
 
+  LOG(WARN) << key;
+  LOG(WARN) << builder.slice().toJson();
+
   if (!result.successful()) {
     THROW_AGENCY_EXCEPTION(result);
   }

js/server/modules/@arangodb/cluster.js

@@ -40,8 +40,8 @@ var _ = require('lodash');
 
 const inccv = {'/arango/Current/Version':{'op':'increment'}};
 const incpv = {'/arango/Plan/Version':{'op':'increment'}};
-const agencyDBs = global.ArangoAgency.prefix() + '/Current/Databases/';
+const agencyDBs = '/' + global.ArangoAgency.prefix() + '/Current/Databases/';
-const agencyCols = global.ArangoAgency.prefix() + '/Current/Collections/';
+const agencyCols = '/' + global.ArangoAgency.prefix() + '/Current/Collections/';
 
 var endpointToURL = function (endpoint) {
   if (endpoint.substr(0, 6) === 'ssl://') {
@@ -311,7 +311,7 @@ function createLocalDatabases (plannedDatabases, currentDatabases) {
   var createDatabaseAgency = function (payload) {
     var envelope = {};
     envelope[agencyDBs + payload.name + '/' + ourselves] = payload;
-    global.ArangoAgency.write([[envelope, inccv]]);
+    global.ArangoAgency.write([[envelope],[inccv]]);
   };
 
   var db = require('internal').db;
@@ -370,7 +370,7 @@ function dropLocalDatabases (plannedDatabases) {
     try {
       var envelope = {};
       envelope[agencyDBs + payload.name + '/' + ourselves] = {"op":"delete"};
-      global.ArangoAgency.write([[envelope, inccv]]);
+      global.ArangoAgency.write([[envelope],[inccv]]);
     } catch (err) {
       // ignore errors
     }
@@ -425,7 +425,7 @@ function cleanupCurrentDatabases (currentDatabases) {
     try {
       var envelope = {};
       envelope[agencyDBs + payload.name + '/' + ourselves] = {"op":"delete"};
-      global.ArangoAgency.write([[envelope, inccv]]);
+      global.ArangoAgency.write([[envelope],[inccv]]);
     } catch (err) {
       // ignore errors
     }
@@ -470,15 +470,17 @@ function handleDatabaseChanges (plan, current) {
 // / @brief create collections if they exist in the plan but not locally
 // //////////////////////////////////////////////////////////////////////////////
 
-function createLocalCollections (plannedCollections, planVersion,
+function createLocalCollections (
-  currentCollections,
+  plannedCollections, planVersion, currentCollections, takeOverResponsibility) {
-  takeOverResponsibility) {
+
   var ourselves = global.ArangoServerState.id();
 
-  var createCollectionAgency = function (database, shard, collInfo, error) {
+  var createCollectionAgency = function (database, shard, collInfo, err) {
-    var payload = { error: error.error,
+
-      errorNum: error.errorNum,
+    var payload = {
-      errorMessage: error.errorMessage,
+      error: err.error,
+      errorNum: err.errorNum,
+      errorMessage: err.errorMessage,
       satellite: collInfo.replicationFactor === 0,
       indexes: collInfo.indexes,
       servers: [ ourselves ],
@@ -486,18 +488,20 @@ function createLocalCollections (plannedCollections, planVersion,
 
     console.debug('creating Current/Collections/' + database + '/' +
                   collInfo.planId + '/' + shard);
-    global.ArangoAgency.set('Current/Collections/' + database + '/' + collInfo.planId + '/' + shard, payload);
-    global.ArangoAgency.increaseVersion('Current/Version');
-    global.ArangoAgency.increaseVersion('Plan/Version');
 
     var envelope = {};
     envelope[agencyCols + database + '/' + collInfo.planId + '/' + shard] = payload;
-    //global.ArangoAgency.write([[envelope, inccv]]);
+    console.info(envelope);
+    global.ArangoAgency.set('Current/Collections/' + database + '/' +
+                            collInfo.planId + '/' + shard, payload)
+    global.ArangoAgency.write([[inccv]]);
     
     console.debug('creating Current/Collections/' + database + '/' +
                   collInfo.planId + '/' + shard + ' done.');
   };
 
+  var takeOver = createCollectionAgency;
+
   var db = require('internal').db;
   db._useDatabase('_system');
 
@@ -635,7 +639,7 @@ function createLocalCollections (plannedCollections, planVersion,
                     if (error.error) {
                       if (takeOverResponsibility && !didWrite) {
                         if (isLeader) {
-                          createCollectionAgency(database, shard, collInfo, error);
+                          takeOver(database, shard, collInfo, error);
                         }
                       }
                       continue; // No point to look for properties and
@@ -718,7 +722,7 @@ function createLocalCollections (plannedCollections, planVersion,
 
                     if ((takeOverResponsibility && !didWrite && isLeader) ||
                       (!didWrite && isLeader && !wasLeader)) {
-                      createCollectionAgency(database, shard, collInfo, error);
+                      takeOver(database, shard, collInfo, error);
                     }
                   }
                 }
@@ -778,7 +782,7 @@ function dropLocalCollections (plannedCollections, currentCollections) {
                     id + '/' + shardID);
       var envelope = {};
       envelope[agencyCols + database + '/' + id + '/' + shardID] = {"op":"delete"};
-      global.ArangoAgency.write([[envelope, inccv]]);
+      global.ArangoAgency.write([[envelope],[inccv]]);
       console.debug('dropping Current/Collections/' + database + '/' +
                     id + '/' + shardID + ' done.');
     } catch (err) {
@@ -888,7 +892,7 @@ function cleanupCurrentCollections (plannedCollections, currentCollections) {
 
       var envelope = {};
       envelope[agencyCols + database + '/' + collection + '/' + shardID] = {"op": "delete"};
-      global.ArangoAgency.write([[envelope, inccv]]);
+      global.ArangoAgency.write([[envelope],[inccv]]);
       
       console.debug('cleaning Current/Collections/' + database + '/' +
                     collection + '/' + shardID + ' done.');

scripts/startStandAloneAgency.sh

@@ -158,7 +158,7 @@ if [ "$GOSSIP_MODE" = "0" ]; then
    GOSSIP_PEERS=" --agency.endpoint $TRANSPORT://localhost:$BASE"
 fi
 
-rm -rf agency
+#rm -rf agency
 mkdir -p agency
 PIDS=""