SERVER-21678 setting fromMigrate flag for deletes in oplog

12 files changed, 212 insertions, 11 deletions

diff --git a/buildscripts/resmokeconfig/suites/sharding_auth.yml b/buildscripts/resmokeconfig/suites/sharding_auth.yml
index 578632650e0..39f36e53586 100644
--- a/buildscripts/resmokeconfig/suites/sharding_auth.yml
+++ b/buildscripts/resmokeconfig/suites/sharding_auth.yml
@@ -24,6 +24,7 @@ selector:
     - jstests/sharding/cleanup_orphaned_cmd_during_movechunk.js         # SERVER-21713
     - jstests/sharding/cleanup_orphaned_cmd_during_movechunk_hashed.js  # SERVER-21713
     - jstests/sharding/migration_with_source_deletes.js                 # SERVER-21713
+    - jstests/sharding/migration_sets_fromMigrate_flag.js               # SERVER-21713
 
 executor:
   js_test:
diff --git a/buildscripts/resmokeconfig/suites/sharding_auth_audit.yml b/buildscripts/resmokeconfig/suites/sharding_auth_audit.yml
index bed76c61261..f5a65432ce7 100644
--- a/buildscripts/resmokeconfig/suites/sharding_auth_audit.yml
+++ b/buildscripts/resmokeconfig/suites/sharding_auth_audit.yml
@@ -24,6 +24,7 @@ selector:
     - jstests/sharding/cleanup_orphaned_cmd_during_movechunk.js         # SERVER-21713
     - jstests/sharding/cleanup_orphaned_cmd_during_movechunk_hashed.js  # SERVER-21713
     - jstests/sharding/migration_with_source_deletes.js                 # SERVER-21713
+    - jstests/sharding/migration_sets_fromMigrate_flag.js               # SERVER-21713
 
 executor:
   js_test:
