SERVER-42723 New shard with new database can be ignored by change streams

Author: gormanb

buildscripts/resmokeconfig/suites/sharding_last_stable_mongos_and_mixed_shards.yml

@@ -24,6 +24,7 @@ selector:
   - jstests/sharding/prepared_txn_metadata_refresh.js
   # Enable when 4.4 becomes last stable
   - jstests/sharding/bulk_insert.js
+  - jstests/sharding/change_streams_new_shard_new_database.js
   - jstests/sharding/clear_jumbo.js
   - jstests/sharding/comment_field.js
   - jstests/sharding/covered_shard_key_indexes.js

jstests/change_streams/required_as_first_stage.js

@@ -20,14 +20,14 @@ assertErrorCode(
     40602);
 
 let error = assert.throws(() => coll.aggregate([{$sort: {x: 1}}, {$changeStream: {}}]));
-assert.contains(error.code, [40602, 50988], "Unexpected error: " + tojson(error));
+assert.contains(error.code, [40602], "Unexpected error: " + tojson(error));
 
 error = assert.throws(
     () => coll.aggregate([{$sort: {x: 1}}, {$changeStream: {}}], {allowDiskUse: true}));
-assert.contains(error.code, [40602, 50988], "Unexpected error: " + tojson(error));
+assert.contains(error.code, [40602], "Unexpected error: " + tojson(error));
 
 error = assert.throws(() => coll.aggregate([{$group: {_id: "$x"}}, {$changeStream: {}}]));
-assert.contains(error.code, [40602, 50988], "Unexpected error: " + tojson(error));
+assert.contains(error.code, [40602], "Unexpected error: " + tojson(error));
 
 // This one has a different error code because of conflicting host type requirements: the $group
 // needs to merge on a shard, but the $changeStream needs to merge on mongos. This doesn't

jstests/libs/override_methods/continuous_stepdown.js

