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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
|
// Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
#pragma once
#include "distributor_message_sender_stub.h"
#include <tests/common/dummystoragelink.h>
#include <tests/common/testhelper.h>
#include <tests/common/teststorageapp.h>
#include <vespa/storage/common/hostreporter/hostinfo.h>
#include <vespa/storage/frameworkimpl/component/distributorcomponentregisterimpl.h>
#include <vespa/storage/storageutil/utils.h>
#include <vespa/storageapi/message/state.h>
#include <vespa/storageframework/defaultimplementation/clock/fakeclock.h>
namespace storage {
namespace framework { struct TickingThreadPool; }
namespace distributor {
class Distributor;
class DistributorBucketSpace;
class DistributorBucketSpaceRepo;
class DistributorStripe;
class DistributorStripeComponent;
class DistributorStripeOperationContext;
class DistributorStripePool;
class ExternalOperationHandler;
class IdealStateManager;
class Operation;
class StripeBucketDBUpdater;
// TODO STRIPE rename to DistributorStripeTestUtil?
class DistributorTestUtil : private DoneInitializeHandler
{
public:
DistributorTestUtil();
~DistributorTestUtil();
/**
* Sets up the storage link chain.
*/
void createLinks();
void setTypeRepo(const std::shared_ptr<const document::DocumentTypeRepo> &repo);
void close();
/**
* Returns a string with the nodes currently stored in the bucket
* database for the given bucket.
*/
std::string getNodes(document::BucketId id);
/**
* Returns a string with the ideal state nodes for the given bucket.
*/
std::string getIdealStr(document::BucketId id, const lib::ClusterState& state);
/**
* Adds the ideal nodes for the given bucket and the given cluster state
* to the bucket database.
*/
void addIdealNodes(const lib::ClusterState& state, const document::BucketId& id);
/**
* Adds all the ideal nodes for the given bucket to the bucket database.
*/
void addIdealNodes(const document::BucketId& id);
/**
* Parses the given string to a set of node => bucket info data,
* and inserts them as nodes in the given bucket.
* Format:
* "node1=checksum/docs/size,node2=checksum/docs/size"
*/
void addNodesToBucketDB(const document::Bucket& bucket, const std::string& nodeStr);
// As the above, but always inserts into default bucket space
void addNodesToBucketDB(const document::BucketId& id, const std::string& nodeStr);
/**
* Removes the given bucket from the bucket database.
*/
void removeFromBucketDB(const document::BucketId& id);
/**
* Inserts the given bucket information for the given bucket and node in
* the bucket database.
*/
void insertBucketInfo(document::BucketId id,
uint16_t node,
uint32_t checksum,
uint32_t count,
uint32_t size,
bool trusted = false,
bool active = false);
/**
* Inserts the given bucket information for the given bucket and node in
* the bucket database.
*/
void insertBucketInfo(document::BucketId id,
uint16_t node,
const api::BucketInfo& info,
bool trusted = false,
bool active = false);
std::string dumpBucket(const document::BucketId& bucket);
/**
* Replies to message idx sent upwards with the given result code.
* If idx = -1, replies to the last command received upwards.
*/
void sendReply(Operation& op,
int idx = -1,
api::ReturnCode::Result result = api::ReturnCode::OK);
StripeBucketDBUpdater& getBucketDBUpdater();
IdealStateManager& getIdealStateManager();
ExternalOperationHandler& getExternalOperationHandler();
storage::distributor::DistributorStripeComponent& distributor_component();
storage::distributor::DistributorStripeOperationContext& operation_context();
Distributor& getDistributor() noexcept { return *_distributor; }
const Distributor& getDistributor() const noexcept { return *_distributor; }
bool tick();
DistributorConfiguration& getConfig();
vdstestlib::DirConfig& getDirConfig() {
return _config;
}
// TODO explicit notion of bucket spaces for tests
DistributorBucketSpace &getDistributorBucketSpace();
BucketDatabase& getBucketDatabase(); // Implicit default space only
BucketDatabase& getBucketDatabase(document::BucketSpace space);
const BucketDatabase& getBucketDatabase() const; // Implicit default space only
const BucketDatabase& getBucketDatabase(document::BucketSpace space) const;
DistributorBucketSpaceRepo &getBucketSpaceRepo();
const DistributorBucketSpaceRepo &getBucketSpaceRepo() const;
DistributorBucketSpaceRepo& getReadOnlyBucketSpaceRepo();
const DistributorBucketSpaceRepo& getReadOnlyBucketSpaceRepo() const;
[[nodiscard]] bool distributor_is_in_recovery_mode() const noexcept;
[[nodiscard]] const lib::ClusterStateBundle& current_distributor_cluster_state_bundle() const noexcept;
[[nodiscard]] std::string active_ideal_state_operations() const;
[[nodiscard]] const PendingMessageTracker& pending_message_tracker() const noexcept;
[[nodiscard]] PendingMessageTracker& pending_message_tracker() noexcept;
[[nodiscard]] std::chrono::steady_clock::duration db_memory_sample_interval() const noexcept;
const lib::Distribution& getDistribution() const;
// "End to end" distribution change trigger, which will invoke the bucket
// DB updater as expected based on the previous and new cluster state
// and config.
void triggerDistributionChange(std::shared_ptr<lib::Distribution> distr);
framework::defaultimplementation::FakeClock& getClock() { return _node->getClock(); }
DistributorComponentRegister& getComponentRegister() { return _node->getComponentRegister(); }
DistributorComponentRegisterImpl& getComponentRegisterImpl() { return _node->getComponentRegister(); }
StorageComponent& getComponent() {
if (_component.get() == 0) {
_component.reset(new storage::DistributorComponent(
_node->getComponentRegister(), "distributor_test_utils"));
}
return *_component;
}
void setupDistributor(int redundancy,
int nodeCount,
const std::string& systemState,
uint32_t earlyReturn = false,
bool requirePrimaryToBeWritten = true);
void setup_distributor(int redundancy,
int node_count,
const lib::ClusterStateBundle& state,
uint32_t early_return = false,
bool require_primary_to_be_written = true);
void setRedundancy(uint32_t redundancy);
void notifyDoneInitializing() override {}
// Must implement this for storage server interface for now
virtual api::Timestamp getUniqueTimestamp() {
return _component->getUniqueTimestamp();
}
void disableBucketActivationInConfig(bool disable);
BucketDatabase::Entry getBucket(const document::Bucket& bucket) const;
// Gets bucket entry from default space only
BucketDatabase::Entry getBucket(const document::BucketId& bId) const;
std::vector<document::BucketSpace> getBucketSpaces() const;
DistributorMessageSenderStub& sender() noexcept { return _sender; }
const DistributorMessageSenderStub& sender() const noexcept { return _sender; }
void setSystemState(const lib::ClusterState& systemState);
// Must be called prior to createLinks() to have any effect
void set_num_distributor_stripes(uint32_t n_stripes) noexcept {
_num_distributor_stripes = n_stripes;
}
protected:
vdstestlib::DirConfig _config;
std::unique_ptr<TestDistributorApp> _node;
std::unique_ptr<framework::TickingThreadPool> _threadPool;
std::unique_ptr<DistributorStripePool> _stripe_pool;
std::unique_ptr<Distributor> _distributor;
std::unique_ptr<storage::DistributorComponent> _component;
DistributorMessageSenderStub _sender;
DistributorMessageSenderStub _senderDown;
HostInfo _hostInfo;
struct MessageSenderImpl : public ChainedMessageSender {
DistributorMessageSenderStub& _sender;
DistributorMessageSenderStub& _senderDown;
MessageSenderImpl(DistributorMessageSenderStub& up, DistributorMessageSenderStub& down)
: _sender(up), _senderDown(down) {}
void sendUp(const std::shared_ptr<api::StorageMessage>& msg) override {
_sender.send(msg);
}
void sendDown(const std::shared_ptr<api::StorageMessage>& msg) override {
_senderDown.send(msg);
}
};
MessageSenderImpl _messageSender;
uint32_t _num_distributor_stripes;
void enableDistributorClusterState(vespalib::stringref state);
void enable_distributor_cluster_state(const lib::ClusterStateBundle& state);
};
}
}
|
