diff options
author | Jessica Yu <jessica.yu@mongodb.com> | 2017-07-17 13:41:53 -0400 |
---|---|---|
committer | Jessica Yu <jessica.yu@mongodb.com> | 2017-07-22 09:52:38 -0400 |
commit | 0bef84ea178a17327e95ef07afb82c6948df1c1f (patch) | |
tree | 6b11a754d355dfd7c7550462f63911c94f7e1d43 /src/mongo/s | |
parent | db9f34fdec33e2250dad53d7d2d3a829507b3f95 (diff) | |
download | mongo-0bef84ea178a17327e95ef07afb82c6948df1c1f.tar.gz |
SERVER-29660 move the logic of the enableSharding into the new _configsvrEnableSharding command
SERVER-30168 Make the _configsvrEnableSharding command idempotent
Diffstat (limited to 'src/mongo/s')
-rw-r--r-- | src/mongo/s/catalog/SConscript | 3 | ||||
-rw-r--r-- | src/mongo/s/catalog/sharding_catalog_client.h | 35 | ||||
-rw-r--r-- | src/mongo/s/catalog/sharding_catalog_client_impl.cpp | 54 | ||||
-rw-r--r-- | src/mongo/s/catalog/sharding_catalog_client_impl.h | 28 | ||||
-rw-r--r-- | src/mongo/s/catalog/sharding_catalog_client_mock.cpp | 6 | ||||
-rw-r--r-- | src/mongo/s/catalog/sharding_catalog_client_mock.h | 4 | ||||
-rw-r--r-- | src/mongo/s/catalog/sharding_catalog_enable_sharding_test.cpp | 200 | ||||
-rw-r--r-- | src/mongo/s/catalog/sharding_catalog_manager.h | 17 | ||||
-rw-r--r-- | src/mongo/s/catalog/sharding_catalog_manager_database_operations_impl.cpp | 93 | ||||
-rw-r--r-- | src/mongo/s/catalog/sharding_catalog_test.cpp | 272 | ||||
-rw-r--r-- | src/mongo/s/commands/cluster_enable_sharding_cmd.cpp | 42 | ||||
-rw-r--r-- | src/mongo/s/commands/cluster_map_reduce_cmd.cpp | 9 |
12 files changed, 387 insertions, 376 deletions
diff --git a/src/mongo/s/catalog/SConscript b/src/mongo/s/catalog/SConscript index 367c101675d..215dee2c731 100644 --- a/src/mongo/s/catalog/SConscript +++ b/src/mongo/s/catalog/SConscript @@ -142,6 +142,7 @@ env.Library( 'sharding_catalog_manager_chunk_operations_impl.cpp', 'sharding_catalog_manager_collection_operations_impl.cpp', 'sharding_catalog_manager.cpp', + 'sharding_catalog_manager_database_operations_impl.cpp', 'sharding_catalog_manager_shard_operations_impl.cpp', 'sharding_catalog_manager_zone_operations_impl.cpp', ], @@ -153,6 +154,7 @@ env.Library( '$BUILD_DIR/mongo/executor/network_interface', '$BUILD_DIR/mongo/s/client/sharding_client', '$BUILD_DIR/mongo/s/coreshard', + 'sharding_catalog_client', ], ) @@ -174,6 +176,7 @@ env.CppUnitTest( 'sharding_catalog_assign_key_range_to_zone_test.cpp', 'sharding_catalog_commit_chunk_migration_test.cpp', 'sharding_catalog_config_initialization_test.cpp', + 'sharding_catalog_enable_sharding_test.cpp', 'sharding_catalog_merge_chunks_test.cpp', 'sharding_catalog_remove_shard_from_zone_test.cpp', 'sharding_catalog_shard_collection_test.cpp', diff --git a/src/mongo/s/catalog/sharding_catalog_client.h b/src/mongo/s/catalog/sharding_catalog_client.h index d7045f8821b..6baceb95da3 100644 --- a/src/mongo/s/catalog/sharding_catalog_client.h +++ b/src/mongo/s/catalog/sharding_catalog_client.h @@ -56,6 +56,7 @@ class DatabaseType; class LogicalTime; class NamespaceString; class OperationContext; +class ShardingCatalogManager; class ShardKeyPattern; class ShardRegistry; class ShardType; @@ -93,6 +94,13 @@ enum ShardDrainingStatus { class ShardingCatalogClient { MONGO_DISALLOW_COPYING(ShardingCatalogClient); + // Allows ShardingCatalogManager to access _checkDbDoesNotExist + // TODO: move _checkDbDoesNotExist to ShardingCatalogManager when + // ShardingCatalogClient::createDatabaseCommand, the other caller of this function, + // is moved into ShardingCatalogManager. + // SERVER-30022. + friend class ShardingCatalogManager; + public: // Constant to use for configuration data majority writes static const WriteConcernOptions kMajorityWriteConcern; @@ -112,18 +120,6 @@ public: virtual void shutDown(OperationContext* opCtx) = 0; /** - * Creates a new database or updates the sharding status for an existing one. Cannot be - * used for the admin/config/local DBs, which should not be created or sharded manually - * anyways. - * - * Returns Status::OK on success or any error code indicating the failure. These are some - * of the known failures: - * - DatabaseDifferCase - database already exists, but with a different case - * - ShardNotFound - could not find a shard to place the DB on - */ - virtual Status enableSharding(OperationContext* opCtx, const std::string& dbName) = 0; - - /** * Tries to remove a shard. To completely remove a shard from a sharded cluster, * the data residing in that shard must be moved to the remaining shards in the * cluster by "draining" chunks from that shard. @@ -424,6 +420,21 @@ public: protected: ShardingCatalogClient() = default; + +private: + /** + * Checks that the given database name doesn't already exist in the config.databases + * collection, including under different casing. Optional db can be passed and will + * be set with the database details if the given dbName exists. + * + * Returns OK status if the db does not exist. + * Some known errors include: + * NamespaceExists if it exists with the same casing + * DatabaseDifferCase if it exists under different casing. + */ + virtual Status _checkDbDoesNotExist(OperationContext* opCtx, + const std::string& dbName, + DatabaseType* db) = 0; }; } // namespace mongo diff --git a/src/mongo/s/catalog/sharding_catalog_client_impl.cpp b/src/mongo/s/catalog/sharding_catalog_client_impl.cpp index 327b1409a68..ec735f84453 100644 --- a/src/mongo/s/catalog/sharding_catalog_client_impl.cpp +++ b/src/mongo/s/catalog/sharding_catalog_client_impl.cpp @@ -317,60 +317,6 @@ StatusWith<ShardId> ShardingCatalogClientImpl::_selectShardForNewDatabase( return candidateShardId; } -Status ShardingCatalogClientImpl::enableSharding(OperationContext* opCtx, - const std::string& dbName) { - invariant(nsIsDbOnly(dbName)); - - if (dbName == NamespaceString::kConfigDb || dbName == NamespaceString::kAdminDb) { - return { - ErrorCodes::IllegalOperation, - str::stream() << "Enabling sharding on system configuration databases is not allowed"}; - } - - // Lock the database globally to prevent conflicts with simultaneous database - // creation/modification. - auto scopedDistLock = getDistLockManager()->lock( - opCtx, dbName, "enableSharding", DistLockManager::kDefaultLockTimeout); - if (!scopedDistLock.isOK()) { - return scopedDistLock.getStatus(); - } - - // Check for case sensitivity violations - DatabaseType db; - - Status status = _checkDbDoesNotExist(opCtx, dbName, &db); - if (status.isOK()) { - // Database does not exist, create a new entry - auto newShardIdStatus = - _selectShardForNewDatabase(opCtx, Grid::get(opCtx)->shardRegistry()); - if (!newShardIdStatus.isOK()) { - return newShardIdStatus.getStatus(); - } - - const ShardId& newShardId = newShardIdStatus.getValue(); - - log() << "Placing [" << dbName << "] on: " << newShardId; - - db.setName(dbName); - db.setPrimary(newShardId); - db.setSharded(true); - } else if (status.code() == ErrorCodes::NamespaceExists) { - if (db.getSharded()) { - return Status(ErrorCodes::AlreadyInitialized, - str::stream() << "sharding already enabled for database " << dbName); - } - - // Database exists, so just update it - db.setSharded(true); - } else { - return status; - } - - log() << "Enabling sharding for database [" << dbName << "] in config db"; - - return updateDatabase(opCtx, dbName, db); -} - Status ShardingCatalogClientImpl::_log(OperationContext* opCtx, const StringData& logCollName, const std::string& what, diff --git a/src/mongo/s/catalog/sharding_catalog_client_impl.h b/src/mongo/s/catalog/sharding_catalog_client_impl.h index c55899ec91c..68536fb7df8 100644 --- a/src/mongo/s/catalog/sharding_catalog_client_impl.h +++ b/src/mongo/s/catalog/sharding_catalog_client_impl.h @@ -48,6 +48,14 @@ class TaskExecutor; * Implements the catalog client for reading from replica set config servers. */ class ShardingCatalogClientImpl final : public ShardingCatalogClient { + + // Allows ShardingCatalogManager to access _selectShardForNewDatabase + // TODO: move _selectShardForNewDatabase to ShardingCatalogManager, when + // ShardingCatalogClient::createDatabaseCommand, the other caller of this function, + // is moved into ShardingCatalogManager. + // SERVER-30022. + friend class ShardingCatalogManager; + public: /* * Updates (or if "upsert" is true, creates) catalog data for the sharded collection "collNs" by @@ -70,8 +78,6 @@ public: void shutDown(OperationContext* opCtx) override; - Status enableSharding(OperationContext* opCtx, const std::string& dbName) override; - Status updateDatabase(OperationContext* opCtx, const std::string& dbName, const DatabaseType& db) override; @@ -190,6 +196,10 @@ public: repl::ReadConcernLevel readConcernLevel) override; private: + Status _checkDbDoesNotExist(OperationContext* opCtx, + const std::string& dbName, + DatabaseType* db) override; + /** * Selects an optimal shard on which to place a newly created database from the set of * available shards. Will return ShardNotFound if shard could not be found. @@ -217,20 +227,6 @@ private: const WriteConcernOptions& writeConcern); /** - * Checks that the given database name doesn't already exist in the config.databases - * collection, including under different casing. Optional db can be passed and will - * be set with the database details if the given dbName exists. - * - * Returns OK status if the db does not exist. - * Some known errors include: - * NamespaceExists if it exists with the same casing - * DatabaseDifferCase if it exists under different casing. - */ - Status _checkDbDoesNotExist(OperationContext* opCtx, - const std::string& dbName, - DatabaseType* db); - - /** * Creates the specified collection name in the config database. */ Status _createCappedConfigCollection(OperationContext* opCtx, diff --git a/src/mongo/s/catalog/sharding_catalog_client_mock.cpp b/src/mongo/s/catalog/sharding_catalog_client_mock.cpp index b37fdbc5ec2..90e7c4dd8a7 100644 --- a/src/mongo/s/catalog/sharding_catalog_client_mock.cpp +++ b/src/mongo/s/catalog/sharding_catalog_client_mock.cpp @@ -226,4 +226,10 @@ StatusWith<std::vector<KeysCollectionDocument>> ShardingCatalogClientMock::getNe return {ErrorCodes::InternalError, "Method not implemented"}; } +Status ShardingCatalogClientMock::_checkDbDoesNotExist(OperationContext* opCtx, + const string& dbName, + DatabaseType* db) { + return {ErrorCodes::InternalError, "Method not implemented"}; +} + } // namespace mongo diff --git a/src/mongo/s/catalog/sharding_catalog_client_mock.h b/src/mongo/s/catalog/sharding_catalog_client_mock.h index 1965dfb9f2d..574e3ad30d3 100644 --- a/src/mongo/s/catalog/sharding_catalog_client_mock.h +++ b/src/mongo/s/catalog/sharding_catalog_client_mock.h @@ -158,6 +158,10 @@ public: private: std::unique_ptr<DistLockManager> _distLockManager; + + Status _checkDbDoesNotExist(OperationContext* opCtx, + const std::string& dbName, + DatabaseType* db) override; }; } // namespace mongo diff --git a/src/mongo/s/catalog/sharding_catalog_enable_sharding_test.cpp b/src/mongo/s/catalog/sharding_catalog_enable_sharding_test.cpp new file mode 100644 index 00000000000..7cccb916106 --- /dev/null +++ b/src/mongo/s/catalog/sharding_catalog_enable_sharding_test.cpp @@ -0,0 +1,200 @@ +/** + * Copyright (C) 2017 MongoDB Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License, version 3, + * as published by the Free Software Foundation. + * + * 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 + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see <http://www.gnu.org/licenses/>. + * + * 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 GNU Affero General 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_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kSharding + +#include "mongo/platform/basic.h" + +#include <pcrecpp.h> + +#include "mongo/bson/json.h" +#include "mongo/client/remote_command_targeter_mock.h" +#include "mongo/db/commands.h" +#include "mongo/db/query/query_request.h" +#include "mongo/db/repl/read_concern_args.h" +#include "mongo/executor/task_executor.h" +#include "mongo/rpc/get_status_from_command_result.h" +#include "mongo/rpc/metadata/repl_set_metadata.h" +#include "mongo/rpc/metadata/tracking_metadata.h" +#include "mongo/s/catalog/dist_lock_catalog_impl.h" +#include "mongo/s/catalog/sharding_catalog_manager.h" +#include "mongo/s/catalog/type_database.h" +#include "mongo/s/catalog/type_locks.h" +#include "mongo/s/catalog/type_shard.h" +#include "mongo/s/catalog/type_tags.h" +#include "mongo/s/chunk_version.h" +#include "mongo/s/client/shard_registry.h" +#include "mongo/s/config_server_test_fixture.h" +#include "mongo/s/write_ops/batched_command_response.h" +#include "mongo/s/write_ops/batched_insert_request.h" +#include "mongo/s/write_ops/batched_update_request.h" +#include "mongo/stdx/future.h" +#include "mongo/util/log.h" +#include "mongo/util/scopeguard.h" +#include "mongo/util/time_support.h" + +namespace mongo { +namespace { + +using executor::RemoteCommandRequest; +using std::vector; + +class EnableShardingTest : public ConfigServerTestFixture {}; + +TEST_F(EnableShardingTest, noDBExists) { + ShardType shard; + shard.setName("shard0"); + shard.setHost("shard0:12"); + + ASSERT_OK(setupShards(vector<ShardType>{shard})); + + auto shardTargeter = RemoteCommandTargeterMock::get( + uassertStatusOK(shardRegistry()->getShard(operationContext(), ShardId("shard0"))) + ->getTargeter()); + shardTargeter->setFindHostReturnValue(HostAndPort("shard0:12")); + + auto future = launchAsync([&] { + ON_BLOCK_EXIT([&] { Client::destroy(); }); + Client::initThreadIfNotAlready("Test"); + auto opCtx = cc().makeOperationContext(); + auto status = ShardingCatalogManager::get(opCtx.get())->enableSharding(opCtx.get(), "db1"); + ASSERT_OK(status); + }); + + // list databases for checking shard size. + onCommand([](const RemoteCommandRequest& request) { + ASSERT_EQ(HostAndPort("shard0:12"), request.target); + ASSERT_EQ("admin", request.dbname); + ASSERT_BSONOBJ_EQ(BSON("listDatabases" << 1 << "maxTimeMS" << 600000), request.cmdObj); + + ASSERT_BSONOBJ_EQ( + ReadPreferenceSetting(ReadPreference::PrimaryPreferred).toContainingBSON(), + rpc::TrackingMetadata::removeTrackingData(request.metadata)); + + return fromjson(R"({ + databases: [], + totalSize: 1, + ok: 1 + })"); + }); + + future.timed_get(kFutureTimeout); +} + +TEST_F(EnableShardingTest, lockBusy) { + std::string db = "db2"; + ASSERT_OK(distLockCatalog() + ->grabLock(operationContext(), + db, + OID::gen(), + "dummyWho", + "dummyProcessId", + Date_t::now(), + "dummyReason") + .getStatus()); + + auto status = + ShardingCatalogManager::get(operationContext())->enableSharding(operationContext(), db); + ASSERT_EQ(ErrorCodes::LockBusy, status.code()); +} + +TEST_F(EnableShardingTest, dbExistsWithDifferentCase) { + ShardType shard; + shard.setName("shard0"); + shard.setHost("shard0:12"); + + ASSERT_OK(setupShards(vector<ShardType>{shard})); + + setupDatabase("Db3", shard.getName(), false); + + auto status = + ShardingCatalogManager::get(operationContext())->enableSharding(operationContext(), "db3"); + ASSERT_EQ(ErrorCodes::DatabaseDifferCase, status.code()); + ASSERT_FALSE(status.reason().empty()); +} + +TEST_F(EnableShardingTest, dbExists) { + ShardType shard; + shard.setName("shard0"); + shard.setHost("shard0:12"); + + ASSERT_OK(setupShards(vector<ShardType>{shard})); + + setupDatabase("db4", shard.getName(), false); + + auto status = + ShardingCatalogManager::get(operationContext())->enableSharding(operationContext(), "db4"); + ASSERT_OK(status); +} + +TEST_F(EnableShardingTest, succeedsWhenTheDatabaseIsAlreadySharded) { + ShardType shard; + shard.setName("shard0"); + shard.setHost("shard0:12"); + + ASSERT_OK(setupShards(vector<ShardType>{shard})); + + setupDatabase("db5", shard.getName(), true); + + auto status = + ShardingCatalogManager::get(operationContext())->enableSharding(operationContext(), "db5"); + ASSERT_OK(status); +} + +TEST_F(EnableShardingTest, dbExistsInvalidFormat) { + ShardType shard; + shard.setName("shard0"); + shard.setHost("shard0:12"); + + ASSERT_OK(setupShards(vector<ShardType>{shard})); + + // Set up database with bad type for primary field. + ASSERT_OK(catalogClient()->insertConfigDocument(operationContext(), + DatabaseType::ConfigNS, + BSON("_id" + << "db6" + << "primary" + << 12 + << "partitioned" + << false), + ShardingCatalogClient::kMajorityWriteConcern)); + + auto status = + ShardingCatalogManager::get(operationContext())->enableSharding(operationContext(), "db6"); + ASSERT_EQ(ErrorCodes::TypeMismatch, status.code()); +} + +TEST_F(EnableShardingTest, noDBExistsNoShards) { + auto status = + ShardingCatalogManager::get(operationContext())->enableSharding(operationContext(), "db7"); + ASSERT_EQ(ErrorCodes::ShardNotFound, status.code()); + ASSERT_FALSE(status.reason().empty()); +} + +} // namespace +} // namespace mongo diff --git a/src/mongo/s/catalog/sharding_catalog_manager.h b/src/mongo/s/catalog/sharding_catalog_manager.h index 3fc84f723cc..254a894aaa4 100644 --- a/src/mongo/s/catalog/sharding_catalog_manager.h +++ b/src/mongo/s/catalog/sharding_catalog_manager.h @@ -175,6 +175,23 @@ public: const ShardId& toShard); // + // Database Operations + // + + /** + * Creates a new database or updates the sharding status for an existing one. Cannot be + * used for the admin/config/local DBs, which should not be created or sharded manually + * anyways. + * + * Returns Status::OK on success or any error code indicating the failure. These are some + * of the known failures: + * - DatabaseDifferCase - database already exists, but with a different case + * - ShardNotFound - could not find a shard to place the DB on + */ + Status enableSharding(OperationContext* opCtx, const std::string& dbName); + + + // // Collection Operations // diff --git a/src/mongo/s/catalog/sharding_catalog_manager_database_operations_impl.cpp b/src/mongo/s/catalog/sharding_catalog_manager_database_operations_impl.cpp new file mode 100644 index 00000000000..9580bdd51dc --- /dev/null +++ b/src/mongo/s/catalog/sharding_catalog_manager_database_operations_impl.cpp @@ -0,0 +1,93 @@ +/** + * Copyright (C) 2017 MongoDB Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License, version 3, + * as published by the Free Software Foundation. + * + * 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 + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see <http://www.gnu.org/licenses/>. + * + * 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 GNU Affero General 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_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kSharding + +#include "mongo/s/catalog/sharding_catalog_manager.h" + +#include "mongo/db/namespace_string.h" +#include "mongo/s/catalog/sharding_catalog_client_impl.h" +#include "mongo/s/catalog/type_database.h" +#include "mongo/s/client/shard.h" +#include "mongo/s/grid.h" +#include "mongo/util/log.h" + +namespace mongo { + +Status ShardingCatalogManager::enableSharding(OperationContext* opCtx, const std::string& dbName) { + invariant(nsIsDbOnly(dbName)); + + if (dbName == NamespaceString::kConfigDb || dbName == NamespaceString::kAdminDb) { + return { + ErrorCodes::IllegalOperation, + str::stream() << "Enabling sharding on system configuration databases is not allowed"}; + } + + // Lock the database globally to prevent conflicts with simultaneous database + // creation/modification. + auto scopedDistLock = Grid::get(opCtx)->catalogClient()->getDistLockManager()->lock( + opCtx, dbName, "enableSharding", DistLockManager::kDefaultLockTimeout); + if (!scopedDistLock.isOK()) { + return scopedDistLock.getStatus(); + } + + // Check for case sensitivity violations + DatabaseType db; + Status status = Grid::get(opCtx)->catalogClient()->_checkDbDoesNotExist(opCtx, dbName, &db); + if (status.isOK()) { + // Database does not exist, create a new entry + auto newShardIdStatus = ShardingCatalogClientImpl::_selectShardForNewDatabase( + opCtx, Grid::get(opCtx)->shardRegistry()); + if (!newShardIdStatus.isOK()) { + return newShardIdStatus.getStatus(); + } + + const ShardId& newShardId = newShardIdStatus.getValue(); + + log() << "Placing [" << dbName << "] on: " << newShardId; + + db.setName(dbName); + db.setPrimary(newShardId); + db.setSharded(true); + } else if (status.code() == ErrorCodes::NamespaceExists) { + if (db.getSharded()) { + return Status::OK(); + } + + // Database exists, so just update it + db.setSharded(true); + } else { + return status; + } + + log() << "Enabling sharding for database [" << dbName << "] in config db"; + + return Grid::get(opCtx)->catalogClient()->updateDatabase(opCtx, dbName, db); +} + +} // namespace mongo diff --git a/src/mongo/s/catalog/sharding_catalog_test.cpp b/src/mongo/s/catalog/sharding_catalog_test.cpp index 0e91371dfcd..9e751477f90 100644 --- a/src/mongo/s/catalog/sharding_catalog_test.cpp +++ b/src/mongo/s/catalog/sharding_catalog_test.cpp @@ -1744,278 +1744,6 @@ TEST_F(ShardingCatalogClientTest, createDatabaseDuplicateKeyOnInsert) { future.timed_get(kFutureTimeout); } -TEST_F(ShardingCatalogClientTest, EnableShardingNoDBExists) { - configTargeter()->setFindHostReturnValue(HostAndPort("config:123")); - - vector<ShardType> shards; - ShardType shard; - shard.setName("shard0"); - shard.setHost("shard0:12"); - - setupShards(vector<ShardType>{shard}); - - auto shardTargeter = RemoteCommandTargeterMock::get( - uassertStatusOK(shardRegistry()->getShard(operationContext(), ShardId("shard0"))) - ->getTargeter()); - shardTargeter->setFindHostReturnValue(HostAndPort("shard0:12")); - - distLock()->expectLock( - [](StringData name, StringData whyMessage, Milliseconds) { - ASSERT_EQ("test", name); - ASSERT_FALSE(whyMessage.empty()); - }, - Status::OK()); - - auto future = launchAsync([this] { - auto status = catalogClient()->enableSharding(operationContext(), "test"); - ASSERT_OK(status); - }); - - // Query to find if db already exists in config. - onFindCommand([this](const RemoteCommandRequest& request) { - ASSERT_BSONOBJ_EQ(getReplSecondaryOkMetadata(), - rpc::TrackingMetadata::removeTrackingData(request.metadata)); - const NamespaceString nss(request.dbname, request.cmdObj.firstElement().String()); - ASSERT_EQ(DatabaseType::ConfigNS, nss.toString()); - - auto queryResult = QueryRequest::makeFromFindCommand(nss, request.cmdObj, false); - ASSERT_OK(queryResult.getStatus()); - - const auto& query = queryResult.getValue(); - BSONObj expectedQuery(fromjson(R"({ _id: { $regex: "^test$", $options: "i" }})")); - - ASSERT_EQ(DatabaseType::ConfigNS, query->ns()); - ASSERT_BSONOBJ_EQ(expectedQuery, query->getFilter()); - ASSERT_BSONOBJ_EQ(BSONObj(), query->getSort()); - ASSERT_EQ(1, query->getLimit().get()); - - checkReadConcern(request.cmdObj, Timestamp(0, 0), repl::OpTime::kUninitializedTerm); - - return vector<BSONObj>{}; - }); - - // list databases for checking shard size. - onCommand([](const RemoteCommandRequest& request) { - ASSERT_EQ(HostAndPort("shard0:12"), request.target); - ASSERT_EQ("admin", request.dbname); - ASSERT_BSONOBJ_EQ(BSON("listDatabases" << 1 << "maxTimeMS" << 600000), request.cmdObj); - - ASSERT_BSONOBJ_EQ( - ReadPreferenceSetting(ReadPreference::PrimaryPreferred).toContainingBSON(), - rpc::TrackingMetadata::removeTrackingData(request.metadata)); - - return fromjson(R"({ - databases: [], - totalSize: 1, - ok: 1 - })"); - }); - - onCommand([](const RemoteCommandRequest& request) { - ASSERT_EQ(HostAndPort("config:123"), request.target); - ASSERT_EQ("config", request.dbname); - - ASSERT_BSONOBJ_EQ(BSON(rpc::kReplSetMetadataFieldName << 1), - rpc::TrackingMetadata::removeTrackingData(request.metadata)); - - BSONObj expectedCmd(fromjson(R"({ - update: "databases", - updates: [{ - q: { _id: "test" }, - u: { _id: "test", primary: "shard0", partitioned: true }, - multi: false, - upsert: true - }], - bypassDocumentValidation: false, - ordered: true, - writeConcern: { w: "majority", wtimeout: 15000 }, - maxTimeMS: 30000 - })")); - - ASSERT_BSONOBJ_EQ(expectedCmd, request.cmdObj); - - return fromjson(R"({ - nModified: 0, - n: 1, - upserted: [ - { _id: "test", primary: "shard0", partitioned: true } - ], - ok: 1 - })"); - }); - - future.timed_get(kFutureTimeout); -} - -TEST_F(ShardingCatalogClientTest, EnableShardingLockBusy) { - configTargeter()->setFindHostReturnValue(HostAndPort("config:123")); - - distLock()->expectLock([](StringData, StringData, Milliseconds) {}, - {ErrorCodes::LockBusy, "lock taken"}); - - auto status = catalogClient()->enableSharding(operationContext(), "test"); - ASSERT_EQ(ErrorCodes::LockBusy, status.code()); -} - -TEST_F(ShardingCatalogClientTest, EnableShardingDBExistsWithDifferentCase) { - configTargeter()->setFindHostReturnValue(HostAndPort("config:123")); - - vector<ShardType> shards; - ShardType shard; - shard.setName("shard0"); - shard.setHost("shard0:12"); - - setupShards(vector<ShardType>{shard}); - - distLock()->expectLock([](StringData, StringData, Milliseconds) {}, Status::OK()); - - auto future = launchAsync([this] { - auto status = catalogClient()->enableSharding(operationContext(), "test"); - ASSERT_EQ(ErrorCodes::DatabaseDifferCase, status.code()); - ASSERT_FALSE(status.reason().empty()); - }); - - // Query to find if db already exists in config. - onFindCommand([](const RemoteCommandRequest& request) { - BSONObj existingDoc(fromjson(R"({ _id: "Test", primary: "shard0", partitioned: true })")); - return vector<BSONObj>{existingDoc}; - }); - - future.timed_get(kFutureTimeout); -} - -TEST_F(ShardingCatalogClientTest, EnableShardingDBExists) { - configTargeter()->setFindHostReturnValue(HostAndPort("config:123")); - - vector<ShardType> shards; - ShardType shard; - shard.setName("shard0"); - shard.setHost("shard0:12"); - - setupShards(vector<ShardType>{shard}); - - distLock()->expectLock([](StringData, StringData, Milliseconds) {}, Status::OK()); - - auto future = launchAsync([this] { - auto status = catalogClient()->enableSharding(operationContext(), "test"); - ASSERT_OK(status); - }); - - // Query to find if db already exists in config. - onFindCommand([](const RemoteCommandRequest& request) { - BSONObj existingDoc(fromjson(R"({ _id: "test", primary: "shard2", partitioned: false })")); - return vector<BSONObj>{existingDoc}; - }); - - onCommand([](const RemoteCommandRequest& request) { - ASSERT_EQ(HostAndPort("config:123"), request.target); - ASSERT_EQ("config", request.dbname); - - ASSERT_BSONOBJ_EQ(BSON(rpc::kReplSetMetadataFieldName << 1), - rpc::TrackingMetadata::removeTrackingData(request.metadata)); - - BSONObj expectedCmd(fromjson(R"({ - update: "databases", - updates: [{ - q: { _id: "test" }, - u: { _id: "test", primary: "shard2", partitioned: true }, - multi: false, - upsert: true - }], - bypassDocumentValidation: false, - ordered: true, - writeConcern: { w: "majority", wtimeout: 15000 }, - maxTimeMS: 30000 - })")); - - ASSERT_BSONOBJ_EQ(expectedCmd, request.cmdObj); - - return fromjson(R"({ - nModified: 0, - n: 1, - upserted: [ - { _id: "test", primary: "shard2", partitioned: true } - ], - ok: 1 - })"); - }); - - future.timed_get(kFutureTimeout); -} - -TEST_F(ShardingCatalogClientTest, EnableShardingFailsWhenTheDatabaseIsAlreadySharded) { - configTargeter()->setFindHostReturnValue(HostAndPort("config:123")); - - vector<ShardType> shards; - ShardType shard; - shard.setName("shard0"); - shard.setHost("shard0:12"); - - setupShards(vector<ShardType>{shard}); - - distLock()->expectLock([](StringData, StringData, Milliseconds) {}, Status::OK()); - - auto future = launchAsync([this] { - auto status = catalogClient()->enableSharding(operationContext(), "test"); - ASSERT_EQ(status.code(), ErrorCodes::AlreadyInitialized); - }); - - // Query to find if db already exists in config and it is sharded. - onFindCommand([](const RemoteCommandRequest& request) { - BSONObj existingDoc(fromjson(R"({ _id: "test", primary: "shard2", partitioned: true })")); - return vector<BSONObj>{existingDoc}; - }); - - future.timed_get(kFutureTimeout); -} - -TEST_F(ShardingCatalogClientTest, EnableShardingDBExistsInvalidFormat) { - configTargeter()->setFindHostReturnValue(HostAndPort("config:123")); - - vector<ShardType> shards; - ShardType shard; - shard.setName("shard0"); - shard.setHost("shard0:12"); - - setupShards(vector<ShardType>{shard}); - - distLock()->expectLock([](StringData, StringData, Milliseconds) {}, Status::OK()); - - auto future = launchAsync([this] { - auto status = catalogClient()->enableSharding(operationContext(), "test"); - ASSERT_EQ(ErrorCodes::TypeMismatch, status.code()); - }); - - // Query to find if db already exists in config. - onFindCommand([](const RemoteCommandRequest& request) { - // Bad type for primary field. - BSONObj existingDoc(fromjson(R"({ _id: "test", primary: 12, partitioned: false })")); - return vector<BSONObj>{existingDoc}; - }); - - future.timed_get(kFutureTimeout); -} - -TEST_F(ShardingCatalogClientTest, EnableShardingNoDBExistsNoShards) { - configTargeter()->setFindHostReturnValue(HostAndPort("config:123")); - - distLock()->expectLock([](StringData, StringData, Milliseconds) {}, Status::OK()); - - auto future = launchAsync([this] { - auto status = catalogClient()->enableSharding(operationContext(), "test"); - ASSERT_EQ(ErrorCodes::ShardNotFound, status.code()); - ASSERT_FALSE(status.reason().empty()); - }); - - // Query to find if db already exists in config. - onFindCommand([](const RemoteCommandRequest& request) { return vector<BSONObj>{}; }); - - // Query for config.shards reload. - onFindCommand([](const RemoteCommandRequest& request) { return vector<BSONObj>{}; }); - - future.timed_get(kFutureTimeout); -} - TEST_F(ShardingCatalogClientTest, BasicReadAfterOpTime) { configTargeter()->setFindHostReturnValue(HostAndPort("TestHost1")); diff --git a/src/mongo/s/commands/cluster_enable_sharding_cmd.cpp b/src/mongo/s/commands/cluster_enable_sharding_cmd.cpp index d7fa8e95ed1..d33823630dc 100644 --- a/src/mongo/s/commands/cluster_enable_sharding_cmd.cpp +++ b/src/mongo/s/commands/cluster_enable_sharding_cmd.cpp @@ -30,8 +30,6 @@ #include "mongo/platform/basic.h" -#include <set> - #include "mongo/db/audit.h" #include "mongo/db/auth/action_set.h" #include "mongo/db/auth/action_type.h" @@ -41,8 +39,11 @@ #include "mongo/db/commands.h" #include "mongo/s/catalog/sharding_catalog_client.h" #include "mongo/s/catalog_cache.h" +#include "mongo/s/client/shard_registry.h" +#include "mongo/s/commands/cluster_commands_helpers.h" #include "mongo/s/grid.h" #include "mongo/util/log.h" +#include "mongo/util/scopeguard.h" namespace mongo { namespace { @@ -91,29 +92,30 @@ public: const BSONObj& cmdObj, std::string& errmsg, BSONObjBuilder& result) { - const std::string dbname = parseNs("", cmdObj); - - uassert( - ErrorCodes::InvalidNamespace, - str::stream() << "invalid db name specified: " << dbname, - NamespaceString::validDBName(dbname, NamespaceString::DollarInDbNameBehavior::Allow)); - - if (dbname == NamespaceString::kAdminDb || dbname == NamespaceString::kConfigDb || - dbname == NamespaceString::kLocalDb) { - errmsg = "can't shard " + dbname + " database"; - return false; + const std::string db = parseNs("", cmdObj); + + // Invalidate the routing table cache entry for this database so that we reload the + // collection the next time it's accessed, even if we receive a failure, e.g. NetworkError. + ON_BLOCK_EXIT([opCtx, db] { Grid::get(opCtx)->catalogCache()->purgeDatabase(db); }); + + auto configShard = Grid::get(opCtx)->shardRegistry()->getConfigShard(); + auto cmdResponseStatus = uassertStatusOK(configShard->runCommandWithFixedRetryAttempts( + opCtx, + ReadPreferenceSetting(ReadPreference::PrimaryOnly), + "admin", + Command::appendPassthroughFields(cmdObj, BSON("_configsvrEnableSharding" << db)), + Shard::RetryPolicy::kIdempotent)); + uassertStatusOK(cmdResponseStatus.commandStatus); + + if (!cmdResponseStatus.writeConcernStatus.isOK()) { + appendWriteConcernErrorToCmdResponse( + configShard->getId(), cmdResponseStatus.response["writeConcernError"], result); } - uassertStatusOK(Grid::get(opCtx)->catalogClient()->enableSharding(opCtx, dbname)); - audit::logEnableSharding(Client::getCurrent(), dbname); - - // Make sure to force update of any stale metadata - Grid::get(opCtx)->catalogCache()->purgeDatabase(dbname); - return true; } -} enableShardingCmd; +} clusterEnableShardingCmd; } // namespace } // namespace mongo diff --git a/src/mongo/s/commands/cluster_map_reduce_cmd.cpp b/src/mongo/s/commands/cluster_map_reduce_cmd.cpp index 16f44fa2723..78ce450a3d6 100644 --- a/src/mongo/s/commands/cluster_map_reduce_cmd.cpp +++ b/src/mongo/s/commands/cluster_map_reduce_cmd.cpp @@ -608,11 +608,16 @@ private: static CachedCollectionRoutingInfo createShardedOutputCollection(OperationContext* opCtx, const NamespaceString& nss, const BSONObjSet& splitPts) { - auto const catalogClient = Grid::get(opCtx)->catalogClient(); auto const catalogCache = Grid::get(opCtx)->catalogCache(); // Enable sharding on the output db - Status status = catalogClient->enableSharding(opCtx, nss.db().toString()); + auto status = + Grid::get(opCtx)->shardRegistry()->getConfigShard()->runCommandWithFixedRetryAttempts( + opCtx, + ReadPreferenceSetting(ReadPreference::PrimaryOnly), + "admin", + BSON("_configsvrEnableSharding" << nss.db().toString()), + Shard::RetryPolicy::kIdempotent); // If the database has sharding already enabled, we can ignore the error if (status.isOK()) { |