diff options
4 files changed, 137 insertions, 9 deletions
diff --git a/jstests/sharding/chunk_history_window.js b/jstests/sharding/chunk_history_window.js new file mode 100644 index 00000000000..85a0b09f462 --- /dev/null +++ b/jstests/sharding/chunk_history_window.js @@ -0,0 +1,116 @@ +/** + * Test that config servers keep chunk history for up to minSnapshotHistoryWindowInSeconds. + * + * @tags: [ + * requires_fcv_46, + * ] + * + * - Create a one-chunk sharded collection, its history is [{validAfter: T0}]. + * - Insert a document at timestamp insertTS. + * - Move the chunk, its history is [{validAfter: T1}, {validAfter: T0}], where T1 > insertTS > T0. + * - Until now > insertTS + window - margin, read at insertTS and assert success. + * - After now > T0 + window + margin, T0 is expired. Move the chunk, triggering a history cleanup. + * - History is [{validAfter: T2}, {validAfter: T1}], where T2 > T1 > insertTS > T0. + * - Read at insertTS and assert failure with StaleChunkHistory. + * - Read at T2 - 1 sec, assert success. + */ +(function() { +"use strict"; + +load("jstests/sharding/libs/sharded_transactions_helpers.js"); + +const configHistoryWindowSecs = 10; +const st = new ShardingTest({ + shards: {rs0: {nodes: 2}, rs1: {nodes: 2}}, + other: { + configOptions: { + setParameter: { + minSnapshotHistoryWindowInSeconds: configHistoryWindowSecs, + logComponentVerbosity: tojson({sharding: {verbosity: 2}}) + } + }, + rsOptions: {setParameter: {minSnapshotHistoryWindowInSeconds: 600}} + } +}); + +const primaryAdmin = st.rs0.getPrimary().getDB("admin"); +assert.eq(assert + .commandWorked( + primaryAdmin.runCommand({getParameter: 1, minSnapshotHistoryWindowInSeconds: 1})) + .minSnapshotHistoryWindowInSeconds, + 600); + +const configAdmin = st.configRS.getPrimary().getDB("admin"); +assert.eq(assert + .commandWorked( + configAdmin.runCommand({getParameter: 1, minSnapshotHistoryWindowInSeconds: 1})) + .minSnapshotHistoryWindowInSeconds, + 10); + +const mongosDB = st.s.getDB(jsTestName()); +const mongosColl = mongosDB.test; +const ns = `${jsTestName()}.test`; + +assert.commandWorked(mongosDB.adminCommand({enableSharding: mongosDB.getName()})); +st.ensurePrimaryShard(mongosDB.getName(), st.rs0.getURL()); +st.shardColl(mongosColl, {_id: 1}, false); + +const configChunks = st.configRS.getPrimary().getDB("config")["chunks"]; +const origChunk = configChunks.findOne({ns: ns}); +jsTestLog(`Original chunk: ${tojson(origChunk)}`); +assert.eq(1, origChunk.history.length, tojson(origChunk)); +let result = mongosDB.runCommand({insert: "test", documents: [{_id: 0}]}); +const insertTS = assert.commandWorked(result).operationTime; +jsTestLog(`Inserted one document at ${insertTS}`); +assert.lte(origChunk.history[0].validAfter, insertTS, `history: ${tojson(origChunk.history)}`); + +jsTestLog("Move chunk to shard 1, create second history entry"); +assert.commandWorked(st.s.adminCommand({moveChunk: ns, find: {_id: 0}, to: st.shard1.shardName})); +const postMoveChunkTime = Date.now(); +let chunk = configChunks.findOne({_id: origChunk._id}); +jsTestLog(`Chunk: ${tojson(chunk)}`); +assert.eq(2, chunk.history.length, tojson(chunk)); + +// Test history window with 1s margin. +const testMarginMS = 1000; + +// Test that reading from a snapshot at insertTS is valid for up to configHistoryWindowSecs +// minus the testMarginMS (as a buffer). +const testWindowMS = configHistoryWindowSecs * 1000 - testMarginMS; +while (Date.now() - insertTS < testWindowMS) { + // Test that reading from a snapshot at insertTS is still valid. + assert.commandWorked(mongosDB.runCommand( + {find: "test", readConcern: {level: "snapshot", atClusterTime: insertTS}})); + + chunk = configChunks.findOne({_id: origChunk._id}); + assert.eq(2, chunk.history.length, tojson(chunk)); + sleep(50); +} + +// Sleep until our most recent chunk move is before the oldest history in our window. +const chunkExpirationTime = postMoveChunkTime + configHistoryWindowSecs * 1000; +sleep(chunkExpirationTime + testMarginMS - Date.now()); + +jsTestLog("Move chunk back to shard 0 to trigger history cleanup"); +assert.commandWorked(st.s.adminCommand({moveChunk: ns, find: {_id: 0}, to: st.shard0.shardName})); +chunk = configChunks.findOne({_id: origChunk._id}); +jsTestLog(`Chunk: ${tojson(chunk)}`); +// Oldest history entry was deleted: we added one and deleted one, still have two. +assert.eq(2, chunk.history.length, tojson(chunk)); +assert.gte(chunk.history[1].validAfter, insertTS, `history: ${tojson(chunk.history)}`); + +flushRoutersAndRefreshShardMetadata(st, {ns}); + +// Test that reading from a snapshot at insertTS returns StaleChunkHistory: the shards have enough +// history but the config servers don't. +assert.commandFailedWithCode( + mongosDB.runCommand({find: "test", readConcern: {level: "snapshot", atClusterTime: insertTS}}), + ErrorCodes.StaleChunkHistory); + +// One second before the newest history entry is valid (check we don't delete *all* old entries). +var recentTS = Timestamp(chunk.history[0].validAfter.getTime() - 1, 0); +assert.commandWorked( + mongosDB.runCommand({find: "test", readConcern: {level: "snapshot", atClusterTime: recentTS}})); + +st.stop(); +})(); diff --git a/src/mongo/db/s/SConscript b/src/mongo/db/s/SConscript index 6ad8bf9f6a8..c932e48c2e2 100644 --- a/src/mongo/db/s/SConscript +++ b/src/mongo/db/s/SConscript @@ -255,7 +255,9 @@ env.Library( ], LIBDEPS_PRIVATE=[ '$BUILD_DIR/mongo/db/commands/mongod_fcv', + '$BUILD_DIR/mongo/db/snapshot_window_options', '$BUILD_DIR/mongo/s/catalog/sharding_catalog_client_impl', + '$BUILD_DIR/mongo/util/options_parser/options_parser', ], ) diff --git a/src/mongo/db/s/config/sharding_catalog_manager_chunk_operations.cpp b/src/mongo/db/s/config/sharding_catalog_manager_chunk_operations.cpp index 7d4e635daec..1b30456d4de 100644 --- a/src/mongo/db/s/config/sharding_catalog_manager_chunk_operations.cpp +++ b/src/mongo/db/s/config/sharding_catalog_manager_chunk_operations.cpp @@ -45,6 +45,7 @@ #include "mongo/db/repl/repl_client_info.h" #include "mongo/db/s/sharding_logging.h" #include "mongo/db/server_options.h" +#include "mongo/db/snapshot_window_options_gen.h" #include "mongo/logv2/log.h" #include "mongo/rpc/get_status_from_command_result.h" #include "mongo/s/catalog/sharding_catalog_client.h" @@ -777,19 +778,28 @@ StatusWith<BSONObj> ShardingCatalogManager::commitChunkMigration( // Copy the complete history. auto newHistory = origChunk.getValue().getHistory(); - const int kHistorySecs = 10; - invariant(validAfter); - // Update the history of the migrated chunk. - // Drop the history that is too old (10 seconds of history for now). - // TODO SERVER-33831 to update the old history removal policy. + // Drop old history. Keep at least 1 entry so ChunkInfo::getShardIdAt finds valid history for + // any query younger than the history window. if (!MONGO_unlikely(skipExpiringOldChunkHistory.shouldFail())) { - while (!newHistory.empty() && - newHistory.back().getValidAfter().getSecs() + kHistorySecs < + const int kHistorySecs = 10; + auto windowInSeconds = std::max(minSnapshotHistoryWindowInSeconds.load(), kHistorySecs); + int entriesDeleted = 0; + while (newHistory.size() > 1 && + newHistory.back().getValidAfter().getSecs() + windowInSeconds < validAfter.get().getSecs()) { newHistory.pop_back(); + ++entriesDeleted; + } + + logv2::DynamicAttributes attrs; + attrs.add("entriesDeleted", entriesDeleted); + if (!newHistory.empty()) { + attrs.add("oldestEntryValidAfter", newHistory.back().getValidAfter()); } + + LOGV2_DEBUG(4778500, 1, "Deleted old chunk history entries", attrs); } if (!newHistory.empty() && newHistory.front().getValidAfter() >= validAfter.get()) { diff --git a/src/mongo/db/s/config/sharding_catalog_manager_commit_chunk_migration_test.cpp b/src/mongo/db/s/config/sharding_catalog_manager_commit_chunk_migration_test.cpp index ef7326db8d9..5d0ed77c047 100644 --- a/src/mongo/db/s/config/sharding_catalog_manager_commit_chunk_migration_test.cpp +++ b/src/mongo/db/s/config/sharding_catalog_manager_commit_chunk_migration_test.cpp @@ -248,8 +248,8 @@ TEST_F(CommitChunkMigrate, CheckCorrectOpsCommandNoCtlTrimHistory) { auto chunkDoc0 = uassertStatusOK(getChunkDoc(operationContext(), chunkMin)); ASSERT_EQ("shard1", chunkDoc0.getShard().toString()); ASSERT_EQ(mver.getValue(), chunkDoc0.getVersion()); - // The history should be updated. - ASSERT_EQ(1UL, chunkDoc0.getHistory().size()); + // The new history entry should be added, but the old one preserved. + ASSERT_EQ(2UL, chunkDoc0.getHistory().size()); ASSERT_EQ(validAfter, chunkDoc0.getHistory().front().getValidAfter()); } |