SERVER-56644 Create oplog views for resharding and tenant migration on startup

(cherry picked from commit c6e8fed25eaa9ca8fa43517ad8dca9d74308274f)

10 files changed, 83 insertions, 11 deletions

diff --git a/jstests/multiVersion/oplog_views_for_resharding_and_tenant_migration.js b/jstests/multiVersion/oplog_views_for_resharding_and_tenant_migration.js
new file mode 100644
index 00000000000..5ec13841a8e
--- /dev/null
+++ b/jstests/multiVersion/oplog_views_for_resharding_and_tenant_migration.js
@@ -0,0 +1,43 @@
+/**
+ * Tests that oplog views for resharding and tenant migration are created when a replica set is
+ * upgraded from 4.4.
+ */
+(function() {
+"use strict";
+load("jstests/multiVersion/libs/multi_rs.js");  // For 'upgradeSet()'
+
+const kReshardingOplogViewName = "system.resharding.slimOplogForGraphLookup";
+const kTenantMigrationOplogViewName = "system.tenantMigration.oplogView";
+
+function assertOplogViewsNotExist(node) {
+    const localDb = node.getDB("local");
+    const reshardingViewRes = localDb.runCommand(
+        {listCollections: 1, filter: {type: "view", name: kReshardingOplogViewName}});
+    assert.eq(0, reshardingViewRes.cursor.firstBatch.length);
+    const tenantMigrationRes = localDb.runCommand(
+        {listCollections: 1, filter: {type: "view", name: kTenantMigrationOplogViewName}});
+    assert.eq(0, tenantMigrationRes.cursor.firstBatch.length);
+}
+
+function assertOplogViewsExist(node) {
+    const localDb = node.getDB("local");
+    const reshardingViewRes = localDb.runCommand(
+        {listCollections: 1, filter: {type: "view", name: kReshardingOplogViewName}});
+    assert.eq(1, reshardingViewRes.cursor.firstBatch.length);
+    const tenantMigrationRes = localDb.runCommand(
+        {listCollections: 1, filter: {type: "view", name: kTenantMigrationOplogViewName}});
+    assert.eq(1, tenantMigrationRes.cursor.firstBatch.length);
+}
+
+// Set up a replica set in v4.4.
+const rst = new ReplSetTest({nodes: 2, nodeOptions: {binVersion: "last-lts"}});
+rst.startSet();
+rst.initiate();
+rst.nodes.forEach(node => assertOplogViewsNotExist(node));
+
+// Upgrade the replica set.
+rst.upgradeSet({binVersion: "latest"});
+rst.nodes.forEach(node => assertOplogViewsExist(node));
+
+rst.stopSet();
+})();
\ No newline at end of file
diff --git a/src/mongo/db/mongod_main.cpp b/src/mongo/db/mongod_main.cpp
index fed12fe6c3f..48a003c2209 100644
--- a/src/mongo/db/mongod_main.cpp
+++ b/src/mongo/db/mongod_main.cpp
@@ -143,6 +143,7 @@
 #include "mongo/db/s/resharding/resharding_donor_service.h"
 #include "mongo/db/s/resharding/resharding_op_observer.h"
 #include "mongo/db/s/resharding/resharding_recipient_service.h"
+#include "mongo/db/s/resharding_util.h"
 #include "mongo/db/s/shard_server_op_observer.h"
 #include "mongo/db/s/sharding_ddl_coordinator_service.h"
 #include "mongo/db/s/sharding_initialization_mongod.h"
@@ -702,6 +703,13 @@ ExitCode _initAndListen(ServiceContext* serviceContext, int listenPort) {
         if (replSettings.usingReplSets() || !gInternalValidateFeaturesAsPrimary) {
             serverGlobalParams.validateFeaturesAsPrimary.store(false);
         }
+
+        if (replSettings.usingReplSets()) {
+            Lock::GlobalWrite lk(startupOpCtx.get());
+            OldClientContext ctx(startupOpCtx.get(), NamespaceString::kRsOplogNamespace.ns());
+            createSlimOplogView(startupOpCtx.get(), ctx.db());
+            tenant_migration_util::createOplogViewForTenantMigrations(startupOpCtx.get(), ctx.db());
+        }
     }
 
     startClientCursorMonitor();
