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

1940 lines
63 KiB
C++
Raw Permalink Blame History

///////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2018 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 Jan Steemann
/// @author Max Neunhoeffer
////////////////////////////////////////////////////////////////////////////////
#include "AgencyComm.h"
#include "ApplicationFeatures/ApplicationServer.h"
#include "RestServer/ServerFeature.h"
#include <thread>
#ifdef DEBUG_SYNC_REPLICATION
#include <atomic>
#endif
#include <velocypack/Iterator.h>
#include <velocypack/Sink.h>
#include <velocypack/velocypack-aliases.h>
#include <set>
#include "Basics/ReadLocker.h"
#include "Basics/StringBuffer.h"
#include "Basics/StringUtils.h"
#include "Basics/VelocyPackHelper.h"
#include "Basics/WriteLocker.h"
#include "Cluster/ClusterComm.h"
#include "Cluster/ServerState.h"
#include "Endpoint/Endpoint.h"
#include "GeneralServer/AuthenticationFeature.h"
#include "Logger/Logger.h"
#include "Random/RandomGenerator.h"
#include "Rest/HttpRequest.h"
#include "Rest/HttpResponse.h"
#include "SimpleHttpClient/GeneralClientConnection.h"
#include "SimpleHttpClient/SimpleHttpClient.h"
#include "SimpleHttpClient/SimpleHttpResult.h"
#include <boost/uuid/uuid.hpp>
#include <boost/uuid/uuid_generators.hpp>
#include <boost/uuid/uuid_io.hpp>
using namespace arangodb;
using namespace arangodb::application_features;
using namespace arangodb::httpclient;
using namespace arangodb::rest;
#ifdef DEBUG_SYNC_REPLICATION
static std::atomic<uint64_t> debugUniqId(1);
bool AgencyComm::syncReplDebug = false;
#endif
static void addEmptyVPackObject(std::string const& name, VPackBuilder& builder) {
builder.add(VPackValue(name));
VPackObjectBuilder c(&builder);
}
const std::vector<std::string> AgencyTransaction::TypeUrl(
{"/read", "/write", "/transact", "/transient"});
// -----------------------------------------------------------------------------
// --SECTION-- AgencyPrecondition
// -----------------------------------------------------------------------------
AgencyPrecondition::AgencyPrecondition()
: type(AgencyPrecondition::Type::NONE), empty(true) {}
AgencyPrecondition::AgencyPrecondition(std::string const& key, Type t, bool e)
: key(AgencyCommManager::path(key)), type(t), empty(e) {}
AgencyPrecondition::AgencyPrecondition(std::string const& key, Type t, VPackSlice const& s)
: key(AgencyCommManager::path(key)), type(t), empty(false), value(s) {}
void AgencyPrecondition::toVelocyPack(VPackBuilder& builder) const {
if (type != AgencyPrecondition::Type::NONE) {
builder.add(VPackValue(key));
{
VPackObjectBuilder preconditionDefinition(&builder);
switch (type) {
case AgencyPrecondition::Type::EMPTY:
builder.add("oldEmpty", VPackValue(empty));
break;
case AgencyPrecondition::Type::VALUE:
builder.add("old", value);
break;
// mop: useless compiler warning :S
case AgencyPrecondition::Type::NONE:
break;
}
}
}
}
void AgencyPrecondition::toGeneralBuilder(VPackBuilder& builder) const {
if (type != AgencyPrecondition::Type::NONE) {
VPackObjectBuilder preconditionDefinition(&builder);
builder.add(VPackValue(key));
{
VPackObjectBuilder preconditionDefinition(&builder);
switch (type) {
case AgencyPrecondition::Type::EMPTY:
builder.add("oldEmpty", VPackValue(empty));
break;
case AgencyPrecondition::Type::VALUE:
builder.add("old", value);
break;
// mop: useless compiler warning :S
case AgencyPrecondition::Type::NONE:
break;
}
}
} else {
VPackObjectBuilder guard(&builder);
}
}
// -----------------------------------------------------------------------------
// --SECTION-- AgencyOperation
// -----------------------------------------------------------------------------
AgencyOperation::AgencyOperation(std::string const& key)
: _key(AgencyCommManager::path(key)), _opType() {
_opType.type = AgencyOperationType::Type::READ;
}
AgencyOperation::AgencyOperation(std::string const& key, AgencySimpleOperationType opType)
: _key(AgencyCommManager::path(key)), _opType() {
_opType.type = AgencyOperationType::Type::SIMPLE;
_opType.simple = opType;
}
AgencyOperation::AgencyOperation(std::string const& key,
AgencyValueOperationType opType, VPackSlice value)
: _key(AgencyCommManager::path(key)), _opType(), _value(value) {
_opType.type = AgencyOperationType::Type::VALUE;
_opType.value = opType;
}
AgencyOperation::AgencyOperation(std::string const& key, AgencyValueOperationType opType,
VPackSlice newValue, VPackSlice oldValue)
: _key(AgencyCommManager::path(key)), _opType(), _value(newValue), _value2(oldValue) {
_opType.type = AgencyOperationType::Type::VALUE;
_opType.value = opType;
}
AgencyOperationType AgencyOperation::type() const { return _opType; }
void AgencyOperation::toVelocyPack(VPackBuilder& builder) const {
builder.add(VPackValue(_key));
{
VPackObjectBuilder valueOperation(&builder);
builder.add("op", VPackValue(_opType.toString()));
if (_opType.type == AgencyOperationType::Type::VALUE) {
if (_opType.value == AgencyValueOperationType::OBSERVE ||
_opType.value == AgencyValueOperationType::UNOBSERVE) {
builder.add("url", _value);
} else if (_opType.value == AgencyValueOperationType::ERASE) {
builder.add("val", _value);
} else if (_opType.value == AgencyValueOperationType::REPLACE) {
builder.add("new", _value);
builder.add("val", _value2);
} else {
builder.add("new", _value);
}
if (_ttl > 0) {
builder.add("ttl", VPackValue(_ttl));
}
}
}
}
void AgencyOperation::toGeneralBuilder(VPackBuilder& builder) const {
if (_opType.type == AgencyOperationType::Type::READ) {
builder.add(VPackValue(_key));
} else {
VPackObjectBuilder operation(&builder);
builder.add(VPackValue(_key));
VPackObjectBuilder valueOperation(&builder);
builder.add("op", VPackValue(_opType.toString()));
if (_opType.type == AgencyOperationType::Type::VALUE) {
if (_opType.value == AgencyValueOperationType::OBSERVE ||
_opType.value == AgencyValueOperationType::UNOBSERVE) {
builder.add("url", _value);
} else {
builder.add("new", _value);
}
if (_ttl > 0) {
builder.add("ttl", VPackValue(_ttl));
}
}
}
}
// -----------------------------------------------------------------------------
// --SECTION-- AgencyTransaction
// -----------------------------------------------------------------------------
std::string AgencyTransaction::toJson() const {
VPackBuilder builder;
builder.openArray();
toVelocyPack(builder);
builder.close();
return builder.toJson();
}
// -----------------------------------------------------------------------------
// --SECTION-- AgencyWriteTransaction
// -----------------------------------------------------------------------------
void AgencyWriteTransaction::toVelocyPack(VPackBuilder& builder) const {
VPackArrayBuilder guard(&builder);
{
VPackObjectBuilder guard2(&builder); // Writes
for (AgencyOperation const& operation : operations) {
operation.toVelocyPack(builder);
}
}
if (preconditions.size() > 0) {
VPackObjectBuilder guard3(&builder); // Preconditions
for (AgencyPrecondition const& precondition : preconditions) {
precondition.toVelocyPack(builder);
}
} else {
VPackObjectBuilder guard3(&builder);
}
builder.add(VPackValue(clientId)); // Transactions
}
bool AgencyWriteTransaction::validate(AgencyCommResult const& result) const {
return (result.slice().isObject() && result.slice().hasKey("results") &&
result.slice().get("results").isArray());
}
std::string AgencyWriteTransaction::randomClientId() {
std::string uuid = to_string(boost::uuids::random_generator()());
auto ss = ServerState::instance();
if (ss != nullptr && !ss->getId().empty()) {
return ss->getId() + ":" + uuid;
}
return uuid;
}
// -----------------------------------------------------------------------------
// --SECTION-- AgencyTransientTransaction
// -----------------------------------------------------------------------------
void AgencyTransientTransaction::toVelocyPack(VPackBuilder& builder) const {
VPackArrayBuilder guard(&builder);
{
VPackObjectBuilder guard2(&builder);
for (AgencyOperation const& operation : operations) {
operation.toVelocyPack(builder);
}
}
if (preconditions.size() > 0) {
VPackObjectBuilder guard3(&builder);
for (AgencyPrecondition const& precondition : preconditions) {
precondition.toVelocyPack(builder);
}
}
}
bool AgencyTransientTransaction::validate(AgencyCommResult const& result) const {
return (result.slice().isArray() && result.slice().length() > 0 &&
result.slice()[0].isBool() && result.slice()[0].getBool() == true);
}
// -----------------------------------------------------------------------------
// --SECTION-- AgencyGeneralTransaction
// -----------------------------------------------------------------------------
/*
void AgencyGeneralTransaction::toVelocyPack(VPackBuilder& builder) const {
for (auto const& trx : transactions) {
auto opers = std::get<0>(trx);
auto precs = std::get<1>(trx);
TRI_ASSERT(!opers.empty());
if (!opers.empty()) {
if (opers[0].type().type == AgencyOperationType::Type::READ) {
for (auto const& op : opers) {
VPackArrayBuilder guard(&builder);
op.toGeneralBuilder(builder);
}
} else {
VPackArrayBuilder guard(&builder);
{ VPackObjectBuilder o(&builder); // Writes
for (AgencyOperation const& oper : opers) {
oper.toVelocyPack(builder);
}}
{ VPackObjectBuilder p(&builder); // Preconditions
if (!precs.empty()) {
for (AgencyPrecondition const& prec : precs) {
prec.toVelocyPack(builder);
}}}
builder.add(VPackValue(clientId)); // Transactions
}
}
}
}
void AgencyGeneralTransaction::push_back(AgencyOperation const& op) {
transactions.emplace_back(
TransactionType(std::vector<AgencyOperation>(1, op),
std::vector<AgencyPrecondition>(0)));
}
void AgencyGeneralTransaction::push_back(
std::pair<AgencyOperation,AgencyPrecondition> const& oper) {
transactions.emplace_back(
TransactionType(std::vector<AgencyOperation>(1,oper.first),
std::vector<AgencyPrecondition>(1,oper.second)));
}
bool AgencyGeneralTransaction::validate(AgencyCommResult const& result) const {
return (result.slice().isArray() &&
result.slice().length() >= 1); // >= transactions.size()
}*/
// -----------------------------------------------------------------------------
// --SECTION-- AgencyReadTransaction
// -----------------------------------------------------------------------------
void AgencyReadTransaction::toVelocyPack(VPackBuilder& builder) const {
VPackArrayBuilder guard2(&builder);
for (std::string const& key : keys) {
builder.add(VPackValue(key));
}
}
bool AgencyReadTransaction::validate(AgencyCommResult const& result) const {
return (result.slice().isArray() && result.slice().length() == 1);
}
// -----------------------------------------------------------------------------
// --SECTION-- AgencyCommResult
// -----------------------------------------------------------------------------
AgencyCommResult::AgencyCommResult()
: _location(), _message(), _body(), _values(), _statusCode(0), _connected(false), _sent(false) {}
AgencyCommResult::AgencyCommResult(int code, std::string const& message,
std::string const& clientId)
: _location(),
_message(message),
_body(),
_values(),
_statusCode(code),
_connected(false),
_sent(false) {}
AgencyCommResult::AgencyCommResult(AgencyCommResult&& other) noexcept
: _location(std::move(other._location)),
_message(std::move(other._message)),
_body(std::move(other._body)),
_values(std::move(other._values)),
_statusCode(other._statusCode),
_connected(other._connected),
_sent(other._sent),
_vpack(std::move(other._vpack)) {
other._statusCode = 0;
other._connected = false;
other._sent = false;
}
AgencyCommResult& AgencyCommResult::operator=(AgencyCommResult&& other) noexcept {
if (this != &other) {
_location = std::move(other._location);
_message = std::move(other._message);
_body = std::move(other._body);
_values = std::move(other._values);
_statusCode = other._statusCode;
_connected = other._connected;
_sent = other._sent;
_vpack = std::move(other._vpack);
other._statusCode = 0;
other._connected = false;
other._sent = false;
}
return *this;
}
void AgencyCommResult::set(int code, std::string const& message) {
_message = message;
_statusCode = code;
_location.clear();
_body.clear();
_values.clear();
_vpack.reset();
}
bool AgencyCommResult::connected() const { return _connected; }
int AgencyCommResult::httpCode() const { return _statusCode; }
bool AgencyCommResult::sent() const { return _sent; }
int AgencyCommResult::errorCode() const {
try {
if (!_body.empty()) {
std::shared_ptr<VPackBuilder> bodyBuilder = VPackParser::fromJson(_body);
VPackSlice body = bodyBuilder->slice();
if (!body.isObject()) {
return 0;
}
// get "errorCode" attribute (0 if not exist)
return basics::VelocyPackHelper::getNumericValue<int>(body, "errorCode", 0);
}
} catch (VPackException const&) {
}
return 0;
}
std::string AgencyCommResult::errorMessage() const {
if (!_message.empty()) {
// return stored message first if set
return _message;
}
if (!_connected) {
return std::string("unable to connect to agency");
}
try {
std::shared_ptr<VPackBuilder> bodyBuilder = VPackParser::fromJson(_body);
VPackSlice body = bodyBuilder->slice();
if (!body.isObject()) {
return "";
}
// get "message" attribute ("" if not exist)
return arangodb::basics::VelocyPackHelper::getStringValue(body, "message",
"");
} catch (VPackException const& e) {
std::string message("VPackException parsing body (" + _body + "): " + e.what());
return std::string(message);
}
}
std::string AgencyCommResult::errorDetails() const {
std::string const errorMessage = this->errorMessage();
if (errorMessage.empty()) {
return _message;
}
return _message + " (" + errorMessage + ")";
}
void AgencyCommResult::clear() {
// clear existing values. They free themselves
_values.clear();
_location = "";
_message = "";
_body = "";
_vpack.reset();
_statusCode = 0;
_sent = false;
_connected = false;
}
VPackSlice AgencyCommResult::slice() const { return _vpack->slice(); }
void AgencyCommResult::toVelocyPack(VPackBuilder& builder) const {
{ VPackObjectBuilder dump(&builder);
builder.add("location", VPackValue(_location));
builder.add("message", VPackValue(_message));
builder.add("sent", VPackValue(_sent));
builder.add("body", VPackValue(_body));
if (_vpack != nullptr) {
if (_vpack->isClosed()) {
builder.add("vpack", _vpack->slice());
}
}
builder.add("statusCode", VPackValue(_statusCode));
builder.add(VPackValue("values"));
{ VPackObjectBuilder v(&builder);
for (auto const& value : _values) {
builder.add(VPackValue(value.first));
auto const& entry = value.second;
{ VPackObjectBuilder vv(&builder);
builder.add("index", VPackValue(entry._index));
builder.add("isDir", VPackValue(entry._isDir));
if (entry._vpack != nullptr && entry._vpack->isClosed()) {
builder.add("vpack", entry._vpack->slice());
}}}
}}
}
VPackBuilder AgencyCommResult::toVelocyPack() const {
VPackBuilder builder;
toVelocyPack(builder);
return builder;
}
namespace std {
ostream& operator<< (ostream& out, AgencyCommResult const& a) {
out << a.toVelocyPack().toJson();
return out;
}}
// -----------------------------------------------------------------------------
// --SECTION-- AgencyCommManager
// -----------------------------------------------------------------------------
std::unique_ptr<AgencyCommManager> AgencyCommManager::MANAGER;
AgencyConnectionOptions AgencyCommManager::CONNECTION_OPTIONS(15.0, 120.0, 120.0, 100);
void AgencyCommManager::initialize(std::string const& prefix) {
MANAGER.reset(new AgencyCommManager(prefix));
}
void AgencyCommManager::shutdown() { MANAGER.reset(); }
std::string AgencyCommManager::path() {
if (MANAGER == nullptr) {
return "";
}
return MANAGER->_prefix;
}
std::string AgencyCommManager::path(std::string const& p1) {
if (MANAGER == nullptr) {
return "";
}
return MANAGER->_prefix + "/" + basics::StringUtils::trim(p1, "/");
}
std::string AgencyCommManager::path(std::string const& p1, std::string const& p2) {
if (MANAGER == nullptr) {
return "";
}
return MANAGER->_prefix + "/" + basics::StringUtils::trim(p1, "/") + "/" +
basics::StringUtils::trim(p2, "/");
}
std::vector<std::string> AgencyCommManager::slicePath(std::string const& p1) {
std::string const p2 = basics::StringUtils::trim(p1, "/");
std::vector<std::string> split = basics::StringUtils::split(p2, '/');
if (split.size() > 0 && split[0] != AgencyCommManager::path()) {
split.insert(split.begin(), AgencyCommManager::path());
}
return split;
}
std::string AgencyCommManager::generateStamp() {
time_t tt = time(nullptr);
struct tm tb;
char buffer[21];
TRI_gmtime(tt, &tb);
size_t len = ::strftime(buffer, sizeof(buffer), "%Y-%m-%dT%H:%M:%SZ", &tb);
return std::string(buffer, len);
}
bool AgencyCommManager::start() {
AgencyComm comm;
bool ok = comm.ensureStructureInitialized();
LOG_TOPIC("d8ce6", DEBUG, Logger::AGENCYCOMM)
<< "structures " << (ok ? "are" : "failed to") << " initialize";
return ok;
}
void AgencyCommManager::stop() {
MUTEX_LOCKER(locker, _lock);
for (auto& i : _unusedConnections) {
i.second.clear();
}
_unusedConnections.clear();
}
std::unique_ptr<GeneralClientConnection> AgencyCommManager::acquire(std::string& endpoint) {
std::unique_ptr<GeneralClientConnection> connection;
MUTEX_LOCKER(locker, _lock);
if (_endpoints.empty()) {
return nullptr;
} else {
if (endpoint.empty()) {
endpoint = _endpoints.front();
LOG_TOPIC("8ca12", DEBUG, Logger::AGENCYCOMM)
<< "Using endpoint " << endpoint << " for agency communication, full selection:";
}
if (!_unusedConnections[endpoint].empty()) {
connection.reset(_unusedConnections[endpoint].back().release());
_unusedConnections[endpoint].pop_back();
} else {
connection = createNewConnection();
}
}
LOG_TOPIC("64bc2", TRACE, Logger::AGENCYCOMM)
<< "acquiring agency connection '" << connection.get()
<< "' for endpoint '" << endpoint << "'";
return connection;
}
void AgencyCommManager::release(std::unique_ptr<httpclient::GeneralClientConnection> connection,
std::string const& endpoint) {
MUTEX_LOCKER(locker, _lock);
releaseNonLocking(std::move(connection), endpoint);
}
void AgencyCommManager::releaseNonLocking(std::unique_ptr<httpclient::GeneralClientConnection> connection,
std::string const& endpoint) {
if (_endpoints.front() == endpoint) {
LOG_TOPIC("ff659", TRACE, Logger::AGENCYCOMM)
<< "releasing agency connection '" << connection.get()
<< "', active endpoint '" << endpoint << "'";
} else {
LOG_TOPIC("8549f", TRACE, Logger::AGENCYCOMM)
<< "releasing agency connection '" << connection.get()
<< "', inactive endpoint '" << endpoint << "'";
}
_unusedConnections[endpoint].emplace_back(std::move(connection));
}
void AgencyCommManager::failed(std::unique_ptr<httpclient::GeneralClientConnection> connection,
std::string const& endpoint) {
MUTEX_LOCKER(locker, _lock);
failedNonLocking(std::move(connection), endpoint);
}
void AgencyCommManager::failedNonLocking(std::unique_ptr<httpclient::GeneralClientConnection> connection,
std::string const& endpoint) {
if (_endpoints.front() == endpoint) {
LOG_TOPIC("1a7b9", TRACE, Logger::AGENCYCOMM)
<< "failed agency connection '" << connection.get()
<< "', active endpoint " << endpoint << "'";
} else {
LOG_TOPIC("90592", TRACE, Logger::AGENCYCOMM)
<< "failed agency connection '" << connection.get()
<< "', inactive endpoint " << endpoint << "'";
}
switchCurrentEndpoint();
// Note that we intentionally let the connection go out of scope, because
// it failed, so we would have to reopen it anyway. We keep the other
// unused connections to the same endpoint which might be in the cache.
}
std::string AgencyCommManager::redirect(std::unique_ptr<httpclient::GeneralClientConnection> connection,
std::string const& endpoint,
std::string const& location, std::string& url) {
MUTEX_LOCKER(locker, _lock);
std::string specification;
size_t delim = std::string::npos;
if (location.substr(0, 7) == "http://") {
specification = "http+tcp://" + location.substr(7);
delim = specification.find_first_of('/', 12);
} else if (location.substr(0, 8) == "https://") {
specification = "http+ssl://" + location.substr(8);
delim = specification.find_first_of('/', 13);
}
// invalid location header
if (delim == std::string::npos) {
failedNonLocking(std::move(connection), endpoint);
return "";
}
std::string rest = specification.substr(delim);
specification = Endpoint::unifiedForm(specification.substr(0, delim));
LOG_TOPIC("1ef82", TRACE, Logger::AGENCYCOMM)
<< "redirect: location = " << location
<< ", specification = " << specification << ", url = " << rest;
if (endpoint == specification) {
LOG_TOPIC("14be3", DEBUG, Logger::AGENCYCOMM)
<< "got an agency redirect back to the old agency '" << endpoint << "'";
failedNonLocking(std::move(connection), endpoint);
return "";
}
// The following is useful if another thread learned about a redirect first.
// In this case we simply release the connection and let the client acquire
// another one for the (new) active endpoint.
if (endpoint != _endpoints.front()) {
LOG_TOPIC("88d87", DEBUG, Logger::AGENCYCOMM)
<< "ignoring an agency redirect to '" << specification
<< "' from inactive endpoint '" << endpoint << "'";
releaseNonLocking(std::move(connection), endpoint);
return "";
}
url = rest;
// Move redirected endpoint to the front to make it active:
_endpoints.erase(std::remove(_endpoints.begin(), _endpoints.end(), specification),
_endpoints.end());
LOG_TOPIC("e6514", DEBUG, Logger::AGENCYCOMM) << "Got an agency redirect from '" << endpoint
<< "' to '" << specification << "'";
_endpoints.push_front(specification);
return specification;
}
void AgencyCommManager::addEndpoint(std::string const& endpoint) {
MUTEX_LOCKER(locker, _lock);
std::string normalized = Endpoint::unifiedForm(endpoint);
auto iter = _endpoints.begin();
for (; iter != _endpoints.end(); ++iter) {
if (*iter == normalized) {
break;
}
}
if (iter == _endpoints.end()) {
LOG_TOPIC("b3062", DEBUG, Logger::AGENCYCOMM) << "using agency endpoint '" << normalized << "'";
_endpoints.emplace_back(normalized);
}
}
void AgencyCommManager::updateEndpoints(std::vector<std::string> const& newEndpoints) {
std::set<std::string> updatedSet;
for (std::string const& endp : newEndpoints) {
updatedSet.emplace(Endpoint::unifiedForm(endp));
}
MUTEX_LOCKER(locker, _lock);
std::set<std::string> currentSet;
currentSet.insert(_endpoints.begin(), _endpoints.end());
std::set<std::string> toRemove;
std::set_difference(currentSet.begin(), currentSet.end(), updatedSet.begin(),
updatedSet.end(), std::inserter(toRemove, toRemove.begin()));
std::set<std::string> toAdd;
std::set_difference(updatedSet.begin(), updatedSet.end(), currentSet.begin(),
currentSet.end(), std::inserter(toAdd, toAdd.begin()));
for (std::string const& rem : toRemove) {
LOG_TOPIC("7fc66", INFO, Logger::AGENCYCOMM) << "Removing endpoint " << rem << " from agent pool";
_endpoints.erase(std::remove(_endpoints.begin(), _endpoints.end(), rem),
_endpoints.end());
}
for (std::string const& add : toAdd) {
LOG_TOPIC("acda0", INFO, Logger::AGENCYCOMM) << "Adding endpoint " << add << " to agent pool";
_endpoints.emplace_back(add);
}
}
std::string AgencyCommManager::endpointsString() const {
return basics::StringUtils::join(endpoints(), ", ");
}
std::vector<std::string> AgencyCommManager::endpoints() const {
std::vector<std::string> result;
MUTEX_LOCKER(locker, _lock);
result.insert(result.begin(), _endpoints.begin(), _endpoints.end());
return result;
}
std::unique_ptr<GeneralClientConnection> AgencyCommManager::createNewConnection() {
if (_endpoints.empty()) {
LOG_TOPIC("3398c", TRACE, Logger::AGENCYCOMM)
<< "no agency endpoint is know, cannot create connection";
return nullptr;
}
std::string const& spec = _endpoints.front();
std::unique_ptr<Endpoint> endpoint(Endpoint::clientFactory(spec));
if (endpoint.get() == nullptr) {
LOG_TOPIC("82087", ERR, arangodb::Logger::AGENCYCOMM)
<< "invalid value for "
<< "--server.endpoint ('" << spec << "')";
THROW_ARANGO_EXCEPTION(TRI_ERROR_BAD_PARAMETER);
}
return std::unique_ptr<GeneralClientConnection>(
GeneralClientConnection::factory(endpoint, CONNECTION_OPTIONS._requestTimeout,
CONNECTION_OPTIONS._connectTimeout,
CONNECTION_OPTIONS._connectRetries, 0));
}
void AgencyCommManager::switchCurrentEndpoint() {
_lock.assertLockedByCurrentThread();
if (_endpoints.empty()) {
return;
}
std::string current = _endpoints.front();
_endpoints.pop_front();
_endpoints.push_back(current);
LOG_TOPIC("479c8", TRACE, Logger::AGENCYCOMM)
<< "switching active agency endpoint from '" << current << "' to '"
<< _endpoints.front() << "'";
}
// -----------------------------------------------------------------------------
// --SECTION-- AgencyComm
// -----------------------------------------------------------------------------
std::string const AgencyComm::AGENCY_URL_PREFIX = "/_api/agency";
AgencyCommResult AgencyComm::sendServerState(double ttl) {
// construct JSON value { "status": "...", "time": "..." }
VPackBuilder builder;
try {
builder.openObject();
std::string const status =
ServerState::stateToString(ServerState::instance()->getState());
builder.add("status", VPackValue(status));
std::string const stamp = AgencyCommManager::generateStamp();
builder.add("time", VPackValue(stamp));
builder.close();
} catch (...) {
return AgencyCommResult();
}
AgencyCommResult result(
setTransient("Sync/ServerStates/" + ServerState::instance()->getId(),
builder.slice(), ttl));
return result;
}
std::string AgencyComm::version() {
AgencyCommResult result =
sendWithFailover(arangodb::rest::RequestType::GET,
AgencyCommManager::CONNECTION_OPTIONS._requestTimeout,
"/_api/version", VPackSlice::noneSlice());
if (result.successful()) {
return result._body;
}
return "";
}
AgencyCommResult AgencyComm::createDirectory(std::string const& key) {
VPackBuilder builder;
{ VPackObjectBuilder dir(&builder); }
AgencyOperation operation(key, AgencyValueOperationType::SET, builder.slice());
AgencyWriteTransaction transaction(operation);
return sendTransactionWithFailover(transaction);
}
AgencyCommResult AgencyComm::setValue(std::string const& key,
std::string const& value, double ttl) {
VPackBuilder builder;
builder.add(VPackValue(value));
AgencyOperation operation(key, AgencyValueOperationType::SET, builder.slice());
operation._ttl = static_cast<uint64_t>(ttl);
AgencyWriteTransaction transaction(operation);
return sendTransactionWithFailover(transaction);
}
AgencyCommResult AgencyComm::setValue(std::string const& key,
arangodb::velocypack::Slice const& slice, double ttl) {
AgencyOperation operation(key, AgencyValueOperationType::SET, slice);
operation._ttl = static_cast<uint64_t>(ttl);
AgencyWriteTransaction transaction(operation);
return sendTransactionWithFailover(transaction);
}
AgencyCommResult AgencyComm::setTransient(std::string const& key,
arangodb::velocypack::Slice const& slice,
double ttl) {
AgencyOperation operation(key, AgencyValueOperationType::SET, slice);
operation._ttl = static_cast<uint64_t>(ttl);
AgencyTransientTransaction transaction(operation);
return sendTransactionWithFailover(transaction);
}
bool AgencyComm::exists(std::string const& key) {
AgencyCommResult result = getValues(key);
if (!result.successful()) {
return false;
}
auto parts = arangodb::basics::StringUtils::split(key, "/");
std::vector<std::string> allParts;
allParts.reserve(parts.size() + 1);
allParts.push_back(AgencyCommManager::path());
allParts.insert(allParts.end(), parts.begin(), parts.end());
VPackSlice slice = result.slice()[0].get(allParts);
return !slice.isNone();
}
AgencyCommResult AgencyComm::getValues(std::string const& key) {
std::string url = AgencyComm::AGENCY_URL_PREFIX + "/read";
VPackBuilder builder;
{
VPackArrayBuilder root(&builder);
{
VPackArrayBuilder keys(&builder);
builder.add(VPackValue(AgencyCommManager::path(key)));
}
}
AgencyCommResult result =
sendWithFailover(arangodb::rest::RequestType::POST,
AgencyCommManager::CONNECTION_OPTIONS._requestTimeout,
url, builder.slice());
if (!result.successful()) {
return result;
}
try {
result.setVPack(VPackParser::fromJson(result.bodyRef()));
if (!result.slice().isArray()) {
result.set(500, "got invalid result structure for getValues response");
return result;
}
if (result.slice().length() != 1) {
result.set(500,
"got invalid result structure length for getValues response");
return result;
}
result._body.clear();
result._statusCode = 200;
} catch (std::exception const& e) {
LOG_TOPIC("a6906", ERR, Logger::AGENCYCOMM) << "Error transforming result: " << e.what();
result.clear();
} catch (...) {
LOG_TOPIC("5391b", ERR, Logger::AGENCYCOMM)
<< "Error transforming result: out of memory";
result.clear();
}
return result;
}
AgencyCommResult AgencyComm::dump() {
std::string url = AgencyComm::AGENCY_URL_PREFIX + "/state";
AgencyCommResult result =
sendWithFailover(
arangodb::rest::RequestType::GET,
AgencyCommManager::CONNECTION_OPTIONS._requestTimeout,
url, VPackSlice::noneSlice());
if (!result.successful()) {
return result;
}
try {
result.setVPack(VPackParser::fromJson(result.bodyRef()));
result._body.clear();
result._statusCode = 200;
} catch (std::exception const& e) {
LOG_TOPIC("8da8e", ERR, Logger::AGENCYCOMM) << "Error transforming result: " << e.what();
result.clear();
} catch (...) {
LOG_TOPIC("b3eef", ERR, Logger::AGENCYCOMM)
<< "Error transforming result: out of memory";
result.clear();
}
return result;
}
AgencyCommResult AgencyComm::removeValues(std::string const& key, bool recursive) {
AgencyWriteTransaction transaction(AgencyOperation(key, AgencySimpleOperationType::DELETE_OP));
return sendTransactionWithFailover(transaction);
}
AgencyCommResult AgencyComm::increment(std::string const& key) {
AgencyWriteTransaction transaction(
AgencyOperation(key, AgencySimpleOperationType::INCREMENT_OP));
return sendTransactionWithFailover(transaction);
}
AgencyCommResult AgencyComm::casValue(std::string const& key,
arangodb::velocypack::Slice const& json,
bool prevExist, double ttl, double timeout) {
VPackBuilder newBuilder;
newBuilder.add(json);
AgencyOperation operation(key, AgencyValueOperationType::SET, newBuilder.slice());
AgencyPrecondition precondition(key, AgencyPrecondition::Type::EMPTY, !prevExist);
if (ttl >= 0.0) {
operation._ttl = static_cast<uint64_t>(ttl);
}
VPackBuilder preBuilder;
precondition.toVelocyPack(preBuilder);
AgencyWriteTransaction transaction(operation, precondition);
return sendTransactionWithFailover(transaction, timeout);
}
AgencyCommResult AgencyComm::casValue(std::string const& key, VPackSlice const& oldJson,
VPackSlice const& newJson, double ttl, double timeout) {
VPackBuilder newBuilder;
newBuilder.add(newJson);
VPackBuilder oldBuilder;
oldBuilder.add(oldJson);
AgencyOperation operation(key, AgencyValueOperationType::SET, newBuilder.slice());
AgencyPrecondition precondition(key, AgencyPrecondition::Type::VALUE,
oldBuilder.slice());
if (ttl >= 0.0) {
operation._ttl = static_cast<uint64_t>(ttl);
}
AgencyWriteTransaction transaction(operation, precondition);
return sendTransactionWithFailover(transaction, timeout);
}
uint64_t AgencyComm::uniqid(uint64_t count, double timeout) {
#ifdef DEBUG_SYNC_REPLICATION
if (AgencyComm::syncReplDebug == true) {
return debugUniqId++;
}
#endif
static int const maxTries = 1000000;
// this is pretty much forever, but we simply cannot continue at all
// if we do not get a unique id from the agency.
int tries = 0;
AgencyCommResult result;
uint64_t oldValue = 0;
while (tries++ < maxTries) {
result = getValues("Sync/LatestID");
if (!result.successful()) {
std::this_thread::sleep_for(std::chrono::microseconds(500000));
continue;
}
VPackSlice oldSlice = result.slice()[0].get(std::vector<std::string>(
{AgencyCommManager::path(), "Sync", "LatestID"}));
if (!(oldSlice.isSmallInt() || oldSlice.isUInt())) {
LOG_TOPIC("b30d9", WARN, Logger::AGENCYCOMM)
<< "Sync/LatestID in agency is not an unsigned integer, fixing...";
try {
VPackBuilder builder;
builder.add(VPackValue(0));
// create the key on the fly
setValue("Sync/LatestID", builder.slice(), 0.0);
} catch (...) {
// Could not build local key. Try again
}
continue;
}
// If we get here, slice is pointing to an unsigned integer, which
// is the value in the agency.
oldValue = 0;
try {
oldValue = oldSlice.getUInt();
} catch (...) {
}
uint64_t const newValue = oldValue + count;
VPackBuilder newBuilder;
try {
newBuilder.add(VPackValue(newValue));
} catch (...) {
std::this_thread::sleep_for(std::chrono::microseconds(500000));
continue;
}
result = casValue("Sync/LatestID", oldSlice, newBuilder.slice(), 0.0, timeout);
if (result.successful()) {
break;
}
// The cas did not work, simply try again!
}
return oldValue;
}
AgencyCommResult AgencyComm::registerCallback(std::string const& key,
std::string const& endpoint) {
VPackBuilder builder;
builder.add(VPackValue(endpoint));
AgencyOperation operation(key, AgencyValueOperationType::OBSERVE, builder.slice());
AgencyWriteTransaction transaction(operation);
AgencyCommResult res;
for (size_t i = 0; i < 3; ++i) {
res = sendTransactionWithFailover(transaction);
if (res.successful()) {
return res;
}
}
return res;
}
AgencyCommResult AgencyComm::unregisterCallback(std::string const& key,
std::string const& endpoint) {
VPackBuilder builder;
builder.add(VPackValue(endpoint));
AgencyOperation operation(key, AgencyValueOperationType::UNOBSERVE, builder.slice());
AgencyWriteTransaction transaction(operation);
return sendTransactionWithFailover(transaction);
}
bool AgencyComm::lockRead(std::string const& key, double ttl, double timeout) {
VPackBuilder builder;
try {
builder.add(VPackValue("READ"));
} catch (...) {
return false;
}
return lock(key, ttl, timeout, builder.slice());
}
bool AgencyComm::lockWrite(std::string const& key, double ttl, double timeout) {
VPackBuilder builder;
try {
builder.add(VPackValue("WRITE"));
} catch (...) {
return false;
}
return lock(key, ttl, timeout, builder.slice());
}
bool AgencyComm::unlockRead(std::string const& key, double timeout) {
VPackBuilder builder;
try {
builder.add(VPackValue("READ"));
} catch (...) {
return false;
}
return unlock(key, builder.slice(), timeout);
}
bool AgencyComm::unlockWrite(std::string const& key, double timeout) {
VPackBuilder builder;
try {
builder.add(VPackValue("WRITE"));
} catch (...) {
return false;
}
return unlock(key, builder.slice(), timeout);
}
AgencyCommResult AgencyComm::sendTransactionWithFailover(AgencyTransaction const& transaction,
double timeout) {
std::string url = AgencyComm::AGENCY_URL_PREFIX;
url += transaction.path();
VPackBuilder builder;
{
VPackArrayBuilder guard(&builder);
transaction.toVelocyPack(builder);
}
LOG_TOPIC("4e440", TRACE, Logger::AGENCYCOMM)
<< "sending " << builder.toJson() << "'" << url << "'";
AgencyCommResult result =
sendWithFailover(arangodb::rest::RequestType::POST,
(timeout == 0.0) ? AgencyCommManager::CONNECTION_OPTIONS._requestTimeout
: timeout,
url, builder.slice());
if (!result.successful() &&
result.httpCode() != (int)arangodb::rest::ResponseCode::PRECONDITION_FAILED) {
return result;
}
try {
result.setVPack(VPackParser::fromJson(result.bodyRef()));
if (!transaction.validate(result)) {
result.set(500, std::string("validation failed for response to URL " + url));
LOG_TOPIC("f2083", DEBUG, Logger::AGENCYCOMM)
<< "validation failed for url: " << url
<< ", type: " << transaction.typeName()
<< ", sent: " << builder.toJson() << ", received: " << result.bodyRef();
return result;
}
result._body.clear();
} catch (std::exception const& e) {
LOG_TOPIC("e13a5", ERR, Logger::AGENCYCOMM)
<< "Error transforming result: " << e.what()
<< ", status code: " << result._statusCode
<< ", incriminating body: " << result.bodyRef() << ", url: " << url
<< ", timeout: " << timeout << ", data sent: " << builder.toJson();
result.clear();
} catch (...) {
LOG_TOPIC("18245", ERR, Logger::AGENCYCOMM)
<< "Error transforming result: out of memory";
result.clear();
}
return result;
}
bool AgencyComm::ensureStructureInitialized() {
LOG_TOPIC("748e2", TRACE, Logger::AGENCYCOMM) << "checking if agency is initialized";
while (!application_features::ApplicationServer::isStopping() &&
shouldInitializeStructure()) {
LOG_TOPIC("17e16", TRACE, Logger::AGENCYCOMM)
<< "Agency is fresh. Needs initial structure.";
if (tryInitializeStructure()) {
LOG_TOPIC("4c5aa", TRACE, Logger::AGENCYCOMM)
<< "Successfully initialized agency";
break;
}
LOG_TOPIC("63f7b", 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("9d265", TRACE, Logger::AGENCYCOMM)
<< "Waiting for agency to get initialized";
std::this_thread::sleep_for(std::chrono::seconds(1));
}
return true;
}
bool AgencyComm::lock(std::string const& key, double ttl, double timeout,
VPackSlice const& slice) {
if (ttl == 0.0) {
ttl = AgencyCommManager::CONNECTION_OPTIONS._lockTimeout;
}
if (timeout == 0.0) {
timeout = AgencyCommManager::CONNECTION_OPTIONS._lockTimeout;
}
unsigned long sleepTime = INITIAL_SLEEP_TIME;
double const end = TRI_microtime() + timeout;
VPackBuilder builder;
try {
builder.add(VPackValue("UNLOCKED"));
} catch (...) {
return false;
}
VPackSlice oldSlice = builder.slice();
while (true) {
AgencyCommResult result = casValue(key + "/Lock", oldSlice, slice, ttl, timeout);
if (!result.successful() &&
result.httpCode() == (int)arangodb::rest::ResponseCode::PRECONDITION_FAILED) {
// key does not yet exist. create it now
result = casValue(key + "/Lock", slice, false, ttl, timeout);
}
if (result.successful()) {
return true;
}
std::this_thread::sleep_for(std::chrono::microseconds(sleepTime));
if (sleepTime < MAX_SLEEP_TIME) {
sleepTime += INITIAL_SLEEP_TIME;
}
if (TRI_microtime() >= end) {
return false;
}
}
TRI_ASSERT(false);
return false;
}
bool AgencyComm::unlock(std::string const& key, VPackSlice const& slice, double timeout) {
if (timeout == 0.0) {
timeout = AgencyCommManager::CONNECTION_OPTIONS._lockTimeout;
}
unsigned long sleepTime = INITIAL_SLEEP_TIME;
double const end = TRI_microtime() + timeout;
VPackBuilder builder;
try {
builder.add(VPackValue("UNLOCKED"));
} catch (...) {
// Out of Memory
return false;
}
VPackSlice newSlice = builder.slice();
while (true) {
AgencyCommResult result = casValue(key + "/Lock", slice, newSlice, 0.0, timeout);
if (result.successful()) {
return true;
}
std::this_thread::sleep_for(std::chrono::microseconds(sleepTime));
if (sleepTime < MAX_SLEEP_TIME) {
sleepTime += INITIAL_SLEEP_TIME;
}
if (TRI_microtime() >= end) {
return false;
}
}
TRI_ASSERT(false);
return false;
}
AgencyCommResult AgencyComm::sendWithFailover(arangodb::rest::RequestType method,
double const timeout,
std::string const& initialUrl,
VPackSlice inBody) {
AgencyCommResult result;
if (!AgencyCommManager::isEnabled()) {
LOG_TOPIC("42fae", ERR, Logger::AGENCYCOMM)
<< "No AgencyCommManager. Inappropriate agent usage?";
result.set(503, "No AgencyCommManager. Inappropriate agent usage?");
return result;
} // if
std::string endpoint;
std::unique_ptr<GeneralClientConnection> connection =
AgencyCommManager::MANAGER->acquire(endpoint);
std::vector<std::string> clientIds;
VPackSlice body = inBody.resolveExternals();
if (body.isArray()) {
// In the writing case we want to find all transactions with client IDs
// and remember these IDs:
for (auto const& query : VPackArrayIterator(body)) {
if (query.isArray() && query.length() == 3 && query[0].isObject() &&
query[2].isString()) {
clientIds.push_back(query[2].copyString());
}
}
}
std::string url;
std::chrono::duration<double> waitInterval(.0); // seconds
auto started = std::chrono::steady_clock::now();
auto timeOut = started + std::chrono::duration<double>(timeout);
double conTimeout = 1.0;
int tries = 0;
auto waitSomeTime = [&waitInterval, &result]() -> bool {
// Returning true means timeout because of shutdown:
if (application_features::ApplicationServer::isStopping()) {
LOG_TOPIC("53e58", INFO, Logger::AGENCYCOMM)
<< "Unsuccessful AgencyComm: Timeout because of shutdown "
<< "errorCode: " << result.errorCode()
<< " errorMessage: " << result.errorMessage()
<< " errorDetails: " << result.errorDetails();
return true;
}
// Raise waits to a maximum 10 seconds
auto waitUntil = std::chrono::steady_clock::now() + waitInterval;
std::this_thread::sleep_until(waitUntil);
if (waitInterval.count() == 0.0) {
waitInterval = std::chrono::duration<double>(0.25);
} else if (waitInterval.count() < 5.0) {
waitInterval *= 1.0749292929292;
}
// Check again for shutdown, since some time has passed:
if (application_features::ApplicationServer::isStopping()) {
LOG_TOPIC("afe45", INFO, Logger::AGENCYCOMM)
<< "Unsuccessful AgencyComm: Timeout because of shutdown "
<< "errorCode: " << result.errorCode()
<< " errorMessage: " << result.errorMessage()
<< " errorDetails: " << result.errorDetails();
return true;
}
return false;
};
bool isInquiry = false; // Set to true whilst we investigate a potentially
// failed transaction.
static std::string const writeURL{"/_api/agency/write"};
bool isWriteTrans = (initialUrl == writeURL);
while (true) { // will be left by timeout eventually
// If for some reason we did not find an agency endpoint, we bail out:
if (connection == nullptr) {
LOG_TOPIC("87cd7", ERR, Logger::AGENCYCOMM) << "No agency endpoints.";
result.set(400, "No endpoints for agency found.");
break;
}
// From second loop on delay a bit:
if (tries++ > 0) {
if (waitSomeTime()) {
break; // timeout because of shutdown
}
}
// timeout exit strategy:
if (std::chrono::steady_clock::now() > timeOut) {
result.set(0, "timeout in AgencyComm operation");
break;
}
double elapsed =
1.e-2 * (std::round(1.e+2 * std::chrono::duration<double>(
std::chrono::steady_clock::now() - started)
.count()));
// Some reporting:
if (tries > 20) {
std::string serverState = application_features::ApplicationServer::server->stringifyState();
LOG_TOPIC("2f181", INFO, Logger::AGENCYCOMM)
<< "Flaky agency communication to " << endpoint
<< ". Unsuccessful consecutive tries: " << tries << " (" << elapsed
<< "s). Network checks advised."
<< " Server " << serverState << ".";
}
if (1 < tries) {
LOG_TOPIC("20afa", DEBUG, Logger::AGENCYCOMM)
<< "Retrying agency communication at '" << endpoint
<< ". Unsuccessful consecutive tries: " << tries << " (" << elapsed
<< "s). Network checks advised.";
}
if (!isInquiry) {
// try to send transaction; if we fail completely, retry
try {
std::string bodyString;
if (!body.isNone()) {
bodyString = body.toJson();
}
url = initialUrl; // Attention: overwritten by redirect below!
result = send(connection.get(), method, conTimeout, url, bodyString);
#ifdef ARANGODB_ENABLE_MAINTAINER_MODE
if (!clientIds.empty()) {
if (clientIds[0] == "INTEGRATION_TEST_INQUIRY_ERROR_0") {
result._statusCode = 0;
} else if (clientIds[0] == "INTEGRATION_TEST_INQUIRY_ERROR_503") {
result._statusCode = 503;
}
}
#endif
} catch (...) {
// Rotate to new agent endpoint:
AgencyCommManager::MANAGER->failed(std::move(connection), endpoint);
endpoint.clear();
connection = AgencyCommManager::MANAGER->acquire(endpoint);
continue;
}
// got a result or shutdown, we are done
if (result.successful() || application_features::ApplicationServer::isStopping()) {
AgencyCommManager::MANAGER->release(std::move(connection), endpoint);
break;
}
// do not retry on client errors
if (result._statusCode >= 400 && result._statusCode <= 499) {
AgencyCommManager::MANAGER->release(std::move(connection), endpoint);
break;
}
// Note that in case of a timeout or 503 we do not know whether the
// operation has been executed. In this case we should inquire about
// the operation. If it actually was done, we are good. If not, we
// can retry. If in doubt, we have to retry inquire until the global
// timeout is reached.
// Also note that we only inquire about WriteTransactions.
if (isWriteTrans && !clientIds.empty() && result._sent &&
(result._statusCode == 0 || result._statusCode == 503)) {
isInquiry = true;
conTimeout = 16.0;
}
// This leaves the redirect, timeout and 503 cases, which are handled
// below.
} else {
// isInquiry case, we are getting here because a previous try to send
// the transaction lead to isInquiry == true because we got a timeout
// or a 503.
VPackBuilder b;
{
VPackArrayBuilder ab(&b);
for (auto const& i : clientIds) {
b.add(VPackValue(i));
}
}
LOG_TOPIC("9b411", DEBUG, Logger::AGENCYCOMM)
<< "Failed agency comm (" << result._statusCode << ")! "
<< "Inquiring about clientIds " << clientIds << ".";
url = "/_api/agency/inquire"; // attention: overwritten by redirect!
result = send(connection.get(), method, conTimeout, url, b.toJson());
// Inquire returns a body like write, if the transactions are not known,
// the list of results is empty.
// _statusCode can be 200 or 404 (if nothing was found)
if (result.successful()) {
std::shared_ptr<VPackBuilder> resultBody = VPackParser::fromJson(result._body);
VPackSlice outer = resultBody->slice();
// If we get an answer, and it contains a "results" key,
// we release the connection and break out of the loop letting the
// inquiry result go to the client. Otherwise try again.
if (outer.isObject() && outer.hasKey("results")) {
VPackSlice results = outer.get("results");
if (results.length() > 0) {
LOG_TOPIC("507f5", DEBUG, Logger::AGENCYCOMM) << "Inquired " << resultBody->toJson();
AgencyCommManager::MANAGER->release(std::move(connection), endpoint);
break;
} else {
// Nothing known, so do a retry of the original operation:
LOG_TOPIC("a25ef", DEBUG, Logger::AGENCYCOMM)
<< "Nothing known, do a retry.";
isInquiry = false;
continue;
}
} else {
// How odd, we are supposed to get at least {results=[...]}, let's retry...
isInquiry = false;
continue;
}
} else if (result._statusCode == 404) {
// clientId was not found, agency has never heard of this trx, retry:
isInquiry = false;
continue;
}
// This can still be a timeout or 503 case, both are handled below
} // end of inquiry case
// Now do some stuff which is needed in both cases:
// sometimes the agency will return a 307 (temporary redirect)
// in this case we have to pick it up and use the new location returned
if (result._statusCode == (int)arangodb::rest::ResponseCode::TEMPORARY_REDIRECT) {
// Note that this may overwrite url, but we do not care.
endpoint = AgencyCommManager::MANAGER->redirect(std::move(connection), endpoint,
result._location, url);
connection = AgencyCommManager::MANAGER->acquire(endpoint);
waitInterval = std::chrono::duration<double>(0.0);
continue;
}
// In case of a timeout, we increase the patience:
if (result._statusCode == 0) {
if (conTimeout < 33.0) { // double until we have 64s
conTimeout *= 2;
}
}
if (result._statusCode == 0 || result._statusCode == 503) {
// Rotate to new agent endpoint:
AgencyCommManager::MANAGER->failed(std::move(connection), endpoint);
endpoint.clear();
connection = AgencyCommManager::MANAGER->acquire(endpoint);
}
}
// Log error
if (!result.successful() && result.httpCode() != 412) {
LOG_TOPIC("78466", DEBUG, Logger::AGENCYCOMM)
<< "Unsuccessful AgencyComm: "
<< "errorCode: " << result.errorCode()
<< " errorMessage: " << result.errorMessage()
<< " errorDetails: " << result.errorDetails();
}
return result;
}
AgencyCommResult AgencyComm::send(arangodb::httpclient::GeneralClientConnection* connection,
arangodb::rest::RequestType method, double timeout,
std::string const& url, std::string const& body) {
TRI_ASSERT(connection != nullptr);
if (method == arangodb::rest::RequestType::GET ||
method == arangodb::rest::RequestType::HEAD ||
method == arangodb::rest::RequestType::DELETE_REQ) {
TRI_ASSERT(body.empty());
}
TRI_ASSERT(!url.empty());
AgencyCommResult result;
LOG_TOPIC("47733", TRACE, Logger::AGENCYCOMM)
<< "sending " << arangodb::HttpRequest::translateMethod(method)
<< " request to agency at endpoint '"
<< connection->getEndpoint()->specification() << "', url '" << url
<< "': " << body;
arangodb::httpclient::SimpleHttpClientParams params(timeout, false);
AuthenticationFeature* af = AuthenticationFeature::instance();
TRI_ASSERT(af != nullptr);
params.setJwt(af->tokenCache().jwtToken());
params.keepConnectionOnDestruction(true);
arangodb::httpclient::SimpleHttpClient client(connection, params);
// set up headers
std::unordered_map<std::string, std::string> headers;
if (method == arangodb::rest::RequestType::POST) {
// the agency needs this content-type for the body
headers["content-type"] = "application/json";
}
// send the actual request
std::unique_ptr<arangodb::httpclient::SimpleHttpResult> response(
client.request(method, url, body.c_str(), body.size(), headers));
if (response == nullptr) {
result._message = "could not send request to agency";
LOG_TOPIC("4366b", TRACE, Logger::AGENCYCOMM) << "could not send request to agency";
return result;
}
result._sent = response->haveSentRequestFully();
if (!response->isComplete()) {
result._message = "sending request to agency failed";
LOG_TOPIC("f305c", TRACE, Logger::AGENCYCOMM) << "sending request to agency failed";
return result;
}
result._connected = true;
if (response->getHttpReturnCode() == (int)arangodb::rest::ResponseCode::TEMPORARY_REDIRECT) {
// temporary redirect. now save location header
bool found = false;
result._location = response->getHeaderField(StaticStrings::Location, found);
LOG_TOPIC("92680", TRACE, Logger::AGENCYCOMM)
<< "redirecting to location: '" << result._location << "'";
if (!found) {
// a 307 without a location header does not make any sense
result._message = "invalid agency response (header missing)";
return result;
}
}
result._message = response->getHttpReturnMessage();
result._statusCode = response->getHttpReturnCode();
basics::StringBuffer& sb = response->getBody();
result._body = std::string(sb.c_str(), sb.length());
LOG_TOPIC("97dc9", TRACE, Logger::AGENCYCOMM)
<< "request to agency returned status code " << result._statusCode
<< ", message: '" << result._message << "', body: '" << result._body << "'";
return result;
}
bool AgencyComm::tryInitializeStructure() {
VPackBuilder builder;
try {
VPackObjectBuilder b(&builder);
builder.add( // Cluster Id --------------------------
"Cluster", VPackValue(to_string(boost::uuids::random_generator()())));
builder.add(VPackValue("Agency")); // Agency ------------------------------
{
VPackObjectBuilder a(&builder);
builder.add("Definition", VPackValue(1));
}
builder.add(VPackValue("Current")); // Current ----------------------------
{
VPackObjectBuilder c(&builder);
addEmptyVPackObject("AsyncReplication", builder);
builder.add(VPackValue("Collections"));
{
VPackObjectBuilder d(&builder);
addEmptyVPackObject("_system", builder);
}
builder.add("Version", VPackValue(1));
addEmptyVPackObject("ShardsCopied", builder);
addEmptyVPackObject("NewServers", builder);
addEmptyVPackObject("Coordinators", builder);
builder.add("Lock", VPackValue("UNLOCKED"));
addEmptyVPackObject("DBServers", builder);
addEmptyVPackObject("Singles", builder);
builder.add(VPackValue("ServersRegistered"));
{
VPackObjectBuilder c(&builder);
builder.add("Version", VPackValue(1));
}
addEmptyVPackObject("Databases", builder);
}
builder.add("InitDone", VPackValue(true)); // InitDone
builder.add(VPackValue("Plan")); // Plan ----------------------------------
{
VPackObjectBuilder c(&builder);
addEmptyVPackObject("AsyncReplication", builder);
addEmptyVPackObject("Coordinators", builder);
builder.add(VPackValue("Databases"));
{
VPackObjectBuilder d(&builder);
builder.add(VPackValue("_system"));
{
VPackObjectBuilder d2(&builder);
builder.add("name", VPackValue("_system"));
builder.add("id", VPackValue("1"));
}
}
builder.add("Lock", VPackValue("UNLOCKED"));
addEmptyVPackObject("DBServers", builder);
addEmptyVPackObject("Singles", builder);
builder.add("Version", VPackValue(1));
builder.add(VPackValue("Collections"));
{
VPackObjectBuilder d(&builder);
addEmptyVPackObject("_system", builder);
}
builder.add(VPackValue("Views"));
{
VPackObjectBuilder d(&builder);
addEmptyVPackObject("_system", builder);
}
}
builder.add(VPackValue("Sync")); // Sync ----------------------------------
{
VPackObjectBuilder c(&builder);
builder.add("LatestID", VPackValue(1));
addEmptyVPackObject("Problems", builder);
builder.add("UserVersion", VPackValue(1));
addEmptyVPackObject("ServerStates", builder);
builder.add("HeartbeatIntervalMs", VPackValue(1000));
}
builder.add(VPackValue("Supervision")); // Supervision --------------------
{
VPackObjectBuilder c(&builder);
addEmptyVPackObject("Health", builder);
addEmptyVPackObject("Shards", builder);
addEmptyVPackObject("DBServers", builder);
}
builder.add(VPackValue("Target")); // Target ------------------------------
{
VPackObjectBuilder c(&builder);
builder.add("NumberOfCoordinators", VPackSlice::nullSlice());
builder.add("NumberOfDBServers", VPackSlice::nullSlice());
builder.add(VPackValue("CleanedServers"));
{ VPackArrayBuilder dd(&builder); }
builder.add(VPackValue("ToBeCleanedServers"));
{ VPackArrayBuilder dd(&builder); }
builder.add(VPackValue("FailedServers"));
{ VPackObjectBuilder dd(&builder); }
builder.add("Lock", VPackValue("UNLOCKED"));
// MapLocalToID is not used for anything since 3.4. It was used in
// previous versions to store server ids from --cluster.my-local-info that
// were mapped to server UUIDs
addEmptyVPackObject("MapLocalToID", builder);
addEmptyVPackObject("Failed", builder);
addEmptyVPackObject("Finished", builder);
addEmptyVPackObject("Pending", builder);
addEmptyVPackObject("ToDo", builder);
builder.add("Version", VPackValue(1));
}
} catch (std::exception const& e) {
LOG_TOPIC("bef45", ERR, Logger::AGENCYCOMM)
<< "Couldn't create initializing structure " << e.what();
return false;
} catch (...) {
LOG_TOPIC("4b5c2", ERR, Logger::AGENCYCOMM)
<< "Couldn't create initializing structure";
return false;
}
try {
LOG_TOPIC("58ffe", TRACE, Logger::AGENCYCOMM)
<< "Initializing agency with " << builder.toJson();
AgencyWriteTransaction initTransaction(
AgencyOperation("", AgencyValueOperationType::SET, builder.slice()),
AgencyPrecondition("Plan", AgencyPrecondition::Type::EMPTY, true));
AgencyCommResult result = sendTransactionWithFailover(initTransaction);
if (result.httpCode() == TRI_ERROR_HTTP_UNAUTHORIZED) {
LOG_TOPIC("a695d", ERR, Logger::AUTHENTICATION)
<< "Cannot authenticate with agency,"
<< " check value of --server.jwt-secret";
}
return result.successful();
} catch (std::exception const& e) {
LOG_TOPIC("0174e", FATAL, Logger::AGENCYCOMM)
<< "Fatal error initializing agency " << e.what();
FATAL_ERROR_EXIT();
} catch (...) {
LOG_TOPIC("6cc28", FATAL, Logger::AGENCYCOMM) << "Fatal error initializing agency";
FATAL_ERROR_EXIT();
}
}
bool AgencyComm::shouldInitializeStructure() {
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("32781", FATAL, Logger::STARTUP) << "Unauthorized. Wrong credentials.";
FATAL_ERROR_EXIT();
}
// Agency not ready yet
LOG_TOPIC("36253", TRACE, Logger::AGENCYCOMM)
<< "waiting for agency to become ready";
continue;
} else {
// Sanity
if (result.slice().isArray() && result.slice().length() == 1) {
// No plan entry? Should initialize
if (result.slice()[0].isObject() && result.slice()[0].length() == 0) {
LOG_TOPIC("98732", DEBUG, Logger::AGENCYCOMM)
<< "agency initialization should be performed";
return true;
} else {
LOG_TOPIC("abedb", DEBUG, Logger::AGENCYCOMM)
<< "agency initialization under way or done";
return false;
}
} else {
// Should never get here
TRI_ASSERT(false);
if (nFail++ < 3) {
LOG_TOPIC("fed52", DEBUG, Logger::AGENCYCOMM) << "What the hell just happened?";
} else {
LOG_TOPIC("54fea", 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));
}
return false;
}
Reference in New Issue View Git Blame Copy Permalink