SERVER-29660 move the logic of the enableSharding into the new _configsvrEnableSharding command

SERVER-30168 Make the _configsvrEnableSharding command idempotent

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()) {