mirror of https://gitee.com/bigwinds/arangodb
240 lines
9.1 KiB
C++
240 lines
9.1 KiB
C++
////////////////////////////////////////////////////////////////////////////////
|
|
/// 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
|
|
////////////////////////////////////////////////////////////////////////////////
|
|
|
|
#ifndef ARANGOD_REPLICATION_DATABASE_INITIAL_SYNCER_H
|
|
#define ARANGOD_REPLICATION_DATABASE_INITIAL_SYNCER_H 1
|
|
|
|
#include "Basics/Result.h"
|
|
#include "Cluster/ServerState.h"
|
|
#include "Replication/InitialSyncer.h"
|
|
#include "Replication/utilities.h"
|
|
#include "Utils/SingleCollectionTransaction.h"
|
|
|
|
struct TRI_vocbase_t;
|
|
|
|
namespace arangodb {
|
|
|
|
class LogicalCollection;
|
|
class DatabaseInitialSyncer;
|
|
class ReplicationApplierConfiguration;
|
|
|
|
class DatabaseInitialSyncer final : public InitialSyncer {
|
|
friend ::arangodb::Result handleSyncKeysMMFiles(DatabaseInitialSyncer& syncer,
|
|
arangodb::LogicalCollection* col,
|
|
std::string const& keysId);
|
|
friend ::arangodb::Result handleSyncKeysRocksDB(DatabaseInitialSyncer& syncer,
|
|
arangodb::LogicalCollection* col,
|
|
std::string const& keysId);
|
|
friend ::arangodb::Result syncChunkRocksDB(
|
|
DatabaseInitialSyncer& syncer, SingleCollectionTransaction* trx,
|
|
InitialSyncerIncrementalSyncStats& stats, std::string const& keysId,
|
|
uint64_t chunkId, std::string const& lowString,
|
|
std::string const& highString, std::vector<std::string> const& markers);
|
|
|
|
public:
|
|
/// @brief apply phases
|
|
typedef enum {
|
|
PHASE_NONE,
|
|
PHASE_INIT,
|
|
PHASE_VALIDATE,
|
|
PHASE_DROP_CREATE,
|
|
PHASE_DUMP
|
|
} SyncPhase;
|
|
|
|
struct Configuration {
|
|
/// @brief replication applier config (from the base Syncer)
|
|
ReplicationApplierConfiguration const& applier;
|
|
/// @brief the compaction barrier state (from the base Syncer)
|
|
replutils::BarrierInfo& barrier; // TODO worker safety
|
|
/// @brief the dump batch state (from the base InitialSyncer)
|
|
replutils::BatchInfo& batch; // TODO worker safety
|
|
/// @brief the client connection (from the base Syncer)
|
|
replutils::Connection& connection;
|
|
/// @brief whether or not we have flushed the WAL on the remote server
|
|
bool flushed; // TODO worker safety
|
|
/// @brief info about master node (from the base Syncer)
|
|
replutils::MasterInfo& master;
|
|
/// @brief the progress info (from the base InitialSyncer)
|
|
replutils::ProgressInfo& progress; // TODO worker safety
|
|
/// @brief the syncer state (from the base Syncer)
|
|
SyncerState& state;
|
|
/// @brief the database to dump
|
|
TRI_vocbase_t& vocbase;
|
|
|
|
explicit Configuration(ReplicationApplierConfiguration const&,
|
|
replutils::BarrierInfo&, replutils::BatchInfo&,
|
|
replutils::Connection&, bool, replutils::MasterInfo&,
|
|
replutils::ProgressInfo&, SyncerState& state, TRI_vocbase_t&);
|
|
|
|
bool isChild() const; // TODO worker safety
|
|
};
|
|
|
|
DatabaseInitialSyncer(TRI_vocbase_t& vocbase,
|
|
ReplicationApplierConfiguration const& configuration);
|
|
|
|
/// @brief run method, performs a full synchronization
|
|
Result run(bool incremental) override {
|
|
return runWithInventory(incremental, velocypack::Slice::noneSlice());
|
|
}
|
|
|
|
/// @brief run method, performs a full synchronization with the
|
|
/// given list of collections.
|
|
Result runWithInventory(bool incremental, velocypack::Slice collections);
|
|
|
|
TRI_vocbase_t* resolveVocbase(velocypack::Slice const& slice) override {
|
|
return &_config.vocbase;
|
|
}
|
|
|
|
/// @brief translate a phase to a phase name
|
|
std::string translatePhase(SyncPhase phase) const {
|
|
switch (phase) {
|
|
case PHASE_INIT:
|
|
return "init";
|
|
case PHASE_VALIDATE:
|
|
return "validate";
|
|
case PHASE_DROP_CREATE:
|
|
return "drop-create";
|
|
case PHASE_DUMP:
|
|
return "dump";
|
|
case PHASE_NONE:
|
|
break;
|
|
}
|
|
return "none";
|
|
}
|
|
|
|
// TODO worker safety
|
|
TRI_vocbase_t& vocbase() const {
|
|
TRI_ASSERT(vocbases().size() == 1);
|
|
return vocbases().begin()->second.database();
|
|
}
|
|
|
|
/// @brief check whether the initial synchronization should be aborted
|
|
// TODO worker-safety
|
|
bool isAborted() const override;
|
|
|
|
/// @brief insert the batch id and barrier ID.
|
|
/// For use in globalinitialsyncer
|
|
// TODO worker safety
|
|
void useAsChildSyncer(replutils::MasterInfo const& info, SyncerId const syncerId, uint64_t barrierId,
|
|
double barrierUpdateTime, uint64_t batchId, double batchUpdateTime) {
|
|
_state.syncerId = syncerId;
|
|
_state.isChildSyncer = true;
|
|
_state.master = info;
|
|
_state.barrier.id = barrierId;
|
|
_state.barrier.updateTime = barrierUpdateTime;
|
|
_config.batch.id = batchId;
|
|
_config.batch.updateTime = batchUpdateTime;
|
|
}
|
|
|
|
/// @brief last time the barrier was extended or created
|
|
/// The barrier prevents the deletion of WAL files for mmfiles
|
|
double barrierUpdateTime() const { return _state.barrier.updateTime; }
|
|
|
|
/// @brief last time the batch was extended or created
|
|
/// The batch prevents compaction in mmfiles and keeps a snapshot
|
|
/// in rocksdb for a constant view of the data
|
|
double batchUpdateTime() const { return _config.batch.updateTime; }
|
|
|
|
/// @brief fetch the server's inventory, public method
|
|
Result getInventory(arangodb::velocypack::Builder& builder);
|
|
|
|
private:
|
|
/// @brief order a new chunk from the /dump API
|
|
void 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);
|
|
|
|
/// @brief fetch the server's inventory
|
|
Result fetchInventory(arangodb::velocypack::Builder& builder);
|
|
|
|
/// @brief set a progress message
|
|
void setProgress(std::string const& msg);
|
|
|
|
/// @brief send a WAL flush command
|
|
Result sendFlush();
|
|
|
|
/// @brief handle a single dump marker
|
|
// TODO worker-safety
|
|
Result parseCollectionDumpMarker(transaction::Methods&, arangodb::LogicalCollection*,
|
|
arangodb::velocypack::Slice const&);
|
|
|
|
/// @brief apply the data from a collection dump
|
|
// TODO worker-safety
|
|
Result parseCollectionDump(transaction::Methods&, LogicalCollection* col,
|
|
httpclient::SimpleHttpResult*, uint64_t&);
|
|
|
|
/// @brief determine the number of documents in a collection
|
|
int64_t getSize(arangodb::LogicalCollection const& col);
|
|
|
|
/// @brief incrementally fetch data from a collection
|
|
// TODO worker safety
|
|
Result fetchCollectionDump(arangodb::LogicalCollection*,
|
|
std::string const& leaderColl, TRI_voc_tick_t);
|
|
|
|
/// @brief incrementally fetch data from a collection
|
|
// TODO worker safety
|
|
Result fetchCollectionSync(arangodb::LogicalCollection*,
|
|
std::string const& leaderColl, TRI_voc_tick_t);
|
|
|
|
/// @brief changes the properties of a collection, based on the VelocyPack
|
|
/// provided
|
|
Result changeCollection(arangodb::LogicalCollection*, arangodb::velocypack::Slice const&);
|
|
|
|
/// @brief handle the information about a collection
|
|
// TODO worker-safety
|
|
Result handleCollection(arangodb::velocypack::Slice const&,
|
|
arangodb::velocypack::Slice const&, bool incremental, SyncPhase);
|
|
|
|
/// @brief handle the inventory response of the master
|
|
Result handleCollectionsAndViews(arangodb::velocypack::Slice const& colls,
|
|
arangodb::velocypack::Slice const& views,
|
|
bool incremental);
|
|
|
|
/// @brief iterate over all collections from an array and apply an action
|
|
Result iterateCollections(
|
|
std::vector<std::pair<arangodb::velocypack::Slice, arangodb::velocypack::Slice>> const&,
|
|
bool incremental, SyncPhase);
|
|
|
|
/// @brief create non-existing views locally
|
|
Result handleViewCreation(VPackSlice const& views);
|
|
|
|
/// @brief send a "start batch" command
|
|
/// @param patchCount (optional)
|
|
/// Try to patch count of this collection (must be a collection name).
|
|
/// Only effective with the incremental sync.
|
|
Result batchStart(std::string const& patchCount = "");
|
|
|
|
/// @brief send an "extend batch" command
|
|
Result batchExtend();
|
|
|
|
/// @brief send a "finish batch" command
|
|
Result batchFinish();
|
|
|
|
Configuration _config;
|
|
};
|
|
|
|
} // namespace arangodb
|
|
|
|
#endif
|