SERVER-52812 Unify the implicit createDatabase/enableSharding flow

This change makes both enableSharding and the implicit createDatabase
from the routers to go through the same _configsvrCreateDatabase
command. This command has the sole responsibility now of creating (or
ensuring it is created) a database with the specified name, optional
primary flag and optional enableSharding field.

1 files changed, 6 insertions, 57 deletions

diff --git a/src/mongo/db/s/config/configsvr_enable_sharding_command.cpp b/src/mongo/db/s/config/configsvr_enable_sharding_command.cpp
index 7ebcb2e0d14..6a20e63a6a0 100644
--- a/src/mongo/db/s/config/configsvr_enable_sharding_command.cpp
+++ b/src/mongo/db/s/config/configsvr_enable_sharding_command.cpp
@@ -37,15 +37,10 @@
 #include "mongo/db/auth/authorization_session.h"
 #include "mongo/db/client.h"
 #include "mongo/db/commands.h"
-#include "mongo/db/dbdirectclient.h"
 #include "mongo/db/field_parser.h"
 #include "mongo/db/operation_context.h"
-#include "mongo/db/ops/write_ops.h"
 #include "mongo/db/repl/read_concern_args.h"
 #include "mongo/db/s/config/sharding_catalog_manager.h"
-#include "mongo/db/s/dist_lock_manager.h"
-#include "mongo/s/catalog/type_database.h"
-#include "mongo/s/catalog_cache.h"
 #include "mongo/s/grid.h"
 #include "mongo/util/scopeguard.h"
 
@@ -113,58 +108,12 @@ public:
         const std::string dbname = parseNs("", cmdObj);
 
         auto shardElem = cmdObj[kShardNameField];
-        ShardId shardId = shardElem.ok() ? ShardId(shardElem.String()) : ShardId();
-
-        // If assigned, check that the shardId is valid
-        uassert(ErrorCodes::BadValue,
-                str::stream() << "invalid shard name: " << shardId,
-                !shardElem.ok() || shardId.isValid());
-
-        uassert(
-            ErrorCodes::InvalidNamespace,
-            str::stream() << "invalid db name specified: " << dbname,
-            NamespaceString::validDBName(dbname, NamespaceString::DollarInDbNameBehavior::Allow));
-
-        if (dbname == NamespaceString::kAdminDb || dbname == NamespaceString::kLocalDb) {
-            uasserted(ErrorCodes::InvalidOptions,
-                      str::stream() << "can't shard " + dbname + " database");
-        }
-
-        // Make sure to force update of any stale metadata
-        ON_BLOCK_EXIT([opCtx, dbname] { Grid::get(opCtx)->catalogCache()->purgeDatabase(dbname); });
-
-        // For an existing database, the enableSharding operation is just adding the {sharded: true}
-        // field to config.database. First do an optimistic attempt to add it and if the write
-        // succeeds do not go through the createDatabase flow.
-        DBDirectClient client(opCtx);
-        auto response = UpdateOp::parseResponse([&] {
-            write_ops::Update updateOp(DatabaseType::ConfigNS);
-            updateOp.setUpdates({[&] {
-                BSONObjBuilder queryFilterBuilder;
-                queryFilterBuilder.append(DatabaseType::name.name(), dbname);
-                if (shardId.isValid())
-                    queryFilterBuilder.append(DatabaseType::primary.name(), shardId.toString());
-                auto updateModification = write_ops::UpdateModification(
-                    write_ops::UpdateModification::parseFromClassicUpdate(
-                        BSON("$set" << BSON(DatabaseType::sharded(true)))));
-                write_ops::UpdateOpEntry updateEntry(queryFilterBuilder.obj(), updateModification);
-                updateEntry.setMulti(false);
-                updateEntry.setUpsert(false);
-                return updateEntry;
-            }()});
-
-            auto response = client.runCommand(updateOp.serialize({}));
-            return response->getCommandReply();
-        }());
-
-        // If an entry for the database was found it can be assumed that it was either updated or
-        // already had 'sharded' enabled, so we can assume success
-        if (response.getN() != 1) {
-            auto dbDistLock = uassertStatusOK(DistLockManager::get(opCtx)->lock(
-                opCtx, dbname, "enableSharding", DistLockManager::kDefaultLockTimeout));
-
-            ShardingCatalogManager::get(opCtx)->enableSharding(opCtx, dbname, shardId);
-        }
+        ShardingCatalogManager::get(opCtx)->createDatabase(
+            opCtx,
+            dbname,
+            shardElem.ok() ? boost::optional<ShardId>(shardElem.String())
+                           : boost::optional<ShardId>(),
+            true);
 
         audit::logEnableSharding(Client::getCurrent(), dbname);