diff options
Diffstat (limited to 'src/mongo/db/s/resharding')
47 files changed, 413 insertions, 982 deletions
diff --git a/src/mongo/db/s/resharding/document_source_resharding_add_resume_id.cpp b/src/mongo/db/s/resharding/document_source_resharding_add_resume_id.cpp index 2695ed842c0..75cb7be7049 100644 --- a/src/mongo/db/s/resharding/document_source_resharding_add_resume_id.cpp +++ b/src/mongo/db/s/resharding/document_source_resharding_add_resume_id.cpp @@ -101,7 +101,8 @@ StageConstraints DocumentSourceReshardingAddResumeId::constraints( ChangeStreamRequirement::kDenylist); } -Value DocumentSourceReshardingAddResumeId::serialize(const SerializationOptions& opts) const { +Value DocumentSourceReshardingAddResumeId::serialize( + boost::optional<ExplainOptions::Verbosity> explain) const { return Value(Document{{kStageName, Value(Document{})}}); } diff --git a/src/mongo/db/s/resharding/document_source_resharding_add_resume_id.h b/src/mongo/db/s/resharding/document_source_resharding_add_resume_id.h index 4fb27980c68..31cbd97c694 100644 --- a/src/mongo/db/s/resharding/document_source_resharding_add_resume_id.h +++ b/src/mongo/db/s/resharding/document_source_resharding_add_resume_id.h @@ -53,7 +53,7 @@ public: DocumentSource::GetModPathsReturn getModifiedPaths() const final; - Value serialize(const SerializationOptions& opts = SerializationOptions{}) const final override; + Value serialize(boost::optional<ExplainOptions::Verbosity> explain) const; StageConstraints constraints(Pipeline::SplitState pipeState) const final; 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 8260cf3e4cc..9d412147b7f 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 @@ -112,7 +112,7 @@ StageConstraints DocumentSourceReshardingIterateTransaction::constraints( } Value DocumentSourceReshardingIterateTransaction::serialize( - const SerializationOptions& opts) const { + boost::optional<ExplainOptions::Verbosity> explain) const { return Value( Document{{kStageName, Value(Document{{kIncludeCommitTransactionTimestampFieldName, @@ -134,7 +134,7 @@ DepsTracker::State DocumentSourceReshardingIterateTransaction::getDependencies( DocumentSource::GetModPathsReturn DocumentSourceReshardingIterateTransaction::getModifiedPaths() const { - return {DocumentSource::GetModPathsReturn::Type::kAllPaths, OrderedPathSet{}, {}}; + return {DocumentSource::GetModPathsReturn::Type::kAllPaths, std::set<std::string>{}, {}}; } DocumentSource::GetNextResult DocumentSourceReshardingIterateTransaction::doGetNext() { diff --git a/src/mongo/db/s/resharding/document_source_resharding_iterate_transaction.h b/src/mongo/db/s/resharding/document_source_resharding_iterate_transaction.h index 8c792116e6c..9589cb64a08 100644 --- a/src/mongo/db/s/resharding/document_source_resharding_iterate_transaction.h +++ b/src/mongo/db/s/resharding/document_source_resharding_iterate_transaction.h @@ -66,7 +66,7 @@ public: DocumentSource::GetModPathsReturn getModifiedPaths() const final; - Value serialize(const SerializationOptions& opts = SerializationOptions{}) const final override; + Value serialize(boost::optional<ExplainOptions::Verbosity> explain) const; StageConstraints constraints(Pipeline::SplitState pipeState) const final; 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 157876391d3..f8a186e3bcd 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 @@ -90,11 +90,12 @@ StageConstraints DocumentSourceReshardingOwnershipMatch::constraints( ChangeStreamRequirement::kDenylist); } -Value DocumentSourceReshardingOwnershipMatch::serialize(const SerializationOptions& opts) const { +Value DocumentSourceReshardingOwnershipMatch::serialize( + boost::optional<ExplainOptions::Verbosity> explain) const { return Value{Document{{kStageName, DocumentSourceReshardingOwnershipMatchSpec( _recipientShardId, _reshardingKey.getKeyPattern()) - .toBSON(opts)}}}; + .toBSON()}}}; } DepsTracker::State DocumentSourceReshardingOwnershipMatch::getDependencies( @@ -108,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, OrderedPathSet{}, {}}; + return {DocumentSource::GetModPathsReturn::Type::kFiniteSet, std::set<std::string>{}, {}}; } DocumentSource::GetNextResult DocumentSourceReshardingOwnershipMatch::doGetNext() { diff --git a/src/mongo/db/s/resharding/document_source_resharding_ownership_match.h b/src/mongo/db/s/resharding/document_source_resharding_ownership_match.h index b7da07a5a57..7a6db2bc125 100644 --- a/src/mongo/db/s/resharding/document_source_resharding_ownership_match.h +++ b/src/mongo/db/s/resharding/document_source_resharding_ownership_match.h @@ -58,7 +58,7 @@ public: DocumentSource::GetModPathsReturn getModifiedPaths() const final; - Value serialize(const SerializationOptions& opts = SerializationOptions{}) const final override; + Value serialize(boost::optional<ExplainOptions::Verbosity> explain) const final; StageConstraints constraints(Pipeline::SplitState pipeState) const final; diff --git a/src/mongo/db/s/resharding/recipient_document.idl b/src/mongo/db/s/resharding/recipient_document.idl index 1eda4620b8c..e3128de0db5 100644 --- a/src/mongo/db/s/resharding/recipient_document.idl +++ b/src/mongo/db/s/resharding/recipient_document.idl @@ -82,11 +82,6 @@ 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_agg_test.cpp b/src/mongo/db/s/resharding/resharding_agg_test.cpp index 99ad259c7d0..f2c5345e907 100644 --- a/src/mongo/db/s/resharding/resharding_agg_test.cpp +++ b/src/mongo/db/s/resharding/resharding_agg_test.cpp @@ -239,7 +239,6 @@ repl::DurableOplogEntry makeApplyOpsOplog(std::vector<BSONObj> operations, {}, UUID::gen(), false /* fromMigrate */, - boost::none, // checkExistenceForDiffInsert 0 /* version */, applyOpsBuilder.obj(), /* o */ boost::none, /* o2 */ @@ -1452,7 +1451,6 @@ TEST_F(ReshardingAggWithStorageTest, RetryableFindAndModifyWithImageLookup) { kCrudNs, kCrudUUID, false /* fromMigrate */, - boost::none, // checkExistenceForDiffInsert 0 /* version */, BSON("$set" << BSON("y" << 1)), /* o1 */ BSON("_id" << 2), /* o2 */ diff --git a/src/mongo/db/s/resharding/resharding_collection_cloner.cpp b/src/mongo/db/s/resharding/resharding_collection_cloner.cpp index e51d2e2ed1f..9132438053a 100644 --- a/src/mongo/db/s/resharding/resharding_collection_cloner.cpp +++ b/src/mongo/db/s/resharding/resharding_collection_cloner.cpp @@ -29,6 +29,8 @@ #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> @@ -37,6 +39,7 @@ #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" @@ -274,7 +277,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 error to allow the collection metadata information to be + // and retry once on a StaleConfig exception 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 5700f0326ae..374dcd6538f 100644 --- a/src/mongo/db/s/resharding/resharding_coordinator_commit_monitor.cpp +++ b/src/mongo/db/s/resharding/resharding_coordinator_commit_monitor.cpp @@ -164,21 +164,13 @@ CoordinatorCommitMonitor::queryRemainingOperationTimeForRecipients() const { uassertStatusOKWithContext(status, errorContext); const auto remainingTime = extractOperationRemainingTime(shardResponse.data); - - // 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(); + // 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 (remainingTime.value() > maxRemainingTime) { - maxRemainingTime = remainingTime.value(); + if (remainingTime && remainingTime.get() > maxRemainingTime) { + maxRemainingTime = remainingTime.get(); } } @@ -211,20 +203,12 @@ ExecutorFuture<void> CoordinatorCommitMonitor::_makeFuture() const { "Encountered an error while querying recipients, will retry shortly", "error"_attr = status); - // 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()}; + return RemainingOperationTimes{Milliseconds(0), Milliseconds::max()}; }) .then([this, anchor = shared_from_this()](RemainingOperationTimes remainingTimes) { auto metrics = ReshardingMetrics::get(cc().getServiceContext()); - // 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)); + metrics->setMinRemainingOperationTime(remainingTimes.min); + metrics->setMaxRemainingOperationTime(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 2fe3075f1fc..e7f90cc41fa 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,8 +92,6 @@ protected: void tearDown() override; void mockCommandForRecipients(Milliseconds remainingOperationTime); - void mockOmitRemainingMillisForRecipients(); - void mockOmitRemainingMillisForOneRecipient(); void mockRemaingOperationTimesCommandForRecipients( CoordinatorCommitMonitor::RemainingOperationTimes remainingOperationTimes); @@ -182,31 +180,6 @@ 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; @@ -291,20 +264,6 @@ 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 ea3451f1c4a..85c14b6478f 100644 --- a/src/mongo/db/s/resharding/resharding_coordinator_service.cpp +++ b/src/mongo/db/s/resharding/resharding_coordinator_service.cpp @@ -37,7 +37,6 @@ #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" @@ -95,7 +94,6 @@ 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()); @@ -111,14 +109,14 @@ Date_t getCurrentTime() { return svcCtx->getFastClockSource()->now(); } -void assertNumDocsMatchedEqualsExpected(const BatchedCommandRequest& request, - const BSONObj& response, - int expected) { - auto numDocsMatched = response.getIntField("n"); +void assertNumDocsModifiedMatchesExpected(const BatchedCommandRequest& request, + const BSONObj& response, + int expected) { + auto numDocsModified = response.getIntField("n"); uassert(5030401, str::stream() << "Expected to match " << expected << " docs, but only matched " - << numDocsMatched << " for write request " << request.toString(), - expected == numDocsMatched); + << numDocsModified << " for write request " << request.toString(), + expected == numDocsModified); } void appendShardEntriesToSetBuilder(const ReshardingCoordinatorDocument& coordinatorDoc, @@ -217,15 +215,15 @@ void writeToCoordinatorStateNss(OperationContext* opCtx, } }()); - auto expectedNumMatched = (request.getBatchType() == BatchedCommandRequest::BatchType_Insert) + auto expectedNumModified = (request.getBatchType() == BatchedCommandRequest::BatchType_Insert) ? boost::none : boost::make_optional(1); auto res = ShardingCatalogManager::get(opCtx)->writeToConfigDocumentInTxn( opCtx, NamespaceString::kConfigReshardingOperationsNamespace, request, txnNumber); - if (expectedNumMatched) { - assertNumDocsMatchedEqualsExpected(request, res, *expectedNumMatched); + if (expectedNumModified) { + assertNumDocsModifiedMatchesExpected(request, res, *expectedNumModified); } } @@ -387,7 +385,7 @@ void updateConfigCollectionsForOriginalNss(OperationContext* opCtx, auto res = ShardingCatalogManager::get(opCtx)->writeToConfigDocumentInTxn( opCtx, CollectionType::ConfigNS, request, txnNumber); - assertNumDocsMatchedEqualsExpected(request, res, 1 /* expected */); + assertNumDocsModifiedMatchesExpected(request, res, 1 /* expected */); } void writeToConfigCollectionsForTempNss(OperationContext* opCtx, @@ -478,15 +476,15 @@ void writeToConfigCollectionsForTempNss(OperationContext* opCtx, } }()); - auto expectedNumMatched = (request.getBatchType() == BatchedCommandRequest::BatchType_Insert) + auto expectedNumModified = (request.getBatchType() == BatchedCommandRequest::BatchType_Insert) ? boost::none : boost::make_optional(1); auto res = ShardingCatalogManager::get(opCtx)->writeToConfigDocumentInTxn( opCtx, CollectionType::ConfigNS, request, txnNumber); - if (expectedNumMatched) { - assertNumDocsMatchedEqualsExpected(request, res, *expectedNumMatched); + if (expectedNumModified) { + assertNumDocsModifiedMatchesExpected(request, res, *expectedNumModified); } } @@ -507,33 +505,43 @@ 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 and specified tag documents. - resharding::removeChunkDocs(opCtx, 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); + 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. */ @@ -584,14 +592,28 @@ CollectionType createTempReshardingCollectionType( return collType; } -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(); +void cleanupSourceConfigCollections(OperationContext* opCtx, + const ReshardingCoordinatorDocument& coordinatorDoc) { + using Doc = Document; + using Arr = std::vector<Value>; + using V = Value; - uassertStatusOK(catalogClient->removeConfigDocuments( - opCtx, ChunkType::ConfigNS, chunksQuery, kMajorityWriteConcern)); + 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); + + removeChunkAndTagsDocs(opCtx, removeTagsQuery, coordinatorDoc.getSourceUUID()); } void writeDecisionPersistedState(OperationContext* opCtx, @@ -600,49 +622,21 @@ void writeDecisionPersistedState(OperationContext* opCtx, Timestamp newCollectionTimestamp) { // No need to bump originalNss version because its epoch will be changed. - executeMetadataChangesInTxn( - opCtx, - [&coordinatorDoc, &newCollectionEpoch, &newCollectionTimestamp](OperationContext* opCtx, - TxnNumber txnNumber) { - // Update the config.reshardingOperations entry - writeToCoordinatorStateNss(opCtx, coordinatorDoc, txnNumber); + executeMetadataChangesInTxn(opCtx, [&](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); - - // 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); + // Remove the config.collections entry for the temporary collection + writeToConfigCollectionsForTempNss( + opCtx, coordinatorDoc, boost::none, boost::none, 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); + // 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 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); + updateChunkAndTagsDocsForTempNss(opCtx, coordinatorDoc, newCollectionEpoch, txnNumber); + }); } void insertCoordDocAndChangeOrigCollEntry(OperationContext* opCtx, @@ -754,21 +748,16 @@ void writeStateTransitionAndCatalogUpdatesThenBumpShardVersions( ShardingCatalogClient::kLocalWriteConcern); } -ReshardingCoordinatorDocument removeCoordinatorDocAndReshardingFields( - OperationContext* opCtx, - const ReshardingCoordinatorDocument& coordinatorDoc, - boost::optional<Status> abortReason) { +void removeCoordinatorDocAndReshardingFields(OperationContext* opCtx, + const ReshardingCoordinatorDocument& coordinatorDoc, + boost::optional<Status> abortReason) { // If the coordinator needs to abort and isn't in kInitializing, additional collections need to // be cleaned up in the final transaction. Otherwise, cleanup for abort and success are the // same. const bool wasDecisionPersisted = - coordinatorDoc.getState() >= CoordinatorStateEnum::kCommitting; + coordinatorDoc.getState() == CoordinatorStateEnum::kCommitting; invariant((wasDecisionPersisted && !abortReason) || abortReason); - if (coordinatorDoc.getState() > CoordinatorStateEnum::kCommitting) { - return coordinatorDoc; - } - ReshardingCoordinatorDocument updatedCoordinatorDoc = coordinatorDoc; updatedCoordinatorDoc.setState(CoordinatorStateEnum::kDone); emplaceTruncatedAbortReasonIfExists(updatedCoordinatorDoc, abortReason); @@ -802,7 +791,6 @@ ReshardingCoordinatorDocument removeCoordinatorDocAndReshardingFields( opCtx, updatedCoordinatorDoc, boost::none, boost::none, txnNumber); }, ShardingCatalogClient::kLocalWriteConcern); - return updatedCoordinatorDoc; } } // namespace resharding @@ -855,12 +843,7 @@ ReshardingCoordinatorExternalStateImpl::calculateParticipantShardsAndChunks( // The database primary must always be a recipient to ensure it ends up with consistent // collection metadata. - const auto dbPrimaryShard = - uassertStatusOK(Grid::get(opCtx)->catalogCache()->getDatabaseWithRefresh( - opCtx, coordinatorDoc.getSourceNss().db())) - ->getPrimary(); - - recipientShardIds.emplace(dbPrimaryShard); + recipientShardIds.emplace(cm.dbPrimary()); if (const auto& chunks = coordinatorDoc.getPresetReshardedChunks()) { auto version = calculateChunkVersionForInitialChunks(opCtx); @@ -1089,7 +1072,6 @@ ReshardingCoordinatorService::ReshardingCoordinator::_tellAllParticipantsReshard _cancelableOpCtxFactory.emplace(_ctHolder->getStepdownToken(), _markKilledExecutor); }) - .then([this] { return _waitForMajority(_ctHolder->getStepdownToken()); }) .then([this, executor]() { pauseBeforeTellDonorToRefresh.pauseWhileSet(); _establishAllDonorsAsParticipants(executor); @@ -1120,7 +1102,8 @@ ExecutorFuture<void> ReshardingCoordinatorService::ReshardingCoordinator::_initi return resharding::WithAutomaticRetry([this, executor] { return ExecutorFuture<void>(**executor) .then([this, executor] { _insertCoordDocAndChangeOrigCollEntry(); }) - .then([this, executor] { _calculateParticipantsAndChunksThenWriteToDisk(); }); + .then([this, executor] { _calculateParticipantsAndChunksThenWriteToDisk(); }) + .then([this] { return _waitForMajority(_ctHolder->getAbortToken()); }); }) .onTransientError([](const Status& status) { LOGV2(5093703, @@ -1259,11 +1242,33 @@ ReshardingCoordinatorService::ReshardingCoordinator::_commitAndFinishReshardOper const ReshardingCoordinatorDocument& updatedCoordinatorDoc) noexcept { return resharding::WithAutomaticRetry([this, executor, updatedCoordinatorDoc] { return ExecutorFuture<void>(**executor) - .then( - [this, executor, updatedCoordinatorDoc] { _commit(updatedCoordinatorDoc); }); + .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); + }); }) .onTransientError([](const Status& status) { - LOGV2(7698801, + LOGV2(5093705, "Resharding coordinator encountered transient error while committing", "error"_attr = status); }) @@ -1271,83 +1276,25 @@ ReshardingCoordinatorService::ReshardingCoordinator::_commitAndFinishReshardOper .until<Status>([](const Status& status) { return status.isOK(); }) .on(**executor, _ctHolder->getStepdownToken()) .onError([this, executor](Status status) { - if (status == ErrorCodes::TransactionTooLargeForCache) { - return _onAbortCoordinatorAndParticipants(executor, status); + { + auto opCtx = _cancelableOpCtxFactory->makeOperationContext(&cc()); + reshardingPauseCoordinatorBeforeStartingErrorFlow.pauseWhileSet(opCtx.get()); } - 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 { - pauseBeforeCTHolderInitialization.pauseWhileSet(); - - auto abortCalled = [&] { - stdx::lock_guard<Latch> lk(_abortCalledMutex); - _ctHolder = std::make_unique<CoordinatorCancellationTokenHolder>(stepdownToken); - return _abortCalled; - }(); - - if (abortCalled) { - _ctHolder->abort(); - } - + _ctHolder = std::make_unique<CoordinatorCancellationTokenHolder>(stepdownToken); _markKilledExecutor->startup(); _cancelableOpCtxFactory.emplace(_ctHolder->getAbortToken(), _markKilledExecutor); @@ -1358,15 +1305,8 @@ SemiFuture<void> ReshardingCoordinatorService::ReshardingCoordinator::run( }) .onCompletion([this, executor](Status status) { auto opCtx = _cancelableOpCtxFactory->makeOperationContext(&cc()); - 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(); - }); + reshardingPauseCoordinatorBeforeCompletion.pauseWhileSetAndNotCanceled( + opCtx.get(), _ctHolder->getStepdownToken()); { auto lg = stdx::lock_guard(_fulfillmentMutex); @@ -1407,8 +1347,6 @@ SemiFuture<void> ReshardingCoordinatorService::ReshardingCoordinator::run( ->onStepDown(ReshardingMetrics::Role::kCoordinator); } - _logStatsOnCompletion(status.isOK()); - if (!status.isOK()) { { auto lg = stdx::lock_guard(_fulfillmentMutex); @@ -1505,15 +1443,7 @@ ReshardingCoordinatorService::ReshardingCoordinator::_onAbortCoordinatorAndParti } void ReshardingCoordinatorService::ReshardingCoordinator::abort() { - auto ctHolderInitialized = [&] { - stdx::lock_guard<Latch> lk(_abortCalledMutex); - _abortCalled = true; - return !(_ctHolder == nullptr); - }(); - - if (ctHolderInitialized) { - _ctHolder->abort(); - } + _ctHolder->abort(); } boost::optional<BSONObj> ReshardingCoordinatorService::ReshardingCoordinator::reportForCurrentOp( @@ -1612,12 +1542,10 @@ void ReshardingCoordinatorService::ReshardingCoordinator:: // Remove the presetReshardedChunks and zones from the coordinator document to reduce // the possibility of the document reaching the BSONObj size constraint. - ShardKeyPattern shardKey(updatedCoordinatorDoc.getReshardingKey()); std::vector<BSONObj> zones; if (updatedCoordinatorDoc.getZones()) { zones = buildTagsDocsFromZones(updatedCoordinatorDoc.getTempReshardingNss(), - *updatedCoordinatorDoc.getZones(), - shardKey); + *updatedCoordinatorDoc.getZones()); } updatedCoordinatorDoc.setPresetReshardedChunks(boost::none); updatedCoordinatorDoc.setZones(boost::none); @@ -1811,11 +1739,11 @@ ReshardingCoordinatorService::ReshardingCoordinator::_awaitAllRecipientsInStrict .thenRunOn(**executor); } -void ReshardingCoordinatorService::ReshardingCoordinator::_commit( +Future<void> ReshardingCoordinatorService::ReshardingCoordinator::_commit( const ReshardingCoordinatorDocument& coordinatorDoc) { if (_coordinatorDoc.getState() > CoordinatorStateEnum::kBlockingWrites) { invariant(_coordinatorDoc.getState() != CoordinatorStateEnum::kAborting); - return; + return Status::OK(); } ReshardingCoordinatorDocument updatedCoordinatorDoc = coordinatorDoc; @@ -1841,6 +1769,8 @@ void ReshardingCoordinatorService::ReshardingCoordinator::_commit( // Update the in memory state installCoordinatorDoc(opCtx.get(), updatedCoordinatorDoc); + + return Status::OK(); } ExecutorFuture<void> @@ -2073,22 +2003,4 @@ void ReshardingCoordinatorService::ReshardingCoordinator::_updateChunkImbalanceM } } -void ReshardingCoordinatorService::ReshardingCoordinator::_logStatsOnCompletion(bool success) { - BSONObjBuilder builder; - BSONObjBuilder statsBuilder; - builder.append("uuid", _coordinatorDoc.getReshardingUUID().toBSON()); - builder.append("status", success ? "success" : "failed"); - statsBuilder.append("ns", _coordinatorDoc.getSourceNss().toString()); - statsBuilder.append("sourceUUID", _coordinatorDoc.getSourceUUID().toBSON()); - statsBuilder.append("newUUID", _coordinatorDoc.getReshardingUUID().toBSON()); - statsBuilder.append("newShardKey", _coordinatorDoc.getReshardingKey().toBSON()); - if (_coordinatorDoc.getStartTime()) { - statsBuilder.append("startTime", *_coordinatorDoc.getStartTime()); - } - statsBuilder.append("endTime", getCurrentTime()); - - builder.append("statistics", statsBuilder.obj()); - LOGV2(7763800, "Resharding complete", "info"_attr = builder.obj()); -} - } // namespace mongo diff --git a/src/mongo/db/s/resharding/resharding_coordinator_service.h b/src/mongo/db/s/resharding/resharding_coordinator_service.h index 694070c126c..a24569ecc44 100644 --- a/src/mongo/db/s/resharding/resharding_coordinator_service.h +++ b/src/mongo/db/s/resharding/resharding_coordinator_service.h @@ -50,17 +50,14 @@ CollectionType createTempReshardingCollectionType( const ChunkVersion& chunkVersion, const BSONObj& collation); -void removeChunkDocs(OperationContext* opCtx, const UUID& collUUID); +void cleanupSourceConfigCollections(OperationContext* opCtx, + const ReshardingCoordinatorDocument& coordinatorDoc); 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); @@ -72,10 +69,9 @@ void writeParticipantShardsAndTempCollInfo(OperationContext* opCtx, void writeStateTransitionAndCatalogUpdatesThenBumpShardVersions( OperationContext* opCtx, const ReshardingCoordinatorDocument& coordinatorDoc); -ReshardingCoordinatorDocument removeCoordinatorDocAndReshardingFields( - OperationContext* opCtx, - const ReshardingCoordinatorDocument& coordinatorDoc, - boost::optional<Status> abortReason = boost::none); +void removeCoordinatorDocAndReshardingFields(OperationContext* opCtx, + const ReshardingCoordinatorDocument& coordinatorDoc, + boost::optional<Status> abortReason = boost::none); } // namespace resharding class ReshardingCoordinatorExternalState { @@ -408,10 +404,11 @@ 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'. */ - void _commit(const ReshardingCoordinatorDocument& updatedDoc); + Future<void> _commit(const ReshardingCoordinatorDocument& updatedDoc); /** * Waits on _reshardingCoordinatorObserver to notify that: @@ -501,11 +498,6 @@ private: // Waits for majority replication of the latest opTime unless token is cancelled. SemiFuture<void> _waitForMajority(const CancellationToken& token); - /** - * Print a log containing the information of this resharding operation. - */ - void _logStatsOnCompletion(bool success); - // The unique key for a given resharding operation. InstanceID is an alias for BSONObj. The // value of this is the UUID that will be used as the collection UUID for the new sharded // collection. The object looks like: {_id: 'reshardingUUID'} @@ -542,13 +534,6 @@ 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. @@ -568,10 +553,6 @@ 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 33511df9fd4..6a5197b4c41 100644 --- a/src/mongo/db/s/resharding/resharding_coordinator_service_test.cpp +++ b/src/mongo/db/s/resharding/resharding_coordinator_service_test.cpp @@ -128,6 +128,10 @@ class ReshardingCoordinatorServiceTest : public ConfigServerTestFixture { public: using ReshardingCoordinator = ReshardingCoordinatorService::ReshardingCoordinator; + // TODO (SERVER-65302): Use wiredTiger. + ReshardingCoordinatorServiceTest() + : ConfigServerTestFixture(Options{}.engine("ephemeralForTest")) {} + std::unique_ptr<repl::PrimaryOnlyService> makeService(ServiceContext* serviceContext) { return std::make_unique<ReshardingCoordinatorServiceForTest>(serviceContext); } @@ -164,6 +168,7 @@ public: dynamic_cast<OpObserverRegistry*>(getServiceContext()->getOpObserver()); invariant(_opObserverRegistry); + _opObserverRegistry->addObserver(std::make_unique<OpObserverImpl>()); _opObserverRegistry->addObserver(std::make_unique<ReshardingOpObserver>()); _opObserverRegistry->addObserver( std::make_unique<CoordinatorOpObserverForTest>(_controller)); @@ -213,9 +218,14 @@ public: std::shared_ptr<ReshardingCoordinatorService::ReshardingCoordinator> getCoordinatorIfExists( OperationContext* opCtx, repl::PrimaryOnlyService::InstanceID instanceId) { - auto [coordinatorOpt, _] = ReshardingCoordinatorService::ReshardingCoordinator::lookup( + auto coordinatorOpt = ReshardingCoordinatorService::ReshardingCoordinator::lookup( opCtx, _service, instanceId); - return coordinatorOpt ? *coordinatorOpt : nullptr; + if (!coordinatorOpt) { + return nullptr; + } + + auto coordinator = *coordinatorOpt; + return coordinator ? coordinator : nullptr; } ReshardingCoordinatorDocument getCoordinatorDoc(OperationContext* opCtx) { @@ -913,16 +923,5 @@ 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 521cb84471e..a8fb4d83889 100644 --- a/src/mongo/db/s/resharding/resharding_coordinator_test.cpp +++ b/src/mongo/db/s/resharding/resharding_coordinator_test.cpp @@ -584,8 +584,7 @@ protected: std::vector<BSONObj> zones; if (expectedCoordinatorDoc.getZones()) { zones = buildTagsDocsFromZones(expectedCoordinatorDoc.getTempReshardingNss(), - *expectedCoordinatorDoc.getZones(), - _newShardKey); + *expectedCoordinatorDoc.getZones()); } expectedCoordinatorDoc.setZones(boost::none); expectedCoordinatorDoc.setPresetReshardedChunks(boost::none); @@ -637,7 +636,7 @@ protected: ReshardingCoordinatorDocument expectedCoordinatorDoc, std::vector<ChunkType> expectedChunks, std::vector<TagsType> expectedZones) { - removeChunkDocs(opCtx, expectedCoordinatorDoc.getSourceUUID()); + cleanupSourceConfigCollections(opCtx, expectedCoordinatorDoc); // Check that chunks and tags entries previously under the temporary namespace have been // correctly updated to the original namespace @@ -648,13 +647,10 @@ protected: void removeCoordinatorDocAndReshardingFieldsExpectSuccess( OperationContext* opCtx, const ReshardingCoordinatorDocument& coordinatorDoc) { - auto updatedCoordinatorDoc = removeCoordinatorDocAndReshardingFields(opCtx, coordinatorDoc); - ASSERT_EQUALS(int(updatedCoordinatorDoc.getState()), int(CoordinatorStateEnum::kDone)); + removeCoordinatorDocAndReshardingFields(opCtx, coordinatorDoc); - // Check that the on disk document is same as the in memory document returned above. auto expectedCoordinatorDoc = coordinatorDoc; expectedCoordinatorDoc.setState(CoordinatorStateEnum::kDone); - ASSERT_BSONOBJ_EQ(updatedCoordinatorDoc.toBSON(), expectedCoordinatorDoc.toBSON()); // Check that the entry is removed from config.reshardingOperations DBDirectClient client(opCtx); @@ -665,12 +661,12 @@ protected: // Check that the resharding fields are removed from the config.collections entry and // allowMigrations is set back to true. auto expectedOriginalCollType = makeOriginalCollectionCatalogEntry( - updatedCoordinatorDoc, + expectedCoordinatorDoc, boost::none, _finalEpoch, opCtx->getServiceContext()->getPreciseClockSource()->now()); assertOriginalCollectionCatalogEntryMatchesExpected( - opCtx, expectedOriginalCollType, updatedCoordinatorDoc); + opCtx, expectedOriginalCollType, expectedCoordinatorDoc); } void transitionToErrorExpectSuccess(ErrorCodes::Error errorCode) { 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 8635e389cf4..9893b2b0f2e 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/exception_util.h" +#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/curop.h" #include "mongo/db/dbhelpers.h" #include "mongo/db/namespace_string.h" @@ -44,7 +44,6 @@ #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" @@ -72,6 +71,28 @@ 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, @@ -98,11 +119,11 @@ void ensureOplogCollectionsDropped(OperationContext* opCtx, // Drop the conflict stash collection for this donor. auto stashNss = getLocalConflictStashNamespace(sourceUUID, donor.getShardId()); - mongo::sharding_ddl_util::ensureCollectionDroppedNoChangeEvent(opCtx, stashNss); + ensureCollectionDropped(opCtx, stashNss); // Drop the oplog buffer collection for this donor. auto oplogBufferNss = getLocalOplogBufferNamespace(sourceUUID, donor.getShardId()); - mongo::sharding_ddl_util::ensureCollectionDroppedNoChangeEvent(opCtx, oplogBufferNss); + ensureCollectionDropped(opCtx, oplogBufferNss); } } @@ -282,8 +303,7 @@ void updateSessionRecord(OperationContext* opCtx, BSONObj o2Field, std::vector<StmtId> stmtIds, boost::optional<repl::OpTime> preImageOpTime, - boost::optional<repl::OpTime> postImageOpTime, - NamespaceString sourceNss) { + boost::optional<repl::OpTime> postImageOpTime) { invariant(opCtx->getLogicalSessionId()); invariant(opCtx->getTxnNumber()); @@ -297,7 +317,7 @@ void updateSessionRecord(OperationContext* opCtx, oplogEntry.setOpType(repl::OpTypeEnum::kNoop); oplogEntry.setObject(SessionCatalogMigration::kSessionOplogTag); oplogEntry.setObject2(std::move(o2Field)); - oplogEntry.setNss(std::move(sourceNss)); + oplogEntry.setNss({}); oplogEntry.setSessionId(sessionId); oplogEntry.setTxnNumber(txnNumber); oplogEntry.setStatementIds(stmtIds); 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 af24152aaf2..9f2a332ef6c 100644 --- a/src/mongo/db/s/resharding/resharding_data_copy_util.h +++ b/src/mongo/db/s/resharding/resharding_data_copy_util.h @@ -60,6 +60,16 @@ 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 @@ -144,14 +154,13 @@ void updateSessionRecord(OperationContext* opCtx, BSONObj o2Field, std::vector<StmtId> stmtIds, boost::optional<repl::OpTime> preImageOpTime, - boost::optional<repl::OpTime> postImageOpTime, - NamespaceString sourceNss); + boost::optional<repl::OpTime> postImageOpTime); /** * Calls and returns the value from the supplied lambda function. * - * 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. + * 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. */ template <typename Callable> auto withOneStaleConfigRetry(OperationContext* opCtx, Callable&& callable) { diff --git a/src/mongo/db/s/resharding/resharding_data_replication.cpp b/src/mongo/db/s/resharding/resharding_data_replication.cpp index 5223b1c5c0b..a5918e3ea77 100644 --- a/src/mongo/db/s/resharding/resharding_data_replication.cpp +++ b/src/mongo/db/s/resharding/resharding_data_replication.cpp @@ -475,9 +475,4 @@ ReshardingDonorOplogId ReshardingDataReplication::getOplogApplierResumeId( : ReshardingDonorOplogId{minFetchTimestamp, minFetchTimestamp}; } -ReshardingDataReplication::~ReshardingDataReplication() { - shutdown(); - join(); -} - } // namespace mongo diff --git a/src/mongo/db/s/resharding/resharding_data_replication.h b/src/mongo/db/s/resharding/resharding_data_replication.h index 61355447a40..a2705abd6f6 100644 --- a/src/mongo/db/s/resharding/resharding_data_replication.h +++ b/src/mongo/db/s/resharding/resharding_data_replication.h @@ -136,7 +136,6 @@ private: struct TrustedInitTag {}; public: - virtual ~ReshardingDataReplication(); static std::unique_ptr<ReshardingDataReplicationInterface> make( OperationContext* opCtx, ReshardingMetrics* metrics, 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 0c26b0acd02..e942dcd139f 100644 --- a/src/mongo/db/s/resharding/resharding_destined_recipient_test.cpp +++ b/src/mongo/db/s/resharding/resharding_destined_recipient_test.cpp @@ -131,16 +131,14 @@ public: StaticCatalogClient(std::vector<ShardType> shards) : _shards(std::move(shards)) {} StatusWith<repl::OpTimeWith<std::vector<ShardType>>> getAllShards( - OperationContext* opCtx, - repl::ReadConcernLevel readConcern, - bool excludeDraining) override { + OperationContext* opCtx, repl::ReadConcernLevel readConcern) override { return repl::OpTimeWith<std::vector<ShardType>>(_shards); } - std::vector<CollectionType> getCollections(OperationContext* opCtx, - StringData dbName, - repl::ReadConcernLevel readConcernLevel, - const BSONObj& sort) override { + std::vector<CollectionType> getCollections( + OperationContext* opCtx, + StringData dbName, + repl::ReadConcernLevel readConcernLevel) override { return _colls; } @@ -239,7 +237,7 @@ protected: createChunks(env.version.epoch(), env.sourceUuid, env.version.getTimestamp(), "y"), boost::none); - ASSERT_OK(onDbVersionMismatchNoExcept(opCtx, kNss.db(), boost::none)); + forceDatabaseRefresh(opCtx, kNss.db()); forceShardFilteringMetadataRefresh(opCtx, kNss); if (refreshTempNss) @@ -312,18 +310,22 @@ TEST_F(DestinedRecipientTest, TestGetDestinedRecipientThrowsOnBlockedRefresh) { auto opCtx = operationContext(); auto env = setupReshardingEnv(opCtx, false); - AutoGetCollection coll(opCtx, kNss, MODE_IX); - OperationShardingState::setShardRole(opCtx, kNss, env.version, env.dbVersion); + { + AutoGetCollection coll(opCtx, kNss, MODE_IX); + OperationShardingState::setShardRole(opCtx, kNss, env.version, env.dbVersion); + + FailPointEnableBlock failPoint("blockCollectionCacheLookup"); + ASSERT_THROWS_WITH_CHECK(ShardingWriteRouter(opCtx, kNss, Grid::get(opCtx)->catalogCache()), + ShardCannotRefreshDueToLocksHeldException, + [&](const ShardCannotRefreshDueToLocksHeldException& ex) { + const auto refreshInfo = + ex.extraInfo<ShardCannotRefreshDueToLocksHeldInfo>(); + ASSERT(refreshInfo); + ASSERT_EQ(refreshInfo->getNss(), env.tempNss); + }); + } - FailPointEnableBlock failPoint("blockCollectionCacheLookup"); - ASSERT_THROWS_WITH_CHECK(ShardingWriteRouter(opCtx, kNss, Grid::get(opCtx)->catalogCache()), - ShardCannotRefreshDueToLocksHeldException, - [&](const ShardCannotRefreshDueToLocksHeldException& ex) { - const auto refreshInfo = - ex.extraInfo<ShardCannotRefreshDueToLocksHeldInfo>(); - ASSERT(refreshInfo); - ASSERT_EQ(refreshInfo->getNss(), env.tempNss); - }); + auto sw = catalogCache()->getCollectionRoutingInfoWithRefresh(opCtx, env.tempNss); } TEST_F(DestinedRecipientTest, TestOpObserverSetsDestinedRecipientOnInserts) { 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 99ab47fc11a..e9a466a1cb6 100644 --- a/src/mongo/db/s/resharding/resharding_donor_recipient_common.cpp +++ b/src/mongo/db/s/resharding/resharding_donor_recipient_common.cpp @@ -51,35 +51,10 @@ 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. @@ -92,10 +67,7 @@ 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. - ensureStateDocumentInserted<StateMachine>(opCtx, doc); - - reshardingInterruptAfterInsertStateMachineDocument.execute( - [&opCtx](const BSONObj& data) { opCtx->markKilled(); }); + StateMachine::insertStateDocument(opCtx, doc); auto registry = repl::PrimaryOnlyServiceRegistry::get(opCtx->getServiceContext()); auto service = registry->lookupServiceByName(Service::kServiceName); @@ -110,6 +82,17 @@ 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()))); } } @@ -155,13 +138,6 @@ 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, @@ -352,16 +328,7 @@ 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_recipient_common.h b/src/mongo/db/s/resharding/resharding_donor_recipient_common.h index f6d6a259282..2efba26f659 100644 --- a/src/mongo/db/s/resharding/resharding_donor_recipient_common.h +++ b/src/mongo/db/s/resharding/resharding_donor_recipient_common.h @@ -38,15 +38,11 @@ namespace resharding { using ReshardingFields = TypeCollectionReshardingFields; /** - * Looks up the StateMachine by the 'reshardingUUID'. Returns boost::none in the following cases: - * 1. The state machine does not exist. - * 2. In certain cases when the node is shutting down. - * Additionally returns a bool indicating if the node is stepping or shutting down to disambiguate - * the two. + * Looks up the StateMachine by the 'reshardingUUID'. If it does not exist, returns boost::none. */ template <class Service, class StateMachine, class ReshardingDocument> -std::pair<boost::optional<std::shared_ptr<StateMachine>>, bool> -tryGetReshardingStateMachineAndShutdownState(OperationContext* opCtx, const UUID& reshardingUUID) { +boost::optional<std::shared_ptr<StateMachine>> tryGetReshardingStateMachine( + OperationContext* opCtx, const UUID& reshardingUUID) { auto instanceId = BSON(ReshardingDocument::kReshardingUUIDFieldName << reshardingUUID); auto registry = repl::PrimaryOnlyServiceRegistry::get(opCtx->getServiceContext()); auto service = registry->lookupServiceByName(Service::kServiceName); @@ -54,39 +50,6 @@ tryGetReshardingStateMachineAndShutdownState(OperationContext* opCtx, const UUID } /** - * Same as tryGetReshardingStateMachineAndShutdownState, except does not return the shutdown state. - */ -template <class Service, class StateMachine, class ReshardingDocument> -boost::optional<std::shared_ptr<StateMachine>> tryGetReshardingStateMachine( - OperationContext* opCtx, const UUID& reshardingUUID) { - auto [instance, _] = - tryGetReshardingStateMachineAndShutdownState<Service, StateMachine, ReshardingDocument>( - opCtx, reshardingUUID); - return instance; -} - -/** - * Same as tryGetReshardingStateMachine, except throws if we were stepping or shutting down when we - * tried to access the PrimaryOnlyService. Use this function in situations where you need to - * guarantee that a return of boost::none means that there is no state document on disk for the - * associated state machine. - */ -template <class Service, class StateMachine, class ReshardingDocument> -boost::optional<std::shared_ptr<StateMachine>> tryGetReshardingStateMachineAndThrowIfShuttingDown( - OperationContext* opCtx, const UUID& reshardingUUID) { - auto [instance, steppingOrShuttingDown] = - tryGetReshardingStateMachineAndShutdownState<Service, StateMachine, ReshardingDocument>( - opCtx, reshardingUUID); - - uassert(ErrorCodes::InterruptedDueToReplStateChange, - "Unable to get resharding state machine, if it exists, because the node is " - "stepping or shutting down.", - !steppingOrShuttingDown); - - return instance; -} - -/** * The following functions construct a ReshardingDocument from the given 'reshardingFields'. */ ReshardingDonorDocument constructDonorDocumentFromReshardingFields( diff --git a/src/mongo/db/s/resharding/resharding_donor_service.cpp b/src/mongo/db/s/resharding/resharding_donor_service.cpp index 2e75751ba01..ce60e58f0e1 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/exception_util.h" +#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/db_raii.h" #include "mongo/db/dbdirectclient.h" #include "mongo/db/dbhelpers.h" @@ -47,7 +47,6 @@ #include "mongo/db/persistent_task_store.h" #include "mongo/db/repl/repl_client_info.h" #include "mongo/db/repl/wait_for_majority_service.h" -#include "mongo/db/s/collection_sharding_runtime.h" #include "mongo/db/s/recoverable_critical_section_service.h" #include "mongo/db/s/resharding/resharding_change_event_o2_field_gen.h" #include "mongo/db/s/resharding/resharding_data_copy_util.h" @@ -56,7 +55,6 @@ #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" @@ -67,7 +65,6 @@ namespace mongo { -MONGO_FAIL_POINT_DEFINE(reshardingPauseDonorBeforeCatalogCacheRefresh); MONGO_FAIL_POINT_DEFINE(reshardingDonorFailsAfterTransitionToDonatingOplogEntries); MONGO_FAIL_POINT_DEFINE(removeDonorDocFailpoint); @@ -180,9 +177,8 @@ public: } } - void refreshCollectionPlacementInfo(OperationContext* opCtx, - const NamespaceString& sourceNss) override { - onShardVersionMismatch(opCtx, sourceNss, boost::none); + void clearFilteringMetadata(OperationContext* opCtx) { + resharding::clearFilteringMetadata(opCtx, true /* scheduleAsyncRefresh */); } }; @@ -371,15 +367,8 @@ ExecutorFuture<void> ReshardingDonorService::DonorStateMachine::_finishReshardin { auto opCtx = _cancelableOpCtxFactory->makeOperationContext(&cc()); - std::initializer_list<NamespaceString> namespacesToRefresh{ - _metadata.getSourceNss(), _metadata.getTempReshardingNss()}; - - // Clear filtering metadata for the source and temp resharding nss. - for (const auto& nss : namespacesToRefresh) { - AutoGetCollection autoColl(opCtx.get(), nss, MODE_IX); - CollectionShardingRuntime::get(opCtx.get(), nss) - ->clearFilteringMetadata(opCtx.get()); - } + + _externalState->clearFilteringMetadata(opCtx.get()); RecoverableCriticalSectionService::get(opCtx.get()) ->releaseRecoverableCriticalSection( @@ -389,13 +378,6 @@ ExecutorFuture<void> ReshardingDonorService::DonorStateMachine::_finishReshardin ShardingCatalogClient::kLocalWriteConcern); _metrics()->leaveCriticalSection(getCurrentTime()); - - // We force a refresh to make sure that the placement information is updated in - // cache after abort decision before the donor state document is deleted. - for (const auto& nss : namespacesToRefresh) { - _externalState->refreshCollectionPlacementInfo(opCtx.get(), nss); - _externalState->waitForCollectionFlush(opCtx.get(), nss); - } } auto opCtx = _cancelableOpCtxFactory->makeOperationContext(&cc()); @@ -585,8 +567,6 @@ 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()); } @@ -815,7 +795,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); - mongo::sharding_ddl_util::ensureCollectionDroppedNoChangeEvent( + resharding::data_copy::ensureCollectionDropped( opCtx.get(), _metadata.getSourceNss(), _metadata.getSourceUUID()); } diff --git a/src/mongo/db/s/resharding/resharding_donor_service.h b/src/mongo/db/s/resharding/resharding_donor_service.h index 7b5331f93ac..b50c88b6af5 100644 --- a/src/mongo/db/s/resharding/resharding_donor_service.h +++ b/src/mongo/db/s/resharding/resharding_donor_service.h @@ -298,8 +298,7 @@ public: const BSONObj& query, const BSONObj& update) = 0; - virtual void refreshCollectionPlacementInfo(OperationContext* opCtx, - const NamespaceString& sourceNss) = 0; + virtual void clearFilteringMetadata(OperationContext* opCtx) = 0; }; } // namespace mongo 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 16e1507be1c..663b8c28136 100644 --- a/src/mongo/db/s/resharding/resharding_donor_service_test.cpp +++ b/src/mongo/db/s/resharding/resharding_donor_service_test.cpp @@ -34,6 +34,7 @@ #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" @@ -52,7 +53,6 @@ #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" @@ -84,8 +84,7 @@ public: const BSONObj& query, const BSONObj& update) override {} - void refreshCollectionPlacementInfo(OperationContext* opCtx, - const NamespaceString& sourceNss) override {} + void clearFilteringMetadata(OperationContext* opCtx) override {} }; class DonorOpObserverForTest : public OpObserverForTest { @@ -162,8 +161,7 @@ public: void createSourceCollection(OperationContext* opCtx, const ReshardingDonorDocument& donorDoc) { CollectionOptions options; options.uuid = donorDoc.getSourceUUID(); - mongo::sharding_ddl_util::ensureCollectionDroppedNoChangeEvent(opCtx, - donorDoc.getSourceNss()); + resharding::data_copy::ensureCollectionDropped(opCtx, donorDoc.getSourceNss()); resharding::data_copy::ensureCollectionExists(opCtx, donorDoc.getSourceNss(), options); } @@ -171,8 +169,7 @@ public: const ReshardingDonorDocument& donorDoc) { CollectionOptions options; options.uuid = donorDoc.getReshardingUUID(); - mongo::sharding_ddl_util::ensureCollectionDroppedNoChangeEvent( - opCtx, donorDoc.getTempReshardingNss()); + resharding::data_copy::ensureCollectionDropped(opCtx, donorDoc.getTempReshardingNss()); resharding::data_copy::ensureCollectionExists( opCtx, donorDoc.getTempReshardingNss(), options); } @@ -418,10 +415,8 @@ TEST_F(ReshardingDonorServiceTest, StepDownStepUpEachTransition) { DonorStateMachine::insertStateDocument(opCtx.get(), doc); return DonorStateMachine::getOrCreate(opCtx.get(), _service, doc.toBSON()); } else { - auto [maybeDonor, isPausedOrShutdown] = - DonorStateMachine::lookup(opCtx.get(), _service, instanceId); - ASSERT_TRUE(maybeDonor); - ASSERT_FALSE(isPausedOrShutdown); + auto maybeDonor = DonorStateMachine::lookup(opCtx.get(), _service, instanceId); + ASSERT_TRUE(bool(maybeDonor)); // Allow the transition to prevState to succeed on this primary-only service // instance. @@ -482,10 +477,8 @@ TEST_F(ReshardingDonorServiceTest, StepDownStepUpEachTransition) { } // Finally complete the operation and ensure its success. - auto [maybeDonor, isPausedOrShutdown] = - DonorStateMachine::lookup(opCtx.get(), _service, instanceId); - ASSERT_TRUE(maybeDonor); - ASSERT_FALSE(isPausedOrShutdown); + auto maybeDonor = DonorStateMachine::lookup(opCtx.get(), _service, instanceId); + ASSERT_TRUE(bool(maybeDonor)); auto donor = *maybeDonor; stateTransitionsGuard.unset(DonorStateEnum::kDone); @@ -615,10 +608,8 @@ TEST_F(ReshardingDonorServiceTest, CompletesWithStepdownAfterAbort) { donor.reset(); stepUp(opCtx.get()); - auto [maybeDonor, isPausedOrShutdown] = - DonorStateMachine::lookup(opCtx.get(), _service, instanceId); - ASSERT_TRUE(maybeDonor); - ASSERT_FALSE(isPausedOrShutdown); + auto maybeDonor = DonorStateMachine::lookup(opCtx.get(), _service, instanceId); + ASSERT_TRUE(bool(maybeDonor)); donor = *maybeDonor; doneTransitionGuard.reset(); diff --git a/src/mongo/db/s/resharding/resharding_future_util.cpp b/src/mongo/db/s/resharding/resharding_future_util.cpp index 62c95fbaf3f..849a14bd80e 100644 --- a/src/mongo/db/s/resharding/resharding_future_util.cpp +++ b/src/mongo/db/s/resharding/resharding_future_util.cpp @@ -51,42 +51,19 @@ 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) { - 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 { + return whenAllSucceedOn(futures, executor) + .onError([futures, executor, cancelSource](Status originalError) mutable { cancelSource.cancel(); - return whenAll(runAllInlineUnsafe(futures)) + return whenAll(thenRunAllOn(futures, executor)) .ignoreValue() - .unsafeToInlineFuture() + .thenRunOn(executor) .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 78847350d0c..0bb858a963c 100644 --- a/src/mongo/db/s/resharding/resharding_future_util.h +++ b/src/mongo/db/s/resharding/resharding_future_util.h @@ -131,8 +131,7 @@ public: status.isA<ErrorCategory::CursorInvalidatedError>() || status == ErrorCodes::Interrupted || status.isA<ErrorCategory::CancellationError>() || - status.isA<ErrorCategory::NotPrimaryError>() || - status.isA<ErrorCategory::NetworkTimeoutError>()) { + status.isA<ErrorCategory::NotPrimaryError>()) { // 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 deleted file mode 100644 index e37a13a314b..00000000000 --- a/src/mongo/db/s/resharding/resharding_future_util_test.cpp +++ /dev/null @@ -1,100 +0,0 @@ -/** - * 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 b42c9df5c9e..6e4e4e041e9 100644 --- a/src/mongo/db/s/resharding/resharding_metrics.cpp +++ b/src/mongo/db/s/resharding/resharding_metrics.cpp @@ -406,60 +406,6 @@ 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(); @@ -538,9 +484,6 @@ 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); @@ -691,10 +634,6 @@ 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); } @@ -723,6 +662,22 @@ 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 9ba80d35417..a6964c9d611 100644 --- a/src/mongo/db/s/resharding/resharding_metrics.h +++ b/src/mongo/db/s/resharding/resharding_metrics.h @@ -35,7 +35,6 @@ #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" @@ -72,29 +71,6 @@ 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; @@ -137,6 +113,11 @@ 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.cpp b/src/mongo/db/s/resharding/resharding_op_observer.cpp index cd2fb324c8e..ba1f768f552 100644 --- a/src/mongo/db/s/resharding/resharding_op_observer.cpp +++ b/src/mongo/db/s/resharding/resharding_op_observer.cpp @@ -48,7 +48,7 @@ std::shared_ptr<ReshardingCoordinatorObserver> getReshardingCoordinatorObserver( OperationContext* opCtx, const BSONObj& reshardingId) { auto registry = repl::PrimaryOnlyServiceRegistry::get(opCtx->getServiceContext()); auto service = registry->lookupServiceByName(ReshardingCoordinatorService::kServiceName); - auto [instance, _] = + auto instance = ReshardingCoordinatorService::ReshardingCoordinator::lookup(opCtx, service, reshardingId); iassert( diff --git a/src/mongo/db/s/resharding/resharding_op_observer.h b/src/mongo/db/s/resharding/resharding_op_observer.h index e8affe3ef4a..30d319a041d 100644 --- a/src/mongo/db/s/resharding/resharding_op_observer.h +++ b/src/mongo/db/s/resharding/resharding_op_observer.h @@ -228,10 +228,6 @@ 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 42c5a1543e5..9a2f6f2750e 100644 --- a/src/mongo/db/s/resharding/resharding_oplog_application.cpp +++ b/src/mongo/db/s/resharding/resharding_oplog_application.cpp @@ -29,9 +29,11 @@ #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/exception_util.h" +#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/dbhelpers.h" #include "mongo/db/index/index_access_method.h" #include "mongo/db/logical_session_cache.h" @@ -77,8 +79,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 error to - // allow the collection metadata information to be recovered. + // and leave it to ReshardingOplogBatchApplier::applyBatch() to retry on a StaleConfig exception + // 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_applier.cpp b/src/mongo/db/s/resharding/resharding_oplog_applier.cpp index 71d156170d4..766ac67f266 100644 --- a/src/mongo/db/s/resharding/resharding_oplog_applier.cpp +++ b/src/mongo/db/s/resharding/resharding_oplog_applier.cpp @@ -236,8 +236,7 @@ void ReshardingOplogApplier::_clearAppliedOpsAndStoreProgress(OperationContext* store.upsert( opCtx, BSON(ReshardingOplogApplierProgress::kOplogSourceIdFieldName << _sourceId.toBSON()), - builder.obj(), - WriteConcerns::kLocalWriteConcern); + builder.obj()); _env->metrics()->onOplogEntriesApplied(_currentBatchToApply.size()); if (ShardingDataTransformMetrics::isEnabled()) { _env->metricsNew()->onOplogEntriesApplied(_currentBatchToApply.size()); diff --git a/src/mongo/db/s/resharding/resharding_oplog_applier_test.cpp b/src/mongo/db/s/resharding/resharding_oplog_applier_test.cpp index 774a654ce8f..02334d1f4bf 100644 --- a/src/mongo/db/s/resharding/resharding_oplog_applier_test.cpp +++ b/src/mongo/db/s/resharding/resharding_oplog_applier_test.cpp @@ -249,7 +249,6 @@ public: kCrudNs, kCrudUUID, false /* fromMigrate */, - boost::none, // checkExistenceForDiffInsert 0 /* version */, obj1, obj2, 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 c7efa6d33e4..4ff29b42d30 100644 --- a/src/mongo/db/s/resharding/resharding_oplog_batch_applier.cpp +++ b/src/mongo/db/s/resharding/resharding_oplog_batch_applier.cpp @@ -29,8 +29,12 @@ #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" @@ -79,8 +83,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 error to allow the collection metadata information to - // be recovered. + // on a StaleConfig exception 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 064ca34c9fa..30811c8d2aa 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/exception_util.h" +#include "mongo/db/concurrency/write_conflict_exception.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,22 +194,23 @@ ExecutorFuture<void> ReshardingOplogFetcher::_reschedule( } bool ReshardingOplogFetcher::iterate(Client* client, CancelableOperationContextFactory factory) { - 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(); + 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 { 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 99707c61b96..e7c512e2669 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/exception_util.h" +#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/db_raii.h" #include "mongo/db/dbdirectclient.h" #include "mongo/db/dbhelpers.h" @@ -144,9 +144,7 @@ public: StaticCatalogClient(std::vector<ShardId> shardIds) : _shardIds(std::move(shardIds)) {} StatusWith<repl::OpTimeWith<std::vector<ShardType>>> getAllShards( - OperationContext* opCtx, - repl::ReadConcernLevel readConcern, - bool excludeDraining) override { + OperationContext* opCtx, repl::ReadConcernLevel readConcern) override { std::vector<ShardType> shardTypes; for (const auto& shardId : _shardIds) { const ConnectionString cs = ConnectionString::forReplicaSet( @@ -668,44 +666,6 @@ 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 95c29d0825c..44251fa39dc 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/exception_util.h" +#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/dbdirectclient.h" #include "mongo/db/operation_context.h" #include "mongo/db/repl/oplog_entry.h" @@ -108,7 +108,6 @@ boost::optional<SharedSemiFuture<void>> ReshardingOplogSessionApplication::tryAp invariant(op.getTxnNumber()); invariant(op.get_id()); - auto sourceNss = op.getNss(); auto lsid = *op.getSessionId(); if (isInternalSessionForNonRetryableWrite(lsid)) { // Skip internal sessions for non-retryable writes since they only support transactions @@ -158,8 +157,7 @@ boost::optional<SharedSemiFuture<void>> ReshardingOplogSessionApplication::tryAp std::move(o2Field), std::move(stmtIds), std::move(preImageOpTime), - std::move(postImageOpTime), - std::move(sourceNss)); + std::move(postImageOpTime)); }); } diff --git a/src/mongo/db/s/resharding/resharding_recipient_service.cpp b/src/mongo/db/s/resharding/resharding_recipient_service.cpp index 087e3cb603f..5d49e0cdb22 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/exception_util.h" +#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/dbdirectclient.h" #include "mongo/db/dbhelpers.h" #include "mongo/db/ops/delete.h" @@ -56,7 +56,6 @@ #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" @@ -146,8 +145,6 @@ 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>([] { @@ -169,7 +166,6 @@ ReshardingRecipientService::RecipientStateMachine::RecipientStateMachine( return donor.getShardId() == myShardId; }) != _donorShards.end(); }()) { - invariant(_externalState); } @@ -557,11 +553,6 @@ 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()); @@ -761,7 +752,7 @@ void ReshardingRecipientService::RecipientStateMachine::_cleanupReshardingCollec opCtx.get(), _metadata.getReshardingUUID(), _metadata.getSourceUUID(), _donorShards); if (aborted) { - mongo::sharding_ddl_util::ensureCollectionDroppedNoChangeEvent( + resharding::data_copy::ensureCollectionDropped( opCtx.get(), _metadata.getTempReshardingNss(), _metadata.getReshardingUUID()); } } @@ -817,53 +808,27 @@ 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(cloningStartTime); + _metrics()->startCopyingDocuments(getCurrentTime()); } 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); - _metrics()->endCopyingDocuments(oplogApplicationStartTime); - _metrics()->startApplyingOplogEntries(oplogApplicationStartTime); + auto currentTime = getCurrentTime(); + _metrics()->endCopyingDocuments(currentTime); + _metrics()->startApplyingOplogEntries(currentTime); } 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); - _metrics()->endApplyingOplogEntries(oplogApplicationStopTime); + auto currentTime = getCurrentTime(); + _metrics()->endApplyingOplogEntries(currentTime); } void ReshardingRecipientService::RecipientStateMachine::_transitionToError( @@ -1015,9 +980,6 @@ void ReshardingRecipientService::RecipientStateMachine::_updateRecipientDocument setBuilder.append(ReshardingRecipientDocument::kDonorShardsFieldName, donorShardsArrayBuilder.arr()); - - setBuilder.append(ReshardingRecipientDocument::kApproxBytesToCopyFieldName, - cloneDetails->approxBytesToCopy); } if (configStartTime) { @@ -1025,8 +987,6 @@ void ReshardingRecipientService::RecipientStateMachine::_updateRecipientDocument *configStartTime); } - setBuilder.append(ReshardingRecipientDocument::kMetricsFieldName, _timeIntervals.toBSON()); - setBuilder.doneFast(); } @@ -1044,7 +1004,6 @@ void ReshardingRecipientService::RecipientStateMachine::_updateRecipientDocument if (cloneDetails) { _cloneTimestamp = cloneDetails->cloneTimestamp; _donorShards = std::move(cloneDetails->donorShards); - _approxBytesToCopy = cloneDetails->approxBytesToCopy; } if (configStartTime) { @@ -1104,6 +1063,7 @@ 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()); @@ -1113,6 +1073,7 @@ 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); }) @@ -1178,13 +1139,8 @@ void ReshardingRecipientService::RecipientStateMachine::_restoreMetrics( } } - _metrics()->onStepUp(_recipientCtx.getState(), - ReshardingMetrics::ReshardingRecipientCountsAndMetrics{documentCountCopied, - documentBytesCopied, - oplogEntriesFetched, - oplogEntriesApplied, - _approxBytesToCopy, - _timeIntervals}); + _metrics()->restoreForCurrentOp( + documentCountCopied, documentBytesCopied, oplogEntriesFetched, oplogEntriesApplied); } 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 86d3236b9c9..e1fe504a970 100644 --- a/src/mongo/db/s/resharding/resharding_recipient_service.h +++ b/src/mongo/db/s/resharding/resharding_recipient_service.h @@ -288,8 +288,6 @@ 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_external_state.cpp b/src/mongo/db/s/resharding/resharding_recipient_service_external_state.cpp index b99f1f60437..2dc584953d6 100644 --- a/src/mongo/db/s/resharding/resharding_recipient_service_external_state.cpp +++ b/src/mongo/db/s/resharding/resharding_recipient_service_external_state.cpp @@ -124,8 +124,13 @@ RecipientStateMachineExternalStateImpl::getCollectionOptions(OperationContext* o StringData reason) { // Load the collection options from the primary shard for the database. return _withShardVersionRetry(opCtx, nss, reason, [&] { + auto cm = getShardedCollectionRoutingInfo(opCtx, nss); return MigrationDestinationManager::getCollectionOptions( - opCtx, NamespaceStringOrUUID{nss.db().toString(), uuid}, afterClusterTime); + opCtx, + NamespaceStringOrUUID{nss.db().toString(), uuid}, + cm.dbPrimary(), + cm, + afterClusterTime); }); } 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 b598f32baf1..43ccab32bec 100644 --- a/src/mongo/db/s/resharding/resharding_recipient_service_test.cpp +++ b/src/mongo/db/s/resharding/resharding_recipient_service_test.cpp @@ -48,13 +48,10 @@ #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 { @@ -211,8 +208,6 @@ 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 { @@ -227,6 +222,7 @@ 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)); } @@ -267,8 +263,7 @@ public: const ReshardingRecipientDocument& recipientDoc) { CollectionOptions options; options.uuid = recipientDoc.getSourceUUID(); - mongo::sharding_ddl_util::ensureCollectionDroppedNoChangeEvent(opCtx, - recipientDoc.getSourceNss()); + resharding::data_copy::ensureCollectionDropped(opCtx, recipientDoc.getSourceNss()); resharding::data_copy::ensureCollectionExists(opCtx, recipientDoc.getSourceNss(), options); } @@ -379,10 +374,9 @@ TEST_F(ReshardingRecipientServiceTest, StepDownStepUpEachTransition) { RecipientStateMachine::insertStateDocument(opCtx.get(), doc); return RecipientStateMachine::getOrCreate(opCtx.get(), _service, doc.toBSON()); } else { - auto [maybeRecipient, isPausedOrShutdown] = + auto maybeRecipient = RecipientStateMachine::lookup(opCtx.get(), _service, instanceId); - ASSERT_TRUE(maybeRecipient); - ASSERT_FALSE(isPausedOrShutdown); + ASSERT_TRUE(bool(maybeRecipient)); // Allow the transition to prevState to succeed on this primary-only service // instance. @@ -429,10 +423,8 @@ TEST_F(ReshardingRecipientServiceTest, StepDownStepUpEachTransition) { } // Finally complete the operation and ensure its success. - auto [maybeRecipient, isPausedOrShutdown] = - RecipientStateMachine::lookup(opCtx.get(), _service, instanceId); - ASSERT_TRUE(maybeRecipient); - ASSERT_FALSE(isPausedOrShutdown); + auto maybeRecipient = RecipientStateMachine::lookup(opCtx.get(), _service, instanceId); + ASSERT_TRUE(bool(maybeRecipient)); auto recipient = *maybeRecipient; @@ -477,10 +469,8 @@ TEST_F(ReshardingRecipientServiceTest, OpCtxKilledWhileRestoringMetrics) { stepUp(opCtx.get()); // After the failpoint is disabled, the operation should succeed. - auto [maybeRecipient, isPausedOrShutdown] = - RecipientStateMachine::lookup(opCtx.get(), _service, instanceId); - ASSERT_TRUE(maybeRecipient); - ASSERT_FALSE(isPausedOrShutdown); + auto maybeRecipient = RecipientStateMachine::lookup(opCtx.get(), _service, instanceId); + ASSERT_TRUE(bool(maybeRecipient)); recipient = *maybeRecipient; notifyReshardingCommitting(opCtx.get(), *recipient, doc); ASSERT_OK(recipient->getCompletionFuture().getNoThrow()); @@ -506,89 +496,69 @@ DEATH_TEST_REGEX_F(ReshardingRecipientServiceTest, CommitFn, "4457001.*tripwire" TEST_F(ReshardingRecipientServiceTest, DropsTemporaryReshardingCollectionOnAbort) { auto metrics = ReshardingRecipientServiceTest::metrics(); for (bool isAlsoDonor : {false, true}) { - 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); - } - - auto doc = makeStateDocument(isAlsoDonor); - auto instanceId = BSON(ReshardingRecipientDocument::kReshardingUUIDFieldName - << doc.getReshardingUUID()); + LOGV2(5551107, + "Running case", + "test"_attr = _agent.getTestName(), + "isAlsoDonor"_attr = isAlsoDonor); - auto opCtx = makeOperationContext(); + boost::optional<PauseDuringStateTransitions> doneTransitionGuard; + doneTransitionGuard.emplace(controller(), RecipientStateEnum::kDone); - if (isAlsoDonor) { - // If the recipient is also a donor, the original collection should already exist on - // this shard. - createSourceCollection(opCtx.get(), doc); - } + auto doc = makeStateDocument(isAlsoDonor); + auto instanceId = + BSON(ReshardingRecipientDocument::kReshardingUUIDFieldName << doc.getReshardingUUID()); - RecipientStateMachine::insertStateDocument(opCtx.get(), doc); - auto recipient = - RecipientStateMachine::getOrCreate(opCtx.get(), _service, doc.toBSON()); + auto opCtx = makeOperationContext(); - 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); - } + if (isAlsoDonor) { + // If the recipient is also a donor, the original collection should already exist on + // this shard. + createSourceCollection(opCtx.get(), doc); + } - recipient->abort(false); + RecipientStateMachine::insertStateDocument(opCtx.get(), doc); + auto recipient = RecipientStateMachine::getOrCreate(opCtx.get(), _service, doc.toBSON()); - stateTransitionGuard->wait(RecipientStateEnum::kDone); + notifyToStartCloning(opCtx.get(), *recipient, doc); + recipient->abort(false); - stepDown(); + doneTransitionGuard->wait(RecipientStateEnum::kDone); + stepDown(); - ASSERT_EQ(recipient->getCompletionFuture().getNoThrow(), - ErrorCodes::InterruptedDueToReplStateChange); + ASSERT_EQ(recipient->getCompletionFuture().getNoThrow(), + ErrorCodes::InterruptedDueToReplStateChange); - recipient.reset(); - stepUp(opCtx.get()); + recipient.reset(); + stepUp(opCtx.get()); - auto [maybeRecipient, isPausedOrShutdown] = - RecipientStateMachine::lookup(opCtx.get(), _service, instanceId); - ASSERT_TRUE(maybeRecipient); - ASSERT_FALSE(isPausedOrShutdown); - recipient = *maybeRecipient; + auto maybeRecipient = RecipientStateMachine::lookup(opCtx.get(), _service, instanceId); + ASSERT_TRUE(bool(maybeRecipient)); + recipient = *maybeRecipient; - stateTransitionGuard.reset(); - recipient->abort(false); + doneTransitionGuard.reset(); + recipient->abort(false); - ASSERT_OK(recipient->getCompletionFuture().getNoThrow()); - checkStateDocumentRemoved(opCtx.get()); + ASSERT_OK(recipient->getCompletionFuture().getNoThrow()); + checkStateDocumentRemoved(opCtx.get()); - 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()); - } + 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)); - } + // 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_LESS_THAN_OR_EQUALS(result.obj().getField("countReshardingFailures").numberLong(), 4); + ASSERT_EQ(result.obj().getField("countReshardingFailures").numberLong(), 2); } TEST_F(ReshardingRecipientServiceTest, RenamesTemporaryReshardingCollectionWhenDone) { @@ -778,10 +748,9 @@ TEST_F(ReshardingRecipientServiceTest, RestoreMetricsAfterStepUp) { RecipientStateMachine::insertStateDocument(opCtx.get(), doc); return RecipientStateMachine::getOrCreate(opCtx.get(), _service, doc.toBSON()); } else { - auto [maybeRecipient, isPausedOrShutdown] = + auto maybeRecipient = RecipientStateMachine::lookup(opCtx.get(), _service, instanceId); - ASSERT_TRUE(maybeRecipient); - ASSERT_FALSE(isPausedOrShutdown); + ASSERT_TRUE(bool(maybeRecipient)); // Allow the transition to prevState to succeed on this primary-only service // instance. @@ -847,33 +816,22 @@ TEST_F(ReshardingRecipientServiceTest, RestoreMetricsAfterStepUp) { } // Step down before the transition to state can complete. stateTransitionsGuard.wait(state); - - 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) { + if (state == RecipientStateEnum::kStrictConsistency) { + 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.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(), @@ -882,11 +840,7 @@ 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 96a08c210fa..7ab491286f3 100644 --- a/src/mongo/db/s/resharding/resharding_txn_cloner.cpp +++ b/src/mongo/db/s/resharding/resharding_txn_cloner.cpp @@ -40,6 +40,7 @@ #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" @@ -190,8 +191,7 @@ boost::optional<SharedSemiFuture<void>> ReshardingTxnCloner::doOneRecord( TransactionParticipant::kDeadEndSentinel, {kIncompleteHistoryStmtId}, boost::none /* preImageOpTime */, - boost::none /* postImageOpTime */, - {}); + boost::none /* postImageOpTime */); }); } diff --git a/src/mongo/db/s/resharding/resharding_txn_cloner_test.cpp b/src/mongo/db/s/resharding/resharding_txn_cloner_test.cpp index b77db34ef11..e4f4f1f2b61 100644 --- a/src/mongo/db/s/resharding/resharding_txn_cloner_test.cpp +++ b/src/mongo/db/s/resharding/resharding_txn_cloner_test.cpp @@ -40,7 +40,6 @@ #include "mongo/db/logical_session_cache_noop.h" #include "mongo/db/persistent_task_store.h" #include "mongo/db/pipeline/process_interface/shardsvr_process_interface.h" -#include "mongo/db/query/cursor_response.h" #include "mongo/db/repl/storage_interface_impl.h" #include "mongo/db/repl/wait_for_majority_service.h" #include "mongo/db/s/resharding/resharding_server_parameters_gen.h" @@ -130,9 +129,7 @@ class ReshardingTxnClonerTest : public ShardServerTestFixture { StaticCatalogClient(std::vector<ShardId> shardIds) : _shardIds(std::move(shardIds)) {} StatusWith<repl::OpTimeWith<std::vector<ShardType>>> getAllShards( - OperationContext* opCtx, - repl::ReadConcernLevel readConcern, - bool excludeDraining) override { + OperationContext* opCtx, repl::ReadConcernLevel readConcern) override { std::vector<ShardType> shardTypes; for (const auto& shardId : _shardIds) { const ConnectionString cs = ConnectionString::forReplicaSet( diff --git a/src/mongo/db/s/resharding/resharding_util.cpp b/src/mongo/db/s/resharding/resharding_util.cpp index 3e0db3b383e..27d319264dd 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/exception_util.h" +#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/exec/document_value/document.h" #include "mongo/db/namespace_string.h" #include "mongo/db/op_observer.h" @@ -207,14 +207,11 @@ void checkForOverlappingZones(std::vector<ReshardingZoneType>& zones) { } std::vector<BSONObj> buildTagsDocsFromZones(const NamespaceString& tempNss, - const std::vector<ReshardingZoneType>& zones, - const ShardKeyPattern& shardKey) { + const std::vector<ReshardingZoneType>& zones) { std::vector<BSONObj> tags; tags.reserve(zones.size()); - for (auto& zone : zones) { - auto min = shardKey.getKeyPattern().extendRangeBound(zone.getMin(), false); - auto max = shardKey.getKeyPattern().extendRangeBound(zone.getMax(), false); - ChunkRange range(min, max); + for (const auto& zone : zones) { + ChunkRange range(zone.getMin(), zone.getMax()); TagsType tag(tempNss, zone.getZone().toString(), range); tags.push_back(tag.toBSON()); } diff --git a/src/mongo/db/s/resharding/resharding_util.h b/src/mongo/db/s/resharding/resharding_util.h index f0e60e99555..856d7cbb081 100644 --- a/src/mongo/db/s/resharding/resharding_util.h +++ b/src/mongo/db/s/resharding/resharding_util.h @@ -260,8 +260,7 @@ void checkForOverlappingZones(std::vector<ReshardingZoneType>& zones); * Builds documents to insert into config.tags from zones provided to reshardCollection cmd. */ std::vector<BSONObj> buildTagsDocsFromZones(const NamespaceString& tempNss, - const std::vector<ReshardingZoneType>& zones, - const ShardKeyPattern& shardKey); + const std::vector<ReshardingZoneType>& zones); /** * Creates a pipeline that can be serialized into a query for fetching oplog entries. `startAfter` |