SERVER-30023 Move createDatabase to configserver

SERVER-30022 Move helper functions for createDatabase to mongod

17 files changed, 621 insertions, 636 deletions

diff --git a/jstests/core/views/views_all_commands.js b/jstests/core/views/views_all_commands.js
index a070f9bb99b..ada7a495123 100644
--- a/jstests/core/views/views_all_commands.js
+++ b/jstests/core/views/views_all_commands.js
@@ -66,6 +66,7 @@
         _configsvrCommitChunkMerge: {skip: isAnInternalCommand},
         _configsvrCommitChunkMigration: {skip: isAnInternalCommand},
         _configsvrCommitChunkSplit: {skip: isAnInternalCommand},
+        _configsvrCreateDatabase: {skip: isAnInternalCommand},
         _configsvrEnableSharding: {skip: isAnInternalCommand},
         _configsvrMoveChunk: {skip: isAnInternalCommand},
         _configsvrMovePrimary: {skip: isAnInternalCommand},
diff --git a/src/mongo/db/s/SConscript b/src/mongo/db/s/SConscript
index 83672f29974..d3c7c682c0b 100644
--- a/src/mongo/db/s/SConscript
+++ b/src/mongo/db/s/SConscript
@@ -164,6 +164,7 @@ env.Library(
         'config/configsvr_add_shard_to_zone_command.cpp',
         'config/configsvr_commit_chunk_migration_command.cpp',
         'config/configsvr_control_balancer_command.cpp',
+        'config/configsvr_create_database_command.cpp',
         'config/configsvr_enable_sharding_command.cpp',
         'config/configsvr_merge_chunk_command.cpp',
         'config/configsvr_move_chunk_command.cpp',
diff --git a/src/mongo/db/s/config/configsvr_create_database_command.cpp b/src/mongo/db/s/config/configsvr_create_database_command.cpp
new file mode 100644
index 00000000000..f752ffa6a18
--- /dev/null
+++ b/src/mongo/db/s/config/configsvr_create_database_command.cpp
@@ -0,0 +1,126 @@
+/**
+ *    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 <set>
+
+#include "mongo/db/audit.h"
+#include "mongo/db/auth/action_set.h"
+#include "mongo/db/auth/action_type.h"
+#include "mongo/db/auth/authorization_session.h"
+#include "mongo/db/client.h"
+#include "mongo/db/commands.h"
+#include "mongo/db/operation_context.h"
+#include "mongo/s/catalog/sharding_catalog_manager.h"
+#include "mongo/s/catalog/type_database.h"
+#include "mongo/s/catalog_cache.h"
+#include "mongo/s/grid.h"
+#include "mongo/s/request_types/create_database_gen.h"
+#include "mongo/util/log.h"
+#include "mongo/util/scopeguard.h"
+
+namespace mongo {
+
+using std::shared_ptr;
+using std::set;
+using std::string;
+
+namespace {
+
+/**
+ * Internal sharding command run on config servers to create a database.
+ * Call with { _configsvrCreateDatabase: <string dbName> }
+ */
+class ConfigSvrCreateDatabaseCommand : public BasicCommand {
+public:
+    ConfigSvrCreateDatabaseCommand() : BasicCommand("_configsvrCreateDatabase") {}
+
+    virtual bool slaveOk() const {
+        return false;
+    }
+
+    virtual bool adminOnly() const {
+        return true;
+    }
+
+    virtual bool supportsWriteConcern(const BSONObj& cmd) const override {
+        return true;
+    }
+
+    virtual void help(std::stringstream& help) const override {
+        help << "Internal command, which is exported by the sharding config server. Do not call "
+                "directly. Create a database.";
+    }
+
+    virtual Status checkAuthForCommand(Client* client,
+                                       const std::string& dbname,
+                                       const BSONObj& cmdObj) override {
+        if (!AuthorizationSession::get(client)->isAuthorizedForActionsOnResource(
+                ResourcePattern::forClusterResource(), ActionType::internal)) {
+            return Status(ErrorCodes::Unauthorized, "Unauthorized");
+        }
+
+        return Status::OK();
+    }
+
+    bool run(OperationContext* opCtx,
+             const std::string& dbname_unused,
+             const BSONObj& cmdObj,
+             BSONObjBuilder& result) {
+
+        if (serverGlobalParams.clusterRole != ClusterRole::ConfigServer) {
+            return appendCommandStatus(
+                result,
+                Status(ErrorCodes::IllegalOperation,
+                       "_configsvrCreateDatabase can only be run on config servers"));
+        }
+
+        auto createDatabaseRequest = ConfigsvrCreateDatabase::parse(
+            IDLParserErrorContext("ConfigsvrCreateDatabase"), cmdObj);
+        const string dbname = createDatabaseRequest.get_configsvrCreateDatabase().toString();
+
+        uassert(
+            ErrorCodes::InvalidNamespace,
+            str::stream() << "invalid db name specified: " << dbname,
+            NamespaceString::validDBName(dbname, NamespaceString::DollarInDbNameBehavior::Allow));
+
+        // Make sure to force update of any stale metadata
+        ON_BLOCK_EXIT([opCtx, dbname] { Grid::get(opCtx)->catalogCache()->purgeDatabase(dbname); });
+
+        uassertStatusOK(ShardingCatalogManager::get(opCtx)->createDatabase(opCtx, dbname));
+
+        return true;
+    }
+
+} configsvrCreateDatabaseCmd;
+
+}  // namespace
+}  // namespace mongo
diff --git a/src/mongo/s/SConscript b/src/mongo/s/SConscript
index b850b18ed73..f743a4f9acd 100644
--- a/src/mongo/s/SConscript
+++ b/src/mongo/s/SConscript
@@ -109,6 +109,7 @@ env.Library(
 env.Library(
     target='sharding_request_types',
     source=[
+        env.Idlc('request_types/create_database.idl')[0],
         env.Idlc('request_types/move_primary.idl')[0],
         env.Idlc('request_types/shard_collection.idl')[0],
     ],
diff --git a/src/mongo/s/catalog/SConscript b/src/mongo/s/catalog/SConscript
index 38d3406f9c1..a0d2be936ac 100644
--- a/src/mongo/s/catalog/SConscript
+++ b/src/mongo/s/catalog/SConscript
@@ -176,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_create_database_test.cpp',
         'sharding_catalog_enable_sharding_test.cpp',
         'sharding_catalog_merge_chunks_test.cpp',
         'sharding_catalog_remove_shard_from_zone_test.cpp',
diff --git a/src/mongo/s/catalog/sharding_catalog_client.h b/src/mongo/s/catalog/sharding_catalog_client.h
index 6baceb95da3..2069a90e5d3 100644
--- a/src/mongo/s/catalog/sharding_catalog_client.h
+++ b/src/mongo/s/catalog/sharding_catalog_client.h
@@ -94,11 +94,7 @@ 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.
+    // Allows ShardingCatalogManager to access _exhaustiveFindOnConfig
     friend class ShardingCatalogManager;
 
 public:
@@ -344,20 +340,6 @@ public:
                                          BatchedCommandResponse* response) = 0;
 
     /**
-     * Creates a new database entry for the specified database name in the configuration
-     * metadata and sets the specified shard as primary.
-     *
-     * @param dbName name of the database (case sensitive)
-     *
-     * Returns Status::OK on success or any error code indicating the failure. These are some
-     * of the known failures:
-     *  - NamespaceExists - database already exists
-     *  - DatabaseDifferCase - database already exists, but with a different case
-     *  - ShardNotFound - could not find a shard to place the DB on
-     */
-    virtual Status createDatabase(OperationContext* opCtx, const std::string& dbName) = 0;
-
-    /**
      * Directly inserts a document in the specified namespace on the config server. The document
      * must have an _id index. Must only be used for insertions in the 'config' database.
      *
@@ -422,19 +404,14 @@ 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;
+    virtual StatusWith<repl::OpTimeWith<std::vector<BSONObj>>> _exhaustiveFindOnConfig(
+        OperationContext* opCtx,
+        const ReadPreferenceSetting& readPref,
+        const repl::ReadConcernLevel& readConcern,
+        const NamespaceString& nss,
+        const BSONObj& query,
+        const BSONObj& sort,
+        boost::optional<long long> limit) = 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 6d2fbca60e4..4e013100eb2 100644
--- a/src/mongo/s/catalog/sharding_catalog_client_impl.cpp
+++ b/src/mongo/s/catalog/sharding_catalog_client_impl.cpp
@@ -186,52 +186,6 @@ Status ShardingCatalogClientImpl::updateDatabase(OperationContext* opCtx,
     return Status::OK();
 }
 
-Status ShardingCatalogClientImpl::createDatabase(OperationContext* opCtx,
-                                                 const std::string& dbName) {
-    invariant(nsIsDbOnly(dbName));
-
-    // The admin and config databases should never be explicitly created. They "just exist",
-    // i.e. getDatabase will always return an entry for them.
-    invariant(dbName != "admin");
-    invariant(dbName != "config");
-
-    // Lock the database globally to prevent conflicts with simultaneous database creation.
-    auto scopedDistLock = getDistLockManager()->lock(
-        opCtx, dbName, "createDatabase", DistLockManager::kDefaultLockTimeout);
-    if (!scopedDistLock.isOK()) {
-        return scopedDistLock.getStatus();
-    }
-
-    // check for case sensitivity violations
-    Status status = _checkDbDoesNotExist(opCtx, dbName, nullptr);
-    if (!status.isOK()) {
-        return status;
-    }
-
-    // Database does not exist, pick a shard and 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;
-
-    DatabaseType db;
-    db.setName(dbName);
-    db.setPrimary(newShardId);
-    db.setSharded(false);
-
-    status = insertConfigDocument(
-        opCtx, DatabaseType::ConfigNS, db.toBSON(), ShardingCatalogClient::kMajorityWriteConcern);
-    if (status.code() == ErrorCodes::DuplicateKey) {
-        return Status(ErrorCodes::NamespaceExists, "database " + dbName + " already exists");
-    }
-
-    return status;
-}
-
 Status ShardingCatalogClientImpl::logAction(OperationContext* opCtx,
                                             const std::string& what,
                                             const std::string& ns,
@@ -278,45 +232,6 @@ Status ShardingCatalogClientImpl::logChange(OperationContext* opCtx,
     return _log(opCtx, kChangeLogCollectionName, what, ns, detail, writeConcern);
 }
 
-// static
-StatusWith<ShardId> ShardingCatalogClientImpl::_selectShardForNewDatabase(
-    OperationContext* opCtx, ShardRegistry* shardRegistry) {
-    vector<ShardId> allShardIds;
-
-    shardRegistry->getAllShardIds(&allShardIds);
-    if (allShardIds.empty()) {
-        shardRegistry->reload(opCtx);
-        shardRegistry->getAllShardIds(&allShardIds);
-
-        if (allShardIds.empty()) {
-            return Status(ErrorCodes::ShardNotFound, "No shards found");
-        }
-    }
-
-    ShardId candidateShardId = allShardIds[0];
-
-    auto candidateSizeStatus = shardutil::retrieveTotalShardSize(opCtx, candidateShardId);
-    if (!candidateSizeStatus.isOK()) {
-        return candidateSizeStatus.getStatus();
-    }
-
-    for (size_t i = 1; i < allShardIds.size(); i++) {
-        const ShardId shardId = allShardIds[i];
-
-        const auto sizeStatus = shardutil::retrieveTotalShardSize(opCtx, shardId);
-        if (!sizeStatus.isOK()) {
-            return sizeStatus.getStatus();
-        }
-
-        if (sizeStatus.getValue() < candidateSizeStatus.getValue()) {
-            candidateSizeStatus = sizeStatus;
-            candidateShardId = shardId;
-        }
-    }
-
-    return candidateShardId;
-}
-
 Status ShardingCatalogClientImpl::_log(OperationContext* opCtx,
                                        const StringData& logCollName,
                                        const std::string& what,
@@ -1388,53 +1303,6 @@ Status ShardingCatalogClientImpl::removeConfigDocuments(OperationContext* opCtx,
     return response.toStatus();
 }
 
-Status ShardingCatalogClientImpl::_checkDbDoesNotExist(OperationContext* opCtx,
-                                                       const string& dbName,
-                                                       DatabaseType* db) {
-    BSONObjBuilder queryBuilder;
-    queryBuilder.appendRegex(
-        DatabaseType::name(), (string) "^" + pcrecpp::RE::QuoteMeta(dbName) + "$", "i");
-
-    auto findStatus = _exhaustiveFindOnConfig(opCtx,
-                                              kConfigReadSelector,
-                                              repl::ReadConcernLevel::kMajorityReadConcern,
-                                              NamespaceString(DatabaseType::ConfigNS),
-                                              queryBuilder.obj(),
-                                              BSONObj(),
-                                              1);
-    if (!findStatus.isOK()) {
-        return findStatus.getStatus();
-    }
-
-    const auto& docs = findStatus.getValue().value;
-    if (docs.empty()) {
-        return Status::OK();
-    }
-
-    BSONObj dbObj = docs.front();
-    std::string actualDbName = dbObj[DatabaseType::name()].String();
-    if (actualDbName == dbName) {
-        if (db) {
-            auto parseDBStatus = DatabaseType::fromBSON(dbObj);
-            if (!parseDBStatus.isOK()) {
-                return parseDBStatus.getStatus();
-            }
-
-            *db = parseDBStatus.getValue();
-        }
-
-        return Status(ErrorCodes::NamespaceExists,
-                      str::stream() << "database " << dbName << " already exists");
-    }
-
-    return Status(ErrorCodes::DatabaseDifferCase,
-                  str::stream() << "can't have 2 databases that just differ on case "
-                                << " have: "
-                                << actualDbName
-                                << " want to add: "
-                                << dbName);
-}
-
 Status ShardingCatalogClientImpl::_createCappedConfigCollection(
     OperationContext* opCtx,
     StringData collName,
@@ -1509,7 +1377,7 @@ StatusWith<long long> ShardingCatalogClientImpl::_runCountCommandOnConfig(Operat
 StatusWith<repl::OpTimeWith<vector<BSONObj>>> ShardingCatalogClientImpl::_exhaustiveFindOnConfig(
     OperationContext* opCtx,
     const ReadPreferenceSetting& readPref,
-    repl::ReadConcernLevel readConcern,
+    const repl::ReadConcernLevel& readConcern,
     const NamespaceString& nss,
     const BSONObj& query,
     const BSONObj& sort,
diff --git a/src/mongo/s/catalog/sharding_catalog_client_impl.h b/src/mongo/s/catalog/sharding_catalog_client_impl.h
index 68536fb7df8..e184c078fff 100644
--- a/src/mongo/s/catalog/sharding_catalog_client_impl.h
+++ b/src/mongo/s/catalog/sharding_catalog_client_impl.h
@@ -49,13 +49,6 @@ class TaskExecutor;
  */
 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
@@ -82,8 +75,6 @@ public:
                           const std::string& dbName,
                           const DatabaseType& db) override;
 
