1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
|
// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
#pragma once
#include "bucketlistmerger.h"
#include "distributor_component.h"
#include "distributormessagesender.h"
#include "operation_routing_snapshot.h"
#include "outdated_nodes_map.h"
#include "pendingclusterstate.h"
#include <vespa/document/bucket/bucket.h>
#include <vespa/storage/common/message_guard.h>
#include <vespa/storageapi/message/bucket.h>
#include <vespa/storageapi/messageapi/messagehandler.h>
#include <vespa/storageframework/generic/clock/timer.h>
#include <vespa/storageframework/generic/status/statusreporter.h>
#include <vespa/vdslib/state/clusterstate.h>
#include <atomic>
#include <list>
#include <mutex>
namespace vespalib::xml {
class XmlOutputStream;
class XmlAttribute;
}
namespace storage::distributor {
struct BucketSpaceDistributionConfigs;
class BucketSpaceDistributionContext;
class ClusterStateBundleActivationListener;
class DistributorInterface;
class NodeSupportedFeaturesRepo;
class StripeAccessor;
class StripeAccessGuard;
class TopLevelBucketDBUpdater : public framework::StatusReporter,
public api::MessageHandler
{
public:
using OutdatedNodesMap = dbtransition::OutdatedNodesMap;
TopLevelBucketDBUpdater(const DistributorNodeContext& node_ctx,
DistributorOperationContext& op_ctx,
DistributorInterface& distributor_interface,
ChainedMessageSender& chained_sender,
std::shared_ptr<const lib::Distribution> bootstrap_distribution,
StripeAccessor& stripe_accessor,
ClusterStateBundleActivationListener* state_activation_listener);
~TopLevelBucketDBUpdater() override;
void flush();
bool onSetSystemState(const std::shared_ptr<api::SetSystemStateCommand>& cmd) override;
bool onActivateClusterStateVersion(const std::shared_ptr<api::ActivateClusterStateVersionCommand>& cmd) override;
bool onRequestBucketInfoReply(const std::shared_ptr<api::RequestBucketInfoReply> & repl) override;
vespalib::string getReportContentType(const framework::HttpUrlPath&) const override;
bool reportStatus(std::ostream&, const framework::HttpUrlPath&) const override;
void resend_delayed_messages();
void storage_distribution_changed(const BucketSpaceDistributionConfigs& configs);
void bootstrap_distribution_config(std::shared_ptr<const lib::Distribution>);
void propagate_distribution_config(const BucketSpaceDistributionConfigs& configs);
vespalib::string report_xml_status(vespalib::xml::XmlOutputStream& xos, const framework::HttpUrlPath&) const;
void print(std::ostream& out, bool verbose, const std::string& indent) const;
void set_stale_reads_enabled(bool enabled) noexcept {
_stale_reads_enabled.store(enabled, std::memory_order_relaxed);
}
bool stale_reads_enabled() const noexcept {
return _stale_reads_enabled.load(std::memory_order_relaxed);
}
private:
friend class DistributorStripeTestUtil;
friend class TopLevelDistributorTestUtil;
// Only to be used by tests that want to ensure both the TopLevelBucketDBUpdater _and_ the Distributor
// components agree on the currently active cluster state bundle.
// Transitively invokes Distributor::enableClusterStateBundle
void simulate_cluster_state_bundle_activation(const lib::ClusterStateBundle& activated_state,
bool has_bucket_ownership_transfer = false);
bool should_defer_state_enabling() const noexcept;
bool has_pending_cluster_state() const;
void attempt_accept_reply_by_current_pending_state(const std::shared_ptr<api::RequestBucketInfoReply>& repl);
bool is_pending_cluster_state_completed() const;
void process_completed_pending_cluster_state(StripeAccessGuard& guard);
void activate_pending_cluster_state(StripeAccessGuard& guard);
void ensure_transition_timer_started();
void complete_transition_timer();
void remove_superfluous_buckets(StripeAccessGuard& guard,
const lib::ClusterStateBundle& new_state,
bool is_distribution_config_change);
void reply_to_previous_pending_cluster_state_if_any();
void reply_to_activation_with_actual_version(
const api::ActivateClusterStateVersionCommand& cmd,
uint32_t actualVersion);
void enable_current_cluster_state_bundle_in_distributor_and_stripes(StripeAccessGuard& guard);
void add_current_state_to_cluster_state_history();
void propagate_active_state_bundle_internally(bool has_bucket_ownership_transfer);
void maybe_inject_simulated_db_pruning_delay();
void maybe_inject_simulated_db_merging_delay();
// TODO STRIPE remove once distributor component dependencies have been pruned
StripeAccessor& _stripe_accessor;
ClusterStateBundleActivationListener* _state_activation_listener;
lib::ClusterStateBundle _active_state_bundle;
const DistributorNodeContext& _node_ctx;
DistributorOperationContext& _op_ctx;
DistributorInterface& _distributor_interface;
std::unique_ptr<PendingClusterState> _pending_cluster_state;
std::list<PendingClusterState::Summary> _history;
DistributorMessageSender& _sender;
ChainedMessageSender& _chained_sender;
OutdatedNodesMap _outdated_nodes_map;
framework::MilliSecTimer _transition_timer;
std::shared_ptr<const NodeSupportedFeaturesRepo> _node_supported_features_repo;
std::atomic<bool> _stale_reads_enabled;
};
}
|
