diff options
Diffstat (limited to 'src/mongo/db/s/migration_source_manager.cpp')
| -rw-r--r-- | src/mongo/db/s/migration_source_manager.cpp | 55 |
1 files changed, 20 insertions, 35 deletions
diff --git a/src/mongo/db/s/migration_source_manager.cpp b/src/mongo/db/s/migration_source_manager.cpp index 9ca15fe9fad..e4ed5e9df6d 100644 --- a/src/mongo/db/s/migration_source_manager.cpp +++ b/src/mongo/db/s/migration_source_manager.cpp @@ -33,6 +33,7 @@ #include "mongo/bson/bsonobjbuilder.h" #include "mongo/db/catalog_raii.h" +#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/logical_session_cache.h" #include "mongo/db/logical_session_id_helpers.h" #include "mongo/db/op_observer.h" @@ -81,6 +82,8 @@ const WriteConcernOptions kMajorityWriteConcern(WriteConcernOptions::kMajority, WriteConcernOptions::SyncMode::UNSET, WriteConcernOptions::kWriteConcernTimeoutMigration); +std::string kEmptyErrMsgForMoveTimingHelper; + /** * Best-effort attempt to ensure the recipient shard has refreshed its routing table to * 'newCollVersion'. Fires and forgets an asychronous remote setShardVersion command. @@ -177,6 +180,7 @@ MigrationSourceManager::MigrationSourceManager(OperationContext* opCtx, _args.getMin(), _args.getMax(), 6, // Total number of steps + &kEmptyErrMsgForMoveTimingHelper, _args.getToShard(), _args.getFromShard()) { invariant(!_opCtx->lockState()->isLocked()); @@ -362,17 +366,7 @@ MigrationSourceManager::~MigrationSourceManager() { invariant(!_cloneDriver); _stats.totalDonorMoveChunkTimeMillis.addAndFetch(_entireOpTimer.millis()); - if (_state == kDone) { - _completion.emplaceValue(); - } else { - std::string errMsg = "Migration not completed"; - if (_coordinator) { - const auto& migrationId = _coordinator->getMigrationId(); - errMsg = str::stream() << "Migration " << migrationId << " not completed"; - } - auto status = Status{ErrorCodes::Interrupted, errMsg}; - _completion.setError(status); - } + _completion.emplaceValue(); } void MigrationSourceManager::startClone() { @@ -426,12 +420,6 @@ void MigrationSourceManager::startClone() { _state = kCloning; } - // Refreshing the collection routing information after starting the clone driver will give us a - // stable view on whether the recipient is owning other chunks of the collection (a condition - // that will be later evaluated). - uassertStatusOK( - Grid::get(_opCtx)->catalogCache()->getCollectionRoutingInfoWithRefresh(_opCtx, nss())); - if (replEnabled) { auto const readConcernArgs = repl::ReadConcernArgs( replCoord->getMyLastAppliedOpTime(), repl::ReadConcernLevel::kLocalReadConcern); @@ -477,12 +465,11 @@ void MigrationSourceManager::enterCriticalSection() { _stats.totalDonorChunkCloneTimeMillis.addAndFetch(_cloneAndCommitTimer.millis()); _cloneAndCommitTimer.reset(); - const auto cm = - uassertStatusOK(Grid::get(_opCtx)->catalogCache()->getCollectionRoutingInfo(_opCtx, nss())); + const auto& metadata = _getCurrentMetadataAndCheckEpoch(); // Check that there are no chunks on the recepient shard. Write an oplog event for change // streams if this is the first migration to the recipient. - if (!cm.getVersion(_args.getToShard()).isSet()) { + if (!metadata.getChunkManager()->getVersion(_args.getToShard()).isSet()) { migrationutil::notifyChangeStreamsOnRecipientFirstChunk( _opCtx, nss(), _args.getFromShard(), _args.getToShard(), _collectionUUID); } @@ -492,7 +479,7 @@ void MigrationSourceManager::enterCriticalSection() { // NOTE: The 'migrateChunkToNewShard' oplog message written by the above call to // 'notifyChangeStreamsOnRecipientFirstChunk' depends on this majority write to carry its // local write to majority committed. - uassertStatusOKWithContext(ShardingStateRecovery::startMetadataOp(_opCtx), "Start metadata op"); + uassertStatusOK(ShardingStateRecovery::startMetadataOp(_opCtx)); LOGV2_DEBUG_OPTIONS(4817402, 2, @@ -510,13 +497,17 @@ void MigrationSourceManager::enterCriticalSection() { // time inclusive of the migration config commit update from accessing secondary data. // Note: this write must occur after the critSec flag is set, to ensure the secondary refresh // will stall behind the flag. - uassertStatusOKWithContext( - shardmetadatautil::updateShardCollectionsEntry( - _opCtx, - BSON(ShardCollectionType::kNssFieldName << nss().ns()), - BSON("$inc" << BSON(ShardCollectionType::kEnterCriticalSectionCounterFieldName << 1)), - false /*upsert*/), - "Persist critical section signal for secondaries"); + Status signalStatus = shardmetadatautil::updateShardCollectionsEntry( + _opCtx, + BSON(ShardCollectionType::kNssFieldName << nss().ns()), + BSON("$inc" << BSON(ShardCollectionType::kEnterCriticalSectionCounterFieldName << 1)), + false /*upsert*/); + if (!signalStatus.isOK()) { + uasserted( + ErrorCodes::OperationFailed, + str::stream() << "Failed to persist critical section signal for secondaries due to: " + << signalStatus.toString()); + } LOGV2(22017, "Migration successfully entered critical section", @@ -693,7 +684,7 @@ void MigrationSourceManager::commitChunkMetadataOnConfig() { _stats.totalCriticalSectionCommitTimeMillis.addAndFetch(t.millis()); - LOGV2(6107801, + LOGV2(4817403, "Exiting commit critical section", "migrationId"_attr = _coordinator->getMigrationId(), "durationMillis"_attr = t.millis()); @@ -814,12 +805,6 @@ void MigrationSourceManager::_cleanup(bool completeMigration) noexcept { }(); if (_state == kCriticalSection || _state == kCloneCompleted || _state == kCommittingOnConfig) { - LOGV2_DEBUG_OPTIONS(4817403, - 2, - {logv2::LogComponent::kShardMigrationPerf}, - "Finished critical section", - "migrationId"_attr = _coordinator->getMigrationId()); - LOGV2(6107802, "Finished critical section", "migrationId"_attr = _coordinator->getMigrationId(), |