mirror of https://gitee.com/bigwinds/arangodb
3179 lines
109 KiB
C++
3179 lines
109 KiB
C++
////////////////////////////////////////////////////////////////////////////////
|
|
/// DISCLAIMER
|
|
///
|
|
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
|
|
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
|
|
///
|
|
/// Licensed under the Apache License, Version 2.0 (the "License");
|
|
/// you may not use this file except in compliance with the License.
|
|
/// You may obtain a copy of the License at
|
|
///
|
|
/// http://www.apache.org/licenses/LICENSE-2.0
|
|
///
|
|
/// Unless required by applicable law or agreed to in writing, software
|
|
/// distributed under the License is distributed on an "AS IS" BASIS,
|
|
/// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
/// See the License for the specific language governing permissions and
|
|
/// limitations under the License.
|
|
///
|
|
/// Copyright holder is ArangoDB GmbH, Cologne, Germany
|
|
///
|
|
/// @author Jan Steemann
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
|
|
#include "RestReplicationHandler.h"
|
|
|
|
#include "Aql/Query.h"
|
|
#include "Aql/QueryRegistry.h"
|
|
#include "Basics/ConditionLocker.h"
|
|
#include "Basics/ReadLocker.h"
|
|
#include "Basics/Result.h"
|
|
#include "Basics/RocksDBUtils.h"
|
|
#include "Basics/VelocyPackHelper.h"
|
|
#include "Basics/WriteLocker.h"
|
|
#include "Cluster/ClusterComm.h"
|
|
#include "Cluster/ClusterFeature.h"
|
|
#include "Cluster/ClusterHelpers.h"
|
|
#include "Cluster/ClusterMethods.h"
|
|
#include "Cluster/FollowerInfo.h"
|
|
#include "Cluster/ResignShardLeadership.h"
|
|
#include "GeneralServer/AuthenticationFeature.h"
|
|
#include "Indexes/Index.h"
|
|
#include "Replication/DatabaseInitialSyncer.h"
|
|
#include "Replication/DatabaseReplicationApplier.h"
|
|
#include "Replication/GlobalInitialSyncer.h"
|
|
#include "Replication/GlobalReplicationApplier.h"
|
|
#include "Replication/ReplicationApplierConfiguration.h"
|
|
#include "Replication/ReplicationClients.h"
|
|
#include "Replication/ReplicationFeature.h"
|
|
#include "RestServer/DatabaseFeature.h"
|
|
#include "RestServer/QueryRegistryFeature.h"
|
|
#include "RestServer/ServerIdFeature.h"
|
|
#include "Sharding/ShardingInfo.h"
|
|
#include "StorageEngine/EngineSelectorFeature.h"
|
|
#include "StorageEngine/PhysicalCollection.h"
|
|
#include "StorageEngine/StorageEngine.h"
|
|
#include "Transaction/StandaloneContext.h"
|
|
#include "Utils/CollectionNameResolver.h"
|
|
#include "Utils/OperationOptions.h"
|
|
#include "Utils/SingleCollectionTransaction.h"
|
|
#include "VocBase/LogicalCollection.h"
|
|
#include "VocBase/LogicalView.h"
|
|
|
|
#include <velocypack/Builder.h>
|
|
#include <velocypack/Collection.h>
|
|
#include <velocypack/Iterator.h>
|
|
#include <velocypack/Parser.h>
|
|
#include <velocypack/Slice.h>
|
|
#include <velocypack/velocypack-aliases.h>
|
|
|
|
using namespace arangodb;
|
|
using namespace arangodb::basics;
|
|
using namespace arangodb::rest;
|
|
|
|
namespace {
|
|
std::string const dataString("data");
|
|
std::string const keyString("key");
|
|
std::string const typeString("type");
|
|
} // namespace
|
|
|
|
uint64_t const RestReplicationHandler::_defaultChunkSize = 128 * 1024;
|
|
uint64_t const RestReplicationHandler::_maxChunkSize = 128 * 1024 * 1024;
|
|
std::chrono::hours const RestReplicationHandler::_tombstoneTimeout =
|
|
std::chrono::hours(24);
|
|
|
|
basics::ReadWriteLock RestReplicationHandler::_tombLock;
|
|
std::unordered_map<std::string, std::chrono::time_point<std::chrono::steady_clock>>
|
|
RestReplicationHandler::_tombstones = {};
|
|
|
|
static aql::QueryId ExtractReadlockId(VPackSlice slice) {
|
|
TRI_ASSERT(slice.isString());
|
|
return StringUtils::uint64(slice.copyString());
|
|
}
|
|
|
|
static bool ignoreHiddenEnterpriseCollection(std::string const& name, bool force) {
|
|
#ifdef USE_ENTERPRISE
|
|
if (!force && name[0] == '_') {
|
|
if (strncmp(name.c_str(), "_local_", 7) == 0 ||
|
|
strncmp(name.c_str(), "_from_", 6) == 0 || strncmp(name.c_str(), "_to_", 4) == 0) {
|
|
LOG_TOPIC("944c4", WARN, arangodb::Logger::REPLICATION)
|
|
<< "Restore ignoring collection " << name
|
|
<< ". Will be created via SmartGraphs of a full dump. If you want to "
|
|
<< "restore ONLY this collection use 'arangorestore --force'. "
|
|
<< "However this is not recommended and you should instead restore "
|
|
<< "the EdgeCollection of the SmartGraph instead.";
|
|
return true;
|
|
}
|
|
}
|
|
#endif
|
|
return false;
|
|
}
|
|
|
|
static Result checkPlanLeaderDirect(std::shared_ptr<LogicalCollection> const& col,
|
|
std::string const& claimLeaderId) {
|
|
std::vector<std::string> agencyPath = {"Plan",
|
|
"Collections",
|
|
col->vocbase().name(),
|
|
std::to_string(col->planId()),
|
|
"shards",
|
|
col->name()};
|
|
|
|
std::string shardAgencyPathString = StringUtils::join(agencyPath, '/');
|
|
|
|
AgencyComm ac;
|
|
AgencyCommResult res = ac.getValues(shardAgencyPathString);
|
|
|
|
if (res.successful()) {
|
|
// This is bullshit. Why does the *fancy* AgencyComm Manager
|
|
// prepend the agency url with `arango` but in the end returns an object
|
|
// that is prepended by `arango`! WTF!?
|
|
VPackSlice plan = res.slice().at(0).get(AgencyCommManager::path()).get(agencyPath);
|
|
TRI_ASSERT(plan.isArray() && plan.length() > 0);
|
|
|
|
VPackSlice leaderSlice = plan.at(0);
|
|
TRI_ASSERT(leaderSlice.isString());
|
|
if (leaderSlice.isEqualString(claimLeaderId)) {
|
|
return Result{};
|
|
}
|
|
}
|
|
|
|
return Result{TRI_ERROR_FORBIDDEN};
|
|
}
|
|
|
|
static Result restoreDataParser(char const* ptr, char const* pos,
|
|
std::string const& collectionName, int line,
|
|
std::string& key, VPackBuilder& builder,
|
|
VPackSlice& doc, TRI_replication_operation_e& type) {
|
|
builder.clear();
|
|
|
|
try {
|
|
VPackParser parser(builder, builder.options);
|
|
parser.parse(ptr, static_cast<size_t>(pos - ptr));
|
|
} catch (std::exception const& ex) {
|
|
// Could not even build the string
|
|
return Result{TRI_ERROR_HTTP_CORRUPTED_JSON,
|
|
"received invalid JSON data for collection '" + collectionName +
|
|
"' on line " + std::to_string(line) + ": " + ex.what()};
|
|
} catch (...) {
|
|
return Result{TRI_ERROR_INTERNAL};
|
|
}
|
|
|
|
VPackSlice const slice = builder.slice();
|
|
|
|
if (!slice.isObject()) {
|
|
return Result{TRI_ERROR_HTTP_CORRUPTED_JSON,
|
|
"received invalid JSON data for collection '" +
|
|
collectionName + "' on line " + std::to_string(line) +
|
|
": data is no object"};
|
|
}
|
|
|
|
type = REPLICATION_INVALID;
|
|
|
|
for (auto const& pair : VPackObjectIterator(slice, true)) {
|
|
if (!pair.key.isString()) {
|
|
return Result{TRI_ERROR_HTTP_CORRUPTED_JSON,
|
|
"received invalid JSON data for collection '" +
|
|
collectionName + "' on line " + std::to_string(line) +
|
|
": got a non-string key"};
|
|
}
|
|
|
|
if (pair.key.isEqualString(::typeString)) {
|
|
if (pair.value.isNumber()) {
|
|
int v = pair.value.getNumericValue<int>();
|
|
if (v == 2301) {
|
|
// pre-3.0 type for edges
|
|
type = REPLICATION_MARKER_DOCUMENT;
|
|
} else {
|
|
type = static_cast<TRI_replication_operation_e>(v);
|
|
}
|
|
}
|
|
} else if (pair.key.isEqualString(::dataString)) {
|
|
if (pair.value.isObject()) {
|
|
doc = pair.value;
|
|
|
|
VPackSlice k = doc.get(StaticStrings::KeyString);
|
|
if (k.isString()) {
|
|
key = k.copyString();
|
|
}
|
|
}
|
|
} else if (pair.key.isEqualString(::keyString)) {
|
|
if (key.empty()) {
|
|
key = pair.value.copyString();
|
|
}
|
|
}
|
|
}
|
|
|
|
if (type == REPLICATION_MARKER_DOCUMENT && !doc.isObject()) {
|
|
return Result{
|
|
TRI_ERROR_HTTP_BAD_PARAMETER,
|
|
"got document marker without object contents for collection '" + collectionName +
|
|
"' on line " + std::to_string(line) + ": " + doc.toJson()};
|
|
}
|
|
|
|
if (key.empty()) {
|
|
return Result{TRI_ERROR_HTTP_BAD_PARAMETER,
|
|
"received invalid JSON data for collection '" + collectionName +
|
|
"' on line " + std::to_string(line) + ": empty key"};
|
|
}
|
|
|
|
return Result{TRI_ERROR_NO_ERROR};
|
|
}
|
|
|
|
RestReplicationHandler::RestReplicationHandler(GeneralRequest* request, GeneralResponse* response)
|
|
: RestVocbaseBaseHandler(request, response) {}
|
|
|
|
RestReplicationHandler::~RestReplicationHandler() {}
|
|
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
/// @brief creates an error if called on a coordinator server
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
|
|
bool RestReplicationHandler::isCoordinatorError() {
|
|
if (_vocbase.type() == TRI_VOCBASE_TYPE_COORDINATOR) {
|
|
generateError(rest::ResponseCode::NOT_IMPLEMENTED, TRI_ERROR_CLUSTER_UNSUPPORTED,
|
|
"replication API is not supported on a coordinator");
|
|
|
|
return true;
|
|
}
|
|
|
|
return false;
|
|
}
|
|
|
|
std::string const RestReplicationHandler::LoggerState = "logger-state";
|
|
std::string const RestReplicationHandler::LoggerTickRanges =
|
|
"logger-tick-ranges";
|
|
std::string const RestReplicationHandler::LoggerFirstTick = "logger-first-tick";
|
|
std::string const RestReplicationHandler::LoggerFollow = "logger-follow";
|
|
std::string const RestReplicationHandler::OpenTransactions =
|
|
"determine-open-transactions";
|
|
std::string const RestReplicationHandler::Batch = "batch";
|
|
std::string const RestReplicationHandler::Barrier = "barrier";
|
|
std::string const RestReplicationHandler::Inventory = "inventory";
|
|
std::string const RestReplicationHandler::Keys = "keys";
|
|
std::string const RestReplicationHandler::Dump = "dump";
|
|
std::string const RestReplicationHandler::RestoreCollection =
|
|
"restore-collection";
|
|
std::string const RestReplicationHandler::RestoreIndexes = "restore-indexes";
|
|
std::string const RestReplicationHandler::RestoreData = "restore-data";
|
|
std::string const RestReplicationHandler::RestoreView = "restore-view";
|
|
std::string const RestReplicationHandler::Sync = "sync";
|
|
std::string const RestReplicationHandler::MakeSlave = "make-slave";
|
|
std::string const RestReplicationHandler::ServerId = "server-id";
|
|
std::string const RestReplicationHandler::ApplierConfig = "applier-config";
|
|
std::string const RestReplicationHandler::ApplierStart = "applier-start";
|
|
std::string const RestReplicationHandler::ApplierStop = "applier-stop";
|
|
std::string const RestReplicationHandler::ApplierState = "applier-state";
|
|
std::string const RestReplicationHandler::ApplierStateAll = "applier-state-all";
|
|
std::string const RestReplicationHandler::ClusterInventory = "clusterInventory";
|
|
std::string const RestReplicationHandler::AddFollower = "addFollower";
|
|
std::string const RestReplicationHandler::RemoveFollower = "removeFollower";
|
|
std::string const RestReplicationHandler::SetTheLeader = "set-the-leader";
|
|
std::string const RestReplicationHandler::HoldReadLockCollection =
|
|
"holdReadLockCollection";
|
|
|
|
// main function that dispatches the different routes and commands
|
|
RestStatus RestReplicationHandler::execute() {
|
|
// extract the request type
|
|
auto const type = _request->requestType();
|
|
auto const& suffixes = _request->suffixes();
|
|
|
|
size_t const len = suffixes.size();
|
|
|
|
if (len >= 1) {
|
|
std::string const& command = suffixes[0];
|
|
|
|
if (command == LoggerState) {
|
|
if (type != rest::RequestType::GET) {
|
|
goto BAD_CALL;
|
|
}
|
|
handleCommandLoggerState();
|
|
} else if (command == LoggerTickRanges) {
|
|
if (type != rest::RequestType::GET) {
|
|
goto BAD_CALL;
|
|
}
|
|
if (isCoordinatorError()) {
|
|
return RestStatus::DONE;
|
|
}
|
|
handleCommandLoggerTickRanges();
|
|
} else if (command == LoggerFirstTick) {
|
|
if (type != rest::RequestType::GET) {
|
|
goto BAD_CALL;
|
|
}
|
|
if (isCoordinatorError()) {
|
|
return RestStatus::DONE;
|
|
}
|
|
handleCommandLoggerFirstTick();
|
|
} else if (command == LoggerFollow) {
|
|
if (type != rest::RequestType::GET && type != rest::RequestType::PUT) {
|
|
goto BAD_CALL;
|
|
}
|
|
if (isCoordinatorError()) {
|
|
return RestStatus::DONE;
|
|
}
|
|
// track the number of parallel invocations of the tailing API
|
|
auto* rf = application_features::ApplicationServer::getFeature<ReplicationFeature>(
|
|
"Replication");
|
|
// this may throw when too many threads are going into tailing
|
|
rf->trackTailingStart();
|
|
|
|
auto guard = scopeGuard([rf]() { rf->trackTailingEnd(); });
|
|
|
|
handleCommandLoggerFollow();
|
|
} else if (command == OpenTransactions) {
|
|
if (type != rest::RequestType::GET) {
|
|
goto BAD_CALL;
|
|
}
|
|
handleCommandDetermineOpenTransactions();
|
|
} else if (command == Batch) {
|
|
// access batch context in context manager
|
|
// example call: curl -XPOST --dump - --data '{}'
|
|
// http://localhost:5555/_api/replication/batch
|
|
// the object may contain a "ttl" for the context
|
|
|
|
// POST - create batch id / handle
|
|
// PUT - extend batch id / handle
|
|
// DEL - delete batchid
|
|
|
|
if (ServerState::instance()->isCoordinator()) {
|
|
handleTrampolineCoordinator();
|
|
} else {
|
|
handleCommandBatch();
|
|
}
|
|
} else if (command == Barrier) {
|
|
if (isCoordinatorError()) {
|
|
return RestStatus::DONE;
|
|
}
|
|
handleCommandBarrier();
|
|
} else if (command == Inventory) {
|
|
// get overview of collections and indexes followed by some extra data
|
|
// example call: curl --dump -
|
|
// http://localhost:5555/_api/replication/inventory?batchId=75
|
|
|
|
// {
|
|
// collections : [ ... ],
|
|
// "state" : {
|
|
// "running" : true,
|
|
// "lastLogTick" : "10528",
|
|
// "lastUncommittedLogTick" : "10531",
|
|
// "totalEvents" : 3782,
|
|
// "time" : "2017-07-19T21:50:59Z"
|
|
// },
|
|
// "tick" : "10531"
|
|
// }
|
|
|
|
if (type != rest::RequestType::GET) {
|
|
goto BAD_CALL;
|
|
}
|
|
if (ServerState::instance()->isCoordinator()) {
|
|
handleTrampolineCoordinator();
|
|
} else {
|
|
handleCommandInventory();
|
|
}
|
|
} else if (command == Keys) {
|
|
// preconditions for calling this route are unclear and undocumented --
|
|
// FIXME
|
|
if (type != rest::RequestType::GET && type != rest::RequestType::POST &&
|
|
type != rest::RequestType::PUT && type != rest::RequestType::DELETE_REQ) {
|
|
goto BAD_CALL;
|
|
}
|
|
|
|
if (isCoordinatorError()) {
|
|
return RestStatus::DONE;
|
|
}
|
|
|
|
if (type == rest::RequestType::POST) {
|
|
// has to be called first will bind the iterator to a collection
|
|
|
|
// example: curl -XPOST --dump -
|
|
// 'http://localhost:5555/_db/_system/_api/replication/keys/?collection=_users&batchId=169'
|
|
// ; echo
|
|
// returns
|
|
// { "id": <context id - int>,
|
|
// "count": <number of documents in collection - int>
|
|
// }
|
|
handleCommandCreateKeys();
|
|
} else if (type == rest::RequestType::GET) {
|
|
// curl --dump -
|
|
// 'http://localhost:5555/_db/_system/_api/replication/keys/123?collection=_users'
|
|
// ; echo # id is batchid
|
|
handleCommandGetKeys();
|
|
} else if (type == rest::RequestType::PUT) {
|
|
handleCommandFetchKeys();
|
|
} else if (type == rest::RequestType::DELETE_REQ) {
|
|
handleCommandRemoveKeys();
|
|
}
|
|
} else if (command == Dump) {
|
|
// works on collections
|
|
// example: curl --dump -
|
|
// 'http://localhost:5555/_db/_system/_api/replication/dump?collection=test&batchId=115'
|
|
// requires batch-id
|
|
// does internally an
|
|
// - get inventory
|
|
// - purge local
|
|
// - dump remote to local
|
|
|
|
if (type != rest::RequestType::GET) {
|
|
goto BAD_CALL;
|
|
}
|
|
|
|
if (ServerState::instance()->isCoordinator()) {
|
|
handleTrampolineCoordinator();
|
|
} else {
|
|
handleCommandDump();
|
|
}
|
|
} else if (command == RestoreCollection) {
|
|
if (type != rest::RequestType::PUT) {
|
|
goto BAD_CALL;
|
|
}
|
|
|
|
handleCommandRestoreCollection();
|
|
} else if (command == RestoreIndexes) {
|
|
if (type != rest::RequestType::PUT) {
|
|
goto BAD_CALL;
|
|
}
|
|
|
|
handleCommandRestoreIndexes();
|
|
} else if (command == RestoreData) {
|
|
if (type != rest::RequestType::PUT) {
|
|
goto BAD_CALL;
|
|
}
|
|
handleCommandRestoreData();
|
|
} else if (command == RestoreView) {
|
|
if (type != rest::RequestType::PUT) {
|
|
goto BAD_CALL;
|
|
}
|
|
|
|
handleCommandRestoreView();
|
|
} else if (command == Sync) {
|
|
if (type != rest::RequestType::PUT) {
|
|
goto BAD_CALL;
|
|
}
|
|
|
|
if (isCoordinatorError()) {
|
|
return RestStatus::DONE;
|
|
}
|
|
|
|
handleCommandSync();
|
|
} else if (command == MakeSlave) {
|
|
if (type != rest::RequestType::PUT) {
|
|
goto BAD_CALL;
|
|
}
|
|
|
|
if (isCoordinatorError()) {
|
|
return RestStatus::DONE;
|
|
}
|
|
|
|
handleCommandMakeSlave();
|
|
} else if (command == ServerId) {
|
|
if (type != rest::RequestType::GET) {
|
|
goto BAD_CALL;
|
|
}
|
|
handleCommandServerId();
|
|
} else if (command == ApplierConfig) {
|
|
if (type == rest::RequestType::GET) {
|
|
handleCommandApplierGetConfig();
|
|
} else {
|
|
if (type != rest::RequestType::PUT) {
|
|
goto BAD_CALL;
|
|
}
|
|
handleCommandApplierSetConfig();
|
|
}
|
|
} else if (command == ApplierStart) {
|
|
if (type != rest::RequestType::PUT) {
|
|
goto BAD_CALL;
|
|
}
|
|
|
|
if (isCoordinatorError()) {
|
|
return RestStatus::DONE;
|
|
}
|
|
|
|
handleCommandApplierStart();
|
|
} else if (command == ApplierStop) {
|
|
if (type != rest::RequestType::PUT) {
|
|
goto BAD_CALL;
|
|
}
|
|
|
|
if (isCoordinatorError()) {
|
|
return RestStatus::DONE;
|
|
}
|
|
|
|
handleCommandApplierStop();
|
|
} else if (command == ApplierState) {
|
|
if (type == rest::RequestType::DELETE_REQ) {
|
|
handleCommandApplierDeleteState();
|
|
} else {
|
|
if (type != rest::RequestType::GET) {
|
|
goto BAD_CALL;
|
|
}
|
|
handleCommandApplierGetState();
|
|
}
|
|
} else if (command == ApplierStateAll) {
|
|
if (type != rest::RequestType::GET) {
|
|
goto BAD_CALL;
|
|
}
|
|
handleCommandApplierGetStateAll();
|
|
} else if (command == ClusterInventory) {
|
|
if (type != rest::RequestType::GET) {
|
|
goto BAD_CALL;
|
|
}
|
|
if (!ServerState::instance()->isCoordinator()) {
|
|
generateError(rest::ResponseCode::FORBIDDEN, TRI_ERROR_CLUSTER_ONLY_ON_COORDINATOR);
|
|
} else {
|
|
handleCommandClusterInventory();
|
|
}
|
|
} else if (command == AddFollower) {
|
|
if (type != rest::RequestType::PUT) {
|
|
goto BAD_CALL;
|
|
}
|
|
if (!ServerState::instance()->isDBServer()) {
|
|
generateError(rest::ResponseCode::FORBIDDEN, TRI_ERROR_CLUSTER_ONLY_ON_DBSERVER);
|
|
} else {
|
|
handleCommandAddFollower();
|
|
}
|
|
} else if (command == RemoveFollower) {
|
|
if (type != rest::RequestType::PUT) {
|
|
goto BAD_CALL;
|
|
}
|
|
if (!ServerState::instance()->isDBServer()) {
|
|
generateError(rest::ResponseCode::FORBIDDEN, TRI_ERROR_CLUSTER_ONLY_ON_DBSERVER);
|
|
} else {
|
|
handleCommandRemoveFollower();
|
|
}
|
|
} else if (command == SetTheLeader) {
|
|
if (type != rest::RequestType::PUT) {
|
|
goto BAD_CALL;
|
|
}
|
|
if (!ServerState::instance()->isDBServer()) {
|
|
generateError(rest::ResponseCode::FORBIDDEN, TRI_ERROR_CLUSTER_ONLY_ON_DBSERVER);
|
|
} else {
|
|
handleCommandSetTheLeader();
|
|
}
|
|
} else if (command == HoldReadLockCollection) {
|
|
if (!ServerState::instance()->isDBServer()) {
|
|
generateError(rest::ResponseCode::FORBIDDEN, TRI_ERROR_CLUSTER_ONLY_ON_DBSERVER);
|
|
} else {
|
|
if (type == rest::RequestType::POST) {
|
|
handleCommandHoldReadLockCollection();
|
|
} else if (type == rest::RequestType::PUT) {
|
|
handleCommandCheckHoldReadLockCollection();
|
|
} else if (type == rest::RequestType::DELETE_REQ) {
|
|
handleCommandCancelHoldReadLockCollection();
|
|
} else if (type == rest::RequestType::GET) {
|
|
handleCommandGetIdForReadLockCollection();
|
|
} else {
|
|
goto BAD_CALL;
|
|
}
|
|
}
|
|
} else {
|
|
generateError(rest::ResponseCode::BAD, TRI_ERROR_HTTP_BAD_PARAMETER,
|
|
std::string("invalid command '") + command + "'");
|
|
}
|
|
|
|
return RestStatus::DONE;
|
|
}
|
|
|
|
BAD_CALL:
|
|
if (len != 1) {
|
|
generateError(rest::ResponseCode::BAD, TRI_ERROR_HTTP_SUPERFLUOUS_SUFFICES,
|
|
"expecting URL /_api/replication/<command>");
|
|
} else {
|
|
generateError(rest::ResponseCode::METHOD_NOT_ALLOWED, TRI_ERROR_HTTP_METHOD_NOT_ALLOWED);
|
|
}
|
|
|
|
return RestStatus::DONE;
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
/// @brief was docuBlock JSF_put_api_replication_makeSlave
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
|
|
void RestReplicationHandler::handleCommandMakeSlave() {
|
|
bool isGlobal = false;
|
|
ReplicationApplier* applier = getApplier(isGlobal);
|
|
if (applier == nullptr) {
|
|
return;
|
|
}
|
|
|
|
bool success = false;
|
|
VPackSlice body = this->parseVPackBody(success);
|
|
if (!success) {
|
|
// error already created
|
|
return;
|
|
}
|
|
|
|
std::string databaseName;
|
|
|
|
if (!isGlobal) {
|
|
databaseName = _vocbase.name();
|
|
}
|
|
|
|
ReplicationApplierConfiguration configuration =
|
|
ReplicationApplierConfiguration::fromVelocyPack(body, databaseName);
|
|
configuration._skipCreateDrop = false;
|
|
|
|
// will throw if invalid
|
|
configuration.validate();
|
|
|
|
// allow access to _users if appropriate
|
|
grantTemporaryRights();
|
|
|
|
// forget about any existing replication applier configuration
|
|
applier->forget();
|
|
applier->reconfigure(configuration);
|
|
applier->startReplication();
|
|
|
|
while (applier->isInitializing()) { // wait for initial sync
|
|
std::this_thread::sleep_for(std::chrono::milliseconds(50));
|
|
if (application_features::ApplicationServer::isStopping()) {
|
|
generateError(Result(TRI_ERROR_SHUTTING_DOWN));
|
|
return;
|
|
}
|
|
}
|
|
// applier->startTailing(lastLogTick, true, barrierId);
|
|
|
|
VPackBuilder result;
|
|
result.openObject();
|
|
applier->toVelocyPack(result);
|
|
result.close();
|
|
generateResult(rest::ResponseCode::OK, result.slice());
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
/// @brief forward a command in the coordinator case
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
|
|
void RestReplicationHandler::handleTrampolineCoordinator() {
|
|
bool useVst = false;
|
|
if (_request->transportType() == Endpoint::TransportType::VST) {
|
|
useVst = true;
|
|
}
|
|
if (_request == nullptr) {
|
|
THROW_ARANGO_EXCEPTION_MESSAGE(TRI_ERROR_INTERNAL, "invalid request");
|
|
}
|
|
|
|
// First check the DBserver component of the body json:
|
|
ServerID DBserver = _request->value("DBserver");
|
|
|
|
if (DBserver.empty()) {
|
|
generateError(rest::ResponseCode::BAD, TRI_ERROR_HTTP_BAD_PARAMETER,
|
|
"need \"DBserver\" parameter");
|
|
return;
|
|
}
|
|
|
|
std::string const& dbname = _request->databaseName();
|
|
|
|
auto headers = std::make_shared<std::unordered_map<std::string, std::string>>(
|
|
arangodb::getForwardableRequestHeaders(_request.get()));
|
|
std::unordered_map<std::string, std::string> values = _request->values();
|
|
std::string params;
|
|
|
|
for (auto const& i : values) {
|
|
if (i.first != "DBserver") {
|
|
if (params.empty()) {
|
|
params.push_back('?');
|
|
} else {
|
|
params.push_back('&');
|
|
}
|
|
params.append(StringUtils::urlEncode(i.first));
|
|
params.push_back('=');
|
|
params.append(StringUtils::urlEncode(i.second));
|
|
}
|
|
}
|
|
|
|
// Set a few variables needed for our work:
|
|
auto cc = ClusterComm::instance();
|
|
if (cc == nullptr) {
|
|
// nullptr happens only during controlled shutdown
|
|
generateError(rest::ResponseCode::SERVICE_UNAVAILABLE,
|
|
TRI_ERROR_SHUTTING_DOWN, "shutting down server");
|
|
return;
|
|
}
|
|
|
|
std::unique_ptr<ClusterCommResult> res;
|
|
if (!useVst) {
|
|
TRI_ASSERT(_request->transportType() == Endpoint::TransportType::HTTP);
|
|
|
|
VPackStringRef body = _request->rawPayload();
|
|
// Send a synchronous request to that shard using ClusterComm:
|
|
res = cc->syncRequest(TRI_NewTickServer(), "server:" + DBserver,
|
|
_request->requestType(),
|
|
"/_db/" + StringUtils::urlEncode(dbname) +
|
|
_request->requestPath() + params,
|
|
body.toString(), *headers, 300.0);
|
|
} else {
|
|
// do we need to handle multiple payloads here - TODO
|
|
// here we switch from vst to http?!
|
|
res = cc->syncRequest(TRI_NewTickServer(), "server:" + DBserver,
|
|
_request->requestType(),
|
|
"/_db/" + StringUtils::urlEncode(dbname) +
|
|
_request->requestPath() + params,
|
|
_request->payload().toJson(), *headers, 300.0);
|
|
}
|
|
|
|
if (res->status == CL_COMM_TIMEOUT) {
|
|
// No reply, we give up:
|
|
generateError(rest::ResponseCode::BAD, TRI_ERROR_CLUSTER_TIMEOUT,
|
|
"timeout within cluster");
|
|
return;
|
|
}
|
|
if (res->status == CL_COMM_BACKEND_UNAVAILABLE) {
|
|
// there is no result
|
|
generateError(rest::ResponseCode::BAD, TRI_ERROR_CLUSTER_CONNECTION_LOST,
|
|
"lost connection within cluster");
|
|
return;
|
|
}
|
|
if (res->status == CL_COMM_ERROR) {
|
|
// This could be a broken connection or an Http error:
|
|
TRI_ASSERT(nullptr != res->result && res->result->isComplete());
|
|
// In this case a proper HTTP error was reported by the DBserver,
|
|
// we simply forward the result.
|
|
// We intentionally fall through here.
|
|
}
|
|
|
|
bool dummy;
|
|
resetResponse(static_cast<rest::ResponseCode>(res->result->getHttpReturnCode()));
|
|
|
|
_response->setContentType(
|
|
res->result->getHeaderField(StaticStrings::ContentTypeHeader, dummy));
|
|
|
|
if (!useVst) {
|
|
HttpResponse* httpResponse = dynamic_cast<HttpResponse*>(_response.get());
|
|
if (_response == nullptr) {
|
|
THROW_ARANGO_EXCEPTION_MESSAGE(TRI_ERROR_INTERNAL,
|
|
"invalid response type");
|
|
}
|
|
httpResponse->body().swap(&(res->result->getBody()));
|
|
} else {
|
|
std::shared_ptr<VPackBuilder> builder = res->result->getBodyVelocyPack();
|
|
std::shared_ptr<VPackBuffer<uint8_t>> buf = builder->steal();
|
|
_response->setPayload(std::move(*buf),
|
|
true); // do we need to generate the body?!
|
|
}
|
|
|
|
auto const& resultHeaders = res->result->getHeaderFields();
|
|
for (auto const& it : resultHeaders) {
|
|
_response->setHeader(it.first, it.second);
|
|
}
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
/// @brief was docuBlock JSF_get_api_replication_cluster_inventory
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
|
|
void RestReplicationHandler::handleCommandClusterInventory() {
|
|
std::string const& dbName = _request->databaseName();
|
|
bool includeSystem = _request->parsedValue("includeSystem", true);
|
|
|
|
ClusterInfo* ci = ClusterInfo::instance();
|
|
std::vector<std::shared_ptr<LogicalCollection>> cols = ci->getCollections(dbName);
|
|
VPackBuilder resultBuilder;
|
|
resultBuilder.openObject();
|
|
resultBuilder.add("collections", VPackValue(VPackValueType::Array));
|
|
for (std::shared_ptr<LogicalCollection> const& c : cols) {
|
|
// We want to check if the collection is usable and all followers
|
|
// are in sync:
|
|
std::shared_ptr<ShardMap> shardMap = c->shardIds();
|
|
// shardMap is an unordered_map from ShardId (string) to a vector of
|
|
// servers (strings), wrapped in a shared_ptr
|
|
auto cic = ci->getCollectionCurrent(dbName, basics::StringUtils::itoa(c->id()));
|
|
// Check all shards:
|
|
bool isReady = true;
|
|
bool allInSync = true;
|
|
for (auto const& p : *shardMap) {
|
|
auto currentServerList = cic->servers(p.first /* shardId */);
|
|
if (currentServerList.size() == 0 || p.second.size() == 0 ||
|
|
currentServerList[0] != p.second[0] ||
|
|
(!p.second[0].empty() && p.second[0][0] == '_')) {
|
|
isReady = false;
|
|
}
|
|
if (!ClusterHelpers::compareServerLists(p.second, currentServerList)) {
|
|
allInSync = false;
|
|
}
|
|
}
|
|
c->toVelocyPackForClusterInventory(resultBuilder, includeSystem, isReady, allInSync);
|
|
}
|
|
resultBuilder.close(); // collections
|
|
resultBuilder.add("views", VPackValue(VPackValueType::Array));
|
|
LogicalView::enumerate(_vocbase, [&resultBuilder](LogicalView::ptr const& view) -> bool {
|
|
if (view) {
|
|
resultBuilder.openObject();
|
|
view->properties(resultBuilder, LogicalDataSource::Serialization::Inventory);
|
|
// details, !forPersistence because on restore any datasource ids will
|
|
// differ, so need an end-user representation
|
|
resultBuilder.close();
|
|
}
|
|
|
|
return true;
|
|
});
|
|
resultBuilder.close(); // views
|
|
|
|
TRI_voc_tick_t tick = TRI_CurrentTickServer();
|
|
resultBuilder.add("tick", VPackValue(std::to_string(tick)));
|
|
resultBuilder.add("state", VPackValue("unused"));
|
|
resultBuilder.close(); // base
|
|
generateResult(rest::ResponseCode::OK, resultBuilder.slice());
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
/// @brief restores the structure of a collection TODO MOVE
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
|
|
void RestReplicationHandler::handleCommandRestoreCollection() {
|
|
std::shared_ptr<VPackBuilder> parsedRequest;
|
|
|
|
try {
|
|
parsedRequest = _request->toVelocyPackBuilderPtr();
|
|
} catch (arangodb::velocypack::Exception const& e) {
|
|
std::string errorMsg = "invalid JSON: ";
|
|
errorMsg += e.what();
|
|
generateError(rest::ResponseCode::BAD, TRI_ERROR_HTTP_BAD_PARAMETER, errorMsg);
|
|
return;
|
|
} catch (...) {
|
|
generateError(rest::ResponseCode::BAD, TRI_ERROR_HTTP_BAD_PARAMETER,
|
|
"invalid JSON");
|
|
return;
|
|
}
|
|
auto pair = rocksutils::stripObjectIds(parsedRequest->slice());
|
|
VPackSlice const slice = pair.first;
|
|
|
|
bool overwrite = _request->parsedValue<bool>("overwrite", false);
|
|
bool force = _request->parsedValue<bool>("force", false);
|
|
;
|
|
bool ignoreDistributeShardsLikeErrors =
|
|
_request->parsedValue<bool>("ignoreDistributeShardsLikeErrors", false);
|
|
uint64_t numberOfShards =
|
|
_request->parsedValue<uint64_t>(StaticStrings::NumberOfShards, 0);
|
|
uint64_t replicationFactor =
|
|
_request->parsedValue<uint64_t>(StaticStrings::ReplicationFactor, 1);
|
|
uint64_t minReplicationFactor =
|
|
_request->parsedValue<uint64_t>(StaticStrings::MinReplicationFactor, 1);
|
|
|
|
Result res;
|
|
if (ServerState::instance()->isCoordinator()) {
|
|
res = processRestoreCollectionCoordinator(slice, overwrite, force, numberOfShards,
|
|
replicationFactor, minReplicationFactor,
|
|
ignoreDistributeShardsLikeErrors);
|
|
} else {
|
|
res = processRestoreCollection(slice, overwrite, force);
|
|
}
|
|
|
|
if (res.is(TRI_ERROR_ARANGO_DUPLICATE_NAME)) {
|
|
VPackSlice name = slice.get("parameters");
|
|
if (name.isObject()) {
|
|
name = name.get("name");
|
|
if (name.isString() && !name.copyString().empty() && name.copyString()[0] == '_') {
|
|
// system collection, may have been created in the meantime by a
|
|
// background action collection should be there now
|
|
res.reset();
|
|
}
|
|
}
|
|
}
|
|
|
|
if (res.fail()) {
|
|
THROW_ARANGO_EXCEPTION(res);
|
|
}
|
|
|
|
VPackBuilder result;
|
|
result.add(VPackValue(VPackValueType::Object));
|
|
result.add("result", VPackValue(true));
|
|
result.close();
|
|
generateResult(rest::ResponseCode::OK, result.slice());
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
/// @brief restores the indexes of a collection
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
|
|
void RestReplicationHandler::handleCommandRestoreIndexes() {
|
|
std::shared_ptr<VPackBuilder> parsedRequest;
|
|
|
|
try {
|
|
parsedRequest = _request->toVelocyPackBuilderPtr();
|
|
} catch (...) {
|
|
generateError(rest::ResponseCode::BAD, TRI_ERROR_HTTP_BAD_PARAMETER,
|
|
"invalid JSON");
|
|
return;
|
|
}
|
|
VPackSlice const slice = parsedRequest->slice();
|
|
|
|
bool force = _request->parsedValue("force", false);
|
|
|
|
Result res;
|
|
if (ServerState::instance()->isCoordinator()) {
|
|
res = processRestoreIndexesCoordinator(slice, force);
|
|
} else {
|
|
res = processRestoreIndexes(slice, force);
|
|
}
|
|
|
|
if (res.fail()) {
|
|
THROW_ARANGO_EXCEPTION(res);
|
|
}
|
|
|
|
VPackBuilder result;
|
|
result.openObject();
|
|
result.add("result", VPackValue(true));
|
|
result.close();
|
|
generateResult(rest::ResponseCode::OK, result.slice());
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
/// @brief restores the data of a collection
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
|
|
void RestReplicationHandler::handleCommandRestoreData() {
|
|
std::string const& colName = _request->value("collection");
|
|
|
|
if (colName.empty()) {
|
|
generateError(rest::ResponseCode::BAD, TRI_ERROR_HTTP_BAD_PARAMETER,
|
|
"invalid collection parameter, not given");
|
|
return;
|
|
}
|
|
|
|
Result res = processRestoreData(colName);
|
|
|
|
if (res.fail()) {
|
|
if (res.errorMessage().empty()) {
|
|
generateError(GeneralResponse::responseCode(res.errorNumber()), res.errorNumber());
|
|
} else {
|
|
generateError(GeneralResponse::responseCode(res.errorNumber()), res.errorNumber(),
|
|
std::string(TRI_errno_string(res.errorNumber())) + ": " +
|
|
res.errorMessage());
|
|
}
|
|
} else {
|
|
VPackBuilder result;
|
|
result.add(VPackValue(VPackValueType::Object));
|
|
result.add("result", VPackValue(true));
|
|
result.close();
|
|
generateResult(rest::ResponseCode::OK, result.slice());
|
|
}
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
/// @brief restores the structure of a collection TODO MOVE
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
|
|
Result RestReplicationHandler::processRestoreCollection(VPackSlice const& collection,
|
|
bool dropExisting, bool /*force*/) {
|
|
if (!collection.isObject()) {
|
|
return Result(TRI_ERROR_HTTP_BAD_PARAMETER,
|
|
"collection declaration is invalid");
|
|
}
|
|
|
|
VPackSlice const parameters = collection.get("parameters");
|
|
|
|
if (!parameters.isObject()) {
|
|
return Result(TRI_ERROR_HTTP_BAD_PARAMETER,
|
|
"collection parameters declaration is invalid");
|
|
}
|
|
|
|
VPackSlice const indexes = collection.get("indexes");
|
|
|
|
if (!indexes.isArray()) {
|
|
return Result(TRI_ERROR_HTTP_BAD_PARAMETER,
|
|
"collection indexes declaration is invalid");
|
|
}
|
|
|
|
std::string const name =
|
|
arangodb::basics::VelocyPackHelper::getStringValue(parameters, "name", "");
|
|
|
|
if (name.empty()) {
|
|
return Result(TRI_ERROR_HTTP_BAD_PARAMETER, "collection name is missing");
|
|
}
|
|
|
|
if (arangodb::basics::VelocyPackHelper::getBooleanValue(parameters, "deleted", false)) {
|
|
// we don't care about deleted collections
|
|
return Result();
|
|
}
|
|
|
|
grantTemporaryRights();
|
|
|
|
auto* col = _vocbase.lookupCollection(name).get();
|
|
|
|
// drop an existing collection if it exists
|
|
if (col != nullptr) {
|
|
if (dropExisting) {
|
|
auto res = _vocbase.dropCollection(col->id(), true, -1.0);
|
|
|
|
if (res.errorNumber() == TRI_ERROR_FORBIDDEN) {
|
|
// some collections must not be dropped
|
|
|
|
// instead, truncate them
|
|
auto ctx = transaction::StandaloneContext::Create(_vocbase);
|
|
SingleCollectionTransaction trx(ctx, *col, AccessMode::Type::EXCLUSIVE);
|
|
|
|
// to turn off waitForSync!
|
|
trx.addHint(transaction::Hints::Hint::RECOVERY);
|
|
trx.addHint(transaction::Hints::Hint::INTERMEDIATE_COMMITS);
|
|
trx.addHint(transaction::Hints::Hint::ALLOW_RANGE_DELETE);
|
|
res = trx.begin();
|
|
|
|
if (!res.ok()) {
|
|
return res;
|
|
}
|
|
|
|
OperationOptions options;
|
|
OperationResult opRes = trx.truncate(name, options);
|
|
|
|
return trx.finish(opRes.result);
|
|
}
|
|
|
|
if (!res.ok()) {
|
|
return Result(res.errorNumber(),
|
|
std::string("unable to drop collection '") + name +
|
|
"': " + res.errorMessage());
|
|
}
|
|
// intentionally falls through
|
|
} else {
|
|
return Result(TRI_ERROR_ARANGO_DUPLICATE_NAME,
|
|
std::string("unable to create collection '") + name +
|
|
"': " + TRI_errno_string(TRI_ERROR_ARANGO_DUPLICATE_NAME));
|
|
}
|
|
}
|
|
|
|
// now re-create the collection
|
|
int res = createCollection(parameters, &col);
|
|
|
|
if (res != TRI_ERROR_NO_ERROR) {
|
|
return Result(res, std::string("unable to create collection: ") + TRI_errno_string(res));
|
|
}
|
|
|
|
// might be also called on dbservers
|
|
ExecContext const* exe = ExecContext::CURRENT;
|
|
if (name[0] != '_' && exe != nullptr && !exe->isSuperuser() &&
|
|
ServerState::instance()->isSingleServer()) {
|
|
auth::UserManager* um = AuthenticationFeature::instance()->userManager();
|
|
TRI_ASSERT(um != nullptr); // should not get here
|
|
if (um != nullptr) {
|
|
um->updateUser(exe->user(), [&](auth::User& entry) {
|
|
entry.grantCollection(_vocbase.name(), col->name(), auth::Level::RW);
|
|
return TRI_ERROR_NO_ERROR;
|
|
});
|
|
}
|
|
}
|
|
|
|
return Result();
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
/// @brief restores the structure of a collection, coordinator case
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
|
|
Result RestReplicationHandler::processRestoreCollectionCoordinator(
|
|
VPackSlice const& collection, bool dropExisting, bool force,
|
|
uint64_t numberOfShards, uint64_t replicationFactor,
|
|
uint64_t minReplicationFactor, bool ignoreDistributeShardsLikeErrors) {
|
|
if (!collection.isObject()) {
|
|
return Result(TRI_ERROR_HTTP_BAD_PARAMETER,
|
|
"collection declaration is invalid");
|
|
}
|
|
|
|
VPackSlice const parameters = collection.get("parameters");
|
|
|
|
if (!parameters.isObject()) {
|
|
return Result(TRI_ERROR_HTTP_BAD_PARAMETER,
|
|
"collection parameters declaration is invalid");
|
|
}
|
|
|
|
std::string const name =
|
|
arangodb::basics::VelocyPackHelper::getStringValue(parameters, "name", "");
|
|
|
|
if (name.empty()) {
|
|
return Result(TRI_ERROR_HTTP_BAD_PARAMETER, "collection name is missing");
|
|
}
|
|
|
|
if (ignoreHiddenEnterpriseCollection(name, force)) {
|
|
return {TRI_ERROR_NO_ERROR};
|
|
}
|
|
|
|
if (arangodb::basics::VelocyPackHelper::getBooleanValue(parameters, "deleted", false)) {
|
|
// we don't care about deleted collections
|
|
return Result();
|
|
}
|
|
|
|
Result res = ShardingInfo::validateShardsAndReplicationFactor(parameters);
|
|
if (res.fail()) {
|
|
return res;
|
|
}
|
|
|
|
auto& dbName = _vocbase.name();
|
|
ClusterInfo* ci = ClusterInfo::instance();
|
|
|
|
try {
|
|
// in a cluster, we only look up by name:
|
|
std::shared_ptr<LogicalCollection> col = ci->getCollection(dbName, name);
|
|
|
|
// drop an existing collection if it exists
|
|
if (dropExisting) {
|
|
auto result = ci->dropCollectionCoordinator(dbName, std::to_string(col->id()), 0.0);
|
|
|
|
if (TRI_ERROR_FORBIDDEN == result.errorNumber() // forbidden
|
|
|| TRI_ERROR_CLUSTER_MUST_NOT_DROP_COLL_OTHER_DISTRIBUTESHARDSLIKE ==
|
|
result.errorNumber()) {
|
|
// some collections must not be dropped
|
|
auto ctx = transaction::StandaloneContext::Create(_vocbase);
|
|
SingleCollectionTransaction trx(ctx, name, AccessMode::Type::EXCLUSIVE);
|
|
|
|
Result res = trx.begin();
|
|
if (res.fail()) {
|
|
return res;
|
|
}
|
|
|
|
OperationOptions options;
|
|
OperationResult result = trx.truncate(name, options);
|
|
res = trx.finish(result.result);
|
|
if (res.fail()) {
|
|
res.appendErrorMessage(
|
|
". Unable to truncate collection (dropping is forbidden)");
|
|
}
|
|
return res;
|
|
}
|
|
|
|
if (!result.ok()) {
|
|
return Result( // result
|
|
result.errorNumber(), // code
|
|
std::string("unable to drop collection '") + name + "': " + result.errorMessage());
|
|
}
|
|
} else {
|
|
return Result(TRI_ERROR_ARANGO_DUPLICATE_NAME,
|
|
std::string("unable to create collection '") + name +
|
|
"': " + TRI_errno_string(TRI_ERROR_ARANGO_DUPLICATE_NAME));
|
|
}
|
|
} catch (basics::Exception const& ex) {
|
|
LOG_TOPIC("41579", DEBUG, Logger::REPLICATION)
|
|
<< "processRestoreCollectionCoordinator "
|
|
<< "could not drop collection: " << ex.what();
|
|
} catch (...) {
|
|
}
|
|
|
|
// now re-create the collection
|
|
|
|
// Build up new information that we need to merge with the given one
|
|
VPackBuilder toMerge;
|
|
toMerge.openObject();
|
|
|
|
// We always need a new id
|
|
TRI_voc_tick_t newIdTick = ci->uniqid(1);
|
|
std::string newId = StringUtils::itoa(newIdTick);
|
|
toMerge.add("id", VPackValue(newId));
|
|
|
|
// Number of shards. Will be overwritten if not existent
|
|
VPackSlice const numberOfShardsSlice = parameters.get(StaticStrings::NumberOfShards);
|
|
if (!numberOfShardsSlice.isInteger()) {
|
|
// The information does not contain numberOfShards. Overwrite it.
|
|
VPackSlice const shards = parameters.get("shards");
|
|
if (shards.isObject()) {
|
|
numberOfShards = static_cast<uint64_t>(shards.length());
|
|
} else {
|
|
// "shards" not specified
|
|
// now check if numberOfShards property was given
|
|
if (numberOfShards == 0) {
|
|
// We take one shard if no value was given
|
|
numberOfShards = 1;
|
|
}
|
|
}
|
|
TRI_ASSERT(numberOfShards > 0);
|
|
toMerge.add(StaticStrings::NumberOfShards, VPackValue(numberOfShards));
|
|
}
|
|
|
|
// Replication Factor. Will be overwritten if not existent
|
|
VPackSlice const replFactorSlice = parameters.get(StaticStrings::ReplicationFactor);
|
|
VPackSlice const minReplFactorSlice = parameters.get(StaticStrings::MinReplicationFactor);
|
|
|
|
bool isValidReplFactorSlice = replFactorSlice.isInteger() ||
|
|
(replFactorSlice.isString() &&
|
|
replFactorSlice.isEqualString("satellite"));
|
|
|
|
bool isValidMinReplFactorSlice =
|
|
replFactorSlice.isInteger() && minReplFactorSlice.isInteger() &&
|
|
(minReplFactorSlice.getInt() <= replFactorSlice.getInt()) &&
|
|
minReplFactorSlice.getInt() > 0;
|
|
|
|
if (!isValidReplFactorSlice) {
|
|
if (replicationFactor == 0) {
|
|
replicationFactor = application_features::ApplicationServer::getFeature<ClusterFeature>("Cluster")->defaultReplicationFactor();
|
|
if (replicationFactor == 0) {
|
|
replicationFactor = 1;
|
|
}
|
|
}
|
|
TRI_ASSERT(replicationFactor > 0);
|
|
toMerge.add(StaticStrings::ReplicationFactor, VPackValue(replicationFactor));
|
|
}
|
|
|
|
if (!isValidMinReplFactorSlice) {
|
|
if (replFactorSlice.isString() &&
|
|
replFactorSlice.isEqualString("satellite")) {
|
|
minReplicationFactor = 0;
|
|
} else if (minReplicationFactor == 0) {
|
|
minReplicationFactor = 1;
|
|
}
|
|
TRI_ASSERT(minReplicationFactor <= replicationFactor);
|
|
toMerge.add(StaticStrings::MinReplicationFactor, VPackValue(minReplicationFactor));
|
|
}
|
|
|
|
// always use current version number when restoring a collection,
|
|
// because the collection is effectively NEW
|
|
toMerge.add("version", VPackValue(LogicalCollection::VERSION_33));
|
|
if (!name.empty() && name[0] == '_' && !parameters.hasKey(StaticStrings::DataSourceSystem)) {
|
|
// system collection?
|
|
toMerge.add(StaticStrings::DataSourceSystem, VPackValue(true));
|
|
}
|
|
|
|
#ifndef USE_ENTERPRISE
|
|
std::vector<std::string> changes;
|
|
|
|
// when in the community version, we need to turn off specific attributes
|
|
// because they are only supported in enterprise mode
|
|
|
|
// watch out for "isSmart" -> we need to set this to false in the community version
|
|
VPackSlice s = parameters.get(StaticStrings::GraphIsSmart);
|
|
if (s.isBoolean() && s.getBoolean()) {
|
|
// isSmart needs to be turned off in the community version
|
|
toMerge.add(StaticStrings::GraphIsSmart, VPackValue(false));
|
|
changes.push_back("changed 'isSmart' attribute value to false");
|
|
}
|
|
|
|
// "smartGraphAttribute" needs to be set to be removed too
|
|
s = parameters.get(StaticStrings::GraphSmartGraphAttribute);
|
|
if (s.isString() && !s.copyString().empty()) {
|
|
// smartGraphAttribute needs to be removed
|
|
toMerge.add(StaticStrings::GraphSmartGraphAttribute, VPackSlice::nullSlice());
|
|
changes.push_back("removed 'smartGraphAttribute' attribute value");
|
|
}
|
|
|
|
// same for "smartJoinAttribute"
|
|
s = parameters.get(StaticStrings::SmartJoinAttribute);
|
|
if (s.isString() && !s.copyString().empty()) {
|
|
// smartJoinAttribute needs to be removed
|
|
toMerge.add(StaticStrings::SmartJoinAttribute, VPackSlice::nullSlice());
|
|
changes.push_back("removed 'smartJoinAttribute' attribute value");
|
|
}
|
|
|
|
// finally rewrite all enterprise sharding strategies to a simple hash-based strategy
|
|
s = parameters.get("shardingStrategy");
|
|
if (s.isString() && s.copyString().find("enterprise") != std::string::npos) {
|
|
// downgrade sharding strategy to just hash
|
|
toMerge.add("shardingStrategy", VPackValue("hash"));
|
|
changes.push_back("changed 'shardingStrategy' attribute value to 'hash'");
|
|
}
|
|
|
|
s = parameters.get(StaticStrings::ReplicationFactor);
|
|
if (s.isString() && s.copyString() == "satellite") {
|
|
// set "satellite" replicationFactor to the default replication factor
|
|
ClusterFeature* cl = application_features::ApplicationServer::getFeature<ClusterFeature>(
|
|
"Cluster");
|
|
|
|
uint32_t replicationFactor = cl->systemReplicationFactor();
|
|
toMerge.add(StaticStrings::ReplicationFactor, VPackValue(replicationFactor));
|
|
changes.push_back(
|
|
std::string("changed 'replicationFactor' attribute value to ") +
|
|
std::to_string(replicationFactor));
|
|
;
|
|
}
|
|
|
|
if (!changes.empty()) {
|
|
LOG_TOPIC("fc359", INFO, Logger::CLUSTER)
|
|
<< "rewrote info for collection '"
|
|
<< name << "' on restore for usage with the community version. the following changes were applied: "
|
|
<< basics::StringUtils::join(changes, ". ");
|
|
}
|
|
#endif
|
|
|
|
// Always ignore `shadowCollections` they were accidentially dumped in
|
|
// arangodb versions earlier than 3.3.6
|
|
toMerge.add("shadowCollections", arangodb::velocypack::Slice::nullSlice());
|
|
toMerge.close(); // TopLevel
|
|
|
|
VPackSlice const type = parameters.get("type");
|
|
if (!type.isNumber()) {
|
|
return Result(TRI_ERROR_HTTP_BAD_PARAMETER,
|
|
"collection type not given or wrong");
|
|
}
|
|
|
|
VPackSlice const sliceToMerge = toMerge.slice();
|
|
VPackBuilder mergedBuilder =
|
|
VPackCollection::merge(parameters, sliceToMerge, false, true);
|
|
VPackBuilder arrayWrapper;
|
|
arrayWrapper.openArray();
|
|
arrayWrapper.add(mergedBuilder.slice());
|
|
arrayWrapper.close();
|
|
VPackSlice const merged = arrayWrapper.slice();
|
|
|
|
try {
|
|
bool createWaitsForSyncReplication =
|
|
application_features::ApplicationServer::getFeature<ClusterFeature>(
|
|
"Cluster")
|
|
->createWaitsForSyncReplication();
|
|
// in the replication case enforcing the replication factor is absolutely
|
|
// not desired, so it is hardcoded to false
|
|
auto cols =
|
|
ClusterMethods::createCollectionOnCoordinator(_vocbase, merged, ignoreDistributeShardsLikeErrors,
|
|
createWaitsForSyncReplication,
|
|
false, false, nullptr);
|
|
ExecContext const* exe = ExecContext::CURRENT;
|
|
TRI_ASSERT(cols.size() == 1);
|
|
if (name[0] != '_' && exe != nullptr && !exe->isSuperuser()) {
|
|
auth::UserManager* um = AuthenticationFeature::instance()->userManager();
|
|
TRI_ASSERT(um != nullptr); // should not get here
|
|
if (um != nullptr) {
|
|
um->updateUser(ExecContext::CURRENT->user(), [&](auth::User& entry) {
|
|
for (auto const& col : cols) {
|
|
TRI_ASSERT(col != nullptr);
|
|
entry.grantCollection(dbName, col->name(), auth::Level::RW);
|
|
}
|
|
return TRI_ERROR_NO_ERROR;
|
|
});
|
|
}
|
|
}
|
|
} catch (basics::Exception const& ex) {
|
|
// Error, report it.
|
|
return Result(ex.code(), ex.what());
|
|
} catch (std::exception const& ex) {
|
|
return Result(TRI_ERROR_INTERNAL, ex.what());
|
|
}
|
|
// All other errors are thrown to the outside.
|
|
return Result();
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
/// @brief restores the data of a collection
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
|
|
Result RestReplicationHandler::processRestoreData(std::string const& colName) {
|
|
#ifdef USE_ENTERPRISE
|
|
bool force = _request->parsedValue("force", false);
|
|
if (ignoreHiddenEnterpriseCollection(colName, force)) {
|
|
return {TRI_ERROR_NO_ERROR};
|
|
}
|
|
#endif
|
|
|
|
grantTemporaryRights();
|
|
|
|
if (colName == TRI_COL_NAME_USERS) {
|
|
// We need to handle the _users in a special way
|
|
return processRestoreUsersBatch(colName);
|
|
}
|
|
|
|
auto ctx = transaction::StandaloneContext::Create(_vocbase);
|
|
SingleCollectionTransaction trx(ctx, colName, AccessMode::Type::WRITE);
|
|
|
|
trx.addHint(transaction::Hints::Hint::RECOVERY); // to turn off waitForSync!
|
|
|
|
Result res = trx.begin();
|
|
|
|
if (!res.ok()) {
|
|
res.reset(res.errorNumber(), std::string("unable to start transaction (") +
|
|
std::string(__FILE__) + std::string(":") +
|
|
std::to_string(__LINE__) +
|
|
std::string("): ") + res.errorMessage());
|
|
return res;
|
|
}
|
|
|
|
res = processRestoreDataBatch(trx, colName);
|
|
res = trx.finish(res);
|
|
|
|
return res;
|
|
}
|
|
|
|
Result RestReplicationHandler::parseBatch(std::string const& collectionName,
|
|
std::unordered_map<std::string, VPackValueLength>& latest,
|
|
VPackBuilder& allMarkers) {
|
|
VPackOptions options = VPackOptions::Defaults;
|
|
options.checkAttributeUniqueness = true;
|
|
VPackBuilder builder(&options);
|
|
|
|
allMarkers.clear();
|
|
|
|
if (_request->transportType() != Endpoint::TransportType::HTTP) {
|
|
THROW_ARANGO_EXCEPTION_MESSAGE(TRI_ERROR_INTERNAL, "invalid request type");
|
|
}
|
|
|
|
VPackStringRef bodyStr = _request->rawPayload();
|
|
char const* ptr = bodyStr.data();
|
|
char const* end = ptr + bodyStr.size();
|
|
|
|
VPackValueLength currentPos = 0;
|
|
|
|
// First parse and collect all markers, we assemble everything in one
|
|
// large builder holding an array. We keep for each key the latest
|
|
// entry.
|
|
|
|
{
|
|
int line = 0;
|
|
VPackArrayBuilder guard(&allMarkers);
|
|
std::string key;
|
|
while (ptr < end) {
|
|
char const* pos = strchr(ptr, '\n');
|
|
|
|
if (pos == nullptr) {
|
|
pos = end;
|
|
} else {
|
|
*((char*)pos) = '\0';
|
|
++line;
|
|
}
|
|
|
|
if (pos - ptr > 1) {
|
|
// found something
|
|
key.clear();
|
|
VPackSlice doc;
|
|
TRI_replication_operation_e type = REPLICATION_INVALID;
|
|
|
|
Result res =
|
|
restoreDataParser(ptr, pos, collectionName, line, key, builder, doc, type);
|
|
if (res.fail()) {
|
|
return res;
|
|
}
|
|
|
|
// Put into array of all parsed markers:
|
|
allMarkers.add(builder.slice());
|
|
latest[key] = currentPos;
|
|
++currentPos;
|
|
}
|
|
|
|
ptr = pos + 1;
|
|
}
|
|
}
|
|
|
|
return Result{TRI_ERROR_NO_ERROR};
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
/// @brief restores the data of _users collection
|
|
/// Special case, we shall allow to import in this collection,
|
|
/// however it is not sharded by key and we need to replace by name instead of
|
|
/// by key
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
|
|
Result RestReplicationHandler::processRestoreUsersBatch(std::string const& collectionName) {
|
|
std::unordered_map<std::string, VPackValueLength> latest;
|
|
VPackBuilder allMarkers;
|
|
|
|
Result res = parseBatch(collectionName, latest, allMarkers);
|
|
if (res.fail()) {
|
|
return res;
|
|
}
|
|
|
|
VPackSlice allMarkersSlice = allMarkers.slice();
|
|
|
|
std::string aql(
|
|
"FOR u IN @restored UPSERT {user: u.user} INSERT u REPLACE u "
|
|
"INTO @@collection OPTIONS {ignoreErrors: true, silent: true, "
|
|
"waitForSync: false, isRestore: true}");
|
|
|
|
auto bindVars = std::make_shared<VPackBuilder>();
|
|
bindVars->openObject();
|
|
bindVars->add("@collection", VPackValue(collectionName));
|
|
bindVars->add(VPackValue("restored"));
|
|
bindVars->openArray();
|
|
|
|
// Now loop over the markers and insert the docs, ignoring _key and _id
|
|
for (auto const& p : latest) {
|
|
VPackSlice const marker = allMarkersSlice.at(p.second);
|
|
VPackSlice const typeSlice = marker.get(::typeString);
|
|
TRI_replication_operation_e type = REPLICATION_INVALID;
|
|
if (typeSlice.isNumber()) {
|
|
int typeInt = typeSlice.getNumericValue<int>();
|
|
if (typeInt == 2301) { // pre-3.0 type for edges
|
|
type = REPLICATION_MARKER_DOCUMENT;
|
|
} else {
|
|
type = static_cast<TRI_replication_operation_e>(typeInt);
|
|
}
|
|
}
|
|
if (type == REPLICATION_MARKER_DOCUMENT) {
|
|
VPackSlice const doc = marker.get(::dataString);
|
|
TRI_ASSERT(doc.isObject());
|
|
// In the _users case we silently remove the _key value.
|
|
bindVars->openObject();
|
|
for (auto const& it : VPackObjectIterator(doc)) {
|
|
if (arangodb::velocypack::StringRef(it.key) != StaticStrings::KeyString &&
|
|
arangodb::velocypack::StringRef(it.key) != StaticStrings::IdString) {
|
|
bindVars->add(it.key);
|
|
bindVars->add(it.value);
|
|
}
|
|
}
|
|
bindVars->close();
|
|
}
|
|
}
|
|
|
|
bindVars->close(); // restored
|
|
bindVars->close(); // bindVars
|
|
|
|
arangodb::aql::Query query(false, _vocbase, arangodb::aql::QueryString(aql),
|
|
bindVars, nullptr, arangodb::aql::PART_MAIN);
|
|
auto queryRegistry = QueryRegistryFeature::registry();
|
|
TRI_ASSERT(queryRegistry != nullptr);
|
|
|
|
aql::QueryResult queryResult = query.executeSync(queryRegistry);
|
|
|
|
// neither agency nor dbserver should get here
|
|
AuthenticationFeature* af = AuthenticationFeature::instance();
|
|
TRI_ASSERT(af->userManager() != nullptr);
|
|
if (af->userManager() != nullptr) {
|
|
af->userManager()->triggerCacheRevalidation();
|
|
}
|
|
|
|
return queryResult.result;
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
/// @brief restores the data of a collection
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
|
|
Result RestReplicationHandler::processRestoreDataBatch(transaction::Methods& trx,
|
|
std::string const& collectionName) {
|
|
std::unordered_map<std::string, VPackValueLength> latest;
|
|
VPackBuilder allMarkers;
|
|
|
|
Result res = parseBatch(collectionName, latest, allMarkers);
|
|
if (res.fail()) {
|
|
return res;
|
|
}
|
|
|
|
// First remove all keys of which the last marker we saw was a deletion
|
|
// marker:
|
|
VPackSlice allMarkersSlice = allMarkers.slice();
|
|
|
|
VPackBuilder oldBuilder;
|
|
{
|
|
VPackArrayBuilder guard(&oldBuilder);
|
|
|
|
for (auto const& p : latest) {
|
|
VPackSlice const marker = allMarkersSlice.at(p.second);
|
|
VPackSlice const typeSlice = marker.get(::typeString);
|
|
TRI_replication_operation_e type = REPLICATION_INVALID;
|
|
if (typeSlice.isNumber()) {
|
|
int typeInt = typeSlice.getNumericValue<int>();
|
|
if (typeInt == 2301) { // pre-3.0 type for edges
|
|
type = REPLICATION_MARKER_DOCUMENT;
|
|
} else {
|
|
type = static_cast<TRI_replication_operation_e>(typeInt);
|
|
}
|
|
}
|
|
if (type == REPLICATION_MARKER_REMOVE) {
|
|
oldBuilder.add(VPackValue(p.first)); // Add _key
|
|
} else if (type != REPLICATION_MARKER_DOCUMENT) {
|
|
return Result{TRI_ERROR_REPLICATION_UNEXPECTED_MARKER,
|
|
"unexpected marker type " + StringUtils::itoa(type)};
|
|
}
|
|
}
|
|
}
|
|
|
|
// Note that we ignore individual errors here, as long as the main
|
|
// operation did not fail. In particular, we intentionally ignore
|
|
// individual "DOCUMENT NOT FOUND" errors, because they can happen!
|
|
try {
|
|
OperationOptions options;
|
|
options.silent = true;
|
|
options.ignoreRevs = true;
|
|
options.isRestore = true;
|
|
options.waitForSync = false;
|
|
double startTime = TRI_microtime();
|
|
OperationResult opRes = trx.remove(collectionName, oldBuilder.slice(), options);
|
|
double duration = TRI_microtime() - startTime;
|
|
if (opRes.fail()) {
|
|
LOG_TOPIC("e86e3", WARN, Logger::CLUSTER)
|
|
<< "Could not delete " << oldBuilder.slice().length()
|
|
<< " documents for restore: " << opRes.result.errorMessage();
|
|
return opRes.result;
|
|
}
|
|
if (duration > 30) {
|
|
LOG_TOPIC("f4aed", INFO, Logger::PERFORMANCE)
|
|
<< "Restored/deleted " << oldBuilder.slice().length()
|
|
<< " documents in time: " << duration << " seconds.";
|
|
}
|
|
} catch (arangodb::basics::Exception const& ex) {
|
|
LOG_TOPIC("a046b", WARN, Logger::CLUSTER)
|
|
<< "Could not delete documents for restore exception: " << ex.what();
|
|
return Result(ex.code(), ex.what());
|
|
} catch (std::exception const& ex) {
|
|
LOG_TOPIC("677b7", WARN, Logger::CLUSTER)
|
|
<< "Could not delete documents for restore exception: " << ex.what();
|
|
return Result(TRI_ERROR_INTERNAL, ex.what());
|
|
} catch (...) {
|
|
LOG_TOPIC("734da", WARN, Logger::CLUSTER)
|
|
<< "Could not delete documents for restore exception.";
|
|
return Result(TRI_ERROR_INTERNAL);
|
|
}
|
|
|
|
bool const isUsersOnCoordinator = (ServerState::instance()->isCoordinator() &&
|
|
collectionName == TRI_COL_NAME_USERS);
|
|
|
|
// Now try to insert all keys for which the last marker was a document
|
|
// marker, note that these could still be replace markers!
|
|
VPackBuilder builder;
|
|
{
|
|
VPackArrayBuilder guard(&builder);
|
|
|
|
for (auto const& p : latest) {
|
|
VPackSlice const marker = allMarkersSlice.at(p.second);
|
|
VPackSlice const typeSlice = marker.get(::typeString);
|
|
TRI_replication_operation_e type = REPLICATION_INVALID;
|
|
if (typeSlice.isNumber()) {
|
|
int typeInt = typeSlice.getNumericValue<int>();
|
|
if (typeInt == 2301) { // pre-3.0 type for edges
|
|
type = REPLICATION_MARKER_DOCUMENT;
|
|
} else {
|
|
type = static_cast<TRI_replication_operation_e>(typeInt);
|
|
}
|
|
}
|
|
if (type == REPLICATION_MARKER_DOCUMENT) {
|
|
VPackSlice const doc = marker.get(::dataString);
|
|
TRI_ASSERT(doc.isObject());
|
|
if (isUsersOnCoordinator) {
|
|
// In the _users case we silently remove the _key value.
|
|
builder.openObject();
|
|
for (auto const& it : VPackObjectIterator(doc)) {
|
|
if (arangodb::velocypack::StringRef(it.key) != StaticStrings::KeyString &&
|
|
arangodb::velocypack::StringRef(it.key) != StaticStrings::IdString) {
|
|
builder.add(it.key);
|
|
builder.add(it.value);
|
|
}
|
|
}
|
|
builder.close();
|
|
} else {
|
|
builder.add(doc);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
VPackSlice requestSlice = builder.slice();
|
|
OperationResult opRes;
|
|
try {
|
|
OperationOptions options;
|
|
options.silent = false;
|
|
options.ignoreRevs = true;
|
|
options.isRestore = true;
|
|
options.waitForSync = false;
|
|
options.overwrite = true;
|
|
double startTime = TRI_microtime();
|
|
opRes = trx.insert(collectionName, requestSlice, options);
|
|
double duration = TRI_microtime() - startTime;
|
|
if (opRes.fail()) {
|
|
LOG_TOPIC("e6280", WARN, Logger::CLUSTER)
|
|
<< "Could not insert " << requestSlice.length()
|
|
<< " documents for restore: " << opRes.result.errorMessage();
|
|
return opRes.result;
|
|
}
|
|
if (duration > 30) {
|
|
LOG_TOPIC("38f0d", INFO, Logger::PERFORMANCE)
|
|
<< "Restored/inserted " << requestSlice.length()
|
|
<< " documents in time: " << duration << " seconds.";
|
|
}
|
|
} catch (arangodb::basics::Exception const& ex) {
|
|
LOG_TOPIC("8e8e1", WARN, Logger::CLUSTER)
|
|
<< "Could not insert documents for restore exception: " << ex.what();
|
|
return Result(ex.code(), ex.what());
|
|
} catch (std::exception const& ex) {
|
|
LOG_TOPIC("f1e7f", WARN, Logger::CLUSTER)
|
|
<< "Could not insert documents for restore exception: " << ex.what();
|
|
return Result(TRI_ERROR_INTERNAL, ex.what());
|
|
} catch (...) {
|
|
LOG_TOPIC("368bb", WARN, Logger::CLUSTER)
|
|
<< "Could not insert documents for restore exception.";
|
|
return Result(TRI_ERROR_INTERNAL);
|
|
}
|
|
|
|
// Now go through the individual results and check each error, if it was
|
|
// TRI_ERROR_ARANGO_UNIQUE_CONSTRAINT_VIOLATED, then we have to call
|
|
// replace on the document:
|
|
builder.clear(); // documents for replace operation
|
|
VPackSlice resultSlice = opRes.slice();
|
|
{
|
|
VPackArrayBuilder guard(&oldBuilder);
|
|
VPackArrayBuilder guard2(&builder);
|
|
VPackArrayIterator itRequest(requestSlice);
|
|
VPackArrayIterator itResult(resultSlice);
|
|
|
|
while (itRequest.valid()) {
|
|
VPackSlice result = *itResult;
|
|
VPackSlice error = result.get(StaticStrings::Error);
|
|
if (error.isTrue()) {
|
|
error = result.get(StaticStrings::ErrorNum);
|
|
if (error.isNumber()) {
|
|
int code = error.getNumericValue<int>();
|
|
if (code != TRI_ERROR_ARANGO_UNIQUE_CONSTRAINT_VIOLATED) {
|
|
return code;
|
|
}
|
|
builder.add(*itRequest);
|
|
} else {
|
|
return Result(TRI_ERROR_INTERNAL);
|
|
}
|
|
}
|
|
itRequest.next();
|
|
itResult.next();
|
|
}
|
|
}
|
|
|
|
if (builder.slice().length() == 0) {
|
|
// no replace operations to perform
|
|
return Result();
|
|
}
|
|
|
|
try {
|
|
OperationOptions options;
|
|
options.silent = true;
|
|
options.ignoreRevs = true;
|
|
options.isRestore = true;
|
|
options.waitForSync = false;
|
|
double startTime = TRI_microtime();
|
|
opRes = trx.replace(collectionName, builder.slice(), options);
|
|
double duration = TRI_microtime() - startTime;
|
|
if (opRes.fail()) {
|
|
LOG_TOPIC("c708e", WARN, Logger::CLUSTER)
|
|
<< "Could not replace " << builder.slice().length()
|
|
<< " documents for restore: " << opRes.result.errorMessage();
|
|
return opRes.result;
|
|
}
|
|
if (duration > 30) {
|
|
LOG_TOPIC("12b62", INFO, Logger::PERFORMANCE)
|
|
<< "Restored/replaced " << builder.slice().length()
|
|
<< " documents in time: " << duration << " seconds.";
|
|
}
|
|
} catch (arangodb::basics::Exception const& ex) {
|
|
LOG_TOPIC("37b43", WARN, Logger::CLUSTER)
|
|
<< "Could not replace documents for restore exception: " << ex.what();
|
|
return Result(ex.code(), ex.what());
|
|
} catch (std::exception const& ex) {
|
|
LOG_TOPIC("33217", WARN, Logger::CLUSTER)
|
|
<< "Could not replace documents for restore exception: " << ex.what();
|
|
return Result(TRI_ERROR_INTERNAL, ex.what());
|
|
} catch (...) {
|
|
LOG_TOPIC("5f400", WARN, Logger::CLUSTER)
|
|
<< "Could not replace documents for restore exception.";
|
|
return Result(TRI_ERROR_INTERNAL);
|
|
}
|
|
|
|
return Result();
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
/// @brief restores the indexes of a collection
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
|
|
Result RestReplicationHandler::processRestoreIndexes(VPackSlice const& collection, bool force) {
|
|
TRI_ASSERT(!ServerState::instance()->isCoordinator());
|
|
if (!collection.isObject()) {
|
|
std::string errorMsg = "collection declaration is invalid";
|
|
return {TRI_ERROR_HTTP_BAD_PARAMETER, errorMsg};
|
|
}
|
|
|
|
VPackSlice const parameters = collection.get("parameters");
|
|
|
|
if (!parameters.isObject()) {
|
|
std::string errorMsg = "collection parameters declaration is invalid";
|
|
return {TRI_ERROR_HTTP_BAD_PARAMETER, errorMsg};
|
|
}
|
|
|
|
VPackSlice const indexes = collection.get("indexes");
|
|
|
|
if (!indexes.isArray()) {
|
|
std::string errorMsg = "collection indexes declaration is invalid";
|
|
return {TRI_ERROR_HTTP_BAD_PARAMETER, errorMsg};
|
|
}
|
|
|
|
VPackValueLength const n = indexes.length();
|
|
|
|
if (n == 0) {
|
|
// nothing to do
|
|
return TRI_ERROR_NO_ERROR;
|
|
}
|
|
|
|
std::string const name =
|
|
arangodb::basics::VelocyPackHelper::getStringValue(parameters, "name", "");
|
|
|
|
if (name.empty()) {
|
|
std::string errorMsg = "collection name is missing";
|
|
return {TRI_ERROR_HTTP_BAD_PARAMETER, errorMsg};
|
|
}
|
|
|
|
if (arangodb::basics::VelocyPackHelper::getBooleanValue(parameters, "deleted", false)) {
|
|
// we don't care about deleted collections
|
|
return {};
|
|
}
|
|
|
|
std::shared_ptr<LogicalCollection> coll = _vocbase.lookupCollection(name);
|
|
if (!coll) {
|
|
return Result(TRI_ERROR_ARANGO_DATA_SOURCE_NOT_FOUND);
|
|
}
|
|
|
|
Result fres;
|
|
|
|
grantTemporaryRights();
|
|
READ_LOCKER(readLocker, _vocbase._inventoryLock);
|
|
|
|
// look up the collection
|
|
try {
|
|
auto physical = coll->getPhysical();
|
|
TRI_ASSERT(physical != nullptr);
|
|
|
|
for (VPackSlice const& idxDef : VPackArrayIterator(indexes)) {
|
|
// {"id":"229907440927234","type":"hash","unique":false,"fields":["x","Y"]}
|
|
arangodb::velocypack::Slice value = idxDef.get(StaticStrings::IndexType);
|
|
if (value.isString()) {
|
|
std::string const typeString = value.copyString();
|
|
if ((typeString == "primary") || (typeString == "edge")) {
|
|
LOG_TOPIC("0d352", DEBUG, Logger::REPLICATION)
|
|
<< "processRestoreIndexes silently ignoring primary or edge "
|
|
<< "index: " << idxDef.toJson();
|
|
continue;
|
|
}
|
|
}
|
|
|
|
std::shared_ptr<arangodb::Index> idx;
|
|
try {
|
|
bool created = false;
|
|
idx = physical->createIndex(idxDef, /*restore*/ true, created);
|
|
} catch (basics::Exception const& e) {
|
|
if (e.code() == TRI_ERROR_NOT_IMPLEMENTED) {
|
|
continue;
|
|
}
|
|
|
|
std::string errorMsg = "could not create index: " + e.message();
|
|
fres.reset(e.code(), errorMsg);
|
|
break;
|
|
}
|
|
TRI_ASSERT(idx != nullptr);
|
|
}
|
|
} catch (arangodb::basics::Exception const& ex) {
|
|
// fix error handling
|
|
std::string errorMsg =
|
|
"could not create index: " + std::string(TRI_errno_string(ex.code()));
|
|
fres.reset(ex.code(), errorMsg);
|
|
} catch (...) {
|
|
std::string errorMsg = "could not create index: unknown error";
|
|
fres.reset(TRI_ERROR_INTERNAL, errorMsg);
|
|
}
|
|
|
|
return fres;
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
/// @brief restores the indexes of a collection, coordinator case
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
|
|
Result RestReplicationHandler::processRestoreIndexesCoordinator(VPackSlice const& collection,
|
|
bool force) {
|
|
if (!collection.isObject()) {
|
|
std::string errorMsg = "collection declaration is invalid";
|
|
return {TRI_ERROR_HTTP_BAD_PARAMETER, errorMsg};
|
|
}
|
|
VPackSlice const parameters = collection.get("parameters");
|
|
|
|
if (!parameters.isObject()) {
|
|
std::string errorMsg = "collection parameters declaration is invalid";
|
|
return {TRI_ERROR_HTTP_BAD_PARAMETER, errorMsg};
|
|
}
|
|
|
|
VPackSlice indexes = collection.get("indexes");
|
|
|
|
if (!indexes.isArray()) {
|
|
std::string errorMsg = "collection indexes declaration is invalid";
|
|
return {TRI_ERROR_HTTP_BAD_PARAMETER, errorMsg};
|
|
}
|
|
|
|
size_t const n = static_cast<size_t>(indexes.length());
|
|
|
|
if (n == 0) {
|
|
// nothing to do
|
|
return {};
|
|
}
|
|
|
|
std::string name =
|
|
arangodb::basics::VelocyPackHelper::getStringValue(parameters, "name", "");
|
|
|
|
if (name.empty()) {
|
|
std::string errorMsg = "collection indexes declaration is invalid";
|
|
return {TRI_ERROR_HTTP_BAD_PARAMETER, errorMsg};
|
|
}
|
|
|
|
if (ignoreHiddenEnterpriseCollection(name, force)) {
|
|
return {};
|
|
}
|
|
|
|
if (arangodb::basics::VelocyPackHelper::getBooleanValue(parameters, "deleted", false)) {
|
|
// we don't care about deleted collections
|
|
return {};
|
|
}
|
|
|
|
auto& dbName = _vocbase.name();
|
|
|
|
// in a cluster, we only look up by name:
|
|
ClusterInfo* ci = ClusterInfo::instance();
|
|
std::shared_ptr<LogicalCollection> col = ci->getCollectionNT(dbName, name);
|
|
if (col == nullptr) {
|
|
return {TRI_ERROR_ARANGO_DATA_SOURCE_NOT_FOUND,
|
|
ClusterInfo::getCollectionNotFoundMsg(dbName, name)};
|
|
}
|
|
|
|
TRI_ASSERT(col != nullptr);
|
|
|
|
auto cluster = application_features::ApplicationServer::getFeature<ClusterFeature>(
|
|
"Cluster");
|
|
|
|
Result res;
|
|
|
|
for (VPackSlice const& idxDef : VPackArrayIterator(indexes)) {
|
|
VPackSlice type = idxDef.get(StaticStrings::IndexType);
|
|
|
|
if (type.isString() && (type.copyString() == "primary" || type.copyString() == "edge")) {
|
|
// must ignore these types of indexes during restore
|
|
continue;
|
|
}
|
|
|
|
VPackBuilder tmp;
|
|
|
|
res = ci->ensureIndexCoordinator(*col, idxDef, true, tmp,
|
|
cluster->indexCreationTimeout());
|
|
|
|
if (res.fail()) {
|
|
return res.reset(res.errorNumber(), "could not create index: " + res.errorMessage());
|
|
}
|
|
}
|
|
|
|
return res;
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
/// @brief restores the views
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
|
|
void RestReplicationHandler::handleCommandRestoreView() {
|
|
bool parseSuccess = false;
|
|
VPackSlice slice = this->parseVPackBody(parseSuccess);
|
|
|
|
if (!parseSuccess) {
|
|
return; // error message generated in parseVPackBody
|
|
}
|
|
|
|
if (!slice.isObject()) {
|
|
generateError(ResponseCode::BAD, TRI_ERROR_BAD_PARAMETER);
|
|
|
|
return;
|
|
}
|
|
|
|
bool overwrite = _request->parsedValue<bool>("overwrite", false);
|
|
auto nameSlice = slice.get(StaticStrings::DataSourceName);
|
|
auto typeSlice = slice.get(StaticStrings::DataSourceType);
|
|
|
|
if (!nameSlice.isString() || !typeSlice.isString()) {
|
|
generateError(ResponseCode::BAD, TRI_ERROR_BAD_PARAMETER);
|
|
return;
|
|
}
|
|
|
|
LOG_TOPIC("f874e", TRACE, Logger::REPLICATION)
|
|
<< "restoring view: " << nameSlice.copyString();
|
|
|
|
try {
|
|
CollectionNameResolver resolver(_vocbase);
|
|
auto view = resolver.getView(nameSlice.copyString());
|
|
|
|
if (view) {
|
|
if (!overwrite) {
|
|
generateError(GeneralResponse::responseCode(TRI_ERROR_ARANGO_DUPLICATE_NAME),
|
|
TRI_ERROR_ARANGO_DUPLICATE_NAME,
|
|
std::string("unable to restore view '") + nameSlice.copyString() + ": " +
|
|
TRI_errno_string(TRI_ERROR_ARANGO_DUPLICATE_NAME));
|
|
return;
|
|
}
|
|
|
|
auto res = view->drop();
|
|
|
|
if (!res.ok()) {
|
|
generateError(res);
|
|
return;
|
|
}
|
|
}
|
|
|
|
auto res = LogicalView::create(view, _vocbase, slice); // must create() since view was drop()ed
|
|
|
|
if (!res.ok()) {
|
|
generateError(res);
|
|
|
|
return;
|
|
}
|
|
|
|
if (view == nullptr) {
|
|
generateError(rest::ResponseCode::SERVER_ERROR, TRI_ERROR_INTERNAL,
|
|
"problem creating view");
|
|
|
|
return;
|
|
}
|
|
} catch (basics::Exception const& ex) {
|
|
generateError(GeneralResponse::responseCode(ex.code()), ex.code(), ex.message());
|
|
|
|
return;
|
|
} catch (...) {
|
|
generateError(rest::ResponseCode::SERVER_ERROR, TRI_ERROR_INTERNAL,
|
|
"problem creating view");
|
|
|
|
return;
|
|
}
|
|
|
|
VPackBuilder result;
|
|
|
|
result.openObject();
|
|
result.add("result", VPackValue(true));
|
|
result.close();
|
|
generateResult(rest::ResponseCode::OK, result.slice());
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
/// @brief was docuBlock JSF_put_api_replication_serverID
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
|
|
void RestReplicationHandler::handleCommandServerId() {
|
|
VPackBuilder result;
|
|
result.add(VPackValue(VPackValueType::Object));
|
|
std::string const serverId = StringUtils::itoa(ServerIdFeature::getId());
|
|
result.add("serverId", VPackValue(serverId));
|
|
result.close();
|
|
generateResult(rest::ResponseCode::OK, result.slice());
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
/// @brief was docuBlock JSF_put_api_replication_synchronize
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
|
|
void RestReplicationHandler::handleCommandSync() {
|
|
bool isGlobal;
|
|
ReplicationApplier* applier = getApplier(isGlobal);
|
|
if (applier == nullptr) {
|
|
return;
|
|
}
|
|
|
|
bool success = false;
|
|
VPackSlice const body = this->parseVPackBody(success);
|
|
if (!success) {
|
|
// error already created
|
|
return;
|
|
}
|
|
|
|
std::string const endpoint =
|
|
VelocyPackHelper::getStringValue(body, "endpoint", "");
|
|
if (endpoint.empty()) {
|
|
generateError(rest::ResponseCode::BAD, TRI_ERROR_HTTP_BAD_PARAMETER,
|
|
"<endpoint> must be a valid endpoint");
|
|
return;
|
|
}
|
|
|
|
std::string dbname = isGlobal ? "" : _vocbase.name();
|
|
auto config = ReplicationApplierConfiguration::fromVelocyPack(body, dbname);
|
|
|
|
// will throw if invalid
|
|
config.validate();
|
|
std::shared_ptr<InitialSyncer> syncer;
|
|
|
|
if (isGlobal) {
|
|
syncer.reset(new GlobalInitialSyncer(config));
|
|
} else {
|
|
syncer.reset(new DatabaseInitialSyncer(_vocbase, config));
|
|
}
|
|
|
|
Result r = syncer->run(config._incremental);
|
|
|
|
if (r.fail()) {
|
|
LOG_TOPIC("c4818", ERR, Logger::REPLICATION)
|
|
<< "failed to sync: " << r.errorMessage();
|
|
generateError(r);
|
|
return;
|
|
}
|
|
|
|
// FIXME: change response for databases
|
|
VPackBuilder result;
|
|
result.add(VPackValue(VPackValueType::Object));
|
|
result.add("collections", VPackValue(VPackValueType::Array));
|
|
for (auto const& it : syncer->getProcessedCollections()) {
|
|
std::string const cidString = StringUtils::itoa(it.first);
|
|
// Insert a collection
|
|
result.add(VPackValue(VPackValueType::Object));
|
|
result.add("id", VPackValue(cidString));
|
|
result.add("name", VPackValue(it.second));
|
|
result.close(); // one collection
|
|
}
|
|
result.close(); // collections
|
|
|
|
auto tickString = std::to_string(syncer->getLastLogTick());
|
|
result.add("lastLogTick", VPackValue(tickString));
|
|
|
|
bool const keepBarrier = VelocyPackHelper::getBooleanValue(body, "keepBarrier", false);
|
|
if (keepBarrier) {
|
|
auto barrierId = std::to_string(syncer->stealBarrier());
|
|
result.add("barrierId", VPackValue(barrierId));
|
|
}
|
|
|
|
result.close(); // base
|
|
generateResult(rest::ResponseCode::OK, result.slice());
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
/// @brief was docuBlock JSF_put_api_replication_applier
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
|
|
void RestReplicationHandler::handleCommandApplierGetConfig() {
|
|
bool isGlobal;
|
|
ReplicationApplier* applier = getApplier(isGlobal);
|
|
if (applier == nullptr) {
|
|
return;
|
|
}
|
|
|
|
auto configuration = applier->configuration();
|
|
VPackBuilder builder;
|
|
builder.openObject();
|
|
configuration.toVelocyPack(builder, false, false);
|
|
builder.close();
|
|
|
|
generateResult(rest::ResponseCode::OK, builder.slice());
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
/// @brief was docuBlock JSF_put_api_replication_applier_adjust
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
|
|
void RestReplicationHandler::handleCommandApplierSetConfig() {
|
|
bool isGlobal;
|
|
ReplicationApplier* applier = getApplier(isGlobal);
|
|
if (applier == nullptr) {
|
|
return;
|
|
}
|
|
|
|
bool success = false;
|
|
VPackSlice const body = this->parseVPackBody(success);
|
|
if (!success) {
|
|
// error already created
|
|
return;
|
|
}
|
|
|
|
std::string databaseName;
|
|
|
|
if (!isGlobal) {
|
|
databaseName = _vocbase.name();
|
|
}
|
|
|
|
auto config =
|
|
ReplicationApplierConfiguration::fromVelocyPack(applier->configuration(),
|
|
body, databaseName);
|
|
// will throw if invalid
|
|
config.validate();
|
|
|
|
applier->reconfigure(config);
|
|
handleCommandApplierGetConfig();
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
/// @brief was docuBlock JSF_put_api_replication_applier_start
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
|
|
void RestReplicationHandler::handleCommandApplierStart() {
|
|
bool isGlobal;
|
|
ReplicationApplier* applier = getApplier(isGlobal);
|
|
if (applier == nullptr) {
|
|
return;
|
|
}
|
|
|
|
bool found;
|
|
std::string const& value1 = _request->value("from", found);
|
|
|
|
TRI_voc_tick_t initialTick = 0;
|
|
bool useTick = false;
|
|
|
|
if (found) {
|
|
// query parameter "from" specified
|
|
initialTick = static_cast<TRI_voc_tick_t>(StringUtils::uint64(value1));
|
|
useTick = true;
|
|
}
|
|
|
|
TRI_voc_tick_t barrierId = 0;
|
|
std::string const& value2 = _request->value("barrierId", found);
|
|
if (found) {
|
|
// query parameter "barrierId" specified
|
|
barrierId = static_cast<TRI_voc_tick_t>(StringUtils::uint64(value2));
|
|
}
|
|
|
|
applier->startTailing(initialTick, useTick, barrierId);
|
|
handleCommandApplierGetState();
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
/// @brief was docuBlock JSF_put_api_replication_applier_stop
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
|
|
void RestReplicationHandler::handleCommandApplierStop() {
|
|
bool isGlobal;
|
|
ReplicationApplier* applier = getApplier(isGlobal);
|
|
if (applier == nullptr) {
|
|
return;
|
|
}
|
|
|
|
applier->stopAndJoin();
|
|
handleCommandApplierGetState();
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
/// @brief was docuBlock JSF_get_api_replication_applier_state
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
|
|
void RestReplicationHandler::handleCommandApplierGetState() {
|
|
bool isGlobal;
|
|
ReplicationApplier* applier = getApplier(isGlobal);
|
|
if (applier == nullptr) {
|
|
return;
|
|
}
|
|
|
|
VPackBuilder builder;
|
|
builder.openObject();
|
|
applier->toVelocyPack(builder);
|
|
builder.close();
|
|
generateResult(rest::ResponseCode::OK, builder.slice());
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
/// @brief was docuBlock JSF_get_api_replication_applier_state_all
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
|
|
void RestReplicationHandler::handleCommandApplierGetStateAll() {
|
|
if (_request->databaseName() != StaticStrings::SystemDatabase) {
|
|
generateError(
|
|
rest::ResponseCode::FORBIDDEN, TRI_ERROR_FORBIDDEN,
|
|
"global inventory can only be fetched from within _system database");
|
|
return;
|
|
}
|
|
DatabaseFeature* databaseFeature =
|
|
application_features::ApplicationServer::getFeature<DatabaseFeature>(
|
|
"Database");
|
|
|
|
VPackBuilder builder;
|
|
builder.openObject();
|
|
for (auto& name : databaseFeature->getDatabaseNames()) {
|
|
TRI_vocbase_t* vocbase = databaseFeature->lookupDatabase(name);
|
|
|
|
if (vocbase == nullptr) {
|
|
continue;
|
|
}
|
|
|
|
ReplicationApplier* applier = vocbase->replicationApplier();
|
|
|
|
if (applier == nullptr) {
|
|
continue;
|
|
}
|
|
|
|
builder.add(name, VPackValue(VPackValueType::Object));
|
|
applier->toVelocyPack(builder);
|
|
builder.close();
|
|
}
|
|
builder.close();
|
|
|
|
generateResult(rest::ResponseCode::OK, builder.slice());
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
/// @brief delete the state of the replication applier
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
|
|
void RestReplicationHandler::handleCommandApplierDeleteState() {
|
|
bool isGlobal;
|
|
ReplicationApplier* applier = getApplier(isGlobal);
|
|
if (applier == nullptr) {
|
|
return;
|
|
}
|
|
|
|
applier->forget();
|
|
|
|
handleCommandApplierGetState();
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
/// @brief add a follower of a shard to the list of followers
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
|
|
void RestReplicationHandler::handleCommandAddFollower() {
|
|
TRI_ASSERT(ServerState::instance()->isDBServer());
|
|
|
|
bool success = false;
|
|
VPackSlice const body = this->parseVPackBody(success);
|
|
if (!success) {
|
|
// error already created
|
|
return;
|
|
}
|
|
if (!body.isObject()) {
|
|
generateError(rest::ResponseCode::BAD, TRI_ERROR_HTTP_BAD_PARAMETER,
|
|
"body needs to be an object with attributes 'followerId' "
|
|
"and 'shard'");
|
|
return;
|
|
}
|
|
VPackSlice const followerIdSlice = body.get("followerId");
|
|
VPackSlice const readLockIdSlice = body.get("readLockId");
|
|
VPackSlice const shardSlice = body.get("shard");
|
|
VPackSlice const checksumSlice = body.get("checksum");
|
|
if (!followerIdSlice.isString() || !shardSlice.isString() || !checksumSlice.isString()) {
|
|
generateError(
|
|
rest::ResponseCode::BAD, TRI_ERROR_HTTP_BAD_PARAMETER,
|
|
"'followerId', 'shard' and 'checksum' attributes must be strings");
|
|
return;
|
|
}
|
|
|
|
auto col = _vocbase.lookupCollection(shardSlice.copyString());
|
|
if (col == nullptr) {
|
|
generateError(rest::ResponseCode::SERVER_ERROR, TRI_ERROR_ARANGO_DATA_SOURCE_NOT_FOUND,
|
|
"did not find collection");
|
|
return;
|
|
}
|
|
|
|
const std::string followerId = followerIdSlice.copyString();
|
|
LOG_TOPIC("312cc", DEBUG, Logger::REPLICATION)
|
|
<< "Attempt to Add Follower: " << followerId << " to shard "
|
|
<< col->name() << " in database: " << _vocbase.name();
|
|
// Short cut for the case that the collection is empty
|
|
if (readLockIdSlice.isNone()) {
|
|
LOG_TOPIC("aaff2", DEBUG, Logger::REPLICATION)
|
|
<< "Try add follower fast-path (no documents)";
|
|
auto ctx = transaction::StandaloneContext::Create(_vocbase);
|
|
SingleCollectionTransaction trx(ctx, *col, AccessMode::Type::EXCLUSIVE);
|
|
auto res = trx.begin();
|
|
|
|
if (res.ok()) {
|
|
auto countRes = trx.count(col->name(), transaction::CountType::Normal);
|
|
|
|
if (countRes.ok()) {
|
|
VPackSlice nrSlice = countRes.slice();
|
|
uint64_t nr = nrSlice.getNumber<uint64_t>();
|
|
LOG_TOPIC("533c3", DEBUG, Logger::REPLICATION)
|
|
<< "Compare with shortCut Leader: " << nr
|
|
<< " == Follower: " << checksumSlice.copyString();
|
|
if (nr == 0 && checksumSlice.isEqualString("0")) {
|
|
Result res = col->followers()->add(followerId);
|
|
|
|
if (res.fail()) {
|
|
// this will create an error response with the appropriate message
|
|
THROW_ARANGO_EXCEPTION(res);
|
|
}
|
|
|
|
VPackBuilder b;
|
|
{
|
|
VPackObjectBuilder bb(&b);
|
|
b.add(StaticStrings::Error, VPackValue(false));
|
|
}
|
|
|
|
generateResult(rest::ResponseCode::OK, b.slice());
|
|
LOG_TOPIC("c316e", DEBUG, Logger::REPLICATION)
|
|
<< followerId << " is now following on shard " << _vocbase.name()
|
|
<< "/" << col->name();
|
|
return;
|
|
}
|
|
}
|
|
}
|
|
// If we get here, we have to report an error:
|
|
generateError(rest::ResponseCode::FORBIDDEN,
|
|
TRI_ERROR_REPLICATION_SHARD_NONEMPTY, "shard not empty");
|
|
LOG_TOPIC("8bf6e", DEBUG, Logger::REPLICATION)
|
|
<< followerId << " is not yet in sync with " << _vocbase.name() << "/"
|
|
<< col->name();
|
|
return;
|
|
}
|
|
|
|
if (!readLockIdSlice.isString() || readLockIdSlice.getStringLength() == 0) {
|
|
generateError(rest::ResponseCode::BAD, TRI_ERROR_HTTP_BAD_PARAMETER,
|
|
"'readLockId' is not a string or empty");
|
|
return;
|
|
}
|
|
LOG_TOPIC("23b9f", DEBUG, Logger::REPLICATION)
|
|
<< "Try add follower with documents";
|
|
// previous versions < 3.3x might not send the checksum, if mixed clusters
|
|
// get into trouble here we may need to be more lenient
|
|
TRI_ASSERT(checksumSlice.isString() && readLockIdSlice.isString());
|
|
|
|
aql::QueryId readLockId = ExtractReadlockId(readLockIdSlice);
|
|
|
|
// referenceChecksum is the stringified number of documents in the collection
|
|
ResultT<std::string> referenceChecksum =
|
|
computeCollectionChecksum(readLockId, col.get());
|
|
if (!referenceChecksum.ok()) {
|
|
generateError(std::move(referenceChecksum).result());
|
|
return;
|
|
}
|
|
|
|
LOG_TOPIC("40b17", DEBUG, Logger::REPLICATION)
|
|
<< "Compare Leader: " << referenceChecksum.get()
|
|
<< " == Follower: " << checksumSlice.copyString();
|
|
if (!checksumSlice.isEqualString(referenceChecksum.get())) {
|
|
LOG_TOPIC("94ebe", DEBUG, Logger::REPLICATION)
|
|
<< followerId << " is not yet in sync with " << _vocbase.name() << "/"
|
|
<< col->name();
|
|
const std::string checksum = checksumSlice.copyString();
|
|
LOG_TOPIC("592ef", WARN, Logger::REPLICATION)
|
|
<< "Cannot add follower, mismatching checksums. "
|
|
<< "Expected: " << referenceChecksum.get() << " Actual: " << checksum;
|
|
generateError(rest::ResponseCode::BAD, TRI_ERROR_REPLICATION_WRONG_CHECKSUM,
|
|
"'checksum' is wrong. Expected: " + referenceChecksum.get() +
|
|
". Actual: " + checksum);
|
|
return;
|
|
}
|
|
|
|
Result res = col->followers()->add(followerId);
|
|
|
|
if (res.fail()) {
|
|
// this will create an error response with the appropriate message
|
|
THROW_ARANGO_EXCEPTION(res);
|
|
}
|
|
|
|
{ // untrack the (async) replication client, so the WAL may be cleaned
|
|
TRI_server_id_t const serverId = StringUtils::uint64(
|
|
basics::VelocyPackHelper::getStringValue(body, "serverId", ""));
|
|
SyncerId const syncerId = SyncerId{StringUtils::uint64(
|
|
basics::VelocyPackHelper::getStringValue(body, "syncerId", ""))};
|
|
std::string const clientInfo =
|
|
basics::VelocyPackHelper::getStringValue(body, "clientInfo", "");
|
|
|
|
_vocbase.replicationClients().untrack(SyncerId{syncerId}, serverId, clientInfo);
|
|
}
|
|
|
|
VPackBuilder b;
|
|
{
|
|
VPackObjectBuilder bb(&b);
|
|
b.add(StaticStrings::Error, VPackValue(false));
|
|
}
|
|
|
|
LOG_TOPIC("c13d4", DEBUG, Logger::REPLICATION)
|
|
<< followerId << " is now following on shard " << _vocbase.name() << "/"
|
|
<< col->name();
|
|
generateResult(rest::ResponseCode::OK, b.slice());
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
/// @brief remove a follower of a shard from the list of followers
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
|
|
void RestReplicationHandler::handleCommandRemoveFollower() {
|
|
TRI_ASSERT(ServerState::instance()->isDBServer());
|
|
|
|
bool success = false;
|
|
VPackSlice const body = this->parseVPackBody(success);
|
|
if (!success) {
|
|
// error already created
|
|
return;
|
|
}
|
|
if (!body.isObject()) {
|
|
generateError(rest::ResponseCode::BAD, TRI_ERROR_HTTP_BAD_PARAMETER,
|
|
"body needs to be an object with attributes 'followerId' "
|
|
"and 'shard'");
|
|
return;
|
|
}
|
|
VPackSlice const followerId = body.get("followerId");
|
|
VPackSlice const shard = body.get("shard");
|
|
if (!followerId.isString() || !shard.isString()) {
|
|
generateError(rest::ResponseCode::BAD, TRI_ERROR_HTTP_BAD_PARAMETER,
|
|
"'followerId' and 'shard' attributes must be strings");
|
|
return;
|
|
}
|
|
|
|
auto col = _vocbase.lookupCollection(shard.copyString());
|
|
|
|
if (col == nullptr) {
|
|
generateError(rest::ResponseCode::SERVER_ERROR, TRI_ERROR_ARANGO_DATA_SOURCE_NOT_FOUND,
|
|
"did not find collection");
|
|
return;
|
|
}
|
|
|
|
Result res = col->followers()->remove(followerId.copyString());
|
|
|
|
if (res.fail()) {
|
|
// this will create an error response with the appropriate message
|
|
THROW_ARANGO_EXCEPTION(res);
|
|
}
|
|
|
|
VPackBuilder b;
|
|
{
|
|
VPackObjectBuilder bb(&b);
|
|
b.add(StaticStrings::Error, VPackValue(false));
|
|
}
|
|
|
|
generateResult(rest::ResponseCode::OK, b.slice());
|
|
}
|
|
|
|
//////////////////////////////////////////////////////////////////////////////
|
|
/// @brief update the leader of a shard
|
|
//////////////////////////////////////////////////////////////////////////////
|
|
|
|
void RestReplicationHandler::handleCommandSetTheLeader() {
|
|
TRI_ASSERT(ServerState::instance()->isDBServer());
|
|
|
|
bool success = false;
|
|
VPackSlice const body = this->parseVPackBody(success);
|
|
if (!success) {
|
|
// error already created
|
|
return;
|
|
}
|
|
if (!body.isObject()) {
|
|
generateError(rest::ResponseCode::BAD, TRI_ERROR_HTTP_BAD_PARAMETER,
|
|
"body needs to be an object with attributes 'leaderId' "
|
|
"and 'shard'");
|
|
return;
|
|
}
|
|
VPackSlice const leaderIdSlice = body.get("leaderId");
|
|
VPackSlice const oldLeaderIdSlice = body.get("oldLeaderId");
|
|
VPackSlice const shard = body.get("shard");
|
|
if (!leaderIdSlice.isString() || !shard.isString() || !oldLeaderIdSlice.isString()) {
|
|
generateError(rest::ResponseCode::BAD, TRI_ERROR_HTTP_BAD_PARAMETER,
|
|
"'leaderId' and 'shard' attributes must be strings");
|
|
return;
|
|
}
|
|
|
|
std::string leaderId = leaderIdSlice.copyString();
|
|
auto col = _vocbase.lookupCollection(shard.copyString());
|
|
|
|
if (col == nullptr) {
|
|
generateError(rest::ResponseCode::SERVER_ERROR, TRI_ERROR_ARANGO_DATA_SOURCE_NOT_FOUND,
|
|
"did not find collection");
|
|
return;
|
|
}
|
|
|
|
Result res = checkPlanLeaderDirect(col, leaderId);
|
|
if (res.fail()) {
|
|
THROW_ARANGO_EXCEPTION(res);
|
|
}
|
|
|
|
std::string currentLeader = col->followers()->getLeader();
|
|
if (currentLeader == arangodb::maintenance::ResignShardLeadership::LeaderNotYetKnownString) {
|
|
// We have resigned, check that we are the old leader
|
|
currentLeader = ServerState::instance()->getId();
|
|
}
|
|
|
|
if (!oldLeaderIdSlice.isEqualString(currentLeader)) {
|
|
generateError(rest::ResponseCode::FORBIDDEN, TRI_ERROR_FORBIDDEN,
|
|
"old leader not as expected");
|
|
return;
|
|
}
|
|
|
|
col->followers()->setTheLeader(leaderId);
|
|
VPackBuilder b;
|
|
{
|
|
VPackObjectBuilder bb(&b);
|
|
b.add(StaticStrings::Error, VPackValue(false));
|
|
}
|
|
|
|
generateResult(rest::ResponseCode::OK, b.slice());
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
/// @brief hold a read lock on a collection to stop writes temporarily
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
|
|
void RestReplicationHandler::handleCommandHoldReadLockCollection() {
|
|
TRI_ASSERT(ServerState::instance()->isDBServer());
|
|
bool success = false;
|
|
VPackSlice const body = this->parseVPackBody(success);
|
|
if (!success) {
|
|
// error already created
|
|
return;
|
|
}
|
|
|
|
if (!body.isObject()) {
|
|
generateError(rest::ResponseCode::BAD, TRI_ERROR_HTTP_BAD_PARAMETER,
|
|
"body needs to be an object with attributes 'collection', "
|
|
"'ttl' and 'id'");
|
|
return;
|
|
}
|
|
|
|
VPackSlice const collection = body.get("collection");
|
|
VPackSlice const ttlSlice = body.get("ttl");
|
|
VPackSlice const idSlice = body.get("id");
|
|
|
|
if (!collection.isString() || !ttlSlice.isNumber() || !idSlice.isString()) {
|
|
generateError(rest::ResponseCode::BAD, TRI_ERROR_HTTP_BAD_PARAMETER,
|
|
"'collection' must be a string and 'ttl' a number and "
|
|
"'id' a string");
|
|
return;
|
|
}
|
|
|
|
aql::QueryId id = ExtractReadlockId(idSlice);
|
|
auto col = _vocbase.lookupCollection(collection.copyString());
|
|
|
|
if (col == nullptr) {
|
|
generateError(rest::ResponseCode::SERVER_ERROR, TRI_ERROR_ARANGO_DATA_SOURCE_NOT_FOUND,
|
|
"did not find collection");
|
|
return;
|
|
}
|
|
|
|
// 0.0 means using the default timeout (whatever that is)
|
|
double ttl = VelocyPackHelper::getNumericValue(ttlSlice, 0.0);
|
|
|
|
if (col->getStatusLocked() != TRI_VOC_COL_STATUS_LOADED) {
|
|
generateError(rest::ResponseCode::SERVER_ERROR, TRI_ERROR_ARANGO_COLLECTION_NOT_LOADED,
|
|
"collection not loaded");
|
|
return;
|
|
}
|
|
|
|
// This is an optional parameter, it may not be set (backwards compatible)
|
|
// If it is not set it will default to a hard-lock, otherwise we do a
|
|
// potentially faster soft-lock synchronization with a smaller hard-lock
|
|
// phase.
|
|
|
|
bool doSoftLock = VelocyPackHelper::getBooleanValue(body, "doSoftLockOnly", false);
|
|
AccessMode::Type lockType = AccessMode::Type::READ;
|
|
if (!doSoftLock && EngineSelectorFeature::ENGINE->typeName() == "rocksdb") {
|
|
// With not doSoftLock we trigger RocksDB to stop writes on this shard.
|
|
// With a softLock we only stop the WAL from being collected,
|
|
// but still allow writes.
|
|
// This has potential to never ever finish, so we need a short
|
|
// hard lock for the final sync.
|
|
lockType = AccessMode::Type::EXCLUSIVE;
|
|
}
|
|
|
|
LOG_TOPIC("4fac2", DEBUG, Logger::REPLICATION)
|
|
<< "Attempt to create a Lock: " << id << " for shard: " << _vocbase.name()
|
|
<< "/" << col->name() << " of type: " << (doSoftLock ? "soft" : "hard");
|
|
Result res = createBlockingTransaction(id, *col, ttl, lockType);
|
|
if (!res.ok()) {
|
|
generateError(res);
|
|
return;
|
|
}
|
|
|
|
TRI_ASSERT(isLockHeld(id).ok());
|
|
TRI_ASSERT(isLockHeld(id).get() == true);
|
|
|
|
VPackBuilder b;
|
|
{
|
|
VPackObjectBuilder bb(&b);
|
|
b.add(StaticStrings::Error, VPackValue(false));
|
|
}
|
|
|
|
LOG_TOPIC("61a9d", DEBUG, Logger::REPLICATION)
|
|
<< "Shard: " << _vocbase.name() << "/" << col->name()
|
|
<< " is now locked with type: " << (doSoftLock ? "soft" : "hard")
|
|
<< " lock id: " << id;
|
|
generateResult(rest::ResponseCode::OK, b.slice());
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
/// @brief check the holding of a read lock on a collection
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
|
|
void RestReplicationHandler::handleCommandCheckHoldReadLockCollection() {
|
|
TRI_ASSERT(ServerState::instance()->isDBServer());
|
|
|
|
bool success = false;
|
|
VPackSlice const body = this->parseVPackBody(success);
|
|
if (!success) {
|
|
// error already created
|
|
return;
|
|
}
|
|
|
|
if (!body.isObject()) {
|
|
generateError(rest::ResponseCode::BAD, TRI_ERROR_HTTP_BAD_PARAMETER,
|
|
"body needs to be an object with attribute 'id'");
|
|
return;
|
|
}
|
|
VPackSlice const idSlice = body.get("id");
|
|
if (!idSlice.isString()) {
|
|
generateError(rest::ResponseCode::BAD, TRI_ERROR_HTTP_BAD_PARAMETER,
|
|
"'id' needs to be a string");
|
|
return;
|
|
}
|
|
aql::QueryId id = ExtractReadlockId(idSlice);
|
|
LOG_TOPIC("05a75", DEBUG, Logger::REPLICATION) << "Test if Lock " << id << " is still active.";
|
|
auto res = isLockHeld(id);
|
|
if (!res.ok()) {
|
|
generateError(std::move(res).result());
|
|
return;
|
|
}
|
|
|
|
VPackBuilder b;
|
|
{
|
|
VPackObjectBuilder bb(&b);
|
|
b.add(StaticStrings::Error, VPackValue(false));
|
|
b.add("lockHeld", VPackValue(res.get()));
|
|
}
|
|
LOG_TOPIC("ca554", DEBUG, Logger::REPLICATION)
|
|
<< "Lock " << id << " is " << (res.get() ? "still active." : "gone.");
|
|
generateResult(rest::ResponseCode::OK, b.slice());
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
/// @brief cancel the holding of a read lock on a collection
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
|
|
void RestReplicationHandler::handleCommandCancelHoldReadLockCollection() {
|
|
TRI_ASSERT(ServerState::instance()->isDBServer());
|
|
|
|
bool success = false;
|
|
VPackSlice const body = this->parseVPackBody(success);
|
|
if (!success) {
|
|
// error already created
|
|
return;
|
|
}
|
|
|
|
if (!body.isObject()) {
|
|
generateError(rest::ResponseCode::BAD, TRI_ERROR_HTTP_BAD_PARAMETER,
|
|
"body needs to be an object with attribute 'id'");
|
|
return;
|
|
}
|
|
VPackSlice const idSlice = body.get("id");
|
|
if (!idSlice.isString()) {
|
|
generateError(rest::ResponseCode::BAD, TRI_ERROR_HTTP_BAD_PARAMETER,
|
|
"'id' needs to be a string");
|
|
return;
|
|
}
|
|
aql::QueryId id = ExtractReadlockId(idSlice);
|
|
LOG_TOPIC("9a5e3", DEBUG, Logger::REPLICATION) << "Attempt to cancel Lock: " << id;
|
|
|
|
auto res = cancelBlockingTransaction(id);
|
|
if (!res.ok()) {
|
|
LOG_TOPIC("9caf7", DEBUG, Logger::REPLICATION)
|
|
<< "Lock " << id << " not canceled because of: " << res.errorMessage();
|
|
generateError(std::move(res).result());
|
|
return;
|
|
}
|
|
|
|
VPackBuilder b;
|
|
{
|
|
VPackObjectBuilder bb(&b);
|
|
b.add(StaticStrings::Error, VPackValue(false));
|
|
b.add("lockHeld", VPackValue(res.get()));
|
|
}
|
|
|
|
LOG_TOPIC("a58e5", DEBUG, Logger::REPLICATION)
|
|
<< "Lock: " << id << " is now canceled, "
|
|
<< (res.get() ? "it is still in use." : "it is gone.");
|
|
generateResult(rest::ResponseCode::OK, b.slice());
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
/// @brief get ID for a read lock job
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
|
|
void RestReplicationHandler::handleCommandGetIdForReadLockCollection() {
|
|
TRI_ASSERT(ServerState::instance()->isDBServer());
|
|
|
|
std::string id = std::to_string(TRI_NewTickServer());
|
|
VPackBuilder b;
|
|
{
|
|
VPackObjectBuilder bb(&b);
|
|
b.add("id", VPackValue(id));
|
|
}
|
|
|
|
generateResult(rest::ResponseCode::OK, b.slice());
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
/// @brief was docuBlock JSF_get_api_replication_logger_return_state
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
|
|
void RestReplicationHandler::handleCommandLoggerState() {
|
|
StorageEngine* engine = EngineSelectorFeature::ENGINE;
|
|
TRI_ASSERT(engine);
|
|
|
|
VPackBuilder builder;
|
|
auto res = engine->createLoggerState(&_vocbase, builder);
|
|
|
|
if (res.fail()) {
|
|
LOG_TOPIC("c7471", DEBUG, Logger::REPLICATION)
|
|
<< "failed to create logger-state" << res.errorMessage();
|
|
generateError(rest::ResponseCode::BAD, res.errorNumber(), res.errorMessage());
|
|
return;
|
|
}
|
|
|
|
generateResult(rest::ResponseCode::OK, builder.slice());
|
|
}
|
|
|
|
//////////////////////////////////////////////////////////////////////////////
|
|
/// @brief return the first tick available in a logfile
|
|
/// @route GET logger-first-tick
|
|
/// @caller js/client/modules/@arangodb/replication.js
|
|
/// @response VPackObject with minTick of LogfileManager->ranges()
|
|
//////////////////////////////////////////////////////////////////////////////
|
|
void RestReplicationHandler::handleCommandLoggerFirstTick() {
|
|
TRI_voc_tick_t tick = UINT64_MAX;
|
|
Result res = EngineSelectorFeature::ENGINE->firstTick(tick);
|
|
|
|
VPackBuilder b;
|
|
b.add(VPackValue(VPackValueType::Object));
|
|
if (tick == UINT64_MAX || res.fail()) {
|
|
b.add("firstTick", VPackValue(VPackValueType::Null));
|
|
} else {
|
|
auto tickString = std::to_string(tick);
|
|
b.add("firstTick", VPackValue(tickString));
|
|
}
|
|
b.close();
|
|
generateResult(rest::ResponseCode::OK, b.slice());
|
|
}
|
|
|
|
//////////////////////////////////////////////////////////////////////////////
|
|
/// @brief return the available logfile range
|
|
/// @route GET logger-tick-ranges
|
|
/// @caller js/client/modules/@arangodb/replication.js
|
|
/// @response VPackArray, containing info about each datafile
|
|
/// * filename
|
|
/// * status
|
|
/// * tickMin - tickMax
|
|
//////////////////////////////////////////////////////////////////////////////
|
|
|
|
void RestReplicationHandler::handleCommandLoggerTickRanges() {
|
|
StorageEngine* engine = EngineSelectorFeature::ENGINE;
|
|
TRI_ASSERT(engine);
|
|
VPackBuilder b;
|
|
Result res = engine->createTickRanges(b);
|
|
if (res.ok()) {
|
|
generateResult(rest::ResponseCode::OK, b.slice());
|
|
} else {
|
|
generateError(res);
|
|
}
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
/// @brief creates a collection, based on the VelocyPack provided
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
|
|
int RestReplicationHandler::createCollection(VPackSlice slice,
|
|
arangodb::LogicalCollection** dst) {
|
|
TRI_ASSERT(dst != nullptr);
|
|
|
|
if (!slice.isObject()) {
|
|
return TRI_ERROR_HTTP_BAD_PARAMETER;
|
|
}
|
|
|
|
std::string const name =
|
|
arangodb::basics::VelocyPackHelper::getStringValue(slice, "name", "");
|
|
|
|
if (name.empty()) {
|
|
return TRI_ERROR_HTTP_BAD_PARAMETER;
|
|
}
|
|
|
|
std::string const uuid =
|
|
arangodb::basics::VelocyPackHelper::getStringValue(slice,
|
|
"globallyUniqueId", "");
|
|
|
|
TRI_col_type_e const type = static_cast<TRI_col_type_e>(
|
|
arangodb::basics::VelocyPackHelper::getNumericValue<int>(slice, "type",
|
|
int(TRI_COL_TYPE_DOCUMENT)));
|
|
std::shared_ptr<arangodb::LogicalCollection> col;
|
|
|
|
if (!uuid.empty()) {
|
|
col = _vocbase.lookupCollectionByUuid(uuid);
|
|
}
|
|
|
|
if (col != nullptr) {
|
|
col = _vocbase.lookupCollection(name);
|
|
}
|
|
|
|
if (col != nullptr && col->type() == type) {
|
|
// TODO
|
|
// collection already exists. TODO: compare attributes
|
|
return TRI_ERROR_NO_ERROR;
|
|
}
|
|
|
|
// always use current version number when restoring a collection,
|
|
// because the collection is effectively NEW
|
|
VPackBuilder patch;
|
|
patch.openObject();
|
|
patch.add("version", VPackValue(LogicalCollection::VERSION_31));
|
|
if (!name.empty() && name[0] == '_' && !slice.hasKey("isSystem")) {
|
|
// system collection?
|
|
patch.add("isSystem", VPackValue(true));
|
|
}
|
|
patch.add("objectId", VPackSlice::nullSlice());
|
|
patch.add("cid", VPackSlice::nullSlice());
|
|
patch.add("id", VPackSlice::nullSlice());
|
|
patch.close();
|
|
|
|
VPackBuilder builder =
|
|
VPackCollection::merge(slice, patch.slice(),
|
|
/*mergeValues*/ true, /*nullMeansRemove*/ true);
|
|
slice = builder.slice();
|
|
|
|
col = _vocbase.createCollection(slice);
|
|
|
|
if (col == nullptr) {
|
|
return TRI_ERROR_INTERNAL;
|
|
}
|
|
|
|
/* Temporary ASSERTS to prove correctness of new constructor */
|
|
TRI_ASSERT(col->system() == (name[0] == '_'));
|
|
#ifdef ARANGODB_ENABLE_MAINTAINER_MODE
|
|
TRI_voc_cid_t planId = 0;
|
|
VPackSlice const planIdSlice = slice.get("planId");
|
|
if (planIdSlice.isNumber()) {
|
|
planId = static_cast<TRI_voc_cid_t>(planIdSlice.getNumericValue<uint64_t>());
|
|
} else if (planIdSlice.isString()) {
|
|
std::string tmp = planIdSlice.copyString();
|
|
planId = static_cast<TRI_voc_cid_t>(StringUtils::uint64(tmp));
|
|
} else if (planIdSlice.isNone()) {
|
|
// There is no plan ID it has to be equal to collection id
|
|
planId = col->id();
|
|
}
|
|
|
|
TRI_ASSERT(col->planId() == planId);
|
|
#endif
|
|
|
|
if (dst != nullptr) {
|
|
*dst = col.get();
|
|
}
|
|
|
|
return TRI_ERROR_NO_ERROR;
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
/// @brief determine the chunk size
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
|
|
uint64_t RestReplicationHandler::determineChunkSize() const {
|
|
// determine chunk size
|
|
uint64_t chunkSize = _defaultChunkSize;
|
|
|
|
bool found;
|
|
std::string const& value = _request->value("chunkSize", found);
|
|
|
|
if (found) {
|
|
// query parameter "chunkSize" was specified
|
|
chunkSize = StringUtils::uint64(value);
|
|
|
|
// don't allow overly big allocations
|
|
if (chunkSize > _maxChunkSize) {
|
|
chunkSize = _maxChunkSize;
|
|
}
|
|
}
|
|
|
|
return chunkSize;
|
|
}
|
|
|
|
//////////////////////////////////////////////////////////////////////////////
|
|
/// @brief Grant temporary restore rights
|
|
//////////////////////////////////////////////////////////////////////////////
|
|
void RestReplicationHandler::grantTemporaryRights() {
|
|
if (ExecContext::CURRENT != nullptr) {
|
|
if (ExecContext::CURRENT->databaseAuthLevel() == auth::Level::RW) {
|
|
// If you have administrative access on this database,
|
|
// we grant you everything for restore.
|
|
ExecContext::CURRENT = nullptr;
|
|
}
|
|
}
|
|
}
|
|
|
|
ReplicationApplier* RestReplicationHandler::getApplier(bool& global) {
|
|
global = _request->parsedValue("global", false);
|
|
|
|
if (global && _request->databaseName() != StaticStrings::SystemDatabase) {
|
|
generateError(
|
|
rest::ResponseCode::FORBIDDEN, TRI_ERROR_FORBIDDEN,
|
|
"global inventory can only be created from within _system database");
|
|
return nullptr;
|
|
}
|
|
|
|
if (global) {
|
|
auto replicationFeature =
|
|
application_features::ApplicationServer::getFeature<ReplicationFeature>(
|
|
"Replication");
|
|
return replicationFeature->globalReplicationApplier();
|
|
} else {
|
|
return _vocbase.replicationApplier();
|
|
}
|
|
}
|
|
|
|
Result RestReplicationHandler::createBlockingTransaction(aql::QueryId id,
|
|
LogicalCollection& col, double ttl,
|
|
AccessMode::Type access) const {
|
|
// This is a constant JSON structure for Queries.
|
|
// we actually do not need a plan, as we only want the query registry to have
|
|
// a hold of our transaction
|
|
auto planBuilder = std::make_shared<VPackBuilder>(VPackSlice::emptyObjectSlice());
|
|
|
|
auto query = std::make_unique<aql::Query>(false, _vocbase, planBuilder, nullptr, /* options */
|
|
aql::QueryPart::PART_MAIN /* Do locking */
|
|
);
|
|
// NOTE: The collections are on purpose not locked here.
|
|
// To acquire an EXCLUSIVE lock may require time under load,
|
|
// we want to allow to cancel this operation while waiting
|
|
// for the lock.
|
|
|
|
auto queryRegistry = QueryRegistryFeature::registry();
|
|
if (queryRegistry == nullptr) {
|
|
return {TRI_ERROR_SHUTTING_DOWN};
|
|
}
|
|
|
|
{
|
|
auto ctx = transaction::StandaloneContext::Create(_vocbase);
|
|
auto trx = std::make_shared<SingleCollectionTransaction>(ctx, col, access);
|
|
query->setTransactionContext(ctx);
|
|
// Inject will take over responsiblilty of transaction, even on error case.
|
|
query->injectTransaction(std::move(trx));
|
|
}
|
|
auto trx = query->trx();
|
|
TRI_ASSERT(trx != nullptr);
|
|
trx->addHint(transaction::Hints::Hint::LOCK_ENTIRELY);
|
|
|
|
TRI_ASSERT(isLockHeld(id).is(TRI_ERROR_HTTP_NOT_FOUND));
|
|
|
|
try {
|
|
queryRegistry->insert(id, query.get(), ttl, true, true);
|
|
} catch (...) {
|
|
// For compatibility we only return this error
|
|
return {TRI_ERROR_TRANSACTION_INTERNAL, "cannot begin read transaction"};
|
|
}
|
|
// For leak protection in case of errors the unique_ptr
|
|
// is not responsible anymore, it has been handed over to the
|
|
// registry.
|
|
auto q = query.release();
|
|
|
|
// Make sure to return the query after we are done
|
|
auto guard = scopeGuard(
|
|
[this, id, &queryRegistry]() { queryRegistry->close(&_vocbase, id); });
|
|
|
|
if (isTombstoned(id)) {
|
|
try {
|
|
guard.fire();
|
|
// error code does not matter, read only access, so we can roll back.
|
|
// we can ignore the openness here, as it was our thread that had
|
|
// inserted the query just a couple of instructions before
|
|
queryRegistry->destroy(_vocbase.name(), id, TRI_ERROR_QUERY_KILLED, true /*ignoreOpened*/);
|
|
} catch (...) {
|
|
// Maybe thrown in shutdown.
|
|
}
|
|
// DO NOT LOCK in this case, pointless
|
|
return {TRI_ERROR_TRANSACTION_INTERNAL, "transaction already cancelled"};
|
|
}
|
|
|
|
TRI_ASSERT(isLockHeld(id).ok());
|
|
TRI_ASSERT(isLockHeld(id).get() == false);
|
|
|
|
return q->trx()->begin();
|
|
}
|
|
|
|
ResultT<bool> RestReplicationHandler::isLockHeld(aql::QueryId id) const {
|
|
// The query is only hold for long during initial locking
|
|
// there it should return false.
|
|
// In all other cases it is released quickly.
|
|
auto queryRegistry = QueryRegistryFeature::registry();
|
|
if (queryRegistry == nullptr) {
|
|
return ResultT<bool>::error(TRI_ERROR_SHUTTING_DOWN);
|
|
}
|
|
if (_vocbase.isDropped()) {
|
|
return ResultT<bool>::error(TRI_ERROR_ARANGO_DATABASE_NOT_FOUND);
|
|
}
|
|
auto res = queryRegistry->isQueryInUse(&_vocbase, id);
|
|
if (!res.ok()) {
|
|
// API compatibility otherwise just return res...
|
|
return ResultT<bool>::error(TRI_ERROR_HTTP_NOT_FOUND,
|
|
"no hold read lock job found for 'id'");
|
|
} else {
|
|
// We need to invert the result, because:
|
|
// if the query is there, but is in use => we are in the process of
|
|
// getting the lock => lock is not held if the query is there, but is not
|
|
// in use => we are after the process of getting the lock => lock is held
|
|
return ResultT<bool>::success(!res.get());
|
|
}
|
|
}
|
|
|
|
ResultT<bool> RestReplicationHandler::cancelBlockingTransaction(aql::QueryId id) const {
|
|
// This lookup is only required for API compatibility,
|
|
// otherwise an unconditional destroy() would do.
|
|
auto res = isLockHeld(id);
|
|
if (res.ok()) {
|
|
auto queryRegistry = QueryRegistryFeature::registry();
|
|
if (queryRegistry == nullptr) {
|
|
return ResultT<bool>::error(TRI_ERROR_SHUTTING_DOWN);
|
|
}
|
|
try {
|
|
// Code does not matter, read only access, so we can roll back.
|
|
queryRegistry->destroy(_vocbase.name(), id, TRI_ERROR_QUERY_KILLED, false);
|
|
} catch (...) {
|
|
// All errors that show up here can only be
|
|
// triggered if the query is destroyed in between.
|
|
}
|
|
} else {
|
|
registerTombstone(id);
|
|
}
|
|
return res;
|
|
}
|
|
|
|
ResultT<std::string> RestReplicationHandler::computeCollectionChecksum(
|
|
aql::QueryId id, LogicalCollection* col) const {
|
|
auto queryRegistry = QueryRegistryFeature::registry();
|
|
if (queryRegistry == nullptr) {
|
|
return ResultT<std::string>::error(TRI_ERROR_SHUTTING_DOWN);
|
|
}
|
|
|
|
try {
|
|
auto query = queryRegistry->open(&_vocbase, id);
|
|
if (query == nullptr) {
|
|
// Query does not exist. So we assume it got cancelled.
|
|
return ResultT<std::string>::error(TRI_ERROR_TRANSACTION_INTERNAL,
|
|
"read transaction was cancelled");
|
|
}
|
|
TRI_DEFER(queryRegistry->close(&_vocbase, id));
|
|
|
|
uint64_t num = col->numberDocuments(query->trx(), transaction::CountType::Normal);
|
|
return ResultT<std::string>::success(std::to_string(num));
|
|
} catch (...) {
|
|
// Query exists, but is in use.
|
|
// So in Locking phase
|
|
return ResultT<std::string>::error(TRI_ERROR_TRANSACTION_INTERNAL,
|
|
"Read lock not yet acquired!");
|
|
}
|
|
}
|
|
|
|
static std::string IdToTombstoneKey(TRI_vocbase_t& vocbase, aql::QueryId id) {
|
|
return vocbase.name() + "/" + StringUtils::itoa(id);
|
|
}
|
|
|
|
void RestReplicationHandler::timeoutTombstones() const {
|
|
std::unordered_set<std::string> toDelete;
|
|
{
|
|
READ_LOCKER(readLocker, RestReplicationHandler::_tombLock);
|
|
if (RestReplicationHandler::_tombstones.empty()) {
|
|
// Fast path
|
|
return;
|
|
}
|
|
auto now = std::chrono::steady_clock::now();
|
|
for (auto const& it : RestReplicationHandler::_tombstones) {
|
|
if (it.second < now) {
|
|
toDelete.emplace(it.first);
|
|
}
|
|
}
|
|
// Release read lock.
|
|
// If someone writes now we do not realy care.
|
|
}
|
|
if (toDelete.empty()) {
|
|
// nothing todo
|
|
return;
|
|
}
|
|
WRITE_LOCKER(writeLocker, RestReplicationHandler::_tombLock);
|
|
for (auto const& it : toDelete) {
|
|
try {
|
|
RestReplicationHandler::_tombstones.erase(it);
|
|
} catch (...) {
|
|
// erase should not throw.
|
|
TRI_ASSERT(false);
|
|
}
|
|
}
|
|
}
|
|
|
|
bool RestReplicationHandler::isTombstoned(aql::QueryId id) const {
|
|
std::string key = IdToTombstoneKey(_vocbase, id);
|
|
bool isDead = false;
|
|
{
|
|
READ_LOCKER(readLocker, RestReplicationHandler::_tombLock);
|
|
isDead = RestReplicationHandler::_tombstones.find(key) !=
|
|
RestReplicationHandler::_tombstones.end();
|
|
}
|
|
if (!isDead) {
|
|
// Clear Tombstone
|
|
WRITE_LOCKER(writeLocker, RestReplicationHandler::_tombLock);
|
|
try {
|
|
RestReplicationHandler::_tombstones.erase(key);
|
|
} catch (...) {
|
|
// Just ignore, tombstone will be removed by timeout, and IDs are unique
|
|
// anyways
|
|
TRI_ASSERT(false);
|
|
}
|
|
}
|
|
return isDead;
|
|
}
|
|
|
|
void RestReplicationHandler::registerTombstone(aql::QueryId id) const {
|
|
std::string key = IdToTombstoneKey(_vocbase, id);
|
|
{
|
|
WRITE_LOCKER(writeLocker, RestReplicationHandler::_tombLock);
|
|
RestReplicationHandler::_tombstones.emplace(key, std::chrono::steady_clock::now() +
|
|
RestReplicationHandler::_tombstoneTimeout);
|
|
}
|
|
timeoutTombstones();
|
|
}
|