@@ -176,6 +176,9 @@ ContinuousStepdown.configure = function(stepdownOptions,
      * Overrides the ReplSetTest constructor to start the continuous primary stepdown thread.
      */
     ReplSetTest = function ReplSetTestWithContinuousPrimaryStepdown() {
+        // Preserve the original set of nodeOptions passed to the constructor.
+        const origNodeOpts = Object.assign({}, (arguments[0] && arguments[0].nodeOptions) || {});
+
         // Construct the original object
         originalReplSetTest.apply(this, arguments);
 
@@ -185,24 +188,31 @@ ContinuousStepdown.configure = function(stepdownOptions,
         const _originalAwaitLastOpCommitted = this.awaitLastOpCommitted;
 
         /**
-         * Overrides startSet call to increase logging verbosity.
+         * Overrides startSet call to increase logging verbosity. Ensure that we only override the
+         * 'logComponentVerbosity' server parameter, but retain any other parameters that were
+         * supplied during ReplSetTest construction.
          */
         this.startSet = function() {
-            let options = arguments[0] || {};
-
-            if (typeof (options.setParameter) === "string") {
-                var eqIdx = options.setParameter.indexOf("=");
-                if (eqIdx != -1) {
-                    var param = options.setParameter.substring(0, eqIdx);
-                    var value = options.setParameter.substring(eqIdx + 1);
-                    options.setParameter = {};
-                    options.setParameter[param] = value;
+            // Helper function to convert a string representation of setParameter to object form.
+            function setParamToObj(setParam) {
+                if (typeof (setParam) === "string") {
+                    var eqIdx = setParam.indexOf("=");
+                    if (eqIdx != -1) {
+                        var param = setParam.substring(0, eqIdx);
+                        var value = setParam.substring(eqIdx + 1);
+                        return {[param]: value};
+                    }
                 }
+                return Object.assign({}, setParam || {});
             }
+
+            const options = arguments[0] || {};
+
+            options.setParameter = Object.assign(setParamToObj(origNodeOpts.setParameter),
+                                                 setParamToObj(options.setParameter),
+                                                 {logComponentVerbosity: verbositySetting});
             arguments[0] = options;
 
-            options.setParameter = options.setParameter || {};
-            options.setParameter.logComponentVerbosity = verbositySetting;
             return _originalStartSetFn.apply(this, arguments);
         };
 

jstests/sharding/change_stream_show_migration_events.js

@@ -1,4 +1,4 @@
-// Tests the undocumented 'showChunkMigrations' option for change streams.
+// Tests the undocumented 'showMigrationEvents' option for change streams.
 //
 // This test is connecting directly to a shard, and change streams require the getMore command.
 // @tags: [requires_find_command, uses_change_streams]

jstests/sharding/change_streams_new_shard_new_database.js

@@ -0,0 +1,88 @@
+/**
+ * Tests that existing whole-cluster, whole-db and single-collection $changeStreams correctly pick
+ * up events on a newly-added shard when a new unsharded collection is created on it. Exercises the
+ * fix for SERVER-42723.
+ * Tagging as 'requires_find_command' to ensure that this test is not run in the legacy protocol
+ * passthroughs. Legacy getMore fails in cases where it is run on a database or collection which
+ * does not yet exist.
+ * @tags: [uses_change_streams, requires_sharding, requires_find_command]
+ */
+(function() {
+
+"use strict";
+
+const rsNodeOptions = {
+    setParameter: {writePeriodicNoops: true, periodicNoopIntervalSecs: 1}
+};
+const st =
+    new ShardingTest({shards: 1, mongos: 1, rs: {nodes: 1}, other: {rsOptions: rsNodeOptions}});
+
+// We require one 'test' database and a second 'other' database.
+const oldShardDB = st.s.getDB(jsTestName() + "_other");
+const newShardDB = st.s.getDB(jsTestName());
+
+const configDB = st.s.getDB("config");
+const adminDB = st.s.getDB("admin");
+
+const oldShardColl = oldShardDB.coll;
+const newShardColl = newShardDB.test;
+
+// Helper function to add a new ReplSetTest shard into the cluster.
+function addShardToCluster(shardName) {
+    const replTest = new ReplSetTest({name: shardName, nodes: 1, nodeOptions: rsNodeOptions});
+    replTest.startSet({shardsvr: ""});
+    replTest.initiate();
+    assert.commandWorked(st.s.adminCommand({addShard: replTest.getURL(), name: shardName}));
+    return replTest;
+}
+
+// Helper function to confirm that a stream sees an expected sequence of documents.
+function assertAllEventsObserved(changeStream, expectedDocs) {
+    for (let expectedDoc of expectedDocs) {
+        assert.soon(() => changeStream.hasNext());
+        const nextEvent = changeStream.next();
+        assert.docEq(nextEvent.fullDocument, expectedDoc);
+    }
+}
+
+// Open a whole-db change stream on the as yet non-existent database.
+const wholeDBCS = newShardDB.watch();
+
+// Open a single-collection change stream on a namespace within the non-existent database.
+const singleCollCS = newShardColl.watch();
+
+// Open a whole-cluster stream on the deployment.
+const wholeClusterCS = adminDB.aggregate([{$changeStream: {allChangesForCluster: true}}]);
+
+// Insert some data into the 'other' database on the only existing shard. This should ensure that
+// the primary shard of the test database will be created on the second shard, after it is added.
+const insertedDocs = Array.from({length: 20}, (_, i) => ({_id: i}));
+assert.commandWorked(oldShardColl.insert(insertedDocs));
+
+// Verify that the whole-cluster stream sees all these events.
+assertAllEventsObserved(wholeClusterCS, insertedDocs);
+
+// Verify that the other two streams did not see any of the insertions on the 'other' collection.
+for (let csCursor of [wholeDBCS, singleCollCS]) {
+    assert(!csCursor.hasNext());
+}
+
+// Now add a new shard into the cluster...
+const newShard1 = addShardToCluster("newShard1");
+
+// ... create a new database and collection, and verify that they were placed on the new shard....
+assert.commandWorked(newShardDB.runCommand({create: newShardColl.getName()}));
+assert(configDB.databases.findOne({_id: newShardDB.getName(), primary: "newShard1"}));
+
+// ... insert some documents into the new, unsharded collection on the new shard...
+assert.commandWorked(newShardColl.insert(insertedDocs));
+
+// ... and confirm that all the pre-existing streams see all of these events.
+for (let csCursor of [singleCollCS, wholeDBCS, wholeClusterCS]) {
+    assertAllEventsObserved(csCursor, insertedDocs);
+}
+
+// Stop the new shard manually since the ShardingTest doesn't know anything about it.
+st.stop();
+newShard1.stopSet();
+})();

jstests/sharding/change_streams_unsharded_becomes_sharded.js

@@ -97,9 +97,7 @@ function testUnshardedBecomesSharded(collToWatch) {
     ];
 
     // Verify that the cursor on the original shard is still valid and sees new inserted
-    // documents. The 'documentKey' field should now include the shard key, even before a
-    // 'kNewShardDetected' operation has been generated by the migration of a chunk to a new
-    // shard.
+    // documents. The 'documentKey' field should now include the shard key.
     assert.commandWorked(mongosColl.insert({_id: 1, x: 1}));
     assert.commandWorked(mongosCollOther.insert({_id: 1, y: 1}));
     cst.assertNextChangesEqual({cursor: cursor, expectedChanges: [postShardCollectionChanges[0]]});

src/mongo/db/pipeline/document_source_change_stream.cpp

@@ -244,8 +244,7 @@ std::string DocumentSourceChangeStream::getNsRegexForChangeStream(const Namespac
 
 BSONObj DocumentSourceChangeStream::buildMatchFilter(
     const boost::intrusive_ptr<ExpressionContext>& expCtx,
-    Timestamp startFrom,
-    bool startFromInclusive,
+    Timestamp startFromInclusive,
     bool showMigrationEvents) {
     auto nss = expCtx->ns;
 
@@ -297,6 +296,11 @@ BSONObj DocumentSourceChangeStream::buildMatchFilter(
     // 2.1) Normal CRUD ops.
     auto normalOpTypeMatch = BSON("op" << NE << "n");
 
+    // TODO SERVER-44039: we continue to generate 'kNewShardDetected' events for compatibility
+    // with 4.2, even though we no longer rely on them to detect new shards. We may wish to remove
+    // this mechanism in 4.6, or retain it for future cases where a change stream is targeted to a
+    // subset of shards. See SERVER-44039 for details.
+
     // 2.2) A chunk gets migrated to a new shard that doesn't have any chunks.
     auto chunkMigratedNewShardMatch = BSON("op"
                                            << "n"
@@ -326,7 +330,7 @@ BSONObj DocumentSourceChangeStream::buildMatchFilter(
     // Only include CRUD operations tagged "fromMigrate" when the "showMigrationEvents" option is
     // set - exempt all other operations and commands with that tag. Include the resume token, if
     // resuming, so we can verify it was still present in the oplog.
-    return BSON("$and" << BSON_ARRAY(BSON("ts" << (startFromInclusive ? GTE : GT) << startFrom)
+    return BSON("$and" << BSON_ARRAY(BSON("ts" << GTE << startFromInclusive)
                                      << BSON(OR(opMatch, commandAndApplyOpsMatch))));
 }
 
@@ -388,6 +392,7 @@ list<intrusive_ptr<DocumentSource>> buildPipeline(const intrusive_ptr<Expression
         }
     }
 
+    // If we do not have a 'resumeAfter' starting point, check for 'startAtOperationTime'.
     if (auto startAtOperationTime = spec.getStartAtOperationTime()) {
         uassert(40674,
                 "Only one type of resume option is allowed, but multiple were found.",
@@ -396,33 +401,37 @@ list<intrusive_ptr<DocumentSource>> buildPipeline(const intrusive_ptr<Expression
         resumeStage = DocumentSourceShardCheckResumability::create(expCtx, *startFrom);
     }
 
-    // There might not be a starting point if we're on mongos, otherwise we should either have a
-    // 'resumeAfter' starting point, or should start from the latest majority committed operation.
+    // We can only run on a replica set, or through mongoS. Confirm that this is the case.
     auto replCoord = repl::ReplicationCoordinator::get(expCtx->opCtx);
     uassert(
         40573,
         "The $changeStream stage is only supported on replica sets",
         expCtx->inMongos ||
             (replCoord &&
              replCoord->getReplicationMode() == repl::ReplicationCoordinator::Mode::modeReplSet));
-    if (!startFrom && !expCtx->inMongos) {
-        startFrom = replCoord->getMyLastAppliedOpTime().getTimestamp();
+
+    // If we do not have an explicit starting point, we should start from the latest majority
+    // committed operation. If we are on mongoS and do not have a starting point, set it to the
+    // current clusterTime so that all shards start in sync. We always start one tick beyond the
+    // most recent operation, to ensure that the stream does not return it.
+    if (!startFrom) {
+        const auto currentTime = !expCtx->inMongos
+            ? LogicalTime{replCoord->getMyLastAppliedOpTime().getTimestamp()}
+            : LogicalClock::get(expCtx->opCtx)->getClusterTime();
+        startFrom = currentTime.addTicks(1).asTimestamp();
     }
 
-    if (startFrom) {
-        const bool startFromInclusive = (resumeStage != nullptr);
-        stages.push_back(DocumentSourceOplogMatch::create(
-            DocumentSourceChangeStream::buildMatchFilter(
-                expCtx, *startFrom, startFromInclusive, showMigrationEvents),
-            expCtx));
-
-        // If we haven't already populated the initial PBRT, then we are starting from a specific
-        // timestamp rather than a resume token. Initialize the PBRT to a high water mark token.
-        if (expCtx->initialPostBatchResumeToken.isEmpty()) {
-            Timestamp startTime{startFrom->getSecs(), startFrom->getInc() + (!startFromInclusive)};
-            expCtx->initialPostBatchResumeToken =
-                ResumeToken::makeHighWaterMarkToken(startTime).toDocument().toBson();
-        }
+    // We must always build the DSOplogMatch stage even on mongoS, since our validation logic relies
+    // upon the fact that it is always the first stage in the pipeline.
+    stages.push_back(DocumentSourceOplogMatch::create(
+        DocumentSourceChangeStream::buildMatchFilter(expCtx, *startFrom, showMigrationEvents),
+        expCtx));
+
+    // If we haven't already populated the initial PBRT, then we are starting from a specific
+    // timestamp rather than a resume token. Initialize the PBRT to a high water mark token.
+    if (expCtx->initialPostBatchResumeToken.isEmpty()) {
+        expCtx->initialPostBatchResumeToken =
+            ResumeToken::makeHighWaterMarkToken(*startFrom).toDocument().toBson();
     }
 
     // Obtain the current FCV and use it to create the DocumentSourceChangeStreamTransform stage.
@@ -516,12 +525,14 @@ void DocumentSourceChangeStream::assertIsLegalSpecification(
                 (expCtx->ns.isAdminDB() && expCtx->ns.isCollectionlessAggregateNS()));
 
     // Prevent $changeStream from running on internal databases. A stream may run against the
-    // 'admin' database iff 'allChangesForCluster' is true.
+    // 'admin' database iff 'allChangesForCluster' is true. A stream may run against the 'config'
+    // database iff 'allowToRunOnConfigDB' is true.
+    const bool isNotBannedInternalDB =
+        !expCtx->ns.isLocal() && (!expCtx->ns.isConfigDB() || spec.getAllowToRunOnConfigDB());
     uassert(ErrorCodes::InvalidNamespace,
             str::stream() << "$changeStream may not be opened on the internal " << expCtx->ns.db()
                           << " database",
-            expCtx->ns.isAdminDB() ? spec.getAllChangesForCluster()
-                                   : (!expCtx->ns.isLocal() && !expCtx->ns.isConfigDB()));
+            expCtx->ns.isAdminDB() ? spec.getAllChangesForCluster() : isNotBannedInternalDB);
 
     // Prevent $changeStream from running on internal collections in any database.
     uassert(ErrorCodes::InvalidNamespace,

src/mongo/db/pipeline/document_source_change_stream.h

@@ -166,7 +166,6 @@ class DocumentSourceChangeStream final {
      */
     static BSONObj buildMatchFilter(const boost::intrusive_ptr<ExpressionContext>& expCtx,
                                     Timestamp startFrom,
-                                    bool startFromInclusive,
                                     bool showMigrationEvents);
 
     /**

src/mongo/db/pipeline/document_source_change_stream.idl

@@ -109,3 +109,11 @@ structs:
                              deletes may appear that do not reflect actual deletions or insertions
                              of data. Instead they reflect this data moving from one shard to
                              another.
+            allowToRunOnConfigDB:
+                cpp_name: allowToRunOnConfigDB
+                type: bool
+                default: false
+                description: A flag indicating whether the change stream may be opened on the
+                             'config' database, which is usually banned. This flag is used
+                             internally to allow mongoS to open a stream on 'config.shards', in
+                             order to monitor for the addition of new shards to the cluster.

src/mongo/db/pipeline/document_source_change_stream_transform.cpp

@@ -362,14 +362,11 @@ Value DocumentSourceChangeStreamTransform::serialize(
         changeStreamOptions[DocumentSourceChangeStreamSpec::kStartAfterFieldName].missing()) {
         MutableDocument newChangeStreamOptions(changeStreamOptions);
 
-        // Use the current cluster time plus 1 tick since the oplog query will include all
-        // operations/commands equal to or greater than the 'startAtOperationTime' timestamp. In
-        // particular, avoid including the last operation that went through mongos in an attempt to
-        // match the behavior of a replica set more closely.
-        auto clusterTime = LogicalClock::get(pExpCtx->opCtx)->getClusterTime();
-        clusterTime.addTicks(1);
-        newChangeStreamOptions[DocumentSourceChangeStreamSpec::kStartAtOperationTimeFieldName] =
-            Value(clusterTime.asTimestamp());
+        // Configure the serialized $changeStream to start from the initial high-watermark
+        // postBatchResumeToken which we generated while parsing the $changeStream pipeline.
+        invariant(!pExpCtx->initialPostBatchResumeToken.isEmpty());
+        newChangeStreamOptions[DocumentSourceChangeStreamSpec::kResumeAfterFieldName] =
+            Value(pExpCtx->initialPostBatchResumeToken);
         changeStreamOptions = newChangeStreamOptions.freeze();
     }
     return Value(Document{{getSourceName(), changeStreamOptions}});