SERVER-49570 Create ReshardingCoordinator in configsvrReshardCollection

7 files changed, 96 insertions, 2 deletions

diff --git a/src/mongo/db/mongod_main.cpp b/src/mongo/db/mongod_main.cpp
index d654bb7e881..f6809e00bf9 100644
--- a/src/mongo/db/mongod_main.cpp
+++ b/src/mongo/db/mongod_main.cpp
@@ -135,6 +135,7 @@
 #include "mongo/db/s/migration_util.h"
 #include "mongo/db/s/op_observer_sharding_impl.h"
 #include "mongo/db/s/periodic_sharded_index_consistency_checker.h"
+#include "mongo/db/s/resharding/resharding_coordinator_service.h"
 #include "mongo/db/s/shard_server_op_observer.h"
 #include "mongo/db/s/sharding_initialization_mongod.h"
 #include "mongo/db/s/sharding_state_recovery.h"
@@ -305,6 +306,10 @@ void registerPrimaryOnlyServices(ServiceContext* serviceContext) {
     std::unique_ptr<TenantMigrationDonorService> tenantMigrationDonorService =
         std::make_unique<TenantMigrationDonorService>(serviceContext);
     registry->registerService(std::move(tenantMigrationDonorService));
+
+    std::unique_ptr<ReshardingCoordinatorService> reshardingCoordinatorService =
+        std::make_unique<ReshardingCoordinatorService>(serviceContext);
+    registry->registerService(std::move(reshardingCoordinatorService));
 }
 
 MONGO_FAIL_POINT_DEFINE(shutdownAtStartup);
diff --git a/src/mongo/db/namespace_string.cpp b/src/mongo/db/namespace_string.cpp
index 9660368c23e..5e4100ea8ac 100644
--- a/src/mongo/db/namespace_string.cpp
+++ b/src/mongo/db/namespace_string.cpp
@@ -252,7 +252,7 @@ bool NamespaceString::isConfigDotCacheDotChunks() const {
 }
 
 bool NamespaceString::isTemporaryReshardingCollection() const {
-    return coll().startsWith("system.resharding.");
+    return coll().startsWith(kTemporaryReshardingCollectionPrefix);
 }
 
 bool NamespaceString::isReplicated() const {
diff --git a/src/mongo/db/namespace_string.h b/src/mongo/db/namespace_string.h
index 46811e5e6b6..ad3c5e7bbb9 100644
--- a/src/mongo/db/namespace_string.h
+++ b/src/mongo/db/namespace_string.h
@@ -73,6 +73,9 @@ public:
     static constexpr StringData kOrphanCollectionPrefix = "orphan."_sd;
     static constexpr StringData kOrphanCollectionDb = "local"_sd;
 
+    // Prefix for temporary resharding collection.
+    static constexpr StringData kTemporaryReshardingCollectionPrefix = "system.resharding."_sd;
+
     // Namespace for storing configuration data, which needs to be replicated if the server is
     // running as a replica set. Documents in this collection should represent some configuration
     // state of the server, which needs to be recovered/consulted at startup. Each document in this
@@ -118,7 +121,7 @@ public:
     // Namespace for pending range deletions.
     static const NamespaceString kRangeDeletionNamespace;
 
-    // Namespace for the config server's resharding operation state.
+    // Namespace for the coordinator's resharding operation state.
     static const NamespaceString kConfigReshardingOperationsNamespace;
 
     // Namespace for the donor shard's local resharding operation state.
diff --git a/src/mongo/db/s/SConscript b/src/mongo/db/s/SConscript
index 83629f46e81..4f97f81b966 100644
--- a/src/mongo/db/s/SConscript
+++ b/src/mongo/db/s/SConscript
@@ -339,6 +339,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/commands/server_status',
         '$BUILD_DIR/mongo/db/commands/test_commands_enabled',
         '$BUILD_DIR/mongo/db/commands/txn_cmd_request',
+        '$BUILD_DIR/mongo/db/repl/primary_only_service',
         '$BUILD_DIR/mongo/db/repl/repl_coordinator_interface',
         '$BUILD_DIR/mongo/db/repl/replica_set_messages',
         '$BUILD_DIR/mongo/s/commands/shared_cluster_commands',
diff --git a/src/mongo/db/s/config/configsvr_reshard_collection_cmd.cpp b/src/mongo/db/s/config/configsvr_reshard_collection_cmd.cpp
index 7ecbb51e33f..10347347fce 100644
--- a/src/mongo/db/s/config/configsvr_reshard_collection_cmd.cpp
+++ b/src/mongo/db/s/config/configsvr_reshard_collection_cmd.cpp
@@ -34,6 +34,9 @@
 #include "mongo/db/auth/authorization_session.h"
 #include "mongo/db/commands.h"
 #include "mongo/db/query/collation/collator_factory_interface.h"
+#include "mongo/db/repl/primary_only_service.h"
+#include "mongo/db/s/resharding/coordinator_document_gen.h"
+#include "mongo/db/s/resharding/resharding_coordinator_service.h"
 #include "mongo/db/s/resharding_util.h"
 #include "mongo/logv2/log.h"
 #include "mongo/s/catalog/type_tags.h"
@@ -105,15 +108,69 @@ public:
                     "Must specify only one of _presetReshardedChunks or numInitialChunks",
                     !(presetReshardedChunksSpecified && bool(request().getNumInitialChunks())));
 
+            std::set<ShardId> donorShardIds;
+            cm.getAllShardIds(&donorShardIds);
+
             int numInitialChunks;
