diff options
Diffstat (limited to 'src/mongo/db/s/config')
24 files changed, 521 insertions, 1372 deletions
diff --git a/src/mongo/db/s/config/configsvr_abort_reshard_collection_command.cpp b/src/mongo/db/s/config/configsvr_abort_reshard_collection_command.cpp index 348e4bc2b40..9cb503a2979 100644 --- a/src/mongo/db/s/config/configsvr_abort_reshard_collection_command.cpp +++ b/src/mongo/db/s/config/configsvr_abort_reshard_collection_command.cpp @@ -76,7 +76,7 @@ void assertExistsReshardingDocument(OperationContext* opCtx, UUID reshardingUUID } auto assertGetReshardingMachine(OperationContext* opCtx, UUID reshardingUUID) { - auto machine = resharding::tryGetReshardingStateMachineAndThrowIfShuttingDown< + auto machine = resharding::tryGetReshardingStateMachine< ReshardingCoordinatorService, ReshardingCoordinatorService::ReshardingCoordinator, ReshardingCoordinatorDocument>(opCtx, reshardingUUID); diff --git a/src/mongo/db/s/config/configsvr_commit_move_primary_command.cpp b/src/mongo/db/s/config/configsvr_commit_move_primary_command.cpp deleted file mode 100644 index 76fc3f7b5c9..00000000000 --- a/src/mongo/db/s/config/configsvr_commit_move_primary_command.cpp +++ /dev/null @@ -1,102 +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. - */ - -#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kSharding - -#include "mongo/db/auth/authorization_session.h" -#include "mongo/db/commands.h" -#include "mongo/db/repl/read_concern_args.h" -#include "mongo/db/s/config/sharding_catalog_manager.h" -#include "mongo/s/request_types/move_primary_gen.h" - -namespace mongo { -namespace { - -class ConfigsvrCommitMovePrimaryCommand final - : public TypedCommand<ConfigsvrCommitMovePrimaryCommand> { -public: - using Request = ConfigsvrCommitMovePrimary; - - class Invocation final : public InvocationBase { - public: - using InvocationBase::InvocationBase; - - void typedRun(OperationContext* opCtx) { - uassert(ErrorCodes::IllegalOperation, - str::stream() << Request::kCommandName << " can only be run on config servers", - serverGlobalParams.clusterRole == ClusterRole::ConfigServer); - - // Set the operation context read concern level to local for reads into the config - // database. - repl::ReadConcernArgs::get(opCtx) = - repl::ReadConcernArgs(repl::ReadConcernLevel::kLocalReadConcern); - - ShardingCatalogManager::get(opCtx)->commitMovePrimary( - opCtx, - request().getCommandParameter(), - request().getExpectedDatabaseVersion(), - request().getTo()); - } - - private: - NamespaceString ns() const override { - return NamespaceString(request().getDbName()); - } - - bool supportsWriteConcern() const override { - return true; - } - - void doCheckAuthorization(OperationContext* opCtx) const override { - uassert(ErrorCodes::Unauthorized, - "Unauthorized", - AuthorizationSession::get(opCtx->getClient()) - ->isAuthorizedForActionsOnResource(ResourcePattern::forClusterResource(), - ActionType::internal)); - } - }; - -private: - std::string help() const override { - return "Reassign a new primary shard for the given database on the config server. This is " - "an internal command only invokable on the config server, therefore do not call " - "directly."; - } - - bool adminOnly() const override { - return true; - } - - AllowedOnSecondary secondaryAllowed(ServiceContext* context) const override { - return AllowedOnSecondary::kNever; - } -} configsvrCommitMovePrimaryCommand; - -} // namespace -} // namespace mongo diff --git a/src/mongo/db/s/config/configsvr_commit_reshard_collection_command.cpp b/src/mongo/db/s/config/configsvr_commit_reshard_collection_command.cpp index 0e5c9f803c9..fb67df820c4 100644 --- a/src/mongo/db/s/config/configsvr_commit_reshard_collection_command.cpp +++ b/src/mongo/db/s/config/configsvr_commit_reshard_collection_command.cpp @@ -84,7 +84,7 @@ public: UUID reshardingUUID = retrieveReshardingUUID(opCtx, ns()); - auto machine = resharding::tryGetReshardingStateMachineAndThrowIfShuttingDown< + auto machine = resharding::tryGetReshardingStateMachine< ReshardingCoordinatorService, ReshardingCoordinatorService::ReshardingCoordinator, ReshardingCoordinatorDocument>(opCtx, reshardingUUID); diff --git a/src/mongo/db/s/config/configsvr_configure_collection_balancing.cpp b/src/mongo/db/s/config/configsvr_configure_collection_balancing.cpp index c471c70437f..a9db28064b5 100644 --- a/src/mongo/db/s/config/configsvr_configure_collection_balancing.cpp +++ b/src/mongo/db/s/config/configsvr_configure_collection_balancing.cpp @@ -36,7 +36,6 @@ #include "mongo/db/auth/authorization_session.h" #include "mongo/db/auth/privilege.h" #include "mongo/db/commands.h" -#include "mongo/db/commands/feature_compatibility_version.h" #include "mongo/db/operation_context.h" #include "mongo/db/repl/repl_client_info.h" #include "mongo/db/s/balancer/balancer.h" @@ -61,16 +60,6 @@ public: void typedRun(OperationContext* opCtx) { opCtx->setAlwaysInterruptAtStepDownOrUp(); - - // Hold the FCV region to serialize with the setFeatureCompatibilityVersion command - FixedFCVRegion fcvRegion(opCtx); - uassert(ErrorCodes::IllegalOperation, - "_configsvrConfigureCollectionBalancing can only be run when the cluster is in " - "feature " - "compatibility versions greater or equal than 5.3.", - serverGlobalParams.featureCompatibility.isGreaterThanOrEqualTo( - multiversion::FeatureCompatibilityVersion::kVersion_5_3)); - uassert(ErrorCodes::IllegalOperation, str::stream() << Request::kCommandName << " can only be run on config servers", serverGlobalParams.clusterRole == ClusterRole::ConfigServer); diff --git a/src/mongo/db/s/config/configsvr_merge_chunks_command.cpp b/src/mongo/db/s/config/configsvr_merge_chunks_command.cpp index 5df553b5bd3..ea2823dcdf0 100644 --- a/src/mongo/db/s/config/configsvr_merge_chunks_command.cpp +++ b/src/mongo/db/s/config/configsvr_merge_chunks_command.cpp @@ -94,7 +94,8 @@ public: request().getTimestamp(), request().getCollectionUUID(), request().getChunkRange(), - request().getShard())); + request().getShard(), + request().getValidAfter())); return ConfigSvrMergeResponse{ChunkVersion::fromBSONPositionalOrNewerFormat( shardAndCollVers[ChunkVersion::kShardVersionField])}; } diff --git a/src/mongo/db/s/config/configsvr_reshard_collection_cmd.cpp b/src/mongo/db/s/config/configsvr_reshard_collection_cmd.cpp index 07f1248a8c5..f8a12c57f90 100644 --- a/src/mongo/db/s/config/configsvr_reshard_collection_cmd.cpp +++ b/src/mongo/db/s/config/configsvr_reshard_collection_cmd.cpp @@ -193,12 +193,6 @@ public: if (auto zones = request().getZones()) { checkForOverlappingZones(*zones); - for (const auto& zone : *zones) { - uassertStatusOK( - ShardKeyPattern::checkShardKeyIsValidForMetadataStorage(zone.getMin())); - uassertStatusOK( - ShardKeyPattern::checkShardKeyIsValidForMetadataStorage(zone.getMax())); - } } auto coordinatorDoc = diff --git a/src/mongo/db/s/config/configsvr_run_restore_command.cpp b/src/mongo/db/s/config/configsvr_run_restore_command.cpp index 22e3f7b2985..166c10a4e70 100644 --- a/src/mongo/db/s/config/configsvr_run_restore_command.cpp +++ b/src/mongo/db/s/config/configsvr_run_restore_command.cpp @@ -73,25 +73,6 @@ ShouldRestoreDocument shouldRestoreDocument(OperationContext* opCtx, : ShouldRestoreDocument::kNo; } -std::set<std::string> getDatabasesToRestore(OperationContext* opCtx) { - auto findRequest = FindCommandRequest(NamespaceString::kConfigsvrRestoreNamespace); - - std::set<std::string> databasesToRestore; - DBDirectClient client(opCtx); - auto it = client.find(findRequest); - while (it->more()) { - const auto doc = it->next(); - if (!doc.hasField("ns")) { - continue; - } - - NamespaceString nss(doc.getStringField("ns")); - databasesToRestore.emplace(nss.db()); - } - - return databasesToRestore; -} - // Modifications to this map should add new testing in 'sharded_backup_restore.js'. // { config collection namespace -> ( optional nss field name, optional UUID field name ) } const stdx::unordered_map<NamespaceString, @@ -165,7 +146,7 @@ public: // Keeps track of database names for collections restored. Databases with no collections // restored will have their entries removed in the config collections. - std::set<std::string> databasesRestored = getDatabasesToRestore(opCtx); + std::set<std::string> databasesRestored; for (const auto& collectionEntry : kCollectionEntries) { const NamespaceString& nss = collectionEntry.first; @@ -219,6 +200,10 @@ public: "doc"_attr = doc, "shouldRestore"_attr = shouldRestore); + if (shouldRestore == ShouldRestoreDocument::kYes && docNss) { + databasesRestored.insert(docNss->db().toString()); + } + if (shouldRestore == ShouldRestoreDocument::kYes || shouldRestore == ShouldRestoreDocument::kMaybe) { continue; diff --git a/src/mongo/db/s/config/configsvr_set_cluster_parameter_command.cpp b/src/mongo/db/s/config/configsvr_set_cluster_parameter_command.cpp index 24030e5fe57..da76faa6593 100644 --- a/src/mongo/db/s/config/configsvr_set_cluster_parameter_command.cpp +++ b/src/mongo/db/s/config/configsvr_set_cluster_parameter_command.cpp @@ -61,11 +61,6 @@ public: const auto coordinatorCompletionFuture = [&]() -> SharedSemiFuture<void> { FixedFCVRegion fcvRegion(opCtx); - uassert(ErrorCodes::UnknownFeatureCompatibilityVersion, - "FCV is not yet initialized, retry the command after FCV initialization " - "has completed", - serverGlobalParams.featureCompatibility.isVersionInitialized()); - uassert(ErrorCodes::IllegalOperation, "featureFlagClusterWideConfig not enabled", gFeatureFlagClusterWideConfig.isEnabled( diff --git a/src/mongo/db/s/config/initial_split_policy.cpp b/src/mongo/db/s/config/initial_split_policy.cpp index 15c5a345c59..5bdfb55d6c3 100644 --- a/src/mongo/db/s/config/initial_split_policy.cpp +++ b/src/mongo/db/s/config/initial_split_policy.cpp @@ -37,14 +37,12 @@ #include "mongo/db/bson/dotted_path_support.h" #include "mongo/db/catalog/collection_catalog.h" #include "mongo/db/curop.h" -#include "mongo/db/pipeline/document_source.h" #include "mongo/db/pipeline/lite_parsed_pipeline.h" #include "mongo/db/pipeline/process_interface/shardsvr_process_interface.h" #include "mongo/db/pipeline/sharded_agg_helpers.h" #include "mongo/db/s/balancer/balancer_policy.h" #include "mongo/db/s/sharding_state.h" #include "mongo/db/vector_clock.h" -#include "mongo/logv2/log.h" #include "mongo/s/balancer_configuration.h" #include "mongo/s/catalog/type_shard.h" #include "mongo/s/grid.h" @@ -56,30 +54,12 @@ namespace { using ChunkDistributionMap = stdx::unordered_map<ShardId, size_t>; using ZoneShardMap = StringMap<std::vector<ShardId>>; -using boost::intrusive_ptr; - -std::vector<ShardId> getAllNonDrainingShardIdsSorted(OperationContext* opCtx) { - const auto shardsAndOpTime = uassertStatusOKWithContext( - Grid::get(opCtx)->catalogClient()->getAllShards( - opCtx, repl::ReadConcernLevel::kMajorityReadConcern, true /* excludeDraining */), - "Cannot retrieve updated shard list from config server"); - const auto shards = std::move(shardsAndOpTime.value); - const auto lastVisibleOpTime = std::move(shardsAndOpTime.opTime); - - LOGV2_DEBUG(6566600, - 1, - "Successfully retrieved updated shard list from config server", - "nonDrainingShardsNumber"_attr = shards.size(), - "lastVisibleOpTime"_attr = lastVisibleOpTime); - - std::vector<ShardId> shardIds; - std::transform(shards.begin(), - shards.end(), - std::back_inserter(shardIds), - [](const ShardType& shard) { return ShardId(shard.getName()); }); +std::vector<ShardId> getAllShardIdsSorted(OperationContext* opCtx) { + // Many tests assume that chunks will be placed on shards + // according to their IDs in ascending lexical order. + auto shardIds = Grid::get(opCtx)->shardRegistry()->getAllShardIdsNoReload(); std::sort(shardIds.begin(), shardIds.end()); - return shardIds; } @@ -286,8 +266,7 @@ std::unique_ptr<InitialSplitPolicy> InitialSplitPolicy::calculateOptimizationStr const boost::optional<std::vector<BSONObj>>& initialSplitPoints, const std::vector<TagsType>& tags, size_t numShards, - bool collectionIsEmpty, - bool useAutoSplitter) { + bool collectionIsEmpty) { uassert(ErrorCodes::InvalidOptions, str::stream() << "numInitialChunks is only supported when the collection is empty " "and has a hashed field in the shard key pattern", @@ -331,11 +310,7 @@ std::unique_ptr<InitialSplitPolicy> InitialSplitPolicy::calculateOptimizationStr return std::make_unique<SingleChunkOnPrimarySplitPolicy>(); } - if (useAutoSplitter) { - return std::make_unique<AutoSplitInChunksOnPrimaryPolicy>(); - } - - return std::make_unique<SingleChunkOnPrimarySplitPolicy>(); + return std::make_unique<UnoptimizedSplitPolicy>(); } InitialSplitPolicy::ShardCollectionConfig SingleChunkOnPrimarySplitPolicy::createFirstChunks( @@ -359,7 +334,7 @@ InitialSplitPolicy::ShardCollectionConfig SingleChunkOnPrimarySplitPolicy::creat return {std::move(chunks)}; } -InitialSplitPolicy::ShardCollectionConfig AutoSplitInChunksOnPrimaryPolicy::createFirstChunks( +InitialSplitPolicy::ShardCollectionConfig UnoptimizedSplitPolicy::createFirstChunks( OperationContext* opCtx, const ShardKeyPattern& shardKeyPattern, const SplitPolicyParams& params) { @@ -397,7 +372,7 @@ InitialSplitPolicy::ShardCollectionConfig SplitPointsBasedSplitPolicy::createFir const SplitPolicyParams& params) { // On which shards are the generated chunks allowed to be placed. - const auto shardIds = getAllNonDrainingShardIdsSorted(opCtx); + const auto shardIds = getAllShardIdsSorted(opCtx); const auto currentTime = VectorClock::get(opCtx)->getTime(); const auto validAfter = currentTime.clusterTime().asTimestamp(); @@ -428,7 +403,7 @@ InitialSplitPolicy::ShardCollectionConfig AbstractTagsBasedSplitPolicy::createFi const SplitPolicyParams& params) { invariant(!_tags.empty()); - const auto shardIds = getAllNonDrainingShardIdsSorted(opCtx); + const auto shardIds = getAllShardIdsSorted(opCtx); const auto currentTime = VectorClock::get(opCtx)->getTime(); const auto validAfter = currentTime.clusterTime().asTimestamp(); const auto& keyPattern = shardKeyPattern.getKeyPattern(); @@ -675,29 +650,34 @@ std::vector<BSONObj> ReshardingSplitPolicy::createRawPipeline(const ShardKeyPatt std::vector<BSONObj> res; const auto& shardKeyFields = shardKey.getKeyPatternFields(); + + BSONObjBuilder projectValBuilder; BSONObjBuilder sortValBuilder; - using Doc = Document; - using Arr = std::vector<Value>; - using V = Value; - Arr arrayToObjectBuilder; + for (auto&& fieldRef : shardKeyFields) { // If the shard key includes a hashed field and current fieldRef is the hashed field. if (shardKey.isHashedPattern() && fieldRef->dottedField().compare(shardKey.getHashedField().fieldNameStringData()) == 0) { - arrayToObjectBuilder.emplace_back( - Doc{{"k", V{fieldRef->dottedField()}}, - {"v", Doc{{"$toHashedIndexKey", V{"$" + fieldRef->dottedField()}}}}}); + projectValBuilder.append(fieldRef->dottedField(), + BSON("$toHashedIndexKey" + << "$" + fieldRef->dottedField())); } else { - arrayToObjectBuilder.emplace_back(Doc{ - {"k", V{fieldRef->dottedField()}}, - {"v", Doc{{"$ifNull", V{Arr{V{"$" + fieldRef->dottedField()}, V{BSONNULL}}}}}}}); + projectValBuilder.append( + str::stream() << fieldRef->dottedField(), + BSON("$ifNull" << BSON_ARRAY("$" + fieldRef->dottedField() << BSONNULL))); } + sortValBuilder.append(fieldRef->dottedField().toString(), 1); } + + // Do not project _id if it's not part of the shard key. + if (!shardKey.hasId()) { + projectValBuilder.append("_id", 0); + } + res.push_back(BSON("$sample" << BSON("size" << numSplitPoints * samplesPerChunk))); + res.push_back(BSON("$project" << projectValBuilder.obj())); res.push_back(BSON("$sort" << sortValBuilder.obj())); - res.push_back( - Doc{{"$replaceWith", Doc{{"$arrayToObject", Arr{V{arrayToObjectBuilder}}}}}}.toBson()); return res; } @@ -763,12 +743,12 @@ InitialSplitPolicy::ShardCollectionConfig ReshardingSplitPolicy::createFirstChun } { - auto shardIds = getAllNonDrainingShardIdsSorted(opCtx); - for (const auto& shard : shardIds) { + auto allShardIds = getAllShardIdsSorted(opCtx); + for (const auto& shard : allShardIds) { chunkDistribution.emplace(shard, 0); } - zoneToShardMap.emplace("", std::move(shardIds)); + zoneToShardMap.emplace("", std::move(allShardIds)); } std::vector<ChunkType> chunks; @@ -820,34 +800,26 @@ void ReshardingSplitPolicy::_appendSplitPointsFromSample(BSONObjSet* splitPoints while (nextKey && nRemaining > 0) { // if key is hashed, nextKey values are already hashed - auto result = splitPoints->insert(nextKey->getOwned()); + auto result = splitPoints->insert( + dotted_path_support::extractElementsBasedOnTemplate(*nextKey, shardKey.toBSON()) + .getOwned()); + if (result.second) { nRemaining--; } + nextKey = _samples->getNext(); } } std::unique_ptr<ReshardingSplitPolicy::SampleDocumentSource> -ReshardingSplitPolicy::makePipelineDocumentSource_forTest(OperationContext* opCtx, - const NamespaceString& ns, - const ShardKeyPattern& shardKey, - int numInitialChunks, - int samplesPerChunk) { - MakePipelineOptions opts; - opts.attachCursorSource = false; - return _makePipelineDocumentSource( - opCtx, ns, shardKey, numInitialChunks, samplesPerChunk, std::move(opts)); -} - -std::unique_ptr<ReshardingSplitPolicy::SampleDocumentSource> ReshardingSplitPolicy::_makePipelineDocumentSource(OperationContext* opCtx, const NamespaceString& ns, const ShardKeyPattern& shardKey, int numInitialChunks, - int samplesPerChunk, - MakePipelineOptions opts) { + int samplesPerChunk) { auto rawPipeline = createRawPipeline(shardKey, numInitialChunks - 1, samplesPerChunk); + StringMap<ExpressionContext::ResolvedNamespace> resolvedNamespaces; resolvedNamespaces[ns.coll()] = {ns, std::vector<BSONObj>{}}; @@ -861,7 +833,7 @@ ReshardingSplitPolicy::_makePipelineDocumentSource(OperationContext* opCtx, boost::none, /* explain */ false, /* fromMongos */ false, /* needsMerge */ - true, /* allowDiskUse */ + false, /* allowDiskUse */ true, /* bypassDocumentValidation */ false, /* isMapReduceCommand */ ns, @@ -871,10 +843,8 @@ ReshardingSplitPolicy::_makePipelineDocumentSource(OperationContext* opCtx, std::move(resolvedNamespaces), boost::none); /* collUUID */ - expCtx->tempDir = storageGlobalParams.dbpath + "/tmp"; - - return std::make_unique<PipelineDocumentSource>( - Pipeline::makePipeline(rawPipeline, expCtx, opts), samplesPerChunk - 1); + return std::make_unique<PipelineDocumentSource>(Pipeline::makePipeline(rawPipeline, expCtx, {}), + samplesPerChunk - 1); } ReshardingSplitPolicy::PipelineDocumentSource::PipelineDocumentSource( diff --git a/src/mongo/db/s/config/initial_split_policy.h b/src/mongo/db/s/config/initial_split_policy.h index e492e9c4cb2..ced8d519a13 100644 --- a/src/mongo/db/s/config/initial_split_policy.h +++ b/src/mongo/db/s/config/initial_split_policy.h @@ -40,6 +40,7 @@ #include "mongo/s/shard_id.h" #include "mongo/s/shard_key_pattern.h" #include "mongo/util/string_map.h" + namespace mongo { struct SplitPolicyParams { @@ -52,11 +53,6 @@ public: /** * Returns the optimization strategy for building initial chunks based on the input parameters * and the collection state. - * - * The 'useAutoSplitter' flag indicates to the initial split strategy selected that in case the - * collection contains data, it should use the auto splitter to chop that data into chunks - * respective to the configured chunk size. If set to false, the policy will create as large of - * chunks as possible. */ static std::unique_ptr<InitialSplitPolicy> calculateOptimizationStrategy( OperationContext* opCtx, @@ -66,8 +62,7 @@ public: const boost::optional<std::vector<BSONObj>>& initialSplitPoints, const std::vector<TagsType>& tags, size_t numShards, - bool collectionIsEmpty, - bool useAutoSplitter = true /* Controlled by FCV, see the comment */); + bool collectionIsEmpty); virtual ~InitialSplitPolicy() {} @@ -145,7 +140,7 @@ public: * Split point building strategy to be used when no optimizations are available. We send a * splitVector command to the primary shard in order to calculate the appropriate split points. */ -class AutoSplitInChunksOnPrimaryPolicy : public InitialSplitPolicy { +class UnoptimizedSplitPolicy : public InitialSplitPolicy { public: ShardCollectionConfig createFirstChunks(OperationContext* opCtx, const ShardKeyPattern& shardKeyPattern, @@ -290,7 +285,6 @@ public: public: virtual ~SampleDocumentSource(){}; virtual boost::optional<BSONObj> getNext() = 0; - virtual Pipeline* getPipeline_forTest() = 0; }; // Provides documents from a real Pipeline @@ -299,9 +293,6 @@ public: PipelineDocumentSource() = delete; PipelineDocumentSource(SampleDocumentPipeline pipeline, int skip); boost::optional<BSONObj> getNext() override; - Pipeline* getPipeline_forTest() override { - return _pipeline.get(); - } private: SampleDocumentPipeline _pipeline; @@ -337,21 +328,13 @@ public: static constexpr int kDefaultSamplesPerChunk = 10; - static std::unique_ptr<SampleDocumentSource> makePipelineDocumentSource_forTest( - OperationContext* opCtx, - const NamespaceString& ns, - const ShardKeyPattern& shardKey, - int numInitialChunks, - int samplesPerChunk); - private: static std::unique_ptr<SampleDocumentSource> _makePipelineDocumentSource( OperationContext* opCtx, const NamespaceString& ns, const ShardKeyPattern& shardKey, int numInitialChunks, - int samplesPerChunk, - MakePipelineOptions opts = {}); + int samplesPerChunk); /** * Returns a set of split points to ensure that chunk boundaries will align with the zone diff --git a/src/mongo/db/s/config/initial_split_policy_test.cpp b/src/mongo/db/s/config/initial_split_policy_test.cpp index 0ef3bee06d1..75d1f2ae0a1 100644 --- a/src/mongo/db/s/config/initial_split_policy_test.cpp +++ b/src/mongo/db/s/config/initial_split_policy_test.cpp @@ -35,7 +35,6 @@ #include "mongo/db/s/config/config_server_test_fixture.h" #include "mongo/db/s/config/initial_split_policy.h" #include "mongo/db/vector_clock.h" -#include "mongo/logv2/log.h" #include "mongo/s/catalog/type_shard.h" #include "mongo/s/catalog/type_tags.h" #include "mongo/unittest/unittest.h" @@ -1728,10 +1727,6 @@ public: return next; } - Pipeline* getPipeline_forTest() override { - return nullptr; - } - private: std::list<BSONObj> _toReturn; }; @@ -1770,9 +1765,9 @@ TEST_F(ReshardingInitSplitTest, NoZones) { shardRegistry()->reload(operationContext()); std::list<BSONObj> mockSamples; - mockSamples.push_back(BSON("y" << 10)); - mockSamples.push_back(BSON("y" << 20)); - mockSamples.push_back(BSON("y" << 30)); + mockSamples.push_back(BSON("x" << 10 << "y" << 10)); + mockSamples.push_back(BSON("x" << 10 << "y" << 20)); + mockSamples.push_back(BSON("x" << 10 << "y" << 30)); auto mockSampleSource = std::make_unique<MockPipelineSource>(std::move(mockSamples)); @@ -1807,9 +1802,9 @@ TEST_F(ReshardingInitSplitTest, HashedShardKey) { shardRegistry()->reload(operationContext()); std::list<BSONObj> mockSamples; - mockSamples.push_back(BSON("y" << 7766103514953448109LL)); - mockSamples.push_back(BSON("y" << -9117533237618642180LL)); - mockSamples.push_back(BSON("y" << -1196399207910989725LL)); + mockSamples.push_back(BSON("x" << 10 << "y" << 7766103514953448109LL)); + mockSamples.push_back(BSON("x" << 10 << "y" << -9117533237618642180LL)); + mockSamples.push_back(BSON("x" << 10 << "y" << -1196399207910989725LL)); auto mockSampleSource = std::make_unique<MockPipelineSource>(std::move(mockSamples)); @@ -1872,9 +1867,9 @@ TEST_F(ReshardingInitSplitTest, ZonesCoversEntireDomainButInsufficient) { shardRegistry()->reload(operationContext()); std::list<BSONObj> mockSamples; - mockSamples.push_back(BSON("y" << 10)); - mockSamples.push_back(BSON("y" << 20)); - mockSamples.push_back(BSON("y" << 30)); + mockSamples.push_back(BSON("x" << 10 << "y" << 10)); + mockSamples.push_back(BSON("x" << 10 << "y" << 20)); + mockSamples.push_back(BSON("x" << 10 << "y" << 30)); auto mockSampleSource = std::make_unique<MockPipelineSource>(std::move(mockSamples)); @@ -1912,9 +1907,9 @@ TEST_F(ReshardingInitSplitTest, SamplesCoincidingWithZones) { shardRegistry()->reload(operationContext()); std::list<BSONObj> mockSamples; - mockSamples.push_back(BSON("y" << 10)); - mockSamples.push_back(BSON("y" << 20)); - mockSamples.push_back(BSON("y" << 30)); + mockSamples.push_back(BSON("x" << 10 << "y" << 10)); + mockSamples.push_back(BSON("x" << 10 << "y" << 20)); + mockSamples.push_back(BSON("x" << 10 << "y" << 30)); auto mockSampleSource = std::make_unique<MockPipelineSource>(std::move(mockSamples)); diff --git a/src/mongo/db/s/config/sharding_catalog_manager.cpp b/src/mongo/db/s/config/sharding_catalog_manager.cpp index d380fe58468..6fbb06a0668 100644 --- a/src/mongo/db/s/config/sharding_catalog_manager.cpp +++ b/src/mongo/db/s/config/sharding_catalog_manager.cpp @@ -44,7 +44,6 @@ #include "mongo/db/internal_transactions_feature_flag_gen.h" #include "mongo/db/operation_context.h" #include "mongo/db/ops/write_ops.h" -#include "mongo/db/query/cursor_response.h" #include "mongo/db/query/query_request_helper.h" #include "mongo/db/repl/repl_client_info.h" #include "mongo/db/s/balancer/type_migration.h" @@ -447,12 +446,6 @@ Status ShardingCatalogManager::_initConfigIndexes(OperationContext* opCtx) { return result.withContext("couldn't create lock id index on config db"); } - result = configShard->createIndexOnConfig( - opCtx, LocksType::ConfigNS, BSON(LocksType::process() << 1), !unique); - if (!result.isOK()) { - return result.withContext("couldn't create lock process index on config db"); - } - result = configShard->createIndexOnConfig(opCtx, LocksType::ConfigNS, diff --git a/src/mongo/db/s/config/sharding_catalog_manager.h b/src/mongo/db/s/config/sharding_catalog_manager.h index 8ec8fb02108..070c1d3d78f 100644 --- a/src/mongo/db/s/config/sharding_catalog_manager.h +++ b/src/mongo/db/s/config/sharding_catalog_manager.h @@ -31,7 +31,6 @@ #include "mongo/base/status_with.h" #include "mongo/bson/bsonobj.h" -#include "mongo/client/fetcher.h" #include "mongo/db/auth/authorization_session.h" #include "mongo/db/concurrency/d_concurrency.h" #include "mongo/db/logical_session_cache.h" @@ -272,7 +271,8 @@ public: const boost::optional<Timestamp>& timestamp, const UUID& requestCollectionUUID, const ChunkRange& chunkRange, - const ShardId& shardId); + const ShardId& shardId, + const boost::optional<Timestamp>& validAfter); /** * Updates metadata in config.chunks collection to show the given chunk in its new shard. @@ -348,8 +348,7 @@ public: */ void splitOrMarkJumbo(OperationContext* opCtx, const NamespaceString& nss, - const BSONObj& minKey, - boost::optional<int64_t> optMaxChunkSizeBytes); + const BSONObj& minKey); /** * In a transaction, sets the 'allowMigrations' to the requested state and bumps the collection @@ -397,15 +396,6 @@ public: // # TODO SERVER-63983: remove enableSharding paramter when 6.0 becomes lastLTS bool enableSharding = false); - /** - * Updates the metadata in config.databases collection with the new primary shard for the given - * database. This also advances the database's lastmod. - */ - void commitMovePrimary(OperationContext* opCtx, - const StringData& dbName, - const DatabaseVersion& expectedDbVersion, - const ShardId& toShardId); - // // Collection Operations // @@ -606,11 +596,10 @@ private: * Runs a command against a "shard" that is not yet in the cluster and thus not present in the * ShardRegistry. */ - StatusWith<Shard::CommandResponse> _runCommandForAddShard( - OperationContext* opCtx, - std::shared_ptr<RemoteCommandTargeter> targeter, - StringData dbName, - const BSONObj& cmdObj); + StatusWith<Shard::CommandResponse> _runCommandForAddShard(OperationContext* opCtx, + RemoteCommandTargeter* targeter, + StringData dbName, + const BSONObj& cmdObj); /** * Helper method for running a count command against the config server with appropriate error @@ -647,29 +636,7 @@ private: * Sets the current cluster's user-write blocking state on the shard that is being added. */ void _setUserWriteBlockingStateOnNewShard(OperationContext* opCtx, - std::shared_ptr<RemoteCommandTargeter> targeter); - - using FetcherDocsCallbackFn = std::function<bool(const std::vector<BSONObj>& batch)>; - using FetcherStatusCallbackFn = std::function<void(const Status& status)>; - - /** - * Creates a Fetcher task for fetching documents in the given collection on the given shard. - * After the task is scheduled, applies 'processDocsCallback' to each fetched batch and - * 'processStatusCallback' to the fetch status. - */ - std::unique_ptr<Fetcher> _createFetcher(OperationContext* opCtx, - std::shared_ptr<RemoteCommandTargeter> targeter, - const NamespaceString& nss, - const repl::ReadConcernLevel& readConcernLevel, - FetcherDocsCallbackFn processDocsCallback, - FetcherStatusCallbackFn processStatusCallback); - - /** - * Gets the cluster time keys on the given shard and then saves them locally. - */ - Status _pullClusterTimeKeys(OperationContext* opCtx, - std::shared_ptr<RemoteCommandTargeter> targeter); - + RemoteCommandTargeter* targeter); /** * Given a vector of cluster parameters in disk format, sets them locally. */ @@ -680,14 +647,14 @@ private: * Gets the cluster parameters set on the shard and then saves them locally. */ void _pullClusterParametersFromNewShard(OperationContext* opCtx, - std::shared_ptr<RemoteCommandTargeter> targeter); + RemoteCommandTargeter* targeter); /** * Clean all possible leftover cluster parameters on the new added shard and sets the ones * stored on the config server. */ void _pushClusterParametersToNewShard(OperationContext* opCtx, - std::shared_ptr<RemoteCommandTargeter> targeter, + RemoteCommandTargeter* targeter, const std::vector<BSONObj>& clusterParameters); /** @@ -695,20 +662,7 @@ private: * converting from a replica set to a sharded cluster) or set the cluster parameters stored on * the config server in the newly added shard. */ - void _standardizeClusterParameters(OperationContext* opCtx, - std::shared_ptr<RemoteCommandTargeter> targeter); - - /** - * Execute the merge chunk updates using the internal transaction API. - */ - void _mergeChunksInTransaction(OperationContext* opCtx, - const NamespaceString& nss, - const UUID& collectionUUID, - const ChunkVersion& mergeVersion, - const Timestamp& validAfter, - const ChunkRange& chunkRange, - const ShardId& shardId, - std::shared_ptr<std::vector<ChunkType>> chunksToMerge); + void _standardizeClusterParameters(OperationContext* opCtx, RemoteCommandTargeter* targeter); // The owning service context ServiceContext* const _serviceContext; diff --git a/src/mongo/db/s/config/sharding_catalog_manager_add_shard_test.cpp b/src/mongo/db/s/config/sharding_catalog_manager_add_shard_test.cpp index aa483a030a5..0ed9c76d0f0 100644 --- a/src/mongo/db/s/config/sharding_catalog_manager_add_shard_test.cpp +++ b/src/mongo/db/s/config/sharding_catalog_manager_add_shard_test.cpp @@ -41,13 +41,11 @@ #include "mongo/db/ops/write_ops.h" #include "mongo/db/query/cursor_response.h" #include "mongo/db/repl/replication_coordinator_mock.h" -#include "mongo/db/repl/wait_for_majority_service.h" #include "mongo/db/s/add_shard_cmd_gen.h" #include "mongo/db/s/add_shard_util.h" #include "mongo/db/s/config/config_server_test_fixture.h" #include "mongo/db/s/config/sharding_catalog_manager.h" #include "mongo/db/s/type_shard_identity.h" -#include "mongo/db/time_proof_service.h" #include "mongo/idl/cluster_server_parameter_gen.h" #include "mongo/s/catalog/config_server_version.h" #include "mongo/s/catalog/type_changelog.h" @@ -85,32 +83,25 @@ protected: ASSERT_OK(clusterIdLoader->loadClusterId(operationContext(), repl::ReadConcernLevel::kLocalReadConcern)); _clusterId = clusterIdLoader->getClusterId(); - - WaitForMajorityService::get(getServiceContext()).startup(getServiceContext()); - } - - void tearDown() override { - WaitForMajorityService::get(getServiceContext()).shutDown(); - ConfigServerTestFixture::tearDown(); } /** - * addShard validates the host as a shard. It calls "hello" on the host to determine what + * addShard validates the host as a shard. It calls "isMaster" on the host to determine what * kind of host it is -- mongos, regular mongod, config mongod -- and whether the replica set - * details are correct. "helloResponse" defines the response of the "hello" request and + * details are correct. "isMasterResponse" defines the response of the "isMaster" request and * should be a command response BSONObj, or a failed Status. * * ShardingTestFixture::expectGetShards() should be called before this function, otherwise - * addShard will never reach the "hello" command -- a find query is called first. + * addShard will never reach the isMaster command -- a find query is called first. */ - void expectHello(const HostAndPort& target, StatusWith<BSONObj> helloResponse) { - onCommandForAddShard([&, target, helloResponse](const RemoteCommandRequest& request) { + void expectIsMaster(const HostAndPort& target, StatusWith<BSONObj> isMasterResponse) { + onCommandForAddShard([&, target, isMasterResponse](const RemoteCommandRequest& request) { ASSERT_EQ(request.target, target); ASSERT_EQ(request.dbname, "admin"); - ASSERT_BSONOBJ_EQ(request.cmdObj, BSON("hello" << 1)); + ASSERT_BSONOBJ_EQ(request.cmdObj, BSON("isMaster" << 1)); ASSERT_BSONOBJ_EQ(rpc::makeEmptyMetadata(), request.metadata); - return helloResponse; + return isMasterResponse; }); } @@ -234,7 +225,7 @@ protected: ASSERT_EQ(request.dbname, NamespaceString::kClusterParametersNamespace.db()); ASSERT_BSONOBJ_EQ(request.cmdObj, BSON("find" << NamespaceString::kClusterParametersNamespace.coll() - << "maxTimeMS" << 60000 << "readConcern" + << "maxTimeMS" << 30000 << "readConcern" << BSON("level" << "majority"))); auto cursorRes = CursorResponse(NamespaceString::kClusterParametersNamespace, 0, {}); @@ -242,24 +233,6 @@ protected: }); } - void expectClusterTimeKeysPullRequest(const HostAndPort& target) { - onCommandForAddShard([&](const RemoteCommandRequest& request) { - ASSERT_EQ(request.target, target); - ASSERT_BSONOBJ_EQ(request.cmdObj, - BSON("find" << NamespaceString::kKeysCollectionNamespace.coll() - << "maxTimeMS" << 60000 << "readConcern" - << BSON("level" - << "local"))); - - KeysCollectionDocument key(1); - key.setKeysCollectionDocumentBase( - {"dummy", TimeProofService::generateRandomKey(), LogicalTime(Timestamp(105, 0))}); - auto cursorRes = - CursorResponse(NamespaceString::kKeysCollectionNamespace, 0, {key.toBSON()}); - return cursorRes.toBSON(CursorResponse::ResponseType::InitialResponse); - }); - } - /** * Waits for a request for the shardIdentity document to be upserted into a shard from the * config server on addShard. @@ -460,7 +433,11 @@ TEST_F(AddShardTest, CreateShardIdentityUpsertForAddShard) { << shardName << "clusterId" << _clusterId << "configsvrConnectionString" << replicationCoordinator()->getConfigConnectionString().toString()) - << "multi" << false << "upsert" << true))); + << "multi" << false << "upsert" << true)) + << "writeConcern" + << BSON("w" + << "majority" + << "wtimeout" << 60000)); auto addShardCmd = add_shard_util::createAddShardCmd(operationContext(), shardName); auto actualBSON = add_shard_util::createShardIdentityUpsertForAddShard(addShardCmd); ASSERT_BSONOBJ_EQ(expectedBSON, actualBSON); @@ -505,9 +482,9 @@ TEST_F(AddShardTest, StandaloneBasicSuccess) { ASSERT_EQUALS(expectedShardName, shardName); }); - BSONObj commandResponse = BSON("ok" << 1 << "isWritablePrimary" << true << "maxWireVersion" + BSONObj commandResponse = BSON("ok" << 1 << "ismaster" << true << "maxWireVersion" << WireVersion::LATEST_WIRE_VERSION); - expectHello(shardTarget, commandResponse); + expectIsMaster(shardTarget, commandResponse); // Get databases list from new shard expectListDatabases( @@ -520,9 +497,6 @@ TEST_F(AddShardTest, StandaloneBasicSuccess) { expectCollectionDrop(shardTarget, NamespaceString("config", "system.sessions")); - // The shard receives a find to pull all clusterTime keys from the new shard. - expectClusterTimeKeysPullRequest(shardTarget); - // The shard receives the _addShard command expectAddShardCmdReturnSuccess(shardTarget, expectedShardName); @@ -595,9 +569,9 @@ TEST_F(AddShardTest, StandaloneGenerateName) { ASSERT_EQUALS(expectedShardName, shardName); }); - BSONObj commandResponse = BSON("ok" << 1 << "isWritablePrimary" << true << "maxWireVersion" + BSONObj commandResponse = BSON("ok" << 1 << "ismaster" << true << "maxWireVersion" << WireVersion::LATEST_WIRE_VERSION); - expectHello(shardTarget, commandResponse); + expectIsMaster(shardTarget, commandResponse); // Get databases list from new shard expectListDatabases( @@ -610,9 +584,6 @@ TEST_F(AddShardTest, StandaloneGenerateName) { expectCollectionDrop(shardTarget, NamespaceString("config", "system.sessions")); - // The shard receives a find to pull all clusterTime keys from the new shard. - expectClusterTimeKeysPullRequest(shardTarget); - // The shard receives the _addShard command expectAddShardCmdReturnSuccess(shardTarget, expectedShardName); @@ -702,7 +673,7 @@ TEST_F(AddShardTest, UnreachableHost) { }); Status hostUnreachableStatus = Status(ErrorCodes::HostUnreachable, "host unreachable"); - expectHello(shardTarget, hostUnreachableStatus); + expectIsMaster(shardTarget, hostUnreachableStatus); future.timed_get(kLongFutureTimeout); } @@ -727,9 +698,9 @@ TEST_F(AddShardTest, AddMongosAsShard) { ASSERT_EQUALS(ErrorCodes::IllegalOperation, status); }); - expectHello(shardTarget, - BSON("msg" - << "isdbgrid")); + expectIsMaster(shardTarget, + BSON("msg" + << "isdbgrid")); future.timed_get(kLongFutureTimeout); } @@ -755,10 +726,10 @@ TEST_F(AddShardTest, AddReplicaSetShardAsStandalone) { ASSERT_STRING_CONTAINS(status.getStatus().reason(), "use replica set url format"); }); - BSONObj commandResponse = BSON("ok" << 1 << "isWritablePrimary" << true << "setName" + BSONObj commandResponse = BSON("ok" << 1 << "ismaster" << true << "setName" << "myOtherSet" << "maxWireVersion" << WireVersion::LATEST_WIRE_VERSION); - expectHello(shardTarget, commandResponse); + expectIsMaster(shardTarget, commandResponse); future.timed_get(kLongFutureTimeout); } @@ -785,9 +756,9 @@ TEST_F(AddShardTest, AddStandaloneHostShardAsReplicaSet) { ASSERT_STRING_CONTAINS(status.getStatus().reason(), "host did not return a set name"); }); - BSONObj commandResponse = BSON("ok" << 1 << "isWritablePrimary" << true << "maxWireVersion" + BSONObj commandResponse = BSON("ok" << 1 << "ismaster" << true << "maxWireVersion" << WireVersion::LATEST_WIRE_VERSION); - expectHello(shardTarget, commandResponse); + expectIsMaster(shardTarget, commandResponse); future.timed_get(kLongFutureTimeout); } @@ -814,10 +785,10 @@ TEST_F(AddShardTest, ReplicaSetMistmatchedReplicaSetName) { ASSERT_STRING_CONTAINS(status.getStatus().reason(), "does not match the actual set name"); }); - BSONObj commandResponse = BSON("ok" << 1 << "isWritablePrimary" << true << "setName" + BSONObj commandResponse = BSON("ok" << 1 << "ismaster" << true << "setName" << "myOtherSet" << "maxWireVersion" << WireVersion::LATEST_WIRE_VERSION); - expectHello(shardTarget, commandResponse); + expectIsMaster(shardTarget, commandResponse); future.timed_get(kLongFutureTimeout); } @@ -846,10 +817,10 @@ TEST_F(AddShardTest, ShardIsCSRSConfigServer) { }); BSONObj commandResponse = - BSON("ok" << 1 << "isWritablePrimary" << true << "setName" + BSON("ok" << 1 << "ismaster" << true << "setName" << "config" << "configsvr" << true << "maxWireVersion" << WireVersion::LATEST_WIRE_VERSION); - expectHello(shardTarget, commandResponse); + expectIsMaster(shardTarget, commandResponse); future.timed_get(kLongFutureTimeout); } @@ -879,11 +850,11 @@ TEST_F(AddShardTest, ReplicaSetMissingHostsProvidedInSeedList) { BSONArrayBuilder hosts; hosts.append("host1:12345"); - BSONObj commandResponse = BSON("ok" << 1 << "isWritablePrimary" << true << "setName" + BSONObj commandResponse = BSON("ok" << 1 << "ismaster" << true << "setName" << "mySet" << "hosts" << hosts.arr() << "maxWireVersion" << WireVersion::LATEST_WIRE_VERSION); - expectHello(shardTarget, commandResponse); + expectIsMaster(shardTarget, commandResponse); future.timed_get(kLongFutureTimeout); } @@ -914,11 +885,11 @@ TEST_F(AddShardTest, AddShardWithNameConfigFails) { BSONArrayBuilder hosts; hosts.append("host1:12345"); hosts.append("host2:12345"); - BSONObj commandResponse = BSON("ok" << 1 << "isWritablePrimary" << true << "setName" + BSONObj commandResponse = BSON("ok" << 1 << "ismaster" << true << "setName" << "mySet" << "hosts" << hosts.arr() << "maxWireVersion" << WireVersion::LATEST_WIRE_VERSION); - expectHello(shardTarget, commandResponse); + expectIsMaster(shardTarget, commandResponse); future.timed_get(kLongFutureTimeout); } @@ -960,11 +931,11 @@ TEST_F(AddShardTest, ShardContainsExistingDatabase) { BSONArrayBuilder hosts; hosts.append("host1:12345"); hosts.append("host2:12345"); - BSONObj commandResponse = BSON("ok" << 1 << "isWritablePrimary" << true << "setName" + BSONObj commandResponse = BSON("ok" << 1 << "ismaster" << true << "setName" << "mySet" << "hosts" << hosts.arr() << "maxWireVersion" << WireVersion::LATEST_WIRE_VERSION); - expectHello(shardTarget, commandResponse); + expectIsMaster(shardTarget, commandResponse); expectListDatabases(shardTarget, {BSON("name" << existingDB.getName())}); @@ -1004,20 +975,17 @@ TEST_F(AddShardTest, SuccessfullyAddReplicaSet) { BSONArrayBuilder hosts; hosts.append("host1:12345"); hosts.append("host2:12345"); - BSONObj commandResponse = BSON("ok" << 1 << "isWritablePrimary" << true << "setName" + BSONObj commandResponse = BSON("ok" << 1 << "ismaster" << true << "setName" << "mySet" << "hosts" << hosts.arr() << "maxWireVersion" << WireVersion::LATEST_WIRE_VERSION); - expectHello(shardTarget, commandResponse); + expectIsMaster(shardTarget, commandResponse); // Get databases list from new shard expectListDatabases(shardTarget, std::vector<BSONObj>{BSON("name" << discoveredDB.getName())}); expectCollectionDrop(shardTarget, NamespaceString("config", "system.sessions")); - // The shard receives a find to pull all clusterTime keys from the new shard. - expectClusterTimeKeysPullRequest(shardTarget); - // The shard receives the _addShard command expectAddShardCmdReturnSuccess(shardTarget, expectedShardName); @@ -1079,20 +1047,17 @@ TEST_F(AddShardTest, ReplicaSetExtraHostsDiscovered) { BSONArrayBuilder hosts; hosts.append("host1:12345"); hosts.append("host2:12345"); - BSONObj commandResponse = BSON("ok" << 1 << "isWritablePrimary" << true << "setName" + BSONObj commandResponse = BSON("ok" << 1 << "ismaster" << true << "setName" << "mySet" << "hosts" << hosts.arr() << "maxWireVersion" << WireVersion::LATEST_WIRE_VERSION); - expectHello(shardTarget, commandResponse); + expectIsMaster(shardTarget, commandResponse); // Get databases list from new shard expectListDatabases(shardTarget, std::vector<BSONObj>{BSON("name" << discoveredDB.getName())}); expectCollectionDrop(shardTarget, NamespaceString("config", "system.sessions")); - // The shard receives a find to pull all clusterTime keys from the new shard. - expectClusterTimeKeysPullRequest(shardTarget); - // The shard receives the _addShard command expectAddShardCmdReturnSuccess(shardTarget, expectedShardName); @@ -1163,9 +1128,9 @@ TEST_F(AddShardTest, AddShardSucceedsEvenIfAddingDBsFromNewShardFails) { ASSERT_EQUALS(expectedShardName, shardName); }); - BSONObj commandResponse = BSON("ok" << 1 << "isWritablePrimary" << true << "maxWireVersion" + BSONObj commandResponse = BSON("ok" << 1 << "ismaster" << true << "maxWireVersion" << WireVersion::LATEST_WIRE_VERSION); - expectHello(shardTarget, commandResponse); + expectIsMaster(shardTarget, commandResponse); // Get databases list from new shard expectListDatabases( @@ -1178,9 +1143,6 @@ TEST_F(AddShardTest, AddShardSucceedsEvenIfAddingDBsFromNewShardFails) { expectCollectionDrop(shardTarget, NamespaceString("config", "system.sessions")); - // The shard receives a find to pull all clusterTime keys from the new shard. - expectClusterTimeKeysPullRequest(shardTarget); - // The shard receives the _addShard command expectAddShardCmdReturnSuccess(shardTarget, expectedShardName); diff --git a/src/mongo/db/s/config/sharding_catalog_manager_chunk_operations.cpp b/src/mongo/db/s/config/sharding_catalog_manager_chunk_operations.cpp index d1bc8fcd343..172729b575b 100644 --- a/src/mongo/db/s/config/sharding_catalog_manager_chunk_operations.cpp +++ b/src/mongo/db/s/config/sharding_catalog_manager_chunk_operations.cpp @@ -43,7 +43,6 @@ #include "mongo/db/logical_session_cache.h" #include "mongo/db/namespace_string.h" #include "mongo/db/operation_context.h" -#include "mongo/db/query/cursor_response.h" #include "mongo/db/query/distinct_command_gen.h" #include "mongo/db/repl/repl_client_info.h" #include "mongo/db/s/sharding_logging.h" @@ -90,6 +89,76 @@ void appendShortVersion(BufBuilder* out, const ChunkType& chunk) { bb.done(); } +BSONArray buildMergeChunksTransactionUpdates(const std::vector<ChunkType>& chunksToMerge, + const ChunkVersion& mergeVersion, + const boost::optional<Timestamp>& validAfter) { + BSONArrayBuilder updates; + + // Build an update operation to expand the first chunk into the newly merged chunk + { + BSONObjBuilder op; + op.append("op", "u"); + op.appendBool("b", false); // no upsert + op.append("ns", ChunkType::ConfigNS.ns()); + + // expand first chunk into newly merged chunk + ChunkType mergedChunk(chunksToMerge.front()); + mergedChunk.setMax(chunksToMerge.back().getMax()); + + // fill in additional details for sending through transaction + mergedChunk.setVersion(mergeVersion); + mergedChunk.setEstimatedSizeBytes(boost::none); + + invariant(validAfter); + mergedChunk.setHistory({ChunkHistory(validAfter.get(), mergedChunk.getShard())}); + + // add the new chunk information as the update object + op.append("o", mergedChunk.toConfigBSON()); + + // query object + op.append("o2", BSON(ChunkType::name(mergedChunk.getName()))); + + updates.append(op.obj()); + } + + // Build update operations to delete the rest of the chunks to be merged. Remember not + // to delete the first chunk we're expanding + for (size_t i = 1; i < chunksToMerge.size(); ++i) { + BSONObjBuilder op; + op.append("op", "d"); + op.append("ns", ChunkType::ConfigNS.ns()); + + op.append("o", BSON(ChunkType::name(chunksToMerge[i].getName()))); + + updates.append(op.obj()); + } + + return updates.arr(); +} + +BSONArray buildMergeChunksTransactionPrecond(const std::vector<ChunkType>& chunksToMerge, + const ChunkVersion& collVersion) { + BSONArrayBuilder preCond; + + for (const auto& chunk : chunksToMerge) { + BSONObj query = BSON(ChunkType::min(chunk.getMin()) + << ChunkType::max(chunk.getMax()) << ChunkType::collectionUUID() + << chunk.getCollectionUUID()); + + const auto collectionIdentityMatchCondition = + BSON(ChunkType::collectionUUID() + << chunk.getCollectionUUID() << ChunkType::shard(chunk.getShard().toString())); + + BSONObjBuilder b; + b.append("ns", ChunkType::ConfigNS.ns()); + b.append("q", BSON("query" << query << "orderby" << BSON(ChunkType::lastmod() << -1))); + b.append("res", collectionIdentityMatchCondition); + + preCond.append(b.obj()); + } + return preCond.arr(); +} + /** * Check that the chunk still exists and return its metadata. */ @@ -97,11 +166,13 @@ StatusWith<ChunkType> findChunkContainingRange(OperationContext* opCtx, const UUID& uuid, const OID& epoch, const Timestamp& timestamp, - const ChunkRange& range) { + const BSONObj& min, + const BSONObj& max) { const auto chunkQuery = [&]() { BSONObjBuilder queryBuilder; queryBuilder << ChunkType::collectionUUID << uuid; - queryBuilder << ChunkType::min(BSON("$lte" << range.getMin())); + queryBuilder << ChunkType::min(BSON("$lte" << min)); + queryBuilder << ChunkType::max(BSON("$gte" << max)); return queryBuilder.obj(); }(); @@ -113,26 +184,21 @@ StatusWith<ChunkType> findChunkContainingRange(OperationContext* opCtx, repl::ReadConcernLevel::kLocalReadConcern, ChunkType::ConfigNS, chunkQuery, - BSON(ChunkType::min << -1), - 1 /* limit */); + BSONObj(), + 2 /* limit */); if (!findResponseWith.isOK()) { return findResponseWith.getStatus(); } - if (!findResponseWith.getValue().docs.empty()) { - const auto containingChunk = uassertStatusOK(ChunkType::parseFromConfigBSON( - findResponseWith.getValue().docs.front(), epoch, timestamp)); - - if (containingChunk.getRange().covers(range)) { - return containingChunk; - } + if (findResponseWith.getValue().docs.size() != 1) { + return {ErrorCodes::Error(40165), + str::stream() << "Could not find a chunk including bounds [" << min << ", " << max + << "). Cannot execute the migration commit with invalid chunks."}; } - return {ErrorCodes::Error(40165), - str::stream() << "Could not find a chunk including bounds [" << range.getMin() << ", " - << range.getMax() - << "). Cannot execute the migration commit with invalid chunks."}; + return uassertStatusOK( + ChunkType::parseFromConfigBSON(findResponseWith.getValue().docs.front(), epoch, timestamp)); } BSONObj makeCommitChunkTransactionCommand(const NamespaceString& nss, @@ -207,33 +273,6 @@ BSONObj makeCommitChunkTransactionCommand(const NamespaceString& nss, return BSON("applyOps" << updates.arr() << "alwaysUpsert" << false); } -BSONObj buildCountContiguousChunksByBounds(const UUID& collectionUUID, - const std::string& shard, - const std::vector<BSONObj>& boundsForChunks) { - AggregateCommandRequest countRequest(ChunkType::ConfigNS); - - invariant(boundsForChunks.size() > 1); - auto minBoundIt = boundsForChunks.begin(); - auto maxBoundIt = minBoundIt + 1; - - BSONArrayBuilder chunkDocArray; - while (maxBoundIt != boundsForChunks.end()) { - const auto query = BSON(ChunkType::min(*minBoundIt) - << ChunkType::max(*maxBoundIt) << ChunkType::collectionUUID() - << collectionUUID << ChunkType::shard() << shard); - - chunkDocArray.append(query); - ++minBoundIt; - ++maxBoundIt; - } - - std::vector<BSONObj> pipeline; - pipeline.push_back(BSON("$match" << BSON("$or" << chunkDocArray.arr()))); - pipeline.push_back(BSON("$count" << ChunkType::collectionUUID.name())); - countRequest.setPipeline(pipeline); - return countRequest.toBSON({}); -} - /** * Returns a chunk different from the one being migrated or 'none' if one doesn't exist. */ @@ -461,48 +500,6 @@ std::vector<ShardId> getShardsOwningChunksForCollection(OperationContext* opCtx, return shardIds; } -// Checks if the requested split already exists. It is possible that the split operation completed, -// but the router did not receive the response. This would result in the router retrying the split -// operation, in which case it is fine for the request to become a no-op. -auto isSplitAlreadyDone(OperationContext* opCtx, - const ChunkRange& range, - const std::string& shardName, - const ChunkType& origChunk, - const std::vector<BSONObj>& newChunkBounds) { - std::vector<BSONObj> expectedChunksBounds; - expectedChunksBounds.reserve(newChunkBounds.size() + 1); - expectedChunksBounds.push_back(range.getMin()); - expectedChunksBounds.insert( - std::end(expectedChunksBounds), std::begin(newChunkBounds), std::end(newChunkBounds)); - - auto countRequest = buildCountContiguousChunksByBounds( - origChunk.getCollectionUUID(), shardName, expectedChunksBounds); - - const auto expectedChunkCount = expectedChunksBounds.size() - 1; - - auto const configShard = Grid::get(opCtx)->shardRegistry()->getConfigShard(); - - auto countResponse = uassertStatusOK(configShard->runCommandWithFixedRetryAttempts( - opCtx, - ReadPreferenceSetting{ReadPreference::PrimaryOnly}, - NamespaceString::kConfigDb.toString(), - countRequest, - Shard::RetryPolicy::kIdempotent)); - - const auto docCount = [&]() { - auto cursorResponse = - uassertStatusOK(CursorResponse::parseFromBSON(countResponse.response)); - auto firstBatch = cursorResponse.getBatch(); - if (firstBatch.empty()) { - return 0; - } - - auto countObj = firstBatch.front(); - return countObj.getIntField(ChunkType::collectionUUID.name()); - }(); - return size_t(docCount) == expectedChunkCount; -} - } // namespace void ShardingCatalogManager::bumpMajorVersionOneChunkPerShard( @@ -573,11 +570,6 @@ StatusWith<BSONObj> ShardingCatalogManager::commitChunkSplit( const std::vector<BSONObj>& splitPoints, const std::string& shardName, const bool fromChunkSplitter) { - - // Mark opCtx as interruptible to ensure that all reads and writes to the metadata collections - // under the exclusive _kChunkOpLock happen on the same term. - opCtx->setAlwaysInterruptAtStepDownOrUp(); - // Take _kChunkOpLock in exclusive mode to prevent concurrent chunk modifications and generate // strictly monotonously increasing collection versions Lock::ExclusiveLock lk(opCtx, opCtx->lockState(), _kChunkOpLock); @@ -634,26 +626,11 @@ StatusWith<BSONObj> ShardingCatalogManager::commitChunkSplit( std::vector<ChunkType> newChunks; ChunkVersion currentMaxVersion = collVersion; - const auto buildChunkVersionBSON = [](const ChunkVersion& version) { - BSONObjBuilder response; - version.serializeToBSON(kCollectionVersionField, &response); - version.serializeToBSON(ChunkVersion::kShardVersionField, &response); - return response.obj(); - }; auto startKey = range.getMin(); auto newChunkBounds(splitPoints); newChunkBounds.push_back(range.getMax()); - if (isSplitAlreadyDone(opCtx, range, shardName, origChunk.getValue(), newChunkBounds)) { - // In case the request was already fullfilled, we still need to wait until the original - // request is majority written. The timestamp is not known, so we use the system's last - // optime. Otherwise the next RoutingInfo cache refresh from the shard may not see the - // newest information. - repl::ReplClientInfo::forClient(opCtx->getClient()).setLastOpToSystemLastOpTime(opCtx); - return buildChunkVersionBSON(collVersion); - } - auto shouldTakeOriginalChunkID = true; OID chunkID; @@ -713,7 +690,6 @@ StatusWith<BSONObj> ShardingCatalogManager::commitChunkSplit( newChunk.setMin(startKey); newChunk.setMax(endKey); newChunk.setEstimatedSizeBytes(boost::none); - newChunk.setJumbo(false); op.append("o", newChunk.toConfigBSON()); @@ -802,83 +778,10 @@ StatusWith<BSONObj> ShardingCatalogManager::commitChunkSplit( } } - return buildChunkVersionBSON(currentMaxVersion); -} - -void ShardingCatalogManager::_mergeChunksInTransaction( - OperationContext* opCtx, - const NamespaceString& nss, - const UUID& collectionUUID, - const ChunkVersion& mergeVersion, - const Timestamp& validAfter, - const ChunkRange& chunkRange, - const ShardId& shardId, - std::shared_ptr<std::vector<ChunkType>> chunksToMerge) { - withTransaction( - opCtx, ChunkType::ConfigNS, [&, this](OperationContext* opCtx, TxnNumber txnNumber) { - // Construct the new chunk by taking `min` from the first merged chunk and `max` - // from the last. - write_ops::UpdateCommandRequest updateOp(ChunkType::ConfigNS); - updateOp.setUpdates({[&] { - write_ops::UpdateOpEntry entry; - - ChunkType mergedChunk(chunksToMerge->front()); - entry.setQ(BSON(ChunkType::name(mergedChunk.getName()))); - mergedChunk.setMax(chunksToMerge->back().getMax()); - - // Fill in additional details for sending through transaction. - mergedChunk.setVersion(mergeVersion); - mergedChunk.setEstimatedSizeBytes(boost::none); - - mergedChunk.setHistory({ChunkHistory(validAfter, mergedChunk.getShard())}); - - entry.setU(write_ops::UpdateModification::parseFromClassicUpdate( - mergedChunk.toConfigBSON())); - entry.setMulti(false); - - return entry; - }()}); - - const auto updateRes = - writeToConfigDocumentInTxn(opCtx, ChunkType::ConfigNS, updateOp, txnNumber); - - const auto numDocsModified = UpdateOp::parseResponse(updateRes).getN(); - uassert(ErrorCodes::ConflictingOperationInProgress, - str::stream() << "Unexpected number of modified documents during chunks merge " - "commit. Modified " - << numDocsModified << " documents instead of 1", - numDocsModified == 1); - - // Delete the rest of the chunks to be merged. - // Remember not to delete the first chunk we're expanding. - BSONObjBuilder queryBuilder; - queryBuilder << ChunkType::collectionUUID << collectionUUID; - queryBuilder << ChunkType::shard(shardId.toString()); - queryBuilder << ChunkType::min(BSON("$gte" << chunksToMerge->front().getMax())); - queryBuilder << ChunkType::min(BSON("$lt" << chunksToMerge->back().getMax())); - - write_ops::DeleteCommandRequest deleteOp(ChunkType::ConfigNS); - deleteOp.setDeletes({[&] { - write_ops::DeleteOpEntry entry; - entry.setQ(queryBuilder.obj()); - entry.setMulti(true); - return entry; - }()}); - - const auto deleteRes = - writeToConfigDocumentInTxn(opCtx, ChunkType::ConfigNS, deleteOp, txnNumber); - - const auto numDocsDeleted = DeleteOp::parseResponse(deleteRes).getN(); - const int expectedNumDocsDeleted = chunksToMerge->size() - 1; - uassert(ErrorCodes::ConflictingOperationInProgress, - str::stream() << "Unexpected number of deleted documents during chunks merge " - "commit. Deleted " - << numDocsDeleted << " documents instead of " - << expectedNumDocsDeleted, - numDocsDeleted == expectedNumDocsDeleted); - - LOGV2_DEBUG(6583805, 1, "Finished all transaction operations in merge chunk command"); - }); + BSONObjBuilder response; + currentMaxVersion.serializeToBSON(kCollectionVersionField, &response); + currentMaxVersion.serializeToBSON(ChunkVersion::kShardVersionField, &response); + return response.obj(); } StatusWith<BSONObj> ShardingCatalogManager::commitChunksMerge( @@ -888,11 +791,11 @@ StatusWith<BSONObj> ShardingCatalogManager::commitChunksMerge( const boost::optional<Timestamp>& timestamp, const UUID& requestCollectionUUID, const ChunkRange& chunkRange, - const ShardId& shardId) { - - // Mark opCtx as interruptible to ensure that all reads and writes to the metadata collections - // under the exclusive _kChunkOpLock happen on the same term. - opCtx->setAlwaysInterruptAtStepDownOrUp(); + const ShardId& shardId, + const boost::optional<Timestamp>& validAfter) { + if (!validAfter) { + return {ErrorCodes::IllegalOperation, "chunk operation requires validAfter timestamp"}; + } // Take _kChunkOpLock in exclusive mode to prevent concurrent chunk modifications and generate // strictly monotonously increasing collection versions @@ -968,18 +871,11 @@ StatusWith<BSONObj> ShardingCatalogManager::commitChunksMerge( // 3. Prepare the data for the merge // and ensure that the retrieved list of chunks covers the whole range. - - // The `validAfter` field must always be set. If not existing, it means the chunk - // always belonged to the same shard, hence it's valid to set `0` as the time at - // which the chunk started being valid. - Timestamp validAfter{0}; - - auto chunksToMerge = std::make_shared<std::vector<ChunkType>>(); - chunksToMerge->reserve(shardChunksInRangeResponse.docs.size()); + std::vector<ChunkType> chunksToMerge; for (const auto& chunkDoc : shardChunksInRangeResponse.docs) { auto chunk = uassertStatusOK( ChunkType::parseFromConfigBSON(chunkDoc, coll.getEpoch(), coll.getTimestamp())); - if (chunksToMerge->empty()) { + if (chunksToMerge.empty()) { uassert(ErrorCodes::IllegalOperation, str::stream() << "could not merge chunks, shard " << shardId @@ -992,41 +888,46 @@ StatusWith<BSONObj> ShardingCatalogManager::commitChunksMerge( << "could not merge chunks, shard " << shardId << " does not contain a sequence of chunks that exactly fills the range " << chunkRange.toString(), - chunk.getMin().woCompare(chunksToMerge->back().getMax()) == 0); - } - - // Get the `validAfter` field from the most recent chunk placed on the shard - if (!chunk.getHistory().empty()) { - const auto& chunkValidAfter = chunk.getHistory().front().getValidAfter(); - if (validAfter < chunkValidAfter) { - validAfter = chunkValidAfter; - } + chunk.getMin().woCompare(chunksToMerge.back().getMax()) == 0); } - - chunksToMerge->push_back(std::move(chunk)); + chunksToMerge.push_back(std::move(chunk)); } uassert(ErrorCodes::IllegalOperation, str::stream() << "could not merge chunks, shard " << shardId << " does not contain a sequence of chunks that exactly fills the range " << chunkRange.toString(), - !chunksToMerge->empty() && - chunksToMerge->back().getMax().woCompare(chunkRange.getMax()) == 0); + !chunksToMerge.empty() && + chunksToMerge.back().getMax().woCompare(chunkRange.getMax()) == 0); ChunkVersion initialVersion = collVersion; ChunkVersion mergeVersion = initialVersion; mergeVersion.incMinor(); + auto updates = buildMergeChunksTransactionUpdates(chunksToMerge, mergeVersion, validAfter); + auto preCond = buildMergeChunksTransactionPrecond(chunksToMerge, initialVersion); + // 4. apply the batch of updates to local metadata - _mergeChunksInTransaction( - opCtx, nss, coll.getUuid(), mergeVersion, validAfter, chunkRange, shardId, chunksToMerge); + uassertStatusOK(Grid::get(opCtx)->catalogClient()->applyChunkOpsDeprecated( + opCtx, + updates, + preCond, + coll.getUuid(), + nss, + mergeVersion, + WriteConcernOptions(), + repl::ReadConcernLevel::kLocalReadConcern)); // 5. log changes BSONObjBuilder logDetail; + { + BSONArrayBuilder b(logDetail.subarrayStart("merged")); + for (const auto& chunkToMerge : chunksToMerge) { + b.append(chunkToMerge.toConfigBSON()); + } + } initialVersion.appendLegacyWithField(&logDetail, "prevShardVersion"); mergeVersion.appendLegacyWithField(&logDetail, "mergedVersion"); logDetail.append("owningShard", shardId); - chunkRange.append(&logDetail); - logDetail.append("numChunks", static_cast<int>(chunksToMerge->size())); ShardingLogging::get(opCtx)->logChange( opCtx, "merge", nss.ns(), logDetail.obj(), WriteConcernOptions()); @@ -1051,15 +952,6 @@ StatusWith<BSONObj> ShardingCatalogManager::commitChunkMigration( return {ErrorCodes::IllegalOperation, "chunk operation requires validAfter timestamp"}; } - uassertStatusOK( - ShardKeyPattern::checkShardKeyIsValidForMetadataStorage(migratedChunk.getMin())); - uassertStatusOK( - ShardKeyPattern::checkShardKeyIsValidForMetadataStorage(migratedChunk.getMax())); - - // Mark opCtx as interruptible to ensure that all reads and writes to the metadata collections - // under the exclusive _kChunkOpLock happen on the same term. - opCtx->setAlwaysInterruptAtStepDownOrUp(); - // Must hold the shard lock until the entire commit finishes to serialize with removeShard. Lock::SharedLock shardLock(opCtx->lockState(), _kShardMembershipLock); @@ -1149,8 +1041,12 @@ StatusWith<BSONObj> ShardingCatalogManager::commitChunkMigration( migratedChunk.isVersionSet() && migratedChunk.getVersion().isSet()); // Check if range still exists and which shard owns it - auto swCurrentChunk = findChunkContainingRange( - opCtx, coll.getUuid(), coll.getEpoch(), coll.getTimestamp(), migratedChunk.getRange()); + auto swCurrentChunk = findChunkContainingRange(opCtx, + coll.getUuid(), + coll.getEpoch(), + coll.getTimestamp(), + migratedChunk.getMin(), + migratedChunk.getMax()); if (!swCurrentChunk.isOK()) { return swCurrentChunk.getStatus(); @@ -1367,10 +1263,6 @@ void ShardingCatalogManager::upgradeChunksHistory(OperationContext* opCtx, auto const catalogClient = Grid::get(opCtx)->catalogClient(); const auto shardRegistry = Grid::get(opCtx)->shardRegistry(); - // Mark opCtx as interruptible to ensure that all reads and writes to the metadata collections - // under the exclusive _kChunkOpLock happen on the same term. - opCtx->setAlwaysInterruptAtStepDownOrUp(); - // Take _kChunkOpLock in exclusive mode to prevent concurrent chunk splits, merges, and // migrations. Lock::ExclusiveLock lk(opCtx->lockState(), _kChunkOpLock); @@ -1410,12 +1302,10 @@ void ShardingCatalogManager::upgradeChunksHistory(OperationContext* opCtx, }()}); return updateOp; }()); + request.setWriteConcern(ShardingCatalogClient::kLocalWriteConcern.toBSON()); - auto response = configShard->runBatchWriteCommand(opCtx, - Shard::kDefaultConfigCommandTimeout, - request, - ShardingCatalogClient::kLocalWriteConcern, - Shard::RetryPolicy::kIdempotent); + auto response = configShard->runBatchWriteCommand( + opCtx, Shard::kDefaultConfigCommandTimeout, request, Shard::RetryPolicy::kIdempotent); uassertStatusOK(response.toStatus()); uassert(ErrorCodes::Error(5760502), @@ -1505,10 +1395,6 @@ void ShardingCatalogManager::clearJumboFlag(OperationContext* opCtx, const NamespaceString& nss, const OID& collectionEpoch, const ChunkRange& chunk) { - // Mark opCtx as interruptible to ensure that all reads and writes to the metadata collections - // under the exclusive _kChunkOpLock happen on the same term. - opCtx->setAlwaysInterruptAtStepDownOrUp(); - // Take _kChunkOpLock in exclusive mode to prevent concurrent chunk modifications and generate // strictly monotonously increasing collection versions Lock::ExclusiveLock lk(opCtx, opCtx->lockState(), _kChunkOpLock); @@ -1627,10 +1513,6 @@ void ShardingCatalogManager::ensureChunkVersionIsGreaterThan(OperationContext* o const BSONObj& minKey, const BSONObj& maxKey, const ChunkVersion& version) { - // Mark opCtx as interruptible to ensure that all reads and writes to the metadata collections - // under the exclusive _kChunkOpLock happen on the same term. - opCtx->setAlwaysInterruptAtStepDownOrUp(); - // Take _kChunkOpLock in exclusive mode to prevent concurrent chunk modifications and generate // strictly monotonously increasing collection versions Lock::ExclusiveLock lk(opCtx, opCtx->lockState(), _kChunkOpLock); @@ -1828,58 +1710,36 @@ void ShardingCatalogManager::bumpMultipleCollectionVersionsAndChangeMetadataInTx unique_function<void(OperationContext*, TxnNumber)> changeMetadataFunc, const WriteConcernOptions& writeConcern) { - // Mark opCtx as interruptible to ensure that all reads and writes to the metadata collections - // under the exclusive _kChunkOpLock happen on the same term. - opCtx->setAlwaysInterruptAtStepDownOrUp(); - // Take _kChunkOpLock in exclusive mode to prevent concurrent chunk splits, merges, and // migrations Lock::ExclusiveLock lk(opCtx, opCtx->lockState(), _kChunkOpLock); - withTransaction( - opCtx, - NamespaceString::kConfigReshardingOperationsNamespace, - [&collNames, &changeMetadataFunc](OperationContext* opCtx, TxnNumber txnNumber) { - for (const auto& nss : collNames) { - bumpCollectionMinorVersion(opCtx, nss, txnNumber); - } - changeMetadataFunc(opCtx, txnNumber); - }, - writeConcern); + withTransaction(opCtx, + NamespaceString::kConfigReshardingOperationsNamespace, + [&](OperationContext* opCtx, TxnNumber txnNumber) { + for (const auto& nss : collNames) { + bumpCollectionMinorVersion(opCtx, nss, txnNumber); + } + changeMetadataFunc(opCtx, txnNumber); + }, + writeConcern); } void ShardingCatalogManager::splitOrMarkJumbo(OperationContext* opCtx, const NamespaceString& nss, - const BSONObj& minKey, - boost::optional<int64_t> optMaxChunkSizeBytes) { + const BSONObj& minKey) { const auto cm = uassertStatusOK( Grid::get(opCtx)->catalogCache()->getShardedCollectionRoutingInfoWithRefresh(opCtx, nss)); auto chunk = cm.findIntersectingChunkWithSimpleCollation(minKey); try { - const auto maxChunkSizeBytes = [&]() -> int64_t { - if (optMaxChunkSizeBytes.has_value()) { - return *optMaxChunkSizeBytes; - } - - auto coll = Grid::get(opCtx)->catalogClient()->getCollection( - opCtx, nss, repl::ReadConcernLevel::kMajorityReadConcern); - return coll.getMaxChunkSizeBytes().value_or( - Grid::get(opCtx)->getBalancerConfiguration()->getMaxChunkSizeBytes()); - }(); - - // Limit the search to one split point: this code path is reached when a migration fails due - // to ErrorCodes::ChunkTooBig. In case there is a too frequent shard key, only select the - // next key in order to split the range in jumbo chunk + remaining range. - const int limit = 1; - auto splitPoints = uassertStatusOK( - shardutil::selectChunkSplitPoints(opCtx, - chunk.getShardId(), - nss, - cm.getShardKeyPattern(), - ChunkRange(chunk.getMin(), chunk.getMax()), - maxChunkSizeBytes, - limit)); + const auto splitPoints = uassertStatusOK(shardutil::selectChunkSplitPoints( + opCtx, + chunk.getShardId(), + nss, + cm.getShardKeyPattern(), + ChunkRange(chunk.getMin(), chunk.getMax()), + Grid::get(opCtx)->getBalancerConfiguration()->getMaxChunkSizeBytes())); if (splitPoints.empty()) { LOGV2(21873, @@ -1933,9 +1793,6 @@ void ShardingCatalogManager::splitOrMarkJumbo(OperationContext* opCtx, return; } - // Resize the vector because in multiversion scenarios the `autoSplitVector` command may end - // up ignoring the `limit` parameter and returning the whole list of split points. - splitPoints.resize(limit); uassertStatusOK( shardutil::splitChunkAtMultiplePoints(opCtx, chunk.getShardId(), @@ -1957,10 +1814,6 @@ void ShardingCatalogManager::setAllowMigrationsAndBumpOneChunk( bool allowMigrations) { std::set<ShardId> shardsIds; { - // Mark opCtx as interruptible to ensure that all reads and writes to the metadata - // collections under the exclusive _kChunkOpLock happen on the same term. - opCtx->setAlwaysInterruptAtStepDownOrUp(); - // Take _kChunkOpLock in exclusive mode to prevent concurrent chunk splits, merges, and // migrations Lock::ExclusiveLock lk(opCtx, opCtx->lockState(), _kChunkOpLock); @@ -1977,10 +1830,7 @@ void ShardingCatalogManager::setAllowMigrationsAndBumpOneChunk( cm.getAllShardIds(&shardsIds); withTransaction( - opCtx, - CollectionType::ConfigNS, - [this, allowMigrations, &nss, &collectionUUID](OperationContext* opCtx, - TxnNumber txnNumber) { + opCtx, CollectionType::ConfigNS, [&](OperationContext* opCtx, TxnNumber txnNumber) { // Update the 'allowMigrations' field. An unset 'allowMigrations' field implies // 'true'. To ease backwards compatibility we omit 'allowMigrations' instead of // setting it explicitly to 'true'. @@ -2081,13 +1931,11 @@ bool ShardingCatalogManager::clearChunkEstimatedSize(OperationContext* opCtx, co }()}); return updateOp; }()); + request.setWriteConcern(ShardingCatalogClient::kMajorityWriteConcern.toBSON()); auto configShard = Grid::get(opCtx)->shardRegistry()->getConfigShard(); - auto response = configShard->runBatchWriteCommand(opCtx, - Shard::kDefaultConfigCommandTimeout, - request, - ShardingCatalogClient::kMajorityWriteConcern, - Shard::RetryPolicy::kIdempotent); + auto response = configShard->runBatchWriteCommand( + opCtx, Shard::kDefaultConfigCommandTimeout, request, Shard::RetryPolicy::kIdempotent); uassertStatusOK(response.toStatus()); return response.getN() > 0; diff --git a/src/mongo/db/s/config/sharding_catalog_manager_collection_operations.cpp b/src/mongo/db/s/config/sharding_catalog_manager_collection_operations.cpp index 5ad10c12d09..6b2538bfeca 100644 --- a/src/mongo/db/s/config/sharding_catalog_manager_collection_operations.cpp +++ b/src/mongo/db/s/config/sharding_catalog_manager_collection_operations.cpp @@ -288,10 +288,6 @@ std::pair<std::vector<BSONObj>, std::vector<BSONObj>> makeChunkAndTagUpdatesForR void ShardingCatalogManager::refineCollectionShardKey(OperationContext* opCtx, const NamespaceString& nss, const ShardKeyPattern& newShardKeyPattern) { - // Mark opCtx as interruptible to ensure that all reads and writes to the metadata collections - // under the exclusive _kChunkOpLock happen on the same term. - opCtx->setAlwaysInterruptAtStepDownOrUp(); - // Take _kChunkOpLock in exclusive mode to prevent concurrent chunk modifications and generate // strictly monotonously increasing collection versions Lock::ExclusiveLock chunkLk(opCtx, opCtx->lockState(), _kChunkOpLock); @@ -327,80 +323,79 @@ void ShardingCatalogManager::refineCollectionShardKey(OperationContext* opCtx, Timestamp newTimestamp = now.clusterTime().asTimestamp(); collType.setTimestamp(newTimestamp); - auto updateCollectionAndChunksFn = - [this, &nss, &collType, &timers, &newFields](OperationContext* opCtx, TxnNumber txnNumber) { - // Update the config.collections entry for the given namespace. - updateShardingCatalogEntryForCollectionInTxn( - opCtx, nss, collType, false /* upsert */, txnNumber); - - LOGV2(21933, - "refineCollectionShardKey updated collection entry for {namespace}: took " - "{durationMillis} ms. Total time taken: {totalTimeMillis} ms.", - "refineCollectionShardKey updated collection entry", - "namespace"_attr = nss.ns(), - "durationMillis"_attr = timers->executionTimer.millis(), - "totalTimeMillis"_attr = timers->totalTimer.millis()); - timers->executionTimer.reset(); - - if (MONGO_unlikely(hangRefineCollectionShardKeyBeforeUpdatingChunks.shouldFail())) { - LOGV2(21934, "Hit hangRefineCollectionShardKeyBeforeUpdatingChunks failpoint"); - hangRefineCollectionShardKeyBeforeUpdatingChunks.pauseWhileSet(opCtx); - } + auto updateCollectionAndChunksFn = [&](OperationContext* opCtx, TxnNumber txnNumber) { + // Update the config.collections entry for the given namespace. + updateShardingCatalogEntryForCollectionInTxn( + opCtx, nss, collType, false /* upsert */, txnNumber); + + LOGV2(21933, + "refineCollectionShardKey updated collection entry for {namespace}: took " + "{durationMillis} ms. Total time taken: {totalTimeMillis} ms.", + "refineCollectionShardKey updated collection entry", + "namespace"_attr = nss.ns(), + "durationMillis"_attr = timers->executionTimer.millis(), + "totalTimeMillis"_attr = timers->totalTimer.millis()); + timers->executionTimer.reset(); + + if (MONGO_unlikely(hangRefineCollectionShardKeyBeforeUpdatingChunks.shouldFail())) { + LOGV2(21934, "Hit hangRefineCollectionShardKeyBeforeUpdatingChunks failpoint"); + hangRefineCollectionShardKeyBeforeUpdatingChunks.pauseWhileSet(opCtx); + } - auto [chunkUpdates, tagUpdates] = makeChunkAndTagUpdatesForRefine(newFields); + auto [chunkUpdates, tagUpdates] = makeChunkAndTagUpdatesForRefine(newFields); - // Update all config.chunks entries for the given namespace by setting (i) their bounds - // for each new field in the refined key to MinKey (except for the global max chunk - // where the max bounds are set to MaxKey), and unsetting (ii) their jumbo field. - const auto chunksQuery = BSON(ChunkType::collectionUUID << collType.getUuid()); - writeToConfigDocumentInTxn( - opCtx, - ChunkType::ConfigNS, - BatchedCommandRequest::buildPipelineUpdateOp(ChunkType::ConfigNS, - chunksQuery, - chunkUpdates, - false, // upsert - true // useMultiUpdate - ), - txnNumber); + // Update all config.chunks entries for the given namespace by setting (i) their bounds for + // each new field in the refined key to MinKey (except for the global max chunk where the + // max bounds are set to MaxKey), and unsetting (ii) their jumbo field. + const auto chunksQuery = BSON(ChunkType::collectionUUID << collType.getUuid()); + writeToConfigDocumentInTxn( + opCtx, + ChunkType::ConfigNS, + BatchedCommandRequest::buildPipelineUpdateOp(ChunkType::ConfigNS, + chunksQuery, + chunkUpdates, + false, // upsert + true // useMultiUpdate + ), + txnNumber); - LOGV2(21935, - "refineCollectionShardKey: updated chunk entries for {namespace}: took " - "{durationMillis} ms. Total time taken: {totalTimeMillis} ms.", - "refineCollectionShardKey: updated chunk entries", - "namespace"_attr = nss.ns(), - "durationMillis"_attr = timers->executionTimer.millis(), - "totalTimeMillis"_attr = timers->totalTimer.millis()); - timers->executionTimer.reset(); - - // Update all config.tags entries for the given namespace by setting their bounds for - // each new field in the refined key to MinKey (except for the global max tag where the - // max bounds are set to MaxKey). - writeToConfigDocumentInTxn( - opCtx, - TagsType::ConfigNS, - BatchedCommandRequest::buildPipelineUpdateOp(TagsType::ConfigNS, - BSON("ns" << nss.ns()), - tagUpdates, - false, // upsert - true // useMultiUpdate - ), - txnNumber); + LOGV2(21935, + "refineCollectionShardKey: updated chunk entries for {namespace}: took " + "{durationMillis} ms. Total time taken: {totalTimeMillis} ms.", + "refineCollectionShardKey: updated chunk entries", + "namespace"_attr = nss.ns(), + "durationMillis"_attr = timers->executionTimer.millis(), + "totalTimeMillis"_attr = timers->totalTimer.millis()); + timers->executionTimer.reset(); + + // Update all config.tags entries for the given namespace by setting their bounds for + // each new field in the refined key to MinKey (except for the global max tag where the + // max bounds are set to MaxKey). + writeToConfigDocumentInTxn( + opCtx, + TagsType::ConfigNS, + BatchedCommandRequest::buildPipelineUpdateOp(TagsType::ConfigNS, + BSON("ns" << nss.ns()), + tagUpdates, + false, // upsert + true // useMultiUpdate + ), + txnNumber); - LOGV2(21936, - "refineCollectionShardKey: updated zone entries for {namespace}: took " - "{durationMillis} ms. Total time taken: {totalTimeMillis} ms.", - "refineCollectionShardKey: updated zone entries", - "namespace"_attr = nss.ns(), - "durationMillis"_attr = timers->executionTimer.millis(), - "totalTimeMillis"_attr = timers->totalTimer.millis()); + LOGV2(21936, + "refineCollectionShardKey: updated zone entries for {namespace}: took " + "{durationMillis} ms. Total time taken: {totalTimeMillis} ms.", + "refineCollectionShardKey: updated zone entries", + "namespace"_attr = nss.ns(), + "durationMillis"_attr = timers->executionTimer.millis(), + "totalTimeMillis"_attr = timers->totalTimer.millis()); - if (MONGO_unlikely(hangRefineCollectionShardKeyBeforeCommit.shouldFail())) { - LOGV2(21937, "Hit hangRefineCollectionShardKeyBeforeCommit failpoint"); - hangRefineCollectionShardKeyBeforeCommit.pauseWhileSet(opCtx); - } - }; + if (MONGO_unlikely(hangRefineCollectionShardKeyBeforeCommit.shouldFail())) { + LOGV2(21937, "Hit hangRefineCollectionShardKeyBeforeCommit failpoint"); + hangRefineCollectionShardKeyBeforeCommit.pauseWhileSet(opCtx); + } + }; auto updateCollectionAndChunksWithAPIFn = [collType, newFields, nss, timers](const txn_api::TransactionClient& txnClient, @@ -563,42 +558,31 @@ void ShardingCatalogManager::configureCollectionBalancing( boost::optional<bool> defragmentCollection, boost::optional<bool> enableAutoSplitter) { + // Hold the FCV region to serialize with the setFeatureCompatibilityVersion command + FixedFCVRegion fcvRegion(opCtx); + uassert(ErrorCodes::IllegalOperation, + "_configsvrConfigureCollectionBalancing can only be run when the cluster is in feature " + "compatibility versions greater or equal than 5.3.", + serverGlobalParams.featureCompatibility.isGreaterThanOrEqualTo( + multiversion::FeatureCompatibilityVersion::kVersion_5_3)); + uassert(ErrorCodes::InvalidOptions, "invalid configure collection balancing update", chunkSizeMB || defragmentCollection || enableAutoSplitter); - // utility lambda to log the change - auto logConfigureCollectionBalancing = [&]() { - BSONObjBuilder logChangeDetail; - if (chunkSizeMB) { - logChangeDetail.append("chunkSizeMB", chunkSizeMB.get()); - } - - if (defragmentCollection) { - logChangeDetail.append("defragmentCollection", defragmentCollection.get()); - } - - if (enableAutoSplitter) { - logChangeDetail.append("enableAutoSplitter", enableAutoSplitter.get()); - } - - ShardingLogging::get(opCtx)->logChange( - opCtx, "configureCollectionBalancing", nss.ns(), logChangeDetail.obj()); - }; - short updatedFields = 0; BSONObjBuilder updateCmd; { BSONObjBuilder setBuilder(updateCmd.subobjStart("$set")); if (chunkSizeMB && *chunkSizeMB != 0) { - auto chunkSizeBytes = static_cast<int64_t>(*chunkSizeMB) * 1024 * 1024; - bool withinRange = nss == NamespaceString::kLogicalSessionsNamespace - ? (chunkSizeBytes > 0 && chunkSizeBytes <= 1024 * 1024 * 1024) - : ChunkSizeSettingsType::checkMaxChunkSizeValid(chunkSizeBytes); + // verify we got a positive integer in range [1MB, 1GB] uassert(ErrorCodes::InvalidOptions, str::stream() << "Chunk size '" << *chunkSizeMB << "' out of range [1MB, 1GB]", - withinRange); - setBuilder.append(CollectionType::kMaxChunkSizeBytesFieldName, chunkSizeBytes); + *chunkSizeMB > 0 && + *chunkSizeMB < std::numeric_limits<int32_t>::max() / (1024 * 1024) && + ChunkSizeSettingsType::checkMaxChunkSizeValid(*chunkSizeMB * 1024 * 1024)); + setBuilder.append(CollectionType::kMaxChunkSizeBytesFieldName, + *chunkSizeMB * 1024 * 1024); updatedFields++; } if (defragmentCollection) { @@ -624,7 +608,6 @@ void ShardingCatalogManager::configureCollectionBalancing( } if (updatedFields == 0) { - logConfigureCollectionBalancing(); return; } @@ -634,27 +617,26 @@ void ShardingCatalogManager::configureCollectionBalancing( // migrations Lock::ExclusiveLock lk(opCtx, opCtx->lockState(), _kChunkOpLock); - withTransaction(opCtx, - CollectionType::ConfigNS, - [this, &nss, &update](OperationContext* opCtx, TxnNumber txnNumber) { - const auto query = BSON(CollectionType::kNssFieldName << nss.ns()); - const auto res = writeToConfigDocumentInTxn( - opCtx, - CollectionType::ConfigNS, - BatchedCommandRequest::buildUpdateOp(CollectionType::ConfigNS, - query, - update /* update */, - false /* upsert */, - false /* multi */), - txnNumber); - const auto numDocsModified = UpdateOp::parseResponse(res).getN(); - uassert(ErrorCodes::NamespaceNotSharded, - str::stream() << "Expected to match one doc for query " << query - << " but matched " << numDocsModified, - numDocsModified == 1); - - bumpCollectionMinorVersionInTxn(opCtx, nss, txnNumber); - }); + withTransaction( + opCtx, CollectionType::ConfigNS, [&](OperationContext* opCtx, TxnNumber txnNumber) { + const auto query = BSON(CollectionType::kNssFieldName << nss.ns()); + const auto res = writeToConfigDocumentInTxn( + opCtx, + CollectionType::ConfigNS, + BatchedCommandRequest::buildUpdateOp(CollectionType::ConfigNS, + query, + update /* update */, + false /* upsert */, + false /* multi */), + txnNumber); + const auto numDocsModified = UpdateOp::parseResponse(res).getN(); + uassert(ErrorCodes::ConflictingOperationInProgress, + str::stream() << "Expected to match one doc for query " << query + << " but matched " << numDocsModified, + numDocsModified == 1); + + bumpCollectionMinorVersionInTxn(opCtx, nss, txnNumber); + }); // Now any migrations that change the list of shards will see the results of the transaction // during refresh, so it is safe to release the chunk lock. } @@ -672,8 +654,6 @@ void ShardingCatalogManager::configureCollectionBalancing( executor); Balancer::get(opCtx)->notifyPersistedBalancerSettingsChanged(opCtx); - - logConfigureCollectionBalancing(); } void ShardingCatalogManager::renameShardedMetadata( @@ -723,9 +703,7 @@ void ShardingCatalogManager::updateTimeSeriesGranularity(OperationContext* opCtx cm.getAllShardIds(&shardIds); withTransaction( - opCtx, - CollectionType::ConfigNS, - [this, &nss, granularity, &shardIds](OperationContext* opCtx, TxnNumber txnNumber) { + opCtx, CollectionType::ConfigNS, [&](OperationContext* opCtx, TxnNumber txnNumber) { // Update granularity value in config.collections. auto granularityFieldName = CollectionType::kTimeseriesFieldsFieldName + "." + TypeCollectionTimeseriesFields::kGranularityFieldName; diff --git a/src/mongo/db/s/config/sharding_catalog_manager_config_initialization_test.cpp b/src/mongo/db/s/config/sharding_catalog_manager_config_initialization_test.cpp index b320c7f2407..6c95002292b 100644 --- a/src/mongo/db/s/config/sharding_catalog_manager_config_initialization_test.cpp +++ b/src/mongo/db/s/config/sharding_catalog_manager_config_initialization_test.cpp @@ -34,7 +34,7 @@ #include "mongo/bson/json.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/namespace_string.h" #include "mongo/db/operation_context.h" @@ -312,8 +312,6 @@ TEST_F(ConfigInitializationTest, BuildsNecessaryIndexes) { << "_id_"), BSON("v" << 2 << "key" << BSON("ts" << 1) << "name" << "ts_1"), - BSON("v" << 2 << "key" << BSON("process" << 1) << "name" - << "process_1"), BSON("v" << 2 << "key" << BSON("state" << 1 << "process" << 1) << "name" << "state_1_process_1")}; auto expectedShardsIndexes = std::vector<BSONObj>{ diff --git a/src/mongo/db/s/config/sharding_catalog_manager_database_operations.cpp b/src/mongo/db/s/config/sharding_catalog_manager_database_operations.cpp index 561a8911029..934eca19e8e 100644 --- a/src/mongo/db/s/config/sharding_catalog_manager_database_operations.cpp +++ b/src/mongo/db/s/config/sharding_catalog_manager_database_operations.cpp @@ -37,7 +37,6 @@ #include "mongo/db/dbdirectclient.h" #include "mongo/db/namespace_string.h" #include "mongo/db/ops/write_ops.h" -#include "mongo/db/persistent_task_store.h" #include "mongo/db/repl/repl_client_info.h" #include "mongo/db/s/dist_lock_manager.h" #include "mongo/db/server_options.h" @@ -94,14 +93,9 @@ DatabaseType ShardingCatalogManager::createDatabase(OperationContext* opCtx, dbName.toString(), ShardId::kConfigServerId, DatabaseVersion::makeFixed()); } - // It is not allowed to create the 'admin' or 'local' databases, including any alternative - // casing. It is allowed to create the 'config' database (handled by the early return above), - // but only with that exact casing. uassert(ErrorCodes::InvalidOptions, - str::stream() << "Cannot manually create database '" << dbName << "'", - !dbName.equalCaseInsensitive(NamespaceString::kAdminDb) && - !dbName.equalCaseInsensitive(NamespaceString::kLocalDb) && - !dbName.equalCaseInsensitive(NamespaceString::kConfigDb)); + str::stream() << "Cannot manually create database'" << dbName << "'", + dbName != NamespaceString::kAdminDb && dbName != NamespaceString::kLocalDb); uassert(ErrorCodes::InvalidNamespace, str::stream() << "Invalid db name specified: " << dbName, @@ -168,7 +162,7 @@ DatabaseType ShardingCatalogManager::createDatabase(OperationContext* opCtx, // Do another loop, with the db lock held in order to avoid taking the expensive path on // concurrent create database operations dbLock.emplace(DistLockManager::get(opCtx)->lockDirectLocally( - opCtx, str::toLower(dbName), DistLockManager::kDefaultLockTimeout)); + opCtx, dbName, DistLockManager::kDefaultLockTimeout)); } // Expensive createDatabase code path @@ -274,70 +268,4 @@ DatabaseType ShardingCatalogManager::createDatabase(OperationContext* opCtx, return database; } -void ShardingCatalogManager::commitMovePrimary(OperationContext* opCtx, - const StringData& dbName, - const DatabaseVersion& expectedDbVersion, - const ShardId& toShardId) { - // Hold the shard lock until the entire commit finishes to serialize with removeShard. - Lock::SharedLock shardLock(opCtx->lockState(), _kShardMembershipLock); - - const auto toShardDoc = [&] { - DBDirectClient dbClient(opCtx); - return dbClient.findOne(NamespaceString::kConfigsvrShardsNamespace, - BSON(ShardType::name << toShardId)); - }(); - uassert(ErrorCodes::ShardNotFound, - "Requested primary shard {} does not exist"_format(toShardId.toString()), - !toShardDoc.isEmpty()); - - const auto toShardEntry = uassertStatusOK(ShardType::fromBSON(toShardDoc)); - uassert(ErrorCodes::ShardNotFound, - "Requested primary shard {} is draining"_format(toShardId.toString()), - !toShardEntry.getDraining()); - - const auto updateOp = [&] { - const auto query = [&] { - BSONObjBuilder bsonBuilder; - bsonBuilder.append(DatabaseType::kNameFieldName, dbName); - // Include the version in the update filter to be resilient to potential network retries - // and delayed messages. - for (const auto [fieldName, fieldValue] : expectedDbVersion.toBSON()) { - const auto dottedFieldName = DatabaseType::kVersionFieldName + "." + fieldName; - bsonBuilder.appendAs(fieldValue, dottedFieldName); - } - return bsonBuilder.obj(); - }(); - - const auto update = [&] { - auto newDbVersion = expectedDbVersion.makeUpdated(); - const auto now = VectorClock::get(opCtx)->getTime(); - const auto clusterTime = now.clusterTime().asTimestamp(); - - newDbVersion.setTimestamp(clusterTime); - tassert(8235300, - "New database timestamp must be newer than previous one", - newDbVersion.getTimestamp() > expectedDbVersion.getTimestamp()); - - BSONObjBuilder bsonBuilder; - bsonBuilder.append(DatabaseType::kPrimaryFieldName, toShardId); - bsonBuilder.append(DatabaseType::kVersionFieldName, newDbVersion.toBSON()); - return BSON("$set" << bsonBuilder.obj()); - }(); - - write_ops::UpdateCommandRequest updateOp(NamespaceString::kConfigDatabasesNamespace); - updateOp.setUpdates({[&] { - write_ops::UpdateOpEntry entry; - entry.setQ(query); - entry.setU(write_ops::UpdateModification::parseFromClassicUpdate(update)); - return entry; - }()}); - - return updateOp; - }(); - - DBDirectClient dbClient(opCtx); - const auto commandResponse = dbClient.runCommand(updateOp.serialize({})); - uassertStatusOK(getStatusFromWriteCommandReply(commandResponse->getCommandReply())); -} - } // namespace mongo diff --git a/src/mongo/db/s/config/sharding_catalog_manager_database_operations_test.cpp b/src/mongo/db/s/config/sharding_catalog_manager_database_operations_test.cpp deleted file mode 100644 index 029a229d73f..00000000000 --- a/src/mongo/db/s/config/sharding_catalog_manager_database_operations_test.cpp +++ /dev/null @@ -1,100 +0,0 @@ -/** - * Copyright (C) 2023-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/config/config_server_test_fixture.h" -#include "mongo/db/s/config/sharding_catalog_manager.h" - -namespace mongo { -namespace { - -class ShardingCatalogManagerDatabaseOperationsTest : public ConfigServerTestFixture { -public: - void setUp() override { - ConfigServerTestFixture::setUp(); - _opCtx = operationContext(); - } - -protected: - OperationContext* _opCtx; -}; - -TEST_F(ShardingCatalogManagerDatabaseOperationsTest, CreateDatabaseAdminFails) { - ASSERT_THROWS_CODE( - ShardingCatalogManager::get(_opCtx)->createDatabase(_opCtx, "admin"_sd, boost::none), - DBException, - ErrorCodes::InvalidOptions); - - // Alternative capitalizations are also invalid - ASSERT_THROWS_CODE( - ShardingCatalogManager::get(_opCtx)->createDatabase(_opCtx, "Admin"_sd, boost::none), - DBException, - ErrorCodes::InvalidOptions); - - ASSERT_THROWS_CODE( - ShardingCatalogManager::get(_opCtx)->createDatabase(_opCtx, "aDmIn"_sd, boost::none), - DBException, - ErrorCodes::InvalidOptions); -} - -TEST_F(ShardingCatalogManagerDatabaseOperationsTest, CreateDatabaseLocalFails) { - ASSERT_THROWS_CODE( - ShardingCatalogManager::get(_opCtx)->createDatabase(_opCtx, "local"_sd, boost::none), - DBException, - ErrorCodes::InvalidOptions); - - // Alternative capitalizations are also invalid - ASSERT_THROWS_CODE( - ShardingCatalogManager::get(_opCtx)->createDatabase(_opCtx, "Local"_sd, boost::none), - DBException, - ErrorCodes::InvalidOptions); - - ASSERT_THROWS_CODE( - ShardingCatalogManager::get(_opCtx)->createDatabase(_opCtx, "lOcAl"_sd, boost::none), - DBException, - ErrorCodes::InvalidOptions); -} - -TEST_F(ShardingCatalogManagerDatabaseOperationsTest, CreateDatabaseConfig) { - // It is allowed to create the "config" database. - ShardingCatalogManager::get(_opCtx)->createDatabase(_opCtx, "config"_sd, boost::none); - - // But alternative capitalizations are invalid. - ASSERT_THROWS_CODE( - ShardingCatalogManager::get(_opCtx)->createDatabase(_opCtx, "Config"_sd, boost::none), - DBException, - ErrorCodes::InvalidOptions); - - ASSERT_THROWS_CODE( - ShardingCatalogManager::get(_opCtx)->createDatabase(_opCtx, "cOnFiG"_sd, boost::none), - DBException, - ErrorCodes::InvalidOptions); -} - -} // namespace -} // namespace mongo diff --git a/src/mongo/db/s/config/sharding_catalog_manager_merge_chunks_test.cpp b/src/mongo/db/s/config/sharding_catalog_manager_merge_chunks_test.cpp index be08bede01c..d8a006c2286 100644 --- a/src/mongo/db/s/config/sharding_catalog_manager_merge_chunks_test.cpp +++ b/src/mongo/db/s/config/sharding_catalog_manager_merge_chunks_test.cpp @@ -30,15 +30,9 @@ #include "mongo/platform/basic.h" #include "mongo/client/read_preference.h" -#include "mongo/db/dbdirectclient.h" -#include "mongo/db/logical_session_cache_noop.h" #include "mongo/db/namespace_string.h" -#include "mongo/db/read_write_concern_defaults.h" -#include "mongo/db/read_write_concern_defaults_cache_lookup_mock.h" #include "mongo/db/s/config/config_server_test_fixture.h" #include "mongo/db/s/config/sharding_catalog_manager.h" -#include "mongo/db/s/transaction_coordinator_service.h" -#include "mongo/db/session_catalog_mongod.h" #include "mongo/s/catalog/sharding_catalog_client.h" #include "mongo/s/catalog/type_chunk.h" @@ -50,37 +44,18 @@ using unittest::assertGet; class MergeChunkTest : public ConfigServerTestFixture { protected: std::string _shardName = "shard0000"; - void setUp() override { ConfigServerTestFixture::setUp(); - ShardType shard; shard.setName(_shardName); shard.setHost(_shardName + ":12"); setupShards({shard}); - - DBDirectClient client(operationContext()); - client.createCollection(NamespaceString::kSessionTransactionsTableNamespace.ns()); - client.createIndexes(NamespaceString::kSessionTransactionsTableNamespace.ns(), - {MongoDSessionCatalog::getConfigTxnPartialIndexSpec()}); - client.createCollection(CollectionType::ConfigNS.ns()); - - ReadWriteConcernDefaults::create(getServiceContext(), _lookupMock.getFetchDefaultsFn()); - LogicalSessionCache::set(getServiceContext(), std::make_unique<LogicalSessionCacheNoop>()); - TransactionCoordinatorService::get(operationContext()) - ->onShardingInitialization(operationContext(), true); - } - - void tearDown() override { - TransactionCoordinatorService::get(operationContext())->onStepDown(); - ConfigServerTestFixture::tearDown(); } const ShardId _shardId{_shardName}; const NamespaceString _nss1{"TestDB.TestColl1"}; const NamespaceString _nss2{"TestDB.TestColl2"}; const KeyPattern _keyPattern{BSON("x" << 1)}; - ReadWriteConcernDefaultsLookupMock _lookupMock; }; TEST_F(MergeChunkTest, MergeExistingChunksCorrectlyShouldSucceed) { @@ -100,11 +75,6 @@ TEST_F(MergeChunkTest, MergeExistingChunksCorrectlyShouldSucceed) { auto chunk2(chunk); chunk2.setName(OID::gen()); - // set histories - chunk.setHistory({ChunkHistory{Timestamp{100, 0}, _shardId}}); - chunk2.setHistory({ChunkHistory{Timestamp{200, 0}, _shardId}}); - - // set boundaries auto chunkMin = BSON("a" << 1); auto chunkBound = BSON("a" << 5); auto chunkMax = BSON("a" << 10); @@ -117,6 +87,8 @@ TEST_F(MergeChunkTest, MergeExistingChunksCorrectlyShouldSucceed) { setupCollection(_nss1, _keyPattern, {chunk, chunk2}); + Timestamp validAfter{100, 0}; + ChunkRange rangeToBeMerged(chunk.getMin(), chunk2.getMax()); auto versions = assertGet(ShardingCatalogManager::get(operationContext()) @@ -126,7 +98,8 @@ TEST_F(MergeChunkTest, MergeExistingChunksCorrectlyShouldSucceed) { collTimestamp, collUuid, rangeToBeMerged, - _shardId)); + _shardId, + validAfter)); auto collVersion = ChunkVersion::fromBSONPositionalOrNewerFormat(versions["collectionVersion"]); auto shardVersion = ChunkVersion::fromBSONPositionalOrNewerFormat(versions["shardVersion"]); @@ -168,8 +141,7 @@ TEST_F(MergeChunkTest, MergeExistingChunksCorrectlyShouldSucceed) { // Make sure history is there ASSERT_EQ(1UL, mergedChunk.getHistory().size()); - ASSERT_EQ(chunk2.getHistory().front().getValidAfter(), - mergedChunk.getHistory().front().getValidAfter()); + ASSERT_EQ(validAfter, mergedChunk.getHistory().front().getValidAfter()); } TEST_F(MergeChunkTest, MergeSeveralChunksCorrectlyShouldSucceed) { @@ -190,11 +162,6 @@ TEST_F(MergeChunkTest, MergeSeveralChunksCorrectlyShouldSucceed) { chunk2.setName(OID::gen()); chunk3.setName(OID::gen()); - // set histories - chunk.setHistory({ChunkHistory{Timestamp{100, 10}, _shardId}}); - chunk2.setHistory({ChunkHistory{Timestamp{200, 1}, _shardId}}); - chunk3.setHistory({ChunkHistory{Timestamp{50, 0}, _shardId}}); - auto chunkMin = BSON("a" << 1); auto chunkBound = BSON("a" << 5); auto chunkBound2 = BSON("a" << 7); @@ -212,6 +179,8 @@ TEST_F(MergeChunkTest, MergeSeveralChunksCorrectlyShouldSucceed) { setupCollection(_nss1, _keyPattern, {chunk, chunk2, chunk3}); ChunkRange rangeToBeMerged(chunk.getMin(), chunk3.getMax()); + Timestamp validAfter{100, 0}; + ASSERT_OK(ShardingCatalogManager::get(operationContext()) ->commitChunksMerge(operationContext(), _nss1, @@ -219,7 +188,8 @@ TEST_F(MergeChunkTest, MergeSeveralChunksCorrectlyShouldSucceed) { collTimestamp, collUuid, rangeToBeMerged, - _shardId)); + _shardId, + validAfter)); const auto query BSON(ChunkType::collectionUUID() << collUuid); auto findResponse = uassertStatusOK( @@ -250,8 +220,7 @@ TEST_F(MergeChunkTest, MergeSeveralChunksCorrectlyShouldSucceed) { // Make sure history is there ASSERT_EQ(1UL, mergedChunk.getHistory().size()); - ASSERT_EQ(chunk2.getHistory().front().getValidAfter(), - mergedChunk.getHistory().front().getValidAfter()); + ASSERT_EQ(validAfter, mergedChunk.getHistory().front().getValidAfter()); } TEST_F(MergeChunkTest, NewMergeShouldClaimHighestVersion) { @@ -272,10 +241,6 @@ TEST_F(MergeChunkTest, NewMergeShouldClaimHighestVersion) { auto chunk2(chunk); chunk2.setName(OID::gen()); - // set histories - chunk.setHistory({ChunkHistory{Timestamp{100, 0}, _shardId}}); - chunk2.setHistory({ChunkHistory{Timestamp{200, 0}, _shardId}}); - auto chunkMin = BSON("a" << 1); auto chunkBound = BSON("a" << 5); auto chunkMax = BSON("a" << 10); @@ -298,6 +263,8 @@ TEST_F(MergeChunkTest, NewMergeShouldClaimHighestVersion) { setupCollection(_nss1, _keyPattern, {chunk, chunk2, otherChunk}); + Timestamp validAfter{100, 0}; + ASSERT_OK(ShardingCatalogManager::get(operationContext()) ->commitChunksMerge(operationContext(), _nss1, @@ -305,7 +272,8 @@ TEST_F(MergeChunkTest, NewMergeShouldClaimHighestVersion) { collTimestamp, collUuid, rangeToBeMerged, - _shardId)); + _shardId, + validAfter)); const auto query = BSON(ChunkType::collectionUUID() << collUuid); auto findResponse = uassertStatusOK( @@ -336,8 +304,7 @@ TEST_F(MergeChunkTest, NewMergeShouldClaimHighestVersion) { // Make sure history is there ASSERT_EQ(1UL, mergedChunk.getHistory().size()); - ASSERT_EQ(chunk2.getHistory().front().getValidAfter(), - mergedChunk.getHistory().front().getValidAfter()); + ASSERT_EQ(validAfter, mergedChunk.getHistory().front().getValidAfter()); } TEST_F(MergeChunkTest, MergeLeavesOtherChunksAlone) { @@ -357,10 +324,6 @@ TEST_F(MergeChunkTest, MergeLeavesOtherChunksAlone) { auto chunk2(chunk); chunk2.setName(OID::gen()); - // set histories - chunk.setHistory({ChunkHistory{Timestamp{100, 5}, shardId}}); - chunk2.setHistory({ChunkHistory{Timestamp{200, 1}, shardId}}); - auto chunkMin = BSON("a" << 1); auto chunkBound = BSON("a" << 5); auto chunkMax = BSON("a" << 10); @@ -391,7 +354,8 @@ TEST_F(MergeChunkTest, MergeLeavesOtherChunksAlone) { collTimestamp, collUuid, rangeToBeMerged, - shardId)); + shardId, + validAfter)); const auto query = BSON(ChunkType::collectionUUID() << collUuid); auto findResponse = uassertStatusOK( getConfigShard()->exhaustiveFindOnConfig(operationContext(), @@ -434,16 +398,11 @@ TEST_F(MergeChunkTest, NonExistingNamespace) { chunk.setCollectionUUID(UUID::gen()); auto origVersion = ChunkVersion(1, 0, collEpoch, collTimestamp); - chunk.setShard(_shardId); chunk.setVersion(origVersion); // Construct chunk to be merged auto chunk2(chunk); - // set history - chunk.setHistory({ChunkHistory{Timestamp{100, 0}, _shardId}}); - chunk2.setHistory({ChunkHistory{Timestamp{200, 0}, _shardId}}); - auto chunkMin = BSON("a" << 1); auto chunkBound = BSON("a" << 5); auto chunkMax = BSON("a" << 10); @@ -458,6 +417,8 @@ TEST_F(MergeChunkTest, NonExistingNamespace) { setupCollection(_nss1, _keyPattern, {chunk, chunk2}); + Timestamp validAfter{1}; + ASSERT_THROWS(ShardingCatalogManager::get(operationContext()) ->commitChunksMerge(operationContext(), NamespaceString("TestDB.NonExistingColl"), @@ -465,7 +426,8 @@ TEST_F(MergeChunkTest, NonExistingNamespace) { collTimestamp, collUuidAtRequest, rangeToBeMerged, - _shardId), + _shardId, + validAfter), DBException); } @@ -484,10 +446,6 @@ TEST_F(MergeChunkTest, NonMatchingUUIDsOfChunkAndRequestErrors) { // Construct chunk to be merged auto chunk2(chunk); - // set histories - chunk.setHistory({ChunkHistory{Timestamp{100, 0}, _shardId}}); - chunk2.setHistory({ChunkHistory{Timestamp{200, 0}, _shardId}}); - auto chunkMin = BSON("a" << 1); auto chunkBound = BSON("a" << 5); auto chunkMax = BSON("a" << 10); @@ -502,6 +460,8 @@ TEST_F(MergeChunkTest, NonMatchingUUIDsOfChunkAndRequestErrors) { setupCollection(_nss1, _keyPattern, {chunk, chunk2}); + Timestamp validAfter{1}; + auto mergeStatus = ShardingCatalogManager::get(operationContext()) ->commitChunksMerge(operationContext(), _nss1, @@ -509,7 +469,8 @@ TEST_F(MergeChunkTest, NonMatchingUUIDsOfChunkAndRequestErrors) { collTimestamp, requestUuid, rangeToBeMerged, - _shardId); + _shardId, + validAfter); ASSERT_EQ(ErrorCodes::InvalidUUID, mergeStatus); } @@ -533,11 +494,12 @@ TEST_F(MergeChunkTest, MergeAlreadyHappenedSucceeds) { mergedChunk.setName(OID::gen()); mergedChunk.setCollectionUUID(collUuid); mergedChunk.setShard(_shardId); - mergedChunk.setHistory({ChunkHistory{Timestamp{100, 0}, _shardId}}); setupCollection(_nss1, _keyPattern, {mergedChunk}); + Timestamp validAfter{1}; + ASSERT_OK(ShardingCatalogManager::get(operationContext()) ->commitChunksMerge(operationContext(), _nss1, @@ -545,7 +507,8 @@ TEST_F(MergeChunkTest, MergeAlreadyHappenedSucceeds) { collTimestamp, collUuid, rangeToBeMerged, - _shardId)); + _shardId, + validAfter)); // Verify that no change to config.chunks happened. const auto query = BSON(ChunkType::collectionUUID() << collUuid); @@ -592,11 +555,6 @@ TEST_F(MergeChunkTest, MergingChunksWithDollarPrefixShouldSucceed) { auto chunkBound2 = BSON("a" << BSON("$mixKey" << 1)); auto chunkMax = BSON("a" << kMaxBSONKey); - // set histories - chunk1.setHistory({ChunkHistory{Timestamp{100, 9}, _shardId}}); - chunk2.setHistory({ChunkHistory{Timestamp{200, 5}, _shardId}}); - chunk3.setHistory({ChunkHistory{Timestamp{156, 1}, _shardId}}); - // first chunk boundaries chunk1.setMin(chunkMin); chunk1.setMax(chunkBound1); @@ -611,6 +569,7 @@ TEST_F(MergeChunkTest, MergingChunksWithDollarPrefixShouldSucceed) { // Record chunk boundaries for passing into commitChunksMerge ChunkRange rangeToBeMerged(chunk1.getMin(), chunk3.getMax()); + Timestamp validAfter{100, 0}; ASSERT_OK(ShardingCatalogManager::get(operationContext()) ->commitChunksMerge(operationContext(), @@ -619,7 +578,8 @@ TEST_F(MergeChunkTest, MergingChunksWithDollarPrefixShouldSucceed) { collTimestamp, collUuid, rangeToBeMerged, - _shardId)); + _shardId, + validAfter)); const auto query = BSON(ChunkType::collectionUUID() << collUuid); auto findResponse = uassertStatusOK( @@ -650,8 +610,7 @@ TEST_F(MergeChunkTest, MergingChunksWithDollarPrefixShouldSucceed) { // Make sure history is there ASSERT_EQ(1UL, mergedChunk.getHistory().size()); - ASSERT_EQ(chunk2.getHistory().front().getValidAfter(), - mergedChunk.getHistory().front().getValidAfter()); + ASSERT_EQ(validAfter, mergedChunk.getHistory().front().getValidAfter()); } } // namespace diff --git a/src/mongo/db/s/config/sharding_catalog_manager_shard_collection_test.cpp b/src/mongo/db/s/config/sharding_catalog_manager_shard_collection_test.cpp index 1fc9f4e8c85..6b15afaf5b0 100644 --- a/src/mongo/db/s/config/sharding_catalog_manager_shard_collection_test.cpp +++ b/src/mongo/db/s/config/sharding_catalog_manager_shard_collection_test.cpp @@ -40,6 +40,7 @@ #include "mongo/rpc/metadata/tracking_metadata.h" #include "mongo/s/balancer_configuration.h" #include "mongo/s/catalog/type_chunk.h" +#include "mongo/s/client/shard_registry.h" #include "mongo/s/request_types/sharded_ddl_commands_gen.h" #include "mongo/s/shard_key_pattern.h" @@ -143,11 +144,12 @@ TEST_F(CreateFirstChunksTest, NonEmptyCollection_SplitPoints_FromSplitVector_Man targeterFactory()->addTargeterToReturn(connStr, std::move(targeter)); setupShards(kShards); + shardRegistry()->reload(operationContext()); auto uuid = UUID::gen(); CollectionCatalog::write(getServiceContext(), [&](CollectionCatalog& catalog) { - catalog.registerCollection(operationContext(), - std::make_shared<CollectionMock>(uuid, kNamespace)); + catalog.registerCollection( + operationContext(), uuid, std::make_shared<CollectionMock>(kNamespace)); }); auto future = launchAsync([&] { @@ -158,7 +160,7 @@ TEST_F(CreateFirstChunksTest, NonEmptyCollection_SplitPoints_FromSplitVector_Man request.setNumInitialChunks(0); request.setPresplitHashedZones(false); auto optimization = InitialSplitPolicy::calculateOptimizationStrategy( - opCtx.get(), + operationContext(), kShardKeyPattern, request.getNumInitialChunks().get(), request.getPresplitHashedZones().get(), @@ -193,6 +195,7 @@ TEST_F(CreateFirstChunksTest, NonEmptyCollection_SplitPoints_FromClient_ManyChun targeterFactory()->addTargeterToReturn(connStr, std::move(targeter)); setupShards(kShards); + shardRegistry()->reload(operationContext()); auto future = launchAsync([&] { ThreadClient tc("Test", getServiceContext()); @@ -207,7 +210,7 @@ TEST_F(CreateFirstChunksTest, NonEmptyCollection_SplitPoints_FromClient_ManyChun request.setPresplitHashedZones(false); request.setInitialSplitPoints(splitPoints); auto optimization = InitialSplitPolicy::calculateOptimizationStrategy( - opCtx.get(), + operationContext(), kShardKeyPattern, request.getNumInitialChunks().get(), request.getPresplitHashedZones().get(), @@ -231,6 +234,7 @@ TEST_F(CreateFirstChunksTest, NonEmptyCollection_WithZones_OneChunkToPrimary) { ShardType("shard1", "rs1/shard1:123", {"TestZone"}), ShardType("shard2", "rs2/shard2:123")}; setupShards(kShards); + shardRegistry()->reload(operationContext()); std::vector<TagsType> zones{ TagsType(kNamespace, @@ -273,6 +277,7 @@ TEST_F(CreateFirstChunksTest, EmptyCollection_SplitPoints_FromClient_ManyChunksD targeterFactory()->addTargeterToReturn(connStr, std::move(targeter)); setupShards(kShards); + shardRegistry()->reload(operationContext()); auto future = launchAsync([&] { ThreadClient tc("Test", getServiceContext()); @@ -287,7 +292,7 @@ TEST_F(CreateFirstChunksTest, EmptyCollection_SplitPoints_FromClient_ManyChunksD request.setPresplitHashedZones(false); request.setInitialSplitPoints(splitPoints); auto optimization = InitialSplitPolicy::calculateOptimizationStrategy( - opCtx.get(), + operationContext(), kShardKeyPattern, request.getNumInitialChunks().get(), request.getPresplitHashedZones().get(), @@ -298,7 +303,7 @@ TEST_F(CreateFirstChunksTest, EmptyCollection_SplitPoints_FromClient_ManyChunksD ASSERT(optimization->isOptimized()); return optimization->createFirstChunks( - opCtx.get(), kShardKeyPattern, {UUID::gen(), ShardId("shard1")}); + operationContext(), kShardKeyPattern, {UUID::gen(), ShardId("shard1")}); }); const auto& firstChunks = future.default_timed_get(); @@ -322,6 +327,7 @@ TEST_F(CreateFirstChunksTest, EmptyCollection_NoSplitPoints_OneChunkToPrimary) { targeterFactory()->addTargeterToReturn(connStr, std::move(targeter)); setupShards(kShards); + shardRegistry()->reload(operationContext()); auto future = launchAsync([&] { ThreadClient tc("Test", getServiceContext()); @@ -336,7 +342,7 @@ TEST_F(CreateFirstChunksTest, EmptyCollection_NoSplitPoints_OneChunkToPrimary) { request.setPresplitHashedZones(false); request.setInitialSplitPoints(splitPoints); auto optimization = InitialSplitPolicy::calculateOptimizationStrategy( - opCtx.get(), + operationContext(), kShardKeyPattern, request.getNumInitialChunks().get(), request.getPresplitHashedZones().get(), @@ -347,7 +353,7 @@ TEST_F(CreateFirstChunksTest, EmptyCollection_NoSplitPoints_OneChunkToPrimary) { ASSERT(optimization->isOptimized()); return optimization->createFirstChunks( - opCtx.get(), kShardKeyPattern, {UUID::gen(), ShardId("shard1")}); + operationContext(), kShardKeyPattern, {UUID::gen(), ShardId("shard1")}); }); const auto& firstChunks = future.default_timed_get(); @@ -360,6 +366,7 @@ TEST_F(CreateFirstChunksTest, EmptyCollection_WithZones_ManyChunksOnFirstZoneSha ShardType("shard1", "rs1/shard1:123", {"TestZone"}), ShardType("shard2", "rs2/shard2:123")}; setupShards(kShards); + shardRegistry()->reload(operationContext()); std::vector<BSONObj> splitPoints{}; std::vector<TagsType> zones{ diff --git a/src/mongo/db/s/config/sharding_catalog_manager_shard_operations.cpp b/src/mongo/db/s/config/sharding_catalog_manager_shard_operations.cpp index 16c2bde5d02..da32d388438 100644 --- a/src/mongo/db/s/config/sharding_catalog_manager_shard_operations.cpp +++ b/src/mongo/db/s/config/sharding_catalog_manager_shard_operations.cpp @@ -41,6 +41,7 @@ #include "mongo/bson/bsonobj_comparator.h" #include "mongo/bson/util/bson_extract.h" #include "mongo/client/connection_string.h" +#include "mongo/client/fetcher.h" #include "mongo/client/read_preference.h" #include "mongo/client/remote_command_targeter.h" #include "mongo/client/replica_set_monitor.h" @@ -53,7 +54,6 @@ #include "mongo/db/commands/feature_compatibility_version_parser.h" #include "mongo/db/commands/set_cluster_parameter_invocation.h" #include "mongo/db/commands/set_feature_compatibility_version_gen.h" -#include "mongo/db/keys_collection_util.h" #include "mongo/db/namespace_string.h" #include "mongo/db/operation_context.h" #include "mongo/db/persistent_task_store.h" @@ -61,10 +61,8 @@ #include "mongo/db/repl/hello_gen.h" #include "mongo/db/repl/repl_client_info.h" #include "mongo/db/repl/repl_set_config.h" -#include "mongo/db/repl/wait_for_majority_service.h" #include "mongo/db/s/add_shard_cmd_gen.h" #include "mongo/db/s/add_shard_util.h" -#include "mongo/db/s/sharding_config_server_parameters_gen.h" #include "mongo/db/s/sharding_logging.h" #include "mongo/db/s/type_shard_identity.h" #include "mongo/db/s/user_writes_critical_section_document_gen.h" @@ -108,8 +106,6 @@ const WriteConcernOptions kMajorityWriteConcern{WriteConcernOptions::kMajority, WriteConcernOptions::SyncMode::UNSET, WriteConcernOptions::kNoTimeout}; -const Seconds kRemoteCommandTimeout{60}; - /** * Generates a unique name to be given to a newly added shard. */ @@ -157,7 +153,7 @@ StatusWith<std::string> generateNewShardName(OperationContext* opCtx) { StatusWith<Shard::CommandResponse> ShardingCatalogManager::_runCommandForAddShard( OperationContext* opCtx, - std::shared_ptr<RemoteCommandTargeter> targeter, + RemoteCommandTargeter* targeter, StringData dbName, const BSONObj& cmdObj) { auto swHost = targeter->findHost(opCtx, ReadPreferenceSetting{ReadPreference::PrimaryOnly}); @@ -167,7 +163,7 @@ StatusWith<Shard::CommandResponse> ShardingCatalogManager::_runCommandForAddShar auto host = std::move(swHost.getValue()); executor::RemoteCommandRequest request( - host, dbName.toString(), cmdObj, rpc::makeEmptyMetadata(), opCtx, kRemoteCommandTimeout); + host, dbName.toString(), cmdObj, rpc::makeEmptyMetadata(), opCtx, Seconds(60)); executor::RemoteCommandResponse response = Status(ErrorCodes::InternalError, "Internal error running command"); @@ -323,8 +319,8 @@ StatusWith<ShardType> ShardingCatalogManager::_validateHostAsShard( std::shared_ptr<RemoteCommandTargeter> targeter, const std::string* shardProposedName, const ConnectionString& connectionString) { - auto swCommandResponse = - _runCommandForAddShard(opCtx, targeter, NamespaceString::kAdminDb, BSON("hello" << 1)); + auto swCommandResponse = _runCommandForAddShard( + opCtx, targeter.get(), NamespaceString::kAdminDb, BSON("isMaster" << 1)); if (swCommandResponse.getStatus() == ErrorCodes::IncompatibleServerVersion) { return swCommandResponse.getStatus().withReason( str::stream() << "Cannot add " << connectionString.toString() @@ -335,16 +331,17 @@ StatusWith<ShardType> ShardingCatalogManager::_validateHostAsShard( } // Check for a command response error - auto resHelloStatus = std::move(swCommandResponse.getValue().commandStatus); - if (!resHelloStatus.isOK()) { - return resHelloStatus.withContext(str::stream() << "Error running 'hello' against " - << targeter->connectionString().toString()); + auto resIsMasterStatus = std::move(swCommandResponse.getValue().commandStatus); + if (!resIsMasterStatus.isOK()) { + return resIsMasterStatus.withContext(str::stream() + << "Error running isMaster against " + << targeter->connectionString().toString()); } - auto resHello = std::move(swCommandResponse.getValue().response); + auto resIsMaster = std::move(swCommandResponse.getValue().response); // Fail if the node being added is a mongos. - const std::string msg = resHello.getStringField("msg").toString(); + const std::string msg = resIsMaster.getStringField("msg").toString(); if (msg == "isdbgrid") { return {ErrorCodes::IllegalOperation, "cannot add a mongos as a shard"}; } @@ -355,23 +352,23 @@ StatusWith<ShardType> ShardingCatalogManager::_validateHostAsShard( // because of our internal wire version protocol. So we can safely invariant here that the node // is compatible. long long maxWireVersion; - Status status = bsonExtractIntegerField(resHello, "maxWireVersion", &maxWireVersion); + Status status = bsonExtractIntegerField(resIsMaster, "maxWireVersion", &maxWireVersion); if (!status.isOK()) { - return status.withContext(str::stream() << "hello returned invalid 'maxWireVersion' " + return status.withContext(str::stream() << "isMaster returned invalid 'maxWireVersion' " << "field when attempting to add " << connectionString.toString() << " as a shard"); } - // Check whether the host is a writable primary. If not, the replica set may not have been - // initiated. If the connection is a standalone, it will return true for "isWritablePrimary". - bool isWritablePrimary; - status = bsonExtractBooleanField(resHello, "isWritablePrimary", &isWritablePrimary); + // Check whether there is a master. If there isn't, the replica set may not have been + // initiated. If the connection is a standalone, it will return true for isMaster. + bool isMaster; + status = bsonExtractBooleanField(resIsMaster, "ismaster", &isMaster); if (!status.isOK()) { - return status.withContext(str::stream() << "hello returned invalid 'isWritablePrimary' " + return status.withContext(str::stream() << "isMaster returned invalid 'ismaster' " << "field when attempting to add " << connectionString.toString() << " as a shard"); } - if (!isWritablePrimary) { + if (!isMaster) { return {ErrorCodes::NotWritablePrimary, str::stream() << connectionString.toString() @@ -380,7 +377,7 @@ StatusWith<ShardType> ShardingCatalogManager::_validateHostAsShard( } const std::string providedSetName = connectionString.getSetName(); - const std::string foundSetName = resHello["setName"].str(); + const std::string foundSetName = resIsMaster["setName"].str(); // Make sure the specified replica set name (if any) matches the actual shard's replica set if (providedSetName.empty() && !foundSetName.empty()) { @@ -393,7 +390,7 @@ StatusWith<ShardType> ShardingCatalogManager::_validateHostAsShard( if (!providedSetName.empty() && foundSetName.empty()) { return {ErrorCodes::OperationFailed, str::stream() << "host did not return a set name; " - << "is the replica set still initializing? " << resHello}; + << "is the replica set still initializing? " << resIsMaster}; } // Make sure the set name specified in the connection string matches the one where its hosts @@ -405,14 +402,14 @@ StatusWith<ShardType> ShardingCatalogManager::_validateHostAsShard( } // Is it a config server? - if (resHello.hasField("configsvr")) { + if (resIsMaster.hasField("configsvr")) { return {ErrorCodes::OperationFailed, str::stream() << "Cannot add " << connectionString.toString() << " as a shard since it is a config server"}; } - if (resHello.hasField(HelloCommandReply::kIsImplicitDefaultMajorityWCFieldName) && - !resHello.getBoolField(HelloCommandReply::kIsImplicitDefaultMajorityWCFieldName) && + if (resIsMaster.hasField(HelloCommandReply::kIsImplicitDefaultMajorityWCFieldName) && + !resIsMaster.getBoolField(HelloCommandReply::kIsImplicitDefaultMajorityWCFieldName) && !ReadWriteConcernDefaults::get(opCtx).isCWWCSet(opCtx)) { return { ErrorCodes::OperationFailed, @@ -425,11 +422,11 @@ StatusWith<ShardType> ShardingCatalogManager::_validateHostAsShard( "using the setDefaultRWConcern command and try again."}; } - if (resHello.hasField(HelloCommandReply::kCwwcFieldName)) { - auto cwwcOnShard = - WriteConcernOptions::parse(resHello.getObjectField(HelloCommandReply::kCwwcFieldName)) - .getValue() - .toBSON(); + if (resIsMaster.hasField(HelloCommandReply::kCwwcFieldName)) { + auto cwwcOnShard = WriteConcernOptions::parse( + resIsMaster.getObjectField(HelloCommandReply::kCwwcFieldName)) + .getValue() + .toBSON(); auto cachedCWWC = ReadWriteConcernDefaults::get(opCtx).getCWWC(opCtx); if (!cachedCWWC) { @@ -462,20 +459,20 @@ StatusWith<ShardType> ShardingCatalogManager::_validateHostAsShard( if (!providedSetName.empty()) { std::set<std::string> hostSet; - BSONObjIterator iter(resHello["hosts"].Obj()); + BSONObjIterator iter(resIsMaster["hosts"].Obj()); while (iter.more()) { hostSet.insert(iter.next().String()); // host:port } - if (resHello["passives"].isABSONObj()) { - BSONObjIterator piter(resHello["passives"].Obj()); + if (resIsMaster["passives"].isABSONObj()) { + BSONObjIterator piter(resIsMaster["passives"].Obj()); while (piter.more()) { hostSet.insert(piter.next().String()); // host:port } } - if (resHello["arbiters"].isABSONObj()) { - BSONObjIterator piter(resHello["arbiters"].Obj()); + if (resIsMaster["arbiters"].isABSONObj()) { + BSONObjIterator piter(resIsMaster["arbiters"].Obj()); while (piter.more()) { hostSet.insert(piter.next().String()); // host:port } @@ -487,7 +484,7 @@ StatusWith<ShardType> ShardingCatalogManager::_validateHostAsShard( return {ErrorCodes::OperationFailed, str::stream() << "in seed list " << connectionString.toString() << ", host " << host << " does not belong to replica set " << foundSetName - << "; found " << resHello.toString()}; + << "; found " << resIsMaster.toString()}; } } } @@ -529,7 +526,7 @@ Status ShardingCatalogManager::_dropSessionsCollection( } auto swCommandResponse = _runCommandForAddShard( - opCtx, targeter, NamespaceString::kLogicalSessionsNamespace.db(), builder.done()); + opCtx, targeter.get(), NamespaceString::kLogicalSessionsNamespace.db(), builder.done()); if (!swCommandResponse.isOK()) { return swCommandResponse.getStatus(); } @@ -547,7 +544,7 @@ StatusWith<std::vector<std::string>> ShardingCatalogManager::_getDBNamesListFrom auto swCommandResponse = _runCommandForAddShard(opCtx, - targeter, + targeter.get(), NamespaceString::kAdminDb, BSON("listDatabases" << 1 << "nameOnly" << true)); if (!swCommandResponse.isOK()) { @@ -657,11 +654,6 @@ StatusWith<std::string> ShardingCatalogManager::addShard( "collection from the shard manually and try again."); } - auto pullKeysStatus = _pullClusterTimeKeys(opCtx, targeter); - if (!pullKeysStatus.isOK()) { - return pullKeysStatus; - } - // If a name for a shard wasn't provided, generate one if (shardType.getName().empty()) { auto result = generateNewShardName(opCtx); @@ -678,7 +670,7 @@ StatusWith<std::string> ShardingCatalogManager::addShard( // Helper function that runs a command on the to-be shard and returns the status auto runCmdOnNewShard = [this, &opCtx, &targeter](const BSONObj& cmd) -> Status { auto swCommandResponse = - _runCommandForAddShard(opCtx, targeter, NamespaceString::kAdminDb, cmd); + _runCommandForAddShard(opCtx, targeter.get(), NamespaceString::kAdminDb, cmd); if (!swCommandResponse.isOK()) { return swCommandResponse.getStatus(); } @@ -700,10 +692,10 @@ StatusWith<std::string> ShardingCatalogManager::addShard( } // Set the user-writes blocking state on the new shard. - _setUserWriteBlockingStateOnNewShard(opCtx, targeter); + _setUserWriteBlockingStateOnNewShard(opCtx, targeter.get()); // Determine the set of cluster parameters to be used. - _standardizeClusterParameters(opCtx, targeter); + _standardizeClusterParameters(opCtx, targeter.get()); { // Keep the FCV stable across checking the FCV, sending setFCV to the new shard and writing @@ -729,7 +721,7 @@ StatusWith<std::string> ShardingCatalogManager::addShard( auto versionResponse = _runCommandForAddShard(opCtx, - targeter, + targeter.get(), NamespaceString::kAdminDb, setFcvCmd.toBSON(BSON(WriteConcernOptions::kWriteConcernField << opCtx->getWriteConcern().toBSON()))); @@ -1067,8 +1059,8 @@ StatusWith<long long> ShardingCatalogManager::_runCountCommandOnConfig(Operation return result; } -void ShardingCatalogManager::_setUserWriteBlockingStateOnNewShard( - OperationContext* opCtx, std::shared_ptr<RemoteCommandTargeter> targeter) { +void ShardingCatalogManager::_setUserWriteBlockingStateOnNewShard(OperationContext* opCtx, + RemoteCommandTargeter* targeter) { // Delete all the config.user_writes_critical_sections documents from the new shard. { write_ops::DeleteCommandRequest deleteOp( @@ -1134,109 +1126,6 @@ void ShardingCatalogManager::_setUserWriteBlockingStateOnNewShard( }); } -std::unique_ptr<Fetcher> ShardingCatalogManager::_createFetcher( - OperationContext* opCtx, - std::shared_ptr<RemoteCommandTargeter> targeter, - const NamespaceString& nss, - const repl::ReadConcernLevel& readConcernLevel, - FetcherDocsCallbackFn processDocsCallback, - FetcherStatusCallbackFn processStatusCallback) { - auto host = uassertStatusOK( - targeter->findHost(opCtx, ReadPreferenceSetting{ReadPreference::PrimaryOnly})); - - FindCommandRequest findCommand(nss); - const auto readConcern = - repl::ReadConcernArgs(boost::optional<repl::ReadConcernLevel>(readConcernLevel)); - findCommand.setReadConcern(readConcern.toBSONInner()); - const Milliseconds maxTimeMS = - std::min(opCtx->getRemainingMaxTimeMillis(), Milliseconds(kRemoteCommandTimeout)); - findCommand.setMaxTimeMS(durationCount<Milliseconds>(maxTimeMS)); - - auto fetcherCallback = [processDocsCallback, - processStatusCallback](const Fetcher::QueryResponseStatus& dataStatus, - Fetcher::NextAction* nextAction, - BSONObjBuilder* getMoreBob) { - // Throw out any accumulated results on error. - if (!dataStatus.isOK()) { - processStatusCallback(dataStatus.getStatus()); - return; - } - const auto& data = dataStatus.getValue(); - - try { - if (!processDocsCallback(data.documents)) { - *nextAction = Fetcher::NextAction::kNoAction; - } - } catch (DBException& ex) { - processStatusCallback(ex.toStatus()); - return; - } - processStatusCallback(Status::OK()); - - if (!getMoreBob) { - return; - } - getMoreBob->append("getMore", data.cursorId); - getMoreBob->append("collection", data.nss.coll()); - }; - - return std::make_unique<Fetcher>(_executorForAddShard.get(), - host, - nss.db().toString(), - findCommand.toBSON({}), - fetcherCallback, - BSONObj(), /* metadata tracking, only used for shards */ - maxTimeMS, /* command network timeout */ - maxTimeMS /* getMore network timeout */); -} - -Status ShardingCatalogManager::_pullClusterTimeKeys( - OperationContext* opCtx, std::shared_ptr<RemoteCommandTargeter> targeter) { - Status fetchStatus = - Status(ErrorCodes::InternalError, "Internal error running cursor callback in command"); - std::vector<ExternalKeysCollectionDocument> keyDocs; - - auto expireAt = opCtx->getServiceContext()->getFastClockSource()->now() + - Seconds(gNewShardExistingClusterTimeKeysExpirationSecs.load()); - auto fetcher = _createFetcher( - opCtx, - targeter, - NamespaceString::kKeysCollectionNamespace, - repl::ReadConcernLevel::kLocalReadConcern, - [&](const std::vector<BSONObj>& docs) -> bool { - for (const BSONObj& doc : docs) { - keyDocs.push_back(keys_collection_util::makeExternalClusterTimeKeyDoc( - doc.getOwned(), boost::none /* migrationId */, expireAt)); - } - return true; - }, - [&](const Status& status) { fetchStatus = status; }); - - auto scheduleStatus = fetcher->schedule(); - if (!scheduleStatus.isOK()) { - return scheduleStatus; - } - - auto joinStatus = fetcher->join(opCtx); - if (!joinStatus.isOK()) { - return joinStatus; - } - - if (keyDocs.empty()) { - return fetchStatus; - } - - auto opTime = keys_collection_util::storeExternalClusterTimeKeyDocs(opCtx, std::move(keyDocs)); - auto waitStatus = WaitForMajorityService::get(opCtx->getServiceContext()) - .waitUntilMajority(opTime, opCtx->getCancellationToken()) - .getNoThrow(); - if (!waitStatus.isOK()) { - return waitStatus; - } - - return fetchStatus; -} - void ShardingCatalogManager::_setClusterParametersLocally(OperationContext* opCtx, const std::vector<BSONObj>& parameters) { DBDirectClient client(opCtx); @@ -1256,40 +1145,78 @@ void ShardingCatalogManager::_setClusterParametersLocally(OperationContext* opCt } } -void ShardingCatalogManager::_pullClusterParametersFromNewShard( - OperationContext* opCtx, std::shared_ptr<RemoteCommandTargeter> targeter) { +void ShardingCatalogManager::_pullClusterParametersFromNewShard(OperationContext* opCtx, + RemoteCommandTargeter* targeter) { LOGV2(6538600, "Pulling cluster parameters from new shard"); // We can safely query the cluster parameters because the replica set must have been started // with --shardsvr in order to add it into the cluster, and in this mode no setClusterParameter // can be called on the replica set directly. + auto host = uassertStatusOK( + targeter->findHost(opCtx, ReadPreferenceSetting{ReadPreference::PrimaryOnly})); + + const Milliseconds maxTimeMS = + std::min(opCtx->getRemainingMaxTimeMillis(), Milliseconds(Seconds{30})); + BSONObjBuilder findCmdBuilder; + { + FindCommandRequest findCommand(NamespaceString::kClusterParametersNamespace); + auto readConcern = repl::ReadConcernArgs( + boost::optional<repl::ReadConcernLevel>(repl::ReadConcernLevel::kMajorityReadConcern)); + findCommand.setReadConcern(readConcern.toBSONInner()); + findCommand.setMaxTimeMS(durationCount<Milliseconds>(maxTimeMS)); + findCommand.serialize(BSONObj(), &findCmdBuilder); + } // If for some reason the callback never gets invoked, we will return this status in response. - Status fetchStatus = + Status status = Status(ErrorCodes::InternalError, "Internal error running cursor callback in command"); + std::vector<BSONObj> parameters; + auto fetcherCallback = + [this, &status, ¶meters](const Fetcher::QueryResponseStatus& dataStatus, + Fetcher::NextAction* nextAction, + BSONObjBuilder* getMoreBob) { + // Throw out any accumulated results on error + if (!dataStatus.isOK()) { + status = dataStatus.getStatus(); + return; + } + const auto& data = dataStatus.getValue(); + + for (const BSONObj& doc : data.documents) { + parameters.push_back(doc.getOwned()); + } + + status = Status::OK(); + + if (!getMoreBob) { + return; + } + getMoreBob->append("getMore", data.cursorId); + getMoreBob->append("collection", data.nss.coll()); + }; + + Fetcher fetcher(_executorForAddShard.get(), + std::move(host), + NamespaceString::kClusterParametersNamespace.db().toString(), + findCmdBuilder.obj(), + fetcherCallback, + BSONObj(), /* metadata tracking, only used for shards */ + maxTimeMS, /* command network timeout */ + maxTimeMS /* getMore network timeout */); + + uassertStatusOK(fetcher.schedule()); + + uassertStatusOK(fetcher.join(opCtx)); + + uassertStatusOK(status); - auto fetcher = _createFetcher(opCtx, - targeter, - NamespaceString::kClusterParametersNamespace, - repl::ReadConcernLevel::kMajorityReadConcern, - [¶meters](const std::vector<BSONObj>& docs) -> bool { - for (const BSONObj& doc : docs) { - parameters.push_back(doc.getOwned()); - } - return true; - }, - [&fetchStatus](const Status& status) { fetchStatus = status; }); - - uassertStatusOK(fetcher->schedule()); - uassertStatusOK(fetcher->join(opCtx)); - uassertStatusOK(fetchStatus); _setClusterParametersLocally(opCtx, parameters); } void ShardingCatalogManager::_pushClusterParametersToNewShard( OperationContext* opCtx, - std::shared_ptr<RemoteCommandTargeter> targeter, + RemoteCommandTargeter* targeter, const std::vector<BSONObj>& clusterParameters) { LOGV2(6360600, "Pushing cluster parameters into new shard"); @@ -1325,8 +1252,8 @@ void ShardingCatalogManager::_pushClusterParametersToNewShard( } } -void ShardingCatalogManager::_standardizeClusterParameters( - OperationContext* opCtx, std::shared_ptr<RemoteCommandTargeter> targeter) { +void ShardingCatalogManager::_standardizeClusterParameters(OperationContext* opCtx, + RemoteCommandTargeter* targeter) { if (!gFeatureFlagClusterWideConfig.isEnabled(serverGlobalParams.featureCompatibility)) return; diff --git a/src/mongo/db/s/config/sharding_catalog_manager_split_chunk_test.cpp b/src/mongo/db/s/config/sharding_catalog_manager_split_chunk_test.cpp index fb288258edd..cd7f46ebfda 100644 --- a/src/mongo/db/s/config/sharding_catalog_manager_split_chunk_test.cpp +++ b/src/mongo/db/s/config/sharding_catalog_manager_split_chunk_test.cpp @@ -300,72 +300,6 @@ TEST_F(SplitChunkTest, NewSplitShouldClaimHighestVersion) { test(_nss2, Timestamp(42)); } -TEST_F(SplitChunkTest, Idempotency) { - auto test = [&](const NamespaceString& nss, - const Timestamp& collTimestamp, - BSONObj chunkMin, - BSONObj chunkMax, - const std::vector<BSONObj>& splitPoints) { - const auto collEpoch = OID::gen(); - const auto collUuid = UUID::gen(); - - ChunkType chunk; - chunk.setName(OID::gen()); - chunk.setCollectionUUID(collUuid); - - auto origVersion = ChunkVersion(1, 0, collEpoch, collTimestamp); - chunk.setVersion(origVersion); - chunk.setShard(ShardId(_shardName)); - - chunk.setMin(chunkMin); - chunk.setMax(chunkMax); - - setupCollection(nss, _keyPattern, {chunk}); - - const auto doSplit = [&]() { - return ShardingCatalogManager::get(operationContext()) - ->commitChunkSplit(operationContext(), - nss, - collEpoch, - collTimestamp, - ChunkRange(chunkMin, chunkMax), - splitPoints, - "shard0000", - false); - }; - - // Split. - ASSERT_OK(doSplit().getStatus()); - // Retry. - ASSERT_OK(doSplit().getStatus()); - - const auto verifyChunk = [&](BSONObj min, BSONObj max) { - auto chunkDocStatus = - getChunkDoc(operationContext(), collUuid, min, collEpoch, collTimestamp); - ASSERT_OK(chunkDocStatus.getStatus()); - - auto chunkDoc = chunkDocStatus.getValue(); - ASSERT_BSONOBJ_EQ(max, chunkDoc.getMax()); - }; - - // Sanity check. - std::vector<BSONObj> expectedChunkBounds; - expectedChunkBounds.push_back(chunkMin); - expectedChunkBounds.insert( - expectedChunkBounds.end(), splitPoints.begin(), splitPoints.end()); - expectedChunkBounds.push_back(chunkMax); - - for (auto minIt = expectedChunkBounds.begin(); minIt != expectedChunkBounds.end() - 1; - ++minIt) { - auto maxIt = minIt + 1; - verifyChunk(*minIt, *maxIt); - } - }; - - test(_nss1, Timestamp(42), BSON("a" << 1), BSON("a" << 10), {BSON("a" << 5)}); - test(_nss2, Timestamp(42), BSON("a" << 1), BSON("a" << 10), {BSON("a" << 3), BSON("a" << 7)}); -} - TEST_F(SplitChunkTest, PreConditionFailErrors) { auto test = [&](const NamespaceString& nss, const Timestamp& collTimestamp) { const auto collEpoch = OID::gen(); @@ -683,54 +617,5 @@ TEST_F(SplitChunkTest, CantCommitSplitFromChunkSplitterDuringDefragmentation) { false /* fromChunkSplitter*/)); } -TEST_F(SplitChunkTest, SplitJumboChunkShouldUnsetJumboFlag) { - const auto& nss = _nss2; - const auto collTimestamp = Timestamp(42); - const auto collEpoch = OID::gen(); - const auto collUuid = UUID::gen(); - - ChunkType chunk; - chunk.setName(OID::gen()); - chunk.setCollectionUUID(collUuid); - - auto origVersion = ChunkVersion(1, 0, collEpoch, collTimestamp); - chunk.setVersion(origVersion); - chunk.setShard(ShardId(_shardName)); - chunk.setJumbo(true); - - auto chunkMin = BSON("a" << 1); - auto chunkMax = BSON("a" << 10); - chunk.setMin(chunkMin); - chunk.setMax(chunkMax); - - auto chunkSplitPoint = BSON("a" << 5); - std::vector<BSONObj> splitPoints{chunkSplitPoint}; - - setupCollection(nss, _keyPattern, {chunk}); - - ASSERT_EQ(true, chunk.getJumbo()); - - uassertStatusOK(ShardingCatalogManager::get(operationContext()) - ->commitChunkSplit(operationContext(), - nss, - collEpoch, - collTimestamp, - ChunkRange(chunkMin, chunkMax), - splitPoints, - "shard0000", - false /* fromChunkSplitter*/)); - - // Both resulting chunks must not be jumbo - auto chunkDocLeft = - getChunkDoc(operationContext(), collUuid, chunkMin, collEpoch, collTimestamp); - ASSERT_OK(chunkDocLeft.getStatus()); - - auto chunkDocRight = - getChunkDoc(operationContext(), collUuid, chunkSplitPoint, collEpoch, collTimestamp); - ASSERT_OK(chunkDocRight.getStatus()); - - ASSERT_EQ(false, chunkDocLeft.getValue().getJumbo()); - ASSERT_EQ(false, chunkDocRight.getValue().getJumbo()); -} } // namespace } // namespace mongo diff --git a/src/mongo/db/s/config/sharding_catalog_manager_zone_operations.cpp b/src/mongo/db/s/config/sharding_catalog_manager_zone_operations.cpp index f293963ff50..ec31cf0ca9f 100644 --- a/src/mongo/db/s/config/sharding_catalog_manager_zone_operations.cpp +++ b/src/mongo/db/s/config/sharding_catalog_manager_zone_operations.cpp @@ -66,7 +66,7 @@ Status checkForOverlappingZonedKeyRange(OperationContext* opCtx, const ChunkRange& range, const std::string& zoneName, const KeyPattern& shardKeyPattern) { - ZoneInfo zoneInfo; + DistributionStatus chunkDist(nss, ShardToChunksMap{}); auto tagStatus = configServer->exhaustiveFindOnConfig(opCtx, kConfigPrimarySelector, @@ -89,7 +89,7 @@ Status checkForOverlappingZonedKeyRange(OperationContext* opCtx, // Always extend ranges to full shard key to be compatible with tags created before // the zone commands were implemented. const auto& parsedTagDoc = tagParseStatus.getValue(); - auto overlapStatus = zoneInfo.addRangeToZone( + auto overlapStatus = chunkDist.addRangeToZone( ZoneRange(shardKeyPattern.extendRangeBound(parsedTagDoc.getMinKey(), false), shardKeyPattern.extendRangeBound(parsedTagDoc.getMaxKey(), false), parsedTagDoc.getTag())); @@ -99,7 +99,7 @@ Status checkForOverlappingZonedKeyRange(OperationContext* opCtx, } auto overlapStatus = - zoneInfo.addRangeToZone(ZoneRange(range.getMin(), range.getMax(), zoneName)); + chunkDist.addRangeToZone(ZoneRange(range.getMin(), range.getMax(), zoneName)); if (!overlapStatus.isOK()) { return overlapStatus; } |