diff options
author | Jonathan Abrahams <jonathan@mongodb.com> | 2016-11-16 15:33:31 -0500 |
---|---|---|
committer | Jonathan Abrahams <jonathan@mongodb.com> | 2016-11-16 15:33:31 -0500 |
commit | 1cb28f56b1a02fce7e0eb47c35d006595da7dafb (patch) | |
tree | af9a8bf310f5d3309ccd2b48e25827b8185b6633 /jstests | |
parent | 2fae4242b9e8256da203639895d1ecd3fe8e2794 (diff) | |
download | mongo-1cb28f56b1a02fce7e0eb47c35d006595da7dafb.tar.gz |
SERVER-20447 Add concurrency workload that tests sharding functions:
- mergeChunks
- moveChunk
- splitChunk
Diffstat (limited to 'jstests')
9 files changed, 1025 insertions, 1 deletions
diff --git a/jstests/concurrency/fsm_libs/cluster.js b/jstests/concurrency/fsm_libs/cluster.js index 1194410023b..078f73bcdc5 100644 --- a/jstests/concurrency/fsm_libs/cluster.js +++ b/jstests/concurrency/fsm_libs/cluster.js @@ -40,6 +40,7 @@ var Cluster = function(options) { 'sameDB', 'setupFunctions', 'sharded.enabled', + 'sharded.enableAutoSplit', 'sharded.enableBalancer', 'sharded.numMongos', 'sharded.numShards', @@ -53,6 +54,12 @@ var Cluster = function(options) { tojson(allowedKeys)); }); + options.sharded.enableAutoSplit = options.sharded.enableAutoSplit || false; + assert.eq('boolean', typeof options.sharded.enableAutoSplit); + + options.sharded.enableBalancer = options.sharded.enableBalancer || false; + assert.eq('boolean', typeof options.sharded.enableBalancer); + options.masterSlave = options.masterSlave || false; assert.eq('boolean', typeof options.masterSlave); @@ -193,7 +200,10 @@ var Cluster = function(options) { shards: options.sharded.numShards, mongos: options.sharded.numMongos, verbose: verbosityLevel, - other: {enableBalancer: options.sharded.enableBalancer} + other: { + enableAutoSplit: options.sharded.enableAutoSplit, + enableBalancer: options.sharded.enableBalancer, + } }; // TODO: allow 'options' to specify an 'rs' config @@ -463,6 +473,10 @@ var Cluster = function(options) { return this.isSharded() && options.sharded.enableBalancer; }; + this.isAutoSplitEnabled = function isAutoSplitEnabled() { + return this.isSharded() && options.sharded.enableAutoSplit; + }; + this.validateAllCollections = function validateAllCollections(phase) { assert(initialized, 'cluster must be initialized first'); diff --git a/jstests/concurrency/fsm_libs/parse_config.js b/jstests/concurrency/fsm_libs/parse_config.js index 3c365dc5f4c..f5255f18d97 100644 --- a/jstests/concurrency/fsm_libs/parse_config.js +++ b/jstests/concurrency/fsm_libs/parse_config.js @@ -11,6 +11,7 @@ function parseConfig(config) { 'iterations', 'passConnectionCache', 'setup', + 'skip', 'startState', 'states', 'teardown', @@ -74,6 +75,11 @@ function parseConfig(config) { config.setup = config.setup || function() {}; assert.eq('function', typeof config.setup); + config.skip = config.skip || function() { + return {skip: false}; + }; + assert.eq('function', typeof config.skip); + config.teardown = config.teardown || function() {}; assert.eq('function', typeof config.teardown); diff --git a/jstests/concurrency/fsm_libs/runner.js b/jstests/concurrency/fsm_libs/runner.js index 9fe5f5903b8..ec629245dad 100644 --- a/jstests/concurrency/fsm_libs/runner.js +++ b/jstests/concurrency/fsm_libs/runner.js @@ -353,6 +353,17 @@ var runner = (function() { } } + function shouldSkipWorkload(workload, context, cluster) { + var config = context[workload].config; + var result = config.skip.call(config.data, cluster); + if (result.skip) { + var msg = result.msg || ''; + jsTest.log('Skipping workload: ' + workload + ' ' + msg); + return true; + } + return false; + } + function setupWorkload(workload, context, cluster) { var myDB = context[workload].db; var collName = context[workload].collName; @@ -609,6 +620,11 @@ var runner = (function() { } cluster.setup(); + // Filter out workloads that need to be skipped. + bgWorkloads = + bgWorkloads.filter(workload => !shouldSkipWorkload(workload, bgContext, cluster)); + workloads = workloads.filter(workload => !shouldSkipWorkload(workload, context, cluster)); + // Clean up the state left behind by other tests in the concurrency suite // to avoid having too many open files. diff --git a/jstests/concurrency/fsm_workload_helpers/chunks.js b/jstests/concurrency/fsm_workload_helpers/chunks.js new file mode 100644 index 00000000000..2cdbdcc6e7b --- /dev/null +++ b/jstests/concurrency/fsm_workload_helpers/chunks.js @@ -0,0 +1,188 @@ +'use strict'; + +/** + * Provides wrapper functions that perform exponential backoff and allow for + * acceptable errors to be returned from mergeChunks, moveChunk, and splitChunk + * commands. + * + * Also provides functions to help perform assertions about the state of chunks. + * + * Intended for use by workloads testing sharding (i.e., workloads starting with 'sharded_'). + */ + +load('jstests/concurrency/fsm_workload_helpers/server_types.js'); // for isMongos & isMongod + +var ChunkHelper = (function() { + // exponential backoff + function getNextBackoffSleep(curSleep) { + const MAX_BACKOFF_SLEEP = 5000; // milliseconds + + curSleep *= 2; + return Math.min(curSleep, MAX_BACKOFF_SLEEP); + } + + function runCommandWithRetries(db, cmd, acceptableErrorCodes) { + const INITIAL_BACKOFF_SLEEP = 500; // milliseconds + const MAX_RETRIES = 5; + + var acceptableErrorOccurred = function acceptableErrorOccurred(errorCode, + acceptableErrorCodes) { + return acceptableErrorCodes.indexOf(errorCode) > -1; + }; + + var res; + var retries = 0; + var backoffSleep = INITIAL_BACKOFF_SLEEP; + while (retries < MAX_RETRIES) { + retries++; + res = db.adminCommand(cmd); + // If the command worked, exit the loop early. + if (res.ok) { + return res; + } + // Assert command worked or acceptable error occurred. + var msg = tojson({command: cmd, res: res}); + assertWhenOwnColl(acceptableErrorOccurred(res.code, acceptableErrorCodes), msg); + + // When an acceptable error occurs, sleep and then retry. + sleep(backoffSleep); + backoffSleep = getNextBackoffSleep(backoffSleep); + } + + return res; + } + + function splitChunkAtPoint(db, collName, splitPoint) { + var cmd = {split: db[collName].getFullName(), middle: {_id: splitPoint}}; + var acceptableErrorCodes = [ErrorCodes.LockBusy]; + return runCommandWithRetries(db, cmd, acceptableErrorCodes); + } + + function splitChunkWithBounds(db, collName, bounds) { + var cmd = {split: db[collName].getFullName(), bounds: bounds}; + var acceptableErrorCodes = [ErrorCodes.LockBusy]; + return runCommandWithRetries(db, cmd, acceptableErrorCodes); + } + + function moveChunk(db, collName, bounds, toShard, waitForDelete) { + var cmd = { + moveChunk: db[collName].getFullName(), + bounds: bounds, + to: toShard, + _waitForDelete: waitForDelete + }; + var acceptableErrorCodes = + [ErrorCodes.ConflictingOperationInProgress, ErrorCodes.ChunkRangeCleanupPending]; + return runCommandWithRetries(db, cmd, acceptableErrorCodes); + } + + function mergeChunks(db, collName, bounds) { + var cmd = {mergeChunks: db[collName].getFullName(), bounds: bounds}; + var acceptableErrorCodes = [ErrorCodes.LockBusy]; + return runCommandWithRetries(db, cmd, acceptableErrorCodes); + } + + // Take a set of connections to a shard (replica set or standalone mongod), + // or a set of connections to the config servers, and return a connection + // to any node in the set for which ismaster is true. + function getPrimary(connArr) { + assertAlways(Array.isArray(connArr), 'Expected an array but got ' + tojson(connArr)); + + for (var conn of connArr) { + assert(isMongod(conn.getDB('admin')), tojson(conn) + ' is not to a mongod'); + var res = conn.adminCommand({isMaster: 1}); + assertAlways.commandWorked(res); + + if (res.ismaster) { + return conn; + } + } + assertAlways(false, 'No primary found for set: ' + tojson(connArr)); + } + + // Take a set of mongos connections to a sharded cluster and return a + // random connection. + function getRandomMongos(connArr) { + assertAlways(Array.isArray(connArr), 'Expected an array but got ' + tojson(connArr)); + var conn = connArr[Random.randInt(connArr.length)]; + assert(isMongos(conn.getDB('admin')), tojson(conn) + ' is not to a mongos'); + return conn; + } + + // Intended for use on mongos connections only. + // Return all shards containing documents in [lower, upper). + function getShardsForRange(conn, collName, lower, upper) { + assert(isMongos(conn.getDB('admin')), tojson(conn) + ' is not to a mongos'); + var adminDB = conn.getDB('admin'); + var shardVersion = adminDB.runCommand({getShardVersion: collName, fullMetadata: true}); + assertAlways.commandWorked(shardVersion); + // As noted in SERVER-20768, doing a range query with { $lt : X }, where + // X is the _upper bound_ of a chunk, incorrectly targets the shard whose + // _lower bound_ is X. Therefore, if upper !== MaxKey, we use a workaround + // to ensure that only the shard whose lower bound = X is targeted. + var query; + if (upper === MaxKey) { + query = {$and: [{_id: {$gte: lower}}, {_id: {$lt: upper}}]}; + } else { + query = {$and: [{_id: {$gte: lower}}, {_id: {$lte: upper - 1}}]}; + } + var res = conn.getCollection(collName).find(query).explain(); + assertAlways.commandWorked(res); + assertAlways.gt( + res.queryPlanner.winningPlan.shards.length, 0, 'Explain did not have shards key.'); + + var shards = res.queryPlanner.winningPlan.shards.map(shard => shard.shardName); + return {shards: shards, explain: res, query: query, shardVersion: shardVersion}; + } + + // Return the number of docs in [lower, upper) as seen by conn. + function getNumDocs(conn, collName, lower, upper) { + var coll = conn.getCollection(collName); + var query = {$and: [{_id: {$gte: lower}}, {_id: {$lt: upper}}]}; + return coll.find(query).itcount(); + } + + // Intended for use on config or mongos connections only. + // Get number of chunks containing values in [lower, upper). The upper bound on a chunk is + // exclusive, but to capture the chunk we must provide it with less than or equal to 'upper'. + function getNumChunks(conn, lower, upper) { + assert(isMongos(conn.getDB('admin')) || isMongodConfigsvr(conn.getDB('admin')), + tojson(conn) + ' is not to a mongos or a mongod config server'); + var query = {'min._id': {$gte: lower}, 'max._id': {$lte: upper}}; + + return conn.getDB('config').chunks.find(query).itcount(); + } + + // Intended for use on config or mongos connections only. + // For getting chunks containing values in [lower, upper). The upper bound on a chunk is + // exclusive, but to capture the chunk we must provide it with less than or equal to 'upper'. + function getChunks(conn, lower, upper) { + assert(isMongos(conn.getDB('admin')) || isMongodConfigsvr(conn.getDB('admin')), + tojson(conn) + ' is not to a mongos or a mongod config server'); + var query = {'min._id': {$gte: lower}, 'max._id': {$lte: upper}}; + return conn.getDB('config').chunks.find(query).sort({'min._id': 1}).toArray(); + } + + // Intended for use on config or mongos connections only. + // For debug printing chunks containing values in [lower, upper). The upper bound on a chunk is + // exclusive, but to capture the chunk we must provide it with less than or equal to 'upper'. + function stringifyChunks(conn, lower, upper) { + assert(isMongos(conn.getDB('admin')) || isMongodConfigsvr(conn.getDB('admin')), + tojson(conn) + ' is not to a mongos or a mongod config server'); + return getChunks(conn, lower, upper).map(chunk => tojson(chunk)).join('\n'); + } + + return { + splitChunkAtPoint: splitChunkAtPoint, + splitChunkWithBounds: splitChunkWithBounds, + moveChunk: moveChunk, + mergeChunks: mergeChunks, + getPrimary: getPrimary, + getRandomMongos: getRandomMongos, + getShardsForRange: getShardsForRange, + getNumDocs: getNumDocs, + getNumChunks: getNumChunks, + getChunks: getChunks, + stringifyChunks: stringifyChunks + }; +})(); diff --git a/jstests/concurrency/fsm_workload_helpers/server_types.js b/jstests/concurrency/fsm_workload_helpers/server_types.js index 12d4933a052..0ef52935a94 100644 --- a/jstests/concurrency/fsm_workload_helpers/server_types.js +++ b/jstests/concurrency/fsm_workload_helpers/server_types.js @@ -20,6 +20,20 @@ function isMongod(db) { } /** + * Returns true if the process is a mongod configsvr, and false otherwise. + * + */ +function isMongodConfigsvr(db) { + if (!isMongod(db)) { + return false; + } + var res = db.adminCommand('getCmdLineOpts'); + assert.commandWorked(res); + + return res.parsed && res.parsed.sharding && res.parsed.sharding.clusterRole === 'configsvr'; +} + +/** * Returns the name of the current storage engine. * * Throws an error if db is connected to a mongos, or if there is no reported storage engine. diff --git a/jstests/concurrency/fsm_workloads/sharded_base_partitioned.js b/jstests/concurrency/fsm_workloads/sharded_base_partitioned.js new file mode 100644 index 00000000000..9ef02e7267d --- /dev/null +++ b/jstests/concurrency/fsm_workloads/sharded_base_partitioned.js @@ -0,0 +1,193 @@ +'use strict'; + +/** + * Provides an init state that partitions the data space into chunks evenly across threads. + * + * t1's data partition encapsulated in own chunk + * v + * ------------) | [------------) | [------------ < t3's data partition in own chunk + * ^ + * t2's data partition encapsulated in own chunk + * + * Intended to allow mergeChunks, moveChunk, and splitChunk operations to stay + * within the bounds of a thread's partition. + * + * <==t1's partition==> <==t3's partition==> + * + * ---)[--)[----)[---) | [---)[---)[----)[-)[) | [-------)[-)[-------- + * + * <===t2's partition==> + */ + +load('jstests/concurrency/fsm_workload_helpers/chunks.js'); // for chunk helpers + +var $config = (function() { + + var data = { + partitionSize: 1, + // We use a non-hashed shard key of { _id: 1 } so that documents reside on their expected + // shard. The setup function creates documents with sequential numbering and gives + // each shard its own numeric range to work with. + shardKey: {_id: 1}, + }; + + data.makePartition = function makePartition(tid, partitionSize) { + var partition = {}; + partition.lower = tid * partitionSize; + partition.upper = (tid * partitionSize) + partitionSize; + + partition.isLowChunk = (tid === 0) ? true : false; + partition.isHighChunk = (tid === (this.threadCount - 1)) ? true : false; + + partition.chunkLower = partition.isLowChunk ? MinKey : partition.lower; + partition.chunkUpper = partition.isHighChunk ? MaxKey : partition.upper; + + // Unless only 1 thread, verify that we aren't both the high and low chunk. + if (this.threadCount > 1) { + assertAlways(!(partition.isLowChunk && partition.isHighChunk), + 'should not be both the high and low chunk when there is more than 1 ' + + 'thread:\n' + tojson(this)); + } else { + assertAlways(partition.isLowChunk && partition.isHighChunk, + 'should be both the high and low chunk when there is only 1 thread:\n' + + tojson(this)); + } + + return partition; + }; + + // Intended for use on config servers only. + // Get a random chunk within this thread's partition. + data.getRandomChunkInPartition = function getRandomChunkInPartition(conn) { + assert(isMongodConfigsvr(conn.getDB('admin')), 'Not connected to a mongod configsvr'); + assert(this.partition, + 'This function must be called from workloads that partition data across threads.'); + var coll = conn.getDB('config').chunks; + // We must split up these cases because MinKey and MaxKey are not fully comparable. + // This may be due to SERVER-18341, where the Matcher returns false positives in + // comparison predicates with MinKey/MaxKey. + if (this.partition.isLowChunk && this.partition.isHighChunk) { + return coll.aggregate([{$sample: {size: 1}}]).toArray()[0]; + } else if (this.partition.isLowChunk) { + return coll + .aggregate([ + {$match: {'max._id': {$lte: this.partition.chunkUpper}}}, + {$sample: {size: 1}} + ]) + .toArray()[0]; + } else if (this.partition.isHighChunk) { + return coll + .aggregate([ + {$match: {'min._id': {$gte: this.partition.chunkLower}}}, + {$sample: {size: 1}} + ]) + .toArray()[0]; + } else { + return coll + .aggregate([ + { + $match: { + 'min._id': {$gte: this.partition.chunkLower}, + 'max._id': {$lte: this.partition.chunkUpper} + } + }, + {$sample: {size: 1}} + ]) + .toArray()[0]; + } + }; + + // This is used by the extended workloads to perform additional setup for more splitPoints. + data.setupAdditionalSplitPoints = function setupAdditionalSplitPoints(db, collName, partition) { + }; + + var states = (function() { + // Inform this thread about its partition, + // and verify that its partition is encapsulated in a single chunk. + function init(db, collName, connCache) { + // Inform this thread about its partition. + // The tid of each thread is assumed to be in the range [0, this.threadCount). + this.partition = this.makePartition(this.tid, this.partitionSize); + Object.freeze(this.partition); + + // Verify that there is exactly 1 chunk in our partition. + var config = ChunkHelper.getPrimary(connCache.config); + var numChunks = ChunkHelper.getNumChunks( + config, this.partition.chunkLower, this.partition.chunkUpper); + var chunks = ChunkHelper.getChunks(config, MinKey, MaxKey); + var msg = tojson({tid: this.tid, data: this.data, chunks: chunks}); + assertWhenOwnColl.eq(numChunks, 1, msg); + } + + function dummy(db, collName, connCache) { + } + + return {init: init, dummy: dummy}; + })(); + + var transitions = {init: {dummy: 1}, dummy: {dummy: 1}}; + + // Define each thread's data partition, populate it, and encapsulate it in a chunk. + var setup = function setup(db, collName, cluster) { + var dbName = db.getName(); + var ns = db[collName].getFullName(); + var configDB = db.getSiblingDB('config'); + + // Sharding must be enabled on db. + var res = configDB.databases.findOne({_id: dbName}); + var msg = 'db ' + dbName + ' must be sharded.'; + assertAlways(res.partitioned, msg); + + // Sharding must be enabled on db[collName]. + msg = 'collection ' + collName + ' must be sharded.'; + assertAlways.gte(configDB.chunks.find({ns: ns}).itcount(), 1, msg); + + for (var tid = 0; tid < this.threadCount; ++tid) { + // Define this thread's partition. + // The tid of each thread is assumed to be in the range [0, this.threadCount). + var partition = this.makePartition(tid, this.partitionSize); + + // Populate this thread's partition. + var bulk = db[collName].initializeUnorderedBulkOp(); + for (var i = partition.lower; i < partition.upper; ++i) { + bulk.insert({_id: i}); + } + assertAlways.writeOK(bulk.execute()); + + // Add split point for lower end of this thread's partition. + // Since a split point will be created at the low end of each partition, + // in the end each partition will be encompassed in its own chunk. + // It's unnecessary to add a split point for the lower end for the thread + // that has the lowest partition, because its chunk's lower end should be MinKey. + if (!partition.isLowChunk) { + assertWhenOwnColl.commandWorked( + ChunkHelper.splitChunkAtPoint(db, collName, partition.lower)); + } + + this.setupAdditionalSplitPoints(db, collName, partition); + } + + }; + + var skip = function skip(cluster) { + if (!cluster.isSharded() || cluster.isAutoSplitEnabled() || cluster.isBalancerEnabled()) { + return { + skip: true, + msg: 'only runs in a sharded cluster with autoSplit & balancer disabled.' + }; + } + return {skip: false}; + }; + + return { + threadCount: 1, + iterations: 1, + startState: 'init', + states: states, + transitions: transitions, + data: data, + setup: setup, + skip: skip, + passConnectionCache: true + }; +})(); diff --git a/jstests/concurrency/fsm_workloads/sharded_mergeChunks_partitioned.js b/jstests/concurrency/fsm_workloads/sharded_mergeChunks_partitioned.js new file mode 100644 index 00000000000..08545c962bf --- /dev/null +++ b/jstests/concurrency/fsm_workloads/sharded_mergeChunks_partitioned.js @@ -0,0 +1,253 @@ +'use strict'; + +/** + * Extends sharded_base_partitioned.js. + * + * Exercises the concurrent moveChunk operations, with each thread operating on its own set of + * chunks. + */ + +load('jstests/concurrency/fsm_libs/extend_workload.js'); // for extendWorkload +load('jstests/concurrency/fsm_workloads/sharded_base_partitioned.js'); // for $config + +var $config = extendWorkload($config, function($config, $super) { + + $config.iterations = 8; + $config.threadCount = 5; + + $config.data.partitionSize = 100; // number of shard key values + + // Create at least as many additional split points in this thread's partition as there + // will be iterations (to accommodate as many mergeChunks operations in this thread's + // partition as iterations). + // + // This is done in setup rather than in a mergeChunk-specific init state after the + // sharded_base_partitioned.js init state because the states are multi-threaded: + // since the mergeChunks operation used to create the chunks within each partition is not + // guaranteed to succeed (it can fail if another concurrent chunk operation is in progress), + // it is much more complicated to do this setup step in a multi-threaded context. + $config.data.setupAdditionalSplitPoints = function setupAdditionalSplitPoints( + db, collName, partition) { + // Add as many additional split points as iterations. + // Define the inner chunk size as the max size of the range of shard key + // values in each inner chunk within the thread partition as the largest + // whole number that allows for as many inner chunks as iterations without + // exceeding partitionSize. + // + // Diagram for partitionSize = 5, iterations = 4 ==> innerChunkSize = 1: + // [----------] ==> [-|-|-|-|-] + // 0 5 0 1 2 3 4 5 + // + // Diagram for partitionSize = 5, iterations = 2 ==> innerChunkSize = 2: + // [----------] ==> [-|--|--] + // 0 5 0 1 3 5 + // + // Diagram for partitionSize = 5, iterations = 1 ==> innerChunkSize = 5: + // [----------] ==> [-|----] + // 0 5 0 1 5 + var innerChunkSize = Math.floor(this.partitionSize / this.iterations); + for (var i = 0; i < this.iterations; ++i) { + var splitPoint = ((i + 1) * innerChunkSize) + partition.lower; + assertAlways.commandWorked(ChunkHelper.splitChunkAtPoint(db, collName, splitPoint)); + } + }; + + // Override sharded_base_partitioned's init state to prevent the default check + // that only 1 chunk is in our partition and to instead check that there are + // at least as many chunks in our partition as iterations. + $config.states.init = function init(db, collName, connCache) { + // Inform this thread about its partition. + // Each thread has tid in range 0..(n-1) where n is the number of threads. + this.partition = this.makePartition(this.tid, this.partitionSize); + Object.freeze(this.partition); + + var config = ChunkHelper.getPrimary(connCache.config); + + var numChunksInPartition = + ChunkHelper.getNumChunks(config, this.partition.chunkLower, this.partition.chunkUpper); + + // Verify that there is at least one chunk in our partition and that + // there are at least as many chunks in our partition as iterations. + assertWhenOwnColl.gte( + numChunksInPartition, 1, "should be at least one chunk in each thread's partition."); + assertWhenOwnColl.gt(numChunksInPartition, + this.iterations, + "should be more chunks in each thread's partition " + + 'than iterations in order to accomodate that many mergeChunks.'); + }; + + // Merge a random chunk in this thread's partition with its upper neighbor. + $config.states.mergeChunks = function mergeChunks(db, collName, connCache) { + var dbName = db.getName(); + var ns = db[collName].getFullName(); + var config = ChunkHelper.getPrimary(connCache.config); + + var chunk1, chunk2; + var configDB = config.getDB('config'); + + // Skip this iteration if our data partition contains less than 2 chunks. + if (configDB.chunks + .find({ + 'min._id': {$gte: this.partition.lower}, + 'max._id': {$lte: this.partition.upper} + }) + .itcount() < 2) { + return; + } + + // Grab a chunk and its upper neighbor. + chunk1 = this.getRandomChunkInPartition(config); + // If we randomly chose the last chunk, choose the one before it. + if (chunk1.max._id === this.partition.chunkUpper) { + chunk1 = configDB.chunks.findOne({'max._id': chunk1.min._id}); + } + chunk2 = configDB.chunks.findOne({'min._id': chunk1.max._id}); + + // Save the number of documents found in these two chunks' ranges before the mergeChunks + // operation. This will be used to verify that the same number of documents in that + // range are still found after the mergeChunks. + // Choose the mongos randomly to distribute load. + var numDocsBefore = ChunkHelper.getNumDocs( + ChunkHelper.getRandomMongos(connCache.mongos), ns, chunk1.min._id, chunk2.max._id); + + // If the second chunk is not on the same shard as the first, move it, + // because mergeChunks requires the chunks being merged to be on the same shard. + if (chunk2.shard !== chunk1.shard) { + ChunkHelper.moveChunk(db, collName, chunk2.min._id, chunk1.shard, true); + } + + // Verify that no docs were lost in the moveChunk. + var shardPrimary = ChunkHelper.getPrimary(connCache.shards[chunk1.shard]); + var shardNumDocsAfter = + ChunkHelper.getNumDocs(shardPrimary, ns, chunk1.min._id, chunk2.max._id); + var msg = "Chunk1's shard should contain all documents after mergeChunks.\n" + msgBase; + assertWhenOwnColl.eq(shardNumDocsAfter, numDocsBefore, msg); + + // Save the number of chunks before the mergeChunks operation. This will be used + // to verify that the number of chunks after a successful mergeChunks decreases + // by one, or after a failed mergeChunks stays the same. + var numChunksBefore = + ChunkHelper.getNumChunks(config, this.partition.chunkLower, this.partition.chunkUpper); + + // Use chunk_helper.js's mergeChunks wrapper to tolerate acceptable failures + // and to use a limited number of retries with exponential backoff. + var bounds = [{_id: chunk1.min._id}, {_id: chunk2.max._id}]; + var mergeChunksRes = ChunkHelper.mergeChunks(db, collName, bounds); + var chunks = + ChunkHelper.getChunks(config, this.partition.chunkLower, this.partition.chunkUpper); + var msgBase = tojson({ + mergeChunksResult: mergeChunksRes, + chunksInPartition: chunks, + chunk1: chunk1, + chunk2: chunk2 + }); + + // Regardless of whether the mergeChunks operation succeeded or failed, + // verify that the shard chunk1 was on returns all data for the chunk. + var shardPrimary = ChunkHelper.getPrimary(connCache.shards[chunk1.shard]); + var shardNumDocsAfter = + ChunkHelper.getNumDocs(shardPrimary, ns, chunk1.min._id, chunk2.max._id); + var msg = "Chunk1's shard should contain all documents after mergeChunks.\n" + msgBase; + assertWhenOwnColl.eq(shardNumDocsAfter, numDocsBefore, msg); + + // Verify that all config servers have the correct after-state. + // (see comments below for specifics). + for (var conn of connCache.config) { + var res = conn.adminCommand({isMaster: 1}); + assertAlways.commandWorked(res); + if (res.ismaster) { + // If the mergeChunks operation succeeded, verify that there is now one chunk + // between the original chunks' lower and upper bounds. If the operation failed, + // verify that there are still two chunks between the original chunks' lower and + // upper bounds. + var numChunksBetweenOldChunksBounds = + ChunkHelper.getNumChunks(conn, chunk1.min._id, chunk2.max._id); + if (mergeChunksRes.ok) { + msg = 'mergeChunks succeeded but config does not see exactly 1 chunk between ' + + 'the chunk bounds.\n' + msgBase; + assertWhenOwnColl.eq(numChunksBetweenOldChunksBounds, 1, msg); + } else { + msg = 'mergeChunks failed but config does not see exactly 2 chunks between ' + + 'the chunk bounds.\n' + msgBase; + assertWhenOwnColl.eq(numChunksBetweenOldChunksBounds, 2, msg); + } + + // If the mergeChunks operation succeeded, verify that the total number + // of chunks in our partition has decreased by 1. If it failed, verify + // that it has stayed the same. + var numChunksAfter = ChunkHelper.getNumChunks( + config, this.partition.chunkLower, this.partition.chunkUpper); + if (mergeChunksRes.ok) { + msg = 'mergeChunks succeeded but config does not see exactly 1 fewer chunks ' + + 'between the chunk bounds than before.\n' + msgBase; + assertWhenOwnColl.eq(numChunksAfter, numChunksBefore - 1, msg); + } else { + msg = 'mergeChunks failed but config sees a different number of chunks ' + + 'between the chunk bounds.\n' + msgBase; + assertWhenOwnColl.eq(numChunksAfter, numChunksBefore, msg); + } + } + } + + // Verify that all mongos processes see the correct after-state on the shards and configs. + // (see comments below for specifics). + for (var mongos of connCache.mongos) { + // Regardless of if the mergeChunks operation succeeded or failed, verify that each + // mongos sees as many documents in the original chunks' range after the move as there + // were before. + var numDocsAfter = ChunkHelper.getNumDocs(mongos, ns, chunk1.min._id, chunk2.max._id); + msg = 'Mongos sees a different amount of documents between chunk bounds after ' + + 'mergeChunks.\n' + msgBase; + assertWhenOwnColl.eq(numDocsAfter, numDocsBefore, msg); + + // Regardless of if the mergeChunks operation succeeded or failed, verify that each + // mongos sees all data in the original chunks' range only on the shard the original + // chunk was on. + var shardsForChunk = + ChunkHelper.getShardsForRange(mongos, ns, chunk1.min._id, chunk2.max._id); + msg = 'Mongos does not see exactly 1 shard for chunk after mergeChunks.\n' + msgBase + + '\n' + + 'Mongos find().explain() results for chunk: ' + tojson(shardsForChunk); + assertWhenOwnColl.eq(shardsForChunk.shards.length, 1, msg); + msg = 'Mongos sees different shard for chunk than chunk does after mergeChunks.\n' + + msgBase + '\n' + + 'Mongos find().explain() results for chunk: ' + tojson(shardsForChunk); + assertWhenOwnColl.eq(shardsForChunk.shards[0], chunk1.shard, msg); + + // If the mergeChunks operation succeeded, verify that the mongos sees one chunk between + // the original chunks' lower and upper bounds. If the operation failed, verify that the + // mongos still sees two chunks between the original chunks' lower and upper bounds. + var numChunksBetweenOldChunksBounds = + ChunkHelper.getNumChunks(mongos, chunk1.min._id, chunk2.max._id); + if (mergeChunksRes.ok) { + msg = 'mergeChunks succeeded but mongos does not see exactly 1 chunk between ' + + 'the chunk bounds.\n' + msgBase; + assertWhenOwnColl.eq(numChunksBetweenOldChunksBounds, 1, msg); + } else { + msg = 'mergeChunks failed but mongos does not see exactly 2 chunks between ' + + 'the chunk bounds.\n' + msgBase; + assertWhenOwnColl.eq(numChunksBetweenOldChunksBounds, 2, msg); + } + + // If the mergeChunks operation succeeded, verify that the mongos sees that the total + // number of chunks in our partition has decreased by 1. If it failed, verify that it + // has stayed the same. + var numChunksAfter = ChunkHelper.getNumChunks( + mongos, this.partition.chunkLower, this.partition.chunkUpper); + if (mergeChunksRes.ok) { + msg = 'mergeChunks succeeded but mongos does not see exactly 1 fewer chunks ' + + 'between the chunk bounds.\n' + msgBase; + assertWhenOwnColl.eq(numChunksAfter, numChunksBefore - 1, msg); + } else { + msg = 'mergeChunks failed but mongos does not see the same number of chunks ' + + 'between the chunk bounds.\n' + msgBase; + assertWhenOwnColl.eq(numChunksAfter, numChunksBefore, msg); + } + } + + }; + + $config.transitions = {init: {mergeChunks: 1}, mergeChunks: {mergeChunks: 1}}; + + return $config; +}); diff --git a/jstests/concurrency/fsm_workloads/sharded_moveChunk_partitioned.js b/jstests/concurrency/fsm_workloads/sharded_moveChunk_partitioned.js new file mode 100644 index 00000000000..6348342f415 --- /dev/null +++ b/jstests/concurrency/fsm_workloads/sharded_moveChunk_partitioned.js @@ -0,0 +1,182 @@ +'use strict'; + +/** + * Extends sharded_base_partitioned.js. + * + * Exercises the concurrent moveChunk operations, but each thread operates on its own set of + * chunks. + */ + +load('jstests/concurrency/fsm_libs/extend_workload.js'); // for extendWorkload +load('jstests/concurrency/fsm_workloads/sharded_base_partitioned.js'); // for $config + +var $config = extendWorkload($config, function($config, $super) { + + $config.iterations = 5; + $config.threadCount = 5; + + $config.data.partitionSize = 100; // number of shard key values + + // Re-assign a chunk from this thread's partition to a random shard, and + // verify that each node in the cluster affected by the moveChunk operation sees + // the appropriate after-state regardless of whether the operation succeeded or failed. + $config.states.moveChunk = function moveChunk(db, collName, connCache) { + var dbName = db.getName(); + var ns = db[collName].getFullName(); + var config = ChunkHelper.getPrimary(connCache.config); + + // Verify that more than one shard exists in the cluster. If only one shard existed, + // there would be no way to move a chunk from one shard to another. + var numShards = config.getDB('config').shards.find().itcount(); + var msg = 'There must be more than one shard when performing a moveChunks operation\n' + + 'shards: ' + tojson(config.getDB('config').shards.find().toArray()); + assertAlways.gt(numShards, 1, msg); + + // Choose a random chunk in our partition to move. + var chunk = this.getRandomChunkInPartition(config); + var fromShard = chunk.shard; + + // Choose a random shard to move the chunk to. + var shardNames = Object.keys(connCache.shards); + var destinationShards = shardNames.filter(function(shard) { + if (shard !== fromShard) { + return shard; + } + }); + var toShard = destinationShards[Random.randInt(destinationShards.length)]; + + // Save the number of documents in this chunk's range found on the chunk's current shard + // (the fromShard) before the moveChunk operation. This will be used to verify that the + // number of documents in the chunk's range found on the _toShard_ after a _successful_ + // moveChunk operation is the same as numDocsBefore, or that the number of documents in the + // chunk's range found on the _fromShard_ after a _failed_ moveChunk operation is the same + // as numDocsBefore. + // Choose the mongos randomly to distribute load. + var numDocsBefore = ChunkHelper.getNumDocs( + ChunkHelper.getRandomMongos(connCache.mongos), ns, chunk.min._id, chunk.max._id); + + // Save the number of chunks before the moveChunk operation. This will be used + // to verify that the number of chunks after the moveChunk operation remains the same. + var numChunksBefore = + ChunkHelper.getNumChunks(config, this.partition.chunkLower, this.partition.chunkUpper); + + // Randomly choose whether to wait for all documents on the fromShard + // to be deleted before the moveChunk operation returns. + var waitForDelete = Random.rand() < 0.5; + + // Use chunk_helper.js's moveChunk wrapper to tolerate acceptable failures + // and to use a limited number of retries with exponential backoff. + var bounds = [{_id: chunk.min._id}, {_id: chunk.max._id}]; + var moveChunkRes = ChunkHelper.moveChunk(db, collName, bounds, toShard, waitForDelete); + var msgBase = 'Result of moveChunk operation: ' + tojson(moveChunkRes); + + // Verify that the fromShard and toShard have the correct after-state + // (see comments below for specifics). + var fromShardPrimary = ChunkHelper.getPrimary(connCache.shards[fromShard]); + var toShardPrimary = ChunkHelper.getPrimary(connCache.shards[toShard]); + var fromShardNumDocsAfter = + ChunkHelper.getNumDocs(fromShardPrimary, ns, chunk.min._id, chunk.max._id); + var toShardNumDocsAfter = + ChunkHelper.getNumDocs(toShardPrimary, ns, chunk.min._id, chunk.max._id); + // If the moveChunk operation succeeded, verify that the shard the chunk + // was moved to returns all data for the chunk. If waitForDelete was true, + // also verify that the shard the chunk was moved from returns no data for the chunk. + if (moveChunkRes.ok) { + if (waitForDelete) { + msg = 'moveChunk succeeded but original shard still had documents.\n' + msgBase + + ', waitForDelete: ' + waitForDelete + ', bounds: ' + tojson(bounds); + assertWhenOwnColl.eq(fromShardNumDocsAfter, 0, msg); + } + msg = 'moveChunk succeeded but new shard did not contain all documents.\n' + msgBase + + ', waitForDelete: ' + waitForDelete + ', bounds: ' + tojson(bounds); + assertWhenOwnColl.eq(toShardNumDocsAfter, numDocsBefore, msg); + } + // If the moveChunk operation failed, verify that the shard the chunk was + // originally on returns all data for the chunk, and the shard the chunk + // was supposed to be moved to returns no data for the chunk. + else { + msg = 'moveChunk failed but original shard did not contain all documents.\n' + msgBase + + ', waitForDelete: ' + waitForDelete + ', bounds: ' + tojson(bounds); + assertWhenOwnColl.eq(fromShardNumDocsAfter, numDocsBefore, msg); + msg = 'moveChunk failed but new shard had documents.\n' + msgBase + + ', waitForDelete: ' + waitForDelete + ', bounds: ' + tojson(bounds); + assertWhenOwnColl.eq(toShardNumDocsAfter, 0, msg); + } + + // Verify that all config servers have the correct after-state. + // (see comments below for specifics). + for (var conn of connCache.config) { + var res = conn.adminCommand({isMaster: 1}); + assertAlways.commandWorked(res); + if (res.ismaster) { + // If the moveChunk operation succeeded, verify that the config updated the chunk's + // shard with the toShard. If the operation failed, verify that the config kept + // the chunk's shard as the fromShard. + var chunkAfter = conn.getDB('config').chunks.findOne({_id: chunk._id}); + var msg = msgBase + '\nchunkBefore: ' + tojson(chunk) + '\nchunkAfter: ' + + tojson(chunkAfter); + if (moveChunkRes.ok) { + msg = "moveChunk succeeded but chunk's shard was not new shard.\n" + msg; + assertWhenOwnColl.eq(chunkAfter.shard, toShard, msg); + } else { + msg = "moveChunk failed but chunk's shard was not original shard.\n" + msg; + assertWhenOwnColl.eq(chunkAfter.shard, fromShard, msg); + } + + // Regardless of whether the operation succeeded or failed, + // verify that the number of chunks in our partition stayed the same. + var numChunksAfter = ChunkHelper.getNumChunks( + conn, this.partition.chunkLower, this.partition.chunkUpper); + msg = 'Number of chunks in partition seen by config changed with moveChunk.\n' + + msgBase; + assertWhenOwnColl.eq(numChunksBefore, numChunksAfter, msgBase); + } + } + + // Verify that all mongos processes see the correct after-state on the shards and configs. + // (see comments below for specifics). + for (var mongos of connCache.mongos) { + // Regardless of if the moveChunk operation succeeded or failed, + // verify that each mongos sees as many documents in the chunk's + // range after the move as there were before. + var numDocsAfter = ChunkHelper.getNumDocs(mongos, ns, chunk.min._id, chunk.max._id); + msg = + 'Number of chunks in partition seen by mongos changed with moveChunk.\n' + msgBase; + assertWhenOwnColl.eq(numDocsAfter, numDocsBefore, msgBase); + + // If the moveChunk operation succeeded, verify that each mongos sees all data in the + // chunk's range on only the toShard. If the operation failed, verify that each mongos + // sees all data in the chunk's range on only the fromShard. + var shardsForChunk = + ChunkHelper.getShardsForRange(mongos, ns, chunk.min._id, chunk.max._id); + var msg = + msgBase + '\nMongos find().explain() results for chunk: ' + tojson(shardsForChunk); + assertWhenOwnColl.eq(shardsForChunk.shards.length, 1, msg); + if (moveChunkRes.ok) { + msg = 'moveChunk succeeded but chunk was not on new shard.\n' + msg; + assertWhenOwnColl.eq(shardsForChunk.shards[0], toShard, msg); + } else { + msg = 'moveChunk failed but chunk was not on original shard.\n' + msg; + assertWhenOwnColl.eq(shardsForChunk.shards[0], fromShard, msg); + } + + // If the moveChunk operation succeeded, verify that each mongos updated the chunk's + // shard metadata with the toShard. If the operation failed, verify that each mongos + // still sees the chunk's shard metadata as the fromShard. + var chunkAfter = mongos.getDB('config').chunks.findOne({_id: chunk._id}); + var msg = + msgBase + '\nchunkBefore: ' + tojson(chunk) + '\nchunkAfter: ' + tojson(chunkAfter); + if (moveChunkRes.ok) { + msg = "moveChunk succeeded but chunk's shard was not new shard.\n" + msg; + assertWhenOwnColl.eq(chunkAfter.shard, toShard, msg); + } else { + msg = "moveChunk failed but chunk's shard was not original shard.\n" + msg; + assertWhenOwnColl.eq(chunkAfter.shard, fromShard, msg); + } + } + }; + + $config.transitions = {init: {moveChunk: 1}, moveChunk: {moveChunk: 1}}; + + return $config; +}); diff --git a/jstests/concurrency/fsm_workloads/sharded_splitChunk_partitioned.js b/jstests/concurrency/fsm_workloads/sharded_splitChunk_partitioned.js new file mode 100644 index 00000000000..20e90a72d22 --- /dev/null +++ b/jstests/concurrency/fsm_workloads/sharded_splitChunk_partitioned.js @@ -0,0 +1,158 @@ +'use strict'; + +/** + * Extends sharded_base_partitioned.js. + * + * Exercises the concurrent splitChunk operations, but each thread operates on its own set of + * chunks. + */ + +load('jstests/concurrency/fsm_libs/extend_workload.js'); // for extendWorkload +load('jstests/concurrency/fsm_workloads/sharded_base_partitioned.js'); // for $config + +var $config = extendWorkload($config, function($config, $super) { + + $config.iterations = 5; + $config.threadCount = 5; + + $config.data.partitionSize = 100; // number of shard key values + + // Split a random chunk in this thread's partition, and verify that each node + // in the cluster affected by the splitChunk operation sees the appropriate + // after-state regardless of whether the operation succeeded or failed. + $config.states.splitChunk = function splitChunk(db, collName, connCache) { + + var dbName = db.getName(); + var ns = db[collName].getFullName(); + var config = ChunkHelper.getPrimary(connCache.config); + + // Choose a random chunk in our partition to split. + var chunk = this.getRandomChunkInPartition(config); + + // Save the number of documents found in this chunk's range before the splitChunk + // operation. This will be used to verify that the same number of documents in that + // range are found after the splitChunk. + // Choose the mongos randomly to distribute load. + var numDocsBefore = ChunkHelper.getNumDocs( + ChunkHelper.getRandomMongos(connCache.mongos), ns, chunk.min._id, chunk.max._id); + + // Save the number of chunks before the splitChunk operation. This will be used + // to verify that the number of chunks after a successful splitChunk increases + // by one, or after a failed splitChunk stays the same. + var numChunksBefore = + ChunkHelper.getNumChunks(config, this.partition.chunkLower, this.partition.chunkUpper); + + // Use chunk_helper.js's splitChunk wrapper to tolerate acceptable failures + // and to use a limited number of retries with exponential backoff. + var bounds = [{_id: chunk.min._id}, {_id: chunk.max._id}]; + var splitChunkRes = ChunkHelper.splitChunkWithBounds(db, collName, bounds); + var msgBase = 'Result of splitChunk operation: ' + tojson(splitChunkRes); + + // Regardless of whether the splitChunk operation succeeded or failed, + // verify that the shard the original chunk was on returns all data for the chunk. + var shardPrimary = ChunkHelper.getPrimary(connCache.shards[chunk.shard]); + var shardNumDocsAfter = + ChunkHelper.getNumDocs(shardPrimary, ns, chunk.min._id, chunk.max._id); + var msg = 'Shard does not have same number of documents after splitChunk.\n' + msgBase; + assertWhenOwnColl.eq(shardNumDocsAfter, numDocsBefore, msg); + + // Verify that all config servers have the correct after-state. + // (see comments below for specifics). + for (var conn of connCache.config) { + var res = conn.adminCommand({isMaster: 1}); + assertAlways.commandWorked(res); + if (res.ismaster) { + // If the splitChunk operation succeeded, verify that there are now + // two chunks between the old chunk's lower and upper bounds. + // If the operation failed, verify that there is still only one chunk + // between the old chunk's lower and upper bounds. + var numChunksBetweenOldChunksBounds = + ChunkHelper.getNumChunks(conn, chunk.min._id, chunk.max._id); + if (splitChunkRes.ok) { + msg = 'splitChunk succeeded but the config does not see exactly 2 chunks ' + + 'between the chunk bounds.\n' + msgBase; + assertWhenOwnColl.eq(numChunksBetweenOldChunksBounds, 2, msg); + } else { + msg = 'splitChunk failed but the config does not see exactly 1 chunk between ' + + 'the chunk bounds.\n' + msgBase; + assertWhenOwnColl.eq(numChunksBetweenOldChunksBounds, 1, msg); + } + + // If the splitChunk operation succeeded, verify that the total number + // of chunks in our partition has increased by 1. If it failed, verify + // that it has stayed the same. + var numChunksAfter = ChunkHelper.getNumChunks( + config, this.partition.chunkLower, this.partition.chunkUpper); + if (splitChunkRes.ok) { + msg = 'splitChunk succeeded but the config does nnot see exactly 1 more ' + + 'chunk between the chunk bounds.\n' + msgBase; + assertWhenOwnColl.eq(numChunksAfter, numChunksBefore + 1, msg); + } else { + msg = 'splitChunk failed but the config does not see the same number ' + + 'of chunks between the chunk bounds.\n' + msgBase; + assertWhenOwnColl.eq(numChunksAfter, numChunksBefore, msg); + } + } + } + + // Verify that all mongos processes see the correct after-state on the shards and configs. + // (see comments below for specifics). + for (var mongos of connCache.mongos) { + // Regardless of if the splitChunk operation succeeded or failed, verify that each + // mongos sees as many documents in the chunk's range after the move as there were + // before. + var numDocsAfter = ChunkHelper.getNumDocs(mongos, ns, chunk.min._id, chunk.max._id); + + msg = 'Mongos does not see same number of documents after splitChunk.\n' + msgBase; + assertWhenOwnColl.eq(numDocsAfter, numDocsBefore, msgBase); + + // Regardless of if the splitChunk operation succeeded or failed, + // verify that each mongos sees all data in the original chunk's + // range only on the shard the original chunk was on. + var shardsForChunk = + ChunkHelper.getShardsForRange(mongos, ns, chunk.min._id, chunk.max._id); + msg = 'Mongos does not see exactly 1 shard for chunk after splitChunk.\n' + msgBase + + '\n' + + 'Mongos find().explain() results for chunk: ' + tojson(shardsForChunk); + assertWhenOwnColl.eq(shardsForChunk.shards.length, 1, msg); + + msg = 'Mongos sees different shard for chunk than chunk does after splitChunk.\n' + + msgBase + '\n' + + 'Mongos find().explain() results for chunk: ' + tojson(shardsForChunk); + assertWhenOwnColl.eq(shardsForChunk.shards[0], chunk.shard, msg); + + // If the splitChunk operation succeeded, verify that the mongos sees two chunks between + // the old chunk's lower and upper bounds. If the operation failed, verify that the + // mongos still only sees one chunk between the old chunk's lower and upper bounds. + var numChunksBetweenOldChunksBounds = + ChunkHelper.getNumChunks(mongos, chunk.min._id, chunk.max._id); + if (splitChunkRes.ok) { + msg = 'splitChunk succeeded but the mongos does not see exactly 2 chunks ' + + 'between the chunk bounds.\n' + msgBase; + assertWhenOwnColl.eq(numChunksBetweenOldChunksBounds, 2, msg); + } else { + msg = 'splitChunk failed but the mongos does not see exactly 1 chunk between ' + + 'the chunk bounds.\n' + msgBase; + assertWhenOwnColl.eq(numChunksBetweenOldChunksBounds, 1, msg); + } + + // If the splitChunk operation succeeded, verify that the total number of chunks in our + // partition has increased by 1. If it failed, verify that it has stayed the same. + var numChunksAfter = ChunkHelper.getNumChunks( + mongos, this.partition.chunkLower, this.partition.chunkUpper); + if (splitChunkRes.ok) { + msg = 'splitChunk succeeded but the mongos does nnot see exactly 1 more ' + + 'chunk between the chunk bounds.\n' + msgBase; + assertWhenOwnColl.eq(numChunksAfter, numChunksBefore + 1, msg); + } else { + msg = 'splitChunk failed but the mongos does not see the same number ' + + 'of chunks between the chunk bounds.\n' + msgBase; + assertWhenOwnColl.eq(numChunksAfter, numChunksBefore, msg); + } + } + }; + + $config.transitions = {init: {splitChunk: 1}, splitChunk: {splitChunk: 1}}; + + return $config; +}); |