diff options
Diffstat (limited to 'src/mongo/db/s/resharding')
19 files changed, 82 insertions, 33 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 75cb7be7049..2695ed842c0 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,8 +101,7 @@ StageConstraints DocumentSourceReshardingAddResumeId::constraints( ChangeStreamRequirement::kDenylist); } -Value DocumentSourceReshardingAddResumeId::serialize( - boost::optional<ExplainOptions::Verbosity> explain) const { +Value DocumentSourceReshardingAddResumeId::serialize(const SerializationOptions& opts) 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 31cbd97c694..4fb27980c68 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(boost::optional<ExplainOptions::Verbosity> explain) const; + Value serialize(const SerializationOptions& opts = SerializationOptions{}) const final override; 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 8075111e3af..8260cf3e4cc 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( - boost::optional<ExplainOptions::Verbosity> explain) const { + const SerializationOptions& opts) const { return Value( Document{{kStageName, Value(Document{{kIncludeCommitTransactionTimestampFieldName, 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 9589cb64a08..8c792116e6c 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(boost::optional<ExplainOptions::Verbosity> explain) const; + Value serialize(const SerializationOptions& opts = SerializationOptions{}) const final override; 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 3144723bf2d..157876391d3 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,12 +90,11 @@ StageConstraints DocumentSourceReshardingOwnershipMatch::constraints( ChangeStreamRequirement::kDenylist); } -Value DocumentSourceReshardingOwnershipMatch::serialize( - boost::optional<ExplainOptions::Verbosity> explain) const { +Value DocumentSourceReshardingOwnershipMatch::serialize(const SerializationOptions& opts) const { return Value{Document{{kStageName, DocumentSourceReshardingOwnershipMatchSpec( _recipientShardId, _reshardingKey.getKeyPattern()) - .toBSON()}}}; + .toBSON(opts)}}}; } DepsTracker::State DocumentSourceReshardingOwnershipMatch::getDependencies( 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 7a6db2bc125..b7da07a5a57 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(boost::optional<ExplainOptions::Verbosity> explain) const final; + Value serialize(const SerializationOptions& opts = SerializationOptions{}) const final override; StageConstraints constraints(Pipeline::SplitState pipeState) const final; 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..c87cb3667dd 100644 --- a/src/mongo/db/s/resharding/resharding_coordinator_commit_monitor.cpp +++ b/src/mongo/db/s/resharding/resharding_coordinator_commit_monitor.cpp @@ -91,17 +91,19 @@ CoordinatorCommitMonitor::CoordinatorCommitMonitor( std::vector<ShardId> recipientShards, CoordinatorCommitMonitor::TaskExecutorPtr executor, CancellationToken cancelToken, + int delayBeforeInitialQueryMillis, Milliseconds maxDelayBetweenQueries) : _ns(std::move(ns)), _recipientShards(std::move(recipientShards)), _executor(std::move(executor)), _cancelToken(std::move(cancelToken)), _threshold(Milliseconds(gRemainingReshardingOperationTimeThresholdMillis.load())), + _delayBeforeInitialQueryMillis(Milliseconds(delayBeforeInitialQueryMillis)), _maxDelayBetweenQueries(maxDelayBetweenQueries) {} SemiFuture<void> CoordinatorCommitMonitor::waitUntilRecipientsAreWithinCommitThreshold() const { - return _makeFuture() + return _makeFuture(_delayBeforeInitialQueryMillis) .onError([](Status status) { if (ErrorCodes::isCancellationError(status.code()) || ErrorCodes::isInterruption(status.code())) { @@ -195,9 +197,16 @@ CoordinatorCommitMonitor::queryRemainingOperationTimeForRecipients() const { return {minRemainingTime, maxRemainingTime}; } -ExecutorFuture<void> CoordinatorCommitMonitor::_makeFuture() const { +ExecutorFuture<void> CoordinatorCommitMonitor::_makeFuture(Milliseconds delayBetweenQueries) const { return ExecutorFuture<void>(_executor) - .then([this] { return queryRemainingOperationTimeForRecipients(); }) + // Start waiting so that we have a more time to calculate a more realistic remaining time + // estimate. + .then([this, anchor = shared_from_this(), delayBetweenQueries] { + return _executor->sleepFor(delayBetweenQueries, _cancelToken) + .then([this, anchor = std::move(anchor)] { + return queryRemainingOperationTimeForRecipients(); + }); + }) .onError([this](Status status) { if (_cancelToken.isCanceled()) { // Do not retry on cancellation errors. @@ -233,12 +242,10 @@ ExecutorFuture<void> CoordinatorCommitMonitor::_makeFuture() const { // The following ensures that the monitor would never sleep for more than a predefined // maximum delay between querying recipient shards. Thus, it can handle very large, // and potentially inaccurate estimates of the remaining operation time. - auto sleepTime = std::min(remainingTimes.max - _threshold, _maxDelayBetweenQueries); - return _executor->sleepFor(sleepTime, _cancelToken) - .then([this, anchor = std::move(anchor)] { - // We are not canceled yet, so schedule new queries against recipient shards. - return _makeFuture(); - }); + auto delayBetweenQueries = + std::min(remainingTimes.max - _threshold, _maxDelayBetweenQueries); + + return _makeFuture(delayBetweenQueries); }); } diff --git a/src/mongo/db/s/resharding/resharding_coordinator_commit_monitor.h b/src/mongo/db/s/resharding/resharding_coordinator_commit_monitor.h index 64544981ae5..be722fb11bd 100644 --- a/src/mongo/db/s/resharding/resharding_coordinator_commit_monitor.h +++ b/src/mongo/db/s/resharding/resharding_coordinator_commit_monitor.h @@ -72,6 +72,7 @@ public: std::vector<ShardId> recipientShards, TaskExecutorPtr executor, CancellationToken cancelToken, + int delayBeforeInitialQueryMillis, Milliseconds maxDelayBetweenQueries = kMaxDelayBetweenQueries); SemiFuture<void> waitUntilRecipientsAreWithinCommitThreshold() const; @@ -88,7 +89,7 @@ public: RemainingOperationTimes queryRemainingOperationTimeForRecipients() const; private: - ExecutorFuture<void> _makeFuture() const; + ExecutorFuture<void> _makeFuture(Milliseconds delayBetweenQueries) const; static constexpr auto kDiagnosticLogLevel = 0; static constexpr auto kMaxDelayBetweenQueries = Seconds(30); @@ -98,6 +99,8 @@ private: const TaskExecutorPtr _executor; const CancellationToken _cancelToken; const Milliseconds _threshold; + + const Milliseconds _delayBeforeInitialQueryMillis; const Milliseconds _maxDelayBetweenQueries; TaskExecutorPtr _networkExecutor; 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..0804565201c 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 @@ -152,7 +152,7 @@ void CoordinatorCommitMonitorTest::setUp() { _cancellationSource = std::make_unique<CancellationSource>(); _commitMonitor = std::make_shared<CoordinatorCommitMonitor>( - _ns, _recipientShards, _futureExecutor, _cancellationSource->token(), Milliseconds(0)); + _ns, _recipientShards, _futureExecutor, _cancellationSource->token(), 0, Milliseconds(0)); _commitMonitor->setNetworkExecutorForTest(executor()); } diff --git a/src/mongo/db/s/resharding/resharding_coordinator_service.cpp b/src/mongo/db/s/resharding/resharding_coordinator_service.cpp index ea3451f1c4a..988360d6799 100644 --- a/src/mongo/db/s/resharding/resharding_coordinator_service.cpp +++ b/src/mongo/db/s/resharding/resharding_coordinator_service.cpp @@ -98,6 +98,7 @@ MONGO_FAIL_POINT_DEFINE(pauseBeforeInsertCoordinatorDoc); MONGO_FAIL_POINT_DEFINE(pauseBeforeCTHolderInitialization); const std::string kReshardingCoordinatorActiveIndexName = "ReshardingCoordinatorActiveIndex"; +const int kReshardingNumInitialChunksDefault = 90; const Backoff kExponentialBackoff(Seconds(1), Milliseconds::max()); const WriteConcernOptions kMajorityWriteConcern{ WriteConcernOptions::kMajority, WriteConcernOptions::SyncMode::UNSET, Seconds(0)}; @@ -879,7 +880,7 @@ ReshardingCoordinatorExternalStateImpl::calculateParticipantShardsAndChunks( } else { int numInitialChunks = coordinatorDoc.getNumInitialChunks() ? *coordinatorDoc.getNumInitialChunks() - : cm.numChunks(); + : kReshardingNumInitialChunksDefault; ShardKeyPattern shardKey(coordinatorDoc.getReshardingKey()); const auto tempNs = coordinatorDoc.getTempReshardingNss(); @@ -1714,7 +1715,8 @@ void ReshardingCoordinatorService::ReshardingCoordinator::_startCommitMonitor( _coordinatorDoc.getSourceNss(), extractShardIdsFromParticipantEntries(_coordinatorDoc.getRecipientShards()), **executor, - _ctHolder->getCommitMonitorToken()); + _ctHolder->getCommitMonitorToken(), + resharding::gReshardingDelayBeforeRemainingOperationTimeQueryMillis.load()); _commitMonitorQuiesced = _commitMonitor->waitUntilRecipientsAreWithinCommitThreshold() .thenRunOn(**executor) 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 d0b27f00c3c..8635e389cf4 100644 --- a/src/mongo/db/s/resharding/resharding_data_copy_util.cpp +++ b/src/mongo/db/s/resharding/resharding_data_copy_util.cpp @@ -282,7 +282,8 @@ void updateSessionRecord(OperationContext* opCtx, BSONObj o2Field, std::vector<StmtId> stmtIds, boost::optional<repl::OpTime> preImageOpTime, - boost::optional<repl::OpTime> postImageOpTime) { + boost::optional<repl::OpTime> postImageOpTime, + NamespaceString sourceNss) { invariant(opCtx->getLogicalSessionId()); invariant(opCtx->getTxnNumber()); @@ -296,7 +297,7 @@ void updateSessionRecord(OperationContext* opCtx, oplogEntry.setOpType(repl::OpTypeEnum::kNoop); oplogEntry.setObject(SessionCatalogMigration::kSessionOplogTag); oplogEntry.setObject2(std::move(o2Field)); - oplogEntry.setNss({}); + oplogEntry.setNss(std::move(sourceNss)); 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 b51cfc250a5..af24152aaf2 100644 --- a/src/mongo/db/s/resharding/resharding_data_copy_util.h +++ b/src/mongo/db/s/resharding/resharding_data_copy_util.h @@ -144,7 +144,8 @@ void updateSessionRecord(OperationContext* opCtx, BSONObj o2Field, std::vector<StmtId> stmtIds, boost::optional<repl::OpTime> preImageOpTime, - boost::optional<repl::OpTime> postImageOpTime); + boost::optional<repl::OpTime> postImageOpTime, + NamespaceString sourceNss); /** * Calls and returns the value from the supplied lambda function. diff --git a/src/mongo/db/s/resharding/resharding_donor_service.cpp b/src/mongo/db/s/resharding/resharding_donor_service.cpp index 4594c4018a6..2e75751ba01 100644 --- a/src/mongo/db/s/resharding/resharding_donor_service.cpp +++ b/src/mongo/db/s/resharding/resharding_donor_service.cpp @@ -47,6 +47,7 @@ #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" @@ -179,8 +180,9 @@ public: } } - void clearFilteringMetadata(OperationContext* opCtx) { - resharding::clearFilteringMetadata(opCtx, true /* scheduleAsyncRefresh */); + void refreshCollectionPlacementInfo(OperationContext* opCtx, + const NamespaceString& sourceNss) override { + onShardVersionMismatch(opCtx, sourceNss, boost::none); } }; @@ -369,8 +371,15 @@ ExecutorFuture<void> ReshardingDonorService::DonorStateMachine::_finishReshardin { auto opCtx = _cancelableOpCtxFactory->makeOperationContext(&cc()); - - _externalState->clearFilteringMetadata(opCtx.get()); + 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()); + } RecoverableCriticalSectionService::get(opCtx.get()) ->releaseRecoverableCriticalSection( @@ -380,6 +389,13 @@ 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()); diff --git a/src/mongo/db/s/resharding/resharding_donor_service.h b/src/mongo/db/s/resharding/resharding_donor_service.h index b50c88b6af5..7b5331f93ac 100644 --- a/src/mongo/db/s/resharding/resharding_donor_service.h +++ b/src/mongo/db/s/resharding/resharding_donor_service.h @@ -298,7 +298,8 @@ public: const BSONObj& query, const BSONObj& update) = 0; - virtual void clearFilteringMetadata(OperationContext* opCtx) = 0; + virtual void refreshCollectionPlacementInfo(OperationContext* opCtx, + const NamespaceString& sourceNss) = 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 cb358c3f508..16e1507be1c 100644 --- a/src/mongo/db/s/resharding/resharding_donor_service_test.cpp +++ b/src/mongo/db/s/resharding/resharding_donor_service_test.cpp @@ -84,7 +84,8 @@ public: const BSONObj& query, const BSONObj& update) override {} - void clearFilteringMetadata(OperationContext* opCtx) override {} + void refreshCollectionPlacementInfo(OperationContext* opCtx, + const NamespaceString& sourceNss) override {} }; class DonorOpObserverForTest : public OpObserverForTest { 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 80338642587..95c29d0825c 100644 --- a/src/mongo/db/s/resharding/resharding_oplog_session_application.cpp +++ b/src/mongo/db/s/resharding/resharding_oplog_session_application.cpp @@ -108,6 +108,7 @@ 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 @@ -157,7 +158,8 @@ boost::optional<SharedSemiFuture<void>> ReshardingOplogSessionApplication::tryAp std::move(o2Field), std::move(stmtIds), std::move(preImageOpTime), - std::move(postImageOpTime)); + std::move(postImageOpTime), + std::move(sourceNss)); }); } diff --git a/src/mongo/db/s/resharding/resharding_server_parameters.idl b/src/mongo/db/s/resharding/resharding_server_parameters.idl index daaedfc9ea4..7508d30c772 100644 --- a/src/mongo/db/s/resharding/resharding_server_parameters.idl +++ b/src/mongo/db/s/resharding/resharding_server_parameters.idl @@ -151,6 +151,21 @@ server_parameters: validator: gte: 0 + reshardingDelayBeforeRemainingOperationTimeQueryMillis: + description: >- + Initial delay before querying for remaining operation time from recipient shards. + The delay allows for applying more oplog entries before calculating time remaining, giving + a more accurate value. + Note we will have this delay every time we happen to have a failover occur. + set_at: [startup, runtime] + cpp_vartype: AtomicWord<int> + cpp_varname: gReshardingDelayBeforeRemainingOperationTimeQueryMillis + default: + expr: 0 + validator: + gte: 0 + redact: false + reshardingCriticalSectionTimeoutMillis: description: >- The upper limit on how long to wait to hear back from recipient shards reaching strict diff --git a/src/mongo/db/s/resharding/resharding_txn_cloner.cpp b/src/mongo/db/s/resharding/resharding_txn_cloner.cpp index 764a32d68d5..96a08c210fa 100644 --- a/src/mongo/db/s/resharding/resharding_txn_cloner.cpp +++ b/src/mongo/db/s/resharding/resharding_txn_cloner.cpp @@ -190,7 +190,8 @@ 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 3a10e01306c..b77db34ef11 100644 --- a/src/mongo/db/s/resharding/resharding_txn_cloner_test.cpp +++ b/src/mongo/db/s/resharding/resharding_txn_cloner_test.cpp @@ -40,6 +40,7 @@ #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" |