mongo/jstests/sharding/migration_with_source_ops.js

//
// Tests during chunk migration that the recipient does not receive out of range operations from
// the donor.
//
// Pauses the migration on the recipient shard after the initial data chunk cloning is finished.
// This allows time for the donor shard to perform inserts/deletes/updates, half of which are on
// the migrating chunk. The recipient is then set to continue, collecting the mods from the
// donor's transfer mods log, and finishes the migration. A failpoint is set prior to resuming
// in the recipient shard to fail if it receives an out of chunk range insert/delete/update from
// the donor's transfer mods log.
//
// The idea is that the recipient shard should not be collecting inserts/deletes/updates from
// the donor shard's transfer mods log that are not in range and will unnecessarily prevent the
// migration from finishing: the migration can only end when donor's log of mods for the migrating
// chunk is empty.
//

import {
    migrateStepNames,
    moveChunkParallel,
    pauseMigrateAtStep,
    unpauseMigrateAtStep,
    waitForMigrateStep,
} from "jstests/libs/chunk_manipulation_util.js";
import {ShardingTest} from "jstests/libs/shardingtest.js";

let staticMongod = MongoRunner.runMongod({});

/**
 * Start up new sharded cluster, stop balancer that would interfere in manual chunk management.
 */

let st = new ShardingTest({shards: 2, mongos: 1});
st.stopBalancer();

let mongos = st.s0,
    admin = mongos.getDB("admin"),
    dbName = "testDB",
    ns = dbName + ".foo",
    coll = mongos.getCollection(ns),
    donor = st.shard0,
    recipient = st.shard1,
    donorColl = donor.getCollection(ns),
    recipientColl = recipient.getCollection(ns);

/**
 * Exable sharding, and split collection into two chunks.
 */

// Two chunks
// Donor:     [0, 20) [20, 40)
// Recipient:
jsTest.log("Enabling sharding of the collection and pre-splitting into two chunks....");
assert.commandWorked(admin.runCommand({enableSharding: dbName, primaryShard: st.shard0.shardName}));
assert.commandWorked(admin.runCommand({shardCollection: ns, key: {a: 1}}));
assert.commandWorked(admin.runCommand({split: ns, middle: {a: 20}}));

/**
 * Insert data into collection
 */

// 10 documents in each chunk on the donor
jsTest.log("Inserting 20 docs into donor shard, 10 in each chunk....");
for (var i = 0; i < 10; ++i) assert.commandWorked(coll.insert({a: i}));
for (var i = 20; i < 30; ++i) assert.commandWorked(coll.insert({a: i}));
assert.eq(20, coll.count());

/**
 * Set failpoints. Recipient will crash if an out of chunk range data op is
 * received from donor. Recipient will pause migration after cloning chunk data from donor,
 * before checking transfer mods log on donor.
 */

jsTest.log("Setting failpoint failMigrationReceivedOutOfRangeOperation");
assert.commandWorked(
    recipient
        .getDB("admin")
        .runCommand({configureFailPoint: "failMigrationReceivedOutOfRangeOperation", mode: "alwaysOn"}),
);

jsTest.log("Setting chunk migration recipient failpoint so that it pauses after bulk clone step");
pauseMigrateAtStep(recipient, migrateStepNames.cloned);

/**
 * Start a moveChunk in the background. Move chunk [20, 40), which has 10 docs in the
 * range, from shard 0 (donor) to shard 1 (recipient). Migration will pause after
 * cloning step (when it reaches the recipient failpoint).
 */

// Donor:     [0, 20)
// Recipient:    [20, 40)
jsTest.log("Starting migration, pause after cloning...");
let joinMoveChunk = moveChunkParallel(staticMongod, st.s0.host, {a: 20}, null, coll.getFullName(), st.shard1.shardName);

/**
 * Wait for recipient to finish cloning step.
 * THEN delete 10 documents on the donor shard, 5 in the migrating chunk and 5 in the remaining
 *chunk.
 * AND insert 2 documents on the donor shard, 1 in the migrating chunk and 1 in the remaining
 *chunk.
 * AND update 2 documents on the donor shard, 1 in the migrating chunk and 1 in the remaining
 *chunk.
 *
 * This will populate the migration transfer mods log, which the recipient will collect when it
 * is unpaused.
 */

waitForMigrateStep(recipient, migrateStepNames.cloned);

jsTest.log("Deleting 5 docs from each chunk, migrating chunk and remaining chunk...");
assert.commandWorked(coll.remove({$and: [{a: {$gte: 5}}, {a: {$lt: 25}}]}));

jsTest.log("Inserting 1 in the migrating chunk range and 1 in the remaining chunk range...");
assert.commandWorked(coll.insert({a: 10}));
assert.commandWorked(coll.insert({a: 30}));

jsTest.log("Updating 1 in the migrating chunk range and 1 in the remaining chunk range...");
assert.commandWorked(coll.update({a: 0}, {a: 0, updatedData: "updated"}));
assert.commandWorked(coll.update({a: 25}, {a: 25, updatedData: "updated"}));

/**
 * Finish migration. Unpause recipient migration, wait for it to collect
 * the new ops from the donor shard's migration transfer mods log, and finish.
 */

jsTest.log("Continuing and finishing migration...");
unpauseMigrateAtStep(recipient, migrateStepNames.cloned);
joinMoveChunk();

/**
 * Check documents are where they should be: 6 docs in each shard's respective chunk.
 */

jsTest.log("Checking that documents are on the shards they should be...");
assert.eq(6, donorColl.count());
assert.eq(6, recipientColl.count());
assert.eq(12, coll.count());

/**
 * Check that the updated documents are where they should be, one on each shard.
 */

jsTest.log("Checking that documents were updated correctly...");
let donorCollUpdatedNum = donorColl.find({updatedData: "updated"}).count();
assert.eq(1, donorCollUpdatedNum, "Update failed on donor shard during migration!");
let recipientCollUpdatedNum = recipientColl.find({updatedData: "updated"}).count();
assert.eq(1, recipientCollUpdatedNum, "Update failed on recipient shard during migration!");

jsTest.log("DONE!");
MongoRunner.stopMongod(staticMongod);
st.stop();