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

653 lines
21 KiB
C++
Raw Blame History

////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2017 ArangoDB 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 "TtlFeature.h"
#include "Aql/Query.h"
#include "Aql/QueryRegistry.h"
#include "ApplicationFeatures/ApplicationServer.h"
#include "Basics/ConditionLocker.h"
#include "Basics/ConditionVariable.h"
#include "Basics/Exceptions.h"
#include "Basics/MutexLocker.h"
#include "Basics/Thread.h"
#include "Cluster/FollowerInfo.h"
#include "Cluster/ServerState.h"
#include "Indexes/Index.h"
#include "Logger/Logger.h"
#include "ProgramOptions/Parameters.h"
#include "ProgramOptions/ProgramOptions.h"
#include "RestServer/DatabaseFeature.h"
#include "RestServer/QueryRegistryFeature.h"
#include "VocBase/LogicalCollection.h"
#include "VocBase/vocbase.h"
#include <velocypack/Builder.h>
#include <velocypack/Slice.h>
#include <velocypack/velocypack-aliases.h>
#include <chrono>
#include <thread>
using namespace arangodb;
using namespace arangodb::options;
namespace {
// the AQL query to remove documents
std::string const removeQuery("FOR doc IN @@collection FILTER doc.@indexAttribute >= 0 && doc.@indexAttribute <= @stamp SORT doc.@indexAttribute LIMIT @limit REMOVE doc IN @@collection OPTIONS { ignoreErrors: true }");
}
namespace arangodb {
TtlStatistics& TtlStatistics::operator+=(VPackSlice const& other) {
if (!other.isObject()) {
THROW_ARANGO_EXCEPTION_MESSAGE(TRI_ERROR_INTERNAL, "expecting object for statistics");
}
if (other.hasKey("runs")) {
runs += other.get("runs").getNumericValue<uint64_t>();
}
if (other.hasKey("documentsRemoved")) {
documentsRemoved += other.get("documentsRemoved").getNumericValue<uint64_t>();
}
if (other.hasKey("limitReached")) {
limitReached += other.get("limitReached").getNumericValue<uint64_t>();
}
return *this;
}
void TtlStatistics::toVelocyPack(VPackBuilder& builder) const {
builder.openObject();
builder.add("runs", VPackValue(runs));
builder.add("documentsRemoved", VPackValue(documentsRemoved));
builder.add("limitReached", VPackValue(limitReached));
builder.close();
}
void TtlProperties::toVelocyPack(VPackBuilder& builder, bool isActive) const {
builder.openObject();
builder.add("frequency", VPackValue(frequency));
builder.add("maxTotalRemoves", VPackValue(maxTotalRemoves));
builder.add("maxCollectionRemoves", VPackValue(maxCollectionRemoves));
builder.add("onlyLoadedCollections", VPackValue(onlyLoadedCollections));
builder.add("active", VPackValue(isActive));
builder.close();
}
Result TtlProperties::fromVelocyPack(VPackSlice const& slice) {
if (!slice.isObject()) {
return Result(TRI_ERROR_BAD_PARAMETER, "expecting object for properties");
}
try {
uint64_t frequency = this->frequency;
uint64_t maxTotalRemoves = this->maxTotalRemoves;
uint64_t maxCollectionRemoves = this->maxCollectionRemoves;
uint64_t onlyLoadedCollections = this->onlyLoadedCollections;
if (slice.hasKey("frequency")) {
if (!slice.get("frequency").isNumber()) {
return Result(TRI_ERROR_BAD_PARAMETER, "expecting numeric value for frequency");
}
frequency = slice.get("frequency").getNumericValue<uint64_t>();
if (frequency < TtlProperties::minFrequency) {
return Result(TRI_ERROR_BAD_PARAMETER, "too low value for frequency");
}
}
if (slice.hasKey("maxTotalRemoves")) {
if (!slice.get("maxTotalRemoves").isNumber()) {
return Result(TRI_ERROR_BAD_PARAMETER, "expecting numeric value for maxTotalRemoves");
}
maxTotalRemoves = slice.get("maxTotalRemoves").getNumericValue<uint64_t>();
}
if (slice.hasKey("maxCollectionRemoves")) {
if (!slice.get("maxCollectionRemoves").isNumber()) {
return Result(TRI_ERROR_BAD_PARAMETER, "expecting numeric value for maxCollectionRemoves");
}
maxCollectionRemoves = slice.get("maxCollectionRemoves").getNumericValue<uint64_t>();
}
if (slice.hasKey("onlyLoadedCollections")) {
if (!slice.get("onlyLoadedCollections").isBool()) {
return Result(TRI_ERROR_BAD_PARAMETER, "expecting boolean value for onlyLoadedCollections");
}
onlyLoadedCollections = slice.get("onlyLoadedCollections").getBool();
}
this->frequency = frequency;
this->maxTotalRemoves = maxTotalRemoves;
this->maxCollectionRemoves = maxCollectionRemoves;
this->onlyLoadedCollections = onlyLoadedCollections;
return Result();
} catch (arangodb::basics::Exception const& ex) {
return Result(ex.code(), ex.what());
} catch (std::exception const& ex) {
return Result(TRI_ERROR_INTERNAL, ex.what());
}
}
class TtlThread final : public Thread {
public:
explicit TtlThread(TtlFeature* ttlFeature)
: Thread("TTL"),
_ttlFeature(ttlFeature),
_working(false) {
TRI_ASSERT(_ttlFeature != nullptr);
}
~TtlThread() { shutdown(); }
void beginShutdown() override {
Thread::beginShutdown();
// wake up the thread that may be waiting in run()
wakeup();
}
void wakeup() {
// wake up the thread that may be waiting in run()
CONDITION_LOCKER(guard, _condition);
guard.signal();
}
bool isCurrentlyWorking() const {
return _working.load();
}
/// @brief frequency is specified in milliseconds
void setNextStart(uint64_t frequency) {
CONDITION_LOCKER(guard, _condition);
_nextStart = std::chrono::steady_clock::now() + std::chrono::milliseconds(frequency);
}
protected:
void run() override {
TtlProperties properties = _ttlFeature->properties();
setNextStart(properties.frequency);
LOG_TOPIC("c2be7", TRACE, Logger::TTL) << "starting TTL background thread with interval " << properties.frequency << " milliseconds, max removals per run: " << properties.maxTotalRemoves << ", max removals per collection per run " << properties.maxCollectionRemoves;
while (true) {
auto now = std::chrono::steady_clock::now();
while (now < _nextStart) {
if (isStopping()) {
// server shutdown
return;
}
// wait for our start...
CONDITION_LOCKER(guard, _condition);
guard.wait(std::chrono::microseconds(std::chrono::duration_cast<std::chrono::microseconds>(_nextStart - now)));
now = std::chrono::steady_clock::now();
}
if (isStopping()) {
// server shutdown
return;
}
// properties may have changed... update them
properties = _ttlFeature->properties();
setNextStart(properties.frequency);
try {
TtlStatistics stats;
// note: work() will do nothing if isActive() is false
work(stats, properties);
// merge stats
_ttlFeature->updateStats(stats);
} catch (std::exception const& ex) {
LOG_TOPIC("6d28a", WARN, Logger::TTL) << "caught exception in TTL background thread: " << ex.what();
} catch (...) {
LOG_TOPIC("44aa8", WARN, Logger::TTL) << "caught unknown exception in TTL background thread";
}
}
}
private:
/// @brief whether or not the background thread shall continue working
bool isActive() const {
return _ttlFeature->isActive() && !isStopping();
}
void work(TtlStatistics& stats, TtlProperties const& properties) {
if (!isActive()) {
return;
}
// mark ourselves as busy
_working = true;
auto guard = scopeGuard([this]() { _working = false; });
LOG_TOPIC("139af", TRACE, Logger::TTL) << "ttl thread work()";
stats.runs++;
auto queryRegistryFeature =
application_features::ApplicationServer::getFeature<QueryRegistryFeature>(
"QueryRegistry");
auto queryRegistry = queryRegistryFeature->queryRegistry();
double const stamp = TRI_microtime();
uint64_t limitLeft = properties.maxTotalRemoves;
// iterate over all databases
auto db = DatabaseFeature::DATABASE;
for (auto const& name : db->getDatabaseNames()) {
if (!isActive()) {
// feature deactivated (for example, due to running on current follower in
// active failover setup)
return;
}
TRI_vocbase_t* vocbase = db->useDatabase(name);
if (vocbase == nullptr) {
continue;
}
// make sure we decrease the reference counter later
TRI_DEFER(vocbase->release());
LOG_TOPIC("ec905", TRACE, Logger::TTL) << "TTL thread going to process database '" << vocbase->name() << "'";
std::vector<std::shared_ptr<arangodb::LogicalCollection>> collections = vocbase->collections(false);
for (auto const& collection : collections) {
if (!isActive()) {
// feature deactivated (for example, due to running on current follower in
// active failover setup)
return;
}
if (properties.onlyLoadedCollections &&
collection->status() != TRI_VOC_COL_STATUS_LOADED) {
// we only care about collections that are already loaded here.
// otherwise our check may load them all into memory, and sure this is
// something we want to avoid here
continue;
}
if (ServerState::instance()->isDBServer() &&
!collection->followers()->getLeader().empty()) {
// we are a follower for this shard. do not remove any data here, but
// let the leader carry out the removal and replicate it
continue;
}
std::vector<std::shared_ptr<Index>> indexes = collection->getIndexes();
for (auto const& index : indexes) {
// we are only interested in collections with TTL indexes
if (index->type() != Index::TRI_IDX_TYPE_TTL_INDEX) {
continue;
}
// serialize the index description so we can read the "expireAfter" attribute
_builder.clear();
index->toVelocyPack(_builder, Index::makeFlags());
VPackSlice ea = _builder.slice().get(StaticStrings::IndexExpireAfter);
if (!ea.isNumber()) {
// index description somehow invalid
continue;
}
double expireAfter = ea.getNumericValue<double>();
LOG_TOPIC("5cca5", DEBUG, Logger::TTL) << "TTL thread going to work for collection '" << collection->name() << "', expireAfter: " << Logger::FIXED(expireAfter, 0) << ", stamp: " << (stamp - expireAfter) << ", limit: " << std::min(properties.maxCollectionRemoves, limitLeft);
auto bindVars = std::make_shared<VPackBuilder>();
bindVars->openObject();
bindVars->add("@collection", VPackValue(collection->name()));
bindVars->add(VPackValue("indexAttribute"));
bindVars->openArray();
for (auto const& it : index->fields()[0]) {
bindVars->add(VPackValue(it.name));
}
bindVars->close();
bindVars->add("stamp", VPackValue(stamp - expireAfter));
bindVars->add("limit", VPackValue(std::min(properties.maxCollectionRemoves, limitLeft)));
bindVars->close();
aql::Query query(false, *vocbase, aql::QueryString(::removeQuery), bindVars, nullptr, arangodb::aql::PART_MAIN);
aql::QueryResult queryResult = query.executeSync(queryRegistry);
if (queryResult.result.fail()) {
// we can probably live with an error here...
// the thread will try to remove the documents again on next iteration
if (!queryResult.result.is(TRI_ERROR_ARANGO_READ_ONLY) &&
!queryResult.result.is(TRI_ERROR_ARANGO_CONFLICT) &&
!queryResult.result.is(TRI_ERROR_LOCKED) &&
!queryResult.result.is(TRI_ERROR_ARANGO_DATA_SOURCE_NOT_FOUND)) {
LOG_TOPIC("08300", WARN, Logger::TTL) << "error during TTL document removal for collection '" << collection->name() << "': " << queryResult.result.errorMessage();
}
} else {
auto extra = queryResult.extra;
if (extra != nullptr) {
VPackSlice v = extra->slice().get("stats");
if (v.isObject()) {
v = v.get("writesExecuted");
if (v.isNumber()) {
uint64_t removed = v.getNumericValue<uint64_t>();
stats.documentsRemoved += removed;
if (removed > 0) {
LOG_TOPIC("2455e", DEBUG, Logger::TTL) << "TTL thread removed " << removed << " documents for collection '" << collection->name() << "'";
if (limitLeft >= removed) {
limitLeft -= removed;
} else {
limitLeft = 0;
}
}
}
}
}
}
// there can only be one TTL index per collection, so we can abort the loop here
break;
}
if (limitLeft == 0) {
// removed as much as we are allowed to. now stop and remove more in next iteration
++stats.limitReached;
return;
}
if (isStopping()) {
// server has been stopped, so abort our loop(s)
return;
}
}
}
}
private:
TtlFeature* _ttlFeature;
arangodb::basics::ConditionVariable _condition;
/// @brief next time the thread should run
std::chrono::time_point<std::chrono::steady_clock> _nextStart;
/// @brief a builder object we reuse to save a few memory allocations
VPackBuilder _builder;
/// @brief set to true while the TTL thread is actually performing deletions,
/// false otherwise
std::atomic<bool> _working;
};
}
TtlFeature::TtlFeature(
application_features::ApplicationServer& server
)
: ApplicationFeature(server, "Ttl"),
_allowRunning(true),
_active(true) {
startsAfter("DatabasePhase");
startsAfter("ServerPhase");
}
TtlFeature::~TtlFeature() {
shutdownThread();
}
void TtlFeature::collectOptions(std::shared_ptr<ProgramOptions> options) {
options->addSection("ttl", "TTL options");
options->addOption("ttl.frequency", "frequency (in milliseconds) for the TTL background thread invocation. "
"a value of 0 turns the TTL background thread off entirely",
new UInt64Parameter(&_properties.frequency));
options->addOption("ttl.max-total-removes", "maximum number of documents to remove per invocation of the TTL thread",
new UInt64Parameter(&_properties.maxTotalRemoves));
options->addOption("ttl.max-collection-removes", "maximum number of documents to remove per collection",
new UInt64Parameter(&_properties.maxCollectionRemoves));
options->addOption("ttl.only-loaded-collection", "only consider already loaded collections for removal",
new BooleanParameter(&_properties.onlyLoadedCollections));
}
void TtlFeature::validateOptions(std::shared_ptr<ProgramOptions> options) {
if (_properties.maxTotalRemoves == 0) {
LOG_TOPIC("1e299", FATAL, arangodb::Logger::STARTUP)
<< "invalid value for '--ttl.max-total-removes'.";
FATAL_ERROR_EXIT();
}
if (_properties.maxCollectionRemoves == 0) {
LOG_TOPIC("2ab82", FATAL, arangodb::Logger::STARTUP)
<< "invalid value for '--ttl.max-collection-removes'.";
FATAL_ERROR_EXIT();
}
if (_properties.frequency < TtlProperties::minFrequency) {
LOG_TOPIC("ea696", FATAL, arangodb::Logger::STARTUP)
<< "too low value for '--ttl.frequency'.";
FATAL_ERROR_EXIT();
}
}
void TtlFeature::start() {
// the thread will not run on a coordinator or an agency node,
// just locally on DB servers or single servers
if (ServerState::instance()->isCoordinator() ||
ServerState::instance()->isAgent()) {
LOG_TOPIC("e94bb", DEBUG, Logger::TTL) << "turning off TTL feature because of coordinator / agency";
return;
}
DatabaseFeature* databaseFeature =
application_features::ApplicationServer::getFeature<DatabaseFeature>(
"Database");
if (databaseFeature->checkVersion() || databaseFeature->upgrade()) {
LOG_TOPIC("5614a", DEBUG, Logger::TTL) << "turning off TTL feature because of version checking or upgrade procedure";
return;
}
// a frequency of 0 means the thread is not started at all
if (_properties.frequency == 0) {
return;
}
MUTEX_LOCKER(locker, _threadMutex);
if (application_features::ApplicationServer::isStopping()) {
// don't create the thread if we are already shutting down
return;
}
_thread.reset(new TtlThread(this));
if (!_thread->start()) {
LOG_TOPIC("33c33", FATAL, Logger::TTL) << "could not start ttl background thread";
FATAL_ERROR_EXIT();
}
}
void TtlFeature::beginShutdown() {
// this will make the TTL background thread stop as soon as possible
deactivate();
MUTEX_LOCKER(locker, _threadMutex);
if (_thread != nullptr) {
// this will also wake up the thread if it should be sleeping
_thread->beginShutdown();
}
}
void TtlFeature::stop() {
shutdownThread();
}
void TtlFeature::allowRunning(bool value) {
{
MUTEX_LOCKER(locker, _propertiesMutex);
if (value) {
_allowRunning = true;
} else {
_allowRunning = false;
}
}
if (value) {
return;
}
waitForThreadWork();
}
void TtlFeature::waitForThreadWork() {
while (true) {
{
MUTEX_LOCKER(locker, _threadMutex);
if (_thread == nullptr) {
break;
}
_thread->wakeup();
if (!_thread->isCurrentlyWorking()) {
break;
}
}
std::this_thread::sleep_for(std::chrono::milliseconds(5));
}
}
void TtlFeature::activate() {
{
MUTEX_LOCKER(locker, _propertiesMutex);
if (_active) {
// already activated
return;
}
_active = true;
}
LOG_TOPIC("79862", DEBUG, Logger::TTL) << "activated TTL background thread";
}
void TtlFeature::deactivate() {
{
MUTEX_LOCKER(locker, _propertiesMutex);
if (!_active) {
// already deactivated
return;
}
_active = false;
}
waitForThreadWork();
LOG_TOPIC("898a7", DEBUG, Logger::TTL) << "deactivated TTL background thread";
}
bool TtlFeature::isActive() const {
MUTEX_LOCKER(locker, _propertiesMutex);
return _allowRunning && _active;
}
void TtlFeature::statsToVelocyPack(VPackBuilder& builder) const {
MUTEX_LOCKER(locker, _statisticsMutex);
_statistics.toVelocyPack(builder);
}
void TtlFeature::updateStats(TtlStatistics const& stats) {
MUTEX_LOCKER(locker, _statisticsMutex);
_statistics += stats;
}
void TtlFeature::propertiesToVelocyPack(VPackBuilder& builder) const {
MUTEX_LOCKER(locker, _propertiesMutex);
_properties.toVelocyPack(builder, _active);
}
TtlProperties TtlFeature::properties() const {
MUTEX_LOCKER(locker, _propertiesMutex);
return _properties;
}
Result TtlFeature::propertiesFromVelocyPack(VPackSlice const& slice, VPackBuilder& out) {
Result res;
uint64_t frequency;
bool active;
{
MUTEX_LOCKER(locker, _propertiesMutex);
bool const hasActiveFlag = slice.isObject() && slice.hasKey("active");
if (hasActiveFlag && !slice.get("active").isBool()) {
return Result(TRI_ERROR_BAD_PARAMETER, "active flag should be a boolean value");
}
// store properties
res = _properties.fromVelocyPack(slice);
if (!res.fail() && hasActiveFlag) {
// update active flag
_active = slice.get("active").getBool();
}
_properties.toVelocyPack(out, _active);
frequency = _properties.frequency;
active = _active;
}
{
MUTEX_LOCKER(locker, _threadMutex);
if (_thread != nullptr) {
_thread->setNextStart(frequency);
_thread->wakeup();
while (!active && _thread->isCurrentlyWorking()) {
std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
}
}
return res;
}
void TtlFeature::shutdownThread() noexcept {
MUTEX_LOCKER(locker, _threadMutex);
if (_thread != nullptr) {
try {
_thread->beginShutdown();
while (_thread->isRunning()) {
std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
} catch (...) {
}
_thread.reset();
}
}
Reference in New Issue View Git Blame Copy Permalink