mirror of https://gitee.com/bigwinds/arangodb
1553 lines
52 KiB
C++
1553 lines
52 KiB
C++
////////////////////////////////////////////////////////////////////////////////
|
|
/// 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 Kaveh Vahedipour
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
|
|
#include "Supervision.h"
|
|
|
|
#include <thread>
|
|
|
|
#include "Agency/ActiveFailoverJob.h"
|
|
#include "Agency/AddFollower.h"
|
|
#include "Agency/Agent.h"
|
|
#include "Agency/CleanOutServer.h"
|
|
#include "Agency/FailedServer.h"
|
|
#include "Agency/Job.h"
|
|
#include "Agency/JobContext.h"
|
|
#include "Agency/RemoveFollower.h"
|
|
#include "Agency/Store.h"
|
|
#include "ApplicationFeatures/ApplicationServer.h"
|
|
#include "Basics/ConditionLocker.h"
|
|
#include "Basics/MutexLocker.h"
|
|
#include "Cluster/ServerState.h"
|
|
|
|
using namespace arangodb;
|
|
using namespace arangodb::consensus;
|
|
using namespace arangodb::application_features;
|
|
|
|
struct HealthRecord {
|
|
std::string shortName;
|
|
std::string syncTime;
|
|
std::string syncStatus;
|
|
std::string status;
|
|
std::string endpoint;
|
|
std::string advertisedEndpoint;
|
|
std::string lastAcked;
|
|
std::string hostId;
|
|
std::string serverVersion;
|
|
std::string engine;
|
|
size_t version;
|
|
|
|
explicit HealthRecord() : version(0) {}
|
|
|
|
HealthRecord(std::string const& sn, std::string const& ep, std::string const& ho,
|
|
std::string const& en, std::string const& sv, std::string const& ae)
|
|
: shortName(sn),
|
|
endpoint(ep),
|
|
advertisedEndpoint(ae),
|
|
hostId(ho),
|
|
serverVersion(sv),
|
|
engine(en),
|
|
version(0) {}
|
|
|
|
explicit HealthRecord(Node const& node) { *this = node; }
|
|
|
|
HealthRecord& operator=(Node const& node) {
|
|
version = 0;
|
|
if (shortName.empty()) {
|
|
shortName = node.hasAsString("ShortName").first;
|
|
}
|
|
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
|
|
version = 2;
|
|
syncStatus = node.hasAsString("SyncStatus").first;
|
|
if (node.has("SyncTime")) {
|
|
syncTime = node.hasAsString("SyncTime").first;
|
|
}
|
|
if (node.has("LastAckedTime")) {
|
|
lastAcked = node.hasAsString("LastAckedTime").first;
|
|
}
|
|
if (node.has("AdvertisedEndpoint")) {
|
|
version = 3;
|
|
advertisedEndpoint = node.hasAsString("AdvertisedEndpoint").first;
|
|
} else {
|
|
advertisedEndpoint.clear();
|
|
}
|
|
if (node.has("Engine") && node.has("Version")) {
|
|
version = 4;
|
|
engine = node.hasAsString("Engine").first;
|
|
serverVersion = node.hasAsString("Version").first;
|
|
} else {
|
|
engine.clear();
|
|
serverVersion.clear();
|
|
}
|
|
} else if (node.has("LastHeartbeatStatus")) {
|
|
version = 1;
|
|
syncStatus = node.hasAsString("LastHeartbeatStatus").first;
|
|
if (node.has("LastHeartbeatSent")) {
|
|
syncTime = node.hasAsString("LastHeartbeatSent").first;
|
|
}
|
|
if (node.has("LastHeartbeatAcked")) {
|
|
lastAcked = node.hasAsString("LastHeartbeatAcked").first;
|
|
}
|
|
}
|
|
}
|
|
return *this;
|
|
}
|
|
|
|
void toVelocyPack(VPackBuilder& obj) const {
|
|
TRI_ASSERT(obj.isOpenObject());
|
|
obj.add("ShortName", VPackValue(shortName));
|
|
obj.add("Endpoint", VPackValue(endpoint));
|
|
obj.add("Host", VPackValue(hostId));
|
|
obj.add("SyncStatus", VPackValue(syncStatus));
|
|
obj.add("Status", VPackValue(status));
|
|
obj.add("Version", VPackValue(serverVersion));
|
|
obj.add("Engine", VPackValue(engine));
|
|
if (!advertisedEndpoint.empty()) {
|
|
obj.add("AdvertisedEndpoint", VPackValue(advertisedEndpoint));
|
|
}
|
|
obj.add("Timestamp", VPackValue(timepointToString(std::chrono::system_clock::now())));
|
|
obj.add("SyncTime", VPackValue(syncTime));
|
|
obj.add("LastAckedTime", VPackValue(lastAcked));
|
|
}
|
|
|
|
bool statusDiff(HealthRecord const& other) {
|
|
return status != other.status || syncStatus != other.syncStatus ||
|
|
advertisedEndpoint != other.advertisedEndpoint ||
|
|
serverVersion != other.serverVersion || engine != other.engine ||
|
|
hostId != other.hostId || endpoint != other.endpoint;
|
|
}
|
|
|
|
friend std::ostream& operator<<(std::ostream& o, HealthRecord const& hr) {
|
|
VPackBuilder builder;
|
|
{
|
|
VPackObjectBuilder b(&builder);
|
|
hr.toVelocyPack(builder);
|
|
}
|
|
o << builder.toJson();
|
|
return o;
|
|
}
|
|
};
|
|
|
|
// This is initialized in AgencyFeature:
|
|
std::string Supervision::_agencyPrefix = "/arango";
|
|
|
|
Supervision::Supervision()
|
|
: arangodb::CriticalThread("Supervision"),
|
|
_agent(nullptr),
|
|
_snapshot("Supervision"),
|
|
_transient("Transient"),
|
|
_frequency(1.),
|
|
_gracePeriod(10.),
|
|
_okThreshold(5.),
|
|
_jobId(0),
|
|
_jobIdMax(0),
|
|
_haveAborts(false),
|
|
_selfShutdown(false),
|
|
_upgraded(false) {}
|
|
|
|
Supervision::~Supervision() {
|
|
shutdown();
|
|
}
|
|
|
|
static std::string const syncPrefix = "/Sync/ServerStates/";
|
|
static std::string const supervisionPrefix = "/Supervision";
|
|
static std::string const healthPrefix = "/Supervision/Health/";
|
|
static std::string const targetShortID = "/Target/MapUniqueToShortID/";
|
|
static std::string const currentServersRegisteredPrefix =
|
|
"/Current/ServersRegistered";
|
|
static std::string const foxxmaster = "/Current/Foxxmaster";
|
|
|
|
void Supervision::upgradeOne(Builder& builder) {
|
|
_lock.assertLockedByCurrentThread();
|
|
// "/arango/Agency/Definition" not exists or is 0
|
|
if (!_snapshot.has("Agency/Definition")) {
|
|
{
|
|
VPackArrayBuilder trx(&builder);
|
|
{
|
|
VPackObjectBuilder oper(&builder);
|
|
builder.add("/Agency/Definition", VPackValue(1));
|
|
builder.add(VPackValue("/Target/ToDo"));
|
|
{ VPackObjectBuilder empty(&builder); }
|
|
builder.add(VPackValue("/Target/Pending"));
|
|
{ VPackObjectBuilder empty(&builder); }
|
|
}
|
|
{
|
|
VPackObjectBuilder o(&builder);
|
|
builder.add(VPackValue("/Agency/Definition"));
|
|
{
|
|
VPackObjectBuilder prec(&builder);
|
|
builder.add("oldEmpty", VPackValue(true));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
void Supervision::upgradeZero(Builder& builder) {
|
|
_lock.assertLockedByCurrentThread();
|
|
// "/arango/Target/FailedServers" is still an array
|
|
Slice fails = _snapshot.hasAsSlice(failedServersPrefix).first;
|
|
if (fails.isArray()) {
|
|
{
|
|
VPackArrayBuilder trx(&builder);
|
|
{
|
|
VPackObjectBuilder o(&builder);
|
|
builder.add(VPackValue(failedServersPrefix));
|
|
{
|
|
VPackObjectBuilder oo(&builder);
|
|
if (fails.length() > 0) {
|
|
for (VPackSlice fail : VPackArrayIterator(fails)) {
|
|
builder.add(VPackValue(fail.copyString()));
|
|
{ VPackArrayBuilder ooo(&builder); }
|
|
}
|
|
}
|
|
}
|
|
}
|
|
} // trx
|
|
}
|
|
}
|
|
|
|
void Supervision::upgradeHealthRecords(Builder& builder) {
|
|
_lock.assertLockedByCurrentThread();
|
|
// "/arango/Supervision/health" is in old format
|
|
Builder b;
|
|
size_t n = 0;
|
|
|
|
if (_snapshot.has(healthPrefix)) {
|
|
HealthRecord hr;
|
|
{
|
|
VPackObjectBuilder oo(&b);
|
|
for (auto recPair : _snapshot.hasAsChildren(healthPrefix).first) {
|
|
if (recPair.second->has("ShortName") &&
|
|
recPair.second->has("Endpoint")) {
|
|
hr = *recPair.second;
|
|
if (hr.version == 1) {
|
|
++n;
|
|
b.add(VPackValue(recPair.first));
|
|
{
|
|
VPackObjectBuilder ooo(&b);
|
|
hr.toVelocyPack(b);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
if (n > 0) {
|
|
{
|
|
VPackArrayBuilder trx(&builder);
|
|
{
|
|
VPackObjectBuilder o(&builder);
|
|
b.add(healthPrefix, b.slice());
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Upgrade agency, guarded by wakeUp
|
|
void Supervision::upgradeAgency() {
|
|
_lock.assertLockedByCurrentThread();
|
|
|
|
Builder builder;
|
|
{
|
|
VPackArrayBuilder trxs(&builder);
|
|
upgradeZero(builder);
|
|
fixPrototypeChain(builder);
|
|
upgradeOne(builder);
|
|
upgradeHealthRecords(builder);
|
|
}
|
|
|
|
LOG_TOPIC(DEBUG, Logger::AGENCY) << "Upgrading the agency:" << builder.toJson();
|
|
|
|
if (builder.slice().length() > 0) {
|
|
generalTransaction(_agent, builder);
|
|
}
|
|
|
|
_upgraded = true;
|
|
}
|
|
|
|
void handleOnStatusDBServer(Agent* agent, Node const& snapshot,
|
|
HealthRecord& persisted, HealthRecord& transisted,
|
|
std::string const& serverID, uint64_t const& jobId,
|
|
std::shared_ptr<VPackBuilder>& envelope) {
|
|
std::string failedServerPath = failedServersPrefix + "/" + serverID;
|
|
// New condition GOOD:
|
|
if (transisted.status == Supervision::HEALTH_STATUS_GOOD) {
|
|
if (snapshot.has(failedServerPath)) {
|
|
envelope = std::make_shared<VPackBuilder>();
|
|
{
|
|
VPackArrayBuilder a(envelope.get());
|
|
{
|
|
VPackObjectBuilder operations(envelope.get());
|
|
envelope->add(VPackValue(failedServerPath));
|
|
{
|
|
VPackObjectBuilder ccc(envelope.get());
|
|
envelope->add("op", VPackValue("delete"));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
} else if ( // New state: FAILED persisted: GOOD (-> BAD)
|
|
persisted.status == Supervision::HEALTH_STATUS_GOOD &&
|
|
transisted.status != Supervision::HEALTH_STATUS_GOOD) {
|
|
transisted.status = Supervision::HEALTH_STATUS_BAD;
|
|
} else if ( // New state: FAILED persisted: BAD (-> Job)
|
|
persisted.status == Supervision::HEALTH_STATUS_BAD &&
|
|
transisted.status == Supervision::HEALTH_STATUS_FAILED) {
|
|
if (!snapshot.has(failedServerPath)) {
|
|
envelope = std::make_shared<VPackBuilder>();
|
|
FailedServer(snapshot, agent, std::to_string(jobId), "supervision", serverID)
|
|
.create(envelope);
|
|
}
|
|
}
|
|
}
|
|
|
|
void handleOnStatusCoordinator(Agent* agent, Node const& snapshot, HealthRecord& persisted,
|
|
HealthRecord& transisted, std::string const& serverID) {
|
|
|
|
if (transisted.status == Supervision::HEALTH_STATUS_FAILED) {
|
|
// if the current foxxmaster server failed => reset the value to ""
|
|
if (snapshot.hasAsString(foxxmaster).first == serverID) {
|
|
VPackBuilder create;
|
|
{
|
|
VPackArrayBuilder tx(&create);
|
|
{
|
|
VPackObjectBuilder d(&create);
|
|
create.add(foxxmaster, VPackValue(""));
|
|
}
|
|
}
|
|
singleWriteTransaction(agent, create);
|
|
}
|
|
}
|
|
}
|
|
|
|
void handleOnStatusSingle(Agent* agent, Node const& snapshot, HealthRecord& persisted,
|
|
HealthRecord& transisted, std::string const& serverID,
|
|
uint64_t const& jobId, std::shared_ptr<VPackBuilder>& envelope) {
|
|
std::string failedServerPath = failedServersPrefix + "/" + serverID;
|
|
// New condition GOOD:
|
|
if (transisted.status == Supervision::HEALTH_STATUS_GOOD) {
|
|
if (snapshot.has(failedServerPath)) {
|
|
envelope = std::make_shared<VPackBuilder>();
|
|
{
|
|
VPackArrayBuilder a(envelope.get());
|
|
{
|
|
VPackObjectBuilder operations(envelope.get());
|
|
envelope->add(VPackValue(failedServerPath));
|
|
{
|
|
VPackObjectBuilder ccc(envelope.get());
|
|
envelope->add("op", VPackValue("delete"));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
} else if ( // New state: FAILED persisted: GOOD (-> BAD)
|
|
persisted.status == Supervision::HEALTH_STATUS_GOOD &&
|
|
transisted.status != Supervision::HEALTH_STATUS_GOOD) {
|
|
transisted.status = Supervision::HEALTH_STATUS_BAD;
|
|
} else if ( // New state: FAILED persisted: BAD (-> Job)
|
|
persisted.status == Supervision::HEALTH_STATUS_BAD &&
|
|
transisted.status == Supervision::HEALTH_STATUS_FAILED) {
|
|
if (!snapshot.has(failedServerPath)) {
|
|
envelope = std::make_shared<VPackBuilder>();
|
|
ActiveFailoverJob(snapshot, agent, std::to_string(jobId), "supervision", serverID)
|
|
.create(envelope);
|
|
}
|
|
}
|
|
}
|
|
|
|
|
|
void handleOnStatus(Agent* agent, Node const& snapshot, HealthRecord& persisted,
|
|
HealthRecord& transisted, std::string const& serverID,
|
|
uint64_t const& jobId, std::shared_ptr<VPackBuilder>& envelope) {
|
|
if (serverID.compare(0, 4, "PRMR") == 0) {
|
|
handleOnStatusDBServer(agent, snapshot, persisted, transisted, serverID, jobId, envelope);
|
|
} else if (serverID.compare(0, 4, "CRDN") == 0) {
|
|
handleOnStatusCoordinator(agent, snapshot, persisted, transisted, serverID);
|
|
} else if (serverID.compare(0, 4, "SNGL") == 0) {
|
|
handleOnStatusSingle(agent, snapshot, persisted, transisted, serverID, jobId, envelope);
|
|
} else {
|
|
LOG_TOPIC(ERR, Logger::SUPERVISION)
|
|
<< "Unknown server type. No supervision action taken. " << serverID;
|
|
}
|
|
}
|
|
|
|
|
|
// Build transaction for removing unattended servers from health monitoring
|
|
query_t arangodb::consensus::removeTransactionBuilder(std::vector<std::string> const& todelete) {
|
|
query_t del = std::make_shared<Builder>();
|
|
{
|
|
VPackArrayBuilder trxs(del.get());
|
|
{
|
|
VPackArrayBuilder trx(del.get());
|
|
{
|
|
VPackObjectBuilder server(del.get());
|
|
for (auto const& srv : todelete) {
|
|
del->add(VPackValue(Supervision::agencyPrefix() +
|
|
arangodb::consensus::healthPrefix + srv));
|
|
{
|
|
VPackObjectBuilder oper(del.get());
|
|
del->add("op", VPackValue("delete"));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
return del;
|
|
}
|
|
|
|
// Check all DB servers, guarded above doChecks
|
|
std::vector<check_t> Supervision::check(std::string const& type) {
|
|
// Dead lock detection
|
|
_lock.assertLockedByCurrentThread();
|
|
|
|
// Book keeping
|
|
std::vector<check_t> ret;
|
|
auto const& machinesPlanned =
|
|
_snapshot.hasAsChildren(std::string("Plan/") + type).first;
|
|
auto const& serversRegistered =
|
|
_snapshot.hasAsNode(currentServersRegisteredPrefix).first;
|
|
std::vector<std::string> todelete;
|
|
for (auto const& machine : _snapshot.hasAsChildren(healthPrefix).first) {
|
|
if ((type == "DBServers" && machine.first.compare(0, 4, "PRMR") == 0) ||
|
|
(type == "Coordinators" && machine.first.compare(0, 4, "CRDN") == 0) ||
|
|
(type == "Singles" && machine.first.compare(0, 4, "SNGL") == 0)) {
|
|
todelete.push_back(machine.first);
|
|
}
|
|
}
|
|
|
|
// Remove all machines, which are no longer planned
|
|
for (auto const& machine : machinesPlanned) {
|
|
todelete.erase(std::remove(todelete.begin(), todelete.end(), machine.first),
|
|
todelete.end());
|
|
}
|
|
if (!todelete.empty()) {
|
|
_agent->write(removeTransactionBuilder(todelete));
|
|
}
|
|
|
|
// Do actual monitoring
|
|
for (auto const& machine : machinesPlanned) {
|
|
std::string lastHeartbeatStatus, lastHeartbeatAcked, lastHeartbeatTime,
|
|
lastStatus, serverID(machine.first), shortName;
|
|
|
|
// short name arrives asynchronous to machine registering, make sure
|
|
// it has arrived before trying to use it
|
|
auto tmp_shortName =
|
|
_snapshot.hasAsString(targetShortID + serverID + "/ShortName");
|
|
if (tmp_shortName.second) {
|
|
shortName = tmp_shortName.first;
|
|
|
|
// "/arango/Current/ServersRegistered/<server-id>/endpoint"
|
|
std::string endpoint;
|
|
std::string epPath = serverID + "/endpoint";
|
|
if (serversRegistered.has(epPath)) {
|
|
endpoint = serversRegistered.hasAsString(epPath).first;
|
|
}
|
|
// "/arango/Current/ServersRegistered/<server-id>/host"
|
|
std::string hostId;
|
|
std::string hoPath = serverID + "/host";
|
|
if (serversRegistered.has(hoPath)) {
|
|
hostId = serversRegistered.hasAsString(hoPath).first;
|
|
}
|
|
// "/arango/Current/ServersRegistered/<server-id>/serverVersion"
|
|
std::string serverVersion;
|
|
std::string svPath = serverID + "/versionString";
|
|
if (serversRegistered.has(svPath)) {
|
|
serverVersion = serversRegistered.hasAsString(svPath).first;
|
|
}
|
|
// "/arango/Current/ServersRegistered/<server-id>/engine"
|
|
std::string engine;
|
|
std::string enPath = serverID + "/engine";
|
|
if (serversRegistered.has(enPath)) {
|
|
engine = serversRegistered.hasAsString(enPath).first;
|
|
}
|
|
//"/arango/Current/<serverId>/externalEndpoint"
|
|
std::string externalEndpoint;
|
|
std::string extEndPath = serverID + "/advertisedEndpoint";
|
|
if (serversRegistered.has(extEndPath)) {
|
|
externalEndpoint = serversRegistered.hasAsString(extEndPath).first;
|
|
}
|
|
|
|
// Health records from persistence, from transience and a new one
|
|
HealthRecord transist(shortName, endpoint, hostId, engine, serverVersion, externalEndpoint);
|
|
HealthRecord persist(shortName, endpoint, hostId, engine, serverVersion, externalEndpoint);
|
|
|
|
// Get last health entries from transient and persistent key value stores
|
|
if (_transient.has(healthPrefix + serverID)) {
|
|
transist = _transient.hasAsNode(healthPrefix + serverID).first;
|
|
}
|
|
if (_snapshot.has(healthPrefix + serverID)) {
|
|
persist = _snapshot.hasAsNode(healthPrefix + serverID).first;
|
|
}
|
|
|
|
// New health record (start with old add current information from sync)
|
|
// Sync.time is copied to Health.syncTime
|
|
// Sync.status is copied to Health.syncStatus
|
|
std::string syncTime =
|
|
_transient.has(syncPrefix + serverID)
|
|
? _transient.hasAsString(syncPrefix + serverID + "/time").first
|
|
: timepointToString(std::chrono::system_clock::time_point());
|
|
std::string syncStatus =
|
|
_transient.has(syncPrefix + serverID)
|
|
? _transient.hasAsString(syncPrefix + serverID + "/status").first
|
|
: "UNKNOWN";
|
|
|
|
// Last change registered in sync (transient != sync)
|
|
// Either now or value in transient
|
|
auto lastAckedTime = (syncTime != transist.syncTime)
|
|
? std::chrono::system_clock::now()
|
|
: stringToTimepoint(transist.lastAcked);
|
|
transist.lastAcked = timepointToString(lastAckedTime);
|
|
transist.syncTime = syncTime;
|
|
transist.syncStatus = syncStatus;
|
|
|
|
// update volatile values that may change
|
|
transist.advertisedEndpoint = externalEndpoint;
|
|
transist.serverVersion = serverVersion;
|
|
transist.engine = engine;
|
|
transist.hostId = hostId;
|
|
transist.endpoint = endpoint;
|
|
|
|
// Calculate elapsed since lastAcked
|
|
auto elapsed =
|
|
std::chrono::duration<double>(std::chrono::system_clock::now() - lastAckedTime);
|
|
|
|
if (elapsed.count() <= _okThreshold) {
|
|
transist.status = Supervision::HEALTH_STATUS_GOOD;
|
|
} else if (elapsed.count() <= _gracePeriod) {
|
|
transist.status = Supervision::HEALTH_STATUS_BAD;
|
|
} else {
|
|
transist.status = Supervision::HEALTH_STATUS_FAILED;
|
|
}
|
|
|
|
// Status changed?
|
|
bool changed = transist.statusDiff(persist);
|
|
|
|
// Take necessary actions if any
|
|
std::shared_ptr<VPackBuilder> envelope;
|
|
if (changed) {
|
|
handleOnStatus(_agent, _snapshot, persist, transist, serverID, _jobId, envelope);
|
|
}
|
|
|
|
persist = transist; // Now copy Status, SyncStatus from transient to persited
|
|
|
|
// Transient report
|
|
std::shared_ptr<Builder> tReport = std::make_shared<Builder>();
|
|
{
|
|
VPackArrayBuilder transaction(tReport.get()); // Transist Transaction
|
|
std::shared_ptr<VPackBuilder> envelope;
|
|
{
|
|
VPackObjectBuilder operation(tReport.get()); // Operation
|
|
tReport->add(VPackValue(healthPrefix + serverID)); // Supervision/Health
|
|
{
|
|
VPackObjectBuilder oo(tReport.get());
|
|
transist.toVelocyPack(*tReport);
|
|
}
|
|
}
|
|
} // Transaction
|
|
|
|
// Persistent report
|
|
std::shared_ptr<Builder> pReport = nullptr;
|
|
if (changed) {
|
|
pReport = std::make_shared<Builder>();
|
|
{
|
|
VPackArrayBuilder transaction(pReport.get()); // Persist Transaction
|
|
{
|
|
VPackObjectBuilder operation(pReport.get()); // Operation
|
|
pReport->add(VPackValue(healthPrefix + serverID)); // Supervision/Health
|
|
{
|
|
VPackObjectBuilder oo(pReport.get());
|
|
persist.toVelocyPack(*pReport);
|
|
}
|
|
if (envelope != nullptr) { // Failed server
|
|
TRI_ASSERT(envelope->slice().isArray() && envelope->slice()[0].isObject());
|
|
for (VPackObjectIterator::ObjectPair i :
|
|
VPackObjectIterator(envelope->slice()[0])) {
|
|
pReport->add(i.key.copyString(), i.value);
|
|
}
|
|
}
|
|
} // Operation
|
|
if (envelope != nullptr && envelope->slice().length() > 1) { // Preconditions(Job)
|
|
TRI_ASSERT(envelope->slice().isArray() && envelope->slice()[1].isObject());
|
|
pReport->add(envelope->slice()[1]);
|
|
}
|
|
} // Transaction
|
|
}
|
|
|
|
if (!this->isStopping()) {
|
|
// Replicate special event and only then transient store
|
|
if (changed) {
|
|
write_ret_t res = singleWriteTransaction(_agent, *pReport);
|
|
if (res.accepted && res.indices.front() != 0) {
|
|
++_jobId; // Job was booked
|
|
transient(_agent, *tReport);
|
|
}
|
|
} else { // Nothing special just transient store
|
|
transient(_agent, *tReport);
|
|
}
|
|
}
|
|
} else {
|
|
LOG_TOPIC(INFO, Logger::SUPERVISION)
|
|
<< "Short name for << " << serverID
|
|
<< " not yet available. Skipping health check.";
|
|
} // else
|
|
|
|
} // for
|
|
|
|
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();
|
|
|
|
if (_agent == nullptr || this->isStopping()) {
|
|
return false;
|
|
}
|
|
|
|
_agent->executeLockedRead([&]() {
|
|
if (_agent->readDB().has(_agencyPrefix)) {
|
|
_snapshot = _agent->readDB().get(_agencyPrefix);
|
|
}
|
|
if (_agent->transient().has(_agencyPrefix)) {
|
|
_transient = _agent->transient().get(_agencyPrefix);
|
|
}
|
|
});
|
|
|
|
return true;
|
|
}
|
|
|
|
// All checks, guarded by main thread
|
|
bool Supervision::doChecks() {
|
|
_lock.assertLockedByCurrentThread();
|
|
TRI_ASSERT(ServerState::roleToAgencyListKey(ServerState::ROLE_PRIMARY) ==
|
|
"DBServers");
|
|
check(ServerState::roleToAgencyListKey(ServerState::ROLE_PRIMARY));
|
|
TRI_ASSERT(ServerState::roleToAgencyListKey(ServerState::ROLE_COORDINATOR) ==
|
|
"Coordinators");
|
|
check(ServerState::roleToAgencyListKey(ServerState::ROLE_COORDINATOR));
|
|
TRI_ASSERT(ServerState::roleToAgencyListKey(ServerState::ROLE_SINGLE) ==
|
|
"Singles");
|
|
check(ServerState::roleToAgencyListKey(ServerState::ROLE_SINGLE));
|
|
|
|
return true;
|
|
}
|
|
|
|
void Supervision::reportStatus(std::string const& status) {
|
|
bool persist = false;
|
|
query_t report;
|
|
|
|
{ // Do I have to report to agency under
|
|
_lock.assertLockedByCurrentThread();
|
|
if (_snapshot.hasAsString("/Supervision/State/Mode").first != status) {
|
|
// This includes the case that the mode is not set, since status
|
|
// is never empty
|
|
persist = true;
|
|
}
|
|
}
|
|
|
|
report = std::make_shared<VPackBuilder>();
|
|
{
|
|
VPackArrayBuilder trx(report.get());
|
|
{
|
|
VPackObjectBuilder br(report.get());
|
|
report->add(VPackValue("/Supervision/State"));
|
|
{
|
|
VPackObjectBuilder bbr(report.get());
|
|
report->add("Mode", VPackValue(status));
|
|
report->add("Timestamp",
|
|
VPackValue(timepointToString(std::chrono::system_clock::now())));
|
|
}
|
|
}
|
|
}
|
|
|
|
// Importatnt! No reporting in transient for Maintenance mode.
|
|
if (status != "Maintenance") {
|
|
transient(_agent, *report);
|
|
}
|
|
|
|
if (persist) {
|
|
write_ret_t res = singleWriteTransaction(_agent, *report);
|
|
}
|
|
}
|
|
|
|
void Supervision::run() {
|
|
// First wait until somebody has initialized the ArangoDB data, before
|
|
// that running the supervision does not make sense and will indeed
|
|
// lead to horrible errors:
|
|
|
|
std::string const supervisionNode = _agencyPrefix + supervisionPrefix;
|
|
|
|
while (!this->isStopping()) {
|
|
{
|
|
CONDITION_LOCKER(guard, _cv);
|
|
_cv.wait(static_cast<uint64_t>(1000000 * _frequency));
|
|
}
|
|
|
|
bool done = false;
|
|
MUTEX_LOCKER(locker, _lock);
|
|
_agent->executeLockedRead([&]() {
|
|
if (_agent->readDB().has(supervisionNode)) {
|
|
try {
|
|
_snapshot = _agent->readDB().get(supervisionNode);
|
|
if (_snapshot.children().size() > 0) {
|
|
done = true;
|
|
}
|
|
} catch (...) {
|
|
LOG_TOPIC(WARN, Logger::SUPERVISION)
|
|
<< "Main node in agency gone. Contact your db administrator.";
|
|
}
|
|
}
|
|
});
|
|
|
|
if (done) {
|
|
break;
|
|
}
|
|
|
|
LOG_TOPIC(DEBUG, Logger::SUPERVISION) << "Waiting for ArangoDB to "
|
|
"initialize its data.";
|
|
}
|
|
|
|
bool shutdown = false;
|
|
{
|
|
CONDITION_LOCKER(guard, _cv);
|
|
TRI_ASSERT(_agent != nullptr);
|
|
|
|
while (!this->isStopping()) {
|
|
|
|
auto lapStart = std::chrono::steady_clock::now();
|
|
|
|
{
|
|
MUTEX_LOCKER(locker, _lock);
|
|
|
|
if (isShuttingDown()) {
|
|
handleShutdown();
|
|
} else if (_selfShutdown) {
|
|
shutdown = true;
|
|
break;
|
|
}
|
|
|
|
// Only modifiy this condition with extreme care:
|
|
// Supervision needs to wait until the agent has finished leadership
|
|
// preparation or else the local agency snapshot might be behind its
|
|
// last state.
|
|
if (_agent->leading() && _agent->getPrepareLeadership() == 0) {
|
|
if (_jobId == 0 || _jobId == _jobIdMax) {
|
|
getUniqueIds(); // cannot fail but only hang
|
|
}
|
|
|
|
updateSnapshot();
|
|
|
|
if (!_snapshot.has("Supervision/Maintenance")) {
|
|
reportStatus("Normal");
|
|
|
|
if (!_upgraded) {
|
|
upgradeAgency();
|
|
}
|
|
|
|
_haveAborts = false;
|
|
|
|
if (_agent->leaderFor() > 55 || earlyBird()) {
|
|
// 55 seconds is less than a minute, which fits to the
|
|
// 60 seconds timeout in /_admin/cluster/health
|
|
try {
|
|
doChecks();
|
|
} catch (std::exception const& e) {
|
|
LOG_TOPIC(ERR, Logger::SUPERVISION)
|
|
<< e.what() << " " << __FILE__ << " " << __LINE__;
|
|
} catch (...) {
|
|
LOG_TOPIC(ERR, Logger::SUPERVISION)
|
|
<< "Supervision::doChecks() generated an uncaught "
|
|
"exception.";
|
|
}
|
|
} else {
|
|
LOG_TOPIC(INFO, Logger::SUPERVISION)
|
|
<< "Postponing supervision for now, waiting for incoming "
|
|
"heartbeats: "
|
|
<< _agent->leaderFor();
|
|
}
|
|
|
|
handleJobs();
|
|
|
|
} else {
|
|
reportStatus("Maintenance");
|
|
}
|
|
}
|
|
}
|
|
|
|
// If anything was rafted, we need to
|
|
index_t leaderIndex = _agent->index();
|
|
|
|
if (leaderIndex != 0) {
|
|
while (true) { // No point in progressing, if indexes cannot be advanced
|
|
|
|
// avoid getting trapped as last agent standing
|
|
if (this->isStopping()) {
|
|
break;
|
|
}
|
|
|
|
auto result = _agent->waitFor(leaderIndex);
|
|
if (result == Agent::raft_commit_t::TIMEOUT) { // Oh snap
|
|
// Note that we can get UNKNOWN if we have lost leadership or
|
|
// if we are shutting down. In both cases we just leave the loop.
|
|
LOG_TOPIC(WARN, Logger::SUPERVISION) << "Waiting for commits to be done ... ";
|
|
continue;
|
|
} else { // Good we can continue
|
|
break;
|
|
}
|
|
|
|
}
|
|
}
|
|
|
|
auto lapTime = std::chrono::duration_cast<std::chrono::microseconds>(
|
|
std::chrono::steady_clock::now() - lapStart).count();
|
|
|
|
if (lapTime < 1000000) {
|
|
_cv.wait(static_cast<uint64_t>((1000000 - lapTime) * _frequency));
|
|
}
|
|
}
|
|
}
|
|
|
|
if (shutdown) {
|
|
ApplicationServer::server->beginShutdown();
|
|
}
|
|
}
|
|
|
|
// Guarded by caller
|
|
bool Supervision::isShuttingDown() {
|
|
_lock.assertLockedByCurrentThread();
|
|
if (_snapshot.has("Shutdown")) {
|
|
return _snapshot.hasAsBool("Shutdown").first;
|
|
}
|
|
return false;
|
|
}
|
|
|
|
// Guarded by caller
|
|
std::string Supervision::serverHealth(std::string const& serverName) {
|
|
_lock.assertLockedByCurrentThread();
|
|
std::string const serverStatus(healthPrefix + serverName + "/Status");
|
|
return (_snapshot.has(serverStatus)) ? _snapshot.hasAsString(serverStatus).first
|
|
: std::string();
|
|
}
|
|
|
|
// Guarded by caller
|
|
void Supervision::handleShutdown() {
|
|
_lock.assertLockedByCurrentThread();
|
|
|
|
_selfShutdown = true;
|
|
LOG_TOPIC(DEBUG, Logger::SUPERVISION) << "Waiting for clients to shut down";
|
|
auto const& serversRegistered =
|
|
_snapshot.hasAsChildren(currentServersRegisteredPrefix).first;
|
|
bool serversCleared = true;
|
|
for (auto const& server : serversRegistered) {
|
|
if (server.first == "Version") {
|
|
continue;
|
|
}
|
|
|
|
LOG_TOPIC(DEBUG, Logger::SUPERVISION) << "Waiting for " << server.first << " to shutdown";
|
|
|
|
if (serverHealth(server.first) != HEALTH_STATUS_GOOD) {
|
|
LOG_TOPIC(WARN, Logger::SUPERVISION)
|
|
<< "Server " << server.first << " did not shutdown properly it seems!";
|
|
continue;
|
|
}
|
|
serversCleared = false;
|
|
}
|
|
|
|
if (serversCleared) {
|
|
if (_agent->leading()) {
|
|
auto del = std::make_shared<Builder>();
|
|
{
|
|
VPackArrayBuilder txs(del.get());
|
|
{
|
|
VPackArrayBuilder tx(del.get());
|
|
{
|
|
VPackObjectBuilder o(del.get());
|
|
del->add(VPackValue(_agencyPrefix + "/Shutdown"));
|
|
{
|
|
VPackObjectBuilder oo(del.get());
|
|
del->add("op", VPackValue("delete"));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
auto result = _agent->write(del);
|
|
if (result.indices.size() != 1) {
|
|
LOG_TOPIC(ERR, Logger::SUPERVISION)
|
|
<< "Invalid resultsize of " << result.indices.size()
|
|
<< " found during shutdown";
|
|
} else {
|
|
if (!_agent->waitFor(result.indices.at(0))) {
|
|
LOG_TOPIC(ERR, Logger::SUPERVISION)
|
|
<< "Result was not written to followers during shutdown";
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
void Supervision::cleanupLostCollections(Node const& snapshot, AgentInterface* agent,
|
|
std::string const& jobId) {
|
|
std::unordered_set<std::string> failedServers;
|
|
|
|
// Search for failed server
|
|
// Could also use `Target/FailedServers`
|
|
auto const& health = snapshot.hasAsChildren(healthPrefix);
|
|
if (!health.second) {
|
|
return;
|
|
}
|
|
|
|
for (auto const& server : health.first) {
|
|
HealthRecord record(*server.second.get());
|
|
|
|
if (record.status == Supervision::HEALTH_STATUS_FAILED) {
|
|
failedServers.insert(server.first);
|
|
}
|
|
}
|
|
|
|
if (failedServers.size() == 0) {
|
|
return;
|
|
}
|
|
|
|
// Now iterate over all shards and look for failed leaders.
|
|
auto const& collections = snapshot.hasAsChildren("/Current/Collections");
|
|
if (!collections.second) {
|
|
return;
|
|
}
|
|
|
|
auto builder = std::make_shared<VPackBuilder>();
|
|
{
|
|
VPackArrayBuilder trxs(builder.get());
|
|
|
|
for (auto const& database : collections.first) {
|
|
auto const& dbname = database.first;
|
|
|
|
auto const& collections = database.second->children();
|
|
|
|
for (auto const& collection : collections) {
|
|
auto const& colname = collection.first;
|
|
|
|
for (auto const& shard : collection.second->children()) {
|
|
auto const& servers = shard.second->hasAsArray("servers").first;
|
|
|
|
TRI_ASSERT(servers.isArray());
|
|
|
|
if (servers.length() >= 1) {
|
|
TRI_ASSERT(servers[0].isString());
|
|
auto const& servername = servers[0].copyString();
|
|
|
|
if (failedServers.find(servername) != failedServers.end()) {
|
|
// lost shard
|
|
LOG_TOPIC(TRACE, Logger::SUPERVISION)
|
|
<< "Found a lost shard: " << shard.first;
|
|
auto const& shardname = shard.first;
|
|
|
|
auto const& planurl = "/arango/Plan/Collections/" + dbname + "/" +
|
|
colname + "/shards/" + shardname;
|
|
auto const& currenturl = "/arango/Current/Collections/" + dbname +
|
|
"/" + colname + "/" + shardname;
|
|
auto const& healthurl =
|
|
"/arango/Supervision/Health/" + servername + "/Status";
|
|
// check if it exists in Plan
|
|
if (snapshot.has(planurl)) {
|
|
continue;
|
|
}
|
|
// Now remove that shard
|
|
{
|
|
VPackArrayBuilder trx(builder.get());
|
|
{
|
|
VPackObjectBuilder update(builder.get());
|
|
// remove the shard in current
|
|
builder->add(VPackValue(currenturl));
|
|
{
|
|
VPackObjectBuilder op(builder.get());
|
|
builder->add("op", VPackValue("delete"));
|
|
}
|
|
// add a job done entry to "Target/Finished"
|
|
builder->add(VPackValue("/arango/Target/Finished"));
|
|
{
|
|
VPackObjectBuilder op(builder.get());
|
|
builder->add("op", VPackValue("push"));
|
|
builder->add(VPackValue("new"));
|
|
{
|
|
VPackObjectBuilder job(builder.get());
|
|
builder->add("type", VPackValue("cleanUpLostCollection"));
|
|
builder->add("server", VPackValue(shardname));
|
|
builder->add("jobId", VPackValue(jobId));
|
|
builder->add("creator", VPackValue("supervision"));
|
|
builder->add("timeCreated", VPackValue(timepointToString(
|
|
std::chrono::system_clock::now())));
|
|
}
|
|
}
|
|
// increase current version
|
|
builder->add(VPackValue("/arango/Current/Version"));
|
|
{
|
|
VPackObjectBuilder op(builder.get());
|
|
builder->add("op", VPackValue("increment"));
|
|
}
|
|
}
|
|
{
|
|
VPackObjectBuilder precon(builder.get());
|
|
// pre condition:
|
|
// still in current
|
|
// not in plan
|
|
// still failed
|
|
builder->add(VPackValue(planurl));
|
|
{
|
|
VPackObjectBuilder cond(builder.get());
|
|
builder->add("oldEmpty", VPackValue(true));
|
|
}
|
|
builder->add(VPackValue(currenturl));
|
|
{
|
|
VPackObjectBuilder cond(builder.get());
|
|
builder->add("oldEmpty", VPackValue(false));
|
|
}
|
|
builder->add(VPackValue(healthurl));
|
|
{
|
|
VPackObjectBuilder cond(builder.get());
|
|
builder->add("old", VPackValue(Supervision::HEALTH_STATUS_FAILED));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
auto const& trx = builder->slice();
|
|
|
|
if (trx.length() > 0) {
|
|
// do it! fire and forget!
|
|
agent->write(builder);
|
|
}
|
|
}
|
|
|
|
// Guarded by caller
|
|
bool Supervision::handleJobs() {
|
|
_lock.assertLockedByCurrentThread();
|
|
// Do supervision
|
|
|
|
shrinkCluster();
|
|
enforceReplication();
|
|
cleanupLostCollections(_snapshot, _agent, std::to_string(_jobId++));
|
|
readyOrphanedIndexCreations();
|
|
|
|
workJobs();
|
|
|
|
return true;
|
|
}
|
|
|
|
// Guarded by caller
|
|
void Supervision::workJobs() {
|
|
_lock.assertLockedByCurrentThread();
|
|
|
|
bool dummy = false;
|
|
auto todos = _snapshot.hasAsChildren(toDoPrefix).first;
|
|
auto it = todos.begin();
|
|
static std::string const FAILED = "failed";
|
|
|
|
while (it != todos.end()) {
|
|
auto jobNode = *(it->second);
|
|
if (jobNode.hasAsString("type").first.compare(0, FAILED.length(), FAILED) == 0) {
|
|
JobContext(TODO, jobNode.hasAsString("jobId").first, _snapshot, _agent)
|
|
.run(_haveAborts);
|
|
it = todos.erase(it);
|
|
} else {
|
|
++it;
|
|
}
|
|
}
|
|
|
|
// Do not start other jobs, if above resilience jobs aborted stuff
|
|
if (!_haveAborts) {
|
|
for (auto const& todoEnt : todos) {
|
|
auto jobNode = *(todoEnt.second);
|
|
JobContext(TODO, jobNode.hasAsString("jobId").first, _snapshot, _agent)
|
|
.run(dummy);
|
|
}
|
|
}
|
|
|
|
auto pends = _snapshot.hasAsChildren(pendingPrefix).first;
|
|
for (auto const& pendEnt : pends) {
|
|
auto jobNode = *(pendEnt.second);
|
|
JobContext(PENDING, jobNode.hasAsString("jobId").first, _snapshot, _agent)
|
|
.run(dummy);
|
|
}
|
|
|
|
}
|
|
|
|
void Supervision::readyOrphanedIndexCreations() {
|
|
_lock.assertLockedByCurrentThread();
|
|
|
|
if (_snapshot.has(planColPrefix) && _snapshot.has(curColPrefix)) {
|
|
auto const& plannedDBs = _snapshot(planColPrefix).children();
|
|
auto const& currentDBs = _snapshot(curColPrefix);
|
|
|
|
for (auto const& db : plannedDBs) {
|
|
std::string const& dbname = db.first;
|
|
auto const& database = *(db.second);
|
|
auto const& plannedCols = database.children();
|
|
for (auto const& col : plannedCols) {
|
|
auto const& colname = col.first;
|
|
std::string const& colPath = dbname + "/" + colname + "/";
|
|
auto const& collection = *(col.second);
|
|
std::unordered_set<std::string> built;
|
|
Slice indexes;
|
|
if (collection.has("indexes")) {
|
|
indexes = collection("indexes").getArray();
|
|
if (indexes.length() > 0) {
|
|
for (auto const& planIndex : VPackArrayIterator(indexes)) {
|
|
if (planIndex.hasKey(StaticStrings::IndexIsBuilding) &&
|
|
collection.has("shards")) {
|
|
auto const& planId = planIndex.get("id");
|
|
auto const& shards = collection("shards");
|
|
if (collection.has("numberOfShards") &&
|
|
collection("numberOfShards").isUInt()) {
|
|
auto const& nshards = collection("numberOfShards").getUInt();
|
|
if (nshards == 0) {
|
|
continue;
|
|
}
|
|
size_t nIndexes = 0;
|
|
for (auto const& sh : shards.children()) {
|
|
auto const& shname = sh.first;
|
|
|
|
if (currentDBs.has(colPath + shname + "/indexes")) {
|
|
auto const& curIndexes =
|
|
currentDBs(colPath + shname + "/indexes").slice();
|
|
for (auto const& curIndex : VPackArrayIterator(curIndexes)) {
|
|
auto const& curId = curIndex.get("id");
|
|
if (planId == curId) {
|
|
++nIndexes;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
if (nIndexes == nshards) {
|
|
built.emplace(planId.copyString());
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// We have some indexes, that have been built and are isBuilding still
|
|
if (!built.empty()) {
|
|
auto envelope = std::make_shared<Builder>();
|
|
{
|
|
VPackArrayBuilder trxs(envelope.get());
|
|
{
|
|
VPackArrayBuilder trx(envelope.get());
|
|
{
|
|
VPackObjectBuilder operation(envelope.get());
|
|
envelope->add(VPackValue(_agencyPrefix + "/" + PLAN_VERSION));
|
|
{
|
|
VPackObjectBuilder o(envelope.get());
|
|
envelope->add("op", VPackValue("increment"));
|
|
}
|
|
envelope->add(VPackValue(_agencyPrefix + planColPrefix + colPath + "indexes"));
|
|
VPackArrayBuilder value(envelope.get());
|
|
for (auto const& planIndex : VPackArrayIterator(indexes)) {
|
|
if (built.find(planIndex.get("id").copyString()) != built.end()) {
|
|
{
|
|
VPackObjectBuilder props(envelope.get());
|
|
for (auto const& prop : VPackObjectIterator(planIndex)) {
|
|
auto const& key = prop.key.copyString();
|
|
if (key != StaticStrings::IndexIsBuilding) {
|
|
envelope->add(key, prop.value);
|
|
}
|
|
}
|
|
}
|
|
} else {
|
|
envelope->add(planIndex);
|
|
}
|
|
}
|
|
}
|
|
{
|
|
VPackObjectBuilder precondition(envelope.get());
|
|
envelope->add(VPackValue(_agencyPrefix + planColPrefix + colPath + "indexes"));
|
|
envelope->add(indexes);
|
|
}
|
|
}
|
|
}
|
|
|
|
write_ret_t res = _agent->write(envelope);
|
|
if (!res.successful()) {
|
|
LOG_TOPIC(DEBUG, Logger::SUPERVISION)
|
|
<< "failed to report ready index to agency. Will retry.";
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
void Supervision::enforceReplication() {
|
|
_lock.assertLockedByCurrentThread();
|
|
auto const& plannedDBs = _snapshot.hasAsChildren(planColPrefix).first;
|
|
|
|
for (const auto& db_ : plannedDBs) { // Planned databases
|
|
auto const& db = *(db_.second);
|
|
for (const auto& col_ : db.children()) { // Planned collections
|
|
auto const& col = *(col_.second);
|
|
|
|
size_t replicationFactor;
|
|
auto replFact = col.hasAsUInt("replicationFactor");
|
|
if (replFact.second) {
|
|
replicationFactor = replFact.first;
|
|
} else {
|
|
auto replFact2 = col.hasAsString("replicationFactor");
|
|
if (replFact2.second && replFact2.first == "satellite") {
|
|
// satellites => distribute to every server
|
|
auto available = Job::availableServers(_snapshot);
|
|
replicationFactor = Job::countGoodServersInList(_snapshot, available);
|
|
} else {
|
|
LOG_TOPIC(DEBUG, Logger::SUPERVISION)
|
|
<< "no replicationFactor entry in " << col.toJson();
|
|
continue;
|
|
}
|
|
}
|
|
|
|
bool clone = col.has("distributeShardsLike");
|
|
|
|
if (!clone) {
|
|
for (auto const& shard_ : col.hasAsChildren("shards").first) { // Pl shards
|
|
auto const& shard = *(shard_.second);
|
|
VPackBuilder onlyFollowers;
|
|
{
|
|
VPackArrayBuilder guard(&onlyFollowers);
|
|
bool first = true;
|
|
for (auto const& pp : VPackArrayIterator(shard.slice())) {
|
|
if (!first) {
|
|
onlyFollowers.add(pp);
|
|
}
|
|
first = false;
|
|
}
|
|
}
|
|
size_t actualReplicationFactor
|
|
= 1 + Job::countGoodServersInList(_snapshot, onlyFollowers.slice());
|
|
// leader plus GOOD followers
|
|
size_t apparentReplicationFactor = shard.slice().length();
|
|
|
|
if (actualReplicationFactor != replicationFactor ||
|
|
apparentReplicationFactor != replicationFactor) {
|
|
// Check that there is not yet an addFollower or removeFollower
|
|
// or moveShard job in ToDo for this shard:
|
|
auto const& todo = _snapshot.hasAsChildren(toDoPrefix).first;
|
|
bool found = false;
|
|
for (auto const& pair : todo) {
|
|
auto const& job = pair.second;
|
|
auto tmp_type = job->hasAsString("type");
|
|
auto tmp_shard = job->hasAsString("shard");
|
|
if ((tmp_type.first == "addFollower" || tmp_type.first == "removeFollower" ||
|
|
tmp_type.first == "moveShard") &&
|
|
tmp_shard.first == shard_.first) {
|
|
found = true;
|
|
LOG_TOPIC(DEBUG, Logger::SUPERVISION)
|
|
<< "already found "
|
|
"addFollower or removeFollower job in ToDo, not "
|
|
"scheduling "
|
|
"again for shard "
|
|
<< shard_.first;
|
|
break;
|
|
}
|
|
}
|
|
// Check that shard is not locked:
|
|
if (_snapshot.has(blockedShardsPrefix + shard_.first)) {
|
|
found = true;
|
|
}
|
|
if (!found) {
|
|
if (actualReplicationFactor < replicationFactor) {
|
|
AddFollower(_snapshot, _agent, std::to_string(_jobId++),
|
|
"supervision", db_.first, col_.first, shard_.first)
|
|
.create();
|
|
} else if (apparentReplicationFactor > replicationFactor &&
|
|
actualReplicationFactor >= replicationFactor) {
|
|
RemoveFollower(_snapshot, _agent, std::to_string(_jobId++),
|
|
"supervision", db_.first, col_.first, shard_.first)
|
|
.create();
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
void Supervision::fixPrototypeChain(Builder& migrate) {
|
|
_lock.assertLockedByCurrentThread();
|
|
|
|
auto const& snap = _snapshot;
|
|
|
|
std::function<std::string(std::string const&, std::string const&)> resolve;
|
|
resolve = [&](std::string const& db, std::string const& col) {
|
|
std::string s;
|
|
auto tmp_n = snap.hasAsNode(planColPrefix + db + "/" + col);
|
|
if (tmp_n.second) {
|
|
Node const& n = tmp_n.first;
|
|
s = n.hasAsString("distributeShardsLike").first;
|
|
}
|
|
return (s.empty()) ? col : resolve(db, s);
|
|
};
|
|
|
|
for (auto const& database : _snapshot.hasAsChildren(planColPrefix).first) {
|
|
for (auto const& collection : database.second->children()) {
|
|
if (collection.second->has("distributeShardsLike")) {
|
|
auto prototype =
|
|
(*collection.second).hasAsString("distributeShardsLike").first;
|
|
if (!prototype.empty()) {
|
|
std::string u;
|
|
try {
|
|
u = resolve(database.first, prototype);
|
|
} catch (...) {
|
|
}
|
|
if (u != prototype) {
|
|
{
|
|
VPackArrayBuilder trx(&migrate);
|
|
{
|
|
VPackObjectBuilder oper(&migrate);
|
|
migrate.add(planColPrefix + database.first + "/" +
|
|
collection.first + "/" + "distributeShardsLike",
|
|
VPackValue(u));
|
|
}
|
|
{
|
|
VPackObjectBuilder prec(&migrate);
|
|
migrate.add(planColPrefix + database.first + "/" +
|
|
collection.first + "/" + "distributeShardsLike",
|
|
VPackValue(prototype));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Shrink cluster if applicable, guarded by caller
|
|
void Supervision::shrinkCluster() {
|
|
_lock.assertLockedByCurrentThread();
|
|
|
|
auto const& todo = _snapshot.hasAsChildren(toDoPrefix).first;
|
|
auto const& pending = _snapshot.hasAsChildren(pendingPrefix).first;
|
|
|
|
if (!todo.empty() || !pending.empty()) { // This is low priority
|
|
return;
|
|
}
|
|
|
|
// 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");
|
|
|
|
if (_snapshot.hasAsUInt(NDBServers).second) {
|
|
targetNumDBServers = _snapshot.hasAsUInt(NDBServers).first;
|
|
} else {
|
|
LOG_TOPIC(TRACE, Logger::SUPERVISION)
|
|
<< "Targeted number of DB servers not set yet";
|
|
return;
|
|
}
|
|
|
|
// Only if number of servers in target is smaller than the available
|
|
if (targetNumDBServers < availServers.size()) {
|
|
// Minimum 1 DB server must remain
|
|
if (availServers.size() == 1) {
|
|
LOG_TOPIC(DEBUG, Logger::SUPERVISION)
|
|
<< "Only one db server left for operation";
|
|
return;
|
|
}
|
|
|
|
/**
|
|
* mop: TODO instead of using Plan/Collections we should watch out for
|
|
* Plan/ReplicationFactor and Current...when the replicationFactor is not
|
|
* fullfilled we should add a follower to the plan
|
|
* When seeing more servers in Current than replicationFactor we should
|
|
* remove a server.
|
|
* RemoveServer then should be changed so that it really just kills a server
|
|
* after a while...
|
|
* this way we would have implemented changing the replicationFactor and
|
|
* have an awesome new feature
|
|
**/
|
|
// Find greatest replication factor among all collections
|
|
uint64_t maxReplFact = 1;
|
|
auto const& databases = _snapshot.hasAsChildren(planColPrefix).first;
|
|
for (auto const& database : databases) {
|
|
for (auto const& collptr : database.second->children()) {
|
|
auto const& node = *collptr.second;
|
|
if (node.hasAsUInt("replicationFactor").second) {
|
|
auto replFact = node.hasAsUInt("replicationFactor").first;
|
|
if (replFact > maxReplFact) {
|
|
maxReplFact = replFact;
|
|
}
|
|
}
|
|
// Note that this could be a satellite collection, in any case, ignore:
|
|
}
|
|
}
|
|
|
|
// mop: do not account any failedservers in this calculation..the ones
|
|
// having
|
|
// a state of failed still have data of interest to us! We wait indefinately
|
|
// for them to recover or for the user to remove them
|
|
if (maxReplFact < availServers.size()) {
|
|
// Clean out as long as number of available servers is bigger
|
|
// than maxReplFactor and bigger than targeted number of db servers
|
|
if (availServers.size() > maxReplFact && availServers.size() > targetNumDBServers) {
|
|
// Sort servers by name
|
|
std::sort(availServers.begin(), availServers.end());
|
|
|
|
// Schedule last server for cleanout
|
|
bool dummy;
|
|
CleanOutServer(_snapshot, _agent, std::to_string(_jobId++),
|
|
"supervision", availServers.back())
|
|
.run(dummy);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Start thread
|
|
bool Supervision::start() {
|
|
Thread::start();
|
|
return true;
|
|
}
|
|
|
|
// Start thread with agent
|
|
bool Supervision::start(Agent* agent) {
|
|
_agent = agent;
|
|
_frequency = _agent->config().supervisionFrequency();
|
|
_gracePeriod = _agent->config().supervisionGracePeriod();
|
|
return start();
|
|
}
|
|
|
|
static std::string const syncLatest = "/Sync/LatestID";
|
|
|
|
void Supervision::getUniqueIds() {
|
|
_lock.assertLockedByCurrentThread();
|
|
|
|
size_t n = 10000;
|
|
|
|
std::string path = _agencyPrefix + "/Sync/LatestID";
|
|
auto builder = std::make_shared<Builder>();
|
|
{
|
|
VPackArrayBuilder a(builder.get());
|
|
{
|
|
VPackArrayBuilder b(builder.get());
|
|
{
|
|
VPackObjectBuilder c(builder.get());
|
|
{
|
|
builder->add(VPackValue(path));
|
|
VPackObjectBuilder b(builder.get());
|
|
builder->add("op", VPackValue("increment"));
|
|
builder->add("step", VPackValue(n));
|
|
}
|
|
}
|
|
}
|
|
{
|
|
VPackArrayBuilder a(builder.get());
|
|
builder->add(VPackValue(path));
|
|
}
|
|
} // [[{path:{"op":"increment","step":n}}],[path]]
|
|
|
|
auto ret = _agent->transact(builder);
|
|
if (ret.accepted) {
|
|
try {
|
|
_jobIdMax =
|
|
ret.result->slice()[1]
|
|
.get(std::vector<std::string>({"arango", "Sync", "LatestID"}))
|
|
.getUInt();
|
|
_jobId = _jobIdMax - n;
|
|
} catch (std::exception const& e) {
|
|
LOG_TOPIC(ERR, Logger::SUPERVISION)
|
|
<< "Failed to acquire job IDs from agency: " << e.what() << __FILE__
|
|
<< " " << __LINE__;
|
|
}
|
|
}
|
|
}
|
|
|
|
void Supervision::beginShutdown() {
|
|
// Personal hygiene
|
|
Thread::beginShutdown();
|
|
|
|
CONDITION_LOCKER(guard, _cv);
|
|
guard.broadcast();
|
|
}
|