Hawkin
/
arangodb
mirror of https://gitee.com/bigwinds/arangodb
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Port agency fixes for ttl to 3.3. (#8554) 

* precondition plan / version in compaction / store TTL removal independent of local _ttl set
* lol through out check if i should initialize
* cannot depend on ServerFeature
* Agency init loops break when shutting down.
* Port little differences from 3.4 back to 3.3 for arangod/Agency/*
* Port agency tests back from 3.4 to 3.3.
* Lose NotifyCallback.cpp
This commit is contained in:
Max Neunhöffer 2019-04-01 16:59:37 +02:00 committed by GitHub
parent 4b1421f187
commit e260226052
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
60 changed files with 663 additions and 593 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
arangod/Agency/ActiveFailoverJob.cpp
View File

@ -286,7 +286,6 @@ std::string ActiveFailoverJob::findBestFollower() {
<< " determining follower ticks";
return "";
}
VPackSlice resp = res.result->slice();
if (!resp.isArray() || resp.length() == 0) {
LOG_TOPIC(ERR, Logger::SUPERVISION)
@ -294,7 +293,8 @@ std::string ActiveFailoverJob::findBestFollower() {
return "";
}
VPackSlice obj = resp.at(0).get({Job::agencyPrefix, "AsyncReplication"});
VPackSlice obj = resp.at(0).get(
std::vector<std::string>({Job::agencyPrefix, std::string("AsyncReplication")}));
for (VPackObjectIterator::ObjectPair pair : VPackObjectIterator(obj)) {
std::string srvUUID = pair.key.copyString();
bool isAvailable =

1
arangod/Agency/ActiveFailoverJob.h
View File

@ -48,6 +48,7 @@ struct ActiveFailoverJob final : public Job {
std::string findBestFollower();
private:
/// @brief the old leader UUID
std::string _server;
};
} // namespace consensus

2
arangod/Agency/AddFollower.h
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");

148
arangod/Agency/AgencyComm.cpp
View File

@ -1,7 +1,7 @@
///////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");
@ -488,7 +488,7 @@ std::vector<std::string> AgencyCommManager::slicePath(std::string const& p1) {
}
std::string AgencyCommManager::generateStamp() {
time_t tt = time(0);
time_t tt = time(nullptr);
struct tm tb;
char buffer[21];
@ -1181,41 +1181,22 @@ AgencyCommResult AgencyComm::sendTransactionWithFailover(AgencyTransaction const
bool AgencyComm::ensureStructureInitialized() {
LOG_TOPIC(TRACE, Logger::AGENCYCOMM) << "checking if agency is initialized";
while (true) {
while (shouldInitializeStructure()) {
while (!application_features::ApplicationServer::isStopping() &&
shouldInitializeStructure()) {
LOG_TOPIC(TRACE, Logger::AGENCYCOMM)
<< "Agency is fresh. Needs initial structure.";
if (tryInitializeStructure()) {
LOG_TOPIC(TRACE, Logger::AGENCYCOMM)
<< "Agency is fresh. Needs initial structure.";
// mop: we are the chosen one .. great success
if (tryInitializeStructure()) {
LOG_TOPIC(TRACE, Logger::AGENCYCOMM)
<< "Successfully initialized agency";
break;
}
LOG_TOPIC(WARN, Logger::AGENCYCOMM)
<< "Initializing agency failed. We'll try again soon";
// We should really have exclusive access, here, this is strange!
std::this_thread::sleep_for(std::chrono::seconds(1));
}
AgencyCommResult result = getValues("InitDone");
if (result.successful()) {
VPackSlice value = result.slice()[0].get(
std::vector<std::string>({AgencyCommManager::path(), "InitDone"}));
if (value.isBoolean() && value.getBoolean()) {
// expecting a value of "true"
LOG_TOPIC(TRACE, Logger::AGENCYCOMM) << "Found an initialized agency";
break;
}
} else {
if (result.httpCode() == 401) {
// unauthorized
LOG_TOPIC(FATAL, Logger::STARTUP) << "Unauthorized. Wrong credentials.";
FATAL_ERROR_EXIT();
}
<< "Successfully initialized agency";
break;
}
LOG_TOPIC(WARN, Logger::AGENCYCOMM)
<< "Initializing agency failed. We'll try again soon";
// We should really have exclusive access, here, this is strange!
std::this_thread::sleep_for(std::chrono::seconds(1));
LOG_TOPIC(TRACE, Logger::AGENCYCOMM)
<< "Waiting for agency to get initialized";
@ -1346,10 +1327,7 @@ AgencyCommResult AgencyComm::sendWithFailover(arangodb::rest::RequestType method
auto waitSomeTime = [&waitInterval, &result]() -> bool {
// Returning true means timeout because of shutdown:
auto serverFeature = application_features::ApplicationServer::getFeature<ServerFeature>(
"Server");
if (serverFeature->isStopping() ||
!application_features::ApplicationServer::isRetryOK()) {
if (!application_features::ApplicationServer::isRetryOK()) {
LOG_TOPIC(INFO, Logger::AGENCYCOMM)
<< "Unsuccessful AgencyComm: Timeout because of shutdown "
<< "errorCode: " << result.errorCode()
@ -1403,10 +1381,30 @@ AgencyCommResult AgencyComm::sendWithFailover(arangodb::rest::RequestType method
// Some reporting:
if (tries > 20) {
auto serverState = application_features::ApplicationServer::server->state();
std::string serverStateStr;
switch (serverState) {
case arangodb::application_features::ServerState::UNINITIALIZED:
case arangodb::application_features::ServerState::IN_COLLECT_OPTIONS:
case arangodb::application_features::ServerState::IN_VALIDATE_OPTIONS:
case arangodb::application_features::ServerState::IN_PREPARE:
case arangodb::application_features::ServerState::IN_START:
serverStateStr = "in startup";
break;
case arangodb::application_features::ServerState::IN_WAIT:
serverStateStr = "running";
break;
case arangodb::application_features::ServerState::IN_STOP:
case arangodb::application_features::ServerState::IN_UNPREPARE:
case arangodb::application_features::ServerState::STOPPED:
case arangodb::application_features::ServerState::ABORT:
serverStateStr = "in shutdown";
}
LOG_TOPIC(INFO, Logger::AGENCYCOMM)
<< "Flaky agency communication to " << endpoint
<< ". Unsuccessful consecutive tries: " << tries << " (" << elapsed
<< "s). Network checks advised.";
<< "s). Network checks advised."
<< " Server " << serverStateStr << ".";
}
if (1 < tries) {
@ -1768,10 +1766,9 @@ bool AgencyComm::tryInitializeStructure() {
LOG_TOPIC(TRACE, Logger::AGENCYCOMM)
<< "Initializing agency with " << builder.toJson();
AgencyOperation initOperation("", AgencyValueOperationType::SET, builder.slice());
AgencyWriteTransaction initTransaction;
initTransaction.operations.push_back(initOperation);
AgencyWriteTransaction initTransaction(
AgencyOperation("", AgencyValueOperationType::SET, builder.slice()),
AgencyPrecondition("Plan", AgencyPrecondition::Type::EMPTY, true));
AgencyCommResult result = sendTransactionWithFailover(initTransaction);
if (result.httpCode() == TRI_ERROR_HTTP_UNAUTHORIZED) {
@ -1792,19 +1789,60 @@ bool AgencyComm::tryInitializeStructure() {
}
bool AgencyComm::shouldInitializeStructure() {
VPackBuilder builder;
builder.add(VPackValue(false));
// "InitDone" key should not previously exist
auto result = casValue("InitDone", builder.slice(), false, 60.0,
AgencyCommManager::CONNECTION_OPTIONS._requestTimeout);
size_t nFail = 0;
while (!application_features::ApplicationServer::isStopping()) {
auto result = getValues("Plan");
if (!result.successful()) { // Not 200 - 299
if (result.httpCode() == 401) {
// unauthorized
LOG_TOPIC(FATAL, Logger::STARTUP) << "Unauthorized. Wrong credentials.";
FATAL_ERROR_EXIT();
}
// Agency not ready yet
LOG_TOPIC(TRACE, Logger::AGENCYCOMM)
<< "waiting for agency to become ready";
continue;
} else {
// Sanity
if (result.slice().isArray() && result.slice().length() == 1) {
// No plan entry? Should initialise
if (result.slice()[0] == VPackSlice::emptyObjectSlice()) {
LOG_TOPIC(DEBUG, Logger::AGENCYCOMM)
<< "agency initialisation should be performed";
return true;
} else {
LOG_TOPIC(DEBUG, Logger::AGENCYCOMM)
<< "agency initialisation under way or done";
return false;
}
} else {
// Should never get here
TRI_ASSERT(false);
if (nFail++ < 3) {
LOG_TOPIC(DEBUG, Logger::AGENCYCOMM) << "What the hell just happened?";
} else {
LOG_TOPIC(FATAL, Logger::AGENCYCOMM)
<< "Illegal response from agency during bootstrap: "
<< result.slice().toJson();
FATAL_ERROR_EXIT();
}
continue;
}
}
std::this_thread::sleep_for(std::chrono::milliseconds(250));
if (!result.successful()) {
// somebody else has or is initializing the agency
LOG_TOPIC(TRACE, Logger::AGENCYCOMM)
<< "someone else is initializing the agency";
return false;
}
return true;
return false;
}

2
arangod/Agency/AgencyComm.h
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");

7
arangod/Agency/AgencyCommon.h
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2018 ArangoDB GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");
/// you may not use this file except in compliance with the License.
@ -120,10 +120,11 @@ struct log_t {
std::string const& clientId = std::string())
: index(idx),
term(t),
entry(std::make_shared<arangodb::velocypack::Buffer<uint8_t>>(*e.get())),
clientId(clientId),
timestamp(std::chrono::duration_cast<std::chrono::milliseconds>(
std::chrono::system_clock::now().time_since_epoch())) {}
std::chrono::system_clock::now().time_since_epoch())) {
entry = std::make_shared<arangodb::velocypack::Buffer<uint8_t>>(*e.get());
}
friend std::ostream& operator<<(std::ostream& o, log_t const& l) {
o << l.index << " " << l.term << " " << VPackSlice(l.entry->data()).toJson() << " "

8
arangod/Agency/AgencyFeature.cpp
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");
@ -32,12 +32,13 @@
#include "ProgramOptions/Section.h"
#include "RestServer/EndpointFeature.h"
using namespace arangodb;
using namespace arangodb::application_features;
using namespace arangodb::basics;
using namespace arangodb::options;
using namespace arangodb::rest;
namespace arangodb {
consensus::Agent* AgencyFeature::AGENT = nullptr;
AgencyFeature::AgencyFeature(application_features::ApplicationServer* server)
@ -239,6 +240,7 @@ void AgencyFeature::start() {
return;
}
// Available after validateOptions of ClusterFeature
// Find the agency prefix:
auto feature = ApplicationServer::getFeature<ClusterFeature>("Cluster");
if (!feature->agencyPrefix().empty()) {
@ -346,3 +348,5 @@ void AgencyFeature::unprepare() {
// shutdown
_agent.reset();
}
} // namespace arangodb

5
arangod/Agency/AgencyFeature.h
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2018 ArangoDB GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");
/// you may not use this file except in compliance with the License.
@ -26,7 +26,9 @@
#include "ApplicationFeatures/ApplicationFeature.h"
namespace arangodb {
namespace consensus {
class Agent;
}
@ -72,6 +74,7 @@ class AgencyFeature : virtual public application_features::ApplicationFeature {
private:
std::unique_ptr<consensus::Agent> _agent;
};
} // namespace arangodb
#endif

222
arangod/Agency/Agent.cpp
View File

@ -1309,11 +1309,11 @@ void Agent::persistConfiguration(term_t t) {
{
VPackObjectBuilder aa(agency.get());
agency->add("term", VPackValue(t));
agency->add(idStr, VPackValue(id()));
agency->add(activeStr, _config.activeToBuilder()->slice());
agency->add(poolStr, _config.poolToBuilder()->slice());
agency->add(config_t::idStr, VPackValue(id()));
agency->add(config_t::activeStr, _config.activeToBuilder()->slice());
agency->add(config_t::poolStr, _config.poolToBuilder()->slice());
agency->add("size", VPackValue(size()));
agency->add(timeoutMultStr, VPackValue(_config.timeoutMult()));
agency->add(config_t::timeoutMultStr, VPackValue(_config.timeoutMult()));
}
}
}
@ -1427,40 +1427,6 @@ int64_t Agent::leaderFor() const {
_leaderSince;
}
// Notify inactive pool members of configuration change()
void Agent::notifyInactive() const {
auto cc = ClusterComm::instance();
if (cc == nullptr) {
// nullptr only happens during controlled shutdown
return;
}
std::unordered_map<std::string, std::string> pool = _config.pool();
std::string path = "/_api/agency_priv/inform";
Builder out;
{
VPackObjectBuilder o(&out);
out.add("term", VPackValue(term()));
out.add("id", VPackValue(id()));
out.add("active", _config.activeToBuilder()->slice());
out.add("pool", _config.poolToBuilder()->slice());
out.add("min ping", VPackValue(_config.minPing()));
out.add("max ping", VPackValue(_config.maxPing()));
out.add("timeoutMult", VPackValue(_config.timeoutMult()));
}
for (auto const& p : pool) {
if (p.first != id()) {
auto headerFields =
std::make_unique<std::unordered_map<std::string, std::string>>();
cc->asyncRequest("1", 1, p.second, arangodb::rest::RequestType::POST,
path, std::make_shared<std::string>(out.toJson()),
headerFields, nullptr, 1.0, true);
}
}
}
void Agent::updatePeerEndpoint(query_t const& message) {
VPackSlice slice = message->slice();
@ -1495,7 +1461,6 @@ void Agent::updatePeerEndpoint(std::string const& id, std::string const& ep) {
if (_config.updateEndpoint(id, ep)) {
if (!challengeLeadership()) {
persistConfiguration(term());
notifyInactive();
}
}
}
@ -1674,13 +1639,13 @@ Store const& Agent::transient() const {
/// Rebuild from persisted state
void Agent::setPersistedState(VPackSlice const& compaction) {
// Catch up with compacted state, this is only called at startup
_spearhead = compaction.get("readDB");
_spearhead = compaction;
// Catch up with commit
try {
WRITE_LOCKER(oLocker, _outputLock);
CONDITION_LOCKER(guard, _waitForCV);
_readDB = compaction.get("readDB");
_readDB = compaction;
_commitIndex =
arangodb::basics::StringUtils::uint64(compaction.get("_key").copyString());
_waitForCV.broadcast();
@ -1771,107 +1736,115 @@ query_t Agent::gossip(query_t const& in, bool isCallback, size_t version) {
query_t out = std::make_shared<Builder>();
VPackObjectBuilder b(out.get());
{
VPackObjectBuilder b(out.get());
std::unordered_set<std::string> gossipPeers = _config.gossipPeers();
if (!gossipPeers.empty() && !isCallback) {
try {
_config.eraseGossipPeer(endpoint);
} catch (std::exception const& e) {
LOG_TOPIC(ERR, Logger::AGENCY) << __FILE__ << ":" << __LINE__ << " " << e.what();
}
}
std::string err;
config_t::upsert_t upsert = config_t::UNCHANGED;
/// Pool incomplete or the other guy is in my pool: I'll gossip.
if (!_config.poolComplete() || _config.matchPeer(id, endpoint)) {
upsert = _config.upsertPool(pslice, id);
if (upsert == config_t::WRONG) {
LOG_TOPIC(FATAL, Logger::AGENCY) << "Discrepancy in agent pool!";
FATAL_ERROR_EXIT(); /// disagreement over pool membership are fatal!
}
// Wrapped in envelope in RestAgencyPrivHandler
auto pool = _config.pool();
out->add(VPackValue("pool"));
{
VPackObjectBuilder bb(out.get());
for (auto const& i : pool) {
out->add(i.first, VPackValue(i.second));
std::unordered_set<std::string> gossipPeers = _config.gossipPeers();
if (!gossipPeers.empty() && !isCallback) {
try {
_config.eraseGossipPeer(endpoint);
} catch (std::exception const& e) {
LOG_TOPIC(ERR, Logger::AGENCY) << __FILE__ << ":" << __LINE__ << " " << e.what();
}
}
} else { // Pool complete & id's endpoint not matching.
std::string err;
config_t::upsert_t upsert = config_t::UNCHANGED;
// Not leader: redirect / 503
if (challengeLeadership()) {
out->add("redirect", VPackValue(true));
out->add("id", VPackValue(leaderID()));
} else { // leader magic
auto tmp = _config;
tmp.upsertPool(pslice, id);
auto query = std::make_shared<VPackBuilder>();
/// Pool incomplete or the other guy is in my pool: I'll gossip.
if (!_config.poolComplete() || _config.matchPeer(id, endpoint)) {
upsert = _config.upsertPool(pslice, id);
if (upsert == config_t::WRONG) {
LOG_TOPIC(FATAL, Logger::AGENCY) << "Discrepancy in agent pool!";
FATAL_ERROR_EXIT(); /// disagreement over pool membership are fatal!
}
// Wrapped in envelope in RestAgencyPrivHandler
auto pool = _config.pool();
out->add(VPackValue("pool"));
{
VPackArrayBuilder trs(query.get());
VPackObjectBuilder bb(out.get());
for (auto const& i : pool) {
out->add(i.first, VPackValue(i.second));
}
}
} else { // Pool complete & id's endpoint not matching.
// Not leader: redirect / 503
if (challengeLeadership()) {
out->add("redirect", VPackValue(true));
out->add("id", VPackValue(leaderID()));
} else { // leader magic
auto tmp = _config;
tmp.upsertPool(pslice, id);
auto query = std::make_shared<VPackBuilder>();
{
VPackArrayBuilder tr(query.get());
VPackArrayBuilder trs(query.get());
{
VPackObjectBuilder o(query.get());
query->add(VPackValue(RECONFIGURE));
VPackArrayBuilder tr(query.get());
{
VPackObjectBuilder o(query.get());
query->add("op", VPackValue("set"));
query->add(VPackValue("new"));
query->add(VPackValue(RECONFIGURE));
{
VPackObjectBuilder c(query.get());
tmp.toBuilder(*query);
VPackObjectBuilder o(query.get());
query->add("op", VPackValue("set"));
query->add(VPackValue("new"));
{
VPackObjectBuilder c(query.get());
tmp.toBuilder(*query);
}
}
}
}
}
LOG_TOPIC(DEBUG, Logger::AGENCY)
<< "persisting new agency configuration via RAFT: " << query->toJson();
// Do write
write_ret_t ret;
try {
ret = write(query, WriteMode(false, true));
arangodb::consensus::index_t max_index = 0;
if (ret.indices.size() > 0) {
max_index = *std::max_element(ret.indices.begin(), ret.indices.end());
}
if (max_index > 0) { // We have a RAFT index. Wait for the RAFT commit.
auto result = waitFor(max_index);
if (result != Agent::raft_commit_t::OK) {
err =
"failed to retrieve RAFT index for updated agency endpoints";
} else {
auto pool = _config.pool();
out->add(VPackValue("pool"));
{
VPackObjectBuilder bb(out.get());
for (auto const& i : pool) {
out->add(i.first, VPackValue(i.second));
}
}
}
} else {
err = "failed to retrieve RAFT index for updated agency endpoints";
}
} catch (std::exception const& e) {
err = std::string("failed to write new agency to RAFT") + e.what();
LOG_TOPIC(ERR, Logger::AGENCY) << err;
}
}
LOG_TOPIC(DEBUG, Logger::AGENCY)
<< "persisting new agency configuration via RAFT: " << query->toJson();
// Do write
write_ret_t ret;
try {
ret = write(query, WriteMode(false, true));
arangodb::consensus::index_t max_index = 0;
if (ret.indices.size() > 0) {
max_index = *std::max_element(ret.indices.begin(), ret.indices.end());
}
if (max_index > 0) { // We have a RAFT index. Wait for the RAFT commit.
auto result = waitFor(max_index);
if (result != Agent::raft_commit_t::OK) {
err = "failed to retrieve RAFT index for updated agency endpoints";
} else {
auto pool = _config.pool();
out->add(VPackValue("pool"));
{
VPackObjectBuilder bb(out.get());
for (auto const& i : pool) {
out->add(i.first, VPackValue(i.second));
}
}
}
} else {
err = "failed to retrieve RAFT index for updated agency endpoints";
}
} catch (std::exception const& e) {
err = std::string("failed to write new agency to RAFT") + e.what();
LOG_TOPIC(ERR, Logger::AGENCY) << err;
if (!err.empty()) {
out->add(StaticStrings::Code, VPackValue(500));
out->add(StaticStrings::Error, VPackValue(true));
out->add(StaticStrings::ErrorMessage, VPackValue(err));
out->add(StaticStrings::ErrorNum, VPackValue(500));
}
}
if (!err.empty()) {
out->add(StaticStrings::Code, VPackValue(500));
out->add(StaticStrings::Error, VPackValue(true));
out->add(StaticStrings::ErrorMessage, VPackValue(err));
out->add(StaticStrings::ErrorNum, VPackValue(500));
// let gossip loop know that it has new data
if (_inception != nullptr && upsert == config_t::CHANGED) {
_inception->signalConditionVar();
}
}
@ -1879,11 +1852,6 @@ query_t Agent::gossip(query_t const& in, bool isCallback, size_t version) {
LOG_TOPIC(TRACE, Logger::AGENCY) << "Answering with gossip " << out->slice().toJson();
}
// let gossip loop know that it has new data
if (_inception != nullptr && upsert == config_t::CHANGED) {
_inception->signalConditionVar();
}
return out;
}

5
arangod/Agency/Agent.h
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");
@ -322,9 +322,6 @@ class Agent final : public arangodb::Thread, public AgentInterface {
/// @brief Find out, if we've had acknowledged RPCs recent enough
bool challengeLeadership();
/// @brief Notify inactive pool members of changes in configuration
void notifyInactive() const;
/// @brief Leader election delegate
Constituent _constituent;

10
arangod/Agency/AgentCallback.cpp
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");
@ -24,12 +24,14 @@
#include "AgentCallback.h"
#include "Agency/Agent.h"
#include "ApplicationFeatures/ApplicationServer.h"
using namespace arangodb::application_features;
using namespace arangodb::consensus;
using namespace arangodb::velocypack;
AgentCallback::AgentCallback()
: _agent(0), _last(0), _toLog(0), _startTime(0.0) {}
: _agent(nullptr), _last(0), _toLog(0), _startTime(0.0) {}
AgentCallback::AgentCallback(Agent* agent, std::string const& slaveID, index_t last, size_t toLog)
: _agent(agent),
@ -38,7 +40,7 @@ AgentCallback::AgentCallback(Agent* agent, std::string const& slaveID, index_t l
_toLog(toLog),
_startTime(TRI_microtime()) {}
void AgentCallback::shutdown() { _agent = 0; }
void AgentCallback::shutdown() { _agent = nullptr; }
bool AgentCallback::operator()(arangodb::ClusterCommResult* res) {
if (res->status == CL_COMM_SENT) {
@ -88,7 +90,7 @@ bool AgentCallback::operator()(arangodb::ClusterCommResult* res) {
<< "comm_status(" << res->status << "), last(" << _last << "), follower("
<< _slaveID << "), time(" << TRI_microtime() - _startTime << ")";
} else {
if (_agent == nullptr || !_agent->isStopping()) {
if (!ApplicationServer::isStopping() && (_agent == nullptr || !_agent->isStopping())) {
// Do not warn if we are already shutting down:
LOG_TOPIC(WARN, Logger::AGENCY)
<< "Got bad callback from AppendEntriesRPC: "

2
arangod/Agency/AgentCallback.h
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");

22
arangod/Agency/AgentConfiguration.cpp
View File

@ -28,6 +28,28 @@
using namespace arangodb::consensus;
std::string const config_t::idStr = "id";
std::string const config_t::agencySizeStr = "agency size";
std::string const config_t::poolSizeStr = "pool size";
std::string const config_t::minPingStr = "min ping";
std::string const config_t::maxPingStr = "max ping";
std::string const config_t::timeoutMultStr = "timeoutMult";
std::string const config_t::endpointStr = "endpoint";
std::string const config_t::uuidStr = "uuid";
std::string const config_t::poolStr = "pool";
std::string const config_t::gossipPeersStr = "gossipPeers";
std::string const config_t::activeStr = "active";
std::string const config_t::supervisionStr = "supervision";
std::string const config_t::waitForSyncStr = "wait for sync";
std::string const config_t::supervisionFrequencyStr = "supervision frequency";
std::string const config_t::supervisionGracePeriodStr =
"supervision grace period";
std::string const config_t::compactionStepSizeStr = "compaction step size";
std::string const config_t::compactionKeepSizeStr = "compaction keep size";
std::string const config_t::defaultEndpointStr = "tcp://localhost:8529";
std::string const config_t::versionStr = "version";
std::string const config_t::startupStr = "startup";
config_t::config_t()
: _agencySize(0),
_poolSize(0),

45
arangod/Agency/AgentConfiguration.h
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");
@ -36,27 +36,6 @@
namespace arangodb {
namespace consensus {
static const std::string idStr = "id";
static const std::string agencySizeStr = "agency size";
static const std::string poolSizeStr = "pool size";
static const std::string minPingStr = "min ping";
static const std::string maxPingStr = "max ping";
static const std::string timeoutMultStr = "timeoutMult";
static const std::string endpointStr = "endpoint";
static const std::string uuidStr = "uuid";
static const std::string poolStr = "pool";
static const std::string gossipPeersStr = "gossipPeers";
static const std::string activeStr = "active";
static const std::string supervisionStr = "supervision";
static const std::string waitForSyncStr = "wait for sync";
static const std::string supervisionFrequencyStr = "supervision frequency";
static const std::string supervisionGracePeriodStr = "supervision grace period";
static const std::string compactionStepSizeStr = "compaction step size";
static const std::string compactionKeepSizeStr = "compaction keep size";
static const std::string defaultEndpointStr = "tcp://localhost:8529";
static const std::string versionStr = "version";
static const std::string startupStr = "startup";
struct config_t {
private:
std::string _id;
@ -81,9 +60,31 @@ struct config_t {
size_t _version;
std::string _startup;
size_t _maxAppendSize;
mutable arangodb::basics::ReadWriteLock _lock; // guard member variables
public:
static std::string const idStr;
static std::string const agencySizeStr;
static std::string const poolSizeStr;
static std::string const minPingStr;
static std::string const maxPingStr;
static std::string const timeoutMultStr;
static std::string const endpointStr;
static std::string const uuidStr;
static std::string const poolStr;
static std::string const gossipPeersStr;
static std::string const activeStr;
static std::string const supervisionStr;
static std::string const waitForSyncStr;
static std::string const supervisionFrequencyStr;
static std::string const supervisionGracePeriodStr;
static std::string const compactionStepSizeStr;
static std::string const compactionKeepSizeStr;
static std::string const defaultEndpointStr;
static std::string const versionStr;
static std::string const startupStr;
using upsert_t = enum { UNCHANGED = 0, CHANGED, WRONG };
/// @brief default ctor

2
arangod/Agency/AgentInterface.h
View File

@ -58,7 +58,7 @@ class AgentInterface {
virtual bool isCommitted(index_t last_entry) = 0;
// Suffice warnings
virtual ~AgentInterface(){};
virtual ~AgentInterface() {}
};
} // namespace consensus
} // namespace arangodb

2
arangod/Agency/CleanOutServer.h
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");

4
arangod/Agency/Compactor.cpp
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");
@ -66,7 +66,7 @@ void Compactor::run() {
}
} catch (std::exception const& e) {
LOG_TOPIC(ERR, Logger::AGENCY)
<< "Expection during compaction, details: " << e.what();
<< "Exception during compaction, details: " << e.what();
}
}
}

2
arangod/Agency/Compactor.h
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");

15
arangod/Agency/Constituent.cpp
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");
@ -112,7 +112,7 @@ void Constituent::termNoLock(term_t t, std::string const& votedFor) {
if (tmp != t || tmpVotedFor != votedFor) {
LOG_TOPIC(INFO, Logger::AGENCY) << _id << ": changing term or votedFor, "
<< "current role:" << roleStr[_role]
<< "current role: " << roleStr[_role]
<< " term " << t << " votedFor: " << votedFor;
Builder body;
@ -128,17 +128,19 @@ void Constituent::termNoLock(term_t t, std::string const& votedFor) {
TRI_ASSERT(_vocbase != nullptr);
auto ctx = transaction::StandaloneContext::Create(_vocbase);
SingleCollectionTransaction trx(ctx, "election", AccessMode::Type::WRITE);
Result res = trx.begin();
if (!res.ok()) {
THROW_ARANGO_EXCEPTION(res);
}
OperationOptions options;
options.waitForSync = _agent->config().waitForSync();
options.silent = true;
OperationResult result;
if (tmp != t) {
try {
result = trx.insert("election", body.slice(), options);
@ -496,10 +498,9 @@ void Constituent::callElection() {
}
if (!isStopping() && cc != nullptr) {
auto res = ClusterComm::instance()->wait(
"", coordinatorTransactionID, 0, "",
duration<double>(timeout - steady_clock::now()).count());
auto res = cc->wait("", coordinatorTransactionID, 0, "",
duration<double>(timeout - steady_clock::now()).count());
if (res.status == CL_COMM_SENT) {
auto body = res.result->getBodyVelocyPack();
VPackSlice slc = body->slice();

4
arangod/Agency/Constituent.h
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");
@ -156,7 +156,7 @@ class Constituent : public Thread {
std::string _id; // My own id
// Last time an AppendEntriesRPC message has arrived, this is used to
// organise out-of-patience in the follower:
// organize out-of-patience in the follower:
std::atomic<double> _lastHeartbeatSeen;
role_t _role; // My role

11
arangod/Agency/FailedFollower.cpp
View File

@ -94,9 +94,11 @@ bool FailedFollower::create(std::shared_ptr<VPackBuilder> envelope) {
} else {
_jb = envelope;
}
// Todo entry
_jb->add(VPackValue(toDoPrefix + _jobId));
{ VPackObjectBuilder todo(_jb.get());
{
VPackObjectBuilder todo(_jb.get());
_jb->add("creator", VPackValue(_creator));
_jb->add("type", VPackValue("failedFollower"));
_jb->add("database", VPackValue(_database));
@ -104,7 +106,8 @@ bool FailedFollower::create(std::shared_ptr<VPackBuilder> envelope) {
_jb->add("shard", VPackValue(_shard));
_jb->add("fromServer", VPackValue(_from));
_jb->add("jobId", VPackValue(_jobId));
_jb->add("timeCreated", VPackValue(timepointToString(_created))); }
_jb->add("timeCreated", VPackValue(timepointToString(_created)));
}
if (envelope == nullptr) {
_jb->close(); // object
@ -112,7 +115,7 @@ bool FailedFollower::create(std::shared_ptr<VPackBuilder> envelope) {
write_ret_t res = singleWriteTransaction(_agent, *_jb);
return (res.accepted && res.indices.size() == 1 && res.indices[0]);
}
return true;
}

2
arangod/Agency/FailedFollower.h
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");

22
arangod/Agency/FailedLeader.cpp
View File

@ -150,8 +150,8 @@ bool FailedLeader::create(std::shared_ptr<VPackBuilder> b) {
_jb->add("shard", VPackValue(_shard));
_jb->add("fromServer", VPackValue(_from));
_jb->add("jobId", VPackValue(_jobId));
_jb->add("timeCreated", VPackValue(timepointToString(system_clock::now())));}
_jb->add("timeCreated", VPackValue(timepointToString(system_clock::now())));
}
if (b == nullptr) {
_jb->close(); // object
@ -239,7 +239,7 @@ bool FailedLeader::start(bool& aborts) {
// Transactions
Builder pending;
{
VPackArrayBuilder transactions(&pending);
{
@ -432,22 +432,6 @@ JOB_STATUS FailedLeader::status() {
}
if (done) {
// Remove shard to /arango/Target/FailedServers/<server> array
Builder del;
{
VPackArrayBuilder a(&del);
{
VPackObjectBuilder o(&del);
del.add(VPackValue(failedServersPrefix + "/" + _from));
{
VPackObjectBuilder erase(&del);
del.add("op", VPackValue("erase"));
del.add("val", VPackValue(_shard));
}
}
}
write_ret_t res = singleWriteTransaction(_agent, del);
if (finish("", shard)) {
LOG_TOPIC(INFO, Logger::SUPERVISION)
<< "Finished failedLeader for " + _shard + " from " + _from + " to " + _to;

2
arangod/Agency/FailedLeader.h
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");

12
arangod/Agency/FailedServer.cpp
View File

@ -107,7 +107,6 @@ bool FailedServer::start(bool& aborts) {
}
} // Todo entry
// Pending entry
auto transactions = std::make_shared<VPackBuilder>();
{
VPackArrayBuilder a(transactions.get());
@ -115,13 +114,16 @@ bool FailedServer::start(bool& aborts) {
// Operations -------------->
{
VPackObjectBuilder oper(transactions.get());
// Add pending
auto const& databases = _snapshot.hasAsChildren("/Plan/Collections").first;
// auto const& current = _snapshot.hasAsChildren("/Current/Collections").first;
size_t sub = 0;
// FIXME: looks OK, but only the non-clone shards are put into the job
for (auto const& database : databases) {
// dead code auto cdatabase = current.at(database.first)->children();
for (auto const& collptr : database.second->children()) {
auto const& collection = *(collptr.second);
@ -191,7 +193,7 @@ bool FailedServer::start(bool& aborts) {
{
VPackObjectBuilder ts(transactions.get());
transactions->add("timeStarted",
VPackValue(timepointToString(system_clock::now())));
VPackValue(timepointToString(system_clock::now())));
for (auto const& obj : VPackObjectIterator(todo.slice()[0])) {
transactions->add(obj.key.copyString(), obj.value);
}
@ -217,18 +219,20 @@ bool FailedServer::start(bool& aborts) {
if (res.accepted && res.indices.size() == 1 && res.indices[0]) {
LOG_TOPIC(DEBUG, Logger::SUPERVISION)
<< "Pending job for failed DB Server " << _server;
return true;
}
LOG_TOPIC(INFO, Logger::SUPERVISION)
<< "Precondition failed for starting FailedServer " + _jobId;
<< "Precondition failed for starting FailedServer " + _jobId;
return false;
}
bool FailedServer::create(std::shared_ptr<VPackBuilder> envelope) {
LOG_TOPIC(DEBUG, Logger::SUPERVISION)
<< "Todo: Handle failover for db server " + _server;
<< "Todo: Handle failover for db server " + _server;
using namespace std::chrono;
bool selfCreate = (envelope == nullptr); // Do we create ourselves?

2
arangod/Agency/FailedServer.h
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");

2
arangod/Agency/GossipCallback.cpp
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");

2
arangod/Agency/GossipCallback.h
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");

21
arangod/Agency/Inception.cpp
View File

@ -28,7 +28,6 @@
#include "Basics/ConditionLocker.h"
#include "Cluster/ClusterComm.h"
#include "Cluster/ServerState.h"
#include "GeneralServer/RestHandlerFactory.h"
#include <chrono>
#include <numeric>
@ -245,10 +244,19 @@ bool Inception::restartingActiveAgent() {
auto comres =
cc->syncRequest(clientId, 1, p, rest::RequestType::POST, path, greetstr,
std::unordered_map<std::string, std::string>(), 2.0);
if (comres->status == CL_COMM_SENT) {
if (comres->status == CL_COMM_SENT && comres->result->getHttpReturnCode() == 200) {
auto const theirConfigVP = comres->result->getBodyVelocyPack();
auto const& theirConfig = theirConfigVP->slice();
if (!theirConfig.isObject()) {
continue;
}
auto const& tcc = theirConfig.get("configuration");
if (!tcc.isObject() || !tcc.hasKey("id")) {
continue;
}
auto const& theirId = tcc.get("id").copyString();
_agent->updatePeerEndpoint(theirId, p);
@ -405,7 +413,8 @@ void Inception::reportVersionForEp(std::string const& endpoint, size_t version)
// @brief Thread main
void Inception::run() {
while (ServerState::isMaintenance() && !this->isStopping() && !_agent->isStopping()) {
auto server = ServerState::instance();
while (server->isMaintenance() && !this->isStopping() && !_agent->isStopping()) {
std::this_thread::sleep_for(std::chrono::microseconds(1000000));
LOG_TOPIC(DEBUG, Logger::AGENCY)
<< "Waiting for RestHandlerFactory to exit maintenance mode before we "
@ -417,8 +426,7 @@ void Inception::run() {
// Are we starting from persisted pool?
if (config.startup() == "persistence") {
if (restartingActiveAgent()) {
LOG_TOPIC(INFO, Logger::AGENCY)
<< "Activating agent with pool " << _agent->config().pool();
LOG_TOPIC(INFO, Logger::AGENCY) << "Activating agent.";
_agent->ready(true);
} else {
if (!this->isStopping()) {
@ -443,8 +451,7 @@ void Inception::run() {
}
}
LOG_TOPIC(INFO, Logger::AGENCY)
<< "Activating agent with pool " << _agent->config().pool();
LOG_TOPIC(INFO, Logger::AGENCY) << "Activating agent.";
_agent->ready(true);
}

2
arangod/Agency/Inception.h
View File

@ -40,7 +40,7 @@ namespace consensus {
class Agent;
/// @brief This class organises the startup of the agency until the point
/// @brief This class organizes the startup of the agency until the point
/// where the RAFT implementation can commence function
class Inception : public Thread {
public:

2
arangod/Agency/JobContext.cpp
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");

2
arangod/Agency/JobContext.h
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");

9
arangod/Agency/MoveShard.cpp
View File

@ -42,7 +42,8 @@ MoveShard::MoveShard(Node const& snapshot, AgentInterface* agent,
_from(id(from)),
_to(id(to)),
_isLeader(isLeader), // will be initialized properly when information known
_remainsFollower(remainsFollower) {}
_remainsFollower(remainsFollower),
_toServerIsFollower(false) {}
MoveShard::MoveShard(Node const& snapshot, AgentInterface* agent, std::string const& jobId,
std::string const& creator, std::string const& database,
@ -55,7 +56,8 @@ MoveShard::MoveShard(Node const& snapshot, AgentInterface* agent, std::string co
_from(id(from)),
_to(id(to)),
_isLeader(isLeader), // will be initialized properly when information known
_remainsFollower(isLeader) {}
_remainsFollower(isLeader),
_toServerIsFollower(false) {}
MoveShard::MoveShard(Node const& snapshot, AgentInterface* agent,
JOB_STATUS status, std::string const& jobId)
@ -625,6 +627,9 @@ JOB_STATUS MoveShard::pendingLeader() {
pre.add(plan);
}
});
if (!_remainsFollower) {
addIncreasePlanVersion(trx);
}
addPreconditionCollectionStillThere(pre, _database, _collection);
addRemoveJobFromSomewhere(trx, "Pending", _jobId);
Builder job;

2
arangod/Agency/MoveShard.h
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");

44
arangod/Agency/Node.cpp
View File

@ -1,7 +1,7 @@
///////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");
@ -46,6 +46,7 @@ struct Empty {
};
const Node::Children Node::dummyChildren = Node::Children();
const Node Node::_dummyNode = Node("dumm-di-dumm");
/// @brief Split strings by separator
inline static std::vector<std::string> split(const std::string& str, char separator) {
@ -390,11 +391,21 @@ bool Node::addTimeToLive(long millis) {
return true;
}
void Node::timeToLive(TimePoint const& ttl) {
_ttl = ttl;
}
TimePoint const& Node::timeToLive() const {
return _ttl;
}
// remove time to live entry for this node
bool Node::removeTimeToLive() {
if (_ttl != std::chrono::system_clock::time_point()) {
store().removeTTL(uri());
_ttl = std::chrono::system_clock::time_point();
if (_store != nullptr) {
_store->removeTTL(uri());
if (_ttl != std::chrono::system_clock::time_point()) {
_ttl = std::chrono::system_clock::time_point();
}
}
return true;
}
@ -906,6 +917,13 @@ bool Node::isInt() const {
return slice().isInt() || slice().isSmallInt();
}
bool Node::isNumber() const {
if (type() == NODE) {
return false;
}
return slice().isNumber();
}
double Node::getDouble() const {
if (type() == NODE) {
throw StoreException("Must not convert NODE type to double");
@ -990,8 +1008,10 @@ std::pair<uint64_t, bool> Node::hasAsUInt(std::string const& url) const {
// retrieve node, throws if does not exist
try {
Node const& target(operator()(url));
ret_pair.first = target.getUInt();
ret_pair.second = true;
if (target.isNumber()) {
ret_pair.first = target.getUInt();
ret_pair.second = true;
}
} catch (...) {
// do nothing, ret_pair second already false
LOG_TOPIC(DEBUG, Logger::SUPERVISION)
@ -1007,8 +1027,10 @@ std::pair<bool, bool> Node::hasAsBool(std::string const& url) const {
// retrieve node, throws if does not exist
try {
Node const& target(operator()(url));
ret_pair.first = target.getBool();
ret_pair.second = true;
if (target.isBool()) {
ret_pair.first = target.getBool();
ret_pair.second = true;
}
} catch (...) {
// do nothing, ret_pair second already false
LOG_TOPIC(DEBUG, Logger::SUPERVISION)
@ -1026,8 +1048,10 @@ std::pair<std::string, bool> Node::hasAsString(std::string const& url) const {
// retrieve node, throws if does not exist
try {
Node const& target(operator()(url));
ret_pair.first = target.getString();
ret_pair.second = true;
if (target.isString()) {
ret_pair.first = target.getString();
ret_pair.second = true;
}
} catch (...) {
// do nothing, ret_pair second already false
LOG_TOPIC(DEBUG, Logger::SUPERVISION)

67
arangod/Agency/Node.h
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");
@ -213,12 +213,15 @@ class Node {
/// @brief Part of relative path which exists
bool has(std::string const&) const;
/// @brief Is UInt
/// @brief Is Int
bool isInt() const;
/// @brief Is UInt
bool isUInt() const;
/// @brief Is number
bool isNumber() const;
/// @brief Is boolean
bool isBool() const;
@ -228,6 +231,18 @@ class Node {
/// @brief Is string
bool isString() const;
/**
* @brief Get seconds this node still has to live. (Must be guarded by caller)
* @return seconds to live (int64_t::max, if none set)
*/
TimePoint const& timeToLive() const;
/**
* @brief Set expiry for this node
* @param Time point of expiry
*/
void timeToLive(TimePoint const& ttl);
/// @brief accessor to Node object
/// @return second is true if url exists, first populated if second true
std::pair<Node const&, bool> hasAsNode(std::string const&) const;
@ -277,42 +292,47 @@ class Node {
// These two operator() functions could be "protected" once
// unit tests updated.
//
public:
/// @brief Get node specified by path string
Node& operator()(std::string const& path);
/// @brief Get node specified by path string
Node const& operator()(std::string const& path) const;
//
// The protected accessors are the "old" interface. They throw.
// Please use the hasAsXXX replacements.
//
/// @brief Get node specified by path string, always throw if not there
Node const& get(std::string const& path) const;
/// @brief Get integer value (throws if type NODE or if conversion fails)
int64_t getInt() const;
/// @brief Get insigned value (throws if type NODE or if conversion fails)
uint64_t getUInt() const;
/// @brief Get bool value (throws if type NODE or if conversion fails)
bool getBool() const;
/// @brief Get double value (throws if type NODE or if conversion fails)
double getDouble() const;
/// @brief Get string value (throws if type NODE or if conversion fails)
std::string getString() const;
/// @brief Get array value
Slice getArray() const;
/// @brief Get insigned value (throws if type NODE or if conversion fails)
uint64_t getUInt() const;
//
// The protected accessors are the "old" interface. They throw.
// Please use the hasAsXXX replacements.
//
protected:
/// @brief Get node specified by path string, always throw if not there
Node const& get(std::string const& path) const;
/// @brief Get integer value (throws if type NODE or if conversion fails)
int64_t getInt() const;
/// @brief Get bool value (throws if type NODE or if conversion fails)
bool getBool() const;
/// @brief Get double value (throws if type NODE or if conversion fails)
double getDouble() const;
public:
/// @brief Clear key value store
void clear();
// @brief Helper function to return static instance of dummy node below
static Node const& dummyNode() {
return _dummyNode;
}
protected:
/// @brief Add time to live entry
virtual bool addTimeToLive(long millis);
@ -332,6 +352,7 @@ class Node {
mutable bool _vecBufDirty;
bool _isArray;
static Children const dummyChildren;
static Node const _dummyNode;
};

33
arangod/Agency/NotifyCallback.cpp
View File

@ -1,33 +0,0 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");
/// you may not use this file except in compliance with the License.
/// You may obtain a copy of the License at
///
/// http://www.apache.org/licenses/LICENSE-2.0
///
/// Unless required by applicable law or agreed to in writing, software
/// distributed under the License is distributed on an "AS IS" BASIS,
/// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
/// See the License for the specific language governing permissions and
/// limitations under the License.
///
/// Copyright holder is ArangoDB GmbH, Cologne, Germany
///
/// @author Andreas Streichardt
////////////////////////////////////////////////////////////////////////////////
#include "NotifyCallback.h"
using namespace arangodb::consensus;
NotifyCallback::NotifyCallback(std::function<void(bool)> const& cb) : _cb(cb) {}
bool NotifyCallback::operator()(arangodb::ClusterCommResult* res) {
_cb(res->status == CL_COMM_SENT && res->result->getHttpReturnCode() == 200);
return true;
}

49
arangod/Agency/NotifyCallback.h
View File

@ -1,49 +0,0 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");
/// you may not use this file except in compliance with the License.
/// You may obtain a copy of the License at
///
/// http://www.apache.org/licenses/LICENSE-2.0
///
/// Unless required by applicable law or agreed to in writing, software
/// distributed under the License is distributed on an "AS IS" BASIS,
/// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
/// See the License for the specific language governing permissions and
/// limitations under the License.
///
/// Copyright holder is ArangoDB GmbH, Cologne, Germany
///
/// @author Andreas Streichardt
////////////////////////////////////////////////////////////////////////////////
#ifndef ARANGOD_CONSENSUS_NOTIFY_CALLBACK_H
#define ARANGOD_CONSENSUS_NOTIFY_CALLBACK_H 1
#include "Agency/AgencyCommon.h"
#include "Cluster/ClusterComm.h"
#include <velocypack/velocypack-aliases.h>
namespace arangodb {
namespace consensus {
class NotifyCallback : public arangodb::ClusterCommCallback {
public:
explicit NotifyCallback(std::function<void(bool)> const&);
virtual bool operator()(arangodb::ClusterCommResult*) override final;
void shutdown();
private:
std::function<void(bool)> _cb;
};
} // namespace consensus
} // namespace arangodb
#endif

6
arangod/Agency/RemoveFollower.cpp
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2016-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2018-2018 ArangoDB GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");
/// you may not use this file except in compliance with the License.
@ -68,7 +68,7 @@ RemoveFollower::~RemoveFollower() {}
void RemoveFollower::run(bool& aborts) { runHelper("", _shard, aborts); }
bool RemoveFollower::create(std::shared_ptr<VPackBuilder> envelope) {
LOG_TOPIC(INFO, Logger::SUPERVISION)
LOG_TOPIC(DEBUG, Logger::SUPERVISION)
<< "Todo: RemoveFollower(s) "
<< " to shard " << _shard << " in collection " << _collection;
@ -414,7 +414,7 @@ bool RemoveFollower::start(bool&) {
if (res.accepted && res.indices.size() == 1 && res.indices[0]) {
_status = FINISHED;
LOG_TOPIC(INFO, Logger::SUPERVISION)
LOG_TOPIC(DEBUG, Logger::SUPERVISION)
<< "Pending: RemoveFollower(s) to shard " << _shard << " in collection "
<< _collection;
return true;

2
arangod/Agency/RemoveFollower.h
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");

16
arangod/Agency/RestAgencyHandler.cpp
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");
@ -564,18 +564,10 @@ RestStatus RestAgencyHandler::handleConfig() {
RestStatus RestAgencyHandler::handleState() {
VPackBuilder body;
body.add(VPackValue(VPackValueType::Array));
for (auto const& i : _agent->state().get()) {
body.add(VPackValue(VPackValueType::Object));
body.add("index", VPackValue(i.index));
body.add("term", VPackValue(i.term));
if (i.entry != nullptr) {
body.add("query", VPackSlice(i.entry->data()));
}
body.add("clientId", VPackValue(i.clientId));
body.close();
{
VPackObjectBuilder o(&body);
_agent->readDB(body);
}
body.close();
generateResult(rest::ResponseCode::OK, body.slice());
return RestStatus::DONE;
}

2
arangod/Agency/RestAgencyHandler.h
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");

2
arangod/Agency/RestAgencyPrivHandler.cpp
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");

2
arangod/Agency/RestAgencyPrivHandler.h
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");

62
arangod/Agency/State.cpp
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");
@ -104,12 +104,15 @@ bool State::persist(index_t index, term_t term, arangodb::velocypack::Slice cons
SingleCollectionTransaction trx(ctx, "log", AccessMode::Type::WRITE);
trx.addHint(transaction::Hints::Hint::SINGLE_OPERATION);
Result res = trx.begin();
if (!res.ok()) {
THROW_ARANGO_EXCEPTION(res);
}
OperationResult result;
try {
result = trx.insert("log", body.slice(), _options);
} catch (std::exception const& e) {
@ -373,7 +376,7 @@ index_t State::logFollower(query_t const& transactions) {
// Now we must completely erase our log and compaction snapshots and
// start from the snapshot
Store snapshot(_agent, "snapshot");
snapshot = slices[0].get("readDB");
snapshot = slices[0];
if (!storeLogFromSnapshot(snapshot, snapshotIndex, snapshotTerm)) {
LOG_TOPIC(FATAL, Logger::AGENCY)
<< "Could not restore received log snapshot.";
@ -491,6 +494,7 @@ size_t State::removeConflicts(query_t const& transactions, bool gotSnapshot) {
LOG_TOPIC(TRACE, Logger::AGENCY) << "removeConflicts done: ndups=" << ndups
<< " first log entry: " << _log.front().index
<< " last log entry: " << _log.back().index;
break;
} else {
++ndups;
@ -706,8 +710,7 @@ bool State::createCollection(std::string const& name) {
body.add("isSystem", VPackValue(LogicalCollection::IsSystemName(name)));
}
arangodb::LogicalCollection const* collection =
_vocbase->createCollection(body.slice());
auto collection = _vocbase->createCollection(body.slice());
if (collection == nullptr) {
THROW_ARANGO_EXCEPTION_MESSAGE(TRI_errno(), "cannot create collection");
@ -734,7 +737,7 @@ bool State::loadCollections(TRI_vocbase_t* vocbase,
MUTEX_LOCKER(logLock, _logLock);
if (_log.empty()) {
std::shared_ptr<Buffer<uint8_t>> buf = std::make_shared<Buffer<uint8_t>>();
VPackSlice value = arangodb::basics::VelocyPackHelper::EmptyObjectValue();
VPackSlice value = arangodb::velocypack::Slice::emptyObjectSlice();
buf->append(value.startAs<char const>(), value.byteSize());
_log.push_back(log_t(index_t(0), term_t(0), buf, std::string()));
persist(0, 0, value, std::string());
@ -795,7 +798,7 @@ bool State::loadLastCompactedSnapshot(Store& store, index_t& index, term_t& term
VPackSlice i = result[0];
VPackSlice ii = i.resolveExternals();
try {
store = ii.get("readDB");
store = ii;
index = basics::StringUtils::uint64(ii.get("_key").copyString());
term = ii.get("term").getNumber<uint64_t>();
return true;
@ -932,7 +935,6 @@ bool State::loadOrPersistConfiguration() {
auto ctx = std::make_shared<transaction::StandaloneContext>(_vocbase);
SingleCollectionTransaction trx(ctx, "configuration", AccessMode::Type::WRITE);
Result res = trx.begin();
OperationResult result;
@ -991,6 +993,7 @@ bool State::loadRemaining() {
// We know that _cur has been set in loadCompacted to the index of the
// snapshot that was loaded or to 0 if there is no snapshot.
index_t lastIndex = _cur;
for (auto const& i : VPackArrayIterator(result)) {
buffer_t tmp = std::make_shared<arangodb::velocypack::Buffer<uint8_t>>();
@ -1009,7 +1012,7 @@ bool State::loadRemaining() {
// Empty patches :
if (index > lastIndex + 1) {
std::shared_ptr<Buffer<uint8_t>> buf = std::make_shared<Buffer<uint8_t>>();
VPackSlice value = arangodb::basics::VelocyPackHelper::EmptyObjectValue();
VPackSlice value = arangodb::velocypack::Slice::emptyObjectSlice();
buf->append(value.startAs<char const>(), value.byteSize());
term_t term(ii.get("term").getNumber<uint64_t>());
for (index_t i = lastIndex + 1; i < index; ++i) {
@ -1162,6 +1165,7 @@ bool State::compactPersisted(index_t cind, index_t keep) {
bindVars->close();
std::stringstream i_str;
i_str << std::setw(20) << std::setfill('0') << cut;
std::string const aql(std::string("FOR l IN log FILTER l._key < \"") +
@ -1201,6 +1205,7 @@ bool State::removeObsolete(index_t cind) {
THROW_ARANGO_EXCEPTION_MESSAGE(queryResult.code, queryResult.details);
}
}
return true;
}
@ -1209,6 +1214,7 @@ bool State::persistCompactionSnapshot(index_t cind, arangodb::consensus::term_t
arangodb::consensus::Store& snapshot) {
if (checkCollection("compact")) {
std::stringstream i_str;
i_str << std::setw(20) << std::setfill('0') << cind;
Builder store;
@ -1221,12 +1227,12 @@ bool State::persistCompactionSnapshot(index_t cind, arangodb::consensus::term_t
}
store.add("term", VPackValue(static_cast<double>(term)));
store.add("_key", VPackValue(i_str.str()));
store.add("version", VPackValue(2));
}
TRI_ASSERT(_vocbase != nullptr);
auto ctx = std::make_shared<transaction::StandaloneContext>(_vocbase);
SingleCollectionTransaction trx(ctx, "compact", AccessMode::Type::WRITE);
Result res = trx.begin();
if (!res.ok()) {
@ -1235,22 +1241,25 @@ bool State::persistCompactionSnapshot(index_t cind, arangodb::consensus::term_t
OperationResult result;
try {
auto result = trx.insert("compact", store.slice(), _options);
result = trx.insert("compact", store.slice(), _options);
if (!result.ok()) {
if (result.is(TRI_ERROR_ARANGO_UNIQUE_CONSTRAINT_VIOLATED)) {
LOG_TOPIC(DEBUG, Logger::AGENCY)
<< "Failed to insert compacted agency state, will attempt to update: "
<< result.errorMessage();
result = trx.replace("compact", store.slice(), _options);
if (!result.ok()) {
LOG_TOPIC(FATAL, Logger::AGENCY)
<< "Failed to persist compated state: " << result.errorMessage();
}
} else {
LOG_TOPIC(FATAL, Logger::AGENCY)
<< "Failed to persist compated state: " << result.errorMessage();
<< "Failed to persist compacted agency state" << result.errorMessage();
FATAL_ERROR_EXIT();
}
}
}
} catch (std::exception const& e) {
LOG_TOPIC(FATAL, Logger::AGENCY) << "Failed to persist compated state: " << e.what();
LOG_TOPIC(FATAL, Logger::AGENCY)
<< "Failed to persist compacted agency state: " << e.what();
FATAL_ERROR_EXIT();
}
res = trx.finish(result.result);
if (res.ok()) {
@ -1262,6 +1271,7 @@ bool State::persistCompactionSnapshot(index_t cind, arangodb::consensus::term_t
LOG_TOPIC(ERR, Logger::AGENCY)
<< "Failed to persist snapshot for compaction!";
return false;
}
@ -1270,11 +1280,13 @@ bool State::persistCompactionSnapshot(index_t cind, arangodb::consensus::term_t
/// log is empty and something ought to be appended to it rather quickly.
bool State::storeLogFromSnapshot(Store& snapshot, index_t index, term_t term) {
_logLock.assertLockedByCurrentThread();
if (!persistCompactionSnapshot(index, term, snapshot)) {
LOG_TOPIC(ERR, Logger::AGENCY)
<< "Could not persist received log snapshot.";
return false;
}
// Now we need to completely erase our log, both persisted and volatile:
LOG_TOPIC(DEBUG, Logger::AGENCY)
<< "Removing complete log because of new snapshot.";
@ -1316,20 +1328,24 @@ void State::persistActiveAgents(query_t const& active, query_t const& pool) {
MUTEX_LOCKER(guard, _configurationWriteLock);
SingleCollectionTransaction trx(ctx, "configuration", AccessMode::Type::WRITE);
Result res = trx.begin();
if (!res.ok()) {
THROW_ARANGO_EXCEPTION(res);
}
auto result = trx.update("configuration", builder.slice(), _options);
if (result.fail()) {
THROW_ARANGO_EXCEPTION(result.result);
}
res = trx.finish(result.result);
if (!res.ok()) {
THROW_ARANGO_EXCEPTION(res);
}
LOG_TOPIC(DEBUG, Logger::AGENCY)
<< "Updated persisted agency configuration: " << builder.slice().toJson();
}
@ -1454,7 +1470,7 @@ std::shared_ptr<VPackBuilder> State::latestAgencyState(TRI_vocbase_t* vocbase,
// Result can only have length 0 or 1.
VPackSlice ii = result[0].resolveExternals();
buffer_t tmp = std::make_shared<arangodb::velocypack::Buffer<uint8_t>>();
store = ii.get("readDB");
store = ii;
index = arangodb::basics::StringUtils::uint64(ii.get("_key").copyString());
term = ii.get("term").getNumber<uint64_t>();
LOG_TOPIC(INFO, Logger::AGENCY)
@ -1526,13 +1542,11 @@ uint64_t State::toVelocyPack(index_t lastIndex, VPackBuilder& builder) const {
auto bindVars = std::make_shared<VPackBuilder>();
{ VPackObjectBuilder b(bindVars.get()); }
std::string const querystr
= "FOR l IN log FILTER l._key <= 'buf" + stringify(lastIndex) +
"' SORT l._key RETURN {'_key': l._key, 'timestamp': l.timestamp,"
"'clientId': l.clientId, 'request': l.request}";
std::string const logQueryStr = std::string("FOR l IN log FILTER l._key <= '")
+ stringify(lastIndex) + std::string("' SORT l._key RETURN l");
TRI_ASSERT(nullptr != _vocbase); // this check was previously in the Query constructor
arangodb::aql::Query logQuery(false, _vocbase, aql::QueryString(querystr), bindVars,
arangodb::aql::Query logQuery(false, _vocbase, aql::QueryString(logQueryStr), bindVars,
nullptr, arangodb::aql::PART_MAIN);
aql::QueryResult logQueryResult = logQuery.execute(_queryRegistry);

2
arangod/Agency/State.h
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");

103
arangod/Agency/Store.cpp
View File

@ -2,7 +2,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");
@ -122,6 +122,7 @@ Store::Store(Agent* agent, std::string const& name)
Store& Store::operator=(Store const& rhs) {
if (&rhs != this) {
MUTEX_LOCKER(otherLock, rhs._storeLock);
MUTEX_LOCKER(lock, _storeLock);
_agent = rhs._agent;
_timeTable = rhs._timeTable;
_observerTable = rhs._observerTable;
@ -135,6 +136,7 @@ Store& Store::operator=(Store const& rhs) {
Store& Store::operator=(Store&& rhs) {
if (&rhs != this) {
MUTEX_LOCKER(otherLock, rhs._storeLock);
MUTEX_LOCKER(lock, _storeLock);
_agent = std::move(rhs._agent);
_timeTable = std::move(rhs._timeTable);
_observerTable = std::move(rhs._observerTable);
@ -166,7 +168,6 @@ std::vector<apply_ret_t> Store::applyTransactions(query_t const& query,
}
}
if (found) {
LOG_TOPIC(ERR, Logger::AGENCY) << "forbidden";
success.push_back(FORBIDDEN);
continue;
}
@ -337,23 +338,23 @@ std::vector<bool> Store::applyLogEntries(arangodb::velocypack::Builder const& qu
body.add("term", VPackValue(term));
body.add("index", VPackValue(index));
auto ret = in.equal_range(url);
std::string currentKey;
std::map<std::string,std::map<std::string, std::string>> result;
// key -> (modified -> op)
for (auto it = ret.first; it != ret.second; ++it) {
if (currentKey != it->second->key) {
if (!currentKey.empty()) {
body.close();
}
body.add(it->second->key, VPackValue(VPackValueType::Object));
currentKey = it->second->key;
}
body.add(VPackValue(it->second->modified));
{
VPackObjectBuilder b(&body);
body.add("op", VPackValue(it->second->oper));
}
result[it->second->key][it->second->modified] = it->second->oper;
}
if (!currentKey.empty()) {
body.close();
for (auto const& m : result) {
body.add(VPackValue(m.first));
{
VPackObjectBuilder guard(&body);
for (auto const& m2 : m.second) {
body.add(VPackValue(m2.first));
{
VPackObjectBuilder guard2(&body);
body.add("op", VPackValue(m2.second));
}
}
}
}
}
@ -389,26 +390,26 @@ check_ret_t Store::check(VPackSlice const& slice, CheckMode mode) const {
std::string key = precond.key.copyString();
std::vector<std::string> pv = split(key, '/');
Node node("precond");
Node const* node = &Node::dummyNode();
// Check is guarded in ::apply
bool found = _node.has(pv);
if (found) {
node = _node(pv);
node = &_node(pv);
}
if (precond.value.isObject()) {
for (auto const& op : VPackObjectIterator(precond.value)) {
std::string const& oper = op.key.copyString();
if (oper == "old") { // old
if (node != op.value) {
if (*node != op.value) {
ret.push_back(precond.key);
if (mode == FIRST_FAIL) {
break;
}
}
} else if (oper == "oldNot") { // oldNot
if (node == op.value) {
if (*node == op.value) {
ret.push_back(precond.key);
if (mode == FIRST_FAIL) {
break;
@ -423,7 +424,7 @@ check_ret_t Store::check(VPackSlice const& slice, CheckMode mode) const {
break;
}
}
bool isArray = (node.type() == LEAF && node.slice().isArray());
bool isArray = (node->type() == LEAF && node->slice().isArray());
if (op.value.getBool() ? !isArray : isArray) {
ret.push_back(precond.key);
if (mode == FIRST_FAIL) {
@ -447,9 +448,9 @@ check_ret_t Store::check(VPackSlice const& slice, CheckMode mode) const {
}
} else if (oper == "in") { // in
if (found) {
if (node.slice().isArray()) {
if (node->slice().isArray()) {
bool _found = false;
for (auto const& i : VPackArrayIterator(node.slice())) {
for (auto const& i : VPackArrayIterator(node->slice())) {
if (i == op.value) {
_found = true;
continue;
@ -470,9 +471,9 @@ check_ret_t Store::check(VPackSlice const& slice, CheckMode mode) const {
if (!found) {
continue;
} else {
if (node.slice().isArray()) {
if (node->slice().isArray()) {
bool _found = false;
for (auto const& i : VPackArrayIterator(node.slice())) {
for (auto const& i : VPackArrayIterator(node->slice())) {
if (i == op.value) {
_found = true;
continue;
@ -499,7 +500,7 @@ check_ret_t Store::check(VPackSlice const& slice, CheckMode mode) const {
}
}
} else {
if (node != precond.value) {
if (*node != precond.value) {
ret.push_back(precond.key);
if (mode == FIRST_FAIL) {
break;
@ -610,14 +611,25 @@ query_t Store::clearExpired() const {
void Store::dumpToBuilder(Builder& builder) const {
MUTEX_LOCKER(storeLocker, _storeLock);
toBuilder(builder, true);
std::map<std::string, int64_t> clean {};
for (auto const& i : _timeTable) {
auto ts = std::chrono::duration_cast<std::chrono::seconds>(
i.first.time_since_epoch()).count();
auto it = clean.find(i.second);
if (it == clean.end()) {
clean[i.second] = ts;
} else if (ts < it->second) {
it->second = ts;
}
}
{
VPackObjectBuilder guard(&builder);
for (auto const& i : _timeTable) {
auto ts = std::chrono::duration_cast<std::chrono::seconds>(i.first.time_since_epoch())
.count();
builder.add(i.second, VPackValue(ts));
for (auto const& c : clean) {
builder.add(c.first, VPackValue(c.second));
}
}
{
VPackArrayBuilder garray(&builder);
for (auto const& i : _observerTable) {
@ -679,21 +691,29 @@ void Store::clear() {
}
/// Apply a request to my key value store
Store& Store::operator=(VPackSlice const& slice) {
TRI_ASSERT(slice.isArray());
Store& Store::operator=(VPackSlice const& s) {
TRI_ASSERT(s.isObject());
TRI_ASSERT(s.hasKey("readDB"));
auto const& slice = s.get("readDB");
TRI_ASSERT(slice.length() == 4);
MUTEX_LOCKER(storeLocker, _storeLock);
_node.applies(slice[0]);
TRI_ASSERT(slice[1].isObject());
for (auto const& entry : VPackObjectIterator(slice[1])) {
long long tse = entry.value.getInt();
_timeTable.emplace(
std::pair<TimePoint, std::string>(TimePoint(std::chrono::duration<int>(tse)),
entry.key.copyString()));
if (s.hasKey("version")) {
TRI_ASSERT(slice[1].isObject());
for (auto const& entry : VPackObjectIterator(slice[1])) {
if (entry.value.isNumber()) {
auto const& key = entry.key.copyString();
if (_node.has(key)) {
auto tp = TimePoint(std::chrono::seconds(entry.value.getNumber<int>()));
_node(key).timeToLive(tp);
_timeTable.emplace(std::pair<TimePoint, std::string>(tp, key));
}
}
}
}
TRI_ASSERT(slice[2].isArray());
for (auto const& entry : VPackArrayIterator(slice[2])) {
TRI_ASSERT(entry.isObject());
@ -770,11 +790,10 @@ bool Store::has(std::string const& path) const {
/// Remove ttl entry for path, guarded by caller
void Store::removeTTL(std::string const& uri) {
_storeLock.assertLockedByCurrentThread();
if (!_timeTable.empty()) {
for (auto it = _timeTable.cbegin(); it != _timeTable.cend();) {
if (it->second == uri) {
_timeTable.erase(it++);
it = _timeTable.erase(it);
} else {
++it;
}

2
arangod/Agency/Store.h
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");

2
arangod/Agency/StoreCallback.cpp
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");

2
arangod/Agency/StoreCallback.h
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");

84
arangod/Agency/Supervision.cpp
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");
@ -71,6 +71,9 @@ struct HealthRecord {
if (endpoint.empty()) {
endpoint = node.hasAsString("Endpoint").first;
}
if (hostId.empty()) {
hostId = node.hasAsString("Host").first;
}
if (node.has("Status")) {
status = node.hasAsString("Status").first;
if (node.has("SyncStatus")) { // New format
@ -100,9 +103,6 @@ struct HealthRecord {
lastAcked = node.hasAsString("LastHeartbeatAcked").first;
}
}
if (node.has("Host")) {
hostId = node.hasAsString("Host").first;
}
}
return *this;
}
@ -147,10 +147,11 @@ Supervision::Supervision()
_snapshot("Supervision"),
_transient("Transient"),
_frequency(1.),
_gracePeriod(5.),
_okThreshold(1.5),
_gracePeriod(10.),
_okThreshold(5.),
_jobId(0),
_jobIdMax(0),
_haveAborts(false),
_selfShutdown(false),
_upgraded(false) {}
@ -209,7 +210,7 @@ void Supervision::upgradeZero(Builder& builder) {
if (fails.length() > 0) {
for (VPackSlice fail : VPackArrayIterator(fails)) {
builder.add(VPackValue(fail.copyString()));
{ VPackObjectBuilder ooo(&builder); }
{ VPackArrayBuilder ooo(&builder); }
}
}
}
@ -428,8 +429,6 @@ std::vector<check_t> Supervision::check(std::string const& type) {
}
}
// LOG_TOPIC(ERR, Logger::FIXME) << "ServersRegistered: " << serversRegistered;
// Remove all machines, which are no longer planned
for (auto const& machine : machinesPlanned) {
todelete.erase(std::remove(todelete.begin(), todelete.end(), machine.first),
@ -444,8 +443,6 @@ std::vector<check_t> Supervision::check(std::string const& type) {
std::string lastHeartbeatStatus, lastHeartbeatAcked, lastHeartbeatTime,
lastStatus, serverID(machine.first), shortName;
// LOG_TOPIC(ERR, Logger::FIXME) << "ServerID: " << serverID;
// short name arrives asynchronous to machine registering, make sure
// it has arrived before trying to use it
auto tmp_shortName =
@ -532,9 +529,6 @@ std::vector<check_t> Supervision::check(std::string const& type) {
// Status changed?
bool changed = transist.statusDiff(persist);
/*LOG_TOPIC(ERR, Logger::FIXME) << "Changed: " << changed
<< "transist: " << transist << " persist: " << persist;*/
// Take necessary actions if any
std::shared_ptr<VPackBuilder> envelope;
if (changed) {
@ -609,6 +603,53 @@ std::vector<check_t> Supervision::check(std::string const& type) {
return ret;
}
bool Supervision::earlyBird() const {
std::vector<std::string> tpath{"Sync", "ServerStates"};
std::vector<std::string> pdbpath{"Plan", "DBServers"};
std::vector<std::string> pcpath{"Plan", "Coordinators"};
if (!_snapshot.has(pdbpath)) {
LOG_TOPIC(DEBUG, Logger::SUPERVISION)
<< "No Plan/DBServers key in persistent store";
return false;
}
VPackBuilder dbserversB = _snapshot(pdbpath).toBuilder();
VPackSlice dbservers = dbserversB.slice();
if (!_snapshot.has(pcpath)) {
LOG_TOPIC(DEBUG, Logger::SUPERVISION)
<< "No Plan/Coordinators key in persistent store";
return false;
}
VPackBuilder coordinatorsB = _snapshot(pcpath).toBuilder();
VPackSlice coordinators = coordinatorsB.slice();
if (!_transient.has(tpath)) {
LOG_TOPIC(DEBUG, Logger::SUPERVISION)
<< "No Sync/ServerStates key in transient store";
return false;
}
VPackBuilder serverStatesB = _transient(tpath).toBuilder();
VPackSlice serverStates = serverStatesB.slice();
// every db server in plan accounted for in transient store?
for (auto const& server : VPackObjectIterator(dbservers)) {
auto serverId = server.key.copyString();
if (!serverStates.hasKey(serverId)) {
return false;
}
}
// every db server in plan accounted for in transient store?
for (auto const& server : VPackObjectIterator(coordinators)) {
auto serverId = server.key.copyString();
if (!serverStates.hasKey(serverId)) {
return false;
}
}
return true;
}
// Update local agency snapshot, guarded by callers
bool Supervision::updateSnapshot() {
_lock.assertLockedByCurrentThread();
@ -759,7 +800,7 @@ void Supervision::run() {
_haveAborts = false;
if (_agent->leaderFor() > 10) {
if (_agent->leaderFor() > 55 || earlyBird()) {
// 55 seconds is less than a minute, which fits to the
// 60 seconds timeout in /_admin/cluster/health
try {
@ -772,6 +813,11 @@ void Supervision::run() {
<< "Supervision::doChecks() generated an uncaught "
"exception.";
}
} else {
LOG_TOPIC(INFO, Logger::SUPERVISION)
<< "Postponing supervision for now, waiting for incoming "
"heartbeats: "
<< _agent->leaderFor();
}
handleJobs();
@ -820,7 +866,10 @@ void Supervision::run() {
// Guarded by caller
bool Supervision::isShuttingDown() {
_lock.assertLockedByCurrentThread();
return _snapshot.hasAsBool("Shutdown").first;
if (_snapshot.has("Shutdown")) {
return _snapshot.hasAsBool("Shutdown").first;
}
return false;
}
// Guarded by caller
@ -1019,8 +1068,6 @@ void Supervision::cleanupLostCollections(Node const& snapshot, AgentInterface* a
auto const& trx = builder->slice();
LOG_TOPIC(ERR, Logger::FIXME) << "Trx: " << trx.toJson();
if (trx.length() > 0) {
// do it! fire and forget!
agent->write(builder);
@ -1370,6 +1417,7 @@ void Supervision::shrinkCluster() {
// Get servers from plan
auto availServers = Job::availableServers(_snapshot);
// set by external service like Kubernetes / Starter / DCOS
size_t targetNumDBServers;
std::string const NDBServers("/Target/NumberOfDBServers");

38
arangod/Agency/Supervision.h
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");
@ -27,13 +27,11 @@
#include "Agency/AgencyCommon.h"
#include "Agency/AgentInterface.h"
#include "Agency/Node.h"
#include "AgencyCommon.h"
#include "Agency/TimeString.h"
#include "Basics/ConditionVariable.h"
#include "Basics/Mutex.h"
#include "Cluster/CriticalThread.h"
#include <chrono>
namespace arangodb {
namespace consensus {
@ -123,6 +121,9 @@ class Supervision : public arangodb::CriticalThread {
}
private:
/// @brief decide, if we can start supervision ahead of armageddon delay
bool earlyBird() const;
/// @brief Upgrade agency with FailedServers an object from array
void upgradeZero(VPackBuilder&);
@ -226,35 +227,6 @@ class Supervision : public arangodb::CriticalThread {
*/
query_t removeTransactionBuilder(std::vector<std::string> const&);
inline std::string timepointToString(Supervision::TimePoint const& t) {
time_t tt = std::chrono::system_clock::to_time_t(t);
struct tm tb;
size_t const len(21);
char buffer[len];
TRI_gmtime(tt, &tb);
::strftime(buffer, sizeof(buffer), "%Y-%m-%dT%H:%M:%SZ", &tb);
return std::string(buffer, len - 1);
}
inline Supervision::TimePoint stringToTimepoint(std::string const& s) {
if (!s.empty()) {
try {
std::tm tt;
tt.tm_year = std::stoi(s.substr(0, 4)) - 1900;
tt.tm_mon = std::stoi(s.substr(5, 2)) - 1;
tt.tm_mday = std::stoi(s.substr(8, 2));
tt.tm_hour = std::stoi(s.substr(11, 2));
tt.tm_min = std::stoi(s.substr(14, 2));
tt.tm_sec = std::stoi(s.substr(17, 2));
tt.tm_isdst = 0;
auto time_c = TRI_timegm(&tt);
return std::chrono::system_clock::from_time_t(time_c);
} catch (...) {
}
}
return std::chrono::time_point<std::chrono::system_clock>();
}
} // namespace consensus
} // namespace arangodb

61
arangod/Agency/TimeString.h Normal file
View File

@ -0,0 +1,61 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2018 ArangoDB GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");
/// you may not use this file except in compliance with the License.
/// You may obtain a copy of the License at
///
/// http://www.apache.org/licenses/LICENSE-2.0
///
/// Unless required by applicable law or agreed to in writing, software
/// distributed under the License is distributed on an "AS IS" BASIS,
/// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
/// See the License for the specific language governing permissions and
/// limitations under the License.
///
/// Copyright holder is ArangoDB GmbH, Cologne, Germany
///
/// @author Kaveh Vahedipour
////////////////////////////////////////////////////////////////////////////////
#ifndef ARANGOD_CONSENSUS_TIMESTRING_H
#define ARANGOD_CONSENSUS_TIMESTRING_H 1
#include <chrono>
inline std::string timepointToString(std::chrono::system_clock::time_point const& t) {
time_t tt = std::chrono::system_clock::to_time_t(t);
struct tm tb;
size_t const len(21);
char buffer[len];
TRI_gmtime(tt, &tb);
::strftime(buffer, sizeof(buffer), "%Y-%m-%dT%H:%M:%SZ", &tb);
return std::string(buffer, len - 1);
}
inline std::string timepointToString(std::chrono::system_clock::duration const& d) {
return timepointToString(std::chrono::system_clock::time_point() + d);
}
inline std::chrono::system_clock::time_point stringToTimepoint(std::string const& s) {
if (!s.empty()) {
try {
std::tm tt;
tt.tm_year = std::stoi(s.substr(0, 4)) - 1900;
tt.tm_mon = std::stoi(s.substr(5, 2)) - 1;
tt.tm_mday = std::stoi(s.substr(8, 2));
tt.tm_hour = std::stoi(s.substr(11, 2));
tt.tm_min = std::stoi(s.substr(14, 2));
tt.tm_sec = std::stoi(s.substr(17, 2));
tt.tm_isdst = 0;
auto time_c = TRI_timegm(&tt);
return std::chrono::system_clock::from_time_t(time_c);
} catch (...) {
}
}
return std::chrono::time_point<std::chrono::system_clock>();
}
#endif

2
arangod/Agency/v8-agency.cpp
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");

2
arangod/Agency/v8-agency.h
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2018 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");

1
arangod/CMakeLists.txt
View File

@ -87,7 +87,6 @@ SET(ARANGOD_SOURCES
Agency/JobContext.cpp
Agency/MoveShard.cpp
Agency/Node.cpp
Agency/NotifyCallback.cpp
Agency/RemoveFollower.cpp
Agency/RestAgencyHandler.cpp
Agency/RestAgencyPrivHandler.cpp

2
js/client/tests/agency/agency-test.js
View File

@ -676,7 +676,7 @@ function agencyTestSuite () {
writeAndCheck([[{"/a/y":{"op":"set","new":12}}]]);
assertEqual(readAndCheck([["a/y"]]), [{"a":{"y":12}}]);
wait(1.1);
assertEqual(readAndCheck([["/a/y"]]), [{"a":{"y":12}}]);
assertEqual(readAndCheck([["/a/y"]]), [{"a":{}}]);
writeAndCheck([[{"foo/bar":{"op":"set","new":{"baz":12}}}]]);
assertEqual(readAndCheck([["/foo/bar/baz"]]),
[{"foo":{"bar":{"baz":12}}}]);

16
tests/Agency/CleanUpLostCollectionTest.cpp
View File

@ -123,23 +123,9 @@ SECTION("clean up a lost collection when the leader is failed") {
// not empty: /arango/Current/Collections/database/collection/s99
// empty: /arango/Plan/Collections/database/collection/shards/s99
// old: /arango/Supervision/Health/leader/Status == "FAILED"
// 2. Transaction:
// - Operation:
// delete /arango/Current/Collections/database/collection/s16
// push {
// "creator": "supervision",
// "jobId": "1",
// "server": "s99",
// "timeCreated": "2018-09-26T09:25:33Z",
// "type": "cleanUpLostCollection"
// }
// - Preconditions:
// not empty: /arango/Current/Collections/database/collection/s16
// empty: /arango/Plan/Collections/database/collection/shards/s16
// old: /arango/Supervision/Health/failed/Status == "FAILED"
auto const& trxs = q->slice();
REQUIRE(trxs.length() == 2);
REQUIRE(trxs.length() == 1);
auto const& trx1 = trxs[0];
REQUIRE(trx1.length() == 2); // Operation and Precondition

11
tests/Agency/CleanUpLostCollectionTest.json
View File

@ -9,11 +9,6 @@ R"=(
"servers": [
"leader"
]
},
"s16": {
"servers": [
"failed"
]
}
},
"other": {
@ -40,9 +35,6 @@ R"=(
"leader": {
"Status": "FAILED"
},
"failed": {
"Status": "FAILED"
},
"otherleader": {
"Status": "GOOD"
}
@ -58,9 +50,6 @@ R"=(
"leader": {
"ShortName": "leader"
},
"failed": {
"ShortName": "failed"
},
"otherleader": {
"ShortName": "otherleader"
}

38
tests/Agency/FailedLeaderTest.cpp
View File

@ -891,37 +891,23 @@ SECTION("when everything is finished there should be proper cleanup") {
REQUIRE(builder);
Node agency = createNodeFromBuilder(*builder);
size_t numWrites = 0;
Mock<AgentInterface> mockAgent;
When(Method(mockAgent, write)).AlwaysDo([&](query_t const& q, consensus::AgentInterface::WriteMode w) -> write_ret_t {
INFO("Write: " << q->slice().toJson());
numWrites++;
if (numWrites == 1) {
REQUIRE(std::string(q->slice().typeName()) == "array" );
REQUIRE(std::string(q->slice()[0][0].typeName()) == "object");
REQUIRE(q->slice()[0][0].length() == 1);
REQUIRE(std::string(q->slice().typeName()) == "array" );
REQUIRE(q->slice().length() == 1);
REQUIRE(std::string(q->slice()[0].typeName()) == "array");
REQUIRE(q->slice()[0].length() == 1); // we always simply override! no preconditions...
REQUIRE(std::string(q->slice()[0][0].typeName()) == "object");
auto writes = q->slice()[0][0];
REQUIRE(std::string(writes.get("/arango/Target/FailedServers/" + SHARD_LEADER).typeName()) == "object");
REQUIRE(writes.get("/arango/Target/FailedServers/" + SHARD_LEADER).get("op").copyString() == "erase");
REQUIRE(writes.get("/arango/Target/FailedServers/" + SHARD_LEADER).get("val").copyString() == SHARD);
return fakeWriteResult;
} else {
REQUIRE(std::string(q->slice().typeName()) == "array" );
REQUIRE(q->slice().length() == 1);
REQUIRE(std::string(q->slice()[0].typeName()) == "array");
REQUIRE(q->slice()[0].length() == 1); // we always simply override! no preconditions...
REQUIRE(std::string(q->slice()[0][0].typeName()) == "object");
auto writes = q->slice()[0][0];
REQUIRE(std::string(writes.get("/arango/Supervision/Shards/" + SHARD).typeName()) == "object");
REQUIRE(writes.get("/arango/Supervision/Shards/" + SHARD).get("op").copyString() == "delete");
REQUIRE(std::string(writes.get("/arango/Target/Pending/1").get("op").typeName()) == "string");
CHECK(writes.get("/arango/Target/Pending/1").get("op").copyString() == "delete");
CHECK(std::string(writes.get("/arango/Target/Finished/1").typeName()) == "object");
return fakeWriteResult;
}
auto writes = q->slice()[0][0];
REQUIRE(std::string(writes.get("/arango/Supervision/Shards/" + SHARD).typeName()) == "object");
REQUIRE(writes.get("/arango/Supervision/Shards/" + SHARD).get("op").copyString() == "delete");
REQUIRE(std::string(writes.get("/arango/Target/Pending/1").get("op").typeName()) == "string");
CHECK(writes.get("/arango/Target/Pending/1").get("op").copyString() == "delete");
CHECK(std::string(writes.get("/arango/Target/Finished/1").typeName()) == "object");
return fakeWriteResult;
});
When(Method(mockAgent, waitFor)).AlwaysReturn(AgentInterface::raft_commit_t::OK);
AgentInterface &agent = mockAgent.get();