diff --git a/fdbclient/BackupAgentBase.actor.cpp b/fdbclient/BackupAgentBase.actor.cpp index 0ddb9b9ee9..dcce6d05ff 100644 --- a/fdbclient/BackupAgentBase.actor.cpp +++ b/fdbclient/BackupAgentBase.actor.cpp @@ -567,7 +567,7 @@ ACTOR Future readCommitted(Database cx, tr.setOption(FDBTransactionOptions::LOCK_AWARE); if (CLIENT_KNOBS->ENABLE_REPLICA_CONSISTENCY_CHECK_ON_BACKUP_READS) { tr.setOption(FDBTransactionOptions::ENABLE_REPLICA_CONSISTENCY_CHECK); - int64_t requiredReplicas = CLIENT_KNOBS->CONSISTENCY_CHECK_REQUIRED_REPLICAS; + int64_t requiredReplicas = CLIENT_KNOBS->BACKUP_CONSISTENCY_CHECK_REQUIRED_REPLICAS; tr.setOption(FDBTransactionOptions::CONSISTENCY_CHECK_REQUIRED_REPLICAS, StringRef((uint8_t*)&requiredReplicas, sizeof(int64_t))); } diff --git a/fdbclient/ClientKnobs.cpp b/fdbclient/ClientKnobs.cpp index 30ff4ef134..99873f482e 100644 --- a/fdbclient/ClientKnobs.cpp +++ b/fdbclient/ClientKnobs.cpp @@ -200,7 +200,7 @@ void ClientKnobs::initialize(Randomize randomize) { init( BLOB_GRANULE_RESTORE_CHECK_INTERVAL, 10 ); init( BACKUP_CONTAINER_LOCAL_ALLOW_RELATIVE_PATH, false ); init( ENABLE_REPLICA_CONSISTENCY_CHECK_ON_BACKUP_READS, false ); if( randomize && BUGGIFY ) { ENABLE_REPLICA_CONSISTENCY_CHECK_ON_BACKUP_READS = true; } - init( CONSISTENCY_CHECK_REQUIRED_REPLICAS, -2 ); // Do consistency check based on all available storage replicas + init( BACKUP_CONSISTENCY_CHECK_REQUIRED_REPLICAS, -2 ); // Do consistency check based on all available storage replicas init( BULKLOAD_JOB_HISTORY_COUNT_MAX, 10 ); if (randomize && BUGGIFY) BULKLOAD_JOB_HISTORY_COUNT_MAX = deterministicRandom()->randomInt(1, 10); init( BULKLOAD_VERBOSE_LEVEL, 10 ); init( S3CLIENT_VERBOSE_LEVEL, 10 ); diff --git a/fdbclient/ServerKnobs.cpp b/fdbclient/ServerKnobs.cpp index 81cba4e800..043a5205ee 100644 --- a/fdbclient/ServerKnobs.cpp +++ b/fdbclient/ServerKnobs.cpp @@ -177,8 +177,8 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi init( FETCH_KEYS_THROTTLE_PRIORITY_THRESHOLD, 0 ); if( randomize && BUGGIFY ) FETCH_KEYS_THROTTLE_PRIORITY_THRESHOLD = 700; init( ENABLE_REPLICA_CONSISTENCY_CHECK_ON_DATA_MOVEMENT, false ); ENABLE_REPLICA_CONSISTENCY_CHECK_ON_DATA_MOVEMENT = isSimulated; - init( CONSISTENCY_CHECK_REQUIRED_REPLICAS, 1 ); - + init( DATAMOVE_CONSISTENCY_CHECK_REQUIRED_REPLICAS, 1 ); + init( ENABLE_CONSERVATIVE_RELOCATION_WHEN_REPLICA_CONSISTENCY_CHECK, false ); if (isSimulated) ENABLE_CONSERVATIVE_RELOCATION_WHEN_REPLICA_CONSISTENCY_CHECK = deterministicRandom()->coinflip(); init( PROBABILITY_TEAM_REDUNDANT_DATAMOVE_CHOOSE_TRUE_BEST_DEST, 0.0 ); if (isSimulated) PROBABILITY_TEAM_REDUNDANT_DATAMOVE_CHOOSE_TRUE_BEST_DEST = deterministicRandom()->random01(); init( PROBABILITY_TEAM_UNHEALTHY_DATAMOVE_CHOOSE_TRUE_BEST_DEST, 0.0 ); if (isSimulated) PROBABILITY_TEAM_UNHEALTHY_DATAMOVE_CHOOSE_TRUE_BEST_DEST = deterministicRandom()->random01(); diff --git a/fdbclient/include/fdbclient/ClientKnobs.h b/fdbclient/include/fdbclient/ClientKnobs.h index 7f88fe6fbf..d0da346df7 100644 --- a/fdbclient/include/fdbclient/ClientKnobs.h +++ b/fdbclient/include/fdbclient/ClientKnobs.h @@ -200,7 +200,7 @@ public: int BLOB_GRANULE_RESTORE_CHECK_INTERVAL; bool BACKUP_CONTAINER_LOCAL_ALLOW_RELATIVE_PATH; bool ENABLE_REPLICA_CONSISTENCY_CHECK_ON_BACKUP_READS; - int CONSISTENCY_CHECK_REQUIRED_REPLICAS; + int BACKUP_CONSISTENCY_CHECK_REQUIRED_REPLICAS; int BULKLOAD_JOB_HISTORY_COUNT_MAX; // the max number of bulk load job history to keep. The oldest job history will // be removed when the count exceeds this value. Set to 0 to disable history. // Do not set the value to a large number, e.g. <= 10. diff --git a/fdbclient/include/fdbclient/ServerKnobs.h b/fdbclient/include/fdbclient/ServerKnobs.h index d795b850fa..b46d1ee9e1 100644 --- a/fdbclient/include/fdbclient/ServerKnobs.h +++ b/fdbclient/include/fdbclient/ServerKnobs.h @@ -199,8 +199,11 @@ public: // STORAGE_FETCH_KEYS_RATE_LIMIT. int FETCH_KEYS_THROTTLE_PRIORITY_THRESHOLD; - bool ENABLE_REPLICA_CONSISTENCY_CHECK_ON_DATA_MOVEMENT; - int CONSISTENCY_CHECK_REQUIRED_REPLICAS; + bool ENABLE_REPLICA_CONSISTENCY_CHECK_ON_DATA_MOVEMENT; // Enable to check replica consistency on data movement + int DATAMOVE_CONSISTENCY_CHECK_REQUIRED_REPLICAS; // The number of extra replicas to check for replica consistency + // on data movement read range requests by fetchKeys + bool ENABLE_CONSERVATIVE_RELOCATION_WHEN_REPLICA_CONSISTENCY_CHECK; // Enable to slow down relocation when replica + // consistency check on data movement is enabled // Probability that a team redundant data move set TrueBest when get destination team double PROBABILITY_TEAM_REDUNDANT_DATAMOVE_CHOOSE_TRUE_BEST_DEST; diff --git a/fdbrpc/include/fdbrpc/LoadBalance.actor.h b/fdbrpc/include/fdbrpc/LoadBalance.actor.h index 45e4e6f640..d3f1b1dcee 100644 --- a/fdbrpc/include/fdbrpc/LoadBalance.actor.h +++ b/fdbrpc/include/fdbrpc/LoadBalance.actor.h @@ -311,31 +311,55 @@ Future replicaComparison(Req req, } } else if (!srcLB.present() || !srcLB.get().error.present()) { // Verify that the other SS servers in the team have the same data. - state std::vector>>> restOfTeamFutures; - restOfTeamFutures.reserve(ssTeam->size() - 1); + std::vector candidates; + // candidates includes all healthy SS endpoints in the team except the one we already + // have a response from for (int i = 0; i < ssTeam->size(); i++) { RequestStream const* si = &ssTeam->get(i, channel); - if (si->getEndpoint().token.first() != - srcEndpointId) { // don't re-request to SS we already have a response from - if (!IFailureMonitor::failureMonitor().getState(si->getEndpoint()).failed) { - resetReply(req); - restOfTeamFutures.push_back(( - requiredReplicas == BEST_EFFORT - ? timeout(si->tryGetReply(req), FLOW_KNOBS->LOAD_BALANCE_FETCH_REPLICA_TIMEOUT) - : timeout(errorOr(si->getReply(req)), FLOW_KNOBS->LOAD_BALANCE_FETCH_REPLICA_TIMEOUT))); - } else if (requiredReplicas == ALL_REPLICAS) { - TraceEvent(SevWarnAlways, "UnreachableStorageServer") - .detail("SSID", ssTeam->getInterface(i).id()); - throw unreachable_storage_replica(); - } + if (si->getEndpoint().token.first() == srcEndpointId) { + // Don't re-request to SS we already have a response from + continue; + } + if (!IFailureMonitor::failureMonitor().getState(si->getEndpoint()).failed) { + candidates.push_back(si->getEndpoint().token.first()); + } else if (requiredReplicas == ALL_REPLICAS) { + TraceEvent(SevWarnAlways, "UnreachableStorageServer").detail("SSID", ssTeam->getInterface(i).id()); + throw unreachable_storage_replica(); } } - - if (requiredReplicas == BEST_EFFORT || requiredReplicas == ALL_REPLICAS) { - wait(waitForAllReady(restOfTeamFutures)); - } else { - wait(waitForQuorumReplies(&restOfTeamFutures, requiredReplicas)); + int numReplicaToRead = candidates.size(); + if (requiredReplicas != BEST_EFFORT && requiredReplicas != ALL_REPLICAS) { + ASSERT(requiredReplicas > 0); + numReplicaToRead = std::min((int)candidates.size(), requiredReplicas); + if (FLOW_KNOBS->ENABLE_WARNING_READ_CONSISTENCY_CHECK_NOT_ENOUGH_REPLICA && + candidates.size() < requiredReplicas) { + TraceEvent(SevWarn, "ReplicaConsistencyCheckNotEnoughReplica") + .suppressFor(5.0) + .detail("RequiredReplicas", requiredReplicas) + .detail("AvailableReplicas", candidates.size()); + } } + state std::vector>>> restOfTeamFutures; + restOfTeamFutures.reserve(numReplicaToRead); + // Randomly select numReplicaToRead SSes to read from + deterministicRandom()->randomShuffle(candidates); + candidates.erase(candidates.begin() + numReplicaToRead, candidates.end()); + std::unordered_set ssToRead(candidates.begin(), candidates.end()); + + for (int i = 0; i < ssTeam->size(); i++) { + RequestStream const* si = &ssTeam->get(i, channel); + if (!ssToRead.contains(si->getEndpoint().token.first())) { + // Only send requests to the SSes that we randomly selected + continue; + } + resetReply(req); + restOfTeamFutures.push_back( + (requiredReplicas == BEST_EFFORT + ? timeout(si->tryGetReply(req), FLOW_KNOBS->LOAD_BALANCE_FETCH_REPLICA_TIMEOUT) + : timeout(errorOr(si->getReply(req)), FLOW_KNOBS->LOAD_BALANCE_FETCH_REPLICA_TIMEOUT))); + } + + wait(waitForAllReady(restOfTeamFutures)); int numError = 0; int numMismatch = 0; @@ -480,16 +504,31 @@ struct RequestData : NonCopyable { int requiredReplicas) { if (model && (compareReplicas || FLOW_KNOBS->ENABLE_REPLICA_CONSISTENCY_CHECK_ON_READS)) { ASSERT(requestStream != nullptr); - int requiredReplicaCount = - compareReplicas ? requiredReplicas : FLOW_KNOBS->CONSISTENCY_CHECK_REQUIRED_REPLICAS; + if (compareReplicas) { + // In case compareReplicas == true, we may read extra requiredReplicas replica. + // The value of compareReplicas is decided by the caller and the knobs. + // If the caller is fetchKeys, when ENABLE_REPLICA_CONSISTENCY_CHECK_ON_DATA_MOVEMENT is on, + // the value is DATAMOVE_CONSISTENCY_CHECK_REQUIRED_REPLICAS. + // If the caller is backup agents, when ENABLE_REPLICA_CONSISTENCY_CHECK_ON_BACKUP_READS is on, + // the value is BACKUP_CONSISTENCY_CHECK_REQUIRED_REPLICAS. + // Otherwise, the value is 0. + return replicaComparison(request, + response, + requestStream->getEndpoint().token.first(), + alternatives, + channel, + requiredReplicas); + } + // In case ENABLE_REPLICA_CONSISTENCY_CHECK_ON_READS is on, we read extra + // READ_CONSISTENCY_CHECK_REQUIRED_REPLICAS replica and conduct consistency + // check among replica for any read request. return replicaComparison(request, response, requestStream->getEndpoint().token.first(), alternatives, channel, - requiredReplicaCount); + FLOW_KNOBS->READ_CONSISTENCY_CHECK_REQUIRED_REPLICAS); } - return Void(); } diff --git a/fdbserver/DDRelocationQueue.actor.cpp b/fdbserver/DDRelocationQueue.actor.cpp index a860b6d595..cda28312b0 100644 --- a/fdbserver/DDRelocationQueue.actor.cpp +++ b/fdbserver/DDRelocationQueue.actor.cpp @@ -428,25 +428,55 @@ std::string Busyness::toString() { return result; } +double adjustRelocationParallelismForSrc(double srcParallelism) { + double res = srcParallelism; + if (SERVER_KNOBS->ENABLE_CONSERVATIVE_RELOCATION_WHEN_REPLICA_CONSISTENCY_CHECK && + SERVER_KNOBS->ENABLE_REPLICA_CONSISTENCY_CHECK_ON_DATA_MOVEMENT && + srcParallelism >= 1.0 + SERVER_KNOBS->DATAMOVE_CONSISTENCY_CHECK_REQUIRED_REPLICAS) { + // DATAMOVE_CONSISTENCY_CHECK_REQUIRED_REPLICAS is the number of extra replicas that the destination + // servers will read from the source team. + res = res / (1.0 + SERVER_KNOBS->DATAMOVE_CONSISTENCY_CHECK_REQUIRED_REPLICAS); + } + ASSERT(res > 0); + return res; +} + // find the "workFactor" for this, were it launched now int getSrcWorkFactor(RelocateData const& relocation, int singleRegionTeamSize) { + // RELOCATION_PARALLELISM_PER_SOURCE_SERVER is the number of concurrent replications that can be launched on a + // single storage server at a time, given the team size is 1 --- only this storage server is available to serve + // fetchKey read requests from the dest team. + // The real parallelism is adjusted by the number of source servers of a source team that can serve + // fetchKey requests. + // When ENABLE_REPLICA_CONSISTENCY_CHECK_ON_DATA_MOVEMENT is enabled, the fetchKeys on + // destination servers will read from DATAMOVE_CONSISTENCY_CHECK_REQUIRED_REPLICAS + 1 replicas from the source team + // (suppose the team size is large enough). As a result it is possible that the source team can be overloaded by the + // fetchKey read requests. This is especially true when the shard split data movements are launched. So, we + // introduce ENABLE_CONSERVATIVE_RELOCATION_WHEN_REPLICA_CONSISTENCY_CHECK knob to adjust the relocation parallelism + // accordingly. The adjustment is to reduce the relocation parallelism by a factor of + // (1 + DATAMOVE_CONSISTENCY_CHECK_REQUIRED_REPLICAS). if (relocation.bulkLoadTask.present()) return 0; else if (relocation.healthPriority == SERVER_KNOBS->PRIORITY_TEAM_1_LEFT || relocation.healthPriority == SERVER_KNOBS->PRIORITY_TEAM_0_LEFT) - return WORK_FULL_UTILIZATION / SERVER_KNOBS->RELOCATION_PARALLELISM_PER_SOURCE_SERVER; + return WORK_FULL_UTILIZATION / + adjustRelocationParallelismForSrc(SERVER_KNOBS->RELOCATION_PARALLELISM_PER_SOURCE_SERVER); else if (relocation.healthPriority == SERVER_KNOBS->PRIORITY_TEAM_2_LEFT) - return WORK_FULL_UTILIZATION / 2 / SERVER_KNOBS->RELOCATION_PARALLELISM_PER_SOURCE_SERVER; + return WORK_FULL_UTILIZATION / + adjustRelocationParallelismForSrc(2 * SERVER_KNOBS->RELOCATION_PARALLELISM_PER_SOURCE_SERVER); else if (relocation.healthPriority == SERVER_KNOBS->PRIORITY_PERPETUAL_STORAGE_WIGGLE) // we want to set PRIORITY_PERPETUAL_STORAGE_WIGGLE to a reasonably large value // to make this parallelism take effect - return WORK_FULL_UTILIZATION / SERVER_KNOBS->WIGGLING_RELOCATION_PARALLELISM_PER_SOURCE_SERVER; + return WORK_FULL_UTILIZATION / + adjustRelocationParallelismForSrc(SERVER_KNOBS->WIGGLING_RELOCATION_PARALLELISM_PER_SOURCE_SERVER); else if (relocation.priority == SERVER_KNOBS->PRIORITY_MERGE_SHARD) - return WORK_FULL_UTILIZATION / SERVER_KNOBS->MERGE_RELOCATION_PARALLELISM_PER_TEAM; + return WORK_FULL_UTILIZATION / + adjustRelocationParallelismForSrc(SERVER_KNOBS->MERGE_RELOCATION_PARALLELISM_PER_TEAM); else { // for now we assume that any message at a lower priority can best be assumed to have a full team left for // work - - return WORK_FULL_UTILIZATION / singleRegionTeamSize / SERVER_KNOBS->RELOCATION_PARALLELISM_PER_SOURCE_SERVER; + return WORK_FULL_UTILIZATION / + adjustRelocationParallelismForSrc(singleRegionTeamSize * + SERVER_KNOBS->RELOCATION_PARALLELISM_PER_SOURCE_SERVER); } } diff --git a/fdbserver/StorageMetrics.actor.cpp b/fdbserver/StorageMetrics.actor.cpp index 8e8d3cf99c..79bb9cb068 100644 --- a/fdbserver/StorageMetrics.actor.cpp +++ b/fdbserver/StorageMetrics.actor.cpp @@ -28,7 +28,7 @@ CommonStorageCounters::CommonStorageCounters(const std::string& name, : cc(name, id), finishedQueries("FinishedQueries", cc), bytesQueried("BytesQueried", cc), bytesFetched("BytesFetched", cc), bytesInput("BytesInput", cc), mutationBytes("MutationBytes", cc), kvFetched("KVFetched", cc), mutations("Mutations", cc), setMutations("SetMutations", cc), - clearRangeMutations("ClearRangeMutations", cc) { + clearRangeMutations("ClearRangeMutations", cc), fetchKeyErrors("FetchKeyErrors", cc) { if (metrics) { specialCounter(cc, "BytesStored", [metrics]() { return metrics->byteSample.getEstimate(allKeys); }); specialCounter(cc, "BytesReadSampleCount", [metrics]() { return metrics->bytesReadSample.queue.size(); }); diff --git a/fdbserver/include/fdbserver/StorageMetrics.actor.h b/fdbserver/include/fdbserver/StorageMetrics.actor.h index 09cd4d38c8..db3e2b9116 100644 --- a/fdbserver/include/fdbserver/StorageMetrics.actor.h +++ b/fdbserver/include/fdbserver/StorageMetrics.actor.h @@ -206,6 +206,9 @@ struct CommonStorageCounters { // The number of key-value pairs fetched by fetchKeys() Counter kvFetched; + // The number of fetchKeys errors + Counter fetchKeyErrors; + // name and id are the inputs to CounterCollection initialization. If metrics provided, the caller should guarantee // the lifetime of metrics is longer than this counter CommonStorageCounters(const std::string& name, diff --git a/fdbserver/storageserver.actor.cpp b/fdbserver/storageserver.actor.cpp index 7c99623643..a36c73524f 100644 --- a/fdbserver/storageserver.actor.cpp +++ b/fdbserver/storageserver.actor.cpp @@ -1546,15 +1546,15 @@ public: // expensive. Counter pTreeClearSplits; - LatencySample readLatencySample; - LatencySample readKeyLatencySample; - LatencySample readValueLatencySample; - LatencySample readRangeLatencySample; - LatencySample readVersionWaitSample; - LatencySample readQueueWaitSample; - LatencySample kvReadRangeLatencySample; - LatencySample updateLatencySample; - LatencySample updateEncryptionLatencySample; + std::unique_ptr readLatencySample; + std::unique_ptr readKeyLatencySample; + std::unique_ptr readValueLatencySample; + std::unique_ptr readRangeLatencySample; + std::unique_ptr readVersionWaitSample; + std::unique_ptr readQueueWaitSample; + std::unique_ptr kvReadRangeLatencySample; + std::unique_ptr updateLatencySample; + std::unique_ptr updateEncryptionLatencySample; LatencyBands readLatencyBands; std::unique_ptr mappedRangeSample; // Samples getMappedRange latency std::unique_ptr mappedRangeRemoteSample; // Samples getMappedRange remote subquery latency @@ -1594,42 +1594,43 @@ public: changeServerKeysAssigned("ChangeServerKeysAssigned", cc), changeServerKeysUnassigned("ChangeServerKeysUnassigned", cc), kvClearRangesInFetchKeys("KvClearRangesInFetchKeys", cc), - readLatencySample("ReadLatencyMetrics", - self->thisServerID, - SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL, - SERVER_KNOBS->LATENCY_SKETCH_ACCURACY), - readKeyLatencySample("GetKeyMetrics", - self->thisServerID, - SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL, - SERVER_KNOBS->LATENCY_SKETCH_ACCURACY), - readValueLatencySample("GetValueMetrics", - self->thisServerID, - SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL, - SERVER_KNOBS->LATENCY_SKETCH_ACCURACY), - readRangeLatencySample("GetRangeMetrics", - self->thisServerID, - SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL, - SERVER_KNOBS->LATENCY_SKETCH_ACCURACY), - readVersionWaitSample("ReadVersionWaitMetrics", - self->thisServerID, - SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL, - SERVER_KNOBS->LATENCY_SKETCH_ACCURACY), - readQueueWaitSample("ReadQueueWaitMetrics", - self->thisServerID, - SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL, - SERVER_KNOBS->LATENCY_SKETCH_ACCURACY), - kvReadRangeLatencySample("KVGetRangeMetrics", - self->thisServerID, - SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL, - SERVER_KNOBS->LATENCY_SKETCH_ACCURACY), - updateLatencySample("UpdateLatencyMetrics", - self->thisServerID, - SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL, - SERVER_KNOBS->LATENCY_SKETCH_ACCURACY), - updateEncryptionLatencySample("UpdateEncryptionLatencyMetrics", - self->thisServerID, - SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL, - SERVER_KNOBS->LATENCY_SKETCH_ACCURACY), + readLatencySample(std::make_unique("ReadLatencyMetrics", + self->thisServerID, + SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL, + SERVER_KNOBS->LATENCY_SKETCH_ACCURACY)), + readKeyLatencySample(std::make_unique("GetKeyMetrics", + self->thisServerID, + SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL, + SERVER_KNOBS->LATENCY_SKETCH_ACCURACY)), + readValueLatencySample(std::make_unique("GetValueMetrics", + self->thisServerID, + SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL, + SERVER_KNOBS->LATENCY_SKETCH_ACCURACY)), + readRangeLatencySample(std::make_unique("GetRangeMetrics", + self->thisServerID, + SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL, + SERVER_KNOBS->LATENCY_SKETCH_ACCURACY)), + readVersionWaitSample(std::make_unique("ReadVersionWaitMetrics", + self->thisServerID, + SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL, + SERVER_KNOBS->LATENCY_SKETCH_ACCURACY)), + readQueueWaitSample(std::make_unique("ReadQueueWaitMetrics", + self->thisServerID, + SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL, + SERVER_KNOBS->LATENCY_SKETCH_ACCURACY)), + kvReadRangeLatencySample(std::make_unique("KVGetRangeMetrics", + self->thisServerID, + SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL, + SERVER_KNOBS->LATENCY_SKETCH_ACCURACY)), + updateLatencySample(std::make_unique("UpdateLatencyMetrics", + self->thisServerID, + SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL, + SERVER_KNOBS->LATENCY_SKETCH_ACCURACY)), + updateEncryptionLatencySample( + std::make_unique("UpdateEncryptionLatencyMetrics", + self->thisServerID, + SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL, + SERVER_KNOBS->LATENCY_SKETCH_ACCURACY)), readLatencyBands("ReadLatencyBands", self->thisServerID, SERVER_KNOBS->STORAGE_LOGGING_DELAY), mappedRangeSample(std::make_unique("GetMappedRangeMetrics", self->thisServerID, @@ -1647,7 +1648,6 @@ public: self->thisServerID, SERVER_KNOBS->LATENCY_METRICS_LOGGING_INTERVAL, SERVER_KNOBS->LATENCY_SKETCH_ACCURACY)) { - specialCounter(cc, "LastTLogVersion", [self]() { return self->lastTLogVersion; }); specialCounter(cc, "Version", [self]() { return self->version.get(); }); specialCounter(cc, "StorageVersion", [self]() { return self->storageVersion(); }); @@ -2541,7 +2541,7 @@ ACTOR Future getValueQ(StorageServer* data, GetValueRequest req) { // Track time from requestTime through now as read queueing wait time state double queueWaitEnd = g_network->timer(); - data->counters.readQueueWaitSample.addMeasurement(queueWaitEnd - req.requestTime()); + data->counters.readQueueWaitSample->addMeasurement(queueWaitEnd - req.requestTime()); if (req.options.present() && req.options.get().debugID.present()) g_traceBatch.addEvent("GetValueDebug", @@ -2551,7 +2551,7 @@ ACTOR Future getValueQ(StorageServer* data, GetValueRequest req) { state Optional v; Version commitVersion = getLatestCommitVersion(req.ssLatestCommitVersions, data->tag); state Version version = wait(waitForVersion(data, commitVersion, req.version, req.spanContext)); - data->counters.readVersionWaitSample.addMeasurement(g_network->timer() - queueWaitEnd); + data->counters.readVersionWaitSample->addMeasurement(g_network->timer() - queueWaitEnd); if (req.options.present() && req.options.get().debugID.present()) g_traceBatch.addEvent("GetValueDebug", @@ -2649,8 +2649,8 @@ ACTOR Future getValueQ(StorageServer* data, GetValueRequest req) { ++data->counters.finishedQueries; double duration = g_network->timer() - req.requestTime(); - data->counters.readLatencySample.addMeasurement(duration); - data->counters.readValueLatencySample.addMeasurement(duration); + data->counters.readLatencySample->addMeasurement(duration); + data->counters.readValueLatencySample->addMeasurement(duration); if (data->latencyBandConfig.present()) { int maxReadBytes = data->latencyBandConfig.get().readConfig.maxReadBytes.orDefault(std::numeric_limits::max()); @@ -4747,7 +4747,7 @@ ACTOR Future getKeyValuesQ(StorageServer* data, GetKeyValuesRequest req) // Track time from requestTime through now as read queueing wait time state double queueWaitEnd = g_network->timer(); - data->counters.readQueueWaitSample.addMeasurement(queueWaitEnd - req.requestTime()); + data->counters.readQueueWaitSample->addMeasurement(queueWaitEnd - req.requestTime()); try { if (req.options.present() && req.options.get().debugID.present()) @@ -4764,7 +4764,7 @@ ACTOR Future getKeyValuesQ(StorageServer* data, GetKeyValuesRequest req) .detail("DebugID", req.options.present() && req.options.get().debugID.present() ? req.options.get().debugID.get() : UID()); - data->counters.readVersionWaitSample.addMeasurement(g_network->timer() - queueWaitEnd); + data->counters.readVersionWaitSample->addMeasurement(g_network->timer() - queueWaitEnd); data->checkTenantEntry(version, req.tenantInfo, req.options.present() ? req.options.get().lockAware : false); if (req.tenantInfo.hasTenant()) { @@ -4871,7 +4871,7 @@ ACTOR Future getKeyValuesQ(StorageServer* data, GetKeyValuesRequest req) req.options, req.tenantInfo.prefix)); const double duration = g_network->timer() - kvReadRange; - data->counters.kvReadRangeLatencySample.addMeasurement(duration); + data->counters.kvReadRangeLatencySample->addMeasurement(duration); GetKeyValuesReply r = _r; if (req.options.present() && req.options.get().debugID.present()) @@ -4900,6 +4900,7 @@ ACTOR Future getKeyValuesQ(StorageServer* data, GetKeyValuesRequest req) for (int i = 0; i < r.data.size(); i++) { totalByteSize += r.data[i].expectedSize(); } + if (totalByteSize > 0 && SERVER_KNOBS->READ_SAMPLING_ENABLED) { int64_t bytesReadPerKSecond = std::max(totalByteSize, SERVER_KNOBS->EMPTY_READ_PENALTY) / 2; data->metrics.notifyBytesReadPerKSecond(addPrefix(r.data[0].key, req.tenantInfo.prefix, req.arena), @@ -4931,8 +4932,8 @@ ACTOR Future getKeyValuesQ(StorageServer* data, GetKeyValuesRequest req) ++data->counters.finishedQueries; double duration = g_network->timer() - req.requestTime(); - data->counters.readLatencySample.addMeasurement(duration); - data->counters.readRangeLatencySample.addMeasurement(duration); + data->counters.readLatencySample->addMeasurement(duration); + data->counters.readRangeLatencySample->addMeasurement(duration); if (data->latencyBandConfig.present()) { int maxReadBytes = data->latencyBandConfig.get().readConfig.maxReadBytes.orDefault(std::numeric_limits::max()); @@ -6709,7 +6710,7 @@ ACTOR Future getMappedKeyValuesQ(StorageServer* data, GetMappedKeyValuesRe // Track time from requestTime through now as read queueing wait time state double queueWaitEnd = g_network->timer(); - data->counters.readQueueWaitSample.addMeasurement(queueWaitEnd - req.requestTime()); + data->counters.readQueueWaitSample->addMeasurement(queueWaitEnd - req.requestTime()); try { if (req.options.present() && req.options.get().debugID.present()) @@ -6718,7 +6719,7 @@ ACTOR Future getMappedKeyValuesQ(StorageServer* data, GetMappedKeyValuesRe // VERSION_VECTOR change Version commitVersion = getLatestCommitVersion(req.ssLatestCommitVersions, data->tag); state Version version = wait(waitForVersion(data, commitVersion, req.version, span.context)); - data->counters.readVersionWaitSample.addMeasurement(g_network->timer() - queueWaitEnd); + data->counters.readVersionWaitSample->addMeasurement(g_network->timer() - queueWaitEnd); data->checkTenantEntry(version, req.tenantInfo, req.options.present() ? req.options.get().lockAware : false); if (req.tenantInfo.hasTenant()) { @@ -6877,7 +6878,7 @@ ACTOR Future getMappedKeyValuesQ(StorageServer* data, GetMappedKeyValuesRe ++data->counters.finishedGetMappedRangeQueries; double duration = g_network->timer() - req.requestTime(); - data->counters.readLatencySample.addMeasurement(duration); + data->counters.readLatencySample->addMeasurement(duration); data->counters.mappedRangeSample->addMeasurement(duration); if (data->latencyBandConfig.present()) { int maxReadBytes = @@ -7124,12 +7125,12 @@ ACTOR Future getKeyQ(StorageServer* data, GetKeyRequest req) { // Track time from requestTime through now as read queueing wait time state double queueWaitEnd = g_network->timer(); - data->counters.readQueueWaitSample.addMeasurement(queueWaitEnd - req.requestTime()); + data->counters.readQueueWaitSample->addMeasurement(queueWaitEnd - req.requestTime()); try { Version commitVersion = getLatestCommitVersion(req.ssLatestCommitVersions, data->tag); state Version version = wait(waitForVersion(data, commitVersion, req.version, req.spanContext)); - data->counters.readVersionWaitSample.addMeasurement(g_network->timer() - queueWaitEnd); + data->counters.readVersionWaitSample->addMeasurement(g_network->timer() - queueWaitEnd); data->checkTenantEntry(version, req.tenantInfo, req.options.map(&ReadOptions::lockAware).orDefault(false)); if (req.tenantInfo.hasTenant()) { @@ -7192,8 +7193,8 @@ ACTOR Future getKeyQ(StorageServer* data, GetKeyRequest req) { ++data->counters.finishedQueries; double duration = g_network->timer() - req.requestTime(); - data->counters.readLatencySample.addMeasurement(duration); - data->counters.readKeyLatencySample.addMeasurement(duration); + data->counters.readLatencySample->addMeasurement(duration); + data->counters.readKeyLatencySample->addMeasurement(duration); if (data->latencyBandConfig.present()) { int maxReadBytes = @@ -9131,7 +9132,7 @@ ACTOR Future fetchKeys(StorageServer* data, AddingShard* shard) { tr.setOption(FDBTransactionOptions::LOCK_AWARE); if (!isFullRestore && SERVER_KNOBS->ENABLE_REPLICA_CONSISTENCY_CHECK_ON_DATA_MOVEMENT) { tr.setOption(FDBTransactionOptions::ENABLE_REPLICA_CONSISTENCY_CHECK); - int64_t requiredReplicas = SERVER_KNOBS->CONSISTENCY_CHECK_REQUIRED_REPLICAS; + int64_t requiredReplicas = SERVER_KNOBS->DATAMOVE_CONSISTENCY_CHECK_REQUIRED_REPLICAS; tr.setOption(FDBTransactionOptions::CONSISTENCY_CHECK_REQUIRED_REPLICAS, StringRef((uint8_t*)&requiredReplicas, sizeof(int64_t))); } @@ -9320,7 +9321,9 @@ ACTOR Future fetchKeys(StorageServer* data, AddingShard* shard) { data->thisServerID); } } - metricReporter.addFetchedBytes(expectedBlockSize, this_block.size()); + if (!conductBulkLoad) { + metricReporter.addFetchedBytes(expectedBlockSize, this_block.size()); + } totalBytes += expectedBlockSize; if (shard->reason != DataMovementReason::INVALID && @@ -9375,6 +9378,9 @@ ACTOR Future fetchKeys(StorageServer* data, AddingShard* shard) { if (!fetchKeyCanRetry(e)) { throw e; } + if (!conductBulkLoad) { + data->counters.fetchKeyErrors += 1; + } lastError = e; if (lastError.code() == error_code_storage_replica_comparison_error) { // The inconsistency could be because of the inclusion of a rolled back @@ -12616,8 +12622,8 @@ ACTOR Future update(StorageServer* data, bool* pReceivedUpdate) { data->behind = false; } const double duration = g_network->timer() - updateStart; - data->counters.updateEncryptionLatencySample.addMeasurement(decryptionTime); - data->counters.updateLatencySample.addMeasurement(duration); + data->counters.updateEncryptionLatencySample->addMeasurement(decryptionTime); + data->counters.updateLatencySample->addMeasurement(duration); return Void(); // update will get called again ASAP } catch (Error& err) { diff --git a/flow/Knobs.cpp b/flow/Knobs.cpp index a821682ac2..c3d6b919a4 100644 --- a/flow/Knobs.cpp +++ b/flow/Knobs.cpp @@ -309,7 +309,8 @@ void FlowKnobs::initialize(Randomize randomize, IsSimulated isSimulated) { init( TSS_LARGE_TRACE_SIZE, 50000 ); init( LOAD_BALANCE_FETCH_REPLICA_TIMEOUT, 5.0 ); init( ENABLE_REPLICA_CONSISTENCY_CHECK_ON_READS, false ); if( randomize && BUGGIFY ) ENABLE_REPLICA_CONSISTENCY_CHECK_ON_READS = true; - init( CONSISTENCY_CHECK_REQUIRED_REPLICAS, -2 ); // Do consistency check based on all available storage replicas + init( READ_CONSISTENCY_CHECK_REQUIRED_REPLICAS, -2 ); // Do consistency check based on all available storage replicas + init( ENABLE_WARNING_READ_CONSISTENCY_CHECK_NOT_ENOUGH_REPLICA, false); if (randomize && BUGGIFY) { ENABLE_WARNING_READ_CONSISTENCY_CHECK_NOT_ENOUGH_REPLICA = true; } // Health Monitor init( FAILURE_DETECTION_DELAY, 4.0 ); if( randomize && BUGGIFY ) FAILURE_DETECTION_DELAY = 1.0; diff --git a/flow/include/flow/FastAlloc.h b/flow/include/flow/FastAlloc.h index c7019284cd..7ccb94a088 100644 --- a/flow/include/flow/FastAlloc.h +++ b/flow/include/flow/FastAlloc.h @@ -22,6 +22,7 @@ #define FLOW_FASTALLOC_H #pragma once +#include "flow/Error.h" #include "flow/Platform.h" #include "flow/config.h" @@ -234,7 +235,7 @@ force_inline void freeOrMaybeKeepalive(void* ptr) { } inline constexpr int nextFastAllocatedSize(int x) { - assert(x > 0 && x <= 16384); + ASSERT(x > 0 && x <= 16384); if (x <= 16) return 16; else if (x <= 32) diff --git a/flow/include/flow/Knobs.h b/flow/include/flow/Knobs.h index aabf29519b..8ae4a9aae2 100644 --- a/flow/include/flow/Knobs.h +++ b/flow/include/flow/Knobs.h @@ -376,7 +376,8 @@ public: int TSS_LARGE_TRACE_SIZE; double LOAD_BALANCE_FETCH_REPLICA_TIMEOUT; bool ENABLE_REPLICA_CONSISTENCY_CHECK_ON_READS; - int CONSISTENCY_CHECK_REQUIRED_REPLICAS; + int READ_CONSISTENCY_CHECK_REQUIRED_REPLICAS; + bool ENABLE_WARNING_READ_CONSISTENCY_CHECK_NOT_ENOUGH_REPLICA; // Health Monitor int FAILURE_DETECTION_DELAY;