diff --git a/documentation/sphinx/source/request-tracing.rst b/documentation/sphinx/source/request-tracing.rst index 335952fb67..3fc719bf7e 100644 --- a/documentation/sphinx/source/request-tracing.rst +++ b/documentation/sphinx/source/request-tracing.rst @@ -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 ` 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. diff --git a/fdbclient/ActorLineageProfiler.cpp b/fdbclient/ActorLineageProfiler.cpp index fd79c51b88..ff097af198 100644 --- a/fdbclient/ActorLineageProfiler.cpp +++ b/fdbclient/ActorLineageProfiler.cpp @@ -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())]() { diff --git a/fdbclient/ClusterInterface.h b/fdbclient/ClusterInterface.h index 8e8c2abc26..e733731028 100644 --- a/fdbclient/ClusterInterface.h +++ b/fdbclient/ClusterInterface.h @@ -36,6 +36,8 @@ struct ClusterInterface { RequestStream> ping; RequestStream getClientWorkers; RequestStream forceRecovery; + RequestStream moveShard; + RequestStream 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 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 addresses; + ReplyPromise reply; + + MoveShardRequest() {} + MoveShardRequest(KeyRange shard, std::vector addresses) + : shard{ std::move(shard) }, addresses{ std::move(addresses) } {} + + template + 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 reply; + + RepairSystemDataRequest() {} + + template + void serialize(Ar& ar) { + serializer(ar, reply); + } +}; #endif diff --git a/fdbclient/DatabaseContext.h b/fdbclient/DatabaseContext.h index 71834e5942..be444b39bb 100644 --- a/fdbclient/DatabaseContext.h +++ b/fdbclient/DatabaseContext.h @@ -382,6 +382,8 @@ public: Counter transactionsProcessBehind; Counter transactionsThrottled; Counter transactionsExpensiveClearCostEstCount; + Counter transactionGrvFullBatches; + Counter transactionGrvTimedOutBatches; ContinuousSample latencies, readLatencies, commitLatencies, GRVLatencies, mutationsPerCommit, bytesPerCommit; @@ -392,6 +394,7 @@ public: int snapshotRywEnabled; int transactionTracingEnabled; + double verifyCausalReadsProp = 0.0; Future logger; Future throttleExpirer; diff --git a/fdbclient/FluentDSampleIngestor.cpp b/fdbclient/FluentDSampleIngestor.cpp index 89e16f1615..3ecfc40d30 100644 --- a/fdbclient/FluentDSampleIngestor.cpp +++ b/fdbclient/FluentDSampleIngestor.cpp @@ -73,13 +73,14 @@ class SampleSender : public std::enable_shared_from_this 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 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() { diff --git a/fdbclient/GrvProxyInterface.h b/fdbclient/GrvProxyInterface.h index 85ad4d16bc..d4b3b78bcb 100644 --- a/fdbclient/GrvProxyInterface.h +++ b/fdbclient/GrvProxyInterface.h @@ -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. diff --git a/fdbclient/NativeAPI.actor.cpp b/fdbclient/NativeAPI.actor.cpp index b7c88d3479..6359cd5e30 100644 --- a/fdbclient/NativeAPI.actor.cpp +++ b/fdbclient/NativeAPI.actor.cpp @@ -447,7 +447,8 @@ ACTOR Future 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 clientStatusUpdateActor(DatabaseContext* cx) { } } -ACTOR static Future monitorProxiesChange(Reference const> clientDBInfo, +ACTOR Future assertFailure(GrvProxyInterface remote, Future> reply) { + try { + ErrorOr 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 attemptGRVFromOldProxies(std::vector oldProxies, + std::vector newProxies) { + Span span(deterministicRandom()->randomUniqueID(), "VerifyCausalReadRisky"_loc); + std::vector> 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& 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(); + replies.push_back(assertFailure(i, i.getConsistentReadVersion.tryGetReply(req))); + } + return waitForAll(replies); +} + +ACTOR static Future monitorProxiesChange(DatabaseContext* cx, + Reference const> clientDBInfo, AsyncTrigger* triggerVar) { state std::vector curCommitProxies; state std::vector 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(ReferenceSHARD_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(specialKeys.begin, specialKeys.end, /* test */ false)) { dbId = deterministicRandom()->randomUniqueID(); connected = (clientInfo->get().commitProxies.size() && clientInfo->get().grvProxies.size()) @@ -1210,7 +1276,7 @@ DatabaseContext::DatabaseContext(ReferenceSHARD_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, OptionalrandomUInt64(); + 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 readVersionBatcher(DatabaseContext* cx, state Future timeout; state Optional debugID; state bool send_batch; + state Reference batchSizeDist = Histogram::getHistogram(LiteralStringRef("GrvBatcher"), + LiteralStringRef("ClientGrvBatchSize"), + Histogram::Unit::countLinear, + 0, + CLIENT_KNOBS->MAX_BATCH_SIZE * 2); + state Reference batchIntervalDist = + Histogram::getHistogram(LiteralStringRef("GrvBatcher"), + LiteralStringRef("ClientGrvBatchInterval"), + Histogram::Unit::microseconds, + 0, + CLIENT_KNOBS->GRV_BATCH_TIMEOUT * 1000000 * 2); + state Reference grvReplyLatencyDist = Histogram::getHistogram( + LiteralStringRef("GrvBatcher"), LiteralStringRef("ClientGrvReplyLatency"), Histogram::Unit::microseconds); + state double lastRequestTime = now(); state TransactionTagMap tags; @@ -5693,22 +5780,34 @@ ACTOR Future 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 GRVReply; requests.push_back(GRVReply); @@ -5859,7 +5958,7 @@ Future 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, 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>> locations = wait(getKeyRangeLocations(cx, diff --git a/fdbclient/PaxosConfigTransaction.actor.cpp b/fdbclient/PaxosConfigTransaction.actor.cpp index 6baa570dcd..1b22cf0d5a 100644 --- a/fdbclient/PaxosConfigTransaction.actor.cpp +++ b/fdbclient/PaxosConfigTransaction.actor.cpp @@ -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 { diff --git a/fdbclient/ServerKnobs.cpp b/fdbclient/ServerKnobs.cpp index b4c0ef707c..0dd3084ad3 100644 --- a/fdbclient/ServerKnobs.cpp +++ b/fdbclient/ServerKnobs.cpp @@ -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 ); diff --git a/fdbclient/ServerKnobs.h b/fdbclient/ServerKnobs.h index b2cdd015c9..21318ca94c 100644 --- a/fdbclient/ServerKnobs.h +++ b/fdbclient/ServerKnobs.h @@ -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) diff --git a/fdbclient/vexillographer/fdb.options b/fdbclient/vexillographer/fdb.options index 6eede67882..996fae6dc7 100644 --- a/fdbclient/vexillographer/fdb.options +++ b/fdbclient/vexillographer/fdb.options @@ -200,6 +200,8 @@ description is not currently required but encouraged. defaultFor="1100"/>