diff options
Diffstat (limited to 'src/mongo/db/s/resharding')
32 files changed, 760 insertions, 308 deletions
diff --git a/src/mongo/db/s/resharding/document_source_resharding_iterate_transaction.cpp b/src/mongo/db/s/resharding/document_source_resharding_iterate_transaction.cpp index 9d412147b7f..8075111e3af 100644 --- a/src/mongo/db/s/resharding/document_source_resharding_iterate_transaction.cpp +++ b/src/mongo/db/s/resharding/document_source_resharding_iterate_transaction.cpp @@ -134,7 +134,7 @@ DepsTracker::State DocumentSourceReshardingIterateTransaction::getDependencies( DocumentSource::GetModPathsReturn DocumentSourceReshardingIterateTransaction::getModifiedPaths() const { - return {DocumentSource::GetModPathsReturn::Type::kAllPaths, std::set<std::string>{}, {}}; + return {DocumentSource::GetModPathsReturn::Type::kAllPaths, OrderedPathSet{}, {}}; } DocumentSource::GetNextResult DocumentSourceReshardingIterateTransaction::doGetNext() { diff --git a/src/mongo/db/s/resharding/document_source_resharding_ownership_match.cpp b/src/mongo/db/s/resharding/document_source_resharding_ownership_match.cpp index f8a186e3bcd..3144723bf2d 100644 --- a/src/mongo/db/s/resharding/document_source_resharding_ownership_match.cpp +++ b/src/mongo/db/s/resharding/document_source_resharding_ownership_match.cpp @@ -109,7 +109,7 @@ DepsTracker::State DocumentSourceReshardingOwnershipMatch::getDependencies( DocumentSource::GetModPathsReturn DocumentSourceReshardingOwnershipMatch::getModifiedPaths() const { // This stage does not modify or rename any paths. - return {DocumentSource::GetModPathsReturn::Type::kFiniteSet, std::set<std::string>{}, {}}; + return {DocumentSource::GetModPathsReturn::Type::kFiniteSet, OrderedPathSet{}, {}}; } DocumentSource::GetNextResult DocumentSourceReshardingOwnershipMatch::doGetNext() { diff --git a/src/mongo/db/s/resharding/recipient_document.idl b/src/mongo/db/s/resharding/recipient_document.idl index e3128de0db5..1eda4620b8c 100644 --- a/src/mongo/db/s/resharding/recipient_document.idl +++ b/src/mongo/db/s/resharding/recipient_document.idl @@ -82,6 +82,11 @@ structs: startConfigTxnCloneTime: type: date optional: true + approxBytesToCopy: + type: long + description: >- + Approximate number of bytes to copy during cloning + optional: true metrics: type: ReshardingRecipientMetrics description: "Metrics related to this recipient." diff --git a/src/mongo/db/s/resharding/resharding_collection_cloner.cpp b/src/mongo/db/s/resharding/resharding_collection_cloner.cpp index 9132438053a..e51d2e2ed1f 100644 --- a/src/mongo/db/s/resharding/resharding_collection_cloner.cpp +++ b/src/mongo/db/s/resharding/resharding_collection_cloner.cpp @@ -29,8 +29,6 @@ #define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kResharding -#include "mongo/platform/basic.h" - #include "mongo/db/s/resharding/resharding_collection_cloner.h" #include <utility> @@ -39,7 +37,6 @@ #include "mongo/db/catalog/collection.h" #include "mongo/db/catalog_raii.h" #include "mongo/db/client.h" -#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/curop.h" #include "mongo/db/exec/document_value/document.h" #include "mongo/db/logical_session_id_helpers.h" @@ -277,7 +274,7 @@ bool ReshardingCollectionCloner::doOneBatch(OperationContext* opCtx, Pipeline& p // ReshardingOpObserver depends on the collection metadata being known when processing writes to // the temporary resharding collection. We attach shard version IGNORED to the insert operations - // and retry once on a StaleConfig exception to allow the collection metadata information to be + // and retry once on a StaleConfig error to allow the collection metadata information to be // recovered. ScopedSetShardRole scopedSetShardRole(opCtx, _outputNss, diff --git a/src/mongo/db/s/resharding/resharding_coordinator_commit_monitor.cpp b/src/mongo/db/s/resharding/resharding_coordinator_commit_monitor.cpp index 374dcd6538f..5700f0326ae 100644 --- a/src/mongo/db/s/resharding/resharding_coordinator_commit_monitor.cpp +++ b/src/mongo/db/s/resharding/resharding_coordinator_commit_monitor.cpp @@ -164,13 +164,21 @@ CoordinatorCommitMonitor::queryRemainingOperationTimeForRecipients() const { uassertStatusOKWithContext(status, errorContext); const auto remainingTime = extractOperationRemainingTime(shardResponse.data); - // A recipient shard does not report the remaining operation time when there is no data - // to copy and no oplog entry to apply. - if (remainingTime && remainingTime.get() < minRemainingTime) { - minRemainingTime = remainingTime.get(); + + // If any recipient omits the "remainingMillis" field of the response then + // we cannot conclude that it is safe to begin the critical section. + // It is possible that the recipient just had a failover and + // was not able to restore its metrics before it replied to the + // _shardsvrReshardingOperationTime command. + if (!remainingTime) { + maxRemainingTime = Milliseconds::max(); + continue; + } + if (remainingTime.value() < minRemainingTime) { + minRemainingTime = remainingTime.value(); } - if (remainingTime && remainingTime.get() > maxRemainingTime) { - maxRemainingTime = remainingTime.get(); + if (remainingTime.value() > maxRemainingTime) { + maxRemainingTime = remainingTime.value(); } } @@ -203,12 +211,20 @@ ExecutorFuture<void> CoordinatorCommitMonitor::_makeFuture() const { "Encountered an error while querying recipients, will retry shortly", "error"_attr = status); - return RemainingOperationTimes{Milliseconds(0), Milliseconds::max()}; + // On error we definitely cannot begin the critical section. Therefore, + // return Milliseconds::max for remainingTimes.max (remainingTimes.max is used + // for determining whether the critical section should begin). + return RemainingOperationTimes{Milliseconds(-1), Milliseconds::max()}; }) .then([this, anchor = shared_from_this()](RemainingOperationTimes remainingTimes) { auto metrics = ReshardingMetrics::get(cc().getServiceContext()); - metrics->setMinRemainingOperationTime(remainingTimes.min); - metrics->setMaxRemainingOperationTime(remainingTimes.max); + // If remainingTimes.max (or remainingTimes.min) is Milliseconds::max, then use -1 so + // that the scale of the y-axis is still useful when looking at FTDC metrics. + auto clampIfMax = [](Milliseconds t) { + return t != Milliseconds::max() ? t : Milliseconds(-1); + }; + metrics->setMinRemainingOperationTime(clampIfMax(remainingTimes.min)); + metrics->setMaxRemainingOperationTime(clampIfMax(remainingTimes.max)); // Check if all recipient shards are within the commit threshold. if (remainingTimes.max <= _threshold) diff --git a/src/mongo/db/s/resharding/resharding_coordinator_commit_monitor_test.cpp b/src/mongo/db/s/resharding/resharding_coordinator_commit_monitor_test.cpp index e7f90cc41fa..2fe3075f1fc 100644 --- a/src/mongo/db/s/resharding/resharding_coordinator_commit_monitor_test.cpp +++ b/src/mongo/db/s/resharding/resharding_coordinator_commit_monitor_test.cpp @@ -92,6 +92,8 @@ protected: void tearDown() override; void mockCommandForRecipients(Milliseconds remainingOperationTime); + void mockOmitRemainingMillisForRecipients(); + void mockOmitRemainingMillisForOneRecipient(); void mockRemaingOperationTimesCommandForRecipients( CoordinatorCommitMonitor::RemainingOperationTimes remainingOperationTimes); @@ -180,6 +182,31 @@ void CoordinatorCommitMonitorTest::mockCommandForRecipients(Milliseconds remaini _recipientShards.begin(), _recipientShards.end(), [&](const ShardId&) { onCommand(func); }); } +void CoordinatorCommitMonitorTest::mockOmitRemainingMillisForRecipients() { + // Omit remainingMillis from all shard responses. + std::for_each(_recipientShards.begin(), _recipientShards.end(), [this](const ShardId&) { + onCommand([](const executor::RemoteCommandRequest& request) -> StatusWith<BSONObj> { + // Return an empty BSON object. + return BSONObj(); + }); + }); +} + +void CoordinatorCommitMonitorTest::mockOmitRemainingMillisForOneRecipient() { + // Omit remainingMillis from a single recipient. + for (const auto& shard : _recipientShards) { + onCommand([&](const executor::RemoteCommandRequest&) -> StatusWith<BSONObj> { + if (shard == _recipientShards.front()) { + // Return an empty BSON object. + return BSONObj(); + } + auto threshold = Milliseconds(gRemainingReshardingOperationTimeThresholdMillis.load()); + return BSON("remainingMillis" + << durationCount<Milliseconds>(threshold - Milliseconds(1))); + }); + } +} + void CoordinatorCommitMonitorTest::mockRemaingOperationTimesCommandForRecipients( CoordinatorCommitMonitor::RemainingOperationTimes remainingOperationTimes) { bool useMin = true; @@ -264,6 +291,20 @@ TEST_F(CoordinatorCommitMonitorTest, RetriesWhenEncountersErrorsWhileQueryingRec future.get(); } +TEST_F(CoordinatorCommitMonitorTest, BlocksWhenRemainingMillisIsOmitted) { + auto future = getCommitMonitor()->waitUntilRecipientsAreWithinCommitThreshold(); + + mockOmitRemainingMillisForRecipients(); + ASSERT(!future.isReady()); + + // If even a single shard omits remainingMillis, we cannot begin the critical section. + mockOmitRemainingMillisForOneRecipient(); + ASSERT(!future.isReady()); + + respondWithReadyToCommit(); + future.get(); +} + } // namespace } // namespace resharding } // namespace mongo diff --git a/src/mongo/db/s/resharding/resharding_coordinator_service.cpp b/src/mongo/db/s/resharding/resharding_coordinator_service.cpp index 85c14b6478f..df90c6d718a 100644 --- a/src/mongo/db/s/resharding/resharding_coordinator_service.cpp +++ b/src/mongo/db/s/resharding/resharding_coordinator_service.cpp @@ -37,6 +37,7 @@ #include "mongo/bson/bsonobj.h" #include "mongo/bson/json.h" #include "mongo/db/auth/authorization_session_impl.h" +#include "mongo/db/concurrency/exception_util.h" #include "mongo/db/dbdirectclient.h" #include "mongo/db/logical_session_cache.h" #include "mongo/db/ops/write_ops.h" @@ -94,6 +95,7 @@ MONGO_FAIL_POINT_DEFINE(reshardingPauseCoordinatorBeforeStartingErrorFlow); MONGO_FAIL_POINT_DEFINE(reshardingPauseCoordinatorBeforePersistingStateTransition); MONGO_FAIL_POINT_DEFINE(pauseBeforeTellDonorToRefresh); MONGO_FAIL_POINT_DEFINE(pauseBeforeInsertCoordinatorDoc); +MONGO_FAIL_POINT_DEFINE(pauseBeforeCTHolderInitialization); const std::string kReshardingCoordinatorActiveIndexName = "ReshardingCoordinatorActiveIndex"; const Backoff kExponentialBackoff(Seconds(1), Milliseconds::max()); @@ -505,43 +507,33 @@ void insertChunkAndTagDocsForTempNss(OperationContext* opCtx, ShardingCatalogManager::get(opCtx)->insertConfigDocuments(opCtx, TagsType::ConfigNS, newZones); } +void removeTagsDocs(OperationContext* opCtx, const BSONObj& tagsQuery, TxnNumber txnNumber) { + // Remove tag documents with the specified tagsQuery. + const auto tagDeleteOperationHint = BSON(TagsType::ns() << 1 << TagsType::min() << 1); + ShardingCatalogManager::get(opCtx)->writeToConfigDocumentInTxn( + opCtx, + TagsType::ConfigNS, + BatchedCommandRequest::buildDeleteOp(TagsType::ConfigNS, + tagsQuery, // query + true, // multi + tagDeleteOperationHint // hint + ), + txnNumber); +} + // Requires that there be no session information on the opCtx. void removeChunkAndTagsDocs(OperationContext* opCtx, const BSONObj& tagsQuery, const UUID& collUUID) { - // Remove all chunk documents for the original nss. We do not know how many chunk docs - // currently exist, so cannot pass a value for expectedNumModified - const auto chunksQuery = BSON(ChunkType::collectionUUID() << collUUID); - const auto tagDeleteOperationHint = BSON(TagsType::ns() << 1 << TagsType::min() << 1); + // Remove all chunk documents and specified tag documents. + resharding::removeChunkDocs(opCtx, collUUID); + const auto tagDeleteOperationHint = BSON(TagsType::ns() << 1 << TagsType::min() << 1); const auto catalogClient = Grid::get(opCtx)->catalogClient(); - - uassertStatusOK(catalogClient->removeConfigDocuments( - opCtx, ChunkType::ConfigNS, chunksQuery, kMajorityWriteConcern)); uassertStatusOK(catalogClient->removeConfigDocuments( opCtx, TagsType::ConfigNS, tagsQuery, kMajorityWriteConcern, tagDeleteOperationHint)); } -void updateChunkAndTagsDocsForTempNss(OperationContext* opCtx, - const ReshardingCoordinatorDocument& coordinatorDoc, - OID newCollectionEpoch, - TxnNumber txnNumber) { - auto hint = BSON("ns" << 1 << "min" << 1); - auto tagsRequest = BatchedCommandRequest::buildUpdateOp( - TagsType::ConfigNS, - BSON(TagsType::ns(coordinatorDoc.getTempReshardingNss().ns())), // query - BSON("$set" << BSON("ns" << coordinatorDoc.getSourceNss().ns())), // update - false, // upsert - true, // multi - hint // hint - ); - - // Update the 'ns' field to be the original collection namespace for all tags documents that - // currently have 'ns' as the temporary collection namespace - auto tagsRes = ShardingCatalogManager::get(opCtx)->writeToConfigDocumentInTxn( - opCtx, TagsType::ConfigNS, tagsRequest, txnNumber); -} - /** * Executes metadata changes in a transaction without bumping the collection version. */ @@ -592,28 +584,14 @@ CollectionType createTempReshardingCollectionType( return collType; } -void cleanupSourceConfigCollections(OperationContext* opCtx, - const ReshardingCoordinatorDocument& coordinatorDoc) { - using Doc = Document; - using Arr = std::vector<Value>; - using V = Value; - - auto createTagFilter = [](const V value) { - return V{Doc{{"$map", - V{Doc{{"input", V{Doc{{"$objectToArray", value}}}}, - {"in", V{StringData("$$this.k")}}}}}}}; - }; - - - auto skipNewTagsFilter = Doc{ - {"$ne", - Arr{createTagFilter(V{StringData("$min")}), - createTagFilter(V{Doc{{"$literal", coordinatorDoc.getReshardingKey().toBSON()}}})}}}; - - const auto removeTagsQuery = - BSON(TagsType::ns(coordinatorDoc.getSourceNss().ns()) << "$expr" << skipNewTagsFilter); +void removeChunkDocs(OperationContext* opCtx, const UUID& collUUID) { + // Remove all chunk documents for the specified collUUID. We do not know how many chunk docs + // currently exist, so cannot pass a value for expectedNumModified + const auto chunksQuery = BSON(ChunkType::collectionUUID() << collUUID); + const auto catalogClient = Grid::get(opCtx)->catalogClient(); - removeChunkAndTagsDocs(opCtx, removeTagsQuery, coordinatorDoc.getSourceUUID()); + uassertStatusOK(catalogClient->removeConfigDocuments( + opCtx, ChunkType::ConfigNS, chunksQuery, kMajorityWriteConcern)); } void writeDecisionPersistedState(OperationContext* opCtx, @@ -622,21 +600,49 @@ void writeDecisionPersistedState(OperationContext* opCtx, Timestamp newCollectionTimestamp) { // No need to bump originalNss version because its epoch will be changed. - executeMetadataChangesInTxn(opCtx, [&](OperationContext* opCtx, TxnNumber txnNumber) { - // Update the config.reshardingOperations entry - writeToCoordinatorStateNss(opCtx, coordinatorDoc, txnNumber); + executeMetadataChangesInTxn( + opCtx, + [&coordinatorDoc, &newCollectionEpoch, &newCollectionTimestamp](OperationContext* opCtx, + TxnNumber txnNumber) { + // Update the config.reshardingOperations entry + writeToCoordinatorStateNss(opCtx, coordinatorDoc, txnNumber); - // Remove the config.collections entry for the temporary collection - writeToConfigCollectionsForTempNss( - opCtx, coordinatorDoc, boost::none, boost::none, txnNumber); + // Remove the config.collections entry for the temporary collection + writeToConfigCollectionsForTempNss( + opCtx, coordinatorDoc, boost::none, boost::none, txnNumber); - // Update the config.collections entry for the original namespace to reflect the new - // shard key, new epoch, and new UUID - updateConfigCollectionsForOriginalNss( - opCtx, coordinatorDoc, newCollectionEpoch, newCollectionTimestamp, txnNumber); + // Update the config.collections entry for the original namespace to reflect the new + // shard key, new epoch, and new UUID + updateConfigCollectionsForOriginalNss( + opCtx, coordinatorDoc, newCollectionEpoch, newCollectionTimestamp, txnNumber); + + // Delete all of the config.tags entries for the user collection namespace. + const auto removeTagsQuery = BSON(TagsType::ns(coordinatorDoc.getSourceNss().ns())); + removeTagsDocs(opCtx, removeTagsQuery, txnNumber); - updateChunkAndTagsDocsForTempNss(opCtx, coordinatorDoc, newCollectionEpoch, txnNumber); - }); + // Update all of the config.tags entries for the temporary resharding namespace + // to refer to the user collection namespace. + updateTagsDocsForTempNss(opCtx, coordinatorDoc, txnNumber); + }); +} + +void updateTagsDocsForTempNss(OperationContext* opCtx, + const ReshardingCoordinatorDocument& coordinatorDoc, + TxnNumber txnNumber) { + auto hint = BSON("ns" << 1 << "min" << 1); + auto tagsRequest = BatchedCommandRequest::buildUpdateOp( + TagsType::ConfigNS, + BSON(TagsType::ns(coordinatorDoc.getTempReshardingNss().ns())), // query + BSON("$set" << BSON("ns" << coordinatorDoc.getSourceNss().ns())), // update + false, // upsert + true, // multi + hint // hint + ); + + // Update the 'ns' field to be the original collection namespace for all tags documents that + // currently have 'ns' as the temporary collection namespace. + auto tagsRes = ShardingCatalogManager::get(opCtx)->writeToConfigDocumentInTxn( + opCtx, TagsType::ConfigNS, tagsRequest, txnNumber); } void insertCoordDocAndChangeOrigCollEntry(OperationContext* opCtx, @@ -1072,6 +1078,7 @@ ReshardingCoordinatorService::ReshardingCoordinator::_tellAllParticipantsReshard _cancelableOpCtxFactory.emplace(_ctHolder->getStepdownToken(), _markKilledExecutor); }) + .then([this] { return _waitForMajority(_ctHolder->getStepdownToken()); }) .then([this, executor]() { pauseBeforeTellDonorToRefresh.pauseWhileSet(); _establishAllDonorsAsParticipants(executor); @@ -1102,8 +1109,7 @@ ExecutorFuture<void> ReshardingCoordinatorService::ReshardingCoordinator::_initi return resharding::WithAutomaticRetry([this, executor] { return ExecutorFuture<void>(**executor) .then([this, executor] { _insertCoordDocAndChangeOrigCollEntry(); }) - .then([this, executor] { _calculateParticipantsAndChunksThenWriteToDisk(); }) - .then([this] { return _waitForMajority(_ctHolder->getAbortToken()); }); + .then([this, executor] { _calculateParticipantsAndChunksThenWriteToDisk(); }); }) .onTransientError([](const Status& status) { LOGV2(5093703, @@ -1242,33 +1248,11 @@ ReshardingCoordinatorService::ReshardingCoordinator::_commitAndFinishReshardOper const ReshardingCoordinatorDocument& updatedCoordinatorDoc) noexcept { return resharding::WithAutomaticRetry([this, executor, updatedCoordinatorDoc] { return ExecutorFuture<void>(**executor) - .then([this, executor, updatedCoordinatorDoc] { - return _commit(updatedCoordinatorDoc); - }) - .then([this] { return _waitForMajority(_ctHolder->getStepdownToken()); }) - .thenRunOn(**executor) - .then([this, executor] { - _tellAllParticipantsToCommit(_coordinatorDoc.getSourceNss(), executor); - }) - .then([this] { _updateChunkImbalanceMetrics(_coordinatorDoc.getSourceNss()); }) - .then([this, updatedCoordinatorDoc] { - auto opCtx = _cancelableOpCtxFactory->makeOperationContext(&cc()); - resharding::cleanupSourceConfigCollections(opCtx.get(), - updatedCoordinatorDoc); - return Status::OK(); - }) - .then([this, executor] { return _awaitAllParticipantShardsDone(executor); }) - .then([this, executor] { - // Best-effort attempt to trigger a refresh on the participant shards so - // they see the collection metadata without reshardingFields and no longer - // throw ReshardCollectionInProgress. There is no guarantee this logic ever - // runs if the config server primary steps down after having removed the - // coordinator state document. - return _tellAllRecipientsToRefresh(executor); - }); + .then( + [this, executor, updatedCoordinatorDoc] { _commit(updatedCoordinatorDoc); }); }) .onTransientError([](const Status& status) { - LOGV2(5093705, + LOGV2(7698801, "Resharding coordinator encountered transient error while committing", "error"_attr = status); }) @@ -1276,25 +1260,83 @@ ReshardingCoordinatorService::ReshardingCoordinator::_commitAndFinishReshardOper .until<Status>([](const Status& status) { return status.isOK(); }) .on(**executor, _ctHolder->getStepdownToken()) .onError([this, executor](Status status) { - { - auto opCtx = _cancelableOpCtxFactory->makeOperationContext(&cc()); - reshardingPauseCoordinatorBeforeStartingErrorFlow.pauseWhileSet(opCtx.get()); + if (status == ErrorCodes::TransactionTooLargeForCache) { + return _onAbortCoordinatorAndParticipants(executor, status); } + return ExecutorFuture<void>(**executor, status); + }) + .then([this, executor, updatedCoordinatorDoc] { + return resharding::WithAutomaticRetry([this, executor, updatedCoordinatorDoc] { + return ExecutorFuture<void>(**executor) + .then([this] { return _waitForMajority(_ctHolder->getStepdownToken()); }) + .thenRunOn(**executor) + .then([this, executor] { + _tellAllParticipantsToCommit(_coordinatorDoc.getSourceNss(), + executor); + }) + .then([this] { + _updateChunkImbalanceMetrics(_coordinatorDoc.getSourceNss()); + }) + .then([this, updatedCoordinatorDoc] { + auto opCtx = _cancelableOpCtxFactory->makeOperationContext(&cc()); + resharding::removeChunkDocs(opCtx.get(), + updatedCoordinatorDoc.getSourceUUID()); + return Status::OK(); + }) + .then([this, executor] { + return _awaitAllParticipantShardsDone(executor); + }) + .then([this, executor] { + // Best-effort attempt to trigger a refresh on the participant shards + // so they see the collection metadata without reshardingFields and + // no longer throw ReshardCollectionInProgress. There is no guarantee + // this logic ever runs if the config server primary steps down after + // having removed the coordinator state document. + return _tellAllRecipientsToRefresh(executor); + }); + }) + .onTransientError([](const Status& status) { + LOGV2(5093705, + "Resharding coordinator encountered transient error while committing", + "error"_attr = status); + }) + .onUnrecoverableError([](const Status& status) {}) + .until<Status>([](const Status& status) { return status.isOK(); }) + .on(**executor, _ctHolder->getStepdownToken()) + .onError([this, executor](Status status) { + { + auto opCtx = _cancelableOpCtxFactory->makeOperationContext(&cc()); + reshardingPauseCoordinatorBeforeStartingErrorFlow.pauseWhileSet( + opCtx.get()); + } - if (_ctHolder->isSteppingOrShuttingDown()) { - return status; - } + if (_ctHolder->isSteppingOrShuttingDown()) { + return status; + } - LOGV2_FATAL(5277000, + LOGV2_FATAL( + 5277000, "Unrecoverable error past the point resharding was guaranteed to succeed", "error"_attr = redact(status)); + }); }); } SemiFuture<void> ReshardingCoordinatorService::ReshardingCoordinator::run( std::shared_ptr<executor::ScopedTaskExecutor> executor, const CancellationToken& stepdownToken) noexcept { - _ctHolder = std::make_unique<CoordinatorCancellationTokenHolder>(stepdownToken); + pauseBeforeCTHolderInitialization.pauseWhileSet(); + + auto abortCalled = [&] { + stdx::lock_guard<Latch> lk(_abortCalledMutex); + _ctHolder = std::make_unique<CoordinatorCancellationTokenHolder>(stepdownToken); + return _abortCalled; + }(); + + if (abortCalled) { + _ctHolder->abort(); + } + _markKilledExecutor->startup(); _cancelableOpCtxFactory.emplace(_ctHolder->getAbortToken(), _markKilledExecutor); @@ -1305,8 +1347,15 @@ SemiFuture<void> ReshardingCoordinatorService::ReshardingCoordinator::run( }) .onCompletion([this, executor](Status status) { auto opCtx = _cancelableOpCtxFactory->makeOperationContext(&cc()); - reshardingPauseCoordinatorBeforeCompletion.pauseWhileSetAndNotCanceled( - opCtx.get(), _ctHolder->getStepdownToken()); + reshardingPauseCoordinatorBeforeCompletion.executeIf( + [&](const BSONObj&) { + reshardingPauseCoordinatorBeforeCompletion.pauseWhileSetAndNotCanceled( + opCtx.get(), _ctHolder->getStepdownToken()); + }, + [&](const BSONObj& data) { + auto ns = data.getStringField("sourceNamespace"); + return ns.empty() ? true : ns.toString() == _coordinatorDoc.getSourceNss().ns(); + }); { auto lg = stdx::lock_guard(_fulfillmentMutex); @@ -1443,7 +1492,15 @@ ReshardingCoordinatorService::ReshardingCoordinator::_onAbortCoordinatorAndParti } void ReshardingCoordinatorService::ReshardingCoordinator::abort() { - _ctHolder->abort(); + auto ctHolderInitialized = [&] { + stdx::lock_guard<Latch> lk(_abortCalledMutex); + _abortCalled = true; + return !(_ctHolder == nullptr); + }(); + + if (ctHolderInitialized) { + _ctHolder->abort(); + } } boost::optional<BSONObj> ReshardingCoordinatorService::ReshardingCoordinator::reportForCurrentOp( @@ -1739,11 +1796,11 @@ ReshardingCoordinatorService::ReshardingCoordinator::_awaitAllRecipientsInStrict .thenRunOn(**executor); } -Future<void> ReshardingCoordinatorService::ReshardingCoordinator::_commit( +void ReshardingCoordinatorService::ReshardingCoordinator::_commit( const ReshardingCoordinatorDocument& coordinatorDoc) { if (_coordinatorDoc.getState() > CoordinatorStateEnum::kBlockingWrites) { invariant(_coordinatorDoc.getState() != CoordinatorStateEnum::kAborting); - return Status::OK(); + return; } ReshardingCoordinatorDocument updatedCoordinatorDoc = coordinatorDoc; @@ -1769,8 +1826,6 @@ Future<void> ReshardingCoordinatorService::ReshardingCoordinator::_commit( // Update the in memory state installCoordinatorDoc(opCtx.get(), updatedCoordinatorDoc); - - return Status::OK(); } ExecutorFuture<void> diff --git a/src/mongo/db/s/resharding/resharding_coordinator_service.h b/src/mongo/db/s/resharding/resharding_coordinator_service.h index a24569ecc44..3fdd0ae05c1 100644 --- a/src/mongo/db/s/resharding/resharding_coordinator_service.h +++ b/src/mongo/db/s/resharding/resharding_coordinator_service.h @@ -50,14 +50,17 @@ CollectionType createTempReshardingCollectionType( const ChunkVersion& chunkVersion, const BSONObj& collation); -void cleanupSourceConfigCollections(OperationContext* opCtx, - const ReshardingCoordinatorDocument& coordinatorDoc); +void removeChunkDocs(OperationContext* opCtx, const UUID& collUUID); void writeDecisionPersistedState(OperationContext* opCtx, const ReshardingCoordinatorDocument& coordinatorDoc, OID newCollectionEpoch, Timestamp newCollectionTimestamp); +void updateTagsDocsForTempNss(OperationContext* opCtx, + const ReshardingCoordinatorDocument& coordinatorDoc, + TxnNumber txnNumber); + void insertCoordDocAndChangeOrigCollEntry(OperationContext* opCtx, const ReshardingCoordinatorDocument& coordinatorDoc); @@ -404,11 +407,10 @@ private: * Does the following writes: * 1. Updates the config.collections entry for the new sharded collection * 2. Updates config.chunks entries for the new sharded collection - * 3. Updates config.tags for the new sharded collection * * Transitions to 'kCommitting'. */ - Future<void> _commit(const ReshardingCoordinatorDocument& updatedDoc); + void _commit(const ReshardingCoordinatorDocument& updatedDoc); /** * Waits on _reshardingCoordinatorObserver to notify that: @@ -534,6 +536,13 @@ private: MONGO_MAKE_LATCH("ReshardingCoordinatorService::_fulfillmentMutex"); /** + * Must be locked while the _abortCalled is being set to true. + */ + mutable Mutex _abortCalledMutex = + MONGO_MAKE_LATCH("ReshardingCoordinatorService::_abortCalledMutex"); + + + /** * Coordinator does not enter the critical section until this is fulfilled. * Can be set by "commitReshardCollection" command or by metrics determining * that it's okay to proceed. @@ -553,6 +562,10 @@ private: std::shared_ptr<resharding::CoordinatorCommitMonitor> _commitMonitor; std::shared_ptr<ReshardingCoordinatorExternalState> _reshardingCoordinatorExternalState; + + // Used to catch the case when an abort() is called but the cancellation source (_ctHolder) has + // not been initialized. + bool _abortCalled{false}; }; } // namespace mongo diff --git a/src/mongo/db/s/resharding/resharding_coordinator_service_test.cpp b/src/mongo/db/s/resharding/resharding_coordinator_service_test.cpp index 6a5197b4c41..188fc400e29 100644 --- a/src/mongo/db/s/resharding/resharding_coordinator_service_test.cpp +++ b/src/mongo/db/s/resharding/resharding_coordinator_service_test.cpp @@ -923,5 +923,16 @@ TEST_F(ReshardingCoordinatorServiceTest, ReshardingCoordinatorFailsIfMigrationNo } } +TEST_F(ReshardingCoordinatorServiceTest, SuccessfullyAbortReshardOperationImmediately) { + auto pauseBeforeCTHolderInitialization = + globalFailPointRegistry().find("pauseBeforeCTHolderInitialization"); + auto timesEnteredFailPoint = pauseBeforeCTHolderInitialization->setMode(FailPoint::alwaysOn, 0); + auto coordinator = initializeAndGetCoordinator(); + coordinator->abort(); + pauseBeforeCTHolderInitialization->waitForTimesEntered(timesEnteredFailPoint + 1); + pauseBeforeCTHolderInitialization->setMode(FailPoint::off, 0); + coordinator->getCompletionFuture().wait(); +} + } // namespace } // namespace mongo diff --git a/src/mongo/db/s/resharding/resharding_coordinator_test.cpp b/src/mongo/db/s/resharding/resharding_coordinator_test.cpp index a8fb4d83889..974879b2959 100644 --- a/src/mongo/db/s/resharding/resharding_coordinator_test.cpp +++ b/src/mongo/db/s/resharding/resharding_coordinator_test.cpp @@ -636,7 +636,7 @@ protected: ReshardingCoordinatorDocument expectedCoordinatorDoc, std::vector<ChunkType> expectedChunks, std::vector<TagsType> expectedZones) { - cleanupSourceConfigCollections(opCtx, expectedCoordinatorDoc); + removeChunkDocs(opCtx, expectedCoordinatorDoc.getSourceUUID()); // Check that chunks and tags entries previously under the temporary namespace have been // correctly updated to the original namespace diff --git a/src/mongo/db/s/resharding/resharding_data_copy_util.cpp b/src/mongo/db/s/resharding/resharding_data_copy_util.cpp index 9893b2b0f2e..d0b27f00c3c 100644 --- a/src/mongo/db/s/resharding/resharding_data_copy_util.cpp +++ b/src/mongo/db/s/resharding/resharding_data_copy_util.cpp @@ -33,7 +33,7 @@ #include "mongo/db/catalog/rename_collection.h" #include "mongo/db/catalog_raii.h" -#include "mongo/db/concurrency/write_conflict_exception.h" +#include "mongo/db/concurrency/exception_util.h" #include "mongo/db/curop.h" #include "mongo/db/dbhelpers.h" #include "mongo/db/namespace_string.h" @@ -44,6 +44,7 @@ #include "mongo/db/s/resharding/resharding_txn_cloner_progress_gen.h" #include "mongo/db/s/resharding/resharding_util.h" #include "mongo/db/s/session_catalog_migration.h" +#include "mongo/db/s/sharding_ddl_util.h" #include "mongo/db/session_catalog_mongod.h" #include "mongo/db/session_txn_record_gen.h" #include "mongo/db/storage/write_unit_of_work.h" @@ -71,28 +72,6 @@ void ensureCollectionExists(OperationContext* opCtx, }); } -void ensureCollectionDropped(OperationContext* opCtx, - const NamespaceString& nss, - const boost::optional<UUID>& uuid) { - invariant(!opCtx->lockState()->isLocked()); - invariant(!opCtx->lockState()->inAWriteUnitOfWork()); - - writeConflictRetry( - opCtx, "resharding::data_copy::ensureCollectionDropped", nss.toString(), [&] { - AutoGetCollection coll(opCtx, nss, MODE_X); - if (!coll || (uuid && coll->uuid() != uuid)) { - // If the collection doesn't exist or exists with a different UUID, then the - // requested collection has been dropped already. - return; - } - - WriteUnitOfWork wuow(opCtx); - uassertStatusOK(coll.getDb()->dropCollectionEvenIfSystem( - opCtx, nss, {} /* dropOpTime */, true /* markFromMigrate */)); - wuow.commit(); - }); -} - void ensureOplogCollectionsDropped(OperationContext* opCtx, const UUID& reshardingUUID, const UUID& sourceUUID, @@ -119,11 +98,11 @@ void ensureOplogCollectionsDropped(OperationContext* opCtx, // Drop the conflict stash collection for this donor. auto stashNss = getLocalConflictStashNamespace(sourceUUID, donor.getShardId()); - ensureCollectionDropped(opCtx, stashNss); + mongo::sharding_ddl_util::ensureCollectionDroppedNoChangeEvent(opCtx, stashNss); // Drop the oplog buffer collection for this donor. auto oplogBufferNss = getLocalOplogBufferNamespace(sourceUUID, donor.getShardId()); - ensureCollectionDropped(opCtx, oplogBufferNss); + mongo::sharding_ddl_util::ensureCollectionDroppedNoChangeEvent(opCtx, oplogBufferNss); } } diff --git a/src/mongo/db/s/resharding/resharding_data_copy_util.h b/src/mongo/db/s/resharding/resharding_data_copy_util.h index 9f2a332ef6c..b51cfc250a5 100644 --- a/src/mongo/db/s/resharding/resharding_data_copy_util.h +++ b/src/mongo/db/s/resharding/resharding_data_copy_util.h @@ -60,16 +60,6 @@ void ensureCollectionExists(OperationContext* opCtx, const CollectionOptions& options); /** - * Drops the specified collection or returns without error if the collection has already been - * dropped. A particular incarnation of the collection can be dropped by specifying its UUID. - * - * This functions assumes the collection being dropped doesn't have any two-phase index builds - * active on it. - */ -void ensureCollectionDropped(OperationContext* opCtx, - const NamespaceString& nss, - const boost::optional<UUID>& uuid = boost::none); -/** * Removes documents from the oplog applier progress and transaction applier progress collections * that are associated with an in-progress resharding operation. Also drops all oplog buffer * collections and conflict stash collections that are associated with the in-progress resharding @@ -159,8 +149,8 @@ void updateSessionRecord(OperationContext* opCtx, /** * Calls and returns the value from the supplied lambda function. * - * If a StaleConfig exception is thrown during its execution, then this function will attempt to - * refresh the collection and invoke the supplied lambda function a second time. + * If a StaleConfig error is thrown during its execution, then this function will attempt to refresh + * the collection and invoke the supplied lambda function a second time. */ template <typename Callable> auto withOneStaleConfigRetry(OperationContext* opCtx, Callable&& callable) { diff --git a/src/mongo/db/s/resharding/resharding_destined_recipient_test.cpp b/src/mongo/db/s/resharding/resharding_destined_recipient_test.cpp index e942dcd139f..28f10ae9143 100644 --- a/src/mongo/db/s/resharding/resharding_destined_recipient_test.cpp +++ b/src/mongo/db/s/resharding/resharding_destined_recipient_test.cpp @@ -135,10 +135,10 @@ public: return repl::OpTimeWith<std::vector<ShardType>>(_shards); } - std::vector<CollectionType> getCollections( - OperationContext* opCtx, - StringData dbName, - repl::ReadConcernLevel readConcernLevel) override { + std::vector<CollectionType> getCollections(OperationContext* opCtx, + StringData dbName, + repl::ReadConcernLevel readConcernLevel, + const BSONObj& sort) override { return _colls; } @@ -237,7 +237,7 @@ protected: createChunks(env.version.epoch(), env.sourceUuid, env.version.getTimestamp(), "y"), boost::none); - forceDatabaseRefresh(opCtx, kNss.db()); + ASSERT_OK(onDbVersionMismatchNoExcept(opCtx, kNss.db(), boost::none)); forceShardFilteringMetadataRefresh(opCtx, kNss); if (refreshTempNss) diff --git a/src/mongo/db/s/resharding/resharding_donor_recipient_common.cpp b/src/mongo/db/s/resharding/resharding_donor_recipient_common.cpp index e9a466a1cb6..99ab47fc11a 100644 --- a/src/mongo/db/s/resharding/resharding_donor_recipient_common.cpp +++ b/src/mongo/db/s/resharding/resharding_donor_recipient_common.cpp @@ -51,10 +51,35 @@ using DonorStateMachine = ReshardingDonorService::DonorStateMachine; using RecipientStateMachine = ReshardingRecipientService::RecipientStateMachine; namespace { +MONGO_FAIL_POINT_DEFINE(reshardingInterruptAfterInsertStateMachineDocument); + using namespace fmt::literals; const Backoff kExponentialBackoff(Seconds(1), Milliseconds::max()); +template <class StateMachine, class ReshardingDocument> +void ensureStateDocumentInserted(OperationContext* opCtx, const ReshardingDocument& doc) { + try { + StateMachine::insertStateDocument(opCtx, doc); + } catch (const ExceptionFor<ErrorCodes::DuplicateKey>& ex) { + // It's possible that the state document was already previously inserted in the following + // cases: + // 1. The document was inserted previously, but the opCtx was interrupted before the + // state machine was started in-memory with getOrCreate(), e.g. due to a chunk migration + // (see SERVER-74647) + // 2. Similar to the ErrorCategory::NotPrimaryError clause below, it is + // theoretically possible for a series of stepdowns and step-ups to lead a scenario where a + // stale but now re-elected primary attempts to insert the state document when another node + // which was primary had already done so. Again, rather than attempt to prevent replica set + // member state transitions during the shard version refresh, we instead swallow the + // DuplicateKey exception. This is safe because PrimaryOnlyService::onStepUp() will have + // constructed a new instance of the resharding state machine. + auto dupeKeyInfo = ex.extraInfo<DuplicateKeyErrorInfo>(); + invariant(dupeKeyInfo->getDuplicatedKeyValue().binaryEqual( + BSON("_id" << doc.getReshardingUUID()))); + } +} + /* * Creates a ReshardingStateMachine if this node is primary and the ReshardingStateMachine doesn't * already exist. @@ -67,7 +92,10 @@ void createReshardingStateMachine(OperationContext* opCtx, const ReshardingDocum // Inserting the resharding state document must happen synchronously with the shard version // refresh for the w:majority wait from the resharding coordinator to mean that this replica // set shard cannot forget about being a participant. - StateMachine::insertStateDocument(opCtx, doc); + ensureStateDocumentInserted<StateMachine>(opCtx, doc); + + reshardingInterruptAfterInsertStateMachineDocument.execute( + [&opCtx](const BSONObj& data) { opCtx->markKilled(); }); auto registry = repl::PrimaryOnlyServiceRegistry::get(opCtx->getServiceContext()); auto service = registry->lookupServiceByName(Service::kServiceName); @@ -82,17 +110,6 @@ void createReshardingStateMachine(OperationContext* opCtx, const ReshardingDocum // secondary (or primary which stepped down) must do for an active resharding operation upon // refreshing its shard version. The primary is solely responsible for advancing the // participant state as a result of the shard version refresh. - } catch (const ExceptionFor<ErrorCodes::DuplicateKey>& ex) { - // Similar to the ErrorCategory::NotPrimaryError clause above, it is theoretically possible - // for a series of stepdowns and step-ups to lead a scenario where a stale but now - // re-elected primary attempts to insert the state document when another node which was - // primary had already done so. Again, rather than attempt to prevent replica set member - // state transitions during the shard version refresh, we instead swallow the DuplicateKey - // exception. This is safe because PrimaryOnlyService::onStepUp() will have constructed a - // new instance of the resharding state machine. - auto dupeKeyInfo = ex.extraInfo<DuplicateKeyErrorInfo>(); - invariant(dupeKeyInfo->getDuplicatedKeyValue().binaryEqual( - BSON("_id" << doc.getReshardingUUID()))); } } @@ -138,6 +155,13 @@ void processReshardingFieldsForDonorCollection(OperationContext* opCtx, return; } + // We clear the routing information for the temporary resharding namespace to ensure this donor + // shard primary will refresh from the config server and see the chunk distribution for the new + // resharding operation. + auto* catalogCache = Grid::get(opCtx)->catalogCache(); + catalogCache->invalidateCollectionEntry_LINEARIZABLE( + reshardingFields.getDonorFields()->getTempReshardingNss()); + auto donorDoc = constructDonorDocumentFromReshardingFields(nss, metadata, reshardingFields); createReshardingStateMachine<ReshardingDonorService, DonorStateMachine, @@ -328,7 +352,16 @@ void clearFilteringMetadata(OperationContext* opCtx, bool scheduleAsyncRefresh) }); } + auto* catalogCache = Grid::get(opCtx)->catalogCache(); + for (const auto& nss : namespacesToRefresh) { + if (nss.isTemporaryReshardingCollection()) { + // We clear the routing information for the temporary resharding namespace to ensure all + // new donor shard primaries will refresh from the config server and see the chunk + // distribution for the ongoing resharding operation. + catalogCache->invalidateCollectionEntry_LINEARIZABLE(nss); + } + AutoGetCollection autoColl(opCtx, nss, MODE_IX); CollectionShardingRuntime::get(opCtx, nss)->clearFilteringMetadata(opCtx); diff --git a/src/mongo/db/s/resharding/resharding_donor_service.cpp b/src/mongo/db/s/resharding/resharding_donor_service.cpp index ce60e58f0e1..4594c4018a6 100644 --- a/src/mongo/db/s/resharding/resharding_donor_service.cpp +++ b/src/mongo/db/s/resharding/resharding_donor_service.cpp @@ -37,7 +37,7 @@ #include "mongo/db/catalog/drop_collection.h" #include "mongo/db/catalog/rename_collection.h" #include "mongo/db/catalog_raii.h" -#include "mongo/db/concurrency/write_conflict_exception.h" +#include "mongo/db/concurrency/exception_util.h" #include "mongo/db/db_raii.h" #include "mongo/db/dbdirectclient.h" #include "mongo/db/dbhelpers.h" @@ -55,6 +55,7 @@ #include "mongo/db/s/resharding/resharding_metrics.h" #include "mongo/db/s/resharding/resharding_server_parameters_gen.h" #include "mongo/db/s/resharding/resharding_util.h" +#include "mongo/db/s/sharding_ddl_util.h" #include "mongo/db/s/sharding_state.h" #include "mongo/db/write_block_bypass.h" #include "mongo/db/write_concern_options.h" @@ -65,6 +66,7 @@ namespace mongo { +MONGO_FAIL_POINT_DEFINE(reshardingPauseDonorBeforeCatalogCacheRefresh); MONGO_FAIL_POINT_DEFINE(reshardingDonorFailsAfterTransitionToDonatingOplogEntries); MONGO_FAIL_POINT_DEFINE(removeDonorDocFailpoint); @@ -567,6 +569,8 @@ void ReshardingDonorService::DonorStateMachine:: // with a SnapshotUnavailable error response. { auto opCtx = _cancelableOpCtxFactory->makeOperationContext(&cc()); + reshardingPauseDonorBeforeCatalogCacheRefresh.pauseWhileSet(opCtx.get()); + _externalState->refreshCatalogCache(opCtx.get(), _metadata.getTempReshardingNss()); _externalState->waitForCollectionFlush(opCtx.get(), _metadata.getTempReshardingNss()); } @@ -795,7 +799,7 @@ void ReshardingDonorService::DonorStateMachine::_dropOriginalCollectionThenTrans // Allow bypassing user write blocking. The check has already been performed on the // db-primary shard's ReshardCollectionCoordinator. WriteBlockBypass::get(opCtx.get()).set(true); - resharding::data_copy::ensureCollectionDropped( + mongo::sharding_ddl_util::ensureCollectionDroppedNoChangeEvent( opCtx.get(), _metadata.getSourceNss(), _metadata.getSourceUUID()); } diff --git a/src/mongo/db/s/resharding/resharding_donor_service_test.cpp b/src/mongo/db/s/resharding/resharding_donor_service_test.cpp index 663b8c28136..6ff3c17fdcf 100644 --- a/src/mongo/db/s/resharding/resharding_donor_service_test.cpp +++ b/src/mongo/db/s/resharding/resharding_donor_service_test.cpp @@ -34,7 +34,6 @@ #include <boost/optional/optional_io.hpp> #include <utility> -#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/dbdirectclient.h" #include "mongo/db/op_observer_noop.h" #include "mongo/db/op_observer_registry.h" @@ -53,6 +52,7 @@ #include "mongo/db/s/resharding/resharding_donor_service.h" #include "mongo/db/s/resharding/resharding_service_test_helpers.h" #include "mongo/db/s/resharding/resharding_util.h" +#include "mongo/db/s/sharding_ddl_util.h" #include "mongo/logv2/log.h" #include "mongo/s/catalog/sharding_catalog_client.h" #include "mongo/unittest/death_test.h" @@ -161,7 +161,8 @@ public: void createSourceCollection(OperationContext* opCtx, const ReshardingDonorDocument& donorDoc) { CollectionOptions options; options.uuid = donorDoc.getSourceUUID(); - resharding::data_copy::ensureCollectionDropped(opCtx, donorDoc.getSourceNss()); + mongo::sharding_ddl_util::ensureCollectionDroppedNoChangeEvent(opCtx, + donorDoc.getSourceNss()); resharding::data_copy::ensureCollectionExists(opCtx, donorDoc.getSourceNss(), options); } @@ -169,7 +170,8 @@ public: const ReshardingDonorDocument& donorDoc) { CollectionOptions options; options.uuid = donorDoc.getReshardingUUID(); - resharding::data_copy::ensureCollectionDropped(opCtx, donorDoc.getTempReshardingNss()); + mongo::sharding_ddl_util::ensureCollectionDroppedNoChangeEvent( + opCtx, donorDoc.getTempReshardingNss()); resharding::data_copy::ensureCollectionExists( opCtx, donorDoc.getTempReshardingNss(), options); } diff --git a/src/mongo/db/s/resharding/resharding_future_util.cpp b/src/mongo/db/s/resharding/resharding_future_util.cpp index 849a14bd80e..62c95fbaf3f 100644 --- a/src/mongo/db/s/resharding/resharding_future_util.cpp +++ b/src/mongo/db/s/resharding/resharding_future_util.cpp @@ -51,19 +51,42 @@ ExecutorFuture<void> whenAllSucceedOn(const std::vector<SharedSemiFuture<void>>& : ExecutorFuture(executor); } +std::vector<Future<void>> runAllInlineUnsafe(const std::vector<SharedSemiFuture<void>>& futures) { + std::vector<Future<void>> result; + result.reserve(futures.size()); + + for (const auto& future : futures) { + result.emplace_back(future.unsafeToInlineFuture()); + } + + return result; +} + ExecutorFuture<void> cancelWhenAnyErrorThenQuiesce( const std::vector<SharedSemiFuture<void>>& futures, ExecutorPtr executor, CancellationSource cancelSource) { - return whenAllSucceedOn(futures, executor) - .onError([futures, executor, cancelSource](Status originalError) mutable { + if (futures.empty()) { + return ExecutorFuture(executor); + } + // Run all futures inline so that the onError callback is called even if that error was caused + // by the executor shutting down. This causes the logic for whenAllSucceed, whenAll, and the + // onError callback to potentially run on the threads of the setters of the promises + // associated with the input futures. Since this logic is thread safe, not blocking, and does + // not acquire additional resources, this is safe, but beware if making further changes to this + // function. + return whenAllSucceed(runAllInlineUnsafe(futures)) + .unsafeToInlineFuture() + .onError([futures, cancelSource](Status originalError) mutable { cancelSource.cancel(); - return whenAll(thenRunAllOn(futures, executor)) + return whenAll(runAllInlineUnsafe(futures)) .ignoreValue() - .thenRunOn(executor) + .unsafeToInlineFuture() .onCompletion([originalError](auto) { return originalError; }); - }); + }) + .thenRunOn(executor); } + } // namespace mongo::resharding diff --git a/src/mongo/db/s/resharding/resharding_future_util.h b/src/mongo/db/s/resharding/resharding_future_util.h index 0bb858a963c..78847350d0c 100644 --- a/src/mongo/db/s/resharding/resharding_future_util.h +++ b/src/mongo/db/s/resharding/resharding_future_util.h @@ -131,7 +131,8 @@ public: status.isA<ErrorCategory::CursorInvalidatedError>() || status == ErrorCodes::Interrupted || status.isA<ErrorCategory::CancellationError>() || - status.isA<ErrorCategory::NotPrimaryError>()) { + status.isA<ErrorCategory::NotPrimaryError>() || + status.isA<ErrorCategory::NetworkTimeoutError>()) { // Always attempt to retry on any type of retryable error. Also retry on errors // from stray killCursors and killOp commands being run. Cancellation and // NotPrimary errors may indicate the primary-only service Instance will be shut diff --git a/src/mongo/db/s/resharding/resharding_future_util_test.cpp b/src/mongo/db/s/resharding/resharding_future_util_test.cpp new file mode 100644 index 00000000000..e37a13a314b --- /dev/null +++ b/src/mongo/db/s/resharding/resharding_future_util_test.cpp @@ -0,0 +1,100 @@ +/** + * Copyright (C) 2022-present MongoDB, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * <http://www.mongodb.com/licensing/server-side-public-license>. + * + * As a special exception, the copyright holders give permission to link the + * code of portions of this program with the OpenSSL library under certain + * conditions as described in each individual source file and distribute + * linked combinations including the program with the OpenSSL library. You + * must comply with the Server Side Public License in all respects for + * all of the code used other than as permitted herein. If you modify file(s) + * with this exception, you may extend this exception to your version of the + * file(s), but you are not obligated to do so. If you do not wish to do so, + * delete this exception statement from your version. If you delete this + * exception statement from all source files in the program, then also delete + * it in the license file. + */ + +#include "mongo/db/s/resharding/resharding_future_util.h" +#include "mongo/unittest/unittest.h" +#include "mongo/util/concurrency/thread_pool.h" + +namespace mongo { +namespace { +class ReshardingFutureUtilTest : public unittest::Test { +protected: + void setUp() override { + _executor = std::make_shared<ThreadPool>([]() { + ThreadPool::Options options; + options.maxThreads = 2; + return options; + }()); + _executor->startup(); + } + + void tearDown() override { + _executor->shutdown(); + _executor->join(); + } + + std::shared_ptr<ThreadPool> getExecutor() const { + return _executor; + } + +private: + std::shared_ptr<ThreadPool> _executor; +}; + +TEST_F(ReshardingFutureUtilTest, CancelWhenAnyErrorThenQuiesceDuringExecutorShutdown) { + CancellationSource cancelSource; + auto token = cancelSource.token(); + PromiseAndFuture<void> taskThreadsReady; + AtomicWord<int> tasksRunningCount{0}; + AtomicWord<bool> taskWasCancelled{false}; + auto checkSignalReady = [&]() { + auto running = tasksRunningCount.addAndFetch(1); + if (running == 2) { + taskThreadsReady.promise.emplaceValue(); + } + }; + PromiseAndFuture<void> executorShutDownTriggered; + auto quiesced = ExecutorFuture(getExecutor()).then([&]() { + return resharding::cancelWhenAnyErrorThenQuiesce( + {ExecutorFuture(getExecutor()) + .then([&]() { + checkSignalReady(); + executorShutDownTriggered.future.wait(); + uasserted(6791600, "Executor shut down"); + }) + .share(), + ExecutorFuture(getExecutor()) + .then([&]() { + checkSignalReady(); + token.onCancel().wait(); + taskWasCancelled.store(true); + }) + .share()}, + getExecutor(), + cancelSource); + }); + taskThreadsReady.future.wait(); + getExecutor()->shutdown(); + executorShutDownTriggered.promise.emplaceValue(); + auto status = quiesced.getNoThrow(); + ASSERT_EQ(status.code(), 6791600); + ASSERT_TRUE(taskWasCancelled.load()); +} +} // namespace +} // namespace mongo diff --git a/src/mongo/db/s/resharding/resharding_metrics.cpp b/src/mongo/db/s/resharding/resharding_metrics.cpp index 6e4e4e041e9..b42c9df5c9e 100644 --- a/src/mongo/db/s/resharding/resharding_metrics.cpp +++ b/src/mongo/db/s/resharding/resharding_metrics.cpp @@ -406,6 +406,60 @@ void ReshardingMetrics::onStepUp(Role role) noexcept { // instead of starting from the current time. } +void ReshardingMetrics::onStepUp(RecipientStateEnum state, + const ReshardingRecipientCountsAndMetrics& recipientMetrics) { + stdx::lock_guard<Latch> lk(_mutex); + + _emplaceCurrentOpForRole(Role::kRecipient, boost::none); + _onStepUpCalled = true; + + invariant(_currentOp, kNoOperationInProgress); + invariant(_currentOp->documentsCopied == 0, kMetricsSetBeforeRestore); + invariant(_currentOp->bytesCopied == 0, kMetricsSetBeforeRestore); + invariant(_currentOp->oplogEntriesFetched == 0, kMetricsSetBeforeRestore); + invariant(_currentOp->oplogEntriesApplied == 0, kMetricsSetBeforeRestore); + + _currentOp->recipientState = state; + _currentOp->documentsCopied = recipientMetrics.documentCountCopied; + _currentOp->bytesCopied = recipientMetrics.documentBytesCopied; + _currentOp->oplogEntriesFetched = recipientMetrics.oplogEntriesFetched; + _currentOp->oplogEntriesApplied = recipientMetrics.oplogEntriesApplied; + + if (recipientMetrics.approxBytesToCopy) + _currentOp->bytesToCopy = recipientMetrics.approxBytesToCopy.get(); + + + const auto& timeIntervals = recipientMetrics.metrics; + + // Restore in memory state of document copy metrics. + // Not calling startCopyingDocuments or endCopyingDocuments because they acquire a mutex that we + // already have. + // + // Also, note that it is possible for documentCopyInterval->getStart() to be none and for + // documentCopyInterval->getStop() to be not none. That can happen if the cluster is upgraded + // to include code for persisting time intervals during a resharding operation. + // In that case, restore neither the start nor stop time. The resharding coordinator will still + // treat this scenario as the recipient shard being completely caught up after a primary + // failover and engage the critical section too early. + const auto& documentCopyInterval = timeIntervals.getDocumentCopy(); + if (documentCopyInterval && documentCopyInterval->getStart()) { + _currentOp->copyingDocuments.start(documentCopyInterval->getStart().get()); + if (documentCopyInterval->getStop()) { + _currentOp->copyingDocuments.end(documentCopyInterval->getStop().get()); + } + } + // Restore in memory state of oplog application metrics. + // Not calling startApplyingOplogEntries or endApplyingOplogEntries because they acquire a mutex + // that we already have. + const auto& oplogApplicationInterval = timeIntervals.getOplogApplication(); + if (oplogApplicationInterval && oplogApplicationInterval->getStart()) { + _currentOp->applyingOplogEntries.start(oplogApplicationInterval->getStart().get()); + if (oplogApplicationInterval->getStop()) { + _currentOp->applyingOplogEntries.end(oplogApplicationInterval->getStop().get()); + } + } +} + void ReshardingMetrics::onStepUp(DonorStateEnum state, ReshardingDonorMetrics donorMetrics) { stdx::lock_guard<Latch> lk(_mutex); auto operationRuntime = donorMetrics.getOperationRuntime(); @@ -484,6 +538,9 @@ void ReshardingMetrics::setDonorState(DonorStateEnum state) noexcept { void ReshardingMetrics::setRecipientState(RecipientStateEnum state) noexcept { stdx::lock_guard<Latch> lk(_mutex); + if (!_currentOp && state == RecipientStateEnum::kDone) { + return; + } invariant(_currentOp, kNoOperationInProgress); const auto oldState = std::exchange(_currentOp->recipientState, state); @@ -634,6 +691,10 @@ void ReshardingMetrics::enterCriticalSection(Date_t start) { void ReshardingMetrics::leaveCriticalSection(Date_t end) { stdx::lock_guard<Latch> lk(_mutex); + if (!_currentOp) { + return; + } + _currentOp->inCriticalSection.forceEnd(end); } @@ -662,22 +723,6 @@ void ReshardingMetrics::onOplogEntriesApplied(int64_t entries) noexcept { _cumulativeOp->oplogEntriesApplied += entries; } -void ReshardingMetrics::restoreForCurrentOp(int64_t documentCountCopied, - int64_t documentBytesCopied, - int64_t oplogEntriesFetched, - int64_t oplogEntriesApplied) noexcept { - invariant(_currentOp, kNoOperationInProgress); - invariant(_currentOp->documentsCopied == 0, kMetricsSetBeforeRestore); - invariant(_currentOp->bytesCopied == 0, kMetricsSetBeforeRestore); - invariant(_currentOp->oplogEntriesFetched == 0, kMetricsSetBeforeRestore); - invariant(_currentOp->oplogEntriesApplied == 0, kMetricsSetBeforeRestore); - - _currentOp->documentsCopied = documentCountCopied; - _currentOp->bytesCopied = documentBytesCopied; - _currentOp->oplogEntriesFetched = oplogEntriesFetched; - _currentOp->oplogEntriesApplied = oplogEntriesApplied; -} - void ReshardingMetrics::onWriteDuringCriticalSection(int64_t writes) noexcept { stdx::lock_guard<Latch> lk(_mutex); if (!_currentOp) diff --git a/src/mongo/db/s/resharding/resharding_metrics.h b/src/mongo/db/s/resharding/resharding_metrics.h index a6964c9d611..9ba80d35417 100644 --- a/src/mongo/db/s/resharding/resharding_metrics.h +++ b/src/mongo/db/s/resharding/resharding_metrics.h @@ -35,6 +35,7 @@ #include "mongo/bson/bsonobjbuilder.h" #include "mongo/db/namespace_string.h" #include "mongo/db/s/resharding/donor_document_gen.h" +#include "mongo/db/s/resharding/recipient_document_gen.h" #include "mongo/db/service_context.h" #include "mongo/platform/mutex.h" #include "mongo/s/resharding/common_types_gen.h" @@ -71,6 +72,29 @@ public: void onStepUp(DonorStateEnum state, ReshardingDonorMetrics donorMetrics); + struct ReshardingRecipientCountsAndMetrics { + ReshardingRecipientCountsAndMetrics(int64_t documentCountCopied, + int64_t documentBytesCopied, + int64_t oplogEntriesFetched, + int64_t oplogEntriesApplied, + boost::optional<int64_t> approxBytesToCopy, + ReshardingRecipientMetrics metrics) + : documentCountCopied{documentCountCopied}, + documentBytesCopied{documentBytesCopied}, + oplogEntriesFetched{oplogEntriesFetched}, + oplogEntriesApplied{oplogEntriesApplied}, + approxBytesToCopy{approxBytesToCopy}, + metrics{metrics} {} + int64_t documentCountCopied; + int64_t documentBytesCopied; + int64_t oplogEntriesFetched; + int64_t oplogEntriesApplied; + boost::optional<int64_t> approxBytesToCopy; + ReshardingRecipientMetrics metrics; + }; + + void onStepUp(RecipientStateEnum, const ReshardingRecipientCountsAndMetrics&); + // So long as a resharding operation is in progress, the following may be used to update the // state of a donor, a recipient, and a coordinator, respectively. void setDonorState(DonorStateEnum) noexcept; @@ -113,11 +137,6 @@ public: // Allows restoring "oplog entries to apply" metrics. void onOplogEntriesApplied(int64_t entries) noexcept; - void restoreForCurrentOp(int64_t documentCountCopied, - int64_t documentBytesCopied, - int64_t oplogEntriesFetched, - int64_t oplogEntriesApplied) noexcept; - // Allows tracking writes during a critical section when the donor's state is either of // "donating-oplog-entries" or "blocking-writes". void onWriteDuringCriticalSection(int64_t writes) noexcept; diff --git a/src/mongo/db/s/resharding/resharding_op_observer.h b/src/mongo/db/s/resharding/resharding_op_observer.h index 30d319a041d..e8affe3ef4a 100644 --- a/src/mongo/db/s/resharding/resharding_op_observer.h +++ b/src/mongo/db/s/resharding/resharding_op_observer.h @@ -228,6 +228,10 @@ public: size_t numberOfPrePostImagesToWrite, Date_t wallClockTime) override {} + void onTransactionPrepareNonPrimary(OperationContext* opCtx, + const std::vector<repl::OplogEntry>& statements, + const repl::OpTime& prepareOpTime) override {} + void onTransactionAbort(OperationContext* opCtx, boost::optional<OplogSlot> abortOplogEntryOpTime) override {} diff --git a/src/mongo/db/s/resharding/resharding_oplog_application.cpp b/src/mongo/db/s/resharding/resharding_oplog_application.cpp index 9a2f6f2750e..42c5a1543e5 100644 --- a/src/mongo/db/s/resharding/resharding_oplog_application.cpp +++ b/src/mongo/db/s/resharding/resharding_oplog_application.cpp @@ -29,11 +29,9 @@ #define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kResharding -#include "mongo/platform/basic.h" - #include "mongo/db/s/resharding/resharding_oplog_application.h" -#include "mongo/db/concurrency/write_conflict_exception.h" +#include "mongo/db/concurrency/exception_util.h" #include "mongo/db/dbhelpers.h" #include "mongo/db/index/index_access_method.h" #include "mongo/db/logical_session_cache.h" @@ -79,8 +77,8 @@ void runWithTransaction(OperationContext* opCtx, // ReshardingOpObserver depends on the collection metadata being known when processing writes to // the temporary resharding collection. We attach shard version IGNORED to the write operations - // and leave it to ReshardingOplogBatchApplier::applyBatch() to retry on a StaleConfig exception - // to allow the collection metadata information to be recovered. + // and leave it to ReshardingOplogBatchApplier::applyBatch() to retry on a StaleConfig error to + // allow the collection metadata information to be recovered. ScopedSetShardRole scopedSetShardRole(asr.opCtx(), nss, ChunkVersion::IGNORED() /* shardVersion */, diff --git a/src/mongo/db/s/resharding/resharding_oplog_batch_applier.cpp b/src/mongo/db/s/resharding/resharding_oplog_batch_applier.cpp index 4ff29b42d30..c7efa6d33e4 100644 --- a/src/mongo/db/s/resharding/resharding_oplog_batch_applier.cpp +++ b/src/mongo/db/s/resharding/resharding_oplog_batch_applier.cpp @@ -29,12 +29,8 @@ #define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kResharding -#include "mongo/platform/basic.h" - #include "mongo/db/s/resharding/resharding_oplog_batch_applier.h" -#include <memory> - #include "mongo/db/s/operation_sharding_state.h" #include "mongo/db/s/resharding/resharding_data_copy_util.h" #include "mongo/db/s/resharding/resharding_future_util.h" @@ -83,8 +79,8 @@ SemiFuture<void> ReshardingOplogBatchApplier::applyBatch( // ReshardingOpObserver depends on the collection metadata being known // when processing writes to the temporary resharding collection. We // attach shard version IGNORED to the write operations and retry once - // on a StaleConfig exception to allow the collection metadata - // information to be recovered. + // on a StaleConfig error to allow the collection metadata information to + // be recovered. ScopedSetShardRole scopedSetShardRole( opCtx.get(), _crudApplication.getOutputNss(), diff --git a/src/mongo/db/s/resharding/resharding_oplog_fetcher.cpp b/src/mongo/db/s/resharding/resharding_oplog_fetcher.cpp index 30811c8d2aa..064ca34c9fa 100644 --- a/src/mongo/db/s/resharding/resharding_oplog_fetcher.cpp +++ b/src/mongo/db/s/resharding/resharding_oplog_fetcher.cpp @@ -41,7 +41,7 @@ #include "mongo/client/dbclient_connection.h" #include "mongo/client/remote_command_targeter.h" #include "mongo/db/catalog_raii.h" -#include "mongo/db/concurrency/write_conflict_exception.h" +#include "mongo/db/concurrency/exception_util.h" #include "mongo/db/pipeline/aggregate_command_gen.h" #include "mongo/db/repl/read_concern_args.h" #include "mongo/db/repl/read_concern_level.h" @@ -194,23 +194,22 @@ ExecutorFuture<void> ReshardingOplogFetcher::_reschedule( } bool ReshardingOplogFetcher::iterate(Client* client, CancelableOperationContextFactory factory) { - std::shared_ptr<Shard> targetShard; - { - auto opCtxRaii = factory.makeOperationContext(client); - opCtxRaii->checkForInterrupt(); - - StatusWith<std::shared_ptr<Shard>> swDonor = - Grid::get(opCtxRaii.get())->shardRegistry()->getShard(opCtxRaii.get(), _donorShard); - if (!swDonor.isOK()) { - LOGV2_WARNING(5127203, - "Error finding shard in registry, retrying.", - "error"_attr = swDonor.getStatus()); - return true; - } - targetShard = swDonor.getValue(); - } - try { + std::shared_ptr<Shard> targetShard; + { + auto opCtxRaii = factory.makeOperationContext(client); + opCtxRaii->checkForInterrupt(); + + StatusWith<std::shared_ptr<Shard>> swDonor = + Grid::get(opCtxRaii.get())->shardRegistry()->getShard(opCtxRaii.get(), _donorShard); + if (!swDonor.isOK()) { + LOGV2_WARNING(5127203, + "Error finding shard in registry, retrying.", + "error"_attr = swDonor.getStatus()); + return true; + } + targetShard = swDonor.getValue(); + } return consume(client, factory, targetShard.get()); } catch (const ExceptionForCat<ErrorCategory::Interruption>&) { // Defer to the cancellation token for whether the Interruption exception should be retried diff --git a/src/mongo/db/s/resharding/resharding_oplog_fetcher_test.cpp b/src/mongo/db/s/resharding/resharding_oplog_fetcher_test.cpp index e7c512e2669..b3b85758e83 100644 --- a/src/mongo/db/s/resharding/resharding_oplog_fetcher_test.cpp +++ b/src/mongo/db/s/resharding/resharding_oplog_fetcher_test.cpp @@ -36,7 +36,7 @@ #include "mongo/bson/bsonobj.h" #include "mongo/db/client.h" -#include "mongo/db/concurrency/write_conflict_exception.h" +#include "mongo/db/concurrency/exception_util.h" #include "mongo/db/db_raii.h" #include "mongo/db/dbdirectclient.h" #include "mongo/db/dbhelpers.h" @@ -666,6 +666,44 @@ TEST_F(ReshardingOplogFetcherTest, RetriesOnRemoteInterruptionError) { ASSERT_TRUE(moreToCome); } +TEST_F(ReshardingOplogFetcherTest, RetriesOnNetworkTimeoutError) { + const NamespaceString outputCollectionNss("dbtests.outputCollection"); + const NamespaceString dataCollectionNss("dbtests.runFetchIteration"); + + create(outputCollectionNss); + create(dataCollectionNss); + _fetchTimestamp = repl::StorageInterface::get(_svcCtx)->getLatestOplogTimestamp(_opCtx); + + const auto& collectionUUID = [&] { + AutoGetCollection dataColl(_opCtx, dataCollectionNss, LockMode::MODE_IX); + return dataColl->uuid(); + }(); + + auto fetcherJob = launchAsync([&, this] { + ThreadClient tc("RunnerForFetcher", _svcCtx, nullptr); + + ReshardingDonorOplogId startAt{_fetchTimestamp, _fetchTimestamp}; + ReshardingOplogFetcher fetcher(makeFetcherEnv(), + _reshardingUUID, + collectionUUID, + startAt, + _donorShard, + _destinationShard, + outputCollectionNss); + + auto factory = makeCancelableOpCtx(); + return fetcher.iterate(&cc(), factory); + }); + + onCommand([&](const executor::RemoteCommandRequest& request) -> StatusWith<BSONObj> { + // Inject network timeout error. + return {ErrorCodes::NetworkInterfaceExceededTimeLimit, "exceeded network time limit"}; + }); + + auto moreToCome = fetcherJob.timed_get(Seconds(5)); + ASSERT_TRUE(moreToCome); +} + TEST_F(ReshardingOplogFetcherTest, ImmediatelyDoneWhenFinalOpHasAlreadyBeenFetched) { const NamespaceString outputCollectionNss("dbtests.outputCollection"); const NamespaceString dataCollectionNss("dbtests.runFetchIteration"); diff --git a/src/mongo/db/s/resharding/resharding_oplog_session_application.cpp b/src/mongo/db/s/resharding/resharding_oplog_session_application.cpp index 44251fa39dc..80338642587 100644 --- a/src/mongo/db/s/resharding/resharding_oplog_session_application.cpp +++ b/src/mongo/db/s/resharding/resharding_oplog_session_application.cpp @@ -32,7 +32,7 @@ #include "mongo/db/s/resharding/resharding_oplog_session_application.h" #include "mongo/db/catalog_raii.h" -#include "mongo/db/concurrency/write_conflict_exception.h" +#include "mongo/db/concurrency/exception_util.h" #include "mongo/db/dbdirectclient.h" #include "mongo/db/operation_context.h" #include "mongo/db/repl/oplog_entry.h" diff --git a/src/mongo/db/s/resharding/resharding_recipient_service.cpp b/src/mongo/db/s/resharding/resharding_recipient_service.cpp index 5d49e0cdb22..087e3cb603f 100644 --- a/src/mongo/db/s/resharding/resharding_recipient_service.cpp +++ b/src/mongo/db/s/resharding/resharding_recipient_service.cpp @@ -36,7 +36,7 @@ #include "mongo/db/cancelable_operation_context.h" #include "mongo/db/catalog/rename_collection.h" #include "mongo/db/catalog_raii.h" -#include "mongo/db/concurrency/write_conflict_exception.h" +#include "mongo/db/concurrency/exception_util.h" #include "mongo/db/dbdirectclient.h" #include "mongo/db/dbhelpers.h" #include "mongo/db/ops/delete.h" @@ -56,6 +56,7 @@ #include "mongo/db/s/resharding/resharding_recipient_service_external_state.h" #include "mongo/db/s/resharding/resharding_server_parameters_gen.h" #include "mongo/db/s/shard_key_util.h" +#include "mongo/db/s/sharding_ddl_util.h" #include "mongo/db/s/sharding_state.h" #include "mongo/db/write_block_bypass.h" #include "mongo/executor/network_interface_factory.h" @@ -145,6 +146,8 @@ ReshardingRecipientService::RecipientStateMachine::RecipientStateMachine( _recipientCtx{recipientDoc.getMutableState()}, _donorShards{recipientDoc.getDonorShards()}, _cloneTimestamp{recipientDoc.getCloneTimestamp()}, + _timeIntervals{recipientDoc.getMetrics().get_value_or({})}, + _approxBytesToCopy{recipientDoc.getApproxBytesToCopy()}, _externalState{std::move(externalState)}, _startConfigTxnCloneAt{recipientDoc.getStartConfigTxnCloneTime()}, _markKilledExecutor(std::make_shared<ThreadPool>([] { @@ -166,6 +169,7 @@ ReshardingRecipientService::RecipientStateMachine::RecipientStateMachine( return donor.getShardId() == myShardId; }) != _donorShards.end(); }()) { + invariant(_externalState); } @@ -553,6 +557,11 @@ ReshardingRecipientService::RecipientStateMachine::_makeDataReplication(Operatio bool cloningDone) { invariant(_cloneTimestamp); + // We refresh the routing information for the source collection to ensure the + // ReshardingOplogApplier is making its decisions according to the chunk distribution after the + // sharding metadata was frozen. + _externalState->refreshCatalogCache(opCtx, _metadata.getSourceNss()); + auto myShardId = _externalState->myShardId(opCtx->getServiceContext()); auto sourceChunkMgr = _externalState->getShardedCollectionRoutingInfo(opCtx, _metadata.getSourceNss()); @@ -752,7 +761,7 @@ void ReshardingRecipientService::RecipientStateMachine::_cleanupReshardingCollec opCtx.get(), _metadata.getReshardingUUID(), _metadata.getSourceUUID(), _donorShards); if (aborted) { - resharding::data_copy::ensureCollectionDropped( + mongo::sharding_ddl_util::ensureCollectionDroppedNoChangeEvent( opCtx.get(), _metadata.getTempReshardingNss(), _metadata.getReshardingUUID()); } } @@ -808,27 +817,53 @@ void ReshardingRecipientService::RecipientStateMachine::_transitionToCloning( const CancelableOperationContextFactory& factory) { auto newRecipientCtx = _recipientCtx; newRecipientCtx.setState(RecipientStateEnum::kCloning); + auto cloningStartTime = getCurrentTime(); + + // Record cloning start time. + ReshardingMetricsTimeInterval interval; + interval.setStart(cloningStartTime); + _timeIntervals.setDocumentCopy(interval); + _transitionState(std::move(newRecipientCtx), boost::none, boost::none, factory); - _metrics()->startCopyingDocuments(getCurrentTime()); + _metrics()->startCopyingDocuments(cloningStartTime); } void ReshardingRecipientService::RecipientStateMachine::_transitionToApplying( const CancelableOperationContextFactory& factory) { auto newRecipientCtx = _recipientCtx; newRecipientCtx.setState(RecipientStateEnum::kApplying); + auto oplogApplicationStartTime = getCurrentTime(); + + // Record oplog application start time. + ReshardingMetricsTimeInterval interval; + interval.setStart(oplogApplicationStartTime); + _timeIntervals.setOplogApplication(interval); + + // Record document copy stop time. + ReshardingMetricsTimeInterval documentCopy{_timeIntervals.getDocumentCopy().get_value_or({})}; + documentCopy.setStop(oplogApplicationStartTime); + _timeIntervals.setDocumentCopy(documentCopy); + _transitionState(std::move(newRecipientCtx), boost::none, boost::none, factory); - auto currentTime = getCurrentTime(); - _metrics()->endCopyingDocuments(currentTime); - _metrics()->startApplyingOplogEntries(currentTime); + _metrics()->endCopyingDocuments(oplogApplicationStartTime); + _metrics()->startApplyingOplogEntries(oplogApplicationStartTime); } void ReshardingRecipientService::RecipientStateMachine::_transitionToStrictConsistency( const CancelableOperationContextFactory& factory) { auto newRecipientCtx = _recipientCtx; newRecipientCtx.setState(RecipientStateEnum::kStrictConsistency); + auto oplogApplicationStopTime = getCurrentTime(); + + // Record oplog application stop time + ReshardingMetricsTimeInterval oplogApplication{ + _timeIntervals.getOplogApplication().get_value_or({})}; + oplogApplication.setStop(oplogApplicationStopTime); + _timeIntervals.setOplogApplication(oplogApplication); + + _transitionState(std::move(newRecipientCtx), boost::none, boost::none, factory); - auto currentTime = getCurrentTime(); - _metrics()->endApplyingOplogEntries(currentTime); + _metrics()->endApplyingOplogEntries(oplogApplicationStopTime); } void ReshardingRecipientService::RecipientStateMachine::_transitionToError( @@ -980,6 +1015,9 @@ void ReshardingRecipientService::RecipientStateMachine::_updateRecipientDocument setBuilder.append(ReshardingRecipientDocument::kDonorShardsFieldName, donorShardsArrayBuilder.arr()); + + setBuilder.append(ReshardingRecipientDocument::kApproxBytesToCopyFieldName, + cloneDetails->approxBytesToCopy); } if (configStartTime) { @@ -987,6 +1025,8 @@ void ReshardingRecipientService::RecipientStateMachine::_updateRecipientDocument *configStartTime); } + setBuilder.append(ReshardingRecipientDocument::kMetricsFieldName, _timeIntervals.toBSON()); + setBuilder.doneFast(); } @@ -1004,6 +1044,7 @@ void ReshardingRecipientService::RecipientStateMachine::_updateRecipientDocument if (cloneDetails) { _cloneTimestamp = cloneDetails->cloneTimestamp; _donorShards = std::move(cloneDetails->donorShards); + _approxBytesToCopy = cloneDetails->approxBytesToCopy; } if (configStartTime) { @@ -1063,7 +1104,6 @@ ExecutorFuture<void> ReshardingRecipientService::RecipientStateMachine::_startMe const std::shared_ptr<executor::ScopedTaskExecutor>& executor, const CancellationToken& abortToken) { if (_recipientCtx.getState() > RecipientStateEnum::kAwaitingFetchTimestamp) { - _metrics()->onStepUp(ReshardingMetrics::Role::kRecipient); return _restoreMetricsWithRetry(executor, abortToken); } _metrics()->onStart(ReshardingMetrics::Role::kRecipient, getCurrentTime()); @@ -1073,7 +1113,6 @@ ExecutorFuture<void> ReshardingRecipientService::RecipientStateMachine::_startMe ExecutorFuture<void> ReshardingRecipientService::RecipientStateMachine::_restoreMetricsWithRetry( const std::shared_ptr<executor::ScopedTaskExecutor>& executor, const CancellationToken& abortToken) { - _metrics()->setRecipientState(_recipientCtx.getState()); return _retryingCancelableOpCtxFactory ->withAutomaticRetry( [this, executor, abortToken](const auto& factory) { _restoreMetrics(factory); }) @@ -1139,8 +1178,13 @@ void ReshardingRecipientService::RecipientStateMachine::_restoreMetrics( } } - _metrics()->restoreForCurrentOp( - documentCountCopied, documentBytesCopied, oplogEntriesFetched, oplogEntriesApplied); + _metrics()->onStepUp(_recipientCtx.getState(), + ReshardingMetrics::ReshardingRecipientCountsAndMetrics{documentCountCopied, + documentBytesCopied, + oplogEntriesFetched, + oplogEntriesApplied, + _approxBytesToCopy, + _timeIntervals}); } CancellationToken ReshardingRecipientService::RecipientStateMachine::_initAbortSource( diff --git a/src/mongo/db/s/resharding/resharding_recipient_service.h b/src/mongo/db/s/resharding/resharding_recipient_service.h index e1fe504a970..86d3236b9c9 100644 --- a/src/mongo/db/s/resharding/resharding_recipient_service.h +++ b/src/mongo/db/s/resharding/resharding_recipient_service.h @@ -288,6 +288,8 @@ private: RecipientShardContext _recipientCtx; std::vector<DonorShardFetchTimestamp> _donorShards; boost::optional<Timestamp> _cloneTimestamp; + ReshardingRecipientMetrics _timeIntervals; + boost::optional<int64_t> _approxBytesToCopy; const std::unique_ptr<RecipientStateMachineExternalState> _externalState; diff --git a/src/mongo/db/s/resharding/resharding_recipient_service_test.cpp b/src/mongo/db/s/resharding/resharding_recipient_service_test.cpp index 43ccab32bec..0b6c078603d 100644 --- a/src/mongo/db/s/resharding/resharding_recipient_service_test.cpp +++ b/src/mongo/db/s/resharding/resharding_recipient_service_test.cpp @@ -48,10 +48,13 @@ #include "mongo/db/s/resharding/resharding_recipient_service.h" #include "mongo/db/s/resharding/resharding_recipient_service_external_state.h" #include "mongo/db/s/resharding/resharding_service_test_helpers.h" +#include "mongo/db/s/sharding_ddl_util.h" #include "mongo/logv2/log.h" #include "mongo/unittest/death_test.h" +#include "mongo/util/clock_source_mock.h" #include "mongo/util/fail_point.h" + namespace mongo { namespace { @@ -208,6 +211,8 @@ public: */ class ReshardingRecipientServiceTest : public repl::PrimaryOnlyServiceMongoDTest { public: + ReshardingRecipientServiceTest() : PrimaryOnlyServiceMongoDTest(Options{}.useMockClock(true)) {} + using RecipientStateMachine = ReshardingRecipientService::RecipientStateMachine; std::unique_ptr<repl::PrimaryOnlyService> makeService(ServiceContext* serviceContext) override { @@ -222,7 +227,6 @@ public: repl::DropPendingCollectionReaper::set( serviceContext, std::make_unique<repl::DropPendingCollectionReaper>(storageMock.get())); repl::StorageInterface::set(serviceContext, std::move(storageMock)); - _controller = std::make_shared<RecipientStateTransitionController>(); _opObserverRegistry->addObserver(std::make_unique<RecipientOpObserverForTest>(_controller)); } @@ -263,7 +267,8 @@ public: const ReshardingRecipientDocument& recipientDoc) { CollectionOptions options; options.uuid = recipientDoc.getSourceUUID(); - resharding::data_copy::ensureCollectionDropped(opCtx, recipientDoc.getSourceNss()); + mongo::sharding_ddl_util::ensureCollectionDroppedNoChangeEvent(opCtx, + recipientDoc.getSourceNss()); resharding::data_copy::ensureCollectionExists(opCtx, recipientDoc.getSourceNss(), options); } @@ -496,69 +501,87 @@ DEATH_TEST_REGEX_F(ReshardingRecipientServiceTest, CommitFn, "4457001.*tripwire" TEST_F(ReshardingRecipientServiceTest, DropsTemporaryReshardingCollectionOnAbort) { auto metrics = ReshardingRecipientServiceTest::metrics(); for (bool isAlsoDonor : {false, true}) { - LOGV2(5551107, - "Running case", - "test"_attr = _agent.getTestName(), - "isAlsoDonor"_attr = isAlsoDonor); + for (bool waitForMetricsInitialized : {false, true}) { + LOGV2(5551107, + "Running case", + "test"_attr = _agent.getTestName(), + "isAlsoDonor"_attr = isAlsoDonor, + "waitForMetricsInitialized"_attr = waitForMetricsInitialized); + + boost::optional<PauseDuringStateTransitions> stateTransitionGuard; + if (waitForMetricsInitialized) { + std::vector<RecipientStateEnum> recipientStates{ + RecipientStateEnum::kDone, RecipientStateEnum::kCreatingCollection}; + stateTransitionGuard.emplace(controller(), recipientStates); + } else { + stateTransitionGuard.emplace(controller(), RecipientStateEnum::kDone); + } - boost::optional<PauseDuringStateTransitions> doneTransitionGuard; - doneTransitionGuard.emplace(controller(), RecipientStateEnum::kDone); + auto doc = makeStateDocument(isAlsoDonor); + auto instanceId = BSON(ReshardingRecipientDocument::kReshardingUUIDFieldName + << doc.getReshardingUUID()); - auto doc = makeStateDocument(isAlsoDonor); - auto instanceId = - BSON(ReshardingRecipientDocument::kReshardingUUIDFieldName << doc.getReshardingUUID()); + auto opCtx = makeOperationContext(); - auto opCtx = makeOperationContext(); + if (isAlsoDonor) { + // If the recipient is also a donor, the original collection should already exist on + // this shard. + createSourceCollection(opCtx.get(), doc); + } - if (isAlsoDonor) { - // If the recipient is also a donor, the original collection should already exist on - // this shard. - createSourceCollection(opCtx.get(), doc); - } + RecipientStateMachine::insertStateDocument(opCtx.get(), doc); + auto recipient = + RecipientStateMachine::getOrCreate(opCtx.get(), _service, doc.toBSON()); - RecipientStateMachine::insertStateDocument(opCtx.get(), doc); - auto recipient = RecipientStateMachine::getOrCreate(opCtx.get(), _service, doc.toBSON()); + notifyToStartCloning(opCtx.get(), *recipient, doc); + if (waitForMetricsInitialized) { + // Waiting for the metrics to be initialized here causes the second abort to occur + // before the metrics are initialized after the step up, thereby testing a different + // code path. + stateTransitionGuard->wait(RecipientStateEnum::kCreatingCollection); + } - notifyToStartCloning(opCtx.get(), *recipient, doc); - recipient->abort(false); + recipient->abort(false); - doneTransitionGuard->wait(RecipientStateEnum::kDone); - stepDown(); + stateTransitionGuard->wait(RecipientStateEnum::kDone); - ASSERT_EQ(recipient->getCompletionFuture().getNoThrow(), - ErrorCodes::InterruptedDueToReplStateChange); + stepDown(); - recipient.reset(); - stepUp(opCtx.get()); + ASSERT_EQ(recipient->getCompletionFuture().getNoThrow(), + ErrorCodes::InterruptedDueToReplStateChange); - auto maybeRecipient = RecipientStateMachine::lookup(opCtx.get(), _service, instanceId); - ASSERT_TRUE(bool(maybeRecipient)); - recipient = *maybeRecipient; + recipient.reset(); + stepUp(opCtx.get()); - doneTransitionGuard.reset(); - recipient->abort(false); + auto maybeRecipient = RecipientStateMachine::lookup(opCtx.get(), _service, instanceId); + ASSERT_TRUE(bool(maybeRecipient)); + recipient = *maybeRecipient; - ASSERT_OK(recipient->getCompletionFuture().getNoThrow()); - checkStateDocumentRemoved(opCtx.get()); + stateTransitionGuard.reset(); + recipient->abort(false); - if (isAlsoDonor) { - // Verify original collection still exists after aborting. - AutoGetCollection coll(opCtx.get(), doc.getSourceNss(), MODE_IS); - ASSERT_TRUE(bool(coll)); - ASSERT_EQ(coll->uuid(), doc.getSourceUUID()); - } + ASSERT_OK(recipient->getCompletionFuture().getNoThrow()); + checkStateDocumentRemoved(opCtx.get()); - // Verify the temporary collection no longer exists. - { - AutoGetCollection coll(opCtx.get(), doc.getTempReshardingNss(), MODE_IS); - ASSERT_FALSE(bool(coll)); + if (isAlsoDonor) { + // Verify original collection still exists after aborting. + AutoGetCollection coll(opCtx.get(), doc.getSourceNss(), MODE_IS); + ASSERT_TRUE(bool(coll)); + ASSERT_EQ(coll->uuid(), doc.getSourceUUID()); + } + + // Verify the temporary collection no longer exists. + { + AutoGetCollection coll(opCtx.get(), doc.getTempReshardingNss(), MODE_IS); + ASSERT_FALSE(bool(coll)); + } } } BSONObjBuilder result; metrics->serializeCumulativeOpMetrics(&result); - ASSERT_EQ(result.obj().getField("countReshardingFailures").numberLong(), 2); + ASSERT_LESS_THAN_OR_EQUALS(result.obj().getField("countReshardingFailures").numberLong(), 4); } TEST_F(ReshardingRecipientServiceTest, RenamesTemporaryReshardingCollectionWhenDone) { @@ -816,22 +839,33 @@ TEST_F(ReshardingRecipientServiceTest, RestoreMetricsAfterStepUp) { } // Step down before the transition to state can complete. stateTransitionsGuard.wait(state); - if (state == RecipientStateEnum::kStrictConsistency) { - auto currOp = recipient - ->reportForCurrentOp( - MongoProcessInterface::CurrentOpConnectionsMode::kExcludeIdle, - MongoProcessInterface::CurrentOpSessionsMode::kExcludeIdle) - .get(); + + dynamic_cast<ClockSourceMock*>(getServiceContext()->getFastClockSource()) + ->advance(Seconds(1)); + auto currOp = + recipient + ->reportForCurrentOp(MongoProcessInterface::CurrentOpConnectionsMode::kExcludeIdle, + MongoProcessInterface::CurrentOpSessionsMode::kExcludeIdle) + .get(); + + + if (state == RecipientStateEnum::kApplying) { + ASSERT_EQ(currOp.getField("totalApplyTimeElapsedSecs").Long(), 0); + ASSERT_EQ(currOp.getStringField("recipientState"), + RecipientState_serializer(RecipientStateEnum::kCloning)); + ASSERT_GT(currOp.getField("totalCopyTimeElapsedSecs").Long(), 0); + ASSERT_GT(currOp.getField("approxBytesToCopy").Long(), 0); + + } else if (state == RecipientStateEnum::kStrictConsistency) { ASSERT_EQ(currOp.getField("documentsCopied").Long(), 1L); ASSERT_EQ(currOp.getField("bytesCopied").Long(), (long)reshardedDoc.objsize()); ASSERT_EQ(currOp.getStringField("recipientState"), RecipientState_serializer(RecipientStateEnum::kApplying)); + ASSERT_GT(currOp.getField("totalCopyTimeElapsedSecs").Long(), 0); + ASSERT_GT(currOp.getField("totalApplyTimeElapsedSecs").Long(), 0); + ASSERT_GT(currOp.getField("approxBytesToCopy").Long(), 0); + } else if (state == RecipientStateEnum::kDone) { - auto currOp = recipient - ->reportForCurrentOp( - MongoProcessInterface::CurrentOpConnectionsMode::kExcludeIdle, - MongoProcessInterface::CurrentOpSessionsMode::kExcludeIdle) - .get(); ASSERT_EQ(currOp.getField("documentsCopied").Long(), 1L); ASSERT_EQ(currOp.getField("bytesCopied").Long(), (long)reshardedDoc.objsize()); ASSERT_EQ(currOp.getField("oplogEntriesFetched").Long(), @@ -840,7 +874,11 @@ TEST_F(ReshardingRecipientServiceTest, RestoreMetricsAfterStepUp) { oplogEntriesAppliedOnEachDonor * doc.getDonorShards().size()); ASSERT_EQ(currOp.getStringField("recipientState"), RecipientState_serializer(RecipientStateEnum::kStrictConsistency)); + ASSERT_GT(currOp.getField("totalCopyTimeElapsedSecs").Long(), 0); + ASSERT_GT(currOp.getField("totalApplyTimeElapsedSecs").Long(), 0); + ASSERT_GT(currOp.getField("approxBytesToCopy").Long(), 0); } + stepDown(); ASSERT_EQ(recipient->getCompletionFuture().getNoThrow(), diff --git a/src/mongo/db/s/resharding/resharding_txn_cloner.cpp b/src/mongo/db/s/resharding/resharding_txn_cloner.cpp index 7ab491286f3..764a32d68d5 100644 --- a/src/mongo/db/s/resharding/resharding_txn_cloner.cpp +++ b/src/mongo/db/s/resharding/resharding_txn_cloner.cpp @@ -40,7 +40,6 @@ #include "mongo/client/read_preference.h" #include "mongo/db/client.h" #include "mongo/db/concurrency/d_concurrency.h" -#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/curop.h" #include "mongo/db/logical_session_id.h" #include "mongo/db/namespace_string.h" diff --git a/src/mongo/db/s/resharding/resharding_util.cpp b/src/mongo/db/s/resharding/resharding_util.cpp index 27d319264dd..3f06e12a291 100644 --- a/src/mongo/db/s/resharding/resharding_util.cpp +++ b/src/mongo/db/s/resharding/resharding_util.cpp @@ -38,7 +38,7 @@ #include "mongo/bson/bsonobj.h" #include "mongo/bson/json.h" #include "mongo/bson/util/bson_extract.h" -#include "mongo/db/concurrency/write_conflict_exception.h" +#include "mongo/db/concurrency/exception_util.h" #include "mongo/db/exec/document_value/document.h" #include "mongo/db/namespace_string.h" #include "mongo/db/op_observer.h" |