diff --git a/src/mongo/db/namespace_string.cpp b/src/mongo/db/namespace_string.cpp
index 2414afacd6b..02f20c2b13b 100644
--- a/src/mongo/db/namespace_string.cpp
+++ b/src/mongo/db/namespace_string.cpp
@@ -118,6 +118,9 @@ const NamespaceString NamespaceString::kReshardingApplierProgressNamespace(
 const NamespaceString NamespaceString::kReshardingTxnClonerProgressNamespace(
     NamespaceString::kConfigDb, "localReshardingOperations.recipient.progress_txn_cloner");
 
+const NamespaceString NamespaceString::kReshardingOplogView(
+    NamespaceString::kLocalDb, "system.resharding.slimOplogForGraphLookup");
+
 const NamespaceString NamespaceString::kCollectionCriticalSectionsNamespace(
     NamespaceString::kConfigDb, "collection_critical_sections");
 
diff --git a/src/mongo/db/namespace_string.h b/src/mongo/db/namespace_string.h
index f918180f567..d28d28d8421 100644
--- a/src/mongo/db/namespace_string.h
+++ b/src/mongo/db/namespace_string.h
@@ -168,6 +168,9 @@ public:
     // Namespace for storing config.transactions cloner progress for resharding.
     static const NamespaceString kReshardingTxnClonerProgressNamespace;
 
+    // Namespace for view on local.oplog.rs for resharding.
+    static const NamespaceString kReshardingOplogView;
+
     // Namespace for storing config.collectionCriticalSections documents
     static const NamespaceString kCollectionCriticalSectionsNamespace;
 
diff --git a/src/mongo/db/repl/oplog.cpp b/src/mongo/db/repl/oplog.cpp
index f4fabae95bd..27024e566e3 100644
--- a/src/mongo/db/repl/oplog.cpp
+++ b/src/mongo/db/repl/oplog.cpp
@@ -82,7 +82,6 @@
 #include "mongo/db/repl/replication_coordinator.h"
 #include "mongo/db/repl/tenant_migration_access_blocker_util.h"
 #include "mongo/db/repl/tenant_migration_decoration.h"
-#include "mongo/db/repl/tenant_migration_util.h"
 #include "mongo/db/repl/timestamp_block.h"
 #include "mongo/db/repl/transaction_oplog_application.h"
 #include "mongo/db/s/resharding_util.h"
@@ -662,9 +661,6 @@ void createOplog(OperationContext* opCtx,
         uow.commit();
     });
 
-    createSlimOplogView(opCtx, ctx.db());
-    tenant_migration_util::createOplogViewForTenantMigrations(opCtx, ctx.db());
-
     /* sync here so we don't get any surprising lag later when we try to sync */
     service->getStorageEngine()->flushAllFiles(opCtx, /*callerHoldsReadLock*/ false);
 }
diff --git a/src/mongo/db/repl/tenant_migration_recipient_service_test.cpp b/src/mongo/db/repl/tenant_migration_recipient_service_test.cpp
index 9405788fd62..d3eb14fb9b0 100644
--- a/src/mongo/db/repl/tenant_migration_recipient_service_test.cpp
+++ b/src/mongo/db/repl/tenant_migration_recipient_service_test.cpp
@@ -39,6 +39,7 @@
 #include "mongo/config.h"
 #include "mongo/db/client.h"
 #include "mongo/db/commands/feature_compatibility_version_document_gen.h"
+#include "mongo/db/db_raii.h"
 #include "mongo/db/dbdirectclient.h"
 #include "mongo/db/op_observer_impl.h"
 #include "mongo/db/op_observer_registry.h"
@@ -192,6 +193,11 @@ public:
             ReplicationCoordinator::set(serviceContext, std::move(replCoord));
 
             repl::createOplog(opCtx.get());
+            {
+                Lock::GlobalWrite lk(opCtx.get());
+                OldClientContext ctx(opCtx.get(), NamespaceString::kRsOplogNamespace.ns());
+                tenant_migration_util::createOplogViewForTenantMigrations(opCtx.get(), ctx.db());
+            }
 
             // Need real (non-mock) storage for the oplog buffer.
             StorageInterface::set(serviceContext, std::make_unique<StorageInterfaceImpl>());
