SERVER-62681 Create change streams per shard cursor passthrough suite

3 files changed, 116 insertions, 0 deletions

diff --git a/buildscripts/resmokeconfig/suites/change_streams_per_shard_cursor_passthrough.yml b/buildscripts/resmokeconfig/suites/change_streams_per_shard_cursor_passthrough.yml
new file mode 100644
index 00000000000..db39ec22c09
--- /dev/null
+++ b/buildscripts/resmokeconfig/suites/change_streams_per_shard_cursor_passthrough.yml
@@ -0,0 +1,58 @@
+test_kind: js_test
+
+selector:
+  roots:
+  - jstests/change_streams/**/*.js
+  exclude_with_any_tags:
+  ##
+  # The next tags correspond to the special errors thrown by the
+  # set_read_and_write_concerns.js override when it refuses to replace the readConcern or
+  # writeConcern of a particular command. Above each tag are the message(s) that cause the tag to be
+  # warranted.
+  ##
+  # "Cowardly refusing to override write concern of command: ..."
+  - assumes_write_concern_unchanged
+  # Exclude any that assume sharding is disabled
+  - assumes_against_mongod_not_mongos
+  - assumes_unsharded_collection
+
+executor:
+  archive:
+    hooks:
+      - CheckReplDBHash
+      - ValidateCollections
+  config:
+    shell_options:
+      global_vars:
+        TestData:
+          defaultReadConcernLevel: null
+          enableMajorityReadConcern: ''
+      # Enable causal consistency for change streams suites using 1 node replica sets. See
+      # change_streams.yml for detailed explanation.
+      eval: >-
+        var testingReplication = true;
+        load('jstests/libs/override_methods/set_read_and_write_concerns.js');
+        load('jstests/libs/override_methods/implicitly_shard_accessed_collections.js');
+        load('jstests/libs/override_methods/enable_causal_consistency_without_read_pref.js');
+        load('jstests/libs/override_methods/implicit_passthrough_to_shard_changestreams.js');
+  hooks:
+  - class: CheckReplDBHash
+  - class: ValidateCollections
+  - class: CleanEveryN
+    n: 20
+  fixture:
+    class: ShardedClusterFixture
+    mongos_options:
+      bind_ip_all: ''
+      set_parameters:
+        enableTestCommands: 1
+    mongod_options:
+      bind_ip_all: ''
+      set_parameters:
+        enableTestCommands: 1
+        writePeriodicNoops: 1
+        periodicNoopIntervalSecs: 1
+        coordinateCommitReturnImmediatelyAfterPersistingDecision: true
+    num_shards: 1
+    enable_sharding:
+    - test
diff --git a/etc/evergreen.yml b/etc/evergreen.yml
index cd526e85d66..5a974bb60a4 100644
--- a/etc/evergreen.yml
+++ b/etc/evergreen.yml
@@ -4414,6 +4414,13 @@ tasks:
   - func: "run tests"
 
 - <<: *task_template
+  name: change_streams_per_shard_cursor_passthrough
+  tags: []
+  commands:
+  - func: "do setup"
+  - func: "run tests"
+
+- <<: *task_template
   name: disk_wiredtiger
   commands:
   - func: "do setup"
@@ -9084,6 +9091,7 @@ buildvariants:
         --excludeWithAnyTags=incompatible_with_windows_tls
         --excludeWithAnyTags=incompatible_with_shard_merge
   tasks:
+    - name: change_streams_per_shard_cursor_passthrough
     - name: cqf
     - name: cqf_parallel
     - name: compile_and_archive_dist_test_then_package_TG
@@ -10006,6 +10014,7 @@ buildvariants:
       --runAllFeatureFlagTests
       --excludeWithAnyTags=incompatible_with_shard_merge
   tasks: &enterprise-rhel-80-64-bit-dynamic-all-feature-flags-tasks
+  - name: change_streams_per_shard_cursor_passthrough
   - name: cqf
   - name: cqf_parallel
   - name: compile_test_and_package_parallel_core_stream_TG
@@ -11836,6 +11845,7 @@ buildvariants:
         --excludeWithAnyTags=incompatible_with_shard_merge
     separate_debug: off
   tasks:
+  - name: change_streams_per_shard_cursor_passthrough
   - name: cqf
   - name: cqf_parallel
   - name: compile_and_archive_dist_test_then_package_TG
@@ -12080,6 +12090,7 @@ buildvariants:
     scons_cache_scope: shared
     separate_debug: off
   tasks:
+  - name: change_streams_per_shard_cursor_passthrough
   - name: cqf
   - name: cqf_parallel
   - name: compile_and_archive_dist_test_then_package_TG
diff --git a/jstests/libs/override_methods/implicit_passthrough_to_shard_changestreams.js b/jstests/libs/override_methods/implicit_passthrough_to_shard_changestreams.js
new file mode 100644
index 00000000000..8e8779e7ca5
--- /dev/null
+++ b/jstests/libs/override_methods/implicit_passthrough_to_shard_changestreams.js
@@ -0,0 +1,47 @@
+/**
+ * Overrides runCommand to use the $_passthroughToShard parameter. The changestreams per-shard
+ * cursor passthrough suite ensures changestream tests can still run correctly on a single-shard
+ * cluster. By adding this parameter, we pass through to that single shard, running the pipelines
+ * directly on that mongod. This will test the machinery of per-shard cursors via mongos.
+ */
+
+(function() {
+'use strict';
+
+load("jstests/libs/override_methods/override_helpers.js");  // For 'OverrideHelpers'.
+load("jstests/libs/discover_topology.js");                  // For 'DiscoverTopology'.
+
+// To be eligible, a command must be a changeStream request sent to a mongos.
+const isEligibleForPerShardCursor = function(conn, cmdObj) {
+    if (!(cmdObj && cmdObj.aggregate && Array.isArray(cmdObj.pipeline) &&
+          cmdObj.pipeline.length > 0 && typeof cmdObj.pipeline[0].$changeStream == "object" &&
+          cmdObj.pipeline[0].$changeStream.constructor === Object)) {
+        return false;
+    }
+    return conn.isMongos();
+};
+
+const discoverShardId = function(conn) {
+    const topology = DiscoverTopology.findConnectedNodes(conn);
+    const shards = topology.shards;
+    let shardName = Object.keys(shards)[0];
+    return {shard: shardName};
+};
+
+function runCommandWithPassthroughToShard(
+    conn, _dbName, _commandName, commandObj, func, makeFuncArgs) {
+    if (typeof commandObj !== "object" || commandObj === null) {
+        return func.apply(conn, makeFuncArgs(commandObj));
+    }
+    if (!isEligibleForPerShardCursor(conn, commandObj)) {
+        return func.apply(conn, makeFuncArgs(commandObj));
+    }
+    commandObj.$_passthroughToShard = discoverShardId(conn);
+    return func.apply(conn, makeFuncArgs(commandObj));
+}
+
+OverrideHelpers.prependOverrideInParallelShell(
+    "jstests/libs/override_methods/implicit_passthrough_to_shard_changestreams.js");
+
+OverrideHelpers.overrideRunCommand(runCommandWithPassthroughToShard);
+}());