diff options
4 files changed, 168 insertions, 40 deletions
diff --git a/etc/backports_required_for_multiversion_tests.yml b/etc/backports_required_for_multiversion_tests.yml index 494804aa496..0b93d4bbb1f 100644 --- a/etc/backports_required_for_multiversion_tests.yml +++ b/etc/backports_required_for_multiversion_tests.yml @@ -747,6 +747,8 @@ last-continuous: ticket: SERVER-93099 - test_file: jstests/core/index/index_creation_on_different_collations.js ticket: SERVER-93112 + - test_file: jstests/sharding/resharding_collection_cloner_stale_config_retry.js + ticket: SERVER-94151 suites: null last-lts: all: @@ -1554,4 +1556,6 @@ last-lts: ticket: SERVER-93099 - test_file: jstests/core/index/index_creation_on_different_collations.js ticket: SERVER-93112 + - test_file: jstests/sharding/resharding_collection_cloner_stale_config_retry.js + ticket: SERVER-94151 suites: null diff --git a/jstests/sharding/resharding_collection_cloner_stale_config_retry.js b/jstests/sharding/resharding_collection_cloner_stale_config_retry.js new file mode 100644 index 00000000000..400bda679ff --- /dev/null +++ b/jstests/sharding/resharding_collection_cloner_stale_config_retry.js @@ -0,0 +1,106 @@ +/** + * Tests that the reshardingCollectionCloner is resilient to staleConfig errors. + * + * @tags: [ + * uses_atclustertime, + * ] + */ +import {configureFailPoint} from "jstests/libs/fail_point_util.js"; +import {funWithArgs} from "jstests/libs/parallel_shell_helpers.js"; +import {extractUUIDFromObject, getUUIDFromListCollections} from "jstests/libs/uuid_util.js"; +import {CreateShardedCollectionUtil} from "jstests/sharding/libs/create_sharded_collection_util.js"; + +const st = new ShardingTest({mongos: 1, config: 1, shards: 2, rs: {nodes: 1}}); + +const inputCollection = st.s.getCollection("reshardingDb.coll"); + +CreateShardedCollectionUtil.shardCollectionWithChunks(inputCollection, {oldKey: 1}, [ + {min: {oldKey: MinKey}, max: {oldKey: 0}, shard: st.shard0.shardName}, + {min: {oldKey: 0}, max: {oldKey: MaxKey}, shard: st.shard1.shardName}, +]); + +const inputCollectionUUID = + getUUIDFromListCollections(inputCollection.getDB(), inputCollection.getName()); +const inputCollectionUUIDString = extractUUIDFromObject(inputCollectionUUID); + +const temporaryReshardingCollection = + st.s.getCollection(`reshardingDb.system.resharding.${inputCollectionUUIDString}`); + +CreateShardedCollectionUtil.shardCollectionWithChunks(temporaryReshardingCollection, {newKey: 1}, [ + {min: {newKey: MinKey}, max: {newKey: 0}, shard: st.shard0.shardName}, + {min: {newKey: 0}, max: {newKey: MaxKey}, shard: st.shard1.shardName}, +]); + +// The shardCollection command doesn't wait for the config.cache.chunks entries to have been written +// on the primary shard for the database. We manually run the _flushRoutingTableCacheUpdates command +// to guarantee they have been written and are visible with the atClusterTime used by the +// testReshardCloneCollection command. +for (const shard of [st.shard0, st.shard1]) { + assert.commandWorked(shard.rs.getPrimary().adminCommand( + {_flushRoutingTableCacheUpdates: temporaryReshardingCollection.getFullName()})); +} + +const documents = [ + {_id: "a0", info: "stays on shard0", oldKey: -10, newKey: -100}, + {_id: "a1", info: "stays on shard0", oldKey: -11, newKey: -101}, + {_id: "a2", info: "stays on shard0", oldKey: -12, newKey: -102}, + {_id: "b", info: "shard1 moves to shard0", oldKey: 10, newKey: -10}, + {_id: "b1", info: "shard1 moves to shard0", oldKey: 11, newKey: -10}, + {_id: "b2", info: "shard1 moves to shard0", oldKey: 12, newKey: -10}, + {_id: "c", info: "stays on shard1", oldKey: -10, newKey: -110}, + {_id: "c2", info: "stays on shard1", oldKey: -20, newKey: -120}, +]; +assert.commandWorked(inputCollection.insert(documents)); +const originalInsertsTs = inputCollection.getDB().getSession().getOperationTime(); + +for (const shard of [st.shard0, st.shard1]) { + shard.rs.getPrimary() + .getDB(inputCollection.getDB().getName()) + .getSession() + .advanceClusterTime(inputCollection.getDB().getSession().getClusterTime()); +} + +const shard0Primary = st.shard0.rs.getPrimary(); +const outerLoop = configureFailPoint( + shard0Primary, "reshardingCollectionClonerPauseBeforeWriteNaturalOrder", {}, {skip: 1}); + +const staleConfigFP = + configureFailPoint(shard0Primary, "reshardingCollectionClonerShouldFailWithStaleConfig"); + +const reshardShell = + startParallelShell(funWithArgs( + (inputCollectionFullName, + inputCollectionUUID, + shardName, + atClusterTime, + tempCollectionFullName) => { + assert.commandWorked(db.adminCommand({ + testReshardCloneCollection: inputCollectionFullName, + shardKey: {newKey: 1}, + uuid: inputCollectionUUID, + shardId: shardName, + atClusterTime: atClusterTime, + outputNs: tempCollectionFullName, + })); + }, + inputCollection.getFullName(), + inputCollectionUUID, + st.shard0.shardName, + originalInsertsTs, + temporaryReshardingCollection.getFullName()), + shard0Primary.port); + +// Wait until the retry of _writeOnceWithNaturalOrder to turn off the staleConfig fp. +outerLoop.wait(); +staleConfigFP.off(); +outerLoop.off(); + +reshardShell(); + +assert.eq(documents.length, + st.s.getCollection(temporaryReshardingCollection.getFullName()).countDocuments({})); + +// The temporary reshard collection must be dropped before checking metadata integrity. +assert(temporaryReshardingCollection.drop()); + +st.stop(); diff --git a/src/mongo/db/s/resharding/resharding_collection_cloner.cpp b/src/mongo/db/s/resharding/resharding_collection_cloner.cpp index b3e89a0a59f..ff482eb55ef 100644 --- a/src/mongo/db/s/resharding/resharding_collection_cloner.cpp +++ b/src/mongo/db/s/resharding/resharding_collection_cloner.cpp @@ -104,6 +104,8 @@ MONGO_FAIL_POINT_DEFINE(reshardingCollectionClonerAbort); MONGO_FAIL_POINT_DEFINE(reshardingCollectionClonerPauseBeforeAttempt); +MONGO_FAIL_POINT_DEFINE(reshardingCollectionClonerShouldFailWithStaleConfig); +MONGO_FAIL_POINT_DEFINE(reshardingCollectionClonerPauseBeforeWriteNaturalOrder); namespace mongo { namespace { @@ -304,21 +306,21 @@ public: ReshardingCloneFetcher(std::shared_ptr<executor::TaskExecutor> executor, std::shared_ptr<executor::TaskExecutor> cleanupExecutor, CancellationToken cancelToken, - sharded_agg_helpers::DispatchShardPipelineResults dispatchResults, + std::vector<OwnedRemoteCursor> remoteCursors, int batchSizeLimitBytes, int numWriteThreads) : _executor(std::move(executor)), _cleanupExecutor(std::move(cleanupExecutor)), _cancelSource(cancelToken), _factory(_cancelSource.token(), _executor), - _dispatchResults(std::move(dispatchResults)), + _remoteCursors(std::move(remoteCursors)), _numWriteThreads(numWriteThreads), _queues(_numWriteThreads), _activeCursors(0), _openConsumers(0) { constexpr int kQueueDepthPerDonor = 2; MultiProducerSingleConsumerQueue<QueueData>::Options qOptions; - qOptions.maxQueueDepth = _dispatchResults.remoteCursors.size() * kQueueDepthPerDonor; + qOptions.maxQueueDepth = _remoteCursors.size() * kQueueDepthPerDonor; for (auto& queue : _queues) { queue.emplace(qOptions); } @@ -431,16 +433,15 @@ public: } void setupReaderThreads(OperationContext* opCtx) { - auto& remoteCursors = _dispatchResults.remoteCursors; // Network commands can start immediately, so reserve here to avoid the // vector being resized while setting up. - _shardIds.reserve(remoteCursors.size()); - for (int i = 0; i < int(remoteCursors.size()); i++) { + _shardIds.reserve(_remoteCursors.size()); + for (int i = 0; i < int(_remoteCursors.size()); i++) { { std::lock_guard lk(_mutex); _activeCursors++; } - auto& cursor = _dispatchResults.remoteCursors[i]; + auto& cursor = _remoteCursors[i]; GetMoreCommandRequest getMoreRequest( cursor->getCursorResponse().getCursorId(), cursor->getCursorResponse().getNSS().coll().toString()); @@ -540,7 +541,7 @@ private: std::shared_ptr<executor::TaskExecutor> _cleanupExecutor; CancellationSource _cancelSource; CancelableOperationContextFactory _factory; - sharded_agg_helpers::DispatchShardPipelineResults _dispatchResults; + std::vector<OwnedRemoteCursor> _remoteCursors; int _numWriteThreads; std::vector<ExecutorFuture<executor::TaskExecutor::ResponseStatus>> _cmdFutures; std::vector<ExecutorFuture<void>> _writerFutures; @@ -658,20 +659,6 @@ ReshardingCollectionCloner::_queryOnceWithNaturalOrder( std::move(resumeTokenMap), std::move(shardsToSkip)); - return dispatchResults; -} - -void ReshardingCollectionCloner::_writeOnceWithNaturalOrder( - OperationContext* opCtx, - std::shared_ptr<executor::TaskExecutor> executor, - std::shared_ptr<executor::TaskExecutor> cleanupExecutor, - CancellationToken cancelToken, - sharded_agg_helpers::DispatchShardPipelineResults& dispatchResults) { - // If we don't establish any cursors, there is no work to do. Return. - if (dispatchResults.remoteCursors.empty()) { - return; - } - bool hasSplitPipeline = !!dispatchResults.splitPipeline; std::string shardsPipelineStr; std::string mergePipelineStr; @@ -700,13 +687,34 @@ void ReshardingCollectionCloner::_writeOnceWithNaturalOrder( "numProducers"_attr = dispatchResults.numProducers, "hasExchangeSpec"_attr = dispatchResults.exchangeSpec != boost::none); + return dispatchResults; +} + +void ReshardingCollectionCloner::_writeOnceWithNaturalOrder( + OperationContext* opCtx, + std::shared_ptr<executor::TaskExecutor> executor, + std::shared_ptr<executor::TaskExecutor> cleanupExecutor, + CancellationToken cancelToken, + std::vector<OwnedRemoteCursor> remoteCursors) { ReshardingCloneFetcher reshardingCloneFetcher( std::move(executor), std::move(cleanupExecutor), cancelToken, - std::move(dispatchResults), + std::move(remoteCursors), resharding::gReshardingCollectionClonerBatchSizeInBytes.load(), resharding::gReshardingCollectionClonerWriteThreadCount); + + if (reshardingCollectionClonerShouldFailWithStaleConfig.shouldFail()) { + uassert(StaleConfigInfo(_sourceNss, + ShardVersionFactory::make(ChunkVersion::IGNORED(), + boost::optional<CollectionIndexes>( + boost::none)) /* receivedVersion */, + boost::none /* wantedVersion */, + ShardId{"0"}), + str::stream() << "Throwing staleConfig for reshardingCollectionCloner failpoint.", + false); + } + reshardingCloneFetcher .run(opCtx, [this](OperationContext* opCtx, @@ -742,15 +750,29 @@ void ReshardingCollectionCloner::_runOnceWithNaturalOrder( std::shared_ptr<executor::TaskExecutor> executor, std::shared_ptr<executor::TaskExecutor> cleanupExecutor, CancellationToken cancelToken) { - auto dispatchResults = shardVersionRetry( - opCtx, - Grid::get(opCtx)->catalogCache(), - _sourceNss, - "resharding collection cloner fetching with natural order (query stage)"_sd, - [&] { return _queryOnceWithNaturalOrder(opCtx, mongoProcessInterface); }); - + // We can run into StaleConfig errors when cloning collections. To make it + // safe during retry, we retry the whole cloning process and rely on the + // resume token to be correct. Note that the remote cursors need to be reestablished during + // retry since _writeOnceWithNaturalOrder can partially or completely consume them. resharding::data_copy::withOneStaleConfigRetry(opCtx, [&] { - _writeOnceWithNaturalOrder(opCtx, executor, cleanupExecutor, cancelToken, dispatchResults); + auto dispatchResults = shardVersionRetry( + opCtx, + Grid::get(opCtx)->catalogCache(), + _sourceNss, + "resharding collection cloner fetching with natural order (query stage)"_sd, + [&] { return _queryOnceWithNaturalOrder(opCtx, mongoProcessInterface); }); + + // If we don't establish any cursors, there is no work to do. Return. + if (dispatchResults.remoteCursors.empty()) { + return; + } + + reshardingCollectionClonerPauseBeforeWriteNaturalOrder.pauseWhileSet(); + _writeOnceWithNaturalOrder(opCtx, + executor, + cleanupExecutor, + cancelToken, + std::move(dispatchResults.remoteCursors)); }); } @@ -890,9 +912,6 @@ SemiFuture<void> ReshardingCollectionCloner::run( reshardingCollectionClonerPauseBeforeAttempt.pauseWhileSet(); if (reshardingImprovementsEnabled) { auto opCtx = factory.makeOperationContext(&cc()); - // We can run into StaleConfig errors when cloning collections. To make it - // safer during retry, we retry the whole cloning process and rely on the - // resume token to be correct. _runOnceWithNaturalOrder(opCtx.get(), MongoProcessInterface::create(opCtx.get()), executor, diff --git a/src/mongo/db/s/resharding/resharding_collection_cloner.h b/src/mongo/db/s/resharding/resharding_collection_cloner.h index 117a52c52ec..55bc256bf4d 100644 --- a/src/mongo/db/s/resharding/resharding_collection_cloner.h +++ b/src/mongo/db/s/resharding/resharding_collection_cloner.h @@ -133,12 +133,11 @@ private: sharded_agg_helpers::DispatchShardPipelineResults _queryOnceWithNaturalOrder( OperationContext* opCtx, std::shared_ptr<MongoProcessInterface> mongoProcessInterface); - void _writeOnceWithNaturalOrder( - OperationContext* opCtx, - std::shared_ptr<executor::TaskExecutor> executor, - std::shared_ptr<executor::TaskExecutor> cleanupExecutor, - CancellationToken cancelToken, - sharded_agg_helpers::DispatchShardPipelineResults& dispatchResults); + void _writeOnceWithNaturalOrder(OperationContext* opCtx, + std::shared_ptr<executor::TaskExecutor> executor, + std::shared_ptr<executor::TaskExecutor> cleanupExecutor, + CancellationToken cancelToken, + std::vector<OwnedRemoteCursor> remoteCursors); void _runOnceWithNaturalOrder(OperationContext* opCtx, std::shared_ptr<MongoProcessInterface> mongoProcessInterface, |