diff --git a/src/mongo/db/repl/tenant_migration_util.cpp b/src/mongo/db/repl/tenant_migration_util.cpp
index 860fc442eea..2de0b42f6c3 100644
--- a/src/mongo/db/repl/tenant_migration_util.cpp
+++ b/src/mongo/db/repl/tenant_migration_util.cpp
@@ -148,8 +148,12 @@ void createOplogViewForTenantMigrations(OperationContext* opCtx, Database* db) {
             options.pipeline = pipeline.arr();
 
             WriteUnitOfWork wuow(opCtx);
-            uassertStatusOK(
-                db->createView(opCtx, NamespaceString::kTenantMigrationOplogView, options));
+            auto status =
+                db->createView(opCtx, NamespaceString::kTenantMigrationOplogView, options);
+            if (status == ErrorCodes::NamespaceExists) {
+                return;
+            }
+            uassertStatusOK(status);
             wuow.commit();
         });
 }
diff --git a/src/mongo/db/s/SConscript b/src/mongo/db/s/SConscript
index 28c64b3c6a3..8f5434acea1 100644
--- a/src/mongo/db/s/SConscript
+++ b/src/mongo/db/s/SConscript
@@ -174,6 +174,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/pipeline/expression_context',
         '$BUILD_DIR/mongo/db/pipeline/pipeline',
         '$BUILD_DIR/mongo/db/storage/write_unit_of_work',
+        '$BUILD_DIR/mongo/db/views/views',
         '$BUILD_DIR/mongo/s/async_requests_sender',
         '$BUILD_DIR/mongo/s/grid',
         'sharding_api_d',
diff --git a/src/mongo/db/s/resharding/resharding_oplog_fetcher_test.cpp b/src/mongo/db/s/resharding/resharding_oplog_fetcher_test.cpp
index ab0932a9daf..a4bc0e6a5fd 100644
--- a/src/mongo/db/s/resharding/resharding_oplog_fetcher_test.cpp
+++ b/src/mongo/db/s/resharding/resharding_oplog_fetcher_test.cpp
@@ -90,6 +90,12 @@ public:
         _opCtx = operationContext();
         _svcCtx = _opCtx->getServiceContext();
 
+        {
+            Lock::GlobalWrite lk(_opCtx);
+            OldClientContext ctx(_opCtx, NamespaceString::kRsOplogNamespace.ns());
+            createSlimOplogView(_opCtx, ctx.db());
+        }
+
         // Initialize ReshardingMetrics to a recipient state compatible with fetching.
         _metrics = std::make_unique<ReshardingMetrics>(_svcCtx);
         _metrics->onStart();
diff --git a/src/mongo/db/s/resharding_util.cpp b/src/mongo/db/s/resharding_util.cpp
index 1aeb6e36014..874db215615 100644
--- a/src/mongo/db/s/resharding_util.cpp
+++ b/src/mongo/db/s/resharding_util.cpp
@@ -51,6 +51,7 @@
 #include "mongo/db/s/collection_sharding_state.h"
 #include "mongo/db/s/sharding_state.h"
 #include "mongo/db/storage/write_unit_of_work.h"
+#include "mongo/db/views/view_catalog.h"
 #include "mongo/logv2/log.h"
 #include "mongo/rpc/get_status_from_command_result.h"
 #include "mongo/s/async_requests_sender.h"
@@ -233,7 +234,7 @@ void validateZones(const std::vector<mongo::BSONObj>& zones,
 
 void createSlimOplogView(OperationContext* opCtx, Database* db) {
     writeConflictRetry(
-        opCtx, "createReshardingSlimOplog", "local.system.resharding.slimOplogForGraphLookup", [&] {
+        opCtx, "createReshardingSlimOplog", NamespaceString::kReshardingOplogView.ns(), [&] {
             {
                 // Create 'system.views' in a separate WUOW if it does not exist.
                 WriteUnitOfWork wuow(opCtx);
@@ -256,10 +257,11 @@ void createSlimOplogView(OperationContext* opCtx, Database* db) {
             options.viewOn = NamespaceString::kRsOplogNamespace.coll().toString();
             options.pipeline = BSON_ARRAY(getSlimOplogPipeline());
             WriteUnitOfWork wuow(opCtx);
-            uassertStatusOK(
-                db->createView(opCtx,
-                               NamespaceString("local.system.resharding.slimOplogForGraphLookup"),
-                               options));
+            auto status = db->createView(opCtx, NamespaceString::kReshardingOplogView, options);
+            if (status == ErrorCodes::NamespaceExists) {
+                return;
+            }
+            uassertStatusOK(status);
             wuow.commit();
         });
 }