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

1526 lines
51 KiB
C++
Raw Permalink Blame History

////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2017 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 "DatabaseInitialSyncer.h"
#include "ApplicationFeatures/ApplicationServer.h"
#include "Basics/Exceptions.h"
#include "Basics/ReadLocker.h"
#include "Basics/Result.h"
#include "Basics/RocksDBUtils.h"
#include "Basics/StaticStrings.h"
#include "Basics/StringUtils.h"
#include "Basics/VelocyPackHelper.h"
#include "Basics/system-functions.h"
#include "Indexes/Index.h"
#include "Logger/Logger.h"
#include "Replication/DatabaseReplicationApplier.h"
#include "Replication/utilities.h"
#include "SimpleHttpClient/SimpleHttpClient.h"
#include "SimpleHttpClient/SimpleHttpResult.h"
#include "StorageEngine/EngineSelectorFeature.h"
#include "StorageEngine/PhysicalCollection.h"
#include "StorageEngine/StorageEngine.h"
#include "Transaction/Helpers.h"
#include "Transaction/Methods.h"
#include "Transaction/StandaloneContext.h"
#include "Utils/CollectionGuard.h"
#include "Utils/OperationOptions.h"
#include "VocBase/LogicalCollection.h"
#include "VocBase/LogicalView.h"
#include "VocBase/voc-types.h"
#include "VocBase/vocbase.h"
#include <velocypack/Builder.h>
#include <velocypack/Iterator.h>
#include <velocypack/Slice.h>
#include <velocypack/Validator.h>
#include <velocypack/velocypack-aliases.h>
#include <array>
#include <cstring>
// lets keep this experimental until we make it faster
#define VPACK_DUMP 0
namespace {
/// @brief maximum internal value for chunkSize
size_t const maxChunkSize = 10 * 1024 * 1024;
std::chrono::milliseconds sleepTimeFromWaitTime(double waitTime) {
if (waitTime < 1.0) {
return std::chrono::milliseconds(100);
}
if (waitTime < 5.0) {
return std::chrono::milliseconds(200);
}
if (waitTime < 20.0) {
return std::chrono::milliseconds(500);
}
if (waitTime < 60.0) {
return std::chrono::seconds(1);
}
return std::chrono::seconds(2);
}
std::string const kTypeString = "type";
std::string const kDataString = "data";
} // namespace
namespace arangodb {
DatabaseInitialSyncer::Configuration::Configuration(
ReplicationApplierConfiguration const& a, replutils::BarrierInfo& bar,
replutils::BatchInfo& bat, replutils::Connection& c, bool f, replutils::MasterInfo& m,
replutils::ProgressInfo& p, SyncerState& s, TRI_vocbase_t& v)
: applier{a}, barrier{bar}, batch{bat}, connection{c}, flushed{f}, master{m}, progress{p}, state{s}, vocbase{v} {}
bool DatabaseInitialSyncer::Configuration::isChild() const {
return state.isChildSyncer;
}
DatabaseInitialSyncer::DatabaseInitialSyncer(TRI_vocbase_t& vocbase,
ReplicationApplierConfiguration const& configuration)
: InitialSyncer(configuration,
[this](std::string const& msg) -> void { setProgress(msg); }),
_config{_state.applier, _state.barrier, _batch,
_state.connection, false, _state.master,
_progress, _state, vocbase} {
_state.vocbases.try_emplace(vocbase.name(), vocbase);
if (configuration._database.empty()) {
_state.databaseName = vocbase.name();
}
}
/// @brief run method, performs a full synchronization
Result DatabaseInitialSyncer::runWithInventory(bool incremental, VPackSlice dbInventory) {
if (!_config.connection.valid()) {
return Result(TRI_ERROR_INTERNAL, "invalid endpoint");
}
auto res = vocbase().replicationApplier()->preventStart();
if (res.fail()) {
return res;
}
TRI_DEFER(vocbase().replicationApplier()->allowStart());
setAborted(false);
double const startTime = TRI_microtime();
try {
_config.progress.set("fetching master state");
LOG_TOPIC("0a10d", DEBUG, Logger::REPLICATION)
<< "client: getting master state to dump " << vocbase().name();
Result r;
r = sendFlush();
if (r.fail()) {
return r;
}
if (!_config.isChild()) {
r = _config.master.getState(_config.connection, _config.isChild());
if (r.fail()) {
return r;
}
}
TRI_ASSERT(!_config.master.endpoint.empty());
TRI_ASSERT(_config.master.serverId != 0);
TRI_ASSERT(_config.master.majorVersion != 0);
LOG_TOPIC("6fd2b", DEBUG, Logger::REPLICATION) << "client: got master state";
if (incremental) {
if (_config.master.majorVersion == 1 ||
(_config.master.majorVersion == 2 && _config.master.minorVersion <= 6)) {
LOG_TOPIC("15183", WARN, Logger::REPLICATION) << "incremental replication is "
"not supported with a master < "
"ArangoDB 2.7";
incremental = false;
}
}
if (!_config.isChild()) {
// create a WAL logfile barrier that prevents WAL logfile collection
r = _config.barrier.create(_config.connection, _config.master.lastLogTick);
if (r.fail()) {
return r;
}
// enable patching of collection count for ShardSynchronization Job
std::string patchCount = StaticStrings::Empty;
std::string const& engineName = EngineSelectorFeature::ENGINE->typeName();
if (incremental && engineName == "rocksdb" && _config.applier._skipCreateDrop &&
_config.applier._restrictType == ReplicationApplierConfiguration::RestrictType::Include &&
_config.applier._restrictCollections.size() == 1) {
patchCount = *_config.applier._restrictCollections.begin();
}
r = batchStart(patchCount);
if (r.fail()) {
return r;
}
startRecurringBatchExtension();
}
VPackSlice collections, views;
if (dbInventory.isObject()) {
collections = dbInventory.get("collections"); // required
views = dbInventory.get("views"); // optional
}
VPackBuilder inventoryResponse; // hold response data
if (!collections.isArray()) {
// caller did not supply an inventory, we need to fetch it
Result res = fetchInventory(inventoryResponse);
if (!res.ok()) {
return res;
}
// we do not really care about the state response
collections = inventoryResponse.slice().get("collections");
if (!collections.isArray()) {
return Result(TRI_ERROR_REPLICATION_INVALID_RESPONSE,
"collections section is missing from response");
}
views = inventoryResponse.slice().get("views");
}
// strip eventual objectIDs and then dump the collections
auto pair = rocksutils::stripObjectIds(collections);
r = handleCollectionsAndViews(pair.first, views, incremental);
// all done here, do not try to finish batch if master is unresponsive
if (r.isNot(TRI_ERROR_REPLICATION_NO_RESPONSE) && !_config.isChild()) {
batchFinish();
}
if (r.fail()) {
LOG_TOPIC("12556", DEBUG, Logger::REPLICATION)
<< "Error during initial sync: " << r.errorMessage();
}
LOG_TOPIC("055df", DEBUG, Logger::REPLICATION)
<< "initial synchronization with master took: "
<< Logger::FIXED(TRI_microtime() - startTime, 6) << " s. status: "
<< (r.errorMessage().empty() ? "all good" : r.errorMessage());
return r;
} catch (arangodb::basics::Exception const& ex) {
if (!_config.isChild()) {
batchFinish();
}
return Result(ex.code(), ex.what());
} catch (std::exception const& ex) {
if (!_config.isChild()) {
batchFinish();
}
return Result(TRI_ERROR_INTERNAL, ex.what());
} catch (...) {
if (!_config.isChild()) {
batchFinish();
}
return Result(TRI_ERROR_NO_ERROR, "an unknown exception occurred");
}
}
/// @brief fetch the server's inventory, public method for TailingSyncer
Result DatabaseInitialSyncer::getInventory(VPackBuilder& builder) {
if (!_state.connection.valid()) {
return Result(TRI_ERROR_INTERNAL, "invalid endpoint");
}
auto r = batchStart();
if (r.fail()) {
return r;
}
TRI_DEFER(batchFinish());
// caller did not supply an inventory, we need to fetch it
return fetchInventory(builder);
}
/// @brief check whether the initial synchronization should be aborted
bool DatabaseInitialSyncer::isAborted() const {
if (vocbase().server().isStopping() ||
(vocbase().replicationApplier() != nullptr &&
vocbase().replicationApplier()->stopInitialSynchronization())) {
return true;
}
return Syncer::isAborted();
}
void DatabaseInitialSyncer::setProgress(std::string const& msg) {
_config.progress.message = msg;
if (_config.applier._verbose) {
LOG_TOPIC("c6f5f", INFO, Logger::REPLICATION) << msg;
} else {
LOG_TOPIC("d15ed", DEBUG, Logger::REPLICATION) << msg;
}
auto* applier = _config.vocbase.replicationApplier();
if (applier != nullptr) {
applier->setProgress(msg);
}
}
/// @brief send a WAL flush command
Result DatabaseInitialSyncer::sendFlush() {
if (isAborted()) {
return Result(TRI_ERROR_REPLICATION_APPLIER_STOPPED);
}
std::string const& engineName = EngineSelectorFeature::ENGINE->typeName();
if (engineName == "rocksdb" && _state.master.engine == engineName) {
// no WAL flush required for RocksDB. this is only relevant for MMFiles
return Result();
}
std::string const url = "/_admin/wal/flush";
VPackBuilder builder;
builder.openObject();
builder.add("waitForSync", VPackValue(true));
builder.add("waitForCollector", VPackValue(true));
builder.add("waitForCollectorQueue", VPackValue(true));
builder.add("maxWaitTime", VPackValue(300.0));
builder.close();
VPackSlice bodySlice = builder.slice();
std::string const body = bodySlice.toJson();
// send request
_config.progress.set("sending WAL flush command to url " + url);
std::unique_ptr<httpclient::SimpleHttpResult> response;
_config.connection.lease([&](httpclient::SimpleHttpClient* client) {
response.reset(client->retryRequest(rest::RequestType::PUT, url,
body.c_str(), body.size()));
});
if (replutils::hasFailed(response.get())) {
return replutils::buildHttpError(response.get(), url, _config.connection);
}
_config.flushed = true;
return Result();
}
/// @brief handle a single dump marker
Result DatabaseInitialSyncer::parseCollectionDumpMarker(transaction::Methods& trx,
LogicalCollection* coll,
VPackSlice const& marker) {
if (!marker.isObject()) {
return TRI_ERROR_REPLICATION_INVALID_RESPONSE;
}
TRI_replication_operation_e type = REPLICATION_INVALID;
VPackSlice doc;
for (auto it : VPackObjectIterator(marker, true)) {
if (it.key.isEqualString(kTypeString)) {
if (it.value.isNumber()) {
type = static_cast<TRI_replication_operation_e>(it.value.getNumber<int>());
}
} else if (it.key.isEqualString(kDataString)) {
if (it.value.isObject()) {
doc = it.value;
}
}
if (type != REPLICATION_INVALID && doc.isObject()) {
break;
}
}
if (!doc.isObject()) {
return TRI_ERROR_REPLICATION_INVALID_RESPONSE;
}
// key must not be empty, but doc can otherwise be empty
VPackSlice key = doc.get(StaticStrings::KeyString);
if (!key.isString() || key.getStringLength() == 0) {
return TRI_ERROR_REPLICATION_INVALID_RESPONSE;
}
return applyCollectionDumpMarker(trx, coll, type, doc);
}
/// @brief apply the data from a collection dump
Result DatabaseInitialSyncer::parseCollectionDump(transaction::Methods& trx,
LogicalCollection* coll,
httpclient::SimpleHttpResult* response,
uint64_t& markersProcessed) {
TRI_ASSERT(!trx.isSingleOperationTransaction());
basics::StringBuffer const& data = response->getBody();
char const* p = data.begin();
char const* end = p + data.length();
bool found = false;
std::string const& cType =
response->getHeaderField(StaticStrings::ContentTypeHeader, found);
if (found && (cType == StaticStrings::MimeTypeVPack)) {
VPackOptions options;
options.validateUtf8Strings = true;
options.disallowExternals = true;
options.disallowCustom = true;
options.checkAttributeUniqueness = true;
options.unsupportedTypeBehavior = VPackOptions::FailOnUnsupportedType;
VPackValidator validator(&options);
try {
while (p < end) {
ptrdiff_t remaining = end - p;
// throws if the data is invalid
validator.validate(p, remaining, /*isSubPart*/ true);
VPackSlice marker(reinterpret_cast<uint8_t const*>(p));
Result r = parseCollectionDumpMarker(trx, coll, marker);
TRI_ASSERT(!r.is(TRI_ERROR_ARANGO_TRY_AGAIN));
if (r.fail()) {
r.reset(r.errorNumber(),
std::string("received invalid dump data for collection '") +
coll->name() + "'");
return r;
}
++markersProcessed;
p += marker.byteSize();
}
} catch (velocypack::Exception const& e) {
LOG_TOPIC("b9f4f", ERR, Logger::REPLICATION)
<< "Error parsing VPack response: " << e.what();
return Result(e.errorCode(), e.what());
}
} else {
// buffer must end with a NUL byte
TRI_ASSERT(*end == '\0');
VPackBuilder builder;
VPackParser parser(builder);
while (p < end) {
char const* q = strchr(p, '\n');
if (q == nullptr) {
q = end;
}
if (q - p < 2) {
// we are done
return Result();
}
TRI_ASSERT(q <= end);
try {
builder.clear();
parser.parse(p, static_cast<size_t>(q - p));
} catch (velocypack::Exception const& e) {
LOG_TOPIC("746ea", ERR, Logger::REPLICATION)
<< "while parsing collection dump: " << e.what();
return Result(e.errorCode(), e.what());
}
p = q + 1;
Result r = parseCollectionDumpMarker(trx, coll, builder.slice());
TRI_ASSERT(!r.is(TRI_ERROR_ARANGO_TRY_AGAIN));
if (r.fail()) {
return r;
}
++markersProcessed;
}
}
// reached the end
return Result();
}
/// @brief order a new chunk from the /dump API
void DatabaseInitialSyncer::fetchDumpChunk(std::shared_ptr<Syncer::JobSynchronizer> sharedStatus,
std::string const& baseUrl,
arangodb::LogicalCollection* coll,
std::string const& leaderColl,
InitialSyncerDumpStats& stats, int batch,
TRI_voc_tick_t fromTick, uint64_t chunkSize) {
using ::arangodb::basics::StringUtils::itoa;
if (isAborted()) {
sharedStatus->gotResponse(Result(TRI_ERROR_REPLICATION_APPLIER_STOPPED));
return;
}
// check if master & slave use the same storage engine
// if both use RocksDB, there is no need to use an async request for the
// initial batch. this is because with RocksDB there is no initial load
// time for collections as there may be with MMFiles if the collection is
// not yet in memory
std::string const& engineName = EngineSelectorFeature::ENGINE->typeName();
bool const useAsync =
(batch == 1 && (engineName != "rocksdb" || _state.master.engine != engineName));
try {
std::string const typeString = (coll->type() == TRI_COL_TYPE_EDGE ? "edge" : "document");
if (!_config.isChild()) {
batchExtend();
_config.barrier.extend(_config.connection);
}
// assemble URL to call
std::string url = baseUrl + "&from=" + itoa(fromTick) + "&chunkSize=" + itoa(chunkSize);
if (_config.flushed) {
url += "&flush=false";
} else {
// only flush WAL once
url += "&flush=true&flushWait=180";
_config.flushed = true;
}
auto headers = replutils::createHeaders();
if (useAsync) {
// use async mode for first batch
headers[StaticStrings::Async] = "store";
}
#if VPACK_DUMP
int vv = _config.master.majorVersion * 1000000 + _config.master.minorVersion * 1000;
if (vv >= 3003009) {
headers[StaticStrings::Accept] = StaticStrings::MimeTypeVPack;
}
#endif
_config.progress.set(
std::string("fetching master collection dump for collection '") +
coll->name() + "', type: " + typeString + ", id: " + leaderColl +
", batch " + itoa(batch) + ", url: " + url);
++stats.numDumpRequests;
double t = TRI_microtime();
// send request
std::unique_ptr<httpclient::SimpleHttpResult> response;
_config.connection.lease([&](httpclient::SimpleHttpClient* client) {
response.reset(client->retryRequest(rest::RequestType::GET, url, nullptr, 0, headers));
});
t = TRI_microtime() - t;
if (replutils::hasFailed(response.get())) {
stats.waitedForDump += t;
sharedStatus->gotResponse(
replutils::buildHttpError(response.get(), url, _config.connection), t);
return;
}
// use async mode for first batch
if (useAsync) {
bool found = false;
std::string jobId = response->getHeaderField(StaticStrings::AsyncId, found);
if (!found) {
sharedStatus->gotResponse(
Result(TRI_ERROR_REPLICATION_INVALID_RESPONSE,
std::string("got invalid response from master at ") +
_config.master.endpoint + url +
": could not find 'X-Arango-Async' header"));
return;
}
double const startTime = TRI_microtime();
// wait until we get a reasonable response
while (true) {
if (!_config.isChild()) {
batchExtend();
_config.barrier.extend(_config.connection);
}
std::string const jobUrl = "/_api/job/" + jobId;
_config.connection.lease([&](httpclient::SimpleHttpClient* client) {
response.reset(client->request(rest::RequestType::PUT, jobUrl, nullptr, 0));
});
if (response != nullptr && response->isComplete()) {
if (response->hasHeaderField("x-arango-async-id")) {
// got the actual response
break;
}
if (response->getHttpReturnCode() == 404) {
// unknown job, we can abort
sharedStatus->gotResponse(
Result(TRI_ERROR_REPLICATION_NO_RESPONSE,
std::string("job not found on master at ") +
_config.master.endpoint));
return;
}
}
double waitTime = TRI_microtime() - startTime;
if (static_cast<uint64_t>(waitTime * 1000.0 * 1000.0) >=
_config.applier._initialSyncMaxWaitTime) {
sharedStatus->gotResponse(Result(
TRI_ERROR_REPLICATION_NO_RESPONSE,
std::string("timed out waiting for response from master at ") +
_config.master.endpoint));
return;
}
if (isAborted()) {
sharedStatus->gotResponse(Result(TRI_ERROR_REPLICATION_APPLIER_STOPPED));
return;
}
std::chrono::milliseconds sleepTime = ::sleepTimeFromWaitTime(waitTime);
std::this_thread::sleep_for(sleepTime);
}
// fallthrough here in case everything went well
}
stats.waitedForDump += t;
if (replutils::hasFailed(response.get())) {
// failure
sharedStatus->gotResponse(
replutils::buildHttpError(response.get(), url, _config.connection), t);
} else {
// success!
sharedStatus->gotResponse(std::move(response), t);
}
} catch (basics::Exception const& ex) {
sharedStatus->gotResponse(Result(ex.code(), ex.what()));
} catch (std::exception const& ex) {
sharedStatus->gotResponse(Result(TRI_ERROR_INTERNAL, ex.what()));
}
}
/// @brief incrementally fetch data from a collection
Result DatabaseInitialSyncer::fetchCollectionDump(arangodb::LogicalCollection* coll,
std::string const& leaderColl,
TRI_voc_tick_t maxTick) {
using ::arangodb::basics::StringUtils::boolean;
using ::arangodb::basics::StringUtils::itoa;
using ::arangodb::basics::StringUtils::uint64;
using ::arangodb::basics::StringUtils::urlEncode;
if (isAborted()) {
return Result(TRI_ERROR_REPLICATION_APPLIER_STOPPED);
}
std::string const typeString = (coll->type() == TRI_COL_TYPE_EDGE ? "edge" : "document");
InitialSyncerDumpStats stats;
TRI_ASSERT(_config.batch.id); // should not be equal to 0
// assemble base URL
std::string baseUrl =
replutils::ReplicationUrl + "/dump?collection=" + urlEncode(leaderColl) +
"&batchId=" + std::to_string(_config.batch.id) +
"&includeSystem=" + std::string(_config.applier._includeSystem ? "true" : "false") +
"&serverId=" + _state.localServerIdString;
if (maxTick > 0) {
baseUrl += "&to=" + itoa(maxTick + 1);
}
// state variables for the dump
TRI_voc_tick_t fromTick = 0;
int batch = 1;
uint64_t chunkSize = _config.applier._chunkSize;
uint64_t bytesReceived = 0;
uint64_t markersProcessed = 0;
double const startTime = TRI_microtime();
// the shared status will wait in its destructor until all posted
// requests have been completed/canceled!
auto self = shared_from_this();
auto sharedStatus = std::make_shared<Syncer::JobSynchronizer>(self);
// order initial chunk. this will block until the initial response
// has arrived
fetchDumpChunk(sharedStatus, baseUrl, coll, leaderColl, stats, batch, fromTick, chunkSize);
while (true) {
std::unique_ptr<httpclient::SimpleHttpResult> dumpResponse;
// block until we either got a response or were shut down
Result res = sharedStatus->waitForResponse(dumpResponse);
if (res.fail()) {
// no response or error or shutdown
return res;
}
// now we have got a response!
TRI_ASSERT(dumpResponse != nullptr);
if (dumpResponse->hasContentLength()) {
bytesReceived += dumpResponse->getContentLength();
}
bool found;
std::string header =
dumpResponse->getHeaderField(StaticStrings::ReplicationHeaderCheckMore, found);
if (!found) {
return Result(TRI_ERROR_REPLICATION_INVALID_RESPONSE,
std::string("got invalid response from master at ") +
_config.master.endpoint + ": required header " +
StaticStrings::ReplicationHeaderCheckMore +
" is missing in dump response");
}
TRI_voc_tick_t tick;
bool checkMore = boolean(header);
if (checkMore) {
header = dumpResponse->getHeaderField(StaticStrings::ReplicationHeaderLastIncluded, found);
if (!found) {
return Result(TRI_ERROR_REPLICATION_INVALID_RESPONSE,
std::string("got invalid response from master at ") +
_config.master.endpoint + ": required header " +
StaticStrings::ReplicationHeaderLastIncluded +
" is missing in dump response");
}
tick = uint64(header);
if (tick > fromTick) {
fromTick = tick;
} else {
// we got the same tick again, this indicates we're at the end
checkMore = false;
}
}
// increase chunk size for next fetch
if (chunkSize < ::maxChunkSize) {
chunkSize = static_cast<uint64_t>(chunkSize * 1.25);
if (chunkSize > ::maxChunkSize) {
chunkSize = ::maxChunkSize;
}
}
if (checkMore && !isAborted()) {
// already fetch next batch in the background, by posting the
// request to the scheduler, which can run it asynchronously
sharedStatus->request([this, self, &stats, &baseUrl, sharedStatus, coll,
leaderColl, batch, fromTick, chunkSize]() {
fetchDumpChunk(sharedStatus, baseUrl, coll, leaderColl, stats,
batch + 1, fromTick, chunkSize);
});
}
SingleCollectionTransaction trx(transaction::StandaloneContext::Create(vocbase()),
*coll, AccessMode::Type::EXCLUSIVE);
// to turn off waitForSync!
trx.addHint(transaction::Hints::Hint::RECOVERY);
// do not index the operations in our own transaction
trx.addHint(transaction::Hints::Hint::NO_INDEXING);
// smaller batch sizes should work better here
#if VPACK_DUMP
// trx.state()->options().intermediateCommitCount = 128;
// trx.state()->options().intermediateCommitSize = 16 * 1024 * 1024;
#endif
res = trx.begin();
if (!res.ok()) {
return Result(res.errorNumber(),
std::string("unable to start transaction: ") + res.errorMessage());
}
trx.pinData(coll->id()); // will throw when it fails
double t = TRI_microtime();
TRI_ASSERT(!trx.isSingleOperationTransaction());
res = parseCollectionDump(trx, coll, dumpResponse.get(), markersProcessed);
if (res.fail()) {
TRI_ASSERT(!res.is(TRI_ERROR_ARANGO_TRY_AGAIN));
return res;
}
res = trx.commit();
double applyTime = TRI_microtime() - t;
stats.waitedForApply += applyTime;
_config.progress.set(
std::string("fetched master collection dump for collection '") +
coll->name() + "', type: " + typeString + ", id: " + leaderColl +
", batch " + itoa(batch) + ", markers processed: " + itoa(markersProcessed) +
", bytes received: " + itoa(bytesReceived) +
", apply time: " + std::to_string(applyTime) + " s");
if (!res.ok()) {
return res;
}
if (!checkMore || fromTick == 0) {
// done
_config.progress.set(
std::string("finished initial dump for collection '") + coll->name() +
"', type: " + typeString + ", id: " + leaderColl +
", markers processed: " + itoa(markersProcessed) +
", bytes received: " + itoa(bytesReceived) +
", dump requests: " + std::to_string(stats.numDumpRequests) +
", waited for dump: " + std::to_string(stats.waitedForDump) + " s" +
", apply time: " + std::to_string(stats.waitedForApply) + " s" +
", total time: " + std::to_string(TRI_microtime() - startTime) + " s");
return Result();
}
batch++;
if (isAborted()) {
return Result(TRI_ERROR_REPLICATION_APPLIER_STOPPED);
}
}
TRI_ASSERT(false);
return Result(TRI_ERROR_INTERNAL);
}
/// @brief incrementally fetch data from a collection
Result DatabaseInitialSyncer::fetchCollectionSync(arangodb::LogicalCollection* coll,
std::string const& leaderColl,
TRI_voc_tick_t maxTick) {
using ::arangodb::basics::StringUtils::urlEncode;
if (!_config.isChild()) {
batchExtend();
_config.barrier.extend(_config.connection);
}
std::string const baseUrl = replutils::ReplicationUrl + "/keys";
std::string url = baseUrl + "/keys" + "?collection=" + urlEncode(leaderColl) +
"&to=" + std::to_string(maxTick) +
"&serverId=" + _state.localServerIdString +
"&batchId=" + std::to_string(_config.batch.id);
std::string msg =
"fetching collection keys for collection '" + coll->name() + "' from " + url;
_config.progress.set(msg);
// send an initial async request to collect the collection keys on the other
// side
// sending this request in a blocking fashion may require very long to
// complete,
// so we're sending the x-arango-async header here
auto headers = replutils::createHeaders();
headers[StaticStrings::Async] = "store";
std::unique_ptr<httpclient::SimpleHttpResult> response;
_config.connection.lease([&](httpclient::SimpleHttpClient* client) {
response.reset(client->retryRequest(rest::RequestType::POST, url, nullptr, 0, headers));
});
if (replutils::hasFailed(response.get())) {
return replutils::buildHttpError(response.get(), url, _config.connection);
}
bool found = false;
std::string jobId = response->getHeaderField(StaticStrings::AsyncId, found);
if (!found) {
return Result(TRI_ERROR_REPLICATION_INVALID_RESPONSE,
std::string("got invalid response from master at ") +
_config.master.endpoint + url +
": could not find 'X-Arango-Async' header");
}
double const startTime = TRI_microtime();
while (true) {
if (!_config.isChild()) {
batchExtend();
_config.barrier.extend(_config.connection);
}
std::string const jobUrl = "/_api/job/" + jobId;
_config.connection.lease([&](httpclient::SimpleHttpClient* client) {
response.reset(client->request(rest::RequestType::PUT, jobUrl, nullptr, 0));
});
if (response != nullptr && response->isComplete()) {
if (response->hasHeaderField("x-arango-async-id")) {
// job is done, got the actual response
break;
}
if (response->getHttpReturnCode() == 404) {
// unknown job, we can abort
return Result(TRI_ERROR_REPLICATION_NO_RESPONSE,
std::string("job not found on master at ") +
_config.master.endpoint);
}
}
double waitTime = TRI_microtime() - startTime;
if (static_cast<uint64_t>(waitTime * 1000.0 * 1000.0) >= _config.applier._initialSyncMaxWaitTime) {
return Result(TRI_ERROR_REPLICATION_NO_RESPONSE,
std::string(
"timed out waiting for response from master at ") +
_config.master.endpoint);
}
if (isAborted()) {
return Result(TRI_ERROR_REPLICATION_APPLIER_STOPPED);
}
std::chrono::milliseconds sleepTime = ::sleepTimeFromWaitTime(waitTime);
std::this_thread::sleep_for(sleepTime);
}
if (replutils::hasFailed(response.get())) {
return replutils::buildHttpError(response.get(), url, _config.connection);
}
VPackBuilder builder;
Result r = replutils::parseResponse(builder, response.get());
if (r.fail()) {
return Result(TRI_ERROR_REPLICATION_INVALID_RESPONSE,
std::string("got invalid response from master at ") +
_config.master.endpoint + url + ": " + r.errorMessage());
}
VPackSlice const slice = builder.slice();
if (!slice.isObject()) {
return Result(TRI_ERROR_REPLICATION_INVALID_RESPONSE,
std::string("got invalid response from master at ") +
_config.master.endpoint + url + ": response is no object");
}
VPackSlice const keysId = slice.get("id");
if (!keysId.isString()) {
return Result(TRI_ERROR_REPLICATION_INVALID_RESPONSE,
std::string("got invalid response from master at ") +
_config.master.endpoint + url +
": response does not contain valid 'id' attribute");
}
auto shutdown = [&]() -> void {
url = baseUrl + "/" + keysId.copyString();
std::string msg =
"deleting remote collection keys object for collection '" +
coll->name() + "' from " + url;
_config.progress.set(msg);
// now delete the keys we ordered
std::unique_ptr<httpclient::SimpleHttpResult> response;
_config.connection.lease([&](httpclient::SimpleHttpClient* client) {
response.reset(client->retryRequest(rest::RequestType::DELETE_REQ, url, nullptr, 0));
});
};
TRI_DEFER(shutdown());
VPackSlice const count = slice.get("count");
if (!count.isNumber()) {
return Result(TRI_ERROR_REPLICATION_INVALID_RESPONSE,
std::string("got invalid response from master at ") +
_config.master.endpoint + url +
": response does not contain valid 'count' attribute");
}
if (count.getNumber<size_t>() <= 0) {
// remote collection has no documents. now truncate our local collection
SingleCollectionTransaction trx(transaction::StandaloneContext::Create(vocbase()),
*coll, AccessMode::Type::EXCLUSIVE);
trx.addHint(transaction::Hints::Hint::INTERMEDIATE_COMMITS);
trx.addHint(transaction::Hints::Hint::ALLOW_RANGE_DELETE);
Result res = trx.begin();
if (!res.ok()) {
return Result(res.errorNumber(),
std::string("unable to start transaction (") + std::string(__FILE__) +
std::string(":") + std::to_string(__LINE__) +
std::string("): ") + res.errorMessage());
}
OperationOptions options;
if (!_state.leaderId.empty()) {
options.isSynchronousReplicationFrom = _state.leaderId;
}
OperationResult opRes = trx.truncate(coll->name(), options);
if (opRes.fail()) {
return Result(opRes.errorNumber(),
std::string("unable to truncate collection '") + coll->name() +
"': " + TRI_errno_string(opRes.errorNumber()));
}
return trx.finish(opRes.result);
}
// now we can fetch the complete chunk information from the master
try {
return EngineSelectorFeature::ENGINE->handleSyncKeys(*this, *coll, keysId.copyString());
} catch (arangodb::basics::Exception const& ex) {
return Result(ex.code(), ex.what());
} catch (std::exception const& ex) {
return Result(TRI_ERROR_INTERNAL, ex.what());
} catch (...) {
return Result(TRI_ERROR_INTERNAL);
}
}
/// @brief incrementally fetch data from a collection
/// @brief changes the properties of a collection, based on the VelocyPack
/// provided
Result DatabaseInitialSyncer::changeCollection(arangodb::LogicalCollection* col,
VPackSlice const& slice) {
arangodb::CollectionGuard guard(&vocbase(), col->id());
return guard.collection()->properties(slice, false); // always a full-update
}
/// @brief determine the number of documents in a collection
int64_t DatabaseInitialSyncer::getSize(arangodb::LogicalCollection const& col) {
SingleCollectionTransaction trx(transaction::StandaloneContext::Create(vocbase()),
col, AccessMode::Type::READ);
Result res = trx.begin();
if (res.fail()) {
return -1;
}
auto result = trx.count(col.name(), transaction::CountType::Normal);
if (result.result.fail()) {
return -1;
}
VPackSlice s = result.slice();
if (!s.isNumber()) {
return -1;
}
return s.getNumber<int64_t>();
}
/// @brief handle the information about a collection
Result DatabaseInitialSyncer::handleCollection(VPackSlice const& parameters,
VPackSlice const& indexes,
bool incremental, SyncPhase phase) {
using ::arangodb::basics::StringUtils::itoa;
if (isAborted()) {
return Result(TRI_ERROR_REPLICATION_APPLIER_STOPPED);
}
if (!parameters.isObject() || !indexes.isArray()) {
return Result(TRI_ERROR_REPLICATION_INVALID_RESPONSE);
}
if (!_config.isChild()) {
batchExtend();
_config.barrier.extend(_config.connection);
}
std::string const masterName =
basics::VelocyPackHelper::getStringValue(parameters, "name", "");
TRI_voc_cid_t const masterCid = basics::VelocyPackHelper::extractIdValue(parameters);
if (masterCid == 0) {
return Result(TRI_ERROR_REPLICATION_INVALID_RESPONSE,
"collection id is missing in response");
}
std::string const masterUuid =
basics::VelocyPackHelper::getStringValue(parameters, "globallyUniqueId",
"");
VPackSlice const type = parameters.get("type");
if (!type.isNumber()) {
return Result(TRI_ERROR_REPLICATION_INVALID_RESPONSE,
"collection type is missing in response");
}
std::string const typeString = (type.getNumber<int>() == 3 ? "edge" : "document");
std::string const collectionMsg = "collection '" + masterName + "', type " +
typeString + ", id " + itoa(masterCid);
// phase handling
if (phase == PHASE_VALIDATE) {
// validation phase just returns ok if we got here (aborts above if data is
// invalid)
_config.progress.processedCollections.try_emplace(masterCid, masterName);
return Result();
}
// drop and re-create collections locally
// -------------------------------------------------------------------------------------
if (phase == PHASE_DROP_CREATE) {
auto* col = resolveCollection(vocbase(), parameters).get();
if (col == nullptr) {
// not found...
col = vocbase().lookupCollection(masterName).get();
if (col != nullptr && (col->name() != masterName ||
(!masterUuid.empty() && col->guid() != masterUuid))) {
// found another collection with the same name locally.
// in this case we must drop it because we will run into duplicate
// name conflicts otherwise
try {
auto res = vocbase().dropCollection(col->id(), true, -1.0).errorNumber();
if (res == TRI_ERROR_NO_ERROR) {
col = nullptr;
}
} catch (...) {
}
}
}
if (col != nullptr) {
if (!incremental) {
// first look up the collection
if (col != nullptr) {
bool truncate = false;
if (col->name() == TRI_COL_NAME_USERS) {
// better not throw away the _users collection. otherwise it is gone
// and this may be a problem if the
// server crashes in-between.
truncate = true;
}
if (truncate) {
// system collection
_config.progress.set("truncating " + collectionMsg);
SingleCollectionTransaction trx(transaction::StandaloneContext::Create(vocbase()),
*col, AccessMode::Type::EXCLUSIVE);
trx.addHint(transaction::Hints::Hint::INTERMEDIATE_COMMITS);
trx.addHint(transaction::Hints::Hint::ALLOW_RANGE_DELETE);
Result res = trx.begin();
if (!res.ok()) {
return Result(res.errorNumber(),
std::string("unable to truncate ") + collectionMsg +
": " + res.errorMessage());
}
OperationOptions options;
if (!_state.leaderId.empty()) {
options.isSynchronousReplicationFrom = _state.leaderId;
}
OperationResult opRes = trx.truncate(col->name(), options);
if (opRes.fail()) {
return Result(opRes.errorNumber(),
std::string("unable to truncate ") + collectionMsg +
": " + TRI_errno_string(opRes.errorNumber()));
}
res = trx.finish(opRes.result);
if (!res.ok()) {
return Result(res.errorNumber(),
std::string("unable to truncate ") + collectionMsg +
": " + res.errorMessage());
}
} else {
// drop a regular collection
if (_config.applier._skipCreateDrop) {
_config.progress.set("dropping " + collectionMsg +
" skipped because of configuration");
return Result();
}
_config.progress.set("dropping " + collectionMsg);
auto res = vocbase().dropCollection(col->id(), true, -1.0).errorNumber();
if (res != TRI_ERROR_NO_ERROR) {
return Result(res, std::string("unable to drop ") + collectionMsg +
": " + TRI_errno_string(res));
}
}
}
} else {
// incremental case
TRI_ASSERT(incremental);
// collection is already present
_config.progress.set("checking/changing parameters of " + collectionMsg);
return changeCollection(col, parameters);
}
}
std::string msg = "creating " + collectionMsg;
if (_config.applier._skipCreateDrop) {
msg += " skipped because of configuration";
_config.progress.set(msg);
return Result();
}
_config.progress.set(msg);
LOG_TOPIC("7093d", DEBUG, Logger::REPLICATION)
<< "Dump is creating collection " << parameters.toJson();
auto r = createCollection(vocbase(), parameters, &col);
if (r.fail()) {
return Result(r.errorNumber(), std::string("unable to create ") + collectionMsg +
": " + TRI_errno_string(r.errorNumber()) +
". Collection info " + parameters.toJson());
}
return r;
}
// sync collection data
// -------------------------------------------------------------------------------------
else if (phase == PHASE_DUMP) {
_config.progress.set("dumping data for " + collectionMsg);
auto* col = resolveCollection(vocbase(), parameters).get();
if (col == nullptr) {
return Result(TRI_ERROR_ARANGO_DATA_SOURCE_NOT_FOUND,
std::string("cannot dump: ") + collectionMsg +
" not found on slave. Collection info " + parameters.toJson());
}
std::string const& masterColl = !masterUuid.empty() ? masterUuid : itoa(masterCid);
auto res = incremental && getSize(*col) > 0
? fetchCollectionSync(col, masterColl, _config.master.lastLogTick)
: fetchCollectionDump(col, masterColl, _config.master.lastLogTick);
if (!res.ok()) {
return res;
} else if (isAborted()) {
return res.reset(TRI_ERROR_REPLICATION_APPLIER_STOPPED);
}
if (masterName == TRI_COL_NAME_USERS) {
reloadUsers();
}
// schmutz++ creates indexes on DBServers
if (_config.applier._skipCreateDrop) {
_config.progress.set("creating indexes for " + collectionMsg +
" skipped because of configuration");
return res;
}
// now create indexes
TRI_ASSERT(indexes.isArray());
VPackValueLength const numIdx = indexes.length();
if (numIdx > 0) {
if (!_config.isChild()) {
batchExtend();
_config.barrier.extend(_config.connection);
}
_config.progress.set("creating " + std::to_string(numIdx) +
" index(es) for " + collectionMsg);
try {
for (auto const& idxDef : VPackArrayIterator(indexes)) {
if (idxDef.isObject()) {
VPackSlice const type = idxDef.get(StaticStrings::IndexType);
if (type.isString()) {
_config.progress.set("creating index of type " +
type.copyString() + " for " + collectionMsg);
}
}
createIndexInternal(idxDef, *col);
}
} catch (arangodb::basics::Exception const& ex) {
return res.reset(ex.code(), ex.what());
} catch (std::exception const& ex) {
return res.reset(TRI_ERROR_INTERNAL, ex.what());
} catch (...) {
return res.reset(TRI_ERROR_INTERNAL);
}
}
return res;
}
// we won't get here
TRI_ASSERT(false);
return Result(TRI_ERROR_INTERNAL);
}
/// @brief fetch the server's inventory
arangodb::Result DatabaseInitialSyncer::fetchInventory(VPackBuilder& builder) {
std::string url = replutils::ReplicationUrl +
"/inventory?serverId=" + _state.localServerIdString +
"&batchId=" + std::to_string(_config.batch.id);
if (_config.applier._includeSystem) {
url += "&includeSystem=true";
}
if (_config.applier._includeFoxxQueues) {
url += "&includeFoxxQueues=true";
}
// send request
_config.progress.set("fetching master inventory from " + url);
std::unique_ptr<httpclient::SimpleHttpResult> response;
_config.connection.lease([&](httpclient::SimpleHttpClient* client) {
response.reset(client->retryRequest(rest::RequestType::GET, url, nullptr, 0));
});
if (replutils::hasFailed(response.get())) {
if (!_config.isChild()) {
batchFinish();
}
return replutils::buildHttpError(response.get(), url, _config.connection);
}
Result r = replutils::parseResponse(builder, response.get());
if (r.fail()) {
return Result(
r.errorNumber(),
std::string("got invalid response from master at ") + _config.master.endpoint +
url + ": invalid response type for initial data. expecting array");
}
VPackSlice const slice = builder.slice();
if (!slice.isObject()) {
LOG_TOPIC("3b1e6", DEBUG, Logger::REPLICATION)
<< "client: DatabaseInitialSyncer::run - inventoryResponse is not an "
"object";
return Result(TRI_ERROR_REPLICATION_INVALID_RESPONSE,
std::string("got invalid response from master at ") +
_config.master.endpoint + url + ": invalid JSON");
}
return Result();
}
/// @brief handle the inventory response of the master
Result DatabaseInitialSyncer::handleCollectionsAndViews(VPackSlice const& collSlices,
VPackSlice const& viewSlices,
bool incremental) {
TRI_ASSERT(collSlices.isArray());
std::vector<std::pair<VPackSlice, VPackSlice>> systemCollections;
std::vector<std::pair<VPackSlice, VPackSlice>> collections;
for (VPackSlice it : VPackArrayIterator(collSlices)) {
if (!it.isObject()) {
return Result(TRI_ERROR_REPLICATION_INVALID_RESPONSE,
"collection declaration is invalid in response");
}
VPackSlice const parameters = it.get("parameters");
if (!parameters.isObject()) {
return Result(TRI_ERROR_REPLICATION_INVALID_RESPONSE,
"collection parameters declaration is invalid in response");
}
VPackSlice const indexes = it.get("indexes");
if (!indexes.isArray()) {
return Result(TRI_ERROR_REPLICATION_INVALID_RESPONSE,
"collection indexes declaration is invalid in response");
}
std::string const masterName =
basics::VelocyPackHelper::getStringValue(parameters, "name", "");
if (masterName.empty()) {
return Result(TRI_ERROR_REPLICATION_INVALID_RESPONSE,
"collection name is missing in response");
}
if (TRI_ExcludeCollectionReplication(masterName, _config.applier._includeSystem,
_config.applier._includeFoxxQueues)) {
continue;
}
if (basics::VelocyPackHelper::getBooleanValue(parameters, "deleted", false)) {
// we don't care about deleted collections
continue;
}
if (_config.applier._restrictType != ReplicationApplierConfiguration::RestrictType::None) {
auto const it = _config.applier._restrictCollections.find(masterName);
bool found = (it != _config.applier._restrictCollections.end());
if (_config.applier._restrictType == ReplicationApplierConfiguration::RestrictType::Include &&
!found) {
// collection should not be included
continue;
} else if (_config.applier._restrictType ==
ReplicationApplierConfiguration::RestrictType::Exclude &&
found) {
// collection should be excluded
continue;
}
}
if (masterName == StaticStrings::AnalyzersCollection) {
// _analyzers collection has to be restored before view creation
systemCollections.emplace_back(parameters, indexes);
} else {
collections.emplace_back(parameters, indexes);
}
}
// STEP 1: validate collection declarations from master
// ----------------------------------------------------------------------------------
// STEP 2: drop and re-create collections locally if they are also present on
// the master
// ------------------------------------------------------------------------------------
// iterate over all collections from the master...
std::array<SyncPhase, 2> phases{{PHASE_VALIDATE, PHASE_DROP_CREATE}};
for (auto const& phase : phases) {
Result r = iterateCollections(systemCollections, incremental, phase);
if (r.fail()) {
return r;
}
r = iterateCollections(collections, incremental, phase);
if (r.fail()) {
return r;
}
}
// STEP 3: restore data for system collections
// ----------------------------------------------------------------------------------
auto const res = iterateCollections(systemCollections, incremental, PHASE_DUMP);
if (res.fail()) {
return res;
}
// STEP 4: now that the collections exist create the views
// this should be faster than re-indexing afterwards
// ----------------------------------------------------------------------------------
if (!_config.applier._skipCreateDrop &&
_config.applier._restrictCollections.empty() &&
viewSlices.isArray()) {
// views are optional, and 3.3 and before will not send any view data
Result r = handleViewCreation(viewSlices); // no requests to master
if (r.fail()) {
LOG_TOPIC("96cda", ERR, Logger::REPLICATION)
<< "Error during intial sync view creation: " << r.errorMessage();
return r;
}
} else {
_config.progress.set("view creation skipped because of configuration");
}
// STEP 5: sync collection data from master and create initial indexes
// ----------------------------------------------------------------------------------
// now load the data into the collections
return iterateCollections(collections, incremental, PHASE_DUMP);
}
/// @brief iterate over all collections from an array and apply an action
Result DatabaseInitialSyncer::iterateCollections(
std::vector<std::pair<VPackSlice, VPackSlice>> const& collections,
bool incremental, SyncPhase phase) {
std::string phaseMsg("starting phase " + translatePhase(phase) + " with " +
std::to_string(collections.size()) + " collections");
_config.progress.set(phaseMsg);
for (auto const& collection : collections) {
VPackSlice const parameters = collection.first;
VPackSlice const indexes = collection.second;
Result res = handleCollection(parameters, indexes, incremental, phase);
if (res.fail()) {
return res;
}
}
// all ok
return Result();
}
/// @brief create non-existing views locally
Result DatabaseInitialSyncer::handleViewCreation(VPackSlice const& views) {
for (VPackSlice slice : VPackArrayIterator(views)) {
Result res = createView(vocbase(), slice);
if (res.fail()) {
return res;
}
}
return {};
}
Result DatabaseInitialSyncer::batchStart(std::string const& patchCount) {
return _config.batch.start(_config.connection, _config.progress, _config.state.syncerId, patchCount);
}
Result DatabaseInitialSyncer::batchExtend() {
return _config.batch.extend(_config.connection, _config.progress, _config.state.syncerId);
}
Result DatabaseInitialSyncer::batchFinish() {
return _config.batch.finish(_config.connection, _config.progress, _config.state.syncerId);
}
} // namespace arangodb
Reference in New Issue View Git Blame Copy Permalink