diff --git a/jstests/sharding/migration_sets_fromMigrate_flag.js b/jstests/sharding/migration_sets_fromMigrate_flag.js
new file mode 100644
index 00000000000..1be4a8324cc
--- /dev/null
+++ b/jstests/sharding/migration_sets_fromMigrate_flag.js
@@ -0,0 +1,166 @@
+//
+// Tests whether the fromMigrate flag is correctly set during migrations.
+//
+// Tests:
+//      #1 (delete op) fromMigrate is set when recipient shard deletes all documents locally
+//         in the chunk range it is about to receive from the donor shard.
+//      #2 (delete op) fromMigrate is set when the donor shard deletes documents that have
+//         been migrated to another shard.
+//      #3 (insert op) fromMigrate is set when the recipient shard receives chunk migration
+//         data and inserts it.
+//      #4 (update op) fromMigrate is set when an update occurs in the donor shard during
+//         migration and is sent to the recipient via the transfer logs.
+//      #5 fromMigrate is NOT set on donor shard and IS set on the recipient shard when real
+//         delete op is done during chunk migration within the chunk range.
+//
+
+load('./jstests/libs/chunk_manipulation_util.js');
+
+(function() {
+"use strict";
+
+var staticMongod = MongoRunner.runMongod({});  // For startParallelOps.
+
+/**
+ * Start up new sharded cluster, stop balancer that would interfere in manual chunk management.
+ */
+
+var st = new ShardingTest({ shards : 2, mongos : 1,  rs : { nodes : 3 } });
+st.stopBalancer();
+
+var mongos = st.s0,
+    admin = mongos.getDB('admin'),
+    shards = mongos.getCollection('config.shards').find().toArray(),
+    dbName = "testDB",
+    ns = dbName + ".foo",
+    coll = mongos.getCollection(ns),
+    donor = st.shard0,
+    recipient = st.shard1,
+    donorColl = donor.getCollection(ns),
+    recipientColl = recipient.getCollection(ns),
+    donorLocal = donor.getDB('local'),
+    recipientLocal = recipient.getDB('local');
+
+// Two chunks
+// Donor:     [0, 2) [2, 5)
+// Recipient:
+jsTest.log('Enable sharding of the collection and pre-split into two chunks....');
+
+assert.commandWorked(admin.runCommand({enableSharding: dbName}));
+st.ensurePrimaryShard(dbName, shards[0]._id);
+assert.commandWorked(donorColl.createIndex({_id: 1}));
+assert.commandWorked(admin.runCommand({shardCollection: ns, key: {_id: 1}}));
+assert.commandWorked(admin.runCommand({split: ns, middle: {_id: 2}}));
+
+// 6 documents,
+//        donor: 2 in the first chunk, 3 in the second.
+//    recipient: 1 document (shardkey overlaps with a doc in second chunk of donor shard)
+jsTest.log('Inserting 5 docs into donor shard, 1 doc into the recipient shard....');
+
+for (var i = 0; i < 5; ++i)
+    assert.writeOK(coll.insert({_id: i}));
+assert.eq(5, donorColl.count());
+
+for (var i = 2; i < 3; ++i)
+    assert.writeOK(recipientColl.insert({_id: i}));
+assert.eq(1, recipientColl.count());
+
+/**
+ * Set failpoint: recipient will pause migration after cloning chunk data from donor,
+ * before checking transfer mods log on donor.
+ */
+
+jsTest.log('setting recipient failpoint cloned');
+pauseMigrateAtStep(recipient, migrateStepNames.cloned);
+
+/**
+ * Start moving chunk [2, 5) from donor shard to recipient shard, run in the background.
+ */
+
+// Donor:     [0, 2)
+// Recipient:    [2, 5)
+jsTest.log('Starting chunk migration, pause after cloning...');
+
+var joinMoveChunk = moveChunkParallel(
+    staticMongod,
+    st.s0.host,
+    {_id: 2},
+    null,
+    coll.getFullName(),
+    shards[1]._id);
+
+/**
+ * Wait for recipient to finish cloning.
+ * THEN update 1 document {_id: 3} on donor within the currently migrating chunk.
+ * AND delete 1 document {_id: 4} on donor within the currently migrating chunk.
+ */
+
+waitForMigrateStep(recipient, migrateStepNames.cloned);
+
+jsTest.log('Update 1 doc and delete 1 doc on donor within the currently migrating chunk...');
+
+assert.writeOK(coll.update({_id: 3}, {_id: 3, a: "updated doc"}));
+assert.writeOK(coll.remove({_id: 4}));
+
+/**
+ * Finish migration. Unpause recipient migration, wait for it to collect
+ * the transfer mods log from donor and finish migration.
+ */
+
+jsTest.log('Continuing and finishing migration...');
+unpauseMigrateAtStep(recipient, migrateStepNames.cloned);
+joinMoveChunk();
+
+/**
+ * Check documents are where they should be: 2 docs in donor chunk, 2 docs in recipient chunk
+ * (because third doc in recipient shard's chunk got deleted on the donor shard during migration).
+ */
+
+jsTest.log('Checking that documents are on the shards they should be...');
+
+assert.eq(2, recipientColl.count(), "Recipient shard doesn't have exactly 2 documents!");
+assert.eq(2, donorColl.count(), "Donor shard doesn't have exactly 2 documents!");
+assert.eq(4, coll.count(), "Collection total is not 4!");
+
+/**
+ * Check that the fromMigrate flag has been set correctly in donor and recipient oplogs,
+ */
+
+jsTest.log('Checking donor and recipient oplogs for correct fromMigrate flags...');
+
+var donorOplogRes = donorLocal.oplog.rs.find(
+                        {op: 'd', fromMigrate: true, 'o._id': 2}).count();
+assert.eq(1, donorOplogRes, "fromMigrate flag wasn't set on the donor shard's oplog for " +
+                            "migrating delete op on {_id: 2}! Test #2 failed.");
+
+donorOplogRes = donorLocal.oplog.rs.find(
+                        {op: 'd', fromMigrate: {$exists: false}, 'o._id': 4}).count();
+assert.eq(1, donorOplogRes, "Real delete of {_id: 4} on donor shard incorrectly set the " +
+                            "fromMigrate flag in the oplog! Test #5 failed.");
+
+var recipientOplogRes = recipientLocal.oplog.rs.find(
+                        {op: 'i', fromMigrate: true, 'o._id': 2}).count();
+assert.eq(1, recipientOplogRes, "fromMigrate flag wasn't set on the recipient shard's " +
+                                "oplog for migrating insert op on {_id: 2}! Test #3 failed.");
+
+recipientOplogRes = recipientLocal.oplog.rs.find(
+                        {op: 'd', fromMigrate: true, 'o._id': 2}).count();
+assert.eq(1, recipientOplogRes, "fromMigrate flag wasn't set on the recipient shard's " +
+                                "oplog for delete op on the old {_id: 2} that overlapped " +
+                                "with the chunk about to be copied! Test #1 failed.");
+
+recipientOplogRes = recipientLocal.oplog.rs.find(
+                        {op: 'u', fromMigrate: true, 'o._id': 3}).count();
+assert.eq(1, recipientOplogRes, "fromMigrate flag wasn't set on the recipient shard's " +
+                                "oplog for update op on {_id: 3}! Test #4 failed.");
+
+recipientOplogRes = recipientLocal.oplog.rs.find(
+                        {op: 'd', fromMigrate: true, 'o._id': 4}).count();
+assert.eq(1, recipientOplogRes, "fromMigrate flag wasn't set on the recipient shard's " +
+                                "oplog for delete op on {_id: 4} that occurred during " +
+                                "migration! Test #5 failed.");
+
+jsTest.log('DONE!');
+st.stop();
+
+})()
diff --git a/src/mongo/db/catalog/collection.cpp b/src/mongo/db/catalog/collection.cpp
index 15ea17f585b..10d8e24ea4f 100644
--- a/src/mongo/db/catalog/collection.cpp
+++ b/src/mongo/db/catalog/collection.cpp
@@ -478,10 +478,8 @@ Status Collection::aboutToDeleteCapped(OperationContext* txn,
     return Status::OK();
 }
 