+            std::set<ShardId> recipientShardIds;
             if (presetReshardedChunksSpecified) {
                 const auto chunks = request().get_presetReshardedChunks().get();
                 validateReshardedChunks(
                     chunks, opCtx, ShardKeyPattern(request().getKey()).getKeyPattern());
                 numInitialChunks = chunks.size();
+
+                // Use the provided shardIds from presetReshardedChunks to construct the
+                // recipient list.
+                for (const BSONObj& obj : chunks) {
+                    recipientShardIds.emplace(
+                        obj.getStringField(ReshardedChunk::kRecipientShardIdFieldName));
+                }
             } else {
                 numInitialChunks = request().getNumInitialChunks().get_value_or(cm.numChunks());
+
+                // No presetReshardedChunks were provided, make the recipients list be the same as
+                // the donors list by default.
+                recipientShardIds = donorShardIds;
             }
+
+            // Construct the lists of donor and recipient shard entries, where each ShardEntry is
+            // in state kUnused.
+            std::vector<DonorShardEntry> donorShards;
+            std::transform(donorShardIds.begin(),
+                           donorShardIds.end(),
+                           std::back_inserter(donorShards),
+                           [](const ShardId& shardId) -> DonorShardEntry {
+                               DonorShardEntry entry{shardId};
+                               entry.setState(DonorStateEnum::kUnused);
+                               return entry;
+                           });
+            std::vector<RecipientShardEntry> recipientShards;
+            std::transform(recipientShardIds.begin(),
+                           recipientShardIds.end(),
+                           std::back_inserter(recipientShards),
+                           [](const ShardId& shardId) -> RecipientShardEntry {
+                               RecipientShardEntry entry{shardId};
+                               entry.setState(RecipientStateEnum::kUnused);
+                               return entry;
+                           });
+
+            auto tempReshardingNss = constructTemporaryReshardingNss(nss, cm);
+            auto coordinatorDoc =
+                ReshardingCoordinatorDocument(std::move(tempReshardingNss),
+                                              std::move(CoordinatorStateEnum::kInitializing),
+                                              std::move(donorShards),
+                                              std::move(recipientShards));
+
+            // Generate the resharding metadata for the ReshardingCoordinatorDocument.
+            auto reshardingUUID = UUID::gen();
+            auto commonMetadata =
+                CommonReshardingMetadata(std::move(reshardingUUID), ns(), request().getKey());
+            coordinatorDoc.setCommonReshardingMetadata(std::move(commonMetadata));
+
+            auto registry = repl::PrimaryOnlyServiceRegistry::get(opCtx->getServiceContext());
+            auto service = registry->lookupServiceByName(kReshardingCoordinatorServiceName);
+            auto instance = ReshardingCoordinatorService::ReshardingCoordinator::getOrCreate(
+                service, coordinatorDoc.toBSON());
         }
 
     private:
diff --git a/src/mongo/db/s/resharding_util.cpp b/src/mongo/db/s/resharding_util.cpp
index e11b33f7b15..123c7b66a8b 100644
--- a/src/mongo/db/s/resharding_util.cpp
+++ b/src/mongo/db/s/resharding_util.cpp
@@ -31,6 +31,8 @@
 
 #include "mongo/platform/basic.h"
 
+#include <fmt/format.h>
+
 #include "mongo/bson/bsonobj.h"
 #include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/namespace_string.h"
@@ -47,6 +49,20 @@
 #include "mongo/s/request_types/flush_routing_table_cache_updates_gen.h"
 
 namespace mongo {
+using namespace fmt::literals;
+
+NamespaceString constructTemporaryReshardingNss(const NamespaceString& originalNss,
+                                                const ChunkManager& cm) {
+    auto collectionUUID = cm.getUUID();
+    uassert(ErrorCodes::InvalidUUID,
+            "Cannot reshard collection {} due to missing UUID"_format(originalNss.ns()),
+            collectionUUID);
+    NamespaceString tempReshardingNss(
+        originalNss.db(),
+        "{}{}"_format(NamespaceString::kTemporaryReshardingCollectionPrefix,
+                      collectionUUID->toString()));
+    return tempReshardingNss;
+}
 
 void tellShardsToRefresh(OperationContext* opCtx,
                          const std::vector<ShardId>& shardIds,
diff --git a/src/mongo/db/s/resharding_util.h b/src/mongo/db/s/resharding_util.h
index 517ef531b29..5c07884d0a9 100644
--- a/src/mongo/db/s/resharding_util.h
+++ b/src/mongo/db/s/resharding_util.h
@@ -38,6 +38,7 @@
 #include "mongo/db/s/resharding/donor_oplog_id_gen.h"
 #include "mongo/executor/task_executor.h"
 #include "mongo/s/catalog/type_tags.h"
+#include "mongo/s/catalog_cache.h"
 #include "mongo/s/resharded_chunk_gen.h"
 #include "mongo/s/shard_id.h"
 
@@ -46,6 +47,17 @@ namespace mongo {
 constexpr auto kReshardingOplogPrePostImageOps = "prePostImageOps"_sd;
 
 /**
+ * Constructs the temporary resharding collection's namespace provided the original collection's
+ * namespace and chunk manager.
+ *
+ *      <db>.system.resharding.<existing collection's UUID>
+ *
+ * Note: throws if the original collection does not have a UUID.
+ */
+NamespaceString constructTemporaryReshardingNss(const NamespaceString& originalNss,
+                                                const ChunkManager& cm);
+
+/**
  * Sends _flushRoutingTableCacheUpdatesWithWriteConcern to a list of shards. Throws if one of the
  * shards fails to refresh.
  */