wbtiger
/
foundationdb
mirror of https://github.com/apple/foundationdb.git
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'apple-upstream/master' into version-vector-prototype

This commit is contained in:
Sreenath Bodagala 2021-10-18 19:07:33 +00:00
parent 26f82440c8 faf16fb29e
commit 246f035afe
37 changed files with 1177 additions and 469 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
documentation/sphinx/source/request-tracing.rst
View File

@ -95,3 +95,13 @@ Tracing can be enabled or disabled for individual transactions. The special key
space exposes an API to set a custom trace ID for a transaction, or to disable
tracing for the transaction. See the special key space :ref:`tracing module
documentation <special-key-space-tracing-module>` to learn more.
^^^^^^^^^^^^^^
Trace sampling
^^^^^^^^^^^^^^
By default, all traces are recorded. If tracing is producing too much data,
adjust the trace sample rate with the ``TRACING_SAMPLE_RATE`` knob. Set the
knob to 0.0 to record no traces, to 1.0 to record all traces, or somewhere in
the middle. Traces are sampled as a unit. All individual spans in the trace
will be included in the sample.

3
fdbclient/ActorLineageProfiler.cpp
View File

@ -242,6 +242,9 @@ void sample(LineageReference* lineagePtr) {
if (!lineagePtr->isValid()) {
return;
}
if (!lineagePtr->isAllocated()) {
lineagePtr->allocate();
}
(*lineagePtr)->modify(&NameLineage::actorName) = lineagePtr->actorName();
boost::asio::post(ActorLineageProfiler::instance().context(),
[lineage = LineageReference::addRef(lineagePtr->getPtr())]() {

50
fdbclient/ClusterInterface.h
View File

@ -36,6 +36,8 @@ struct ClusterInterface {
RequestStream<ReplyPromise<Void>> ping;
RequestStream<struct GetClientWorkersRequest> getClientWorkers;
RequestStream<struct ForceRecoveryRequest> forceRecovery;
RequestStream<struct MoveShardRequest> moveShard;
RequestStream<struct RepairSystemDataRequest> repairSystemData;
bool operator==(ClusterInterface const& r) const { return id() == r.id(); }
bool operator!=(ClusterInterface const& r) const { return id() != r.id(); }
@ -45,7 +47,8 @@ struct ClusterInterface {
bool hasMessage() const {
return openDatabase.getFuture().isReady() || failureMonitoring.getFuture().isReady() ||
databaseStatus.getFuture().isReady() || ping.getFuture().isReady() ||
getClientWorkers.getFuture().isReady() || forceRecovery.getFuture().isReady();
getClientWorkers.getFuture().isReady() || forceRecovery.getFuture().isReady() ||
moveShard.getFuture().isReady() || repairSystemData.getFuture().isReady();
}
void initEndpoints() {
@ -55,11 +58,21 @@ struct ClusterInterface {
ping.getEndpoint(TaskPriority::ClusterController);
getClientWorkers.getEndpoint(TaskPriority::ClusterController);
forceRecovery.getEndpoint(TaskPriority::ClusterController);
moveShard.getEndpoint(TaskPriority::ClusterController);
repairSystemData.getEndpoint(TaskPriority::ClusterController);
}
template <class Ar>
void serialize(Ar& ar) {
serializer(ar, openDatabase, failureMonitoring, databaseStatus, ping, getClientWorkers, forceRecovery);
serializer(ar,
openDatabase,
failureMonitoring,
databaseStatus,
ping,
getClientWorkers,
forceRecovery,
moveShard,
repairSystemData);
}
};
@ -291,4 +304,37 @@ struct ForceRecoveryRequest {
}
};
// Request to move a keyrange (shard) to a new team represented as addresses.
struct MoveShardRequest {
constexpr static FileIdentifier file_identifier = 2799592;
KeyRange shard;
std::vector<NetworkAddress> addresses;
ReplyPromise<Void> reply;
MoveShardRequest() {}
MoveShardRequest(KeyRange shard, std::vector<NetworkAddress> addresses)
: shard{ std::move(shard) }, addresses{ std::move(addresses) } {}
template <class Ar>
void serialize(Ar& ar) {
serializer(ar, shard, addresses, reply);
}
};
// Request to trigger a master recovery, and during the following recovery, the system metadata will be
// reconstructed from TLogs, and written to a new SS team.
// This is used when metadata on SSes are lost or corrupted.
struct RepairSystemDataRequest {
constexpr static FileIdentifier file_identifier = 2799593;
ReplyPromise<Void> reply;
RepairSystemDataRequest() {}
template <class Ar>
void serialize(Ar& ar) {
serializer(ar, reply);
}
};
#endif

3
fdbclient/DatabaseContext.h
View File

@ -382,6 +382,8 @@ public:
Counter transactionsProcessBehind;
Counter transactionsThrottled;
Counter transactionsExpensiveClearCostEstCount;
Counter transactionGrvFullBatches;
Counter transactionGrvTimedOutBatches;
ContinuousSample<double> latencies, readLatencies, commitLatencies, GRVLatencies, mutationsPerCommit,
bytesPerCommit;
@ -392,6 +394,7 @@ public:
int snapshotRywEnabled;
int transactionTracingEnabled;
double verifyCausalReadsProp = 0.0;
Future<Void> logger;
Future<Void> throttleExpirer;

11
fdbclient/FluentDSampleIngestor.cpp
View File

@ -73,13 +73,14 @@ class SampleSender : public std::enable_shared_from_this<SampleSender<Protocol,
}
void send(boost::asio::ip::tcp::socket& socket, std::shared_ptr<Buf> const& buf) {
boost::asio::async_write(socket,
boost::asio::const_buffer(buf->data, buf->size),
[buf, this](auto const& ec, size_t) { this->sendCompletionHandler(ec); });
boost::system::error_code ec;
socket.send(boost::asio::const_buffer(buf->data, buf->size), 0, ec);
this->sendCompletionHandler(ec);
}
void send(boost::asio::ip::udp::socket& socket, std::shared_ptr<Buf> const& buf) {
socket.async_send(boost::asio::const_buffer(buf->data, buf->size),
[buf, this](auto const& ec, size_t) { this->sendCompletionHandler(ec); });
boost::system::error_code ec;
socket.send(boost::asio::const_buffer(buf->data, buf->size), 0, ec);
this->sendCompletionHandler(ec);
}
void sendNext() {

3
fdbclient/GrvProxyInterface.h
View File

@ -22,6 +22,9 @@
#ifndef FDBCLIENT_GRVPROXYINTERFACE_H
#define FDBCLIENT_GRVPROXYINTERFACE_H
#pragma once
#include "flow/FileIdentifier.h"
#include "fdbrpc/fdbrpc.h"
#include "fdbclient/FDBTypes.h"
// GrvProxy is proxy primarily specializing on serving GetReadVersion. It also serves health metrics since it
// communicates with RateKeeper to gather health information of the cluster.

147
fdbclient/NativeAPI.actor.cpp
View File

@ -447,7 +447,8 @@ ACTOR Future<Void> databaseLogger(DatabaseContext* cx) {
.detail("MaxMutationsPerCommit", cx->mutationsPerCommit.max())
.detail("MeanBytesPerCommit", cx->bytesPerCommit.mean())
.detail("MedianBytesPerCommit", cx->bytesPerCommit.median())
.detail("MaxBytesPerCommit", cx->bytesPerCommit.max());
.detail("MaxBytesPerCommit", cx->bytesPerCommit.max())
.detail("NumLocalityCacheEntries", cx->locationCache.size());
cx->latencies.clear();
cx->readLatencies.clear();
@ -714,19 +715,82 @@ ACTOR static Future<Void> clientStatusUpdateActor(DatabaseContext* cx) {
}
}
ACTOR static Future<Void> monitorProxiesChange(Reference<AsyncVar<ClientDBInfo> const> clientDBInfo,
ACTOR Future<Void> assertFailure(GrvProxyInterface remote, Future<ErrorOr<GetReadVersionReply>> reply) {
try {
ErrorOr<GetReadVersionReply> res = wait(reply);
if (!res.isError()) {
TraceEvent(SevError, "GotStaleReadVersion")
.detail("Remote", remote.getConsistentReadVersion.getEndpoint().addresses.address.toString())
.detail("Provisional", remote.provisional)
.detail("ReadVersion", res.get().version);
ASSERT_WE_THINK(false);
}
} catch (Error& e) {
if (e.code() == error_code_actor_cancelled) {
throw;
}
// we want this to fail -- so getting here is good, we'll just ignore the error.
}
return Void();
}
Future<Void> attemptGRVFromOldProxies(std::vector<GrvProxyInterface> oldProxies,
std::vector<GrvProxyInterface> newProxies) {
Span span(deterministicRandom()->randomUniqueID(), "VerifyCausalReadRisky"_loc);
std::vector<Future<Void>> replies;
replies.reserve(oldProxies.size());
GetReadVersionRequest req(
span.context, 1, TransactionPriority::IMMEDIATE, GetReadVersionRequest::FLAG_CAUSAL_READ_RISKY);
TraceEvent evt("AttemptGRVFromOldProxies");
evt.detail("NumOldProxies", oldProxies.size()).detail("NumNewProxies", newProxies.size());
auto traceProxies = [&](std::vector<GrvProxyInterface>& proxies, std::string const& key) {
for (int i = 0; i < proxies.size(); ++i) {
auto k = key + std::to_string(i);
evt.detail(k.c_str(), proxies[i].id());
}
};
traceProxies(oldProxies, "OldProxy"s);
traceProxies(newProxies, "NewProxy"s);
evt.log();
for (auto& i : oldProxies) {
req.reply = ReplyPromise<GetReadVersionReply>();
replies.push_back(assertFailure(i, i.getConsistentReadVersion.tryGetReply(req)));
}
return waitForAll(replies);
}
ACTOR static Future<Void> monitorProxiesChange(DatabaseContext* cx,
Reference<AsyncVar<ClientDBInfo> const> clientDBInfo,
AsyncTrigger* triggerVar) {
state std::vector<CommitProxyInterface> curCommitProxies;
state std::vector<GrvProxyInterface> curGrvProxies;
state ActorCollection actors(false);
curCommitProxies = clientDBInfo->get().commitProxies;
curGrvProxies = clientDBInfo->get().grvProxies;
loop {
wait(clientDBInfo->onChange());
if (clientDBInfo->get().commitProxies != curCommitProxies || clientDBInfo->get().grvProxies != curGrvProxies) {
curCommitProxies = clientDBInfo->get().commitProxies;
curGrvProxies = clientDBInfo->get().grvProxies;
triggerVar->trigger();
choose {
when(wait(clientDBInfo->onChange())) {
if (clientDBInfo->get().commitProxies != curCommitProxies ||
clientDBInfo->get().grvProxies != curGrvProxies) {
// This condition is a bit complicated. Here we want to verify that we're unable to receive a read
// version from a proxy of an old generation after a successful recovery. The conditions are:
// 1. We only do this with a configured probability.
// 2. If the old set of Grv proxies is empty, there's nothing to do
// 3. If the new set of Grv proxies is empty, it means the recovery is not complete. So if an old
// Grv proxy still gives out read versions, this would be correct behavior.
// 4. If we see a provisional proxy, it means the recovery didn't complete yet, so the same as (3)
// applies.
if (deterministicRandom()->random01() < cx->verifyCausalReadsProp && !curGrvProxies.empty() &&
!clientDBInfo->get().grvProxies.empty() && !clientDBInfo->get().grvProxies[0].provisional) {
actors.add(attemptGRVFromOldProxies(curGrvProxies, clientDBInfo->get().grvProxies));
}
curCommitProxies = clientDBInfo->get().commitProxies;
curGrvProxies = clientDBInfo->get().grvProxies;
triggerVar->trigger();
}
}
when(wait(actors.getResult())) { UNSTOPPABLE_ASSERT(false); }
}
}
}
@ -1187,11 +1251,13 @@ DatabaseContext::DatabaseContext(Reference<AsyncVar<Reference<ClusterConnectionF
transactionsFutureVersions("FutureVersions", cc), transactionsNotCommitted("NotCommitted", cc),
transactionsMaybeCommitted("MaybeCommitted", cc), transactionsResourceConstrained("ResourceConstrained", cc),
transactionsProcessBehind("ProcessBehind", cc), transactionsThrottled("Throttled", cc),
transactionsExpensiveClearCostEstCount("ExpensiveClearCostEstCount", cc), latencies(1000), readLatencies(1000),
commitLatencies(1000), GRVLatencies(1000), mutationsPerCommit(1000), bytesPerCommit(1000), outstandingWatches(0),
transactionTracingEnabled(true), taskID(taskID), clientInfo(clientInfo), clientInfoMonitor(clientInfoMonitor),
coordinator(coordinator), apiVersion(apiVersion), mvCacheInsertLocation(0), healthMetricsLastUpdated(0),
detailedHealthMetricsLastUpdated(0), smoothMidShardSize(CLIENT_KNOBS->SHARD_STAT_SMOOTH_AMOUNT),
transactionsExpensiveClearCostEstCount("ExpensiveClearCostEstCount", cc),
transactionGrvFullBatches("NumGrvFullBatches", cc), transactionGrvTimedOutBatches("NumGrvTimedOutBatches", cc),
latencies(1000), readLatencies(1000), commitLatencies(1000), GRVLatencies(1000), mutationsPerCommit(1000),
bytesPerCommit(1000), outstandingWatches(0), transactionTracingEnabled(true), taskID(taskID),
clientInfo(clientInfo), clientInfoMonitor(clientInfoMonitor), coordinator(coordinator), apiVersion(apiVersion),
mvCacheInsertLocation(0), healthMetricsLastUpdated(0), detailedHealthMetricsLastUpdated(0),
smoothMidShardSize(CLIENT_KNOBS->SHARD_STAT_SMOOTH_AMOUNT),
specialKeySpace(std::make_unique<SpecialKeySpace>(specialKeys.begin, specialKeys.end, /* test */ false)) {
dbId = deterministicRandom()->randomUniqueID();
connected = (clientInfo->get().commitProxies.size() && clientInfo->get().grvProxies.size())
@ -1210,7 +1276,7 @@ DatabaseContext::DatabaseContext(Reference<AsyncVar<Reference<ClusterConnectionF
getValueSubmitted.init(LiteralStringRef("NativeAPI.GetValueSubmitted"));
getValueCompleted.init(LiteralStringRef("NativeAPI.GetValueCompleted"));
monitorProxiesInfoChange = monitorProxiesChange(clientInfo, &proxiesChangeTrigger);
monitorProxiesInfoChange = monitorProxiesChange(this, clientInfo, &proxiesChangeTrigger);
tssMismatchHandler = handleTssMismatches(this);
clientStatusUpdater.actor = clientStatusUpdateActor(this);
cacheListMonitor = monitorCacheList(this);
@ -1439,9 +1505,10 @@ DatabaseContext::DatabaseContext(const Error& err)
transactionsFutureVersions("FutureVersions", cc), transactionsNotCommitted("NotCommitted", cc),
transactionsMaybeCommitted("MaybeCommitted", cc), transactionsResourceConstrained("ResourceConstrained", cc),
transactionsProcessBehind("ProcessBehind", cc), transactionsThrottled("Throttled", cc),
transactionsExpensiveClearCostEstCount("ExpensiveClearCostEstCount", cc), latencies(1000), readLatencies(1000),
commitLatencies(1000), GRVLatencies(1000), mutationsPerCommit(1000), bytesPerCommit(1000),
transactionTracingEnabled(true), smoothMidShardSize(CLIENT_KNOBS->SHARD_STAT_SMOOTH_AMOUNT) {}
transactionsExpensiveClearCostEstCount("ExpensiveClearCostEstCount", cc),
transactionGrvFullBatches("NumGrvFullBatches", cc), transactionGrvTimedOutBatches("NumGrvTimedOutBatches", cc),
latencies(1000), readLatencies(1000), commitLatencies(1000), GRVLatencies(1000), mutationsPerCommit(1000),
bytesPerCommit(1000), transactionTracingEnabled(true), smoothMidShardSize(CLIENT_KNOBS->SHARD_STAT_SMOOTH_AMOUNT) {}
// Static constructor used by server processes to create a DatabaseContext
// For internal (fdbserver) use only
@ -1652,6 +1719,9 @@ void DatabaseContext::setOption(FDBDatabaseOptions::Option option, Optional<Stri
validateOptionValueNotPresent(value);
useConfigDatabase = true;
break;
case FDBDatabaseOptions::TEST_CAUSAL_READ_RISKY:
verifyCausalReadsProp = double(extractIntOption(value, 0, 100)) / 100.0;
break;
default:
break;
}
@ -4208,11 +4278,14 @@ void debugAddTags(Transaction* tr) {
}
SpanID generateSpanID(int transactionTracingEnabled) {
uint64_t tid = deterministicRandom()->randomUInt64();
uint64_t txnId = deterministicRandom()->randomUInt64();
if (transactionTracingEnabled > 0) {
return SpanID(tid, deterministicRandom()->randomUInt64());
uint64_t tokenId = deterministicRandom()->random01() <= FLOW_KNOBS->TRACING_SAMPLE_RATE
? deterministicRandom()->randomUInt64()
: 0;
return SpanID(txnId, tokenId);
} else {
return SpanID(tid, 0);
return SpanID(txnId, 0);
}
}
@ -5669,6 +5742,20 @@ ACTOR Future<Void> readVersionBatcher(DatabaseContext* cx,
state Future<Void> timeout;
state Optional<UID> debugID;
state bool send_batch;
state Reference<Histogram> batchSizeDist = Histogram::getHistogram(LiteralStringRef("GrvBatcher"),
LiteralStringRef("ClientGrvBatchSize"),
Histogram::Unit::countLinear,
0,
CLIENT_KNOBS->MAX_BATCH_SIZE * 2);
state Reference<Histogram> batchIntervalDist =
Histogram::getHistogram(LiteralStringRef("GrvBatcher"),
LiteralStringRef("ClientGrvBatchInterval"),
Histogram::Unit::microseconds,
0,
CLIENT_KNOBS->GRV_BATCH_TIMEOUT * 1000000 * 2);
state Reference<Histogram> grvReplyLatencyDist = Histogram::getHistogram(
LiteralStringRef("GrvBatcher"), LiteralStringRef("ClientGrvReplyLatency"), Histogram::Unit::microseconds);
state double lastRequestTime = now();
state TransactionTagMap<uint32_t> tags;
@ -5693,22 +5780,34 @@ ACTOR Future<Void> readVersionBatcher(DatabaseContext* cx,
++tags[tag];
}
if (requests.size() == CLIENT_KNOBS->MAX_BATCH_SIZE)
if (requests.size() == CLIENT_KNOBS->MAX_BATCH_SIZE) {
send_batch = true;
else if (!timeout.isValid())
++cx->transactionGrvFullBatches;
} else if (!timeout.isValid()) {
timeout = delay(batchTime, TaskPriority::GetConsistentReadVersion);
}
}
when(wait(timeout.isValid() ? timeout : Never())) {
send_batch = true;
++cx->transactionGrvTimedOutBatches;
}
when(wait(timeout.isValid() ? timeout : Never())) { send_batch = true; }
// dynamic batching monitors reply latencies
when(double reply_latency = waitNext(replyTimes.getFuture())) {
double target_latency = reply_latency * 0.5;
batchTime = std::min(0.1 * target_latency + 0.9 * batchTime, CLIENT_KNOBS->GRV_BATCH_TIMEOUT);
grvReplyLatencyDist->sampleSeconds(reply_latency);
}
when(wait(collection)) {} // for errors
}
if (send_batch) {
int count = requests.size();
ASSERT(count);
batchSizeDist->sampleRecordCounter(count);
auto requestTime = now();
batchIntervalDist->sampleSeconds(requestTime - lastRequestTime);
lastRequestTime = requestTime;
// dynamic batching
Promise<GetReadVersionReply> GRVReply;
requests.push_back(GRVReply);
@ -5859,7 +5958,7 @@ Future<Version> Transaction::getReadVersion(uint32_t flags) {
}
Location location = "NAPI:getReadVersion"_loc;
UID spanContext = deterministicRandom()->randomUniqueID();
UID spanContext = generateSpanID(cx->transactionTracingEnabled);
auto const req = DatabaseContext::VersionRequest(spanContext, options.tags, info.debugID);
batcher.stream.send(req);
startTime = now();
@ -6223,7 +6322,7 @@ ACTOR Future<std::pair<Optional<StorageMetrics>, int>> waitStorageMetrics(Databa
StorageMetrics permittedError,
int shardLimit,
int expectedShardCount) {
state Span span("NAPI:WaitStorageMetrics"_loc);
state Span span("NAPI:WaitStorageMetrics"_loc, generateSpanID(cx->transactionTracingEnabled));
loop {
std::vector<std::pair<KeyRange, Reference<LocationInfo>>> locations =
wait(getKeyRangeLocations(cx,

5
fdbclient/PaxosConfigTransaction.actor.cpp
View File

@ -58,6 +58,11 @@ class CommitQuorum {
wait(retryBrokenPromise(cti.commit, self->getCommitRequest(generation)));
++self->successful;
} catch (Error& e) {
// self might be destroyed if this actor is canceled
if (e.code() == error_code_actor_cancelled) {
throw;
}
if (e.code() == error_code_not_committed) {
++self->failed;
} else {

8
fdbclient/ServerKnobs.cpp
View File

@ -459,6 +459,7 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi
init( SIM_SHUTDOWN_TIMEOUT, 10 );
init( SHUTDOWN_TIMEOUT, 600 ); if( randomize && BUGGIFY ) SHUTDOWN_TIMEOUT = 60.0;
init( MASTER_SPIN_DELAY, 1.0 ); if( randomize && BUGGIFY ) MASTER_SPIN_DELAY = 10.0;
init( CC_PRUNE_CLIENTS_INTERVAL, 60.0 );
init( CC_CHANGE_DELAY, 0.1 );
init( CC_CLASS_DELAY, 0.01 );
init( WAIT_FOR_GOOD_RECRUITMENT_DELAY, 1.0 );
@ -470,6 +471,8 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi
init( CHECK_OUTSTANDING_INTERVAL, 0.5 ); if( randomize && BUGGIFY ) CHECK_OUTSTANDING_INTERVAL = 0.001;
init( VERSION_LAG_METRIC_INTERVAL, 0.5 ); if( randomize && BUGGIFY ) VERSION_LAG_METRIC_INTERVAL = 10.0;
init( MAX_VERSION_DIFFERENCE, 20 * VERSIONS_PER_SECOND );
init( INITIAL_UPDATE_CROSS_DC_INFO_DELAY, 300 );
init( CHECK_REMOTE_HEALTH_INTERVAL, 60 );
init( FORCE_RECOVERY_CHECK_DELAY, 5.0 );
init( RATEKEEPER_FAILURE_TIME, 1.0 );
init( REPLACE_INTERFACE_DELAY, 60.0 );
@ -484,7 +487,10 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi
init( CC_MAX_EXCLUSION_DUE_TO_HEALTH, 2 );
init( CC_HEALTH_TRIGGER_RECOVERY, false );
init( CC_TRACKING_HEALTH_RECOVERY_INTERVAL, 3600.0 );
init( CC_MAX_HEALTH_RECOVERY_COUNT, 2 );
init( CC_MAX_HEALTH_RECOVERY_COUNT, 5 );
init( CC_HEALTH_TRIGGER_FAILOVER, false );
init( CC_FAILOVER_DUE_TO_HEALTH_MIN_DEGRADATION, 5 );
init( CC_FAILOVER_DUE_TO_HEALTH_MAX_DEGRADATION, 10 );
init( INCOMPATIBLE_PEERS_LOGGING_INTERVAL, 600 ); if( randomize && BUGGIFY ) INCOMPATIBLE_PEERS_LOGGING_INTERVAL = 60.0;
init( EXPECTED_MASTER_FITNESS, ProcessClass::UnsetFit );

13
fdbclient/ServerKnobs.h
View File

@ -381,6 +381,7 @@ public:
double SIM_SHUTDOWN_TIMEOUT;
double SHUTDOWN_TIMEOUT;
double MASTER_SPIN_DELAY;
double CC_PRUNE_CLIENTS_INTERVAL;
double CC_CHANGE_DELAY;
double CC_CLASS_DELAY;
double WAIT_FOR_GOOD_RECRUITMENT_DELAY;
@ -393,6 +394,10 @@ public:
double INCOMPATIBLE_PEERS_LOGGING_INTERVAL;
double VERSION_LAG_METRIC_INTERVAL;
int64_t MAX_VERSION_DIFFERENCE;
double INITIAL_UPDATE_CROSS_DC_INFO_DELAY; // The intial delay in a new Cluster Controller just started to refresh
// the info of remote DC, such as remote DC health, and whether we need
// to take remote DC health info when making failover decision.
double CHECK_REMOTE_HEALTH_INTERVAL; // Remote DC health refresh interval.
double FORCE_RECOVERY_CHECK_DELAY;
double RATEKEEPER_FAILURE_TIME;
double REPLACE_INTERFACE_DELAY;
@ -415,7 +420,13 @@ public:
double CC_TRACKING_HEALTH_RECOVERY_INTERVAL; // The number of recovery count should not exceed
// CC_MAX_HEALTH_RECOVERY_COUNT within
// CC_TRACKING_HEALTH_RECOVERY_INTERVAL.
int CC_MAX_HEALTH_RECOVERY_COUNT;
int CC_MAX_HEALTH_RECOVERY_COUNT; // The max number of recoveries can be triggered due to worker health within
// CC_TRACKING_HEALTH_RECOVERY_INTERVAL
bool CC_HEALTH_TRIGGER_FAILOVER; // Whether to enable health triggered failover in CC.
int CC_FAILOVER_DUE_TO_HEALTH_MIN_DEGRADATION; // The minimum number of degraded servers that can trigger a
// failover.
int CC_FAILOVER_DUE_TO_HEALTH_MAX_DEGRADATION; // The maximum number of degraded servers that can trigger a
// failover.
// Knobs used to select the best policy (via monte carlo)
int POLICY_RATING_TESTS; // number of tests per policy (in order to compare)

2
fdbclient/vexillographer/fdb.options
View File

@ -200,6 +200,8 @@ description is not currently required but encouraged.
defaultFor="1100"/>
<Option name="use_config_database" code="800"
description="Use configuration database." />
<Option name="test_causal_read_risky" code="900"
description="An integer between 0 and 100 (default is 0) expressing the probability that a client will verify it can't read stale data whenever it detects a recovery." />
</Scope>
<Scope name="TransactionOption">

2
fdbrpc/AsyncFileNonDurable.actor.h
View File

@ -457,13 +457,13 @@ private:
void const* data,
int length,
int64_t offset) {
state Standalone<StringRef> dataCopy(StringRef((uint8_t*)data, length));
state ISimulator::ProcessInfo* currentProcess = g_simulator.getCurrentProcess();
state TaskPriority currentTaskID = g_network->getCurrentTask();
wait(g_simulator.onMachine(currentProcess));
state double delayDuration =
g_simulator.speedUpSimulation ? 0.0001 : (deterministicRandom()->random01() * self->maxWriteDelay);
state Standalone<StringRef> dataCopy(StringRef((uint8_t*)data, length));
state Future<bool> startSyncFuture = self->startSyncPromise.getFuture();

6
fdbserver/ApplyMetadataMutation.cpp
View File

@ -866,7 +866,10 @@ private:
MutationRef privatized = m;
privatized.param1 = m.param1.withPrefix(systemKeys.begin, arena);
privatized.param2 = m.param2.withPrefix(systemKeys.begin, arena);
<<<<<<< HEAD
TraceEvent(SevDebug, "SendingPrivatized_ClearTSSMapping", dbgid).detail("M", privatized.toString());
=======
>>>>>>> apple-upstream/master
toCommit->addTag(decodeServerTagValue(tagV.get()));
toCommit->writeTypedMessage(privatized);
}
@ -893,8 +896,11 @@ private:
MutationRef privatized = m;
privatized.param1 = m.param1.withPrefix(systemKeys.begin, arena);
privatized.param2 = m.param2.withPrefix(systemKeys.begin, arena);
<<<<<<< HEAD
TraceEvent(SevDebug, "SendingPrivatized_ClearTSSQuarantine", dbgid)
.detail("M", privatized.toString());
=======
>>>>>>> apple-upstream/master
toCommit->addTag(decodeServerTagValue(tagV.get()));
toCommit->writeTypedMessage(privatized);
}

308
fdbserver/ClusterController.actor.cpp
View File

@ -26,6 +26,7 @@
#include "fdbrpc/FailureMonitor.h"
#include "flow/ActorCollection.h"
#include "flow/SystemMonitor.h"
#include "fdbclient/NativeAPI.actor.h"
#include "fdbserver/BackupInterface.h"
#include "fdbserver/CoordinationInterface.h"
@ -132,6 +133,8 @@ public:
int logGenerations;
bool cachePopulated;
std::map<NetworkAddress, std::pair<double, OpenDatabaseRequest>> clientStatus;
Future<Void> clientCounter;
int clientCount;
DBInfo()
: clientInfo(new AsyncVar<ClientDBInfo>()), serverInfo(new AsyncVar<ServerDBInfo>()),
@ -142,7 +145,9 @@ public:
EnableLocalityLoadBalance::True,
TaskPriority::DefaultEndpoint,
LockAware::True)), // SOMEDAY: Locality!
unfinishedRecoveries(0), logGenerations(0), cachePopulated(false) {}
unfinishedRecoveries(0), logGenerations(0), cachePopulated(false), clientCount(0) {
clientCounter = countClients(this);
}
void setDistributor(const DataDistributorInterface& interf) {
auto newInfo = serverInfo->get();
@ -171,6 +176,22 @@ public:
}
serverInfo->set(newInfo);
}
ACTOR static Future<Void> countClients(DBInfo* self) {
loop {
wait(delay(SERVER_KNOBS->CC_PRUNE_CLIENTS_INTERVAL));
self->clientCount = 0;
for (auto itr = self->clientStatus.begin(); itr != self->clientStatus.end();) {
if (now() - itr->second.first < 2 * SERVER_KNOBS->COORDINATOR_REGISTER_INTERVAL) {
self->clientCount += itr->second.second.clientCount;
++itr;
} else {
itr = self->clientStatus.erase(itr);
}
}
}
}
};
struct UpdateWorkerList {
@ -2233,7 +2254,7 @@ public:
if (db.config.regions.size() > 1 && db.config.regions[0].priority > db.config.regions[1].priority &&
db.config.regions[0].dcId != clusterControllerDcId.get() && versionDifferenceUpdated &&
datacenterVersionDifference < SERVER_KNOBS->MAX_VERSION_DIFFERENCE) {
datacenterVersionDifference < SERVER_KNOBS->MAX_VERSION_DIFFERENCE && remoteDCIsHealthy()) {
checkRegions(db.config.regions);
}
@ -2795,7 +2816,7 @@ public:
void updateWorkerHealth(const UpdateWorkerHealthRequest& req) {
std::string degradedPeersString;
for (int i = 0; i < req.degradedPeers.size(); ++i) {
degradedPeersString += i == 0 ? "" : " " + req.degradedPeers[i].toString();
degradedPeersString += (i == 0 ? "" : " ") + req.degradedPeers[i].toString();
}
TraceEvent("ClusterControllerUpdateWorkerHealth")
.detail("WorkerAddress", req.address)
@ -2929,24 +2950,9 @@ public:
return currentDegradedServersWithinLimit;
}
// Returns true when the cluster controller should trigger a recovery due to degraded servers are used in the
// transaction system in the primary data center.
bool shouldTriggerRecoveryDueToDegradedServers() {
if (degradedServers.size() > SERVER_KNOBS->CC_MAX_EXCLUSION_DUE_TO_HEALTH) {
return false;
}
// Whether the transaction system (in primary DC if in HA setting) contains degraded servers.
bool transactionSystemContainsDegradedServers() {
const ServerDBInfo dbi = db.serverInfo->get();
if (dbi.recoveryState < RecoveryState::ACCEPTING_COMMITS) {
return false;
}
// Do not trigger recovery if the cluster controller is excluded, since the master will change
// anyways once the cluster controller is moved
if (id_worker[clusterControllerProcessId].priorityInfo.isExcluded) {
return false;
}
for (const auto& excludedServer : degradedServers) {
if (dbi.master.addresses().contains(excludedServer)) {
return true;
@ -2985,6 +2991,93 @@ public:
return false;
}
// Whether transaction system in the remote DC, e.g. log router and tlogs in the remote DC, contains degraded
// servers.
bool remoteTransactionSystemContainsDegradedServers() {
if (db.config.usableRegions <= 1) {
return false;
}
for (const auto& excludedServer : degradedServers) {
if (addressInDbAndRemoteDc(excludedServer, db.serverInfo)) {
return true;
}
}
return false;
}
// Returns true if remote DC is healthy and can failover to.
bool remoteDCIsHealthy() {
// When we just start, we ignore any remote DC health info since the current CC may be elected at wrong DC due
// to that all the processes are still starting.
if (machineStartTime() == 0) {
return true;
}
if (now() - machineStartTime() < SERVER_KNOBS->INITIAL_UPDATE_CROSS_DC_INFO_DELAY) {
return true;
}
// When remote DC health is not monitored, we may not know whether the remote is healthy or not. So return false
// here to prevent failover.
if (!remoteDCMonitorStarted) {
return false;
}
return !remoteTransactionSystemContainsDegradedServers();
}
// Returns true when the cluster controller should trigger a recovery due to degraded servers used in the
// transaction system in the primary data center.
bool shouldTriggerRecoveryDueToDegradedServers() {
if (degradedServers.size() > SERVER_KNOBS->CC_MAX_EXCLUSION_DUE_TO_HEALTH) {
return false;
}
if (db.serverInfo->get().recoveryState < RecoveryState::ACCEPTING_COMMITS) {
return false;
}
// Do not trigger recovery if the cluster controller is excluded, since the master will change
// anyways once the cluster controller is moved
if (id_worker[clusterControllerProcessId].priorityInfo.isExcluded) {
return false;
}
return transactionSystemContainsDegradedServers();
}
// Returns true when the cluster controller should trigger a failover due to degraded servers used in the
// transaction system in the primary data center, and no degradation in the remote data center.
bool shouldTriggerFailoverDueToDegradedServers() {
if (db.config.usableRegions <= 1) {
return false;
}
if (SERVER_KNOBS->CC_FAILOVER_DUE_TO_HEALTH_MIN_DEGRADATION >
SERVER_KNOBS->CC_FAILOVER_DUE_TO_HEALTH_MAX_DEGRADATION) {
TraceEvent(SevWarn, "TriggerFailoverDueToDegradedServersInvalidConfig")
.suppressFor(1.0)
.detail("Min", SERVER_KNOBS->CC_FAILOVER_DUE_TO_HEALTH_MIN_DEGRADATION)
.detail("Max", SERVER_KNOBS->CC_FAILOVER_DUE_TO_HEALTH_MAX_DEGRADATION);
return false;
}
if (degradedServers.size() < SERVER_KNOBS->CC_FAILOVER_DUE_TO_HEALTH_MIN_DEGRADATION ||
degradedServers.size() > SERVER_KNOBS->CC_FAILOVER_DUE_TO_HEALTH_MAX_DEGRADATION) {
return false;
}
// Do not trigger recovery if the cluster controller is excluded, since the master will change
// anyways once the cluster controller is moved
if (id_worker[clusterControllerProcessId].priorityInfo.isExcluded) {
return false;
}
return transactionSystemContainsDegradedServers() && !remoteTransactionSystemContainsDegradedServers();
}
int recentRecoveryCountDueToHealth() {
while (!recentHealthTriggeredRecoveryTime.empty() &&
now() - recentHealthTriggeredRecoveryTime.front() > SERVER_KNOBS->CC_TRACKING_HEALTH_RECOVERY_INTERVAL) {
@ -3036,11 +3129,15 @@ public:
PromiseStream<Future<Void>> addActor;
bool versionDifferenceUpdated;
bool remoteDCMonitorStarted;
bool remoteTransactionSystemDegraded;
// recruitX is used to signal when role X needs to be (re)recruited.
// recruitingXID is used to track the ID of X's interface which is being recruited.
// We use AsyncVars to kill (i.e. halt) singletons that have been replaced.
AsyncVar<bool> recruitDistributor;
Optional<UID> recruitingDistributorID;
AsyncVar<bool> recruitRatekeeper;
Optional<UID> recruitingRatekeeperID;
@ -3080,8 +3177,8 @@ public:
clusterControllerDcId(locality.dcId()), id(ccInterface.id()), ac(false), outstandingRequestChecker(Void()),
outstandingRemoteRequestChecker(Void()), startTime(now()), goodRecruitmentTime(Never()),
goodRemoteRecruitmentTime(Never()), datacenterVersionDifference(0), versionDifferenceUpdated(false),
recruitDistributor(false), recruitRatekeeper(false),
clusterControllerMetrics("ClusterController", id.toString()),
remoteDCMonitorStarted(false), remoteTransactionSystemDegraded(false), recruitDistributor(false),
recruitRatekeeper(false), clusterControllerMetrics("ClusterController", id.toString()),
openDatabaseRequests("OpenDatabaseRequests", clusterControllerMetrics),
registerWorkerRequests("RegisterWorkerRequests", clusterControllerMetrics),
getWorkersRequests("GetWorkersRequests", clusterControllerMetrics),
@ -3097,6 +3194,8 @@ public:
serverInfo.myLocality = locality;
db.serverInfo->set(serverInfo);
cx = openDBOnServer(db.serverInfo, TaskPriority::DefaultEndpoint, LockAware::True);
specialCounter(clusterControllerMetrics, "ClientCount", [this]() { return db.clientCount; });
}
~ClusterControllerData() {
@ -4645,6 +4744,31 @@ ACTOR Future<Void> updateDatacenterVersionDifference(ClusterControllerData* self
}
}
// A background actor that periodically checks remote DC health, and `checkOutstandingRequests` if remote DC recovers.
ACTOR Future<Void> updateRemoteDCHealth(ClusterControllerData* self) {
// The purpose of the initial delay is to wait for the cluster to achieve a steady state before checking remote DC
// health, since remote DC healthy may trigger a failover, and we don't want that to happen too frequently.
wait(delay(SERVER_KNOBS->INITIAL_UPDATE_CROSS_DC_INFO_DELAY));
self->remoteDCMonitorStarted = true;
// When the remote DC health just start, we may just recover from a health degradation. Check if we can failback if
// we are currently in the remote DC in the database configuration.
if (!self->remoteTransactionSystemDegraded) {
checkOutstandingRequests(self);
}
loop {
bool oldRemoteTransactionSystemDegraded = self->remoteTransactionSystemDegraded;
self->remoteTransactionSystemDegraded = self->remoteTransactionSystemContainsDegradedServers();
if (oldRemoteTransactionSystemDegraded && !self->remoteTransactionSystemDegraded) {
checkOutstandingRequests(self);
}
wait(delay(SERVER_KNOBS->CHECK_REMOTE_HEALTH_INTERVAL));
}
}
ACTOR Future<Void> doEmptyCommit(Database cx) {
state Transaction tr(cx);
loop {
@ -4953,6 +5077,24 @@ ACTOR Future<Void> workerHealthMonitor(ClusterControllerData* self) {
self->excludedDegradedServers.clear();
TraceEvent("DegradedServerDetectedAndSuggestRecovery").log();
}
} else if (self->shouldTriggerFailoverDueToDegradedServers()) {
double ccUpTime = now() - machineStartTime();
if (SERVER_KNOBS->CC_HEALTH_TRIGGER_FAILOVER &&
ccUpTime > SERVER_KNOBS->INITIAL_UPDATE_CROSS_DC_INFO_DELAY) {
TraceEvent("DegradedServerDetectedAndTriggerFailover").log();
std::vector<Optional<Key>> dcPriority;
auto remoteDcId = self->db.config.regions[0].dcId == self->clusterControllerDcId.get()
? self->db.config.regions[1].dcId
: self->db.config.regions[0].dcId;
// Switch the current primary DC and remote DC in desiredDcIds, so that the remote DC becomes
// the new primary, and the primary DC becomes the new remote.
dcPriority.push_back(remoteDcId);
dcPriority.push_back(self->clusterControllerDcId);
self->desiredDcIds.set(dcPriority);
} else {
TraceEvent("DegradedServerDetectedAndSuggestFailover").detail("CCUpTime", ccUpTime);
}
}
}
@ -5005,6 +5147,7 @@ ACTOR Future<Void> clusterControllerCore(ClusterControllerFullInterface interf,
if (SERVER_KNOBS->CC_ENABLE_WORKER_HEALTH_MONITOR) {
self.addActor.send(workerHealthMonitor(&self));
self.addActor.send(updateRemoteDCHealth(&self));
}
loop choose {
@ -5409,18 +5552,19 @@ TEST_CASE("/fdbserver/clustercontroller/shouldTriggerRecoveryDueToDegradedServer
NetworkAddress backup(IPAddress(0x06060606), 1);
NetworkAddress proxy(IPAddress(0x07070707), 1);
NetworkAddress resolver(IPAddress(0x08080808), 1);
UID testUID(1, 2);
// Create a ServerDBInfo using above addresses.
ServerDBInfo testDbInfo;
testDbInfo.master.changeCoordinators =
RequestStream<struct ChangeCoordinatorsRequest>(Endpoint({ master }, UID(1, 2)));
RequestStream<struct ChangeCoordinatorsRequest>(Endpoint({ master }, testUID));
TLogInterface localTLogInterf;
localTLogInterf.peekMessages = RequestStream<struct TLogPeekRequest>(Endpoint({ tlog }, UID(1, 2)));
localTLogInterf.peekMessages = RequestStream<struct TLogPeekRequest>(Endpoint({ tlog }, testUID));
TLogInterface localLogRouterInterf;
localLogRouterInterf.peekMessages = RequestStream<struct TLogPeekRequest>(Endpoint({ logRouter }, UID(1, 2)));
localLogRouterInterf.peekMessages = RequestStream<struct TLogPeekRequest>(Endpoint({ logRouter }, testUID));
BackupInterface backupInterf;
backupInterf.waitFailure = RequestStream<ReplyPromise<Void>>(Endpoint({ backup }, UID(1, 2)));
backupInterf.waitFailure = RequestStream<ReplyPromise<Void>>(Endpoint({ backup }, testUID));
TLogSet localTLogSet;
localTLogSet.isLocal = true;
localTLogSet.tLogs.push_back(OptionalInterface(localTLogInterf));
@ -5429,7 +5573,7 @@ TEST_CASE("/fdbserver/clustercontroller/shouldTriggerRecoveryDueToDegradedServer
testDbInfo.logSystemConfig.tLogs.push_back(localTLogSet);
TLogInterface sateTLogInterf;
sateTLogInterf.peekMessages = RequestStream<struct TLogPeekRequest>(Endpoint({ satelliteTlog }, UID(1, 2)));
sateTLogInterf.peekMessages = RequestStream<struct TLogPeekRequest>(Endpoint({ satelliteTlog }, testUID));
TLogSet sateTLogSet;
sateTLogSet.isLocal = true;
sateTLogSet.locality = tagLocalitySatellite;
@ -5437,18 +5581,18 @@ TEST_CASE("/fdbserver/clustercontroller/shouldTriggerRecoveryDueToDegradedServer
testDbInfo.logSystemConfig.tLogs.push_back(sateTLogSet);
TLogInterface remoteTLogInterf;
remoteTLogInterf.peekMessages = RequestStream<struct TLogPeekRequest>(Endpoint({ remoteTlog }, UID(1, 2)));
remoteTLogInterf.peekMessages = RequestStream<struct TLogPeekRequest>(Endpoint({ remoteTlog }, testUID));
TLogSet remoteTLogSet;
remoteTLogSet.isLocal = false;
remoteTLogSet.tLogs.push_back(OptionalInterface(remoteTLogInterf));
testDbInfo.logSystemConfig.tLogs.push_back(remoteTLogSet);
GrvProxyInterface proxyInterf;
proxyInterf.getConsistentReadVersion = RequestStream<struct GetReadVersionRequest>(Endpoint({ proxy }, UID(1, 2)));
proxyInterf.getConsistentReadVersion = RequestStream<struct GetReadVersionRequest>(Endpoint({ proxy }, testUID));
testDbInfo.client.grvProxies.push_back(proxyInterf);
ResolverInterface resolverInterf;
resolverInterf.resolve = RequestStream<struct ResolveTransactionBatchRequest>(Endpoint({ resolver }, UID(1, 2)));
resolverInterf.resolve = RequestStream<struct ResolveTransactionBatchRequest>(Endpoint({ resolver }, testUID));
testDbInfo.resolvers.push_back(resolverInterf);
testDbInfo.recoveryState = RecoveryState::ACCEPTING_COMMITS;
@ -5499,4 +5643,108 @@ TEST_CASE("/fdbserver/clustercontroller/shouldTriggerRecoveryDueToDegradedServer
return Void();
}
TEST_CASE("/fdbserver/clustercontroller/shouldTriggerFailoverDueToDegradedServers") {
// Create a testing ClusterControllerData. Most of the internal states do not matter in this test.
ClusterControllerData data(ClusterControllerFullInterface(),
LocalityData(),
ServerCoordinators(Reference<ClusterConnectionFile>(new ClusterConnectionFile())));
NetworkAddress master(IPAddress(0x01010101), 1);
NetworkAddress tlog(IPAddress(0x02020202), 1);
NetworkAddress satelliteTlog(IPAddress(0x03030303), 1);
NetworkAddress remoteTlog(IPAddress(0x04040404), 1);
NetworkAddress logRouter(IPAddress(0x05050505), 1);
NetworkAddress backup(IPAddress(0x06060606), 1);
NetworkAddress proxy(IPAddress(0x07070707), 1);
NetworkAddress proxy2(IPAddress(0x08080808), 1);
NetworkAddress resolver(IPAddress(0x09090909), 1);
UID testUID(1, 2);
data.db.config.usableRegions = 2;
// Create a ServerDBInfo using above addresses.
ServerDBInfo testDbInfo;
testDbInfo.master.changeCoordinators =
RequestStream<struct ChangeCoordinatorsRequest>(Endpoint({ master }, testUID));
TLogInterface localTLogInterf;
localTLogInterf.peekMessages = RequestStream<struct TLogPeekRequest>(Endpoint({ tlog }, testUID));
TLogInterface localLogRouterInterf;
localLogRouterInterf.peekMessages = RequestStream<struct TLogPeekRequest>(Endpoint({ logRouter }, testUID));
BackupInterface backupInterf;
backupInterf.waitFailure = RequestStream<ReplyPromise<Void>>(Endpoint({ backup }, testUID));
TLogSet localTLogSet;
localTLogSet.isLocal = true;
localTLogSet.tLogs.push_back(OptionalInterface(localTLogInterf));
localTLogSet.logRouters.push_back(OptionalInterface(localLogRouterInterf));
localTLogSet.backupWorkers.push_back(OptionalInterface(backupInterf));
testDbInfo.logSystemConfig.tLogs.push_back(localTLogSet);
TLogInterface sateTLogInterf;
sateTLogInterf.peekMessages = RequestStream<struct TLogPeekRequest>(Endpoint({ satelliteTlog }, testUID));
TLogSet sateTLogSet;
sateTLogSet.isLocal = true;
sateTLogSet.locality = tagLocalitySatellite;
sateTLogSet.tLogs.push_back(OptionalInterface(sateTLogInterf));
testDbInfo.logSystemConfig.tLogs.push_back(sateTLogSet);
TLogInterface remoteTLogInterf;
remoteTLogInterf.peekMessages = RequestStream<struct TLogPeekRequest>(Endpoint({ remoteTlog }, testUID));
TLogSet remoteTLogSet;
remoteTLogSet.isLocal = false;
remoteTLogSet.tLogs.push_back(OptionalInterface(remoteTLogInterf));
testDbInfo.logSystemConfig.tLogs.push_back(remoteTLogSet);
GrvProxyInterface grvProxyInterf;
grvProxyInterf.getConsistentReadVersion = RequestStream<struct GetReadVersionRequest>(Endpoint({ proxy }, testUID));
testDbInfo.client.grvProxies.push_back(grvProxyInterf);
CommitProxyInterface commitProxyInterf;
commitProxyInterf.commit = RequestStream<struct CommitTransactionRequest>(Endpoint({ proxy2 }, testUID));
testDbInfo.client.commitProxies.push_back(commitProxyInterf);
ResolverInterface resolverInterf;
resolverInterf.resolve = RequestStream<struct ResolveTransactionBatchRequest>(Endpoint({ resolver }, testUID));
testDbInfo.resolvers.push_back(resolverInterf);
testDbInfo.recoveryState = RecoveryState::ACCEPTING_COMMITS;
// No failover when no degraded servers.
data.db.serverInfo->set(testDbInfo);
ASSERT(!data.shouldTriggerFailoverDueToDegradedServers());
// No failover when small number of degraded servers
data.degradedServers.insert(master);
ASSERT(!data.shouldTriggerFailoverDueToDegradedServers());
data.degradedServers.clear();
// Trigger failover when enough servers in the txn system are degraded.
data.degradedServers.insert(master);
data.degradedServers.insert(tlog);
data.degradedServers.insert(proxy);
data.degradedServers.insert(proxy2);
data.degradedServers.insert(resolver);
ASSERT(data.shouldTriggerFailoverDueToDegradedServers());
// No failover when usable region is 1.
data.db.config.usableRegions = 1;
ASSERT(!data.shouldTriggerFailoverDueToDegradedServers());
data.db.config.usableRegions = 2;
// No failover when remote is also degraded.
data.degradedServers.insert(remoteTlog);
ASSERT(!data.shouldTriggerFailoverDueToDegradedServers());
data.degradedServers.clear();
// No failover when some are not from transaction system
data.degradedServers.insert(NetworkAddress(IPAddress(0x13131313), 1));
data.degradedServers.insert(NetworkAddress(IPAddress(0x13131313), 2));
data.degradedServers.insert(NetworkAddress(IPAddress(0x13131313), 3));
data.degradedServers.insert(NetworkAddress(IPAddress(0x13131313), 4));
data.degradedServers.insert(NetworkAddress(IPAddress(0x13131313), 5));
ASSERT(!data.shouldTriggerFailoverDueToDegradedServers());
data.degradedServers.clear();
return Void();
}
} // namespace

24
fdbserver/DataDistribution.actor.cpp
View File

@ -3096,7 +3096,8 @@ ACTOR Future<Void> printSnapshotTeamsInfo(Reference<DDTeamCollection> self) {
auto const& keys = self->server_status.getKeys();
for (auto const& key : keys) {
server_status.emplace(key, self->server_status.get(key));
// Add to or update the local server_status map
server_status[key] = self->server_status.get(key);
}
TraceEvent("DDPrintSnapshotTeasmInfo", self->distributorId)
@ -3131,13 +3132,22 @@ ACTOR Future<Void> printSnapshotTeamsInfo(Reference<DDTeamCollection> self) {
server = server_info.begin();
for (i = 0; i < server_info.size(); i++) {
const UID& uid = server->first;
TraceEvent("ServerStatus", self->distributorId)
.detail("ServerUID", uid)
.detail("Healthy", !get(server_status, uid).isUnhealthy())
TraceEvent e("ServerStatus", self->distributorId);
e.detail("ServerUID", uid)
.detail("MachineIsValid", server_info[uid]->machine.isValid())
.detail("MachineTeamSize",
server_info[uid]->machine.isValid() ? server_info[uid]->machine->machineTeams.size() : -1)
.detail("Primary", self->primary);
// ServerStatus might not be known if server was very recently added and storageServerFailureTracker()
// has not yet updated self->server_status
// If the UID is not found, do not assume the server is healthy or unhealthy
auto it = server_status.find(uid);
if (it != server_status.end()) {
e.detail("Healthy", !it->second.isUnhealthy());
}
server++;
if (++traceEventsPrinted % SERVER_KNOBS->DD_TEAMS_INFO_PRINT_YIELD_COUNT == 0) {
wait(yield());
@ -3174,7 +3184,11 @@ ACTOR Future<Void> printSnapshotTeamsInfo(Reference<DDTeamCollection> self) {
// Healthy machine has at least one healthy server
for (auto& server : _machine->serversOnMachine) {
if (!get(server_status, server->id).isUnhealthy()) {
// ServerStatus might not be known if server was very recently added and
// storageServerFailureTracker() has not yet updated self->server_status If the UID is not found, do
// not assume the server is healthy
auto it = server_status.find(server->id);
if (it != server_status.end() && !it->second.isUnhealthy()) {
isMachineHealthy = true;
}
}

18
fdbserver/IKeyValueStore.h
View File

@ -47,16 +47,30 @@ public:
virtual Future<Void> commit(
bool sequential = false) = 0; // returns when prior sets and clears are (atomically) durable
virtual Future<Optional<Value>> readValue(KeyRef key, Optional<UID> debugID = Optional<UID>()) = 0;
enum class ReadType {
EAGER,
FETCH,
LOW,
NORMAL,
HIGH,
};
virtual Future<Optional<Value>> readValue(KeyRef key,
ReadType type = ReadType::NORMAL,
Optional<UID> debugID = Optional<UID>()) = 0;
// Like readValue(), but returns only the first maxLength bytes of the value if it is longer
virtual Future<Optional<Value>> readValuePrefix(KeyRef key,
int maxLength,
ReadType type = ReadType::NORMAL,
Optional<UID> debugID = Optional<UID>()) = 0;
// If rowLimit>=0, reads first rows sorted ascending, otherwise reads last rows sorted descending
// The total size of the returned value (less the last entry) will be less than byteLimit
virtual Future<RangeResult> readRange(KeyRangeRef keys, int rowLimit = 1 << 30, int byteLimit = 1 << 30) = 0;
virtual Future<RangeResult> readRange(KeyRangeRef keys,
int rowLimit = 1 << 30,
int byteLimit = 1 << 30,
ReadType type = ReadType::NORMAL) = 0;
// To debug MEMORY_RADIXTREE type ONLY
// Returns (1) how many key & value pairs have been inserted (2) how many nodes have been created (3) how many

15
fdbserver/IPager.h
View File

@ -257,8 +257,10 @@ public:
// The snapshot shall be usable until setOldVersion() is called with a version > v.
virtual Reference<IPagerSnapshot> getReadSnapshot(Version v) = 0;
// Atomically make durable all pending page writes, page frees, and update the metadata string.
virtual Future<Void> commit() = 0;
// Atomically make durable all pending page writes, page frees, and update the metadata string,
// setting the committed version to v
// v must be >= the highest versioned page write.
virtual Future<Void> commit(Version v) = 0;
// Get the latest meta key set or committed
virtual Key getMetaKey() const = 0;
@ -266,9 +268,6 @@ public:
// Set the metakey which will be stored in the next commit
virtual void setMetaKey(KeyRef metaKey) = 0;
// Sets the next commit version
virtual void setCommitVersion(Version v) = 0;
virtual StorageBytes getStorageBytes() const = 0;
virtual int64_t getPageCount() = 0;
@ -282,16 +281,16 @@ public:
virtual Future<Void> init() = 0;
// Returns latest committed version
virtual Version getLatestVersion() const = 0;
virtual Version getLastCommittedVersion() const = 0;
// Returns the oldest readable version as of the most recent committed version
virtual Version getOldestVersion() const = 0;
virtual Version getOldestReadableVersion() const = 0;
// Sets the oldest readable version to be put into affect at the next commit.
// The pager can reuse pages that were freed at a version less than v.
// If any snapshots are in use at a version less than v, the pager can either forcefully
// invalidate them or keep their versions around until the snapshots are no longer in use.
virtual void setOldestVersion(Version v) = 0;
virtual void setOldestReadableVersion(Version v) = 0;
protected:
~IPager2() {} // Destruction should be done using close()/dispose() from the IClosable interface

9
fdbserver/KeyValueStoreCompressTestData.actor.cpp
View File

@ -56,7 +56,7 @@ struct KeyValueStoreCompressTestData final : IKeyValueStore {
void clear(KeyRangeRef range, const Arena* arena = nullptr) override { store->clear(range, arena); }
Future<Void> commit(bool sequential = false) override { return store->commit(sequential); }
Future<Optional<Value>> readValue(KeyRef key, Optional<UID> debugID = Optional<UID>()) override {
Future<Optional<Value>> readValue(KeyRef key, IKeyValueStore::ReadType, Optional<UID> debugID) override {
return doReadValue(store, key, debugID);
}
@ -66,19 +66,20 @@ struct KeyValueStoreCompressTestData final : IKeyValueStore {
// reason, you will need to fix this.
Future<Optional<Value>> readValuePrefix(KeyRef key,
int maxLength,
Optional<UID> debugID = Optional<UID>()) override {
IKeyValueStore::ReadType,
Optional<UID> debugID) override {
return doReadValuePrefix(store, key, maxLength, debugID);
}
// If rowLimit>=0, reads first rows sorted ascending, otherwise reads last rows sorted descending
// The total size of the returned value (less the last entry) will be less than byteLimit
Future<RangeResult> readRange(KeyRangeRef keys, int rowLimit = 1 << 30, int byteLimit = 1 << 30) override {
Future<RangeResult> readRange(KeyRangeRef keys, int rowLimit, int byteLimit, IKeyValueStore::ReadType) override {
return doReadRange(store, keys, rowLimit, byteLimit);
}
private:
ACTOR static Future<Optional<Value>> doReadValue(IKeyValueStore* store, Key key, Optional<UID> debugID) {
Optional<Value> v = wait(store->readValue(key, debugID));
Optional<Value> v = wait(store->readValue(key, IKeyValueStore::ReadType::NORMAL, debugID));
if (!v.present())
return v;
return unpack(v.get());

13
fdbserver/KeyValueStoreMemory.actor.cpp
View File

@ -194,7 +194,7 @@ public:
return c;
}
Future<Optional<Value>> readValue(KeyRef key, Optional<UID> debugID = Optional<UID>()) override {
Future<Optional<Value>> readValue(KeyRef key, IKeyValueStore::ReadType, Optional<UID> debugID) override {
if (recovering.isError())
throw recovering.getError();
if (!recovering.isReady())
@ -208,7 +208,8 @@ public:
Future<Optional<Value>> readValuePrefix(KeyRef key,
int maxLength,
Optional<UID> debugID = Optional<UID>()) override {
IKeyValueStore::ReadType,
Optional<UID> debugID) override {
if (recovering.isError())
throw recovering.getError();
if (!recovering.isReady())
@ -227,7 +228,7 @@ public:
// If rowLimit>=0, reads first rows sorted ascending, otherwise reads last rows sorted descending
// The total size of the returned value (less the last entry) will be less than byteLimit
Future<RangeResult> readRange(KeyRangeRef keys, int rowLimit = 1 << 30, int byteLimit = 1 << 30) override {
Future<RangeResult> readRange(KeyRangeRef keys, int rowLimit, int byteLimit, IKeyValueStore::ReadType) override {
if (recovering.isError())
throw recovering.getError();
if (!recovering.isReady())
@ -826,18 +827,18 @@ private:
ACTOR static Future<Optional<Value>> waitAndReadValue(KeyValueStoreMemory* self, Key key) {
wait(self->recovering);
return self->readValue(key).get();
return static_cast<IKeyValueStore*>(self)->readValue(key).get();
}
ACTOR static Future<Optional<Value>> waitAndReadValuePrefix(KeyValueStoreMemory* self, Key key, int maxLength) {
wait(self->recovering);
return self->readValuePrefix(key, maxLength).get();
return static_cast<IKeyValueStore*>(self)->readValuePrefix(key, maxLength).get();
}
ACTOR static Future<RangeResult> waitAndReadRange(KeyValueStoreMemory* self,
KeyRange keys,
int rowLimit,
int byteLimit) {
wait(self->recovering);
return self->readRange(keys, rowLimit, byteLimit).get();
return static_cast<IKeyValueStore*>(self)->readRange(keys, rowLimit, byteLimit).get();
}
ACTOR static Future<Void> waitAndCommit(KeyValueStoreMemory* self, bool sequential) {
wait(self->recovering);

9
fdbserver/KeyValueStoreRocksDB.actor.cpp
View File

@ -645,21 +645,24 @@ struct RocksDBKeyValueStore : IKeyValueStore {
return res;
}
Future<Optional<Value>> readValue(KeyRef key, Optional<UID> debugID) override {
Future<Optional<Value>> readValue(KeyRef key, IKeyValueStore::ReadType, Optional<UID> debugID) override {
auto a = new Reader::ReadValueAction(key, debugID);
auto res = a->result.getFuture();
readThreads->post(a);
return res;
}
Future<Optional<Value>> readValuePrefix(KeyRef key, int maxLength, Optional<UID> debugID) override {
Future<Optional<Value>> readValuePrefix(KeyRef key,
int maxLength,
IKeyValueStore::ReadType,
Optional<UID> debugID) override {
auto a = new Reader::ReadValuePrefixAction(key, maxLength, debugID);
auto res = a->result.getFuture();
readThreads->post(a);
return res;
}
Future<RangeResult> readRange(KeyRangeRef keys, int rowLimit, int byteLimit) override {
Future<RangeResult> readRange(KeyRangeRef keys, int rowLimit, int byteLimit, IKeyValueStore::ReadType) override {
auto a = new Reader::ReadRangeAction(keys, rowLimit, byteLimit);
auto res = a->result.getFuture();
readThreads->post(a);

21
fdbserver/KeyValueStoreSQLite.actor.cpp
View File

@ -1577,9 +1577,12 @@ public:
void clear(KeyRangeRef range, const Arena* arena = nullptr) override;
Future<Void> commit(bool sequential = false) override;
Future<Optional<Value>> readValue(KeyRef key, Optional<UID> debugID) override;
Future<Optional<Value>> readValuePrefix(KeyRef key, int maxLength, Optional<UID> debugID) override;
Future<RangeResult> readRange(KeyRangeRef keys, int rowLimit = 1 << 30, int byteLimit = 1 << 30) override;
Future<Optional<Value>> readValue(KeyRef key, IKeyValueStore::ReadType, Optional<UID> debugID) override;
Future<Optional<Value>> readValuePrefix(KeyRef key,
int maxLength,
IKeyValueStore::ReadType,
Optional<UID> debugID) override;
Future<RangeResult> readRange(KeyRangeRef keys, int rowLimit, int byteLimit, IKeyValueStore::ReadType) override;
KeyValueStoreSQLite(std::string const& filename,
UID logID,
@ -2192,21 +2195,27 @@ Future<Void> KeyValueStoreSQLite::commit(bool sequential) {
writeThread->post(p);
return f;
}
Future<Optional<Value>> KeyValueStoreSQLite::readValue(KeyRef key, Optional<UID> debugID) {
Future<Optional<Value>> KeyValueStoreSQLite::readValue(KeyRef key, IKeyValueStore::ReadType, Optional<UID> debugID) {
++readsRequested;
auto p = new Reader::ReadValueAction(key, debugID);
auto f = p->result.getFuture();
readThreads->post(p);
return f;
}
Future<Optional<Value>> KeyValueStoreSQLite::readValuePrefix(KeyRef key, int maxLength, Optional<UID> debugID) {
Future<Optional<Value>> KeyValueStoreSQLite::readValuePrefix(KeyRef key,
int maxLength,
IKeyValueStore::ReadType,
Optional<UID> debugID) {
++readsRequested;
auto p = new Reader::ReadValuePrefixAction(key, maxLength, debugID);
auto f = p->result.getFuture();
readThreads->post(p);
return f;
}
Future<RangeResult> KeyValueStoreSQLite::readRange(KeyRangeRef keys, int rowLimit, int byteLimit) {
Future<RangeResult> KeyValueStoreSQLite::readRange(KeyRangeRef keys,
int rowLimit,
int byteLimit,
IKeyValueStore::ReadType) {
++readsRequested;
auto p = new Reader::ReadRangeAction(keys, rowLimit, byteLimit);
auto f = p->result.getFuture();

584
fdbserver/VersionedBTree.actor.cpp
View File

File diff suppressed because it is too large Load Diff

4
fdbserver/WorkerInterface.actor.h
View File

@ -967,6 +967,10 @@ ACTOR Future<Void> backupWorker(BackupInterface bi,
Reference<AsyncVar<ServerDBInfo> const> db);
void registerThreadForProfiling();
// Returns true if `address` is used in the db (indicated by `dbInfo`) transaction system and in the db's remote DC.
bool addressInDbAndRemoteDc(const NetworkAddress& address, Reference<AsyncVar<ServerDBInfo> const> dbInfo);
void updateCpuProfiler(ProfilerRequest req);
namespace oldTLog_4_6 {

3
fdbserver/masterserver.actor.cpp
View File

@ -665,7 +665,8 @@ ACTOR Future<Standalone<CommitTransactionRef>> provisionalMaster(Reference<Maste
loop choose {
when(GetReadVersionRequest req =
waitNext(parent->provisionalGrvProxies[0].getConsistentReadVersion.getFuture())) {
if (req.flags & GetReadVersionRequest::FLAG_CAUSAL_READ_RISKY && parent->lastEpochEnd) {
if ((req.flags & GetReadVersionRequest::FLAG_CAUSAL_READ_RISKY) &&
(req.flags & GetReadVersionRequest::FLAG_USE_PROVISIONAL_PROXIES) && parent->lastEpochEnd) {
GetReadVersionReply rep;
rep.version = parent->lastEpochEnd;
rep.locked = locked;

80
fdbserver/storageserver.actor.cpp
View File

@ -195,15 +195,25 @@ struct StorageServerDisk {
Future<Void> commit() { return storage->commit(); }
// SOMEDAY: Put readNextKeyInclusive in IKeyValueStore
Future<Key> readNextKeyInclusive(KeyRef key) { return readFirstKey(storage, KeyRangeRef(key, allKeys.end)); }
Future<Optional<Value>> readValue(KeyRef key, Optional<UID> debugID = Optional<UID>()) {
return storage->readValue(key, debugID);
Future<Key> readNextKeyInclusive(KeyRef key, IKeyValueStore::ReadType type = IKeyValueStore::ReadType::NORMAL) {
return readFirstKey(storage, KeyRangeRef(key, allKeys.end), type);
}
Future<Optional<Value>> readValuePrefix(KeyRef key, int maxLength, Optional<UID> debugID = Optional<UID>()) {
return storage->readValuePrefix(key, maxLength, debugID);
Future<Optional<Value>> readValue(KeyRef key,
IKeyValueStore::ReadType type = IKeyValueStore::ReadType::NORMAL,
Optional<UID> debugID = Optional<UID>()) {
return storage->readValue(key, type, debugID);
}
Future<RangeResult> readRange(KeyRangeRef keys, int rowLimit = 1 << 30, int byteLimit = 1 << 30) {
return storage->readRange(keys, rowLimit, byteLimit);
Future<Optional<Value>> readValuePrefix(KeyRef key,
int maxLength,
IKeyValueStore::ReadType type = IKeyValueStore::ReadType::NORMAL,
Optional<UID> debugID = Optional<UID>()) {
return storage->readValuePrefix(key, maxLength, type, debugID);
}
Future<RangeResult> readRange(KeyRangeRef keys,
int rowLimit = 1 << 30,
int byteLimit = 1 << 30,
IKeyValueStore::ReadType type = IKeyValueStore::ReadType::NORMAL) {
return storage->readRange(keys, rowLimit, byteLimit, type);
}
KeyValueStoreType getKeyValueStoreType() const { return storage->getType(); }
@ -216,8 +226,8 @@ private:
void writeMutations(const VectorRef<MutationRef>& mutations, Version debugVersion, const char* debugContext);
ACTOR static Future<Key> readFirstKey(IKeyValueStore* storage, KeyRangeRef range) {
RangeResult r = wait(storage->readRange(range, 1));
ACTOR static Future<Key> readFirstKey(IKeyValueStore* storage, KeyRangeRef range, IKeyValueStore::ReadType type) {
RangeResult r = wait(storage->readRange(range, 1, 1 << 30, type));
if (r.size())
return r[0].key;
else
@ -1331,7 +1341,7 @@ ACTOR Future<Void> getValueQ(StorageServer* data, GetValueRequest req) {
path = 1;
} else if (!i || !i->isClearTo() || i->getEndKey() <= req.key) {
path = 2;
Optional<Value> vv = wait(data->storage.readValue(req.key, req.debugID));
Optional<Value> vv = wait(data->storage.readValue(req.key, IKeyValueStore::ReadType::NORMAL, req.debugID));
// Validate that while we were reading the data we didn't lose the version or shard
if (version < data->storageVersion()) {
TEST(true); // transaction_too_old after readValue
@ -1681,7 +1691,8 @@ ACTOR Future<GetKeyValuesReply> readRange(StorageServer* data,
KeyRange range,
int limit,
int* pLimitBytes,
SpanID parentSpan) {
SpanID parentSpan,
IKeyValueStore::ReadType type) {
state GetKeyValuesReply result;
state StorageServer::VersionedData::ViewAtVersion view = data->data().at(version);
state StorageServer::VersionedData::iterator vCurrent = view.end();
@ -1745,7 +1756,7 @@ ACTOR Future<GetKeyValuesReply> readRange(StorageServer* data,
// Read the data on disk up to vCurrent (or the end of the range)
readEnd = vCurrent ? std::min(vCurrent.key(), range.end) : range.end;
RangeResult atStorageVersion =
wait(data->storage.readRange(KeyRangeRef(readBegin, readEnd), limit, *pLimitBytes));
wait(data->storage.readRange(KeyRangeRef(readBegin, readEnd), limit, *pLimitBytes, type));
ASSERT(atStorageVersion.size() <= limit);
if (data->storageVersion() > version)
@ -1826,7 +1837,7 @@ ACTOR Future<GetKeyValuesReply> readRange(StorageServer* data,
readBegin = vCurrent ? std::max(vCurrent->isClearTo() ? vCurrent->getEndKey() : vCurrent.key(), range.begin)
: range.begin;
RangeResult atStorageVersion =
wait(data->storage.readRange(KeyRangeRef(readBegin, readEnd), limit, *pLimitBytes));
wait(data->storage.readRange(KeyRangeRef(readBegin, readEnd), limit, *pLimitBytes, type));
ASSERT(atStorageVersion.size() <= -limit);
if (data->storageVersion() > version)
@ -1883,7 +1894,8 @@ ACTOR Future<Key> findKey(StorageServer* data,
Version version,
KeyRange range,
int* pOffset,
SpanID parentSpan)
SpanID parentSpan,
IKeyValueStore::ReadType type)
// Attempts to find the key indicated by sel in the data at version, within range.
// Precondition: selectorInRange(sel, range)
// If it is found, offset is set to 0 and a key is returned which falls inside range.
@ -1921,7 +1933,8 @@ ACTOR Future<Key> findKey(StorageServer* data,
forward ? KeyRangeRef(sel.getKey(), range.end) : KeyRangeRef(range.begin, keyAfter(sel.getKey())),
(distance + skipEqualKey) * sign,
&maxBytes,
span.context));
span.context,
type));
state bool more = rep.more && rep.data.size() != distance + skipEqualKey;
// If we get only one result in the reverse direction as a result of the data being too large, we could get stuck in
@ -1929,8 +1942,8 @@ ACTOR Future<Key> findKey(StorageServer* data,
if (more && !forward && rep.data.size() == 1) {
TEST(true); // Reverse key selector returned only one result in range read
maxBytes = std::numeric_limits<int>::max();
GetKeyValuesReply rep2 = wait(
readRange(data, version, KeyRangeRef(range.begin, keyAfter(sel.getKey())), -2, &maxBytes, span.context));
GetKeyValuesReply rep2 = wait(readRange(
data, version, KeyRangeRef(range.begin, keyAfter(sel.getKey())), -2, &maxBytes, span.context, type));
rep = rep2;
more = rep.more && rep.data.size() != distance + skipEqualKey;
ASSERT(rep.data.size() == 2 || !more);
@ -1995,6 +2008,8 @@ ACTOR Future<Void> getKeyValuesQ(StorageServer* data, GetKeyValuesRequest req)
{
state Span span("SS:getKeyValues"_loc, { req.spanContext });
state int64_t resultSize = 0;
state IKeyValueStore::ReadType type =
req.isFetchKeys ? IKeyValueStore::ReadType::FETCH : IKeyValueStore::ReadType::NORMAL;
getCurrentLineage()->modify(&TransactionLineage::txID) = req.spanContext.first();
++data->counters.getRangeQueries;
@ -2041,10 +2056,10 @@ ACTOR Future<Void> getKeyValuesQ(StorageServer* data, GetKeyValuesRequest req)
state int offset2;
state Future<Key> fBegin = req.begin.isFirstGreaterOrEqual()
? Future<Key>(req.begin.getKey())
: findKey(data, req.begin, version, shard, &offset1, span.context);
: findKey(data, req.begin, version, shard, &offset1, span.context, type);
state Future<Key> fEnd = req.end.isFirstGreaterOrEqual()
? Future<Key>(req.end.getKey())
: findKey(data, req.end, version, shard, &offset2, span.context);
: findKey(data, req.end, version, shard, &offset2, span.context, type);
state Key begin = wait(fBegin);
state Key end = wait(fEnd);
@ -2084,8 +2099,8 @@ ACTOR Future<Void> getKeyValuesQ(StorageServer* data, GetKeyValuesRequest req)
} else {
state int remainingLimitBytes = req.limitBytes;
GetKeyValuesReply _r =
wait(readRange(data, version, KeyRangeRef(begin, end), req.limit, &remainingLimitBytes, span.context));
GetKeyValuesReply _r = wait(
readRange(data, version, KeyRangeRef(begin, end), req.limit, &remainingLimitBytes, span.context, type));
GetKeyValuesReply r = _r;
if (req.debugID.present())
@ -2162,6 +2177,8 @@ ACTOR Future<Void> getKeyValuesStreamQ(StorageServer* data, GetKeyValuesStreamRe
{
state Span span("SS:getKeyValuesStream"_loc, { req.spanContext });
state int64_t resultSize = 0;
state IKeyValueStore::ReadType type =
req.isFetchKeys ? IKeyValueStore::ReadType::FETCH : IKeyValueStore::ReadType::NORMAL;
req.reply.setByteLimit(SERVER_KNOBS->RANGESTREAM_LIMIT_BYTES);
++data->counters.getRangeStreamQueries;
@ -2209,10 +2226,10 @@ ACTOR Future<Void> getKeyValuesStreamQ(StorageServer* data, GetKeyValuesStreamRe
state int offset2;
state Future<Key> fBegin = req.begin.isFirstGreaterOrEqual()
? Future<Key>(req.begin.getKey())
: findKey(data, req.begin, version, shard, &offset1, span.context);
: findKey(data, req.begin, version, shard, &offset1, span.context, type);
state Future<Key> fEnd = req.end.isFirstGreaterOrEqual()
? Future<Key>(req.end.getKey())
: findKey(data, req.end, version, shard, &offset2, span.context);
: findKey(data, req.end, version, shard, &offset2, span.context, type);
state Key begin = wait(fBegin);
state Key end = wait(fEnd);
if (req.debugID.present())
@ -2261,7 +2278,7 @@ ACTOR Future<Void> getKeyValuesStreamQ(StorageServer* data, GetKeyValuesStreamRe
? 1
: CLIENT_KNOBS->REPLY_BYTE_LIMIT;
GetKeyValuesReply _r =
wait(readRange(data, version, KeyRangeRef(begin, end), req.limit, &byteLimit, span.context));
wait(readRange(data, version, KeyRangeRef(begin, end), req.limit, &byteLimit, span.context, type));
GetKeyValuesStreamReply r(_r);
if (req.debugID.present())
@ -2363,7 +2380,8 @@ ACTOR Future<Void> getKeyQ(StorageServer* data, GetKeyRequest req) {
state KeyRange shard = getShardKeyRange(data, req.sel);
state int offset;
Key k = wait(findKey(data, req.sel, version, shard, &offset, req.spanContext));
Key k =
wait(findKey(data, req.sel, version, shard, &offset, req.spanContext, IKeyValueStore::ReadType::NORMAL));
data->checkChangeCounter(
changeCounter, KeyRangeRef(std::min<KeyRef>(req.sel.getKey(), k), std::max<KeyRef>(req.sel.getKey(), k)));
@ -2461,7 +2479,7 @@ ACTOR Future<Void> doEagerReads(StorageServer* data, UpdateEagerReadInfo* eager)
if (SERVER_KNOBS->ENABLE_CLEAR_RANGE_EAGER_READS) {
std::vector<Future<Key>> keyEnd(eager->keyBegin.size());
for (int i = 0; i < keyEnd.size(); i++)
keyEnd[i] = data->storage.readNextKeyInclusive(eager->keyBegin[i]);
keyEnd[i] = data->storage.readNextKeyInclusive(eager->keyBegin[i], IKeyValueStore::ReadType::EAGER);
state Future<std::vector<Key>> futureKeyEnds = getAll(keyEnd);
state std::vector<Key> keyEndVal = wait(futureKeyEnds);
@ -2470,7 +2488,8 @@ ACTOR Future<Void> doEagerReads(StorageServer* data, UpdateEagerReadInfo* eager)
std::vector<Future<Optional<Value>>> value(eager->keys.size());
for (int i = 0; i < value.size(); i++)
value[i] = data->storage.readValuePrefix(eager->keys[i].first, eager->keys[i].second);
value[i] =
data->storage.readValuePrefix(eager->keys[i].first, eager->keys[i].second, IKeyValueStore::ReadType::EAGER);
state Future<std::vector<Optional<Value>>> futureValues = getAll(value);
std::vector<Optional<Value>> optionalValues = wait(futureValues);
@ -3603,6 +3622,7 @@ private:
data->recoveryVersionSkips.emplace_back(rollbackVersion, currentVersion - rollbackVersion);
} else if (m.type == MutationRef::SetValue && m.param1 == killStoragePrivateKey) {
TraceEvent("StorageServerWorkerRemoved", data->thisServerID).detail("Reason", "KillStorage");
throw worker_removed();
} else if ((m.type == MutationRef::SetValue || m.type == MutationRef::ClearRange) &&
m.param1.substr(1).startsWith(serverTagPrefix)) {
@ -3612,6 +3632,10 @@ private:
if ((m.type == MutationRef::SetValue && !data->isTss() && !matchesThisServer) ||
(m.type == MutationRef::ClearRange &&
((!data->isTSSInQuarantine() && matchesThisServer) || (data->isTss() && matchesTssPair)))) {
TraceEvent("StorageServerWorkerRemoved", data->thisServerID)
.detail("Reason", "ServerTag")
.detail("TagMatches", matchesThisServer)
.detail("IsTSS", data->isTss());
throw worker_removed();
}
if (!data->isTss() && m.type == MutationRef::ClearRange && data->ssPairID.present() &&
@ -3751,6 +3775,7 @@ ACTOR Future<Void> update(StorageServer* data, bool* pReceivedUpdate) {
}
data->tlogCursorReadsLatencyHistogram->sampleSeconds(now() - beforeTLogCursorReads);
if (cursor->popped() > 0) {
TraceEvent("StorageServerWorkerRemoved", data->thisServerID).detail("Reason", "PeekPoppedTLogData");
throw worker_removed();
}
@ -5413,6 +5438,7 @@ ACTOR Future<Void> replaceTSSInterface(StorageServer* self, StorageServerInterfa
if (!pairTagValue.present()) {
TEST(true); // Race where tss was down, pair was removed, tss starts back up
TraceEvent("StorageServerWorkerRemoved", self->thisServerID).detail("Reason", "TssPairMissing");
throw worker_removed();
}

170
fdbserver/worker.actor.cpp
View File

@ -723,6 +723,88 @@ TEST_CASE("/fdbserver/worker/addressInDbAndPrimaryDc") {
} // namespace
bool addressInDbAndRemoteDc(const NetworkAddress& address, Reference<AsyncVar<ServerDBInfo> const> dbInfo) {
const auto& dbi = dbInfo->get();
for (const auto& logSet : dbi.logSystemConfig.tLogs) {
if (logSet.isLocal || logSet.locality == tagLocalitySatellite) {
continue;
}
for (const auto& tlog : logSet.tLogs) {
if (tlog.present() && tlog.interf().addresses().contains(address)) {
return true;
}
}
for (const auto& logRouter : logSet.logRouters) {
if (logRouter.present() && logRouter.interf().addresses().contains(address)) {
return true;
}
}
}
return false;
}
bool addressesInDbAndRemoteDc(const NetworkAddressList& addresses, Reference<AsyncVar<ServerDBInfo> const> dbInfo) {
return addressInDbAndRemoteDc(addresses.address, dbInfo) ||
(addresses.secondaryAddress.present() && addressInDbAndRemoteDc(addresses.secondaryAddress.get(), dbInfo));
}
namespace {
TEST_CASE("/fdbserver/worker/addressInDbAndRemoteDc") {
// Setup a ServerDBInfo for test.
ServerDBInfo testDbInfo;
LocalityData testLocal;
testLocal.set(LiteralStringRef("dcid"), StringRef(std::to_string(1)));
testDbInfo.master.locality = testLocal;
// First, create an empty TLogInterface, and check that it shouldn't be considered as in remote DC.
testDbInfo.logSystemConfig.tLogs.push_back(TLogSet());
testDbInfo.logSystemConfig.tLogs.back().isLocal = true;
testDbInfo.logSystemConfig.tLogs.back().tLogs.push_back(OptionalInterface<TLogInterface>());
ASSERT(!addressInDbAndRemoteDc(g_network->getLocalAddress(), makeReference<AsyncVar<ServerDBInfo>>(testDbInfo)));
TLogInterface localTlog(testLocal);
localTlog.initEndpoints();
testDbInfo.logSystemConfig.tLogs.back().tLogs.push_back(OptionalInterface(localTlog));
ASSERT(!addressInDbAndRemoteDc(g_network->getLocalAddress(), makeReference<AsyncVar<ServerDBInfo>>(testDbInfo)));
// Create a remote TLog, and it should be considered as in remote DC.
LocalityData fakeRemote;
fakeRemote.set(LiteralStringRef("dcid"), StringRef(std::to_string(2)));
TLogInterface remoteTlog(fakeRemote);
remoteTlog.initEndpoints();
testDbInfo.logSystemConfig.tLogs.push_back(TLogSet());
testDbInfo.logSystemConfig.tLogs.back().isLocal = false;
testDbInfo.logSystemConfig.tLogs.back().tLogs.push_back(OptionalInterface(remoteTlog));
ASSERT(addressInDbAndRemoteDc(g_network->getLocalAddress(), makeReference<AsyncVar<ServerDBInfo>>(testDbInfo)));
// Create a remote log router, and it should be considered as in remote DC.
NetworkAddress logRouterAddress(IPAddress(0x26262626), 1);
TLogInterface remoteLogRouter(fakeRemote);
remoteLogRouter.initEndpoints();
remoteLogRouter.peekMessages = RequestStream<struct TLogPeekRequest>(Endpoint({ logRouterAddress }, UID(1, 2)));
testDbInfo.logSystemConfig.tLogs.back().logRouters.push_back(OptionalInterface(remoteLogRouter));
ASSERT(addressInDbAndRemoteDc(logRouterAddress, makeReference<AsyncVar<ServerDBInfo>>(testDbInfo)));
// Create a satellite tlog, and it shouldn't be considered as in remote DC.
testDbInfo.logSystemConfig.tLogs.push_back(TLogSet());
testDbInfo.logSystemConfig.tLogs.back().locality = tagLocalitySatellite;
NetworkAddress satelliteTLogAddress(IPAddress(0x13131313), 1);
TLogInterface satelliteTLog(fakeRemote);
satelliteTLog.initEndpoints();
satelliteTLog.peekMessages = RequestStream<struct TLogPeekRequest>(Endpoint({ satelliteTLogAddress }, UID(1, 2)));
testDbInfo.logSystemConfig.tLogs.back().tLogs.push_back(OptionalInterface(satelliteTLog));
ASSERT(!addressInDbAndRemoteDc(satelliteTLogAddress, makeReference<AsyncVar<ServerDBInfo>>(testDbInfo)));
return Void();
}
} // namespace
// The actor that actively monitors the health of local and peer servers, and reports anomaly to the cluster controller.
ACTOR Future<Void> healthMonitor(Reference<AsyncVar<Optional<ClusterControllerFullInterface>> const> ccInterface,
WorkerInterface interf,
@ -730,49 +812,63 @@ ACTOR Future<Void> healthMonitor(Reference<AsyncVar<Optional<ClusterControllerFu
Reference<AsyncVar<ServerDBInfo> const> dbInfo) {
loop {
Future<Void> nextHealthCheckDelay = Never();
if (dbInfo->get().recoveryState >= RecoveryState::ACCEPTING_COMMITS &&
addressesInDbAndPrimaryDc(interf.addresses(), dbInfo) && ccInterface->get().present()) {
if (dbInfo->get().recoveryState >= RecoveryState::ACCEPTING_COMMITS && ccInterface->get().present()) {
nextHealthCheckDelay = delay(SERVER_KNOBS->WORKER_HEALTH_MONITOR_INTERVAL);
const auto& allPeers = FlowTransport::transport().getAllPeers();
UpdateWorkerHealthRequest req;
for (const auto& [address, peer] : allPeers) {
if (peer->pingLatencies.getPopulationSize() < SERVER_KNOBS->PEER_LATENCY_CHECK_MIN_POPULATION) {
// Ignore peers that don't have enough samples.
// TODO(zhewu): Currently, FlowTransport latency monitor clears ping latency samples on a regular
// basis, which may affect the measurement count. Currently,
// WORKER_HEALTH_MONITOR_INTERVAL is much smaller than the ping clearance interval, so
// it may be ok. If this ends to be a problem, we need to consider keep track of last
// ping latencies logged.
continue;
}
if (!addressInDbAndPrimaryDc(address, dbInfo)) {
// Ignore the servers that are not in the database's transaction system and not in the primary DC.
// Note that currently we are not monitor storage servers, since lagging in storage servers today
// already can trigger server exclusion by data distributor.
continue;
}
bool workerInDb = false;
bool workerInPrimary = false;
if (addressesInDbAndPrimaryDc(interf.addresses(), dbInfo)) {
workerInDb = true;
workerInPrimary = true;
} else if (addressesInDbAndRemoteDc(interf.addresses(), dbInfo)) {
workerInDb = true;
workerInPrimary = false;
}
if (peer->pingLatencies.percentile(SERVER_KNOBS->PEER_LATENCY_DEGRADATION_PERCENTILE) >
SERVER_KNOBS->PEER_LATENCY_DEGRADATION_THRESHOLD ||
peer->timeoutCount / (double)(peer->pingLatencies.getPopulationSize()) >
SERVER_KNOBS->PEER_TIMEOUT_PERCENTAGE_DEGRADATION_THRESHOLD) {
// This is a degraded peer.
TraceEvent("HealthMonitorDetectDegradedPeer")
.suppressFor(30)
.detail("Peer", address)
.detail("Elapsed", now() - peer->lastLoggedTime)
.detail("MinLatency", peer->pingLatencies.min())
.detail("MaxLatency", peer->pingLatencies.max())
.detail("MeanLatency", peer->pingLatencies.mean())
.detail("MedianLatency", peer->pingLatencies.median())
.detail("CheckedPercentile", SERVER_KNOBS->PEER_LATENCY_DEGRADATION_PERCENTILE)
.detail("CheckedPercentileLatency",
peer->pingLatencies.percentile(SERVER_KNOBS->PEER_LATENCY_DEGRADATION_PERCENTILE))
.detail("Count", peer->pingLatencies.getPopulationSize())
.detail("TimeoutCount", peer->timeoutCount);
if (workerInDb) {
for (const auto& [address, peer] : allPeers) {
if (peer->pingLatencies.getPopulationSize() < SERVER_KNOBS->PEER_LATENCY_CHECK_MIN_POPULATION) {
// Ignore peers that don't have enough samples.
// TODO(zhewu): Currently, FlowTransport latency monitor clears ping latency samples on a
// regular
// basis, which may affect the measurement count. Currently,
// WORKER_HEALTH_MONITOR_INTERVAL is much smaller than the ping clearance interval,
// so it may be ok. If this ends to be a problem, we need to consider keep track of
// last ping latencies logged.
continue;
}
req.degradedPeers.push_back(address);
if ((workerInPrimary && addressInDbAndPrimaryDc(address, dbInfo)) ||
(!workerInPrimary && addressInDbAndRemoteDc(address, dbInfo))) {
// Only monitoring the servers that in the primary or remote DC's transaction systems.
// Note that currently we are not monitor storage servers, since lagging in storage servers
// today already can trigger server exclusion by data distributor.
if (peer->pingLatencies.percentile(SERVER_KNOBS->PEER_LATENCY_DEGRADATION_PERCENTILE) >
SERVER_KNOBS->PEER_LATENCY_DEGRADATION_THRESHOLD ||
peer->timeoutCount / (double)(peer->pingLatencies.getPopulationSize()) >
SERVER_KNOBS->PEER_TIMEOUT_PERCENTAGE_DEGRADATION_THRESHOLD) {
// This is a degraded peer.
TraceEvent("HealthMonitorDetectDegradedPeer")
.suppressFor(30)
.detail("Peer", address)
.detail("Elapsed", now() - peer->lastLoggedTime)
.detail("MinLatency", peer->pingLatencies.min())
.detail("MaxLatency", peer->pingLatencies.max())
.detail("MeanLatency", peer->pingLatencies.mean())
.detail("MedianLatency", peer->pingLatencies.median())
.detail("CheckedPercentile", SERVER_KNOBS->PEER_LATENCY_DEGRADATION_PERCENTILE)
.detail(
"CheckedPercentileLatency",
peer->pingLatencies.percentile(SERVER_KNOBS->PEER_LATENCY_DEGRADATION_PERCENTILE))
.detail("Count", peer->pingLatencies.getPopulationSize())
.detail("TimeoutCount", peer->timeoutCount);
req.degradedPeers.push_back(address);
}
}
}
}

3
flow/Knobs.cpp
View File

@ -67,7 +67,8 @@ void FlowKnobs::initialize(Randomize randomize, IsSimulated isSimulated) {
init( HUGE_ARENA_LOGGING_INTERVAL, 5.0 );
init( WRITE_TRACING_ENABLED, true ); if( randomize && BUGGIFY ) WRITE_TRACING_ENABLED = false;
init( TRACING_UDP_LISTENER_PORT, 8889 ); // Only applicable if TracerType is set to a network option.
init( TRACING_SAMPLE_RATE, 1.0 ); // Fraction of traces (not spans) to sample (0 means ignore all traces)
init( TRACING_UDP_LISTENER_PORT, 8889 ); // Only applicable if TracerType is set to a network option
//connectionMonitor
init( CONNECTION_MONITOR_LOOP_TIME, isSimulated ? 0.75 : 1.0 ); if( randomize && BUGGIFY ) CONNECTION_MONITOR_LOOP_TIME = 6.0;

1
flow/Knobs.h
View File

@ -117,6 +117,7 @@ public:
double HUGE_ARENA_LOGGING_INTERVAL;
bool WRITE_TRACING_ENABLED;
double TRACING_SAMPLE_RATE;
int TRACING_UDP_LISTENER_PORT;
// run loop profiling

37
flow/SystemMonitor.cpp
View File

@ -37,6 +37,10 @@ void initializeSystemMonitorMachineState(SystemMonitorMachineState machineState)
::machineState.monitorStartTime = now();
}
double machineStartTime() {
return ::machineState.monitorStartTime;
}
void systemMonitor() {
static StatisticsState statState = StatisticsState();
customSystemMonitor("ProcessMetrics", &statState, true);
@ -157,6 +161,39 @@ SystemStatistics customSystemMonitor(std::string const& eventName, StatisticsSta
.detail("ZoneID", machineState.zoneId)
.detail("MachineID", machineState.machineId);
uint64_t total_memory = 0;
total_memory += FastAllocator<16>::getTotalMemory();
total_memory += FastAllocator<32>::getTotalMemory();
total_memory += FastAllocator<64>::getTotalMemory();
total_memory += FastAllocator<96>::getTotalMemory();
total_memory += FastAllocator<128>::getTotalMemory();
total_memory += FastAllocator<256>::getTotalMemory();
total_memory += FastAllocator<512>::getTotalMemory();
total_memory += FastAllocator<1024>::getTotalMemory();
total_memory += FastAllocator<2048>::getTotalMemory();
total_memory += FastAllocator<4096>::getTotalMemory();
total_memory += FastAllocator<8192>::getTotalMemory();
uint64_t unused_memory = 0;
unused_memory += FastAllocator<16>::getApproximateMemoryUnused();
unused_memory += FastAllocator<32>::getApproximateMemoryUnused();
unused_memory += FastAllocator<64>::getApproximateMemoryUnused();
unused_memory += FastAllocator<96>::getApproximateMemoryUnused();
unused_memory += FastAllocator<128>::getApproximateMemoryUnused();
unused_memory += FastAllocator<256>::getApproximateMemoryUnused();
unused_memory += FastAllocator<512>::getApproximateMemoryUnused();
unused_memory += FastAllocator<1024>::getApproximateMemoryUnused();
unused_memory += FastAllocator<2048>::getApproximateMemoryUnused();
unused_memory += FastAllocator<4096>::getApproximateMemoryUnused();
unused_memory += FastAllocator<8192>::getApproximateMemoryUnused();
if (total_memory > 0) {
TraceEvent("FastAllocMemoryUsage")
.detail("TotalMemory", total_memory)
.detail("UnusedMemory", unused_memory)
.detail("Utilization", format("%f%%", (total_memory - unused_memory) * 100.0 / total_memory));
}
TraceEvent n("NetworkMetrics");
n.detail("Elapsed", currentStats.elapsed)
.detail("CantSleep", netData.countCantSleep - statState->networkState.countCantSleep)

3
flow/SystemMonitor.h
View File

@ -46,6 +46,9 @@ struct SystemMonitorMachineState {
void initializeSystemMonitorMachineState(SystemMonitorMachineState machineState);
// Returns the machine start time. 0 if the system monitor is not initialized.
double machineStartTime();
struct NetworkData {
int64_t bytesSent;
int64_t countPacketsReceived;

4
flow/Tracing.actor.cpp
View File

@ -373,7 +373,7 @@ void openTracer(TracerType type) {
ITracer::~ITracer() {}
Span& Span::operator=(Span&& o) {
if (begin > 0.0) {
if (begin > 0.0 && context.second() > 0) {
end = g_network->now();
g_tracer->trace(*this);
}
@ -388,7 +388,7 @@ Span& Span::operator=(Span&& o) {
}
Span::~Span() {
if (begin > 0.0) {
if (begin > 0.0 && context.second() > 0) {
end = g_network->now();
g_tracer->trace(*this);
}

19
flow/Tracing.h
View File

@ -37,11 +37,19 @@ struct Span {
Span(SpanID context, Location location, std::initializer_list<SpanID> const& parents = {})
: context(context), begin(g_network->now()), location(location), parents(arena, parents.begin(), parents.end()) {
if (parents.size() > 0) {
this->context = SpanID((*parents.begin()).first(), context.second());
// If the parents' token is 0 (meaning the trace should not be
// recorded), set the child token to 0 as well. Otherwise, use the
// existing (likely randomly generated) value.
uint64_t traceId = (*parents.begin()).second() > 0 ? context.second() : 0;
this->context = SpanID((*parents.begin()).first(), traceId);
}
}
Span(Location location, std::initializer_list<SpanID> const& parents = {})
: Span(deterministicRandom()->randomUniqueID(), location, parents) {}
Span(Location location, std::initializer_list<SpanID> const& parents = {}) {
uint64_t tokenId = deterministicRandom()->random01() < FLOW_KNOBS->TRACING_SAMPLE_RATE
? deterministicRandom()->randomUInt64()
: 0;
Span(UID(deterministicRandom()->randomUInt64(), tokenId), location, parents);
}
Span(Location location, SpanID context) : Span(location, { context }) {}
Span(const Span&) = delete;
Span(Span&& o) {
@ -70,12 +78,13 @@ struct Span {
void addParent(SpanID span) {
if (parents.size() == 0) {
uint64_t traceId = (*parents.begin()).second() > 0 ? context.second() : 0;
// Use first parent to set trace ID. This is non-ideal for spans
// with multiple parents, because the trace ID will associate the
// span with only one trace. A workaround is to look at the parent
// relationships instead of the trace ID. Another option in the
// future is to keep a list of trace IDs.
context = SpanID(span.first(), context.second());
context = SpanID(span.first(), traceId);
}
parents.push_back(arena, span);
}
@ -112,7 +121,7 @@ void openTracer(TracerType type);
template <class T>
struct SpannedDeque : Deque<T> {
Span span;
explicit SpannedDeque(Location loc) : span(deterministicRandom()->randomUniqueID(), loc) {}
explicit SpannedDeque(Location loc) : span(loc) {}
SpannedDeque(SpannedDeque&& other) : Deque<T>(std::move(other)), span(std::move(other.span)) {}
SpannedDeque(SpannedDeque const&) = delete;
SpannedDeque& operator=(SpannedDeque const&) = delete;

1
flow/error_definitions.h
View File

@ -78,6 +78,7 @@ ERROR( wrong_connection_file, 1054, "Connection file mismatch")
ERROR( version_already_compacted, 1055, "The requested changes have been compacted away")
ERROR( local_config_changed, 1056, "Local configuration file has changed. Restart and apply these changes" )
ERROR( failed_to_reach_quorum, 1057, "Failed to reach quorum from configuration database nodes. Retry sending these requests" )
ERROR( wrong_format_version, 1058, "Format version not recognize." )
ERROR( broken_promise, 1100, "Broken promise" )
ERROR( operation_cancelled, 1101, "Asynchronous operation cancelled" )

2
flow/flow.cpp
View File

@ -51,9 +51,7 @@ LineageReference getCurrentLineage() {
}
return *currentLineage;
}
#endif
#ifdef ENABLE_SAMPLING
void sample(LineageReference* lineagePtr);
void replaceLineage(LineageReference* lineage) {

5
flow/flow.h
View File

@ -545,6 +545,11 @@ public:
LineageReference() : Reference<ActorLineage>(nullptr), actorName_(""), allocated_(false) {}
explicit LineageReference(ActorLineage* ptr) : Reference<ActorLineage>(ptr), actorName_(""), allocated_(false) {}
LineageReference(const LineageReference& r) : Reference<ActorLineage>(r), actorName_(""), allocated_(false) {}
LineageReference(LineageReference&& r)
: Reference<ActorLineage>(std::forward<LineageReference>(r)), actorName_(r.actorName_), allocated_(r.allocated_) {
r.actorName_ = "";
r.allocated_ = false;
}
void setActorName(const char* name) { actorName_ = name; }
const char* actorName() { return actorName_; }

32
packaging/docker/ycsb/Dockerfile Normal file
View File

@ -0,0 +1,32 @@
FROM openjdk:17-slim AS RUN
WORKDIR /tmp
RUN apt-get update && \
apt-get install -y curl zip && \
curl -Ls https://amazon-eks.s3.amazonaws.com/1.19.6/2021-01-05/bin/linux/amd64/kubectl -o kubectl && \
echo "08ff68159bbcb844455167abb1d0de75bbfe5ae1b051f81ab060a1988027868a kubectl" > kubectl.txt && \
sha256sum -c kubectl.txt && \
mv kubectl /usr/local/bin/kubectl && \
chmod 755 /usr/local/bin/kubectl && \
curl https://awscli.amazonaws.com/awscli-exe-linux-x86_64-2.2.43.zip -o "awscliv2.zip" && \
echo "9a8b3c4e7f72bbcc55e341dce3af42479f2730c225d6d265ee6f9162cfdebdfd awscliv2.zip" > awscliv2.txt && \
sha256sum -c awscliv2.txt && \
unzip -qq awscliv2.zip && \
./aws/install && \
rm -rf /tmp/*
ADD YCSB /YCSB
WORKDIR /YCSB
ENV FDB_NETWORK_OPTION_EXTERNAL_CLIENT_DIRECTORY=/var/dynamic-conf/lib/multiversion/
ENV FDB_NETWORK_OPTION_TRACE_ENABLE=/var/log/fdb-trace-logs
ENV LD_LIBRARY_PATH=/var/dynamic-conf/lib/
ENV BUCKET=backup-112664522426-us-west-2
# TODO: Log4J complains that it's eating the HTracer logs. Even without it, we get per-operation
# time series graphs of throughput, median, 90, 99, 99.9 and 99.99 (in usec).
COPY ycsb/run_ycsb.sh /usr/local/bin/run_ycsb.sh
RUN mkdir -p /var/log/fdb-trace-logs && \
chmod +x /usr/local/bin/run_ycsb.sh
CMD ["run_ycsb.sh"]

18
packaging/docker/ycsb/run_ycsb.sh Normal file
View File

@ -0,0 +1,18 @@
#!/usr/bin/env bash
set -Eeuo pipefail
namespace=$(cat /var/run/secrets/kubernetes.io/serviceaccount/namespace)
echo "WAITING FOR ALL PODS TO COME UP"
while [[ $(kubectl get pods -n ${namespace} -l name=ycsb,run=${RUN_ID} --field-selector=status.phase=Running | grep -cv NAME) -lt ${NUM_PODS} ]]; do
sleep 0.1
done
echo "ALL PODS ARE UP"
echo "RUNNING YCSB"
./bin/ycsb.sh ${MODE} foundationdb -s -P workloads/${WORKLOAD} ${YCSB_ARGS}
echo "YCSB FINISHED"
echo "COPYING HISTOGRAMS TO S3"
aws s3 sync --sse aws:kms --exclude "*" --include "histogram.*" /tmp s3://${BUCKET}/ycsb_histgorams/${namespace}/${POD_NAME}
echo "COPYING HISTOGRAMS TO S3 FINISHED"