-void Collection::deleteDocument(OperationContext* txn,
-                                const RecordId& loc,
-                                bool cappedOK,
-                                bool noWarn) {
+void Collection::deleteDocument(
+    OperationContext* txn, const RecordId& loc, bool fromMigrate, bool cappedOK, bool noWarn) {
     if (isCapped() && !cappedOK) {
         log() << "failing remove on a capped ns " << _ns << endl;
         uasserted(10089, "cannot remove from a capped collection");
@@ -500,7 +498,7 @@ void Collection::deleteDocument(OperationContext* txn,
 
     _recordStore->deleteRecord(txn, loc);
 
-    opObserver->onDelete(txn, ns(), std::move(deleteState));
+    opObserver->onDelete(txn, ns(), std::move(deleteState), fromMigrate);
 }
 
 Counter64 moveCounter;
diff --git a/src/mongo/db/catalog/collection.h b/src/mongo/db/catalog/collection.h
index 5db3fe09984..261ac42bf97 100644
--- a/src/mongo/db/catalog/collection.h
+++ b/src/mongo/db/catalog/collection.h
@@ -243,8 +243,20 @@ public:
      */
     std::vector<std::unique_ptr<RecordCursor>> getManyCursors(OperationContext* txn) const;
 
+    /**
+     * Deletes the document with the given RecordId from the collection.
+     *
+     * 'fromMigrate' indicates whether the delete was induced by a chunk migration, and
+     * so should be ignored by the user as an internal maintenance operation and not a
+     * real delete.
+     * 'loc' key to uniquely identify a record in a collection.
+     * 'cappedOK' if true, allows deletes on capped collections (Cloner::copyDB uses this).
+     * 'noWarn' if unindexing the record causes an error, if noWarn is true the error
+     * will not be logged.
+     */
     void deleteDocument(OperationContext* txn,
                         const RecordId& loc,
+                        bool fromMigrate = false,
                         bool cappedOK = false,
                         bool noWarn = false);
 
diff --git a/src/mongo/db/cloner.cpp b/src/mongo/db/cloner.cpp
index eec4a83e1db..feb93876365 100644
--- a/src/mongo/db/cloner.cpp
+++ b/src/mongo/db/cloner.cpp
@@ -605,7 +605,7 @@ Status Cloner::copyDb(OperationContext* txn,
                 // dupsAllowed in IndexCatalog::_unindexRecord and SERVER-17487.
                 for (set<RecordId>::const_iterator it = dups.begin(); it != dups.end(); ++it) {
                     WriteUnitOfWork wunit(txn);
-                    c->deleteDocument(txn, *it, true, true);
+                    c->deleteDocument(txn, *it, false, true, true);
                     wunit.commit();
                 }
 
diff --git a/src/mongo/db/dbhelpers.cpp b/src/mongo/db/dbhelpers.cpp
index b68c232048f..d1541c2409c 100644
--- a/src/mongo/db/dbhelpers.cpp
+++ b/src/mongo/db/dbhelpers.cpp
@@ -429,7 +429,7 @@ long long Helpers::removeRange(OperationContext* txn,
             if (callback)
                 callback->goingToDelete(obj);
 
-            collection->deleteDocument(txn, rloc);
+            collection->deleteDocument(txn, rloc, fromMigrate);
             wuow.commit();
             numDeleted++;
         }
diff --git a/src/mongo/db/exec/delete.cpp b/src/mongo/db/exec/delete.cpp
index 0e5f72cedc3..5a070fcdc0b 100644
--- a/src/mongo/db/exec/delete.cpp
+++ b/src/mongo/db/exec/delete.cpp
@@ -220,7 +220,7 @@ PlanStage::StageState DeleteStage::work(WorkingSetID* out) {
         // Do the write, unless this is an explain.
         if (!_params.isExplain) {
             WriteUnitOfWork wunit(getOpCtx());
-            _collection->deleteDocument(getOpCtx(), rloc);
+            _collection->deleteDocument(getOpCtx(), rloc, _params.fromMigrate);
             wunit.commit();
         }
 
diff --git a/src/mongo/db/op_observer.h b/src/mongo/db/op_observer.h
index e9c18f4bbe7..406402b093d 100644
--- a/src/mongo/db/op_observer.h
+++ b/src/mongo/db/op_observer.h
@@ -68,10 +68,19 @@ public:
                    bool fromMigrate = false);
     void onUpdate(OperationContext* txn, oplogUpdateEntryArgs args);
     DeleteState aboutToDelete(OperationContext* txn, const NamespaceString& ns, const BSONObj& doc);
+    /**
+     * Handles logging before document is deleted.
+     *
+     * "ns" name of the collection from which deleteState.idDoc will be deleted.
+     * "deleteState" holds information about the deleted document.
+     * "fromMigrate" indicates whether the delete was induced by a chunk migration, and
+     * so should be ignored by the user as an internal maintenance operation and not a
+     * real delete.
+     */
     void onDelete(OperationContext* txn,
                   const NamespaceString& ns,
                   DeleteState deleteState,
-                  bool fromMigrate = false);
+                  bool fromMigrate);
     void onOpMessage(OperationContext* txn, const BSONObj& msgObj);
     void onCreateCollection(OperationContext* txn,
                             const NamespaceString& collectionName,
diff --git a/src/mongo/db/s/migration_source_manager.cpp b/src/mongo/db/s/migration_source_manager.cpp
index 59eb96222fb..09680bae2a3 100644
--- a/src/mongo/db/s/migration_source_manager.cpp
+++ b/src/mongo/db/s/migration_source_manager.cpp
@@ -234,8 +234,8 @@ void MigrationSourceManager::logOp(OperationContext* txn,
         // won't be transferred to the recipient shard. Also ignore ops from
         // _migrateClone and _transferMods since it is impossible to move a chunk
         // to self.
-        // Also ignore out of range deletes when migrating (notInActiveChunk is set in
-        // OpObserver::onDelete)
+        // Also ignore out of range deletes when migrating a chunk (is set
+        // in OpObserver::onDelete)
         return;
     }
 
diff --git a/src/mongo/db/s/migration_source_manager.h b/src/mongo/db/s/migration_source_manager.h
index 8c6251fe49b..076c14fa9f0 100644
--- a/src/mongo/db/s/migration_source_manager.h
+++ b/src/mongo/db/s/migration_source_manager.h
@@ -71,6 +71,10 @@ public:
                BSONObj* patt,
                bool notInActiveChunk);
 
+    /**
+     * Determines whether the given document 'doc' in namespace 'ns' is within the range
+     * of a currently migrating chunk.
+     */
     bool isInMigratingChunk(const NamespaceString& ns, const BSONObj& doc);
 
     /**
diff --git a/src/mongo/s/d_migrate.cpp b/src/mongo/s/d_migrate.cpp
index 6a7abe7f723..9983ccd58df 100644
--- a/src/mongo/s/d_migrate.cpp
+++ b/src/mongo/s/d_migrate.cpp
@@ -430,6 +430,16 @@ public:
 
 }  // namespace
 
+/**
+ * If sharding is enabled, logs the operation for an active migration in the transfer mods log.
+ *
+ * 'ns' name of the collection in which the operation will occur.
+ * 'notInActiveChunk' a true value indicates that either:
+ *      1) the delete is coming from a donor shard in a current chunk migration,
+ *         and so does not need to be entered in this shard's outgoing transfer log.
+ *      2) the document is not within this shard's outgoing chunk migration range,
+ *         and so does not need to be forwarded to the migration recipient via the transfer log.
+ */
 void logOpForSharding(OperationContext* txn,
                       const char* opstr,
                       const char* ns,