diff options
Diffstat (limited to 'src/mongo/db/s/config')
18 files changed, 1126 insertions, 477 deletions
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 new file mode 100644 index 00000000000..76fc3f7b5c9 --- /dev/null +++ b/src/mongo/db/s/config/configsvr_commit_move_primary_command.cpp @@ -0,0 +1,102 @@ +/** + * 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_configure_collection_balancing.cpp b/src/mongo/db/s/config/configsvr_configure_collection_balancing.cpp index a9db28064b5..c471c70437f 100644 --- a/src/mongo/db/s/config/configsvr_configure_collection_balancing.cpp +++ b/src/mongo/db/s/config/configsvr_configure_collection_balancing.cpp @@ -36,6 +36,7 @@ #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" @@ -60,6 +61,16 @@ 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 ea2823dcdf0..5df553b5bd3 100644 --- a/src/mongo/db/s/config/configsvr_merge_chunks_command.cpp +++ b/src/mongo/db/s/config/configsvr_merge_chunks_command.cpp @@ -94,8 +94,7 @@ public: request().getTimestamp(), request().getCollectionUUID(), request().getChunkRange(), - request().getShard(), - request().getValidAfter())); + request().getShard())); return ConfigSvrMergeResponse{ChunkVersion::fromBSONPositionalOrNewerFormat( shardAndCollVers[ChunkVersion::kShardVersionField])}; } 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 166c10a4e70..22e3f7b2985 100644 --- a/src/mongo/db/s/config/configsvr_run_restore_command.cpp +++ b/src/mongo/db/s/config/configsvr_run_restore_command.cpp @@ -73,6 +73,25 @@ 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, @@ -146,7 +165,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; + std::set<std::string> databasesRestored = getDatabasesToRestore(opCtx); for (const auto& collectionEntry : kCollectionEntries) { const NamespaceString& nss = collectionEntry.first; @@ -200,10 +219,6 @@ 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 da76faa6593..24030e5fe57 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,6 +61,11 @@ 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 5bdfb55d6c3..55f2d1c534e 100644 --- a/src/mongo/db/s/config/initial_split_policy.cpp +++ b/src/mongo/db/s/config/initial_split_policy.cpp @@ -37,12 +37,14 @@ #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" @@ -54,6 +56,7 @@ namespace { using ChunkDistributionMap = stdx::unordered_map<ShardId, size_t>; using ZoneShardMap = StringMap<std::vector<ShardId>>; +using boost::intrusive_ptr; std::vector<ShardId> getAllShardIdsSorted(OperationContext* opCtx) { // Many tests assume that chunks will be placed on shards @@ -266,7 +269,8 @@ std::unique_ptr<InitialSplitPolicy> InitialSplitPolicy::calculateOptimizationStr const boost::optional<std::vector<BSONObj>>& initialSplitPoints, const std::vector<TagsType>& tags, size_t numShards, - bool collectionIsEmpty) { + bool collectionIsEmpty, + bool useAutoSplitter) { uassert(ErrorCodes::InvalidOptions, str::stream() << "numInitialChunks is only supported when the collection is empty " "and has a hashed field in the shard key pattern", @@ -310,7 +314,11 @@ std::unique_ptr<InitialSplitPolicy> InitialSplitPolicy::calculateOptimizationStr return std::make_unique<SingleChunkOnPrimarySplitPolicy>(); } - return std::make_unique<UnoptimizedSplitPolicy>(); + if (useAutoSplitter) { + return std::make_unique<AutoSplitInChunksOnPrimaryPolicy>(); + } + + return std::make_unique<SingleChunkOnPrimarySplitPolicy>(); } InitialSplitPolicy::ShardCollectionConfig SingleChunkOnPrimarySplitPolicy::createFirstChunks( @@ -334,7 +342,7 @@ InitialSplitPolicy::ShardCollectionConfig SingleChunkOnPrimarySplitPolicy::creat return {std::move(chunks)}; } -InitialSplitPolicy::ShardCollectionConfig UnoptimizedSplitPolicy::createFirstChunks( +InitialSplitPolicy::ShardCollectionConfig AutoSplitInChunksOnPrimaryPolicy::createFirstChunks( OperationContext* opCtx, const ShardKeyPattern& shardKeyPattern, const SplitPolicyParams& params) { @@ -650,34 +658,29 @@ 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) { - projectValBuilder.append(fieldRef->dottedField(), - BSON("$toHashedIndexKey" - << "$" + fieldRef->dottedField())); + arrayToObjectBuilder.emplace_back( + Doc{{"k", V{fieldRef->dottedField()}}, + {"v", Doc{{"$toHashedIndexKey", V{"$" + fieldRef->dottedField()}}}}}); } else { - projectValBuilder.append( - str::stream() << fieldRef->dottedField(), - BSON("$ifNull" << BSON_ARRAY("$" + fieldRef->dottedField() << BSONNULL))); + arrayToObjectBuilder.emplace_back(Doc{ + {"k", V{fieldRef->dottedField()}}, + {"v", Doc{{"$ifNull", V{Arr{V{"$" + fieldRef->dottedField()}, V{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; } @@ -800,26 +803,34 @@ void ReshardingSplitPolicy::_appendSplitPointsFromSample(BSONObjSet* splitPoints while (nextKey && nRemaining > 0) { // if key is hashed, nextKey values are already hashed - auto result = splitPoints->insert( - dotted_path_support::extractElementsBasedOnTemplate(*nextKey, shardKey.toBSON()) - .getOwned()); - + auto result = splitPoints->insert(nextKey->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) { + int samplesPerChunk, + MakePipelineOptions opts) { auto rawPipeline = createRawPipeline(shardKey, numInitialChunks - 1, samplesPerChunk); - StringMap<ExpressionContext::ResolvedNamespace> resolvedNamespaces; resolvedNamespaces[ns.coll()] = {ns, std::vector<BSONObj>{}}; @@ -833,7 +844,7 @@ ReshardingSplitPolicy::_makePipelineDocumentSource(OperationContext* opCtx, boost::none, /* explain */ false, /* fromMongos */ false, /* needsMerge */ - false, /* allowDiskUse */ + true, /* allowDiskUse */ true, /* bypassDocumentValidation */ false, /* isMapReduceCommand */ ns, @@ -843,8 +854,10 @@ ReshardingSplitPolicy::_makePipelineDocumentSource(OperationContext* opCtx, std::move(resolvedNamespaces), boost::none); /* collUUID */ - return std::make_unique<PipelineDocumentSource>(Pipeline::makePipeline(rawPipeline, expCtx, {}), - samplesPerChunk - 1); + expCtx->tempDir = storageGlobalParams.dbpath + "/tmp"; + + return std::make_unique<PipelineDocumentSource>( + Pipeline::makePipeline(rawPipeline, expCtx, opts), 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 ced8d519a13..e492e9c4cb2 100644 --- a/src/mongo/db/s/config/initial_split_policy.h +++ b/src/mongo/db/s/config/initial_split_policy.h @@ -40,7 +40,6 @@ #include "mongo/s/shard_id.h" #include "mongo/s/shard_key_pattern.h" #include "mongo/util/string_map.h" - namespace mongo { struct SplitPolicyParams { @@ -53,6 +52,11 @@ 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, @@ -62,7 +66,8 @@ public: const boost::optional<std::vector<BSONObj>>& initialSplitPoints, const std::vector<TagsType>& tags, size_t numShards, - bool collectionIsEmpty); + bool collectionIsEmpty, + bool useAutoSplitter = true /* Controlled by FCV, see the comment */); virtual ~InitialSplitPolicy() {} @@ -140,7 +145,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 UnoptimizedSplitPolicy : public InitialSplitPolicy { +class AutoSplitInChunksOnPrimaryPolicy : public InitialSplitPolicy { public: ShardCollectionConfig createFirstChunks(OperationContext* opCtx, const ShardKeyPattern& shardKeyPattern, @@ -285,6 +290,7 @@ public: public: virtual ~SampleDocumentSource(){}; virtual boost::optional<BSONObj> getNext() = 0; + virtual Pipeline* getPipeline_forTest() = 0; }; // Provides documents from a real Pipeline @@ -293,6 +299,9 @@ public: PipelineDocumentSource() = delete; PipelineDocumentSource(SampleDocumentPipeline pipeline, int skip); boost::optional<BSONObj> getNext() override; + Pipeline* getPipeline_forTest() override { + return _pipeline.get(); + } private: SampleDocumentPipeline _pipeline; @@ -328,13 +337,21 @@ 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); + int samplesPerChunk, + MakePipelineOptions opts = {}); /** * 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 75d1f2ae0a1..0ef3bee06d1 100644 --- a/src/mongo/db/s/config/initial_split_policy_test.cpp +++ b/src/mongo/db/s/config/initial_split_policy_test.cpp @@ -35,6 +35,7 @@ #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" @@ -1727,6 +1728,10 @@ public: return next; } + Pipeline* getPipeline_forTest() override { + return nullptr; + } + private: std::list<BSONObj> _toReturn; }; @@ -1765,9 +1770,9 @@ TEST_F(ReshardingInitSplitTest, NoZones) { shardRegistry()->reload(operationContext()); std::list<BSONObj> mockSamples; - mockSamples.push_back(BSON("x" << 10 << "y" << 10)); - mockSamples.push_back(BSON("x" << 10 << "y" << 20)); - mockSamples.push_back(BSON("x" << 10 << "y" << 30)); + mockSamples.push_back(BSON("y" << 10)); + mockSamples.push_back(BSON("y" << 20)); + mockSamples.push_back(BSON("y" << 30)); auto mockSampleSource = std::make_unique<MockPipelineSource>(std::move(mockSamples)); @@ -1802,9 +1807,9 @@ TEST_F(ReshardingInitSplitTest, HashedShardKey) { shardRegistry()->reload(operationContext()); std::list<BSONObj> mockSamples; - mockSamples.push_back(BSON("x" << 10 << "y" << 7766103514953448109LL)); - mockSamples.push_back(BSON("x" << 10 << "y" << -9117533237618642180LL)); - mockSamples.push_back(BSON("x" << 10 << "y" << -1196399207910989725LL)); + mockSamples.push_back(BSON("y" << 7766103514953448109LL)); + mockSamples.push_back(BSON("y" << -9117533237618642180LL)); + mockSamples.push_back(BSON("y" << -1196399207910989725LL)); auto mockSampleSource = std::make_unique<MockPipelineSource>(std::move(mockSamples)); @@ -1867,9 +1872,9 @@ TEST_F(ReshardingInitSplitTest, ZonesCoversEntireDomainButInsufficient) { shardRegistry()->reload(operationContext()); std::list<BSONObj> mockSamples; - mockSamples.push_back(BSON("x" << 10 << "y" << 10)); - mockSamples.push_back(BSON("x" << 10 << "y" << 20)); - mockSamples.push_back(BSON("x" << 10 << "y" << 30)); + mockSamples.push_back(BSON("y" << 10)); + mockSamples.push_back(BSON("y" << 20)); + mockSamples.push_back(BSON("y" << 30)); auto mockSampleSource = std::make_unique<MockPipelineSource>(std::move(mockSamples)); @@ -1907,9 +1912,9 @@ TEST_F(ReshardingInitSplitTest, SamplesCoincidingWithZones) { shardRegistry()->reload(operationContext()); std::list<BSONObj> mockSamples; - mockSamples.push_back(BSON("x" << 10 << "y" << 10)); - mockSamples.push_back(BSON("x" << 10 << "y" << 20)); - mockSamples.push_back(BSON("x" << 10 << "y" << 30)); + mockSamples.push_back(BSON("y" << 10)); + mockSamples.push_back(BSON("y" << 20)); + mockSamples.push_back(BSON("y" << 30)); auto mockSampleSource = std::make_unique<MockPipelineSource>(std::move(mockSamples)); diff --git a/src/mongo/db/s/config/sharding_catalog_manager.h b/src/mongo/db/s/config/sharding_catalog_manager.h index 070c1d3d78f..8ec8fb02108 100644 --- a/src/mongo/db/s/config/sharding_catalog_manager.h +++ b/src/mongo/db/s/config/sharding_catalog_manager.h @@ -31,6 +31,7 @@ #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" @@ -271,8 +272,7 @@ public: const boost::optional<Timestamp>& timestamp, const UUID& requestCollectionUUID, const ChunkRange& chunkRange, - const ShardId& shardId, - const boost::optional<Timestamp>& validAfter); + const ShardId& shardId); /** * Updates metadata in config.chunks collection to show the given chunk in its new shard. @@ -348,7 +348,8 @@ public: */ void splitOrMarkJumbo(OperationContext* opCtx, const NamespaceString& nss, - const BSONObj& minKey); + const BSONObj& minKey, + boost::optional<int64_t> optMaxChunkSizeBytes); /** * In a transaction, sets the 'allowMigrations' to the requested state and bumps the collection @@ -396,6 +397,15 @@ 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 // @@ -596,10 +606,11 @@ 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, - RemoteCommandTargeter* targeter, - StringData dbName, - const BSONObj& cmdObj); + StatusWith<Shard::CommandResponse> _runCommandForAddShard( + OperationContext* opCtx, + std::shared_ptr<RemoteCommandTargeter> targeter, + StringData dbName, + const BSONObj& cmdObj); /** * Helper method for running a count command against the config server with appropriate error @@ -636,7 +647,29 @@ private: * Sets the current cluster's user-write blocking state on the shard that is being added. */ void _setUserWriteBlockingStateOnNewShard(OperationContext* opCtx, - RemoteCommandTargeter* targeter); + 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); + /** * Given a vector of cluster parameters in disk format, sets them locally. */ @@ -647,14 +680,14 @@ private: * Gets the cluster parameters set on the shard and then saves them locally. */ void _pullClusterParametersFromNewShard(OperationContext* opCtx, - RemoteCommandTargeter* targeter); + std::shared_ptr<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, - RemoteCommandTargeter* targeter, + std::shared_ptr<RemoteCommandTargeter> targeter, const std::vector<BSONObj>& clusterParameters); /** @@ -662,7 +695,20 @@ 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, RemoteCommandTargeter* targeter); + 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); // 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 0ed9c76d0f0..db1d21033bd 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,11 +41,13 @@ #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" @@ -83,25 +85,32 @@ 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 "isMaster" on the host to determine what + * addShard validates the host as a shard. It calls "hello" on the host to determine what * kind of host it is -- mongos, regular mongod, config mongod -- and whether the replica set - * details are correct. "isMasterResponse" defines the response of the "isMaster" request and + * details are correct. "helloResponse" defines the response of the "hello" 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 isMaster command -- a find query is called first. + * addShard will never reach the "hello" command -- a find query is called first. */ - void expectIsMaster(const HostAndPort& target, StatusWith<BSONObj> isMasterResponse) { - onCommandForAddShard([&, target, isMasterResponse](const RemoteCommandRequest& request) { + void expectHello(const HostAndPort& target, StatusWith<BSONObj> helloResponse) { + onCommandForAddShard([&, target, helloResponse](const RemoteCommandRequest& request) { ASSERT_EQ(request.target, target); ASSERT_EQ(request.dbname, "admin"); - ASSERT_BSONOBJ_EQ(request.cmdObj, BSON("isMaster" << 1)); + ASSERT_BSONOBJ_EQ(request.cmdObj, BSON("hello" << 1)); ASSERT_BSONOBJ_EQ(rpc::makeEmptyMetadata(), request.metadata); - return isMasterResponse; + return helloResponse; }); } @@ -225,7 +234,7 @@ protected: ASSERT_EQ(request.dbname, NamespaceString::kClusterParametersNamespace.db()); ASSERT_BSONOBJ_EQ(request.cmdObj, BSON("find" << NamespaceString::kClusterParametersNamespace.coll() - << "maxTimeMS" << 30000 << "readConcern" + << "maxTimeMS" << 60000 << "readConcern" << BSON("level" << "majority"))); auto cursorRes = CursorResponse(NamespaceString::kClusterParametersNamespace, 0, {}); @@ -233,6 +242,24 @@ 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. @@ -482,9 +509,9 @@ TEST_F(AddShardTest, StandaloneBasicSuccess) { ASSERT_EQUALS(expectedShardName, shardName); }); - BSONObj commandResponse = BSON("ok" << 1 << "ismaster" << true << "maxWireVersion" + BSONObj commandResponse = BSON("ok" << 1 << "isWritablePrimary" << true << "maxWireVersion" << WireVersion::LATEST_WIRE_VERSION); - expectIsMaster(shardTarget, commandResponse); + expectHello(shardTarget, commandResponse); // Get databases list from new shard expectListDatabases( @@ -497,6 +524,9 @@ 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); @@ -569,9 +599,9 @@ TEST_F(AddShardTest, StandaloneGenerateName) { ASSERT_EQUALS(expectedShardName, shardName); }); - BSONObj commandResponse = BSON("ok" << 1 << "ismaster" << true << "maxWireVersion" + BSONObj commandResponse = BSON("ok" << 1 << "isWritablePrimary" << true << "maxWireVersion" << WireVersion::LATEST_WIRE_VERSION); - expectIsMaster(shardTarget, commandResponse); + expectHello(shardTarget, commandResponse); // Get databases list from new shard expectListDatabases( @@ -584,6 +614,9 @@ 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); @@ -673,7 +706,7 @@ TEST_F(AddShardTest, UnreachableHost) { }); Status hostUnreachableStatus = Status(ErrorCodes::HostUnreachable, "host unreachable"); - expectIsMaster(shardTarget, hostUnreachableStatus); + expectHello(shardTarget, hostUnreachableStatus); future.timed_get(kLongFutureTimeout); } @@ -698,9 +731,9 @@ TEST_F(AddShardTest, AddMongosAsShard) { ASSERT_EQUALS(ErrorCodes::IllegalOperation, status); }); - expectIsMaster(shardTarget, - BSON("msg" - << "isdbgrid")); + expectHello(shardTarget, + BSON("msg" + << "isdbgrid")); future.timed_get(kLongFutureTimeout); } @@ -726,10 +759,10 @@ TEST_F(AddShardTest, AddReplicaSetShardAsStandalone) { ASSERT_STRING_CONTAINS(status.getStatus().reason(), "use replica set url format"); }); - BSONObj commandResponse = BSON("ok" << 1 << "ismaster" << true << "setName" + BSONObj commandResponse = BSON("ok" << 1 << "isWritablePrimary" << true << "setName" << "myOtherSet" << "maxWireVersion" << WireVersion::LATEST_WIRE_VERSION); - expectIsMaster(shardTarget, commandResponse); + expectHello(shardTarget, commandResponse); future.timed_get(kLongFutureTimeout); } @@ -756,9 +789,9 @@ TEST_F(AddShardTest, AddStandaloneHostShardAsReplicaSet) { ASSERT_STRING_CONTAINS(status.getStatus().reason(), "host did not return a set name"); }); - BSONObj commandResponse = BSON("ok" << 1 << "ismaster" << true << "maxWireVersion" + BSONObj commandResponse = BSON("ok" << 1 << "isWritablePrimary" << true << "maxWireVersion" << WireVersion::LATEST_WIRE_VERSION); - expectIsMaster(shardTarget, commandResponse); + expectHello(shardTarget, commandResponse); future.timed_get(kLongFutureTimeout); } @@ -785,10 +818,10 @@ TEST_F(AddShardTest, ReplicaSetMistmatchedReplicaSetName) { ASSERT_STRING_CONTAINS(status.getStatus().reason(), "does not match the actual set name"); }); - BSONObj commandResponse = BSON("ok" << 1 << "ismaster" << true << "setName" + BSONObj commandResponse = BSON("ok" << 1 << "isWritablePrimary" << true << "setName" << "myOtherSet" << "maxWireVersion" << WireVersion::LATEST_WIRE_VERSION); - expectIsMaster(shardTarget, commandResponse); + expectHello(shardTarget, commandResponse); future.timed_get(kLongFutureTimeout); } @@ -817,10 +850,10 @@ TEST_F(AddShardTest, ShardIsCSRSConfigServer) { }); BSONObj commandResponse = - BSON("ok" << 1 << "ismaster" << true << "setName" + BSON("ok" << 1 << "isWritablePrimary" << true << "setName" << "config" << "configsvr" << true << "maxWireVersion" << WireVersion::LATEST_WIRE_VERSION); - expectIsMaster(shardTarget, commandResponse); + expectHello(shardTarget, commandResponse); future.timed_get(kLongFutureTimeout); } @@ -850,11 +883,11 @@ TEST_F(AddShardTest, ReplicaSetMissingHostsProvidedInSeedList) { BSONArrayBuilder hosts; hosts.append("host1:12345"); - BSONObj commandResponse = BSON("ok" << 1 << "ismaster" << true << "setName" + BSONObj commandResponse = BSON("ok" << 1 << "isWritablePrimary" << true << "setName" << "mySet" << "hosts" << hosts.arr() << "maxWireVersion" << WireVersion::LATEST_WIRE_VERSION); - expectIsMaster(shardTarget, commandResponse); + expectHello(shardTarget, commandResponse); future.timed_get(kLongFutureTimeout); } @@ -885,11 +918,11 @@ TEST_F(AddShardTest, AddShardWithNameConfigFails) { BSONArrayBuilder hosts; hosts.append("host1:12345"); hosts.append("host2:12345"); - BSONObj commandResponse = BSON("ok" << 1 << "ismaster" << true << "setName" + BSONObj commandResponse = BSON("ok" << 1 << "isWritablePrimary" << true << "setName" << "mySet" << "hosts" << hosts.arr() << "maxWireVersion" << WireVersion::LATEST_WIRE_VERSION); - expectIsMaster(shardTarget, commandResponse); + expectHello(shardTarget, commandResponse); future.timed_get(kLongFutureTimeout); } @@ -931,11 +964,11 @@ TEST_F(AddShardTest, ShardContainsExistingDatabase) { BSONArrayBuilder hosts; hosts.append("host1:12345"); hosts.append("host2:12345"); - BSONObj commandResponse = BSON("ok" << 1 << "ismaster" << true << "setName" + BSONObj commandResponse = BSON("ok" << 1 << "isWritablePrimary" << true << "setName" << "mySet" << "hosts" << hosts.arr() << "maxWireVersion" << WireVersion::LATEST_WIRE_VERSION); - expectIsMaster(shardTarget, commandResponse); + expectHello(shardTarget, commandResponse); expectListDatabases(shardTarget, {BSON("name" << existingDB.getName())}); @@ -975,17 +1008,20 @@ TEST_F(AddShardTest, SuccessfullyAddReplicaSet) { BSONArrayBuilder hosts; hosts.append("host1:12345"); hosts.append("host2:12345"); - BSONObj commandResponse = BSON("ok" << 1 << "ismaster" << true << "setName" + BSONObj commandResponse = BSON("ok" << 1 << "isWritablePrimary" << true << "setName" << "mySet" << "hosts" << hosts.arr() << "maxWireVersion" << WireVersion::LATEST_WIRE_VERSION); - expectIsMaster(shardTarget, commandResponse); + expectHello(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); @@ -1047,17 +1083,20 @@ TEST_F(AddShardTest, ReplicaSetExtraHostsDiscovered) { BSONArrayBuilder hosts; hosts.append("host1:12345"); hosts.append("host2:12345"); - BSONObj commandResponse = BSON("ok" << 1 << "ismaster" << true << "setName" + BSONObj commandResponse = BSON("ok" << 1 << "isWritablePrimary" << true << "setName" << "mySet" << "hosts" << hosts.arr() << "maxWireVersion" << WireVersion::LATEST_WIRE_VERSION); - expectIsMaster(shardTarget, commandResponse); + expectHello(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); @@ -1128,9 +1167,9 @@ TEST_F(AddShardTest, AddShardSucceedsEvenIfAddingDBsFromNewShardFails) { ASSERT_EQUALS(expectedShardName, shardName); }); - BSONObj commandResponse = BSON("ok" << 1 << "ismaster" << true << "maxWireVersion" + BSONObj commandResponse = BSON("ok" << 1 << "isWritablePrimary" << true << "maxWireVersion" << WireVersion::LATEST_WIRE_VERSION); - expectIsMaster(shardTarget, commandResponse); + expectHello(shardTarget, commandResponse); // Get databases list from new shard expectListDatabases( @@ -1143,6 +1182,9 @@ 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 172729b575b..dfdc04f8517 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 @@ -89,76 +89,6 @@ 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. */ @@ -166,13 +96,11 @@ StatusWith<ChunkType> findChunkContainingRange(OperationContext* opCtx, const UUID& uuid, const OID& epoch, const Timestamp& timestamp, - const BSONObj& min, - const BSONObj& max) { + const ChunkRange& range) { const auto chunkQuery = [&]() { BSONObjBuilder queryBuilder; queryBuilder << ChunkType::collectionUUID << uuid; - queryBuilder << ChunkType::min(BSON("$lte" << min)); - queryBuilder << ChunkType::max(BSON("$gte" << max)); + queryBuilder << ChunkType::min(BSON("$lte" << range.getMin())); return queryBuilder.obj(); }(); @@ -184,21 +112,26 @@ StatusWith<ChunkType> findChunkContainingRange(OperationContext* opCtx, repl::ReadConcernLevel::kLocalReadConcern, ChunkType::ConfigNS, chunkQuery, - BSONObj(), - 2 /* limit */); + BSON(ChunkType::min << -1), + 1 /* limit */); if (!findResponseWith.isOK()) { return findResponseWith.getStatus(); } - 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."}; + if (!findResponseWith.getValue().docs.empty()) { + const auto containingChunk = uassertStatusOK(ChunkType::parseFromConfigBSON( + findResponseWith.getValue().docs.front(), epoch, timestamp)); + + if (containingChunk.getRange().covers(range)) { + return containingChunk; + } } - return uassertStatusOK( - ChunkType::parseFromConfigBSON(findResponseWith.getValue().docs.front(), epoch, timestamp)); + 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."}; } BSONObj makeCommitChunkTransactionCommand(const NamespaceString& nss, @@ -570,6 +503,11 @@ 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); @@ -690,6 +628,7 @@ StatusWith<BSONObj> ShardingCatalogManager::commitChunkSplit( newChunk.setMin(startKey); newChunk.setMax(endKey); newChunk.setEstimatedSizeBytes(boost::none); + newChunk.setJumbo(false); op.append("o", newChunk.toConfigBSON()); @@ -784,6 +723,82 @@ StatusWith<BSONObj> ShardingCatalogManager::commitChunkSplit( return response.obj(); } +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"); + }); +} + StatusWith<BSONObj> ShardingCatalogManager::commitChunksMerge( OperationContext* opCtx, const NamespaceString& nss, @@ -791,11 +806,11 @@ StatusWith<BSONObj> ShardingCatalogManager::commitChunksMerge( const boost::optional<Timestamp>& timestamp, const UUID& requestCollectionUUID, const ChunkRange& chunkRange, - const ShardId& shardId, - const boost::optional<Timestamp>& validAfter) { - if (!validAfter) { - return {ErrorCodes::IllegalOperation, "chunk operation requires validAfter timestamp"}; - } + 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(); // Take _kChunkOpLock in exclusive mode to prevent concurrent chunk modifications and generate // strictly monotonously increasing collection versions @@ -871,11 +886,18 @@ StatusWith<BSONObj> ShardingCatalogManager::commitChunksMerge( // 3. Prepare the data for the merge // and ensure that the retrieved list of chunks covers the whole range. - std::vector<ChunkType> chunksToMerge; + + // 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()); 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 @@ -888,46 +910,41 @@ 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); + chunk.getMin().woCompare(chunksToMerge->back().getMax()) == 0); } - chunksToMerge.push_back(std::move(chunk)); + + // 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; + } + } + + 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 - uassertStatusOK(Grid::get(opCtx)->catalogClient()->applyChunkOpsDeprecated( - opCtx, - updates, - preCond, - coll.getUuid(), - nss, - mergeVersion, - WriteConcernOptions(), - repl::ReadConcernLevel::kLocalReadConcern)); + _mergeChunksInTransaction( + opCtx, nss, coll.getUuid(), mergeVersion, validAfter, chunkRange, shardId, chunksToMerge); // 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()); @@ -952,6 +969,15 @@ 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); @@ -1041,12 +1067,8 @@ 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.getMin(), - migratedChunk.getMax()); + auto swCurrentChunk = findChunkContainingRange( + opCtx, coll.getUuid(), coll.getEpoch(), coll.getTimestamp(), migratedChunk.getRange()); if (!swCurrentChunk.isOK()) { return swCurrentChunk.getStatus(); @@ -1263,6 +1285,10 @@ 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); @@ -1395,6 +1421,10 @@ 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); @@ -1513,6 +1543,10 @@ 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); @@ -1710,36 +1744,58 @@ 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, - [&](OperationContext* opCtx, TxnNumber txnNumber) { - for (const auto& nss : collNames) { - bumpCollectionMinorVersion(opCtx, nss, txnNumber); - } - changeMetadataFunc(opCtx, txnNumber); - }, - writeConcern); + withTransaction( + opCtx, + NamespaceString::kConfigReshardingOperationsNamespace, + [&collNames, &changeMetadataFunc](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) { + const BSONObj& minKey, + boost::optional<int64_t> optMaxChunkSizeBytes) { const auto cm = uassertStatusOK( Grid::get(opCtx)->catalogCache()->getShardedCollectionRoutingInfoWithRefresh(opCtx, nss)); auto chunk = cm.findIntersectingChunkWithSimpleCollation(minKey); try { - const auto splitPoints = uassertStatusOK(shardutil::selectChunkSplitPoints( - opCtx, - chunk.getShardId(), - nss, - cm.getShardKeyPattern(), - ChunkRange(chunk.getMin(), chunk.getMax()), - Grid::get(opCtx)->getBalancerConfiguration()->getMaxChunkSizeBytes())); + 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)); if (splitPoints.empty()) { LOGV2(21873, @@ -1793,6 +1849,9 @@ 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(), @@ -1814,6 +1873,10 @@ 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); @@ -1830,7 +1893,10 @@ void ShardingCatalogManager::setAllowMigrationsAndBumpOneChunk( cm.getAllShardIds(&shardsIds); withTransaction( - opCtx, CollectionType::ConfigNS, [&](OperationContext* opCtx, TxnNumber txnNumber) { + opCtx, + CollectionType::ConfigNS, + [this, allowMigrations, &nss, &collectionUUID](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'. 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 6b2538bfeca..1cb326b2bb5 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,6 +288,10 @@ 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); @@ -323,79 +327,80 @@ void ShardingCatalogManager::refineCollectionShardKey(OperationContext* opCtx, Timestamp newTimestamp = now.clusterTime().asTimestamp(); collType.setTimestamp(newTimestamp); - 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 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 [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, @@ -558,14 +563,6 @@ 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); @@ -575,14 +572,14 @@ void ShardingCatalogManager::configureCollectionBalancing( { BSONObjBuilder setBuilder(updateCmd.subobjStart("$set")); if (chunkSizeMB && *chunkSizeMB != 0) { - // verify we got a positive integer in range [1MB, 1GB] + auto chunkSizeBytes = static_cast<int64_t>(*chunkSizeMB) * 1024 * 1024; + bool withinRange = nss == NamespaceString::kLogicalSessionsNamespace + ? (chunkSizeBytes > 0 && chunkSizeBytes <= 1024 * 1024 * 1024) + : ChunkSizeSettingsType::checkMaxChunkSizeValid(chunkSizeBytes); uassert(ErrorCodes::InvalidOptions, str::stream() << "Chunk size '" << *chunkSizeMB << "' out of range [1MB, 1GB]", - *chunkSizeMB > 0 && - *chunkSizeMB < std::numeric_limits<int32_t>::max() / (1024 * 1024) && - ChunkSizeSettingsType::checkMaxChunkSizeValid(*chunkSizeMB * 1024 * 1024)); - setBuilder.append(CollectionType::kMaxChunkSizeBytesFieldName, - *chunkSizeMB * 1024 * 1024); + withinRange); + setBuilder.append(CollectionType::kMaxChunkSizeBytesFieldName, chunkSizeBytes); updatedFields++; } if (defragmentCollection) { @@ -617,26 +614,27 @@ void ShardingCatalogManager::configureCollectionBalancing( // migrations Lock::ExclusiveLock lk(opCtx, opCtx->lockState(), _kChunkOpLock); - 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); - }); + 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); + }); // 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. } @@ -703,7 +701,9 @@ void ShardingCatalogManager::updateTimeSeriesGranularity(OperationContext* opCtx cm.getAllShardIds(&shardIds); withTransaction( - opCtx, CollectionType::ConfigNS, [&](OperationContext* opCtx, TxnNumber txnNumber) { + opCtx, + CollectionType::ConfigNS, + [this, &nss, granularity, &shardIds](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 6c95002292b..39c741092f9 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/write_conflict_exception.h" +#include "mongo/db/concurrency/exception_util.h" #include "mongo/db/curop.h" #include "mongo/db/namespace_string.h" #include "mongo/db/operation_context.h" 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 934eca19e8e..6a8e082b77e 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,6 +37,7 @@ #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" @@ -93,9 +94,14 @@ 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 != NamespaceString::kAdminDb && dbName != NamespaceString::kLocalDb); + str::stream() << "Cannot manually create database '" << dbName << "'", + !dbName.equalCaseInsensitive(NamespaceString::kAdminDb) && + !dbName.equalCaseInsensitive(NamespaceString::kLocalDb) && + !dbName.equalCaseInsensitive(NamespaceString::kConfigDb)); uassert(ErrorCodes::InvalidNamespace, str::stream() << "Invalid db name specified: " << dbName, @@ -162,7 +168,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, dbName, DistLockManager::kDefaultLockTimeout)); + opCtx, str::toLower(dbName), DistLockManager::kDefaultLockTimeout)); } // Expensive createDatabase code path @@ -268,4 +274,63 @@ 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 = [&] { + const auto newDbVersion = expectedDbVersion.makeUpdated(); + + 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 new file mode 100644 index 00000000000..029a229d73f --- /dev/null +++ b/src/mongo/db/s/config/sharding_catalog_manager_database_operations_test.cpp @@ -0,0 +1,100 @@ +/** + * 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 d8a006c2286..be08bede01c 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,9 +30,15 @@ #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" @@ -44,18 +50,37 @@ 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) { @@ -75,6 +100,11 @@ 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); @@ -87,8 +117,6 @@ 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()) @@ -98,8 +126,7 @@ TEST_F(MergeChunkTest, MergeExistingChunksCorrectlyShouldSucceed) { collTimestamp, collUuid, rangeToBeMerged, - _shardId, - validAfter)); + _shardId)); auto collVersion = ChunkVersion::fromBSONPositionalOrNewerFormat(versions["collectionVersion"]); auto shardVersion = ChunkVersion::fromBSONPositionalOrNewerFormat(versions["shardVersion"]); @@ -141,7 +168,8 @@ TEST_F(MergeChunkTest, MergeExistingChunksCorrectlyShouldSucceed) { // Make sure history is there ASSERT_EQ(1UL, mergedChunk.getHistory().size()); - ASSERT_EQ(validAfter, mergedChunk.getHistory().front().getValidAfter()); + ASSERT_EQ(chunk2.getHistory().front().getValidAfter(), + mergedChunk.getHistory().front().getValidAfter()); } TEST_F(MergeChunkTest, MergeSeveralChunksCorrectlyShouldSucceed) { @@ -162,6 +190,11 @@ 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); @@ -179,8 +212,6 @@ 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, @@ -188,8 +219,7 @@ TEST_F(MergeChunkTest, MergeSeveralChunksCorrectlyShouldSucceed) { collTimestamp, collUuid, rangeToBeMerged, - _shardId, - validAfter)); + _shardId)); const auto query BSON(ChunkType::collectionUUID() << collUuid); auto findResponse = uassertStatusOK( @@ -220,7 +250,8 @@ TEST_F(MergeChunkTest, MergeSeveralChunksCorrectlyShouldSucceed) { // Make sure history is there ASSERT_EQ(1UL, mergedChunk.getHistory().size()); - ASSERT_EQ(validAfter, mergedChunk.getHistory().front().getValidAfter()); + ASSERT_EQ(chunk2.getHistory().front().getValidAfter(), + mergedChunk.getHistory().front().getValidAfter()); } TEST_F(MergeChunkTest, NewMergeShouldClaimHighestVersion) { @@ -241,6 +272,10 @@ 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); @@ -263,8 +298,6 @@ TEST_F(MergeChunkTest, NewMergeShouldClaimHighestVersion) { setupCollection(_nss1, _keyPattern, {chunk, chunk2, otherChunk}); - Timestamp validAfter{100, 0}; - ASSERT_OK(ShardingCatalogManager::get(operationContext()) ->commitChunksMerge(operationContext(), _nss1, @@ -272,8 +305,7 @@ TEST_F(MergeChunkTest, NewMergeShouldClaimHighestVersion) { collTimestamp, collUuid, rangeToBeMerged, - _shardId, - validAfter)); + _shardId)); const auto query = BSON(ChunkType::collectionUUID() << collUuid); auto findResponse = uassertStatusOK( @@ -304,7 +336,8 @@ TEST_F(MergeChunkTest, NewMergeShouldClaimHighestVersion) { // Make sure history is there ASSERT_EQ(1UL, mergedChunk.getHistory().size()); - ASSERT_EQ(validAfter, mergedChunk.getHistory().front().getValidAfter()); + ASSERT_EQ(chunk2.getHistory().front().getValidAfter(), + mergedChunk.getHistory().front().getValidAfter()); } TEST_F(MergeChunkTest, MergeLeavesOtherChunksAlone) { @@ -324,6 +357,10 @@ 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); @@ -354,8 +391,7 @@ TEST_F(MergeChunkTest, MergeLeavesOtherChunksAlone) { collTimestamp, collUuid, rangeToBeMerged, - shardId, - validAfter)); + shardId)); const auto query = BSON(ChunkType::collectionUUID() << collUuid); auto findResponse = uassertStatusOK( getConfigShard()->exhaustiveFindOnConfig(operationContext(), @@ -398,11 +434,16 @@ 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); @@ -417,8 +458,6 @@ TEST_F(MergeChunkTest, NonExistingNamespace) { setupCollection(_nss1, _keyPattern, {chunk, chunk2}); - Timestamp validAfter{1}; - ASSERT_THROWS(ShardingCatalogManager::get(operationContext()) ->commitChunksMerge(operationContext(), NamespaceString("TestDB.NonExistingColl"), @@ -426,8 +465,7 @@ TEST_F(MergeChunkTest, NonExistingNamespace) { collTimestamp, collUuidAtRequest, rangeToBeMerged, - _shardId, - validAfter), + _shardId), DBException); } @@ -446,6 +484,10 @@ 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); @@ -460,8 +502,6 @@ TEST_F(MergeChunkTest, NonMatchingUUIDsOfChunkAndRequestErrors) { setupCollection(_nss1, _keyPattern, {chunk, chunk2}); - Timestamp validAfter{1}; - auto mergeStatus = ShardingCatalogManager::get(operationContext()) ->commitChunksMerge(operationContext(), _nss1, @@ -469,8 +509,7 @@ TEST_F(MergeChunkTest, NonMatchingUUIDsOfChunkAndRequestErrors) { collTimestamp, requestUuid, rangeToBeMerged, - _shardId, - validAfter); + _shardId); ASSERT_EQ(ErrorCodes::InvalidUUID, mergeStatus); } @@ -494,12 +533,11 @@ 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, @@ -507,8 +545,7 @@ TEST_F(MergeChunkTest, MergeAlreadyHappenedSucceeds) { collTimestamp, collUuid, rangeToBeMerged, - _shardId, - validAfter)); + _shardId)); // Verify that no change to config.chunks happened. const auto query = BSON(ChunkType::collectionUUID() << collUuid); @@ -555,6 +592,11 @@ 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); @@ -569,7 +611,6 @@ 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(), @@ -578,8 +619,7 @@ TEST_F(MergeChunkTest, MergingChunksWithDollarPrefixShouldSucceed) { collTimestamp, collUuid, rangeToBeMerged, - _shardId, - validAfter)); + _shardId)); const auto query = BSON(ChunkType::collectionUUID() << collUuid); auto findResponse = uassertStatusOK( @@ -610,7 +650,8 @@ TEST_F(MergeChunkTest, MergingChunksWithDollarPrefixShouldSucceed) { // Make sure history is there ASSERT_EQ(1UL, mergedChunk.getHistory().size()); - ASSERT_EQ(validAfter, mergedChunk.getHistory().front().getValidAfter()); + ASSERT_EQ(chunk2.getHistory().front().getValidAfter(), + mergedChunk.getHistory().front().getValidAfter()); } } // namespace 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 da32d388438..16c2bde5d02 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,7 +41,6 @@ #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" @@ -54,6 +53,7 @@ #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,8 +61,10 @@ #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" @@ -106,6 +108,8 @@ 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. */ @@ -153,7 +157,7 @@ StatusWith<std::string> generateNewShardName(OperationContext* opCtx) { StatusWith<Shard::CommandResponse> ShardingCatalogManager::_runCommandForAddShard( OperationContext* opCtx, - RemoteCommandTargeter* targeter, + std::shared_ptr<RemoteCommandTargeter> targeter, StringData dbName, const BSONObj& cmdObj) { auto swHost = targeter->findHost(opCtx, ReadPreferenceSetting{ReadPreference::PrimaryOnly}); @@ -163,7 +167,7 @@ StatusWith<Shard::CommandResponse> ShardingCatalogManager::_runCommandForAddShar auto host = std::move(swHost.getValue()); executor::RemoteCommandRequest request( - host, dbName.toString(), cmdObj, rpc::makeEmptyMetadata(), opCtx, Seconds(60)); + host, dbName.toString(), cmdObj, rpc::makeEmptyMetadata(), opCtx, kRemoteCommandTimeout); executor::RemoteCommandResponse response = Status(ErrorCodes::InternalError, "Internal error running command"); @@ -319,8 +323,8 @@ StatusWith<ShardType> ShardingCatalogManager::_validateHostAsShard( std::shared_ptr<RemoteCommandTargeter> targeter, const std::string* shardProposedName, const ConnectionString& connectionString) { - auto swCommandResponse = _runCommandForAddShard( - opCtx, targeter.get(), NamespaceString::kAdminDb, BSON("isMaster" << 1)); + auto swCommandResponse = + _runCommandForAddShard(opCtx, targeter, NamespaceString::kAdminDb, BSON("hello" << 1)); if (swCommandResponse.getStatus() == ErrorCodes::IncompatibleServerVersion) { return swCommandResponse.getStatus().withReason( str::stream() << "Cannot add " << connectionString.toString() @@ -331,17 +335,16 @@ StatusWith<ShardType> ShardingCatalogManager::_validateHostAsShard( } // Check for a command response error - auto resIsMasterStatus = std::move(swCommandResponse.getValue().commandStatus); - if (!resIsMasterStatus.isOK()) { - return resIsMasterStatus.withContext(str::stream() - << "Error running isMaster against " - << targeter->connectionString().toString()); + auto resHelloStatus = std::move(swCommandResponse.getValue().commandStatus); + if (!resHelloStatus.isOK()) { + return resHelloStatus.withContext(str::stream() << "Error running 'hello' against " + << targeter->connectionString().toString()); } - auto resIsMaster = std::move(swCommandResponse.getValue().response); + auto resHello = std::move(swCommandResponse.getValue().response); // Fail if the node being added is a mongos. - const std::string msg = resIsMaster.getStringField("msg").toString(); + const std::string msg = resHello.getStringField("msg").toString(); if (msg == "isdbgrid") { return {ErrorCodes::IllegalOperation, "cannot add a mongos as a shard"}; } @@ -352,23 +355,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(resIsMaster, "maxWireVersion", &maxWireVersion); + Status status = bsonExtractIntegerField(resHello, "maxWireVersion", &maxWireVersion); if (!status.isOK()) { - return status.withContext(str::stream() << "isMaster returned invalid 'maxWireVersion' " + return status.withContext(str::stream() << "hello returned invalid 'maxWireVersion' " << "field when attempting to add " << connectionString.toString() << " as a shard"); } - // 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); + // 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); if (!status.isOK()) { - return status.withContext(str::stream() << "isMaster returned invalid 'ismaster' " + return status.withContext(str::stream() << "hello returned invalid 'isWritablePrimary' " << "field when attempting to add " << connectionString.toString() << " as a shard"); } - if (!isMaster) { + if (!isWritablePrimary) { return {ErrorCodes::NotWritablePrimary, str::stream() << connectionString.toString() @@ -377,7 +380,7 @@ StatusWith<ShardType> ShardingCatalogManager::_validateHostAsShard( } const std::string providedSetName = connectionString.getSetName(); - const std::string foundSetName = resIsMaster["setName"].str(); + const std::string foundSetName = resHello["setName"].str(); // Make sure the specified replica set name (if any) matches the actual shard's replica set if (providedSetName.empty() && !foundSetName.empty()) { @@ -390,7 +393,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? " << resIsMaster}; + << "is the replica set still initializing? " << resHello}; } // Make sure the set name specified in the connection string matches the one where its hosts @@ -402,14 +405,14 @@ StatusWith<ShardType> ShardingCatalogManager::_validateHostAsShard( } // Is it a config server? - if (resIsMaster.hasField("configsvr")) { + if (resHello.hasField("configsvr")) { return {ErrorCodes::OperationFailed, str::stream() << "Cannot add " << connectionString.toString() << " as a shard since it is a config server"}; } - if (resIsMaster.hasField(HelloCommandReply::kIsImplicitDefaultMajorityWCFieldName) && - !resIsMaster.getBoolField(HelloCommandReply::kIsImplicitDefaultMajorityWCFieldName) && + if (resHello.hasField(HelloCommandReply::kIsImplicitDefaultMajorityWCFieldName) && + !resHello.getBoolField(HelloCommandReply::kIsImplicitDefaultMajorityWCFieldName) && !ReadWriteConcernDefaults::get(opCtx).isCWWCSet(opCtx)) { return { ErrorCodes::OperationFailed, @@ -422,11 +425,11 @@ StatusWith<ShardType> ShardingCatalogManager::_validateHostAsShard( "using the setDefaultRWConcern command and try again."}; } - if (resIsMaster.hasField(HelloCommandReply::kCwwcFieldName)) { - auto cwwcOnShard = WriteConcernOptions::parse( - resIsMaster.getObjectField(HelloCommandReply::kCwwcFieldName)) - .getValue() - .toBSON(); + if (resHello.hasField(HelloCommandReply::kCwwcFieldName)) { + auto cwwcOnShard = + WriteConcernOptions::parse(resHello.getObjectField(HelloCommandReply::kCwwcFieldName)) + .getValue() + .toBSON(); auto cachedCWWC = ReadWriteConcernDefaults::get(opCtx).getCWWC(opCtx); if (!cachedCWWC) { @@ -459,20 +462,20 @@ StatusWith<ShardType> ShardingCatalogManager::_validateHostAsShard( if (!providedSetName.empty()) { std::set<std::string> hostSet; - BSONObjIterator iter(resIsMaster["hosts"].Obj()); + BSONObjIterator iter(resHello["hosts"].Obj()); while (iter.more()) { hostSet.insert(iter.next().String()); // host:port } - if (resIsMaster["passives"].isABSONObj()) { - BSONObjIterator piter(resIsMaster["passives"].Obj()); + if (resHello["passives"].isABSONObj()) { + BSONObjIterator piter(resHello["passives"].Obj()); while (piter.more()) { hostSet.insert(piter.next().String()); // host:port } } - if (resIsMaster["arbiters"].isABSONObj()) { - BSONObjIterator piter(resIsMaster["arbiters"].Obj()); + if (resHello["arbiters"].isABSONObj()) { + BSONObjIterator piter(resHello["arbiters"].Obj()); while (piter.more()) { hostSet.insert(piter.next().String()); // host:port } @@ -484,7 +487,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 " << resIsMaster.toString()}; + << "; found " << resHello.toString()}; } } } @@ -526,7 +529,7 @@ Status ShardingCatalogManager::_dropSessionsCollection( } auto swCommandResponse = _runCommandForAddShard( - opCtx, targeter.get(), NamespaceString::kLogicalSessionsNamespace.db(), builder.done()); + opCtx, targeter, NamespaceString::kLogicalSessionsNamespace.db(), builder.done()); if (!swCommandResponse.isOK()) { return swCommandResponse.getStatus(); } @@ -544,7 +547,7 @@ StatusWith<std::vector<std::string>> ShardingCatalogManager::_getDBNamesListFrom auto swCommandResponse = _runCommandForAddShard(opCtx, - targeter.get(), + targeter, NamespaceString::kAdminDb, BSON("listDatabases" << 1 << "nameOnly" << true)); if (!swCommandResponse.isOK()) { @@ -654,6 +657,11 @@ 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); @@ -670,7 +678,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.get(), NamespaceString::kAdminDb, cmd); + _runCommandForAddShard(opCtx, targeter, NamespaceString::kAdminDb, cmd); if (!swCommandResponse.isOK()) { return swCommandResponse.getStatus(); } @@ -692,10 +700,10 @@ StatusWith<std::string> ShardingCatalogManager::addShard( } // Set the user-writes blocking state on the new shard. - _setUserWriteBlockingStateOnNewShard(opCtx, targeter.get()); + _setUserWriteBlockingStateOnNewShard(opCtx, targeter); // Determine the set of cluster parameters to be used. - _standardizeClusterParameters(opCtx, targeter.get()); + _standardizeClusterParameters(opCtx, targeter); { // Keep the FCV stable across checking the FCV, sending setFCV to the new shard and writing @@ -721,7 +729,7 @@ StatusWith<std::string> ShardingCatalogManager::addShard( auto versionResponse = _runCommandForAddShard(opCtx, - targeter.get(), + targeter, NamespaceString::kAdminDb, setFcvCmd.toBSON(BSON(WriteConcernOptions::kWriteConcernField << opCtx->getWriteConcern().toBSON()))); @@ -1059,8 +1067,8 @@ StatusWith<long long> ShardingCatalogManager::_runCountCommandOnConfig(Operation return result; } -void ShardingCatalogManager::_setUserWriteBlockingStateOnNewShard(OperationContext* opCtx, - RemoteCommandTargeter* targeter) { +void ShardingCatalogManager::_setUserWriteBlockingStateOnNewShard( + OperationContext* opCtx, std::shared_ptr<RemoteCommandTargeter> targeter) { // Delete all the config.user_writes_critical_sections documents from the new shard. { write_ops::DeleteCommandRequest deleteOp( @@ -1126,6 +1134,109 @@ void ShardingCatalogManager::_setUserWriteBlockingStateOnNewShard(OperationConte }); } +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); @@ -1145,78 +1256,40 @@ void ShardingCatalogManager::_setClusterParametersLocally(OperationContext* opCt } } -void ShardingCatalogManager::_pullClusterParametersFromNewShard(OperationContext* opCtx, - RemoteCommandTargeter* targeter) { +void ShardingCatalogManager::_pullClusterParametersFromNewShard( + OperationContext* opCtx, std::shared_ptr<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 status = + Status fetchStatus = 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, - RemoteCommandTargeter* targeter, + std::shared_ptr<RemoteCommandTargeter> targeter, const std::vector<BSONObj>& clusterParameters) { LOGV2(6360600, "Pushing cluster parameters into new shard"); @@ -1252,8 +1325,8 @@ void ShardingCatalogManager::_pushClusterParametersToNewShard( } } -void ShardingCatalogManager::_standardizeClusterParameters(OperationContext* opCtx, - RemoteCommandTargeter* targeter) { +void ShardingCatalogManager::_standardizeClusterParameters( + OperationContext* opCtx, std::shared_ptr<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 cd7f46ebfda..967dd335d44 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 @@ -617,5 +617,54 @@ 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 |