-    Status createDatabase(OperationContext* opCtx, const std::string& dbName) override;
-
     Status logAction(OperationContext* opCtx,
                      const std::string& what,
                      const std::string& ns,
@@ -196,17 +187,6 @@ 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.
-     */
-    static StatusWith<ShardId> _selectShardForNewDatabase(OperationContext* opCtx,
-                                                          ShardRegistry* shardRegistry);
-
     /**
      * Updates a single document in the specified namespace on the config server. The document must
      * have an _id index. Must only be used for updates to the 'config' database.
@@ -245,11 +225,11 @@ private:
     StatusWith<repl::OpTimeWith<std::vector<BSONObj>>> _exhaustiveFindOnConfig(
         OperationContext* opCtx,
         const ReadPreferenceSetting& readPref,
-        repl::ReadConcernLevel readConcern,
+        const repl::ReadConcernLevel& readConcern,
         const NamespaceString& nss,
         const BSONObj& query,
         const BSONObj& sort,
-        boost::optional<long long> limit);
+        boost::optional<long long> limit) override;
 
     /**
      * Appends a read committed read concern to the request object.
diff --git a/src/mongo/s/catalog/sharding_catalog_client_mock.cpp b/src/mongo/s/catalog/sharding_catalog_client_mock.cpp
index 90e7c4dd8a7..1df6313afec 100644
--- a/src/mongo/s/catalog/sharding_catalog_client_mock.cpp
+++ b/src/mongo/s/catalog/sharding_catalog_client_mock.cpp
@@ -226,9 +226,14 @@ StatusWith<std::vector<KeysCollectionDocument>> ShardingCatalogClientMock::getNe
     return {ErrorCodes::InternalError, "Method not implemented"};
 }
 
-Status ShardingCatalogClientMock::_checkDbDoesNotExist(OperationContext* opCtx,
-                                                       const string& dbName,
-                                                       DatabaseType* db) {
+StatusWith<repl::OpTimeWith<std::vector<BSONObj>>>
+ShardingCatalogClientMock::_exhaustiveFindOnConfig(OperationContext* opCtx,
+                                                   const ReadPreferenceSetting& readPref,
+                                                   const repl::ReadConcernLevel& readConcern,
+                                                   const NamespaceString& nss,
+                                                   const BSONObj& query,
+                                                   const BSONObj& sort,
+                                                   boost::optional<long long> limit) {
     return {ErrorCodes::InternalError, "Method not implemented"};
 }
 
diff --git a/src/mongo/s/catalog/sharding_catalog_client_mock.h b/src/mongo/s/catalog/sharding_catalog_client_mock.h
index 574e3ad30d3..6189ad67660 100644
--- a/src/mongo/s/catalog/sharding_catalog_client_mock.h
+++ b/src/mongo/s/catalog/sharding_catalog_client_mock.h
@@ -159,9 +159,14 @@ public:
 private:
     std::unique_ptr<DistLockManager> _distLockManager;
 
-    Status _checkDbDoesNotExist(OperationContext* opCtx,
-                                const std::string& dbName,
-                                DatabaseType* db) override;
+    StatusWith<repl::OpTimeWith<std::vector<BSONObj>>> _exhaustiveFindOnConfig(
+        OperationContext* opCtx,
+        const ReadPreferenceSetting& readPref,
+        const repl::ReadConcernLevel& readConcern,
+        const NamespaceString& nss,
+        const BSONObj& query,
+        const BSONObj& sort,
+        boost::optional<long long> limit) override;
 };
 
 }  // namespace mongo
diff --git a/src/mongo/s/catalog/sharding_catalog_create_database_test.cpp b/src/mongo/s/catalog/sharding_catalog_create_database_test.cpp
new file mode 100644
index 00000000000..4b0515ef833
--- /dev/null
+++ b/src/mongo/s/catalog/sharding_catalog_create_database_test.cpp
@@ -0,0 +1,219 @@
+/**
+ *    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 CreateDatabaseTest : public ConfigServerTestFixture {};
+
+TEST_F(CreateDatabaseTest, createDatabaseSuccess) {
+    const string dbname = "db1";
+
+    ShardType s0;
+    s0.setName("shard0000");
+    s0.setHost("ShardHost0:27017");
+    ASSERT_OK(setupShards(vector<ShardType>{s0}));
+
+    ShardType s1;
+    s1.setName("shard0001");
+    s1.setHost("ShardHost1:27017");
+    ASSERT_OK(setupShards(vector<ShardType>{s1}));
+
+    ShardType s2;
+    s2.setName("shard0002");
+    s2.setHost("ShardHost2:27017");
+    ASSERT_OK(setupShards(vector<ShardType>{s2}));
+
+    // Prime the shard registry with information about the existing shards
+    shardRegistry()->reload(operationContext());
+
+    // Set up all the target mocks return values.
+    RemoteCommandTargeterMock::get(
+        uassertStatusOK(shardRegistry()->getShard(operationContext(), s0.getName()))->getTargeter())
+        ->setFindHostReturnValue(HostAndPort(s0.getHost()));
+    RemoteCommandTargeterMock::get(
+        uassertStatusOK(shardRegistry()->getShard(operationContext(), s1.getName()))->getTargeter())
+        ->setFindHostReturnValue(HostAndPort(s1.getHost()));
+    RemoteCommandTargeterMock::get(
+        uassertStatusOK(shardRegistry()->getShard(operationContext(), s2.getName()))->getTargeter())
+        ->setFindHostReturnValue(HostAndPort(s2.getHost()));
+
+    // Now actually start the createDatabase work.
+
+    auto future = launchAsync([this, dbname] {
+        ON_BLOCK_EXIT([&] { Client::destroy(); });
+        Client::initThreadIfNotAlready("Test");
+        auto opCtx = cc().makeOperationContext();
+        Status status =
+            ShardingCatalogManager::get(opCtx.get())->createDatabase(opCtx.get(), dbname);
+        ASSERT_OK(status);
+    });
+
+    // Return size information about first shard
+    onCommand([&](const RemoteCommandRequest& request) {
+        ASSERT_EQUALS(s0.getHost(), request.target.toString());
+        ASSERT_EQUALS("admin", request.dbname);
+        string cmdName = request.cmdObj.firstElement().fieldName();
+        ASSERT_EQUALS("listDatabases", cmdName);
+        ASSERT_FALSE(request.cmdObj.hasField(repl::ReadConcernArgs::kReadConcernFieldName));
+
+        ASSERT_BSONOBJ_EQ(
+            ReadPreferenceSetting(ReadPreference::PrimaryPreferred).toContainingBSON(),
+            rpc::TrackingMetadata::removeTrackingData(request.metadata));
+
+        return BSON("ok" << 1 << "totalSize" << 10);
+    });
+
+    // Return size information about second shard
+    onCommand([&](const RemoteCommandRequest& request) {
+        ASSERT_EQUALS(s1.getHost(), request.target.toString());
+        ASSERT_EQUALS("admin", request.dbname);
+        string cmdName = request.cmdObj.firstElement().fieldName();
+        ASSERT_EQUALS("listDatabases", cmdName);
+        ASSERT_FALSE(request.cmdObj.hasField(repl::ReadConcernArgs::kReadConcernFieldName));
+
+        ASSERT_BSONOBJ_EQ(
+            ReadPreferenceSetting(ReadPreference::PrimaryPreferred).toContainingBSON(),
+            rpc::TrackingMetadata::removeTrackingData(request.metadata));
+
+        return BSON("ok" << 1 << "totalSize" << 1);
+    });
+
+    // Return size information about third shard
+    onCommand([&](const RemoteCommandRequest& request) {
+        ASSERT_EQUALS(s2.getHost(), request.target.toString());
+        ASSERT_EQUALS("admin", request.dbname);
+        string cmdName = request.cmdObj.firstElement().fieldName();
+        ASSERT_EQUALS("listDatabases", cmdName);
+
+        ASSERT_BSONOBJ_EQ(
+            ReadPreferenceSetting(ReadPreference::PrimaryPreferred).toContainingBSON(),
+            rpc::TrackingMetadata::removeTrackingData(request.metadata));
+
+        return BSON("ok" << 1 << "totalSize" << 100);
+    });
+
+    future.timed_get(kFutureTimeout);
+}
+
+TEST_F(CreateDatabaseTest, createDatabaseDistLockHeld) {
+    const string dbname = "db2";
+
+    ASSERT_OK(distLockCatalog()
+                  ->grabLock(operationContext(),
+                             dbname,
+                             OID::gen(),
+                             "dummyWho",
+                             "dummyProcessId",
+                             Date_t::now(),
+                             "dummyReason")
+                  .getStatus());
+
+    Status status =
+        ShardingCatalogManager::get(operationContext())->createDatabase(operationContext(), dbname);
+    ASSERT_EQUALS(ErrorCodes::LockBusy, status);
+}
+
+TEST_F(CreateDatabaseTest, createDatabaseDBExists) {
+    const string dbname = "db3";
+
+    ShardType shard;
+    shard.setName("shard0");
+    shard.setHost("shard0:12");
+
+    ASSERT_OK(setupShards(vector<ShardType>{shard}));
+
+    setupDatabase(dbname, shard.getName(), false);
+
+    Status status =
+        ShardingCatalogManager::get(operationContext())->createDatabase(operationContext(), dbname);
+    ASSERT_OK(status);
+}
+
+TEST_F(CreateDatabaseTest, createDatabaseDBExistsDifferentCase) {
+    const string dbname = "db4";
+    const string dbnameDiffCase = "Db4";
+
+    ShardType shard;
+    shard.setName("shard0");
+    shard.setHost("shard0:12");
+
+    ASSERT_OK(setupShards(vector<ShardType>{shard}));
+
+    setupDatabase(dbnameDiffCase, shard.getName(), false);
+
+    Status status =
+        ShardingCatalogManager::get(operationContext())->createDatabase(operationContext(), dbname);
+    ASSERT_EQUALS(ErrorCodes::DatabaseDifferCase, status);
+}
+
+TEST_F(CreateDatabaseTest, createDatabaseNoShards) {
+    const string dbname = "db5";
+
+    Status status =
+        ShardingCatalogManager::get(operationContext())->createDatabase(operationContext(), dbname);
+    ASSERT_EQUALS(ErrorCodes::ShardNotFound, status);
+}
+
+}  // namespace
+}  // namespace mongo
diff --git a/src/mongo/s/catalog/sharding_catalog_manager.h b/src/mongo/s/catalog/sharding_catalog_manager.h
index 8d54f47c7d0..b323469f178 100644
--- a/src/mongo/s/catalog/sharding_catalog_manager.h
+++ b/src/mongo/s/catalog/sharding_catalog_manager.h
@@ -33,8 +33,10 @@
 #include "mongo/db/concurrency/d_concurrency.h"
 #include "mongo/executor/task_executor.h"
 #include "mongo/s/catalog/type_chunk.h"
+#include "mongo/s/catalog/type_database.h"
 #include "mongo/s/catalog/type_shard.h"
 #include "mongo/s/client/shard.h"
+#include "mongo/s/client/shard_registry.h"
 #include "mongo/s/shard_key_pattern.h"
 #include "mongo/stdx/mutex.h"
 
@@ -180,6 +182,20 @@ public:
     //
 
     /**
+     * Creates a new database entry for the specified database name in the configuration
+     * metadata and sets the specified shard as primary.
+     *
+     * @param dbName name of the database (case sensitive)
+     *
+     * Returns Status::OK on success or any error code indicating the failure. These are some
+     * of the known failures:
+     *  - NamespaceExists - database already exists
+     *  - DatabaseDifferCase - database already exists, but with a different case
+     *  - ShardNotFound - could not find a shard to place the DB on
+     */
+    Status createDatabase(OperationContext* opCtx, const std::string& dbName);
+
+    /**
      * 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.
@@ -342,6 +358,27 @@ private:
                                                               const std::string& dbName,
                                                               const BSONObj& cmdObj);
 
+    /**
+     * 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);
+
+    /**
+     * 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.
+     */
+    static StatusWith<ShardId> _selectShardForNewDatabase(OperationContext* opCtx,
+                                                          ShardRegistry* shardRegistry);
+
     // The owning service context
     ServiceContext* const _serviceContext;
 
diff --git a/src/mongo/s/catalog/sharding_catalog_manager_database_operations.cpp b/src/mongo/s/catalog/sharding_catalog_manager_database_operations.cpp
index 9580bdd51dc..26ebb518f4b 100644
--- a/src/mongo/s/catalog/sharding_catalog_manager_database_operations.cpp
+++ b/src/mongo/s/catalog/sharding_catalog_manager_database_operations.cpp
@@ -30,6 +30,8 @@
 
 #include "mongo/s/catalog/sharding_catalog_manager.h"
 
+#include <pcrecpp.h>
+
 #include "mongo/db/namespace_string.h"
 #include "mongo/s/catalog/sharding_catalog_client_impl.h"
 #include "mongo/s/catalog/type_database.h"
@@ -39,6 +41,61 @@
 
 namespace mongo {
 
+using std::string;
+using std::vector;
+
+const ReadPreferenceSetting kConfigReadSelector(ReadPreference::Nearest, TagSet{});
+
+Status ShardingCatalogManager::createDatabase(OperationContext* opCtx, const std::string& dbName) {
+    invariant(nsIsDbOnly(dbName));
+
+    // The admin and config databases should never be explicitly created. They "just exist",
+    // i.e. getDatabase will always return an entry for them.
+    if (dbName == "admin" || dbName == "config") {
+        uasserted(ErrorCodes::InvalidOptions,
+                  str::stream() << "cannot manually create database '" << dbName << "'");
+    }
+
+    // Lock the database globally to prevent conflicts with simultaneous database creation.
+    auto scopedDistLock = Grid::get(opCtx)->catalogClient()->getDistLockManager()->lock(
+        opCtx, dbName, "createDatabase", DistLockManager::kDefaultLockTimeout);
+    if (!scopedDistLock.isOK()) {
+        return scopedDistLock.getStatus();
+    }
+
+    // check for case sensitivity violations
+    Status status = _checkDbDoesNotExist(opCtx, dbName, nullptr);
+    if (!status.isOK()) {
+        if (status.code() == ErrorCodes::NamespaceExists) {
+            return Status::OK();
+        }
+        return status;
+    }
+
+    // Database does not exist, pick a shard and 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;
+
+    DatabaseType db;
+    db.setName(dbName);
+    db.setPrimary(newShardId);
+    db.setSharded(false);
+
+    status = Grid::get(opCtx)->catalogClient()->insertConfigDocument(
+        opCtx, DatabaseType::ConfigNS, db.toBSON(), ShardingCatalogClient::kMajorityWriteConcern);
+    if (status.code() == ErrorCodes::DuplicateKey) {
+        return Status(ErrorCodes::NamespaceExists, "database " + dbName + " already exists");
+    }
+
+    return status;
+}
+
 Status ShardingCatalogManager::enableSharding(OperationContext* opCtx, const std::string& dbName) {
     invariant(nsIsDbOnly(dbName));
 
@@ -58,11 +115,11 @@ Status ShardingCatalogManager::enableSharding(OperationContext* opCtx, const std
 
     // Check for case sensitivity violations
     DatabaseType db;
-    Status status = Grid::get(opCtx)->catalogClient()->_checkDbDoesNotExist(opCtx, dbName, &db);
+    Status status = _checkDbDoesNotExist(opCtx, dbName, &db);
     if (status.isOK()) {
         // Database does not exist, create a new entry
-        auto newShardIdStatus = ShardingCatalogClientImpl::_selectShardForNewDatabase(
-            opCtx, Grid::get(opCtx)->shardRegistry());
+        auto newShardIdStatus =
+            _selectShardForNewDatabase(opCtx, Grid::get(opCtx)->shardRegistry());
         if (!newShardIdStatus.isOK()) {
             return newShardIdStatus.getStatus();
         }
@@ -90,4 +147,52 @@ Status ShardingCatalogManager::enableSharding(OperationContext* opCtx, const std
     return Grid::get(opCtx)->catalogClient()->updateDatabase(opCtx, dbName, db);
 }
 
+Status ShardingCatalogManager::_checkDbDoesNotExist(OperationContext* opCtx,
+                                                    const std::string& dbName,
+                                                    DatabaseType* db) {
+    BSONObjBuilder queryBuilder;
+    queryBuilder.appendRegex(
+        DatabaseType::name(), (string) "^" + pcrecpp::RE::QuoteMeta(dbName) + "$", "i");
+
+    auto findStatus = Grid::get(opCtx)->catalogClient()->_exhaustiveFindOnConfig(
+        opCtx,
+        kConfigReadSelector,
+        repl::ReadConcernLevel::kMajorityReadConcern,
+        NamespaceString(DatabaseType::ConfigNS),
+        queryBuilder.obj(),
+        BSONObj(),
+        1);
+    if (!findStatus.isOK()) {
+        return findStatus.getStatus();
+    }
+
+    const auto& docs = findStatus.getValue().value;
+    if (docs.empty()) {
+        return Status::OK();
+    }
+
+    BSONObj dbObj = docs.front();
+    std::string actualDbName = dbObj[DatabaseType::name()].String();
+    if (actualDbName == dbName) {
+        if (db) {
+            auto parseDBStatus = DatabaseType::fromBSON(dbObj);
+            if (!parseDBStatus.isOK()) {
+                return parseDBStatus.getStatus();
+            }
+
+            *db = parseDBStatus.getValue();
+        }
+
+        return Status(ErrorCodes::NamespaceExists,
+                      str::stream() << "database " << dbName << " already exists");
+    }
+
+    return Status(ErrorCodes::DatabaseDifferCase,
+                  str::stream() << "can't have 2 databases that just differ on case "
+                                << " have: "
+                                << actualDbName
+                                << " want to add: "
+                                << dbName);
+}
+
 }  // namespace mongo
diff --git a/src/mongo/s/catalog/sharding_catalog_manager_shard_operations.cpp b/src/mongo/s/catalog/sharding_catalog_manager_shard_operations.cpp
index 80ea3fe6de3..42e99784c6c 100644
--- a/src/mongo/s/catalog/sharding_catalog_manager_shard_operations.cpp
+++ b/src/mongo/s/catalog/sharding_catalog_manager_shard_operations.cpp
@@ -33,6 +33,7 @@
 #include "mongo/s/catalog/sharding_catalog_manager.h"
 
 #include <iomanip>
+#include <pcrecpp.h>
 #include <set>
 
 #include "mongo/base/status_with.h"
@@ -61,6 +62,7 @@
 #include "mongo/s/client/shard_registry.h"
 #include "mongo/s/cluster_identity_loader.h"
 #include "mongo/s/grid.h"
+#include "mongo/s/shard_util.h"
 #include "mongo/s/write_ops/batched_command_request.h"
 #include "mongo/s/write_ops/batched_command_response.h"
 #include "mongo/util/fail_point_service.h"
@@ -71,6 +73,8 @@
 namespace mongo {
 namespace {
 
+using std::vector;
+
 using CallbackHandle = executor::TaskExecutor::CallbackHandle;
 using CallbackArgs = executor::TaskExecutor::CallbackArgs;
 using RemoteCommandCallbackArgs = executor::TaskExecutor::RemoteCommandCallbackArgs;
@@ -732,4 +736,43 @@ BSONObj ShardingCatalogManager::createShardIdentityUpsertForAddShard(OperationCo
     return request.toBSON();
 }
 
+// static
+StatusWith<ShardId> ShardingCatalogManager::_selectShardForNewDatabase(
+    OperationContext* opCtx, ShardRegistry* shardRegistry) {
+    vector<ShardId> allShardIds;
+
+    shardRegistry->getAllShardIds(&allShardIds);
+    if (allShardIds.empty()) {
+        shardRegistry->reload(opCtx);
+        shardRegistry->getAllShardIds(&allShardIds);
+
+        if (allShardIds.empty()) {
+            return Status(ErrorCodes::ShardNotFound, "No shards found");
+        }
+    }
+
+    ShardId candidateShardId = allShardIds[0];
+
+    auto candidateSizeStatus = shardutil::retrieveTotalShardSize(opCtx, candidateShardId);
+    if (!candidateSizeStatus.isOK()) {
+        return candidateSizeStatus.getStatus();
+    }
+
+    for (size_t i = 1; i < allShardIds.size(); i++) {
+        const ShardId shardId = allShardIds[i];
+
+        const auto sizeStatus = shardutil::retrieveTotalShardSize(opCtx, shardId);
+        if (!sizeStatus.isOK()) {
+            return sizeStatus.getStatus();
+        }
+
+        if (sizeStatus.getValue() < candidateSizeStatus.getValue()) {
+            candidateSizeStatus = sizeStatus;
+            candidateShardId = shardId;
+        }
+    }
+
+    return candidateShardId;
+}
+
 }  // namespace mongo
diff --git a/src/mongo/s/catalog/sharding_catalog_test.cpp b/src/mongo/s/catalog/sharding_catalog_test.cpp
index 9e751477f90..425e08473b2 100644
--- a/src/mongo/s/catalog/sharding_catalog_test.cpp
+++ b/src/mongo/s/catalog/sharding_catalog_test.cpp
@@ -1305,445 +1305,6 @@ TEST_F(ShardingCatalogClientTest, ApplyChunkOpsDeprecatedFailedWithCheck) {
     future.timed_get(kFutureTimeout);
 }
 
-TEST_F(ShardingCatalogClientTest, createDatabaseSuccess) {
-    const string dbname = "databaseToCreate";
-    const HostAndPort configHost("TestHost1");
-    configTargeter()->setFindHostReturnValue(configHost);
-
-    ShardType s0;
-    s0.setName("shard0000");
-    s0.setHost("ShardHost0:27017");
-
-    ShardType s1;
-    s1.setName("shard0001");
-    s1.setHost("ShardHost1:27017");
-
-    ShardType s2;
-    s2.setName("shard0002");
-    s2.setHost("ShardHost2:27017");
-
-    // Prime the shard registry with information about the existing shards
-    auto future = launchAsync([this] { shardRegistry()->reload(operationContext()); });
-
-    onFindCommand([&](const RemoteCommandRequest& request) {
-        ASSERT_EQUALS(configHost, request.target);
-        ASSERT_BSONOBJ_EQ(getReplSecondaryOkMetadata(),
-                          rpc::TrackingMetadata::removeTrackingData(request.metadata));
-
-        const NamespaceString nss(request.dbname, request.cmdObj.firstElement().String());
-        auto query = assertGet(QueryRequest::makeFromFindCommand(nss, request.cmdObj, false));
-
-        ASSERT_EQ(ShardType::ConfigNS, query->ns());
-        ASSERT_BSONOBJ_EQ(BSONObj(), query->getFilter());
-        ASSERT_BSONOBJ_EQ(BSONObj(), query->getSort());
-        ASSERT_FALSE(query->getLimit().is_initialized());
-
-        checkReadConcern(request.cmdObj, Timestamp(0, 0), repl::OpTime::kUninitializedTerm);
-
-        return vector<BSONObj>{s0.toBSON(), s1.toBSON(), s2.toBSON()};
-    });
-
-    future.timed_get(kFutureTimeout);
-
-    // Set up all the target mocks return values.
-    RemoteCommandTargeterMock::get(
-        uassertStatusOK(shardRegistry()->getShard(operationContext(), s0.getName()))->getTargeter())
-        ->setFindHostReturnValue(HostAndPort(s0.getHost()));
-    RemoteCommandTargeterMock::get(
-        uassertStatusOK(shardRegistry()->getShard(operationContext(), s1.getName()))->getTargeter())
-        ->setFindHostReturnValue(HostAndPort(s1.getHost()));
-    RemoteCommandTargeterMock::get(
-        uassertStatusOK(shardRegistry()->getShard(operationContext(), s2.getName()))->getTargeter())
-        ->setFindHostReturnValue(HostAndPort(s2.getHost()));
-
-    // Now actually start the createDatabase work.
-
-    distLock()->expectLock(
-        [dbname](StringData name, StringData whyMessage, Milliseconds waitFor) {}, Status::OK());
-
-
-    future = launchAsync([this, dbname] {
-        Status status = catalogClient()->createDatabase(operationContext(), dbname);
-        ASSERT_OK(status);
-    });
-
-    // Report no databases with the same name already exist
-    onFindCommand([&](const RemoteCommandRequest& request) {
-        ASSERT_EQUALS(configHost, request.target);
-        const NamespaceString nss(request.dbname, request.cmdObj.firstElement().String());
-        ASSERT_BSONOBJ_EQ(getReplSecondaryOkMetadata(),
-                          rpc::TrackingMetadata::removeTrackingData(request.metadata));
-        ASSERT_EQ(DatabaseType::ConfigNS, nss.ns());
-        checkReadConcern(request.cmdObj, Timestamp(0, 0), repl::OpTime::kUninitializedTerm);
-        return vector<BSONObj>{};
-    });
-
-    // Return size information about first shard
-    onCommand([&](const RemoteCommandRequest& request) {
-        ASSERT_EQUALS(s0.getHost(), request.target.toString());
-        ASSERT_EQUALS("admin", request.dbname);
-        string cmdName = request.cmdObj.firstElement().fieldName();
-        ASSERT_EQUALS("listDatabases", cmdName);
-        ASSERT_FALSE(request.cmdObj.hasField(repl::ReadConcernArgs::kReadConcernFieldName));
-
-        ASSERT_BSONOBJ_EQ(
-            ReadPreferenceSetting(ReadPreference::PrimaryPreferred).toContainingBSON(),
-            rpc::TrackingMetadata::removeTrackingData(request.metadata));
-
-        return BSON("ok" << 1 << "totalSize" << 10);
-    });
-
-    // Return size information about second shard
-    onCommand([&](const RemoteCommandRequest& request) {
-        ASSERT_EQUALS(s1.getHost(), request.target.toString());
-        ASSERT_EQUALS("admin", request.dbname);
-        string cmdName = request.cmdObj.firstElement().fieldName();
-        ASSERT_EQUALS("listDatabases", cmdName);
-        ASSERT_FALSE(request.cmdObj.hasField(repl::ReadConcernArgs::kReadConcernFieldName));
-
-        ASSERT_BSONOBJ_EQ(
-            ReadPreferenceSetting(ReadPreference::PrimaryPreferred).toContainingBSON(),
-            rpc::TrackingMetadata::removeTrackingData(request.metadata));
-
-        return BSON("ok" << 1 << "totalSize" << 1);
-    });
-
-    // Return size information about third shard
-    onCommand([&](const RemoteCommandRequest& request) {
-        ASSERT_EQUALS(s2.getHost(), request.target.toString());
-        ASSERT_EQUALS("admin", request.dbname);
-        string cmdName = request.cmdObj.firstElement().fieldName();
-        ASSERT_EQUALS("listDatabases", cmdName);
-
-        ASSERT_BSONOBJ_EQ(
-            ReadPreferenceSetting(ReadPreference::PrimaryPreferred).toContainingBSON(),
-            rpc::TrackingMetadata::removeTrackingData(request.metadata));
-
-        return BSON("ok" << 1 << "totalSize" << 100);
-    });
-
-    // Process insert to config.databases collection
-    onCommand([&](const RemoteCommandRequest& request) {
-        ASSERT_EQUALS(configHost, request.target);
-        ASSERT_EQUALS("config", request.dbname);
-
-        ASSERT_BSONOBJ_EQ(BSON(rpc::kReplSetMetadataFieldName << 1),
-                          rpc::TrackingMetadata::removeTrackingData(request.metadata));
-
-        const auto opMsgRequest = OpMsgRequest::fromDBAndBody(request.dbname, request.cmdObj);
-        const auto insertOp = InsertOp::parse(opMsgRequest);
-        ASSERT_EQUALS(DatabaseType::ConfigNS, insertOp.getNamespace().ns());
-
-        const auto& inserts = insertOp.getDocuments();
-        ASSERT_EQUALS(1U, inserts.size());
-
-        const auto& insert = inserts.front();
-        DatabaseType expectedDb;
-        expectedDb.setName(dbname);
-        expectedDb.setPrimary(
-            ShardId(s1.getName()));  // This is the one we reported with the smallest size
-        expectedDb.setSharded(false);
-
-        ASSERT_BSONOBJ_EQ(expectedDb.toBSON(), insert);
-
-        BatchedCommandResponse response;
-        response.setOk(true);
-        response.setNModified(1);
-
-        return response.toBSON();
-    });
-
-    future.timed_get(kFutureTimeout);
-}
-
-TEST_F(ShardingCatalogClientTest, createDatabaseDistLockHeld) {
-    const string dbname = "databaseToCreate";
-
-
-    configTargeter()->setFindHostReturnValue(HostAndPort("TestHost1"));
-
-    distLock()->expectLock(
-        [dbname](StringData name, StringData whyMessage, Milliseconds waitFor) {
-            ASSERT_EQUALS(dbname, name);
-            ASSERT_EQUALS("createDatabase", whyMessage);
-        },
-        Status(ErrorCodes::LockBusy, "lock already held"));
-
-    Status status = catalogClient()->createDatabase(operationContext(), dbname);
-    ASSERT_EQUALS(ErrorCodes::LockBusy, status);
-}
-
-TEST_F(ShardingCatalogClientTest, createDatabaseDBExists) {
-    const string dbname = "databaseToCreate";
-
-
-    configTargeter()->setFindHostReturnValue(HostAndPort("TestHost1"));
-
-    distLock()->expectLock(
-        [dbname](StringData name, StringData whyMessage, Milliseconds waitFor) {}, Status::OK());
-
-
-    auto future = launchAsync([this, dbname] {
-        Status status = catalogClient()->createDatabase(operationContext(), dbname);
-        ASSERT_EQUALS(ErrorCodes::NamespaceExists, status);
-    });
-
-    onFindCommand([this, dbname](const RemoteCommandRequest& request) {
-        ASSERT_BSONOBJ_EQ(getReplSecondaryOkMetadata(),
-                          rpc::TrackingMetadata::removeTrackingData(request.metadata));
-
-        const NamespaceString nss(request.dbname, request.cmdObj.firstElement().String());
-        auto query = assertGet(QueryRequest::makeFromFindCommand(nss, request.cmdObj, false));
-
-        BSONObjBuilder queryBuilder;
-        queryBuilder.appendRegex(
-            DatabaseType::name(), (string) "^" + pcrecpp::RE::QuoteMeta(dbname) + "$", "i");
-
-        ASSERT_EQ(DatabaseType::ConfigNS, query->ns());
-        ASSERT_BSONOBJ_EQ(queryBuilder.obj(), query->getFilter());
-        checkReadConcern(request.cmdObj, Timestamp(0, 0), repl::OpTime::kUninitializedTerm);
-
-        return vector<BSONObj>{BSON("_id" << dbname)};
-    });
-
-    future.timed_get(kFutureTimeout);
-}
-
-TEST_F(ShardingCatalogClientTest, createDatabaseDBExistsDifferentCase) {
-    const string dbname = "databaseToCreate";
-    const string dbnameDiffCase = "databasetocreate";
-
-
-    configTargeter()->setFindHostReturnValue(HostAndPort("TestHost1"));
-
-    distLock()->expectLock(
-        [dbname](StringData name, StringData whyMessage, Milliseconds waitFor) {}, Status::OK());
-
-
-    auto future = launchAsync([this, dbname] {
-        Status status = catalogClient()->createDatabase(operationContext(), dbname);
-        ASSERT_EQUALS(ErrorCodes::DatabaseDifferCase, status);
-    });
-
-    onFindCommand([this, dbname, dbnameDiffCase](const RemoteCommandRequest& request) {
-        ASSERT_BSONOBJ_EQ(getReplSecondaryOkMetadata(),
-                          rpc::TrackingMetadata::removeTrackingData(request.metadata));
-
-        const NamespaceString nss(request.dbname, request.cmdObj.firstElement().String());
-        auto query = assertGet(QueryRequest::makeFromFindCommand(nss, request.cmdObj, false));
-
-        BSONObjBuilder queryBuilder;
-        queryBuilder.appendRegex(
-            DatabaseType::name(), (string) "^" + pcrecpp::RE::QuoteMeta(dbname) + "$", "i");
-
-        ASSERT_EQ(DatabaseType::ConfigNS, query->ns());
-        ASSERT_BSONOBJ_EQ(queryBuilder.obj(), query->getFilter());
-        checkReadConcern(request.cmdObj, Timestamp(0, 0), repl::OpTime::kUninitializedTerm);
-
-        return vector<BSONObj>{BSON("_id" << dbnameDiffCase)};
-    });
-
-    future.timed_get(kFutureTimeout);
-}
-
-TEST_F(ShardingCatalogClientTest, createDatabaseNoShards) {
-    const string dbname = "databaseToCreate";
-
-
-    configTargeter()->setFindHostReturnValue(HostAndPort("TestHost1"));
-
-    distLock()->expectLock(
-        [dbname](StringData name, StringData whyMessage, Milliseconds waitFor) {}, Status::OK());
-
-
-    auto future = launchAsync([this, dbname] {
-        Status status = catalogClient()->createDatabase(operationContext(), dbname);
-        ASSERT_EQUALS(ErrorCodes::ShardNotFound, status);
-    });
-
-    // Report no databases with the same name already exist
-    onFindCommand([this, dbname](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.ns());
-        checkReadConcern(request.cmdObj, Timestamp(0, 0), repl::OpTime::kUninitializedTerm);
-        return vector<BSONObj>{};
-    });
-
-    // Report no shards exist
-    onFindCommand([this](const RemoteCommandRequest& request) {
-        ASSERT_BSONOBJ_EQ(getReplSecondaryOkMetadata(),
-                          rpc::TrackingMetadata::removeTrackingData(request.metadata));
-        const NamespaceString nss(request.dbname, request.cmdObj.firstElement().String());
-        auto query = assertGet(QueryRequest::makeFromFindCommand(nss, request.cmdObj, false));
-
-        ASSERT_EQ(ShardType::ConfigNS, query->ns());
-        ASSERT_BSONOBJ_EQ(BSONObj(), query->getFilter());
-        ASSERT_BSONOBJ_EQ(BSONObj(), query->getSort());
-        ASSERT_FALSE(query->getLimit().is_initialized());
-
-        checkReadConcern(request.cmdObj, Timestamp(0, 0), repl::OpTime::kUninitializedTerm);
-
-        return vector<BSONObj>{};
-    });
-
-    future.timed_get(kFutureTimeout);
-}
-
-TEST_F(ShardingCatalogClientTest, createDatabaseDuplicateKeyOnInsert) {
-    const string dbname = "databaseToCreate";
-    const HostAndPort configHost("TestHost1");
-    configTargeter()->setFindHostReturnValue(configHost);
-
-    ShardType s0;
-    s0.setName("shard0000");
-    s0.setHost("ShardHost0:27017");
-
-    ShardType s1;
-    s1.setName("shard0001");
-    s1.setHost("ShardHost1:27017");
-
-    ShardType s2;
-    s2.setName("shard0002");
-    s2.setHost("ShardHost2:27017");
-
-    // Prime the shard registry with information about the existing shards
-    auto future = launchAsync([this] { shardRegistry()->reload(operationContext()); });
-
-    onFindCommand([&](const RemoteCommandRequest& request) {
-        ASSERT_EQUALS(configHost, request.target);
-        ASSERT_BSONOBJ_EQ(getReplSecondaryOkMetadata(),
-                          rpc::TrackingMetadata::removeTrackingData(request.metadata));
-        const NamespaceString nss(request.dbname, request.cmdObj.firstElement().String());
-        auto query = assertGet(QueryRequest::makeFromFindCommand(nss, request.cmdObj, false));
-
-        ASSERT_EQ(ShardType::ConfigNS, query->ns());
-        ASSERT_BSONOBJ_EQ(BSONObj(), query->getFilter());
-        ASSERT_BSONOBJ_EQ(BSONObj(), query->getSort());
-        ASSERT_FALSE(query->getLimit().is_initialized());
-
-        checkReadConcern(request.cmdObj, Timestamp(0, 0), repl::OpTime::kUninitializedTerm);
-
-        return vector<BSONObj>{s0.toBSON(), s1.toBSON(), s2.toBSON()};
-    });
-
-    future.timed_get(kFutureTimeout);
-
-    // Set up all the target mocks return values.
-    RemoteCommandTargeterMock::get(
-        uassertStatusOK(shardRegistry()->getShard(operationContext(), s0.getName()))->getTargeter())
-        ->setFindHostReturnValue(HostAndPort(s0.getHost()));
-    RemoteCommandTargeterMock::get(
-        uassertStatusOK(shardRegistry()->getShard(operationContext(), s1.getName()))->getTargeter())
-        ->setFindHostReturnValue(HostAndPort(s1.getHost()));
-    RemoteCommandTargeterMock::get(
-        uassertStatusOK(shardRegistry()->getShard(operationContext(), s2.getName()))->getTargeter())
-        ->setFindHostReturnValue(HostAndPort(s2.getHost()));
-
-    // Now actually start the createDatabase work.
-
-    distLock()->expectLock(
-        [dbname](StringData name, StringData whyMessage, Milliseconds waitFor) {}, Status::OK());
-
-
-    future = launchAsync([this, dbname] {
-        Status status = catalogClient()->createDatabase(operationContext(), dbname);
-        ASSERT_EQUALS(ErrorCodes::NamespaceExists, status);
-    });
-
-    // Report no databases with the same name already exist
-    onFindCommand([&](const RemoteCommandRequest& request) {
-        ASSERT_EQUALS(configHost, request.target);
-        ASSERT_BSONOBJ_EQ(getReplSecondaryOkMetadata(),
-                          rpc::TrackingMetadata::removeTrackingData(request.metadata));
-        const NamespaceString nss(request.dbname, request.cmdObj.firstElement().String());
-        ASSERT_EQ(DatabaseType::ConfigNS, nss.ns());
-        checkReadConcern(request.cmdObj, Timestamp(0, 0), repl::OpTime::kUninitializedTerm);
-        return vector<BSONObj>{};
-    });
-
-    // Return size information about first shard
-    onCommand([&](const RemoteCommandRequest& request) {
-        ASSERT_EQUALS(s0.getHost(), request.target.toString());
-        ASSERT_EQUALS("admin", request.dbname);
-        string cmdName = request.cmdObj.firstElement().fieldName();
-        ASSERT_EQUALS("listDatabases", cmdName);
-        ASSERT_FALSE(request.cmdObj.hasField(repl::ReadConcernArgs::kReadConcernFieldName));
-
-        ASSERT_BSONOBJ_EQ(
-            ReadPreferenceSetting(ReadPreference::PrimaryPreferred).toContainingBSON(),
-            rpc::TrackingMetadata::removeTrackingData(request.metadata));
-
-        return BSON("ok" << 1 << "totalSize" << 10);
-    });
-
-    // Return size information about second shard
-    onCommand([&](const RemoteCommandRequest& request) {
-        ASSERT_EQUALS(s1.getHost(), request.target.toString());
-        ASSERT_EQUALS("admin", request.dbname);
-        string cmdName = request.cmdObj.firstElement().fieldName();
-        ASSERT_EQUALS("listDatabases", cmdName);
-        ASSERT_FALSE(request.cmdObj.hasField(repl::ReadConcernArgs::kReadConcernFieldName));
-
-        ASSERT_BSONOBJ_EQ(
-            ReadPreferenceSetting(ReadPreference::PrimaryPreferred).toContainingBSON(),
-            rpc::TrackingMetadata::removeTrackingData(request.metadata));
-
-        return BSON("ok" << 1 << "totalSize" << 1);
-    });
-
-    // Return size information about third shard
-    onCommand([&](const RemoteCommandRequest& request) {
-        ASSERT_EQUALS(s2.getHost(), request.target.toString());
-        ASSERT_EQUALS("admin", request.dbname);
-        string cmdName = request.cmdObj.firstElement().fieldName();
-        ASSERT_EQUALS("listDatabases", cmdName);
-        ASSERT_FALSE(request.cmdObj.hasField(repl::ReadConcernArgs::kReadConcernFieldName));
-
-        ASSERT_BSONOBJ_EQ(
-            ReadPreferenceSetting(ReadPreference::PrimaryPreferred).toContainingBSON(),
-            rpc::TrackingMetadata::removeTrackingData(request.metadata));
-
-        return BSON("ok" << 1 << "totalSize" << 100);
-    });
-
-    // Process insert to config.databases collection
-    onCommand([&](const RemoteCommandRequest& request) {
-        ASSERT_EQUALS(configHost, request.target);
-        ASSERT_EQUALS("config", request.dbname);
-        ASSERT_FALSE(request.cmdObj.hasField(repl::ReadConcernArgs::kReadConcernFieldName));
-
-        ASSERT_BSONOBJ_EQ(BSON(rpc::kReplSetMetadataFieldName << 1),
-                          rpc::TrackingMetadata::removeTrackingData(request.metadata));
-
-        const auto opMsgRequest = OpMsgRequest::fromDBAndBody(request.dbname, request.cmdObj);
-        const auto insertOp = InsertOp::parse(opMsgRequest);
-        ASSERT_EQUALS(DatabaseType::ConfigNS, insertOp.getNamespace().ns());
-
-        const auto& inserts = insertOp.getDocuments();
-        ASSERT_EQUALS(1U, inserts.size());
-
-        const auto& insert = inserts.front();
-        DatabaseType expectedDb;
-        expectedDb.setName(dbname);
-        expectedDb.setPrimary(
-            ShardId(s1.getName()));  // This is the one we reported with the smallest size
-        expectedDb.setSharded(false);
-
-        ASSERT_BSONOBJ_EQ(expectedDb.toBSON(), insert);
-
-        BatchedCommandResponse response;
-        response.setOk(false);
-        response.setErrCode(ErrorCodes::DuplicateKey);
-        response.setErrMessage("duplicate key");
-
-        return response.toBSON();
-    });
-
-    future.timed_get(kFutureTimeout);
-}
-
 TEST_F(ShardingCatalogClientTest, BasicReadAfterOpTime) {
     configTargeter()->setFindHostReturnValue(HostAndPort("TestHost1"));
 
diff --git a/src/mongo/s/commands/cluster_commands_helpers.cpp b/src/mongo/s/commands/cluster_commands_helpers.cpp
index ffda19bfc27..5c81d01cf9b 100644
--- a/src/mongo/s/commands/cluster_commands_helpers.cpp
+++ b/src/mongo/s/commands/cluster_commands_helpers.cpp
@@ -45,6 +45,7 @@
 #include "mongo/s/client/shard_registry.h"
 #include "mongo/s/client/version_manager.h"
 #include "mongo/s/grid.h"
+#include "mongo/s/request_types/create_database_gen.h"
 #include "mongo/s/shard_id.h"
 #include "mongo/s/stale_exception.h"
 #include "mongo/util/log.h"
@@ -437,8 +438,20 @@ CachedCollectionRoutingInfo getShardedCollection(OperationContext* opCtx,
 StatusWith<CachedDatabaseInfo> createShardDatabase(OperationContext* opCtx, StringData dbName) {
     auto dbStatus = Grid::get(opCtx)->catalogCache()->getDatabase(opCtx, dbName);
     if (dbStatus == ErrorCodes::NamespaceNotFound) {
+        ConfigsvrCreateDatabase configCreateDatabaseRequest;
+        configCreateDatabaseRequest.set_configsvrCreateDatabase(dbName);
+
+        auto configShard = Grid::get(opCtx)->shardRegistry()->getConfigShard();
+
         auto createDbStatus =
-            Grid::get(opCtx)->catalogClient()->createDatabase(opCtx, dbName.toString());
+            uassertStatusOK(configShard->runCommandWithFixedRetryAttempts(
+                                opCtx,
+                                ReadPreferenceSetting(ReadPreference::PrimaryOnly),
+                                "admin",
+                                configCreateDatabaseRequest.toBSON(),
+                                Shard::RetryPolicy::kIdempotent))
+                .commandStatus;
+
         if (createDbStatus.isOK() || createDbStatus == ErrorCodes::NamespaceExists) {
             dbStatus = Grid::get(opCtx)->catalogCache()->getDatabase(opCtx, dbName);
         } else {
diff --git a/src/mongo/s/request_types/create_database.idl b/src/mongo/s/request_types/create_database.idl
new file mode 100644
index 00000000000..f357a005778
--- /dev/null
+++ b/src/mongo/s/request_types/create_database.idl
@@ -0,0 +1,42 @@
+# 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.
+
+# createDatabase IDL File
+
+global:
+    cpp_namespace: "mongo"
+
+imports:
+    - "mongo/idl/basic_types.idl"
+
+structs:
+    ConfigsvrCreateDatabase:
+        description: "The internal createDatabase command on the config server"
+        strict: false
+        fields:
+            _configsvrCreateDatabase:
+                type: string
+                description: "The namespace of the database to be created."
\ No newline at end of file