<commit_before>// Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.

#include "bucketdbupdater.h"
#include "bucket_db_prune_elision.h"
#include "distributor.h"
#include "distributor_bucket_space.h"
#include "distributormetricsset.h"
#include "simpleclusterinformation.h"
#include <vespa/storage/common/bucketoperationlogger.h>
#include <vespa/storageapi/message/persistence.h>
#include <vespa/storageapi/message/removelocation.h>
#include <vespa/vespalib/util/xmlstream.h>

#include <vespa/log/bufferedlogger.h>
LOG_SETUP(".distributor.bucketdb.updater");

using storage::lib::Node;
using storage::lib::NodeType;
using document::BucketSpace;

namespace storage::distributor {

BucketDBUpdater::BucketDBUpdater(Distributor& owner,
                                 DistributorBucketSpaceRepo& bucketSpaceRepo,
                                 DistributorBucketSpaceRepo& readOnlyBucketSpaceRepo,
                                 DistributorMessageSender& sender,
                                 DistributorComponentRegister& compReg)
    : framework::StatusReporter("bucketdb", "Bucket DB Updater"),
      _distributorComponent(owner, bucketSpaceRepo, readOnlyBucketSpaceRepo, compReg, "Bucket DB Updater"),
      _sender(sender),
      _transitionTimer(_distributorComponent.getClock())
{
}

BucketDBUpdater::~BucketDBUpdater() = default;

void
BucketDBUpdater::flush()
{
    for (auto & entry : _sentMessages) {
        // Cannot sendDown MergeBucketReplies during flushing, since
        // all lower links have been closed
        if (entry.second._mergeReplyGuard) {
            entry.second._mergeReplyGuard->resetReply();
        }
    }
    _sentMessages.clear();
}

void
BucketDBUpdater::print(std::ostream& out, bool verbose, const std::string& indent) const
{
    (void) verbose; (void) indent;
    out << "BucketDBUpdater";
}

bool
BucketDBUpdater::shouldDeferStateEnabling() const noexcept
{
    return _distributorComponent.getDistributor().getConfig()
            .allowStaleReadsDuringClusterStateTransitions();
}

bool
BucketDBUpdater::hasPendingClusterState() const
{
    return static_cast<bool>(_pendingClusterState);
}

BucketOwnership
BucketDBUpdater::checkOwnershipInPendingState(const document::Bucket& b) const
{
    if (hasPendingClusterState()) {
        const auto& state(*_pendingClusterState->getNewClusterStateBundle().getDerivedClusterState(b.getBucketSpace()));
        if (!_distributorComponent.ownsBucketInState(state, b)) {
            return BucketOwnership::createNotOwnedInState(state);
        }
    }
    return BucketOwnership::createOwned();
}

const lib::ClusterState*
BucketDBUpdater::pendingClusterStateOrNull(const document::BucketSpace& space) const {
    return (hasPendingClusterState()
            ? _pendingClusterState->getNewClusterStateBundle().getDerivedClusterState(space).get()
            : nullptr);
}

void
BucketDBUpdater::sendRequestBucketInfo(
        uint16_t node,
        const document::Bucket& bucket,
        const std::shared_ptr<MergeReplyGuard>& mergeReplyGuard)
{
    if (!_distributorComponent.storageNodeIsUp(bucket.getBucketSpace(), node)) {
        return;
    }

    std::vector<document::BucketId> buckets;
    buckets.push_back(bucket.getBucketId());

    std::shared_ptr<api::RequestBucketInfoCommand> msg(
            new api::RequestBucketInfoCommand(bucket.getBucketSpace(), buckets));

    LOG(debug,
        "Sending request bucket info command %" PRIu64 " for "
        "bucket %s to node %u",
        msg->getMsgId(),
        bucket.toString().c_str(),
        node);

    msg->setPriority(50);
    msg->setAddress(_distributorComponent.nodeAddress(node));

    _sentMessages[msg->getMsgId()] =
        BucketRequest(node, _distributorComponent.getUniqueTimestamp(),
                      bucket, mergeReplyGuard);
    _sender.sendCommand(msg);
}

void
BucketDBUpdater::recheckBucketInfo(uint32_t nodeIdx,
                                   const document::Bucket& bucket)
{
    sendRequestBucketInfo(nodeIdx, bucket, std::shared_ptr<MergeReplyGuard>());
}

void
BucketDBUpdater::removeSuperfluousBuckets(
        const lib::ClusterStateBundle& newState,
        bool is_distribution_config_change)
{
    const bool move_to_read_only_db = shouldDeferStateEnabling();
    const char* up_states = _distributorComponent.getDistributor().getStorageNodeUpStates();
    for (auto& elem : _distributorComponent.getBucketSpaceRepo()) {
        const auto& newDistribution(elem.second->getDistribution());
        const auto& oldClusterState(elem.second->getClusterState());
        const auto& new_cluster_state = newState.getDerivedClusterState(elem.first);

        // Running a full DB sweep is expensive, so if the cluster state transition does
        // not actually indicate that buckets should possibly be removed, we elide it entirely.
        if (!is_distribution_config_change
            && db_pruning_may_be_elided(oldClusterState, *new_cluster_state, up_states))
        {
            LOG(debug, "Eliding DB pruning for state transition '%s' -> '%s'",
                oldClusterState.toString().c_str(), new_cluster_state->toString().c_str());
            continue;
        }

        auto& bucketDb(elem.second->getBucketDatabase());
        auto& readOnlyDb(_distributorComponent.getReadOnlyBucketSpaceRepo().get(elem.first).getBucketDatabase());

        // Remove all buckets not belonging to this distributor, or
        // being on storage nodes that are no longer up.
        MergingNodeRemover proc(
                oldClusterState,
                *new_cluster_state,
                _distributorComponent.getIndex(),
                newDistribution,
                up_states,
                move_to_read_only_db);

        bucketDb.merge(proc);
        // Non-owned entries vector only has contents if move_to_read_only_db is true.
        // TODO either rewrite in terms of merge() or remove entirely
        for (const auto& db_entry : proc.getNonOwnedEntries()) {
            readOnlyDb.update(db_entry); // TODO Entry move support
        }
    }
}

void
BucketDBUpdater::ensureTransitionTimerStarted()
{<commit_msg>Promote log level for DB pruning elision decisions<commit_after>// Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.

#include "bucketdbupdater.h"
#include "bucket_db_prune_elision.h"
#include "distributor.h"
#include "distributor_bucket_space.h"
#include "distributormetricsset.h"
#include "simpleclusterinformation.h"
#include <vespa/document/bucket/fixed_bucket_spaces.h>
#include <vespa/storage/common/bucketoperationlogger.h>
#include <vespa/storageapi/message/persistence.h>
#include <vespa/storageapi/message/removelocation.h>
#include <vespa/vespalib/util/xmlstream.h>

#include <vespa/log/bufferedlogger.h>
LOG_SETUP(".distributor.bucketdb.updater");

using storage::lib::Node;
using storage::lib::NodeType;
using document::BucketSpace;

namespace storage::distributor {

BucketDBUpdater::BucketDBUpdater(Distributor& owner,
                                 DistributorBucketSpaceRepo& bucketSpaceRepo,
                                 DistributorBucketSpaceRepo& readOnlyBucketSpaceRepo,
                                 DistributorMessageSender& sender,
                                 DistributorComponentRegister& compReg)
    : framework::StatusReporter("bucketdb", "Bucket DB Updater"),
      _distributorComponent(owner, bucketSpaceRepo, readOnlyBucketSpaceRepo, compReg, "Bucket DB Updater"),
      _sender(sender),
      _transitionTimer(_distributorComponent.getClock())
{
}

BucketDBUpdater::~BucketDBUpdater() = default;

void
BucketDBUpdater::flush()
{
    for (auto & entry : _sentMessages) {
        // Cannot sendDown MergeBucketReplies during flushing, since
        // all lower links have been closed
        if (entry.second._mergeReplyGuard) {
            entry.second._mergeReplyGuard->resetReply();
        }
    }
    _sentMessages.clear();
}

void
BucketDBUpdater::print(std::ostream& out, bool verbose, const std::string& indent) const
{
    (void) verbose; (void) indent;
    out << "BucketDBUpdater";
}

bool
BucketDBUpdater::shouldDeferStateEnabling() const noexcept
{
    return _distributorComponent.getDistributor().getConfig()
            .allowStaleReadsDuringClusterStateTransitions();
}

bool
BucketDBUpdater::hasPendingClusterState() const
{
    return static_cast<bool>(_pendingClusterState);
}

BucketOwnership
BucketDBUpdater::checkOwnershipInPendingState(const document::Bucket& b) const
{
    if (hasPendingClusterState()) {
        const auto& state(*_pendingClusterState->getNewClusterStateBundle().getDerivedClusterState(b.getBucketSpace()));
        if (!_distributorComponent.ownsBucketInState(state, b)) {
            return BucketOwnership::createNotOwnedInState(state);
        }
    }
    return BucketOwnership::createOwned();
}

const lib::ClusterState*
BucketDBUpdater::pendingClusterStateOrNull(const document::BucketSpace& space) const {
    return (hasPendingClusterState()
            ? _pendingClusterState->getNewClusterStateBundle().getDerivedClusterState(space).get()
            : nullptr);
}

void
BucketDBUpdater::sendRequestBucketInfo(
        uint16_t node,
        const document::Bucket& bucket,
        const std::shared_ptr<MergeReplyGuard>& mergeReplyGuard)
{
    if (!_distributorComponent.storageNodeIsUp(bucket.getBucketSpace(), node)) {
        return;
    }

    std::vector<document::BucketId> buckets;
    buckets.push_back(bucket.getBucketId());

    std::shared_ptr<api::RequestBucketInfoCommand> msg(
            new api::RequestBucketInfoCommand(bucket.getBucketSpace(), buckets));

    LOG(debug,
        "Sending request bucket info command %" PRIu64 " for "
        "bucket %s to node %u",
        msg->getMsgId(),
        bucket.toString().c_str(),
        node);

    msg->setPriority(50);
    msg->setAddress(_distributorComponent.nodeAddress(node));

    _sentMessages[msg->getMsgId()] =
        BucketRequest(node, _distributorComponent.getUniqueTimestamp(),
                      bucket, mergeReplyGuard);
    _sender.sendCommand(msg);
}

void
BucketDBUpdater::recheckBucketInfo(uint32_t nodeIdx,
                                   const document::Bucket& bucket)
{
    sendRequestBucketInfo(nodeIdx, bucket, std::shared_ptr<MergeReplyGuard>());
}

void
BucketDBUpdater::removeSuperfluousBuckets(
        const lib::ClusterStateBundle& newState,
        bool is_distribution_config_change)
{
    const bool move_to_read_only_db = shouldDeferStateEnabling();
    const char* up_states = _distributorComponent.getDistributor().getStorageNodeUpStates();
    for (auto& elem : _distributorComponent.getBucketSpaceRepo()) {
        const auto& newDistribution(elem.second->getDistribution());
        const auto& oldClusterState(elem.second->getClusterState());
        const auto& new_cluster_state = newState.getDerivedClusterState(elem.first);

        // Running a full DB sweep is expensive, so if the cluster state transition does
        // not actually indicate that buckets should possibly be removed, we elide it entirely.
        if (!is_distribution_config_change
            && db_pruning_may_be_elided(oldClusterState, *new_cluster_state, up_states))
        {
            LOG(info, "[bucket space '%s']: eliding DB pruning for state transition '%s' -> '%s'",
                document::FixedBucketSpaces::to_string(elem.first).data(),
                oldClusterState.toString().c_str(), new_cluster_state->toString().c_str());
            continue;
        }

        auto& bucketDb(elem.second->getBucketDatabase());
        auto& readOnlyDb(_distributorComponent.getReadOnlyBucketSpaceRepo().get(elem.first).getBucketDatabase());

        // Remove all buckets not belonging to this distributor, or
        // being on storage nodes that are no longer up.
        MergingNodeRemover proc(
                oldClusterState,
                *new_cluster_state,
                _distributorComponent.getIndex(),
                newDistribution,
                up_states,
                move_to_read_only_db);

        bucketDb.merge(proc);
        // Non-owned entries vector only has contents if move_to_read_only_db is true.
        // TODO either rewrite in terms of merge() or remove entirely
        for (const auto& db_entry : proc.getNonOwnedEntries()) {
            readOnlyDb.update(db_entry); // TODO Entry move support
        }
    }
}

void
BucketDBUpdater::ensureTransitionTimerStarted()
{
{'fraction_non_alphanumeric': 0.10116917203531377, 'fraction_numerical': 0.002226994352978605, 'mean_word_length': 3.4779202279202277, 'pattern_counts': {'":': 0, '<': 28, '<?xml version=': 0, '>': 48, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6813435', 'n_tokens_mistral': 3553, 'n_tokens_neox': 3482, 'n_words': 701}