diff options
Diffstat (limited to 'src/mongo/db/repl')
31 files changed, 699 insertions, 264 deletions
diff --git a/src/mongo/db/repl/SConscript b/src/mongo/db/repl/SConscript index 05dbeff18a0..31f35d384f8 100644 --- a/src/mongo/db/repl/SConscript +++ b/src/mongo/db/repl/SConscript @@ -631,12 +631,12 @@ env.Library('replica_set_messages', 'is_master_response.cpp', 'member_config.cpp', 'old_update_position_args.cpp', + 'repl_set_config.cpp', 'repl_set_heartbeat_args.cpp', 'repl_set_heartbeat_args_v1.cpp', 'repl_set_heartbeat_response.cpp', 'repl_set_html_summary.cpp', 'repl_set_request_votes_args.cpp', - 'repl_set_config.cpp', 'repl_set_tag.cpp', 'update_position_args.cpp', 'last_vote.cpp', diff --git a/src/mongo/db/repl/bgsync.cpp b/src/mongo/db/repl/bgsync.cpp index 2c04c40bf55..fbbf1c56efd 100644 --- a/src/mongo/db/repl/bgsync.cpp +++ b/src/mongo/db/repl/bgsync.cpp @@ -225,15 +225,16 @@ void BackgroundSync::_runProducer() { } // we want to start when we're no longer primary // start() also loads _lastOpTimeFetched, which we know is set from the "if" - auto txn = cc().makeOperationContext(); - if (getState() == ProducerState::Starting) { - start(txn.get()); + { + auto opCtx = cc().makeOperationContext(); + if (getState() == ProducerState::Starting) { + start(opCtx.get()); + } } - - _produce(txn.get()); + _produce(); } -void BackgroundSync::_produce(OperationContext* opCtx) { +void BackgroundSync::_produce() { if (MONGO_FAIL_POINT(stopReplProducer)) { // This log output is used in js tests so please leave it. log() << "bgsync - stopReplProducer fail point " @@ -266,15 +267,18 @@ void BackgroundSync::_produce(OperationContext* opCtx) { } } - auto storageInterface = StorageInterface::get(opCtx); // find a target to sync from the last optime fetched OpTime lastOpTimeFetched; HostAndPort source; HostAndPort oldSource = _syncSourceHost; SyncSourceResolverResponse syncSourceResp; { - const OpTime minValidSaved = storageInterface->getMinValid(opCtx); - + OpTime minValidSaved; + { + auto opCtx = cc().makeOperationContext(); + auto storageInterface = StorageInterface::get(opCtx.get()); + minValidSaved = storageInterface->getMinValid(opCtx.get()); + } stdx::lock_guard<stdx::mutex> lock(_mutex); if (_state != ProducerState::Running) { return; @@ -397,8 +401,12 @@ void BackgroundSync::_produce(OperationContext* opCtx) { // Set the applied point if unset. This is most likely the first time we've established a sync // source since stepping down or otherwise clearing the applied point. We need to set this here, // before the OplogWriter gets a chance to append to the oplog. - if (storageInterface->getAppliedThrough(opCtx).isNull()) { - storageInterface->setAppliedThrough(opCtx, _replCoord->getMyLastAppliedOpTime()); + { + auto opCtx = cc().makeOperationContext(); + auto storageInterface = StorageInterface::get(opCtx.get()); + if (storageInterface->getAppliedThrough(opCtx.get()).isNull()) { + storageInterface->setAppliedThrough(opCtx.get(), _replCoord->getMyLastAppliedOpTime()); + } } // "lastFetched" not used. Already set in _enqueueDocuments. @@ -521,10 +529,18 @@ void BackgroundSync::_produce(OperationContext* opCtx) { } } - OplogInterfaceLocal localOplog(opCtx, rsOplogName); - RollbackSourceImpl rollbackSource(getConnection, source, rsOplogName); - rollback( - opCtx, localOplog, rollbackSource, syncSourceResp.rbid, _replCoord, storageInterface); + { + auto opCtx = cc().makeOperationContext(); + OplogInterfaceLocal localOplog(opCtx.get(), rsOplogName); + RollbackSourceImpl rollbackSource(getConnection, source, rsOplogName); + auto storageInterface = StorageInterface::get(opCtx.get()); + rollback(opCtx.get(), + localOplog, + rollbackSource, + syncSourceResp.rbid, + _replCoord, + storageInterface); + } // Reset the producer to clear the sync source and the last optime fetched. stop(true); diff --git a/src/mongo/db/repl/bgsync.h b/src/mongo/db/repl/bgsync.h index 6d068967723..00f23e59b1d 100644 --- a/src/mongo/db/repl/bgsync.h +++ b/src/mongo/db/repl/bgsync.h @@ -148,7 +148,7 @@ private: void _run(); // Production thread inner loop. void _runProducer(); - void _produce(OperationContext* txn); + void _produce(); /** * Checks current background sync state before pushing operations into blocking queue and diff --git a/src/mongo/db/repl/collection_cloner.cpp b/src/mongo/db/repl/collection_cloner.cpp index 9dc3674efa0..bbaf04ba3dc 100644 --- a/src/mongo/db/repl/collection_cloner.cpp +++ b/src/mongo/db/repl/collection_cloner.cpp @@ -114,7 +114,8 @@ CollectionCloner::CollectionCloner(executor::TaskExecutor* executor, stdx::placeholders::_2, stdx::placeholders::_3), rpc::ServerSelectionMetadata(true, boost::none).toBSON(), - RemoteCommandRequest::kNoTimeout, + RemoteCommandRequest::kNoTimeout /* find network timeout */, + RemoteCommandRequest::kNoTimeout /* getMore network timeout */, RemoteCommandRetryScheduler::makeRetryPolicy( numInitialSyncListIndexesAttempts, executor::RemoteCommandRequest::kNoTimeout, @@ -518,7 +519,8 @@ void CollectionCloner::_beginCollectionCallback(const executor::TaskExecutor::Ca stdx::placeholders::_3, onCompletionGuard), rpc::ServerSelectionMetadata(true, boost::none).toBSON(), - RemoteCommandRequest::kNoTimeout, + RemoteCommandRequest::kNoTimeout /* find network timeout */, + RemoteCommandRequest::kNoTimeout /* getMore network timeout */, RemoteCommandRetryScheduler::makeRetryPolicy( numInitialSyncCollectionFindAttempts.load(), executor::RemoteCommandRequest::kNoTimeout, diff --git a/src/mongo/db/repl/database_cloner.cpp b/src/mongo/db/repl/database_cloner.cpp index 461c4dfe1b7..d0211fdf058 100644 --- a/src/mongo/db/repl/database_cloner.cpp +++ b/src/mongo/db/repl/database_cloner.cpp @@ -114,7 +114,8 @@ DatabaseCloner::DatabaseCloner(executor::TaskExecutor* executor, stdx::placeholders::_2, stdx::placeholders::_3), rpc::ServerSelectionMetadata(true, boost::none).toBSON(), - RemoteCommandRequest::kNoTimeout, + RemoteCommandRequest::kNoTimeout /* find network timeout */, + RemoteCommandRequest::kNoTimeout /* getMore network timeout */, RemoteCommandRetryScheduler::makeRetryPolicy( numInitialSyncListCollectionsAttempts, executor::RemoteCommandRequest::kNoTimeout, diff --git a/src/mongo/db/repl/databases_cloner.cpp b/src/mongo/db/repl/databases_cloner.cpp index 4a18e1be1b8..c9016c322f4 100644 --- a/src/mongo/db/repl/databases_cloner.cpp +++ b/src/mongo/db/repl/databases_cloner.cpp @@ -231,6 +231,44 @@ void DatabasesCloner::setScheduleDbWorkFn_forTest(const CollectionCloner::Schedu _scheduleDbWorkFn = work; } +StatusWith<std::vector<BSONElement>> DatabasesCloner::parseListDatabasesResponse_forTest( + BSONObj dbResponse) { + return _parseListDatabasesResponse(dbResponse); +} + +void DatabasesCloner::setAdminAsFirst_forTest(std::vector<BSONElement>& dbsArray) { + _setAdminAsFirst(dbsArray); +} + +StatusWith<std::vector<BSONElement>> DatabasesCloner::_parseListDatabasesResponse( + BSONObj dbResponse) { + if (!dbResponse.hasField("databases")) { + return Status(ErrorCodes::BadValue, + "The 'listDatabases' response does not contain a 'databases' field."); + } + BSONElement response = dbResponse["databases"]; + try { + return response.Array(); + } catch (const MsgAssertionException& e) { + return Status(ErrorCodes::BadValue, + "The 'listDatabases' response is unable to be transformed into an array."); + } +} + +void DatabasesCloner::_setAdminAsFirst(std::vector<BSONElement>& dbsArray) { + auto adminIter = std::find_if(dbsArray.begin(), dbsArray.end(), [](BSONElement elem) { + if (!elem.isABSONObj()) { + return false; + } + auto bsonObj = elem.Obj(); + std::string databaseName = bsonObj.getStringField("name"); + return (databaseName == "admin"); + }); + if (adminIter != dbsArray.end()) { + std::iter_swap(adminIter, dbsArray.begin()); + } +} + void DatabasesCloner::_onListDatabaseFinish(const CommandCallbackArgs& cbd) { Status respStatus = cbd.response.status; if (respStatus.isOK()) { @@ -239,24 +277,44 @@ void DatabasesCloner::_onListDatabaseFinish(const CommandCallbackArgs& cbd) { UniqueLock lk(_mutex); if (!respStatus.isOK()) { - LOG(1) << "listDatabases failed: " << respStatus; + LOG(1) << "'listDatabases' failed: " << respStatus; _fail_inlock(&lk, respStatus); return; } - const auto respBSON = cbd.response.data; - // There should not be any cloners yet + // There should not be any cloners yet. invariant(_databaseCloners.size() == 0); - const auto dbsElem = respBSON["databases"].Obj(); - BSONForEach(arrayElement, dbsElem) { + const auto respBSON = cbd.response.data; + + auto databasesArray = _parseListDatabasesResponse(respBSON); + if (!databasesArray.isOK()) { + LOG(1) << "'listDatabases' returned a malformed response: " + << databasesArray.getStatus().toString(); + _fail_inlock(&lk, databasesArray.getStatus()); + return; + } + + auto dbsArray = databasesArray.getValue(); + // Ensure that the 'admin' database is the first element in the array of databases so that it + // will be the first to be cloned. This allows users to authenticate against a database while + // initial sync is occurring. + _setAdminAsFirst(dbsArray); + + for (BSONElement arrayElement : dbsArray) { const BSONObj dbBSON = arrayElement.Obj(); // Check to see if we want to exclude this db from the clone. if (!_includeDbFn(dbBSON)) { - LOG(1) << "excluding db: " << dbBSON; + LOG(1) << "Excluding database from the 'listDatabases' response: " << dbBSON; continue; } + if (!dbBSON.hasField("name")) { + LOG(1) << "Excluding database due to the 'listDatabases' response not containing a " + "'name' field for this entry: " + << dbBSON; + } + const std::string dbName = dbBSON["name"].str(); std::shared_ptr<DatabaseCloner> dbCloner{nullptr}; @@ -321,7 +379,6 @@ void DatabasesCloner::_onListDatabaseFinish(const CommandCallbackArgs& cbd) { // add cloner to list. _databaseCloners.push_back(dbCloner); } - if (_databaseCloners.size() == 0) { if (_status.isOK()) { _succeed_inlock(&lk); diff --git a/src/mongo/db/repl/databases_cloner.h b/src/mongo/db/repl/databases_cloner.h index 53cced9ee4f..13f11f1e2a1 100644 --- a/src/mongo/db/repl/databases_cloner.h +++ b/src/mongo/db/repl/databases_cloner.h @@ -34,6 +34,7 @@ #include "mongo/base/disallow_copying.h" #include "mongo/base/status.h" +#include "mongo/base/status_with.h" #include "mongo/bson/bsonobj.h" #include "mongo/client/fetcher.h" #include "mongo/db/namespace_string.h" @@ -104,6 +105,18 @@ public: */ void setScheduleDbWorkFn_forTest(const CollectionCloner::ScheduleDbWorkFn& scheduleDbWorkFn); + /** + * Calls DatabasesCloner::_setAdminAsFirst. + * For testing only. + */ + void setAdminAsFirst_forTest(std::vector<BSONElement>& dbsArray); + + /** + * Calls DatabasesCloner::_parseListDatabasesResponse. + * For testing only. + */ + StatusWith<std::vector<BSONElement>> parseListDatabasesResponse_forTest(BSONObj dbResponse); + private: bool _isActive_inlock() const; @@ -135,6 +148,23 @@ private: void _onListDatabaseFinish(const CommandCallbackArgs& cbd); + /** + * Takes a vector of BSONElements and scans for an element that contains a 'name' field with the + * value 'admin'. If found, the element is swapped with the first element in the vector. + * Otherwise, return. + * + * Used to parse the BSONResponse returned by listDatabases. + */ + void _setAdminAsFirst(std::vector<BSONElement>& dbsArray); + + /** + * Takes a 'listDatabases' command response and parses the response into a + * vector of BSON elements. + * + * If the input response is malformed, Status ErrorCodes::BadValue will be returned. + */ + StatusWith<std::vector<BSONElement>> _parseListDatabasesResponse(BSONObj dbResponse); + // // All member variables are labeled with one of the following codes indicating the // synchronization rules for accessing them. diff --git a/src/mongo/db/repl/databases_cloner_test.cpp b/src/mongo/db/repl/databases_cloner_test.cpp index 81552201485..e44114e5d58 100644 --- a/src/mongo/db/repl/databases_cloner_test.cpp +++ b/src/mongo/db/repl/databases_cloner_test.cpp @@ -314,6 +314,15 @@ protected: ASSERT_OK(result); }; + std::unique_ptr<DatabasesCloner> makeDummyDatabasesCloner() { + return stdx::make_unique<DatabasesCloner>(&getStorage(), + &getExecutor(), + &getDbWorkThreadPool(), + HostAndPort{"local:1234"}, + [](const BSONObj&) { return true; }, + [](const Status&) {}); + } + private: executor::ThreadPoolMock::Options makeThreadPoolMockOptions() const override; @@ -450,6 +459,87 @@ TEST_F(DBsClonerTest, StartupReturnsInternalErrorAfterSuccessfulStartup) { ASSERT_TRUE(cloner.isActive()); } +TEST_F(DBsClonerTest, ParseAndSetAdminFirstWhenAdminInListDatabasesResponse) { + const Responses responsesWithAdmin = { + {"listDatabases", fromjson("{ok:1, databases:[{name:'a'}, {name:'aab'}, {name:'admin'}]}")}, + {"listDatabases", fromjson("{ok:1, databases:[{name:'admin'}, {name:'a'}, {name:'b'}]}")}, + }; + std::unique_ptr<DatabasesCloner> cloner = makeDummyDatabasesCloner(); + + for (auto&& resp : responsesWithAdmin) { + auto parseResponseStatus = cloner->parseListDatabasesResponse_forTest(resp.second); + ASSERT_TRUE(parseResponseStatus.isOK()); + std::vector<BSONElement> dbNamesArray = parseResponseStatus.getValue(); + cloner->setAdminAsFirst_forTest(dbNamesArray); + ASSERT_EQUALS("admin", dbNamesArray[0].Obj().firstElement().str()); + } +} + +TEST_F(DBsClonerTest, ParseAndAttemptSetAdminFirstWhenAdminNotInListDatabasesResponse) { + const Responses responsesWithoutAdmin = { + {"listDatabases", fromjson("{ok:1, databases:[{name:'a'}, {name:'aab'}, {name:'abc'}]}")}, + {"listDatabases", fromjson("{ok:1, databases:[{name:'foo'}, {name:'a'}, {name:'b'}]}")}, + {"listDatabases", fromjson("{ok:1, databases:[{name:1}, {name:2}, {name:3}]}")}, + }; + std::unique_ptr<DatabasesCloner> cloner = makeDummyDatabasesCloner(); + + for (auto&& resp : responsesWithoutAdmin) { + auto parseResponseStatus = cloner->parseListDatabasesResponse_forTest(resp.second); + ASSERT_TRUE(parseResponseStatus.isOK()); + std::vector<BSONElement> dbNamesArray = parseResponseStatus.getValue(); + std::string expectedResult = dbNamesArray[0].Obj().firstElement().str(); + cloner->setAdminAsFirst_forTest(dbNamesArray); + ASSERT_EQUALS(expectedResult, dbNamesArray[0].Obj().firstElement().str()); + } +} + + +TEST_F(DBsClonerTest, ParseListDatabasesResponseWithMalformedResponses) { + Status expectedResultForNoDatabasesField{ + ErrorCodes::BadValue, + "The 'listDatabases' command response does not contain a databases field."}; + Status expectedResultForNoArrayOfDatabases{ + ErrorCodes::BadValue, + "The 'listDatabases' command response is unable to be transformed into an array."}; + + const Responses responsesWithoutDatabasesField = { + {"listDatabases", fromjson("{ok:1, fake:[{name:'a'}, {name:'aab'}, {name:'foo'}]}")}, + {"listDatabases", fromjson("{ok:1, fake:[{name:'admin'}, {name:'a'}, {name:'b'}]}")}, + }; + + const Responses responsesWithoutArrayOfDatabases = { + {"listDatabases", fromjson("{ok:1, databases:1}")}, + {"listDatabases", fromjson("{ok:1, databases:'abc'}")}, + }; + + const Responses responsesWithInvalidAdminNameField = { + {"listDatabases", fromjson("{ok:1, databases:[{name:'a'}, {name:'aab'}, {fake:'admin'}]}")}, + {"listDatabases", fromjson("{ok:1, databases:[{fake:'admin'}, {name:'a'}, {name:'b'}]}")}, + }; + + std::unique_ptr<DatabasesCloner> cloner = makeDummyDatabasesCloner(); + + for (auto&& resp : responsesWithoutDatabasesField) { + auto parseResponseStatus = cloner->parseListDatabasesResponse_forTest(resp.second); + ASSERT_EQ(parseResponseStatus.getStatus(), expectedResultForNoDatabasesField); + } + + for (auto&& resp : responsesWithoutArrayOfDatabases) { + auto parseResponseStatus = cloner->parseListDatabasesResponse_forTest(resp.second); + ASSERT_EQ(parseResponseStatus.getStatus(), expectedResultForNoArrayOfDatabases); + } + + for (auto&& resp : responsesWithInvalidAdminNameField) { + auto parseResponseStatus = cloner->parseListDatabasesResponse_forTest(resp.second); + ASSERT_TRUE(parseResponseStatus.isOK()); + // We expect no elements to be swapped. + std::vector<BSONElement> dbNamesArray = parseResponseStatus.getValue(); + std::string expectedResult = dbNamesArray[0].Obj().firstElement().str(); + cloner->setAdminAsFirst_forTest(dbNamesArray); + ASSERT_EQUALS(expectedResult, dbNamesArray[0].Obj().firstElement().str()); + } +} + TEST_F(DBsClonerTest, FailsOnListDatabases) { Status result{Status::OK()}; Status expectedResult{ErrorCodes::BadValue, "foo"}; diff --git a/src/mongo/db/repl/initial_syncer.cpp b/src/mongo/db/repl/initial_syncer.cpp index e9a920bb37b..f3bc1a8acd3 100644 --- a/src/mongo/db/repl/initial_syncer.cpp +++ b/src/mongo/db/repl/initial_syncer.cpp @@ -1189,7 +1189,8 @@ Status InitialSyncer::_scheduleLastOplogEntryFetcher_inlock(Fetcher::CallbackFn query, callback, rpc::ServerSelectionMetadata(true, boost::none).toBSON(), - RemoteCommandRequest::kNoTimeout, + RemoteCommandRequest::kNoTimeout /* find network timeout */, + RemoteCommandRequest::kNoTimeout /* getMore network timeout */, RemoteCommandRetryScheduler::makeRetryPolicy( numInitialSyncOplogFindAttempts, executor::RemoteCommandRequest::kNoTimeout, diff --git a/src/mongo/db/repl/master_slave.cpp b/src/mongo/db/repl/master_slave.cpp index 96cfe58efdc..6929d5be297 100644 --- a/src/mongo/db/repl/master_slave.cpp +++ b/src/mongo/db/repl/master_slave.cpp @@ -168,7 +168,7 @@ BSONObj ReplSource::jsobj() { BSONObjBuilder dbsNextPassBuilder; int n = 0; - for (set<string>::iterator i = addDbNextPass.begin(); i != addDbNextPass.end(); i++) { + for (set<std::string>::iterator i = addDbNextPass.begin(); i != addDbNextPass.end(); i++) { n++; dbsNextPassBuilder.appendBool(*i, 1); } @@ -177,7 +177,8 @@ BSONObj ReplSource::jsobj() { BSONObjBuilder incompleteCloneDbsBuilder; n = 0; - for (set<string>::iterator i = incompleteCloneDbs.begin(); i != incompleteCloneDbs.end(); i++) { + for (set<std::string>::iterator i = incompleteCloneDbs.begin(); i != incompleteCloneDbs.end(); + i++) { n++; incompleteCloneDbsBuilder.appendBool(*i, 1); } @@ -188,7 +189,7 @@ BSONObj ReplSource::jsobj() { } void ReplSource::ensureMe(OperationContext* txn) { - string myname = getHostName(); + std::string myname = getHostName(); // local.me is an identifier for a server for getLastError w:2+ bool exists = Helpers::getSingleton(txn, "local.me", _me); @@ -378,10 +379,10 @@ public: } virtual bool run(OperationContext* txn, - const string& ns, + const std::string& ns, BSONObj& cmdObj, int options, - string& errmsg, + std::string& errmsg, BSONObjBuilder& result) { HandshakeArgs handshake; Status status = handshake.initialize(cmdObj); @@ -398,7 +399,7 @@ public: } handshakeCmd; bool replHandshake(DBClientConnection* conn, const OID& myRID) { - string myname = getHostName(); + std::string myname = getHostName(); BSONObjBuilder cmd; cmd.append("handshake", myRID); @@ -450,7 +451,7 @@ void ReplSource::forceResync(OperationContext* txn, const char* requester) { BSONElement e = i.next(); if (e.eoo()) break; - string name = e.embeddedObject().getField("name").valuestr(); + std::string name = e.embeddedObject().getField("name").valuestr(); if (!e.embeddedObject().getBoolField("empty")) { if (name != "local") { if (only.empty() || only == name) { @@ -481,7 +482,7 @@ Status ReplSource::_updateIfDoneWithInitialSync() { return Status::OK(); } -void ReplSource::resyncDrop(OperationContext* txn, const string& dbName) { +void ReplSource::resyncDrop(OperationContext* txn, const std::string& dbName) { log() << "resync: dropping database " << dbName; invariant(txn->lockState()->isW()); @@ -531,13 +532,13 @@ void ReplSource::resync(OperationContext* txn, const std::string& dbName) { static DatabaseIgnorer ___databaseIgnorer; -void DatabaseIgnorer::doIgnoreUntilAfter(const string& db, const Timestamp& futureOplogTime) { +void DatabaseIgnorer::doIgnoreUntilAfter(const std::string& db, const Timestamp& futureOplogTime) { if (futureOplogTime > _ignores[db]) { _ignores[db] = futureOplogTime; } } -bool DatabaseIgnorer::ignoreAt(const string& db, const Timestamp& currentOplogTime) { +bool DatabaseIgnorer::ignoreAt(const std::string& db, const Timestamp& currentOplogTime) { if (_ignores[db].isNull()) { return false; } @@ -627,7 +628,7 @@ bool ReplSource::handleDuplicateDbName(OperationContext* txn, // The database is present on the master and no conflicting databases // are present on the master. Drop any local conflicts. - for (set<string>::const_iterator i = duplicates.begin(); i != duplicates.end(); ++i) { + for (set<std::string>::const_iterator i = duplicates.begin(); i != duplicates.end(); ++i) { ___databaseIgnorer.doIgnoreUntilAfter(*i, lastTime); incompleteCloneDbs.erase(*i); addDbNextPass.erase(*i); @@ -822,10 +823,10 @@ void ReplSource::_sync_pullOpLog_applyOperation(OperationContext* txn, } void ReplSource::syncToTailOfRemoteLog() { - string _ns = ns(); + std::string _ns = ns(); BSONObjBuilder b; if (!only.empty()) { - b.appendRegex("ns", string("^") + pcrecpp::RE::QuoteMeta(only)); + b.appendRegex("ns", std::string("^") + pcrecpp::RE::QuoteMeta(only)); } BSONObj last = oplogReader.findOne(_ns.c_str(), Query(b.done()).sort(BSON("$natural" << -1))); if (!last.isEmpty()) { @@ -873,7 +874,7 @@ public: */ int ReplSource::_sync_pullOpLog(OperationContext* txn, int& nApplied) { int okResultCode = restartSyncAfterSleep; - string ns = string("local.oplog.$") + sourceName(); + std::string ns = std::string("local.oplog.$") + sourceName(); LOG(2) << "sync_pullOpLog " << ns << " syncedTo:" << syncedTo.toStringLong() << '\n'; bool tailing = true; @@ -893,7 +894,7 @@ int ReplSource::_sync_pullOpLog(OperationContext* txn, int& nApplied) { BSONElement e = i.next(); if (e.eoo()) break; - string name = e.embeddedObject().getField("name").valuestr(); + std::string name = e.embeddedObject().getField("name").valuestr(); if (!e.embeddedObject().getBoolField("empty")) { if (name != "local") { if (only.empty() || only == name) { @@ -917,7 +918,7 @@ int ReplSource::_sync_pullOpLog(OperationContext* txn, int& nApplied) { if (!only.empty()) { // note we may here skip a LOT of data table scanning, a lot of work for the master. // maybe append "\\." here? - query.appendRegex("ns", string("^") + pcrecpp::RE::QuoteMeta(only)); + query.appendRegex("ns", std::string("^") + pcrecpp::RE::QuoteMeta(only)); } BSONObj queryObj = query.done(); // e.g. queryObj = { ts: { $gte: syncedTo } } @@ -936,7 +937,7 @@ int ReplSource::_sync_pullOpLog(OperationContext* txn, int& nApplied) { // show any deferred database creates from a previous pass { - set<string>::iterator i = addDbNextPass.begin(); + set<std::string>::iterator i = addDbNextPass.begin(); if (i != addDbNextPass.end()) { BSONObjBuilder b; b.append("ns", *i + '.'); @@ -980,7 +981,7 @@ int ReplSource::_sync_pullOpLog(OperationContext* txn, int& nApplied) { BSONObj op = oplogReader.nextSafe(); BSONElement ts = op.getField("ts"); if (ts.type() != Date && ts.type() != bsonTimestamp) { - string err = op.getStringField("$err"); + std::string err = op.getStringField("$err"); if (!err.empty()) { // 13051 is "tailable cursor requested on non capped collection" if (op.getIntField("code") == 13051) { @@ -1148,7 +1149,7 @@ int ReplSource::sync(OperationContext* txn, int& nApplied) { // FIXME Handle cases where this db isn't on default port, or default port is spec'd in // hostName. - if ((string("localhost") == hostName || string("127.0.0.1") == hostName) && + if ((std::string("localhost") == hostName || std::string("127.0.0.1") == hostName) && serverGlobalParams.port == ServerGlobalParams::DefaultDBPort) { log() << "can't sync from self (localhost). sources configuration may be wrong." << endl; sleepsecs(5); @@ -1293,7 +1294,7 @@ static void replMain(OperationContext* txn) { if (s) { stringstream ss; ss << "sleep " << s << " sec before next pass"; - string msg = ss.str(); + std::string msg = ss.str(); if (!serverGlobalParams.quiet) log() << msg << endl; ReplInfo r(msg.c_str()); diff --git a/src/mongo/db/repl/oplog.cpp b/src/mongo/db/repl/oplog.cpp index 43a1eba5042..76df271ed78 100644 --- a/src/mongo/db/repl/oplog.cpp +++ b/src/mongo/db/repl/oplog.cpp @@ -116,6 +116,11 @@ namespace { // cached copy...so don't rename, drop, etc.!!! Collection* _localOplogCollection = nullptr; +// Specifies whether we abort initial sync when attempting to apply a renameCollection operation. +// If set to true, users risk corrupting their data. This should only be enabled by expert users +// of the server who understand the risks this poses. +MONGO_EXPORT_SERVER_PARAMETER(allowUnsafeRenamesDuringInitialSync, bool, false); + PseudoRandom hashGenerator(std::unique_ptr<SecureRandom>(SecureRandom::create())->nextInt64()); // Synchronizes the section where a new Timestamp is generated and when it actually @@ -663,6 +668,45 @@ std::map<std::string, ApplyOpMetadata> opsMap = { } // namespace +std::pair<BSONObj, NamespaceString> prepForApplyOpsIndexInsert(const BSONElement& fieldO, + const BSONObj& op, + const NamespaceString& requestNss) { + uassert(ErrorCodes::NoSuchKey, + str::stream() << "Missing expected index spec in field 'o': " << op, + !fieldO.eoo()); + uassert(ErrorCodes::TypeMismatch, + str::stream() << "Expected object for index spec in field 'o': " << op, + fieldO.isABSONObj()); + BSONObj indexSpec = fieldO.embeddedObject(); + + std::string indexNs; + uassertStatusOK(bsonExtractStringField(indexSpec, "ns", &indexNs)); + const NamespaceString indexNss(indexNs); + uassert(ErrorCodes::InvalidNamespace, + str::stream() << "Invalid namespace in index spec: " << op, + indexNss.isValid()); + uassert(ErrorCodes::InvalidNamespace, + str::stream() << "Database name mismatch for database (" << requestNss.db() + << ") while creating index: " + << op, + requestNss.db() == indexNss.db()); + + if (!indexSpec["v"]) { + // If the "v" field isn't present in the index specification, then we assume it is a + // v=1 index from an older version of MongoDB. This is because + // (1) we haven't built v=0 indexes as the default for a long time, and + // (2) the index version has been included in the corresponding oplog entry since + // v=2 indexes were introduced. + BSONObjBuilder bob; + + bob.append("v", static_cast<int>(IndexVersion::kV1)); + bob.appendElements(indexSpec); + + indexSpec = bob.obj(); + } + + return std::make_pair(indexSpec, indexNss); +} // @return failure status if an update should have happened and the document DNE. // See replset initial sync code. Status applyOperation_inlock(OperationContext* txn, @@ -688,6 +732,7 @@ Status applyOperation_inlock(OperationContext* txn, o = fieldO.embeddedObject(); const StringData ns = fieldNs.valueStringData(); + NamespaceString requestNss{ns}; BSONObj o2; if (fieldO2.isABSONObj()) @@ -718,27 +763,11 @@ Status applyOperation_inlock(OperationContext* txn, invariant(*opType != 'c'); // commands are processed in applyCommand_inlock() if (*opType == 'i') { - if (nsToCollectionSubstring(ns) == "system.indexes") { - uassert(ErrorCodes::NoSuchKey, - str::stream() << "Missing expected index spec in field 'o': " << op, - !fieldO.eoo()); - uassert(ErrorCodes::TypeMismatch, - str::stream() << "Expected object for index spec in field 'o': " << op, - fieldO.isABSONObj()); - BSONObj indexSpec = fieldO.embeddedObject(); - - std::string indexNs; - uassertStatusOK(bsonExtractStringField(indexSpec, "ns", &indexNs)); - const NamespaceString indexNss(indexNs); - uassert(ErrorCodes::InvalidNamespace, - str::stream() << "Invalid namespace in index spec: " << op, - indexNss.isValid()); - uassert(ErrorCodes::InvalidNamespace, - str::stream() << "Database name mismatch for database (" - << nsToDatabaseSubstring(ns) - << ") while creating index: " - << op, - nsToDatabaseSubstring(ns) == indexNss.db()); + if (requestNss.isSystemDotIndexes()) { + BSONObj indexSpec; + NamespaceString indexNss; + std::tie(indexSpec, indexNss) = + repl::prepForApplyOpsIndexInsert(fieldO, op, requestNss); // Check if collection exists. auto indexCollection = db->getCollection(indexNss); @@ -749,20 +778,6 @@ Status applyOperation_inlock(OperationContext* txn, opCounters->gotInsert(); - if (!indexSpec["v"]) { - // If the "v" field isn't present in the index specification, then we assume it is a - // v=1 index from an older version of MongoDB. This is because - // (1) we haven't built v=0 indexes as the default for a long time, and - // (2) the index version has been included in the corresponding oplog entry since - // v=2 indexes were introduced. - BSONObjBuilder bob; - - bob.append("v", static_cast<int>(IndexVersion::kV1)); - bob.appendElements(indexSpec); - - indexSpec = bob.obj(); - } - bool relaxIndexConstraints = ReplicationCoordinator::get(txn)->shouldRelaxIndexConstraints(indexNss); if (indexSpec["background"].trueValue()) { @@ -863,13 +878,12 @@ Status applyOperation_inlock(OperationContext* txn, BSONObjBuilder b; b.append(o.getField("_id")); - const NamespaceString requestNs(ns); - UpdateRequest request(requestNs); + UpdateRequest request(requestNss); request.setQuery(b.done()); request.setUpdates(o); request.setUpsert(); - UpdateLifecycleImpl updateLifecycle(requestNs); + UpdateLifecycleImpl updateLifecycle(requestNss); request.setLifecycle(&updateLifecycle); UpdateResult res = update(txn, db, request); @@ -894,13 +908,12 @@ Status applyOperation_inlock(OperationContext* txn, str::stream() << "Failed to apply update due to missing _id: " << op.toString(), updateCriteria.hasField("_id")); - const NamespaceString requestNs(ns); - UpdateRequest request(requestNs); + UpdateRequest request(requestNss); request.setQuery(updateCriteria); request.setUpdates(o); request.setUpsert(upsert); - UpdateLifecycleImpl updateLifecycle(requestNs); + UpdateLifecycleImpl updateLifecycle(requestNss); request.setLifecycle(&updateLifecycle); UpdateResult ur = update(txn, db, request); @@ -954,7 +967,12 @@ Status applyOperation_inlock(OperationContext* txn, o.hasField("_id")); if (opType[1] == 0) { - deleteObjects(txn, collection, ns, o, PlanExecutor::YIELD_MANUAL, /*justOne*/ valueB); + deleteObjects(txn, + collection, + requestNss.ns().c_str(), + o, + PlanExecutor::YIELD_MANUAL, + /*justOne*/ valueB); } else verify(opType[1] == 'b'); // "db" advertisement if (incrementOpsAppliedStats) { @@ -970,16 +988,6 @@ Status applyOperation_inlock(OperationContext* txn, 14825, str::stream() << "error in applyOperation : unknown opType " << *opType); } - // AuthorizationManager's logOp method registers a RecoveryUnit::Change and to do so we need - // to a new WriteUnitOfWork, if we dont have a wrapping unit of work already. If we already - // have a wrapping WUOW, the extra nexting is harmless. The logOp really should have been - // done in the WUOW that did the write, but this won't happen because applyOps turns off - // observers. - WriteUnitOfWork wuow(txn); - getGlobalAuthorizationManager()->logOp( - txn, opType, ns.toString().c_str(), o, fieldO2.isABSONObj() ? &o2 : NULL); - wuow.commit(); - return Status::OK(); } @@ -1021,9 +1029,15 @@ Status applyCommand_inlock(OperationContext* txn, // Applying renameCollection during initial sync might lead to data corruption, so we restart // the initial sync. if (!inSteadyStateReplication && o.firstElementFieldName() == std::string("renameCollection")) { - return Status(ErrorCodes::OplogOperationUnsupported, - str::stream() << "Applying renameCollection not supported in initial sync: " - << redact(op)); + if (!allowUnsafeRenamesDuringInitialSync.load()) { + return Status(ErrorCodes::OplogOperationUnsupported, + str::stream() + << "Applying renameCollection not supported in initial sync: " + << redact(op)); + } + warning() << "allowUnsafeRenamesDuringInitialSync set to true. Applying renameCollection " + "operation during initial sync even though it may lead to data corruption: " + << redact(op); } // Applying commands in repl is done under Global W-lock, so it is safe to not @@ -1071,11 +1085,8 @@ Status applyCommand_inlock(OperationContext* txn, break; } default: - if (_oplogCollectionName == masterSlaveOplogName) { - error() << "Failed command " << redact(o) << " on " << nss.db() - << " with status " << status << " during oplog application"; - } else if (curOpToApply.acceptableErrors.find(status.code()) == - curOpToApply.acceptableErrors.end()) { + if (curOpToApply.acceptableErrors.find(status.code()) == + curOpToApply.acceptableErrors.end()) { error() << "Failed command " << redact(o) << " on " << nss.db() << " with status " << status << " during oplog application"; return status; diff --git a/src/mongo/db/repl/oplog.h b/src/mongo/db/repl/oplog.h index b2078e93d28..f51aadb8dcf 100644 --- a/src/mongo/db/repl/oplog.h +++ b/src/mongo/db/repl/oplog.h @@ -104,6 +104,15 @@ void oplogCheckCloseDatabase(OperationContext* txn, Database* db); using IncrementOpsAppliedStatsFn = stdx::function<void()>; /** + * Take the object field of a BSONObj, the BSONObj, and the namespace of + * the operation and perform necessary validation to ensure the BSONObj is a + * properly-formed command to insert into system.indexes. This is only to + * be used for insert operations into system.indexes. It is called via applyOps. + */ +std::pair<BSONObj, NamespaceString> prepForApplyOpsIndexInsert(const BSONElement& fieldO, + const BSONObj& op, + const NamespaceString& requestNss); +/** * Take a non-command op and apply it locally * Used for applying from an oplog * @param inSteadyStateReplication convert some updates to upserts for idempotency reasons diff --git a/src/mongo/db/repl/oplog_fetcher.cpp b/src/mongo/db/repl/oplog_fetcher.cpp index ece0ef266fd..c9b96ab77f7 100644 --- a/src/mongo/db/repl/oplog_fetcher.cpp +++ b/src/mongo/db/repl/oplog_fetcher.cpp @@ -36,6 +36,7 @@ #include "mongo/db/commands/server_status_metric.h" #include "mongo/db/jsobj.h" #include "mongo/db/repl/replication_coordinator.h" +#include "mongo/db/server_parameters.h" #include "mongo/db/stats/timer_stats.h" #include "mongo/rpc/metadata/oplog_query_metadata.h" #include "mongo/rpc/metadata/server_selection_metadata.h" @@ -56,8 +57,12 @@ MONGO_FP_DECLARE(stopReplProducer); namespace { -Seconds kOplogInitialFindMaxTime{60}; -Seconds kOplogQueryNetworkTimeout{65}; // 5 seconds past the find command's 1 minute maxTimeMs +// Number of seconds for the `maxTimeMS` on the initial `find` command. +MONGO_EXPORT_SERVER_PARAMETER(oplogInitialFindMaxSeconds, int, 60); + +// Number of milliseconds to add to the `find` and `getMore` timeouts to calculate the network +// timeout for the requests. +const Milliseconds kNetworkTimeoutBufferMS{5000}; Counter64 readersCreatedStats; ServerStatusMetricField<Counter64> displayReadersCreated("repl.network.readersCreated", @@ -91,17 +96,22 @@ Milliseconds calculateAwaitDataTimeout(const ReplSetConfig& config) { */ BSONObj makeFindCommandObject(const NamespaceString& nss, long long currentTerm, - OpTime lastOpTimeFetched) { + OpTime lastOpTimeFetched, + Milliseconds fetcherMaxTimeMS) { BSONObjBuilder cmdBob; cmdBob.append("find", nss.coll()); cmdBob.append("filter", BSON("ts" << BSON("$gte" << lastOpTimeFetched.getTimestamp()))); cmdBob.append("tailable", true); cmdBob.append("oplogReplay", true); cmdBob.append("awaitData", true); - cmdBob.append("maxTimeMS", durationCount<Milliseconds>(kOplogInitialFindMaxTime)); + cmdBob.append("maxTimeMS", durationCount<Milliseconds>(fetcherMaxTimeMS)); if (currentTerm != OpTime::kUninitializedTerm) { cmdBob.append("term", currentTerm); } + if (serverGlobalParams.featureCompatibility.version.load() == + ServerGlobalParams::FeatureCompatibility::Version::k34) { + cmdBob.append("readConcern", BSON("afterOpTime" << lastOpTimeFetched.toBSON())); + } return cmdBob.obj(); } @@ -450,6 +460,14 @@ BSONObj OplogFetcher::getMetadataObject_forTest() const { } Milliseconds OplogFetcher::getAwaitDataTimeout_forTest() const { + return _getGetMoreMaxTime(); +} + +Milliseconds OplogFetcher::_getFindMaxTime() const { + return Milliseconds(oplogInitialFindMaxSeconds.load() * 1000); +} + +Milliseconds OplogFetcher::_getGetMoreMaxTime() const { return _awaitDataTimeout; } @@ -660,7 +678,7 @@ void OplogFetcher::_callback(const Fetcher::QueryResponseStatus& result, getMoreBob->appendElements(makeGetMoreCommandObject(queryResponse.nss, queryResponse.cursorId, lastCommittedWithCurrentTerm, - _awaitDataTimeout)); + _getGetMoreMaxTime())); } void OplogFetcher::_finishCallback(Status status) { @@ -691,10 +709,11 @@ std::unique_ptr<Fetcher> OplogFetcher::_makeFetcher(long long currentTerm, _executor, _source, _nss.db().toString(), - makeFindCommandObject(_nss, currentTerm, lastFetchedOpTime), + makeFindCommandObject(_nss, currentTerm, lastFetchedOpTime, _getFindMaxTime()), stdx::bind(&OplogFetcher::_callback, this, stdx::placeholders::_1, stdx::placeholders::_3), _metadataObject, - kOplogQueryNetworkTimeout); + _getFindMaxTime() + kNetworkTimeoutBufferMS, + _getGetMoreMaxTime() + kNetworkTimeoutBufferMS); } bool OplogFetcher::_isShuttingDown() const { diff --git a/src/mongo/db/repl/oplog_fetcher.h b/src/mongo/db/repl/oplog_fetcher.h index ad4fde296b1..54bfbabbf8d 100644 --- a/src/mongo/db/repl/oplog_fetcher.h +++ b/src/mongo/db/repl/oplog_fetcher.h @@ -235,6 +235,16 @@ private: void _finishCallback(Status status, OpTimeWithHash opTimeWithHash); /** + * Returns how long the `find` command should wait before timing out. + */ + virtual Milliseconds _getFindMaxTime() const; + + /** + * Returns how long the `getMore` command should wait before timing out. + */ + virtual Milliseconds _getGetMoreMaxTime() const; + + /** * Creates a new instance of the fetcher to tail the remote oplog starting at the given optime. */ std::unique_ptr<Fetcher> _makeFetcher(long long currentTerm, OpTime lastFetchedOpTime); diff --git a/src/mongo/db/repl/repl_client_info.cpp b/src/mongo/db/repl/repl_client_info.cpp index 3e98a0cb9d3..5b77e3aa6f3 100644 --- a/src/mongo/db/repl/repl_client_info.cpp +++ b/src/mongo/db/repl/repl_client_info.cpp @@ -26,6 +26,8 @@ * it in the license file. */ +#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kReplication + #include "mongo/platform/basic.h" #include "mongo/db/repl/repl_client_info.h" @@ -36,6 +38,7 @@ #include "mongo/db/operation_context.h" #include "mongo/db/repl/replication_coordinator_global.h" #include "mongo/util/decorable.h" +#include "mongo/util/log.h" namespace mongo { namespace repl { @@ -48,10 +51,23 @@ void ReplClientInfo::setLastOp(const OpTime& ot) { _lastOp = ot; } + void ReplClientInfo::setLastOpToSystemLastOpTime(OperationContext* txn) { ReplicationCoordinator* replCoord = repl::ReplicationCoordinator::get(txn->getServiceContext()); if (replCoord->isReplEnabled() && txn->writesAreReplicated()) { - setLastOp(replCoord->getMyLastAppliedOpTime()); + auto systemOpTime = replCoord->getMyLastAppliedOpTime(); + + // If the system optime has gone backwards, that must mean that there was a rollback. + // This is safe, but the last op for a Client should never go backwards, so just leave + // the last op for this Client as it was. + if (systemOpTime >= _lastOp) { + _lastOp = systemOpTime; + } else { + log() << "Not setting the last OpTime for this Client from " << _lastOp + << " to the current system time of " << systemOpTime + << " as that would be moving the OpTime backwards. This should only happen if " + "there was a rollback recently"; + } } } diff --git a/src/mongo/db/repl/repl_set_config.cpp b/src/mongo/db/repl/repl_set_config.cpp index 1ad2d245ac6..024c3ff04a5 100644 --- a/src/mongo/db/repl/repl_set_config.cpp +++ b/src/mongo/db/repl/repl_set_config.cpp @@ -35,11 +35,19 @@ #include "mongo/bson/util/bson_check.h" #include "mongo/bson/util/bson_extract.h" #include "mongo/db/jsobj.h" +#include "mongo/db/mongod_options.h" #include "mongo/db/server_options.h" +#include "mongo/db/server_parameters.h" #include "mongo/stdx/functional.h" #include "mongo/util/stringutils.h" namespace mongo { +/** + * Dont run any sharding validations. Can not be combined with --configsvr or shardvr. Intended to + * allow restarting config server or shard as an independent replica set. + */ +MONGO_EXPORT_STARTUP_SERVER_PARAMETER(skipShardingConfigurationChecks, bool, false); + namespace repl { const size_t ReplSetConfig::kMaxMembers; @@ -552,7 +560,8 @@ Status ReplSetConfig::validate() const { "servers cannot have a non-zero slaveDelay"); } } - if (serverGlobalParams.clusterRole != ClusterRole::ConfigServer) { + if (serverGlobalParams.clusterRole != ClusterRole::ConfigServer && + !skipShardingConfigurationChecks) { return Status(ErrorCodes::BadValue, "Nodes being used for config servers must be started with the " "--configsvr flag"); diff --git a/src/mongo/db/repl/replication_coordinator_external_state_impl.cpp b/src/mongo/db/repl/replication_coordinator_external_state_impl.cpp index cfc6dbb55e1..05d070c08fe 100644 --- a/src/mongo/db/repl/replication_coordinator_external_state_impl.cpp +++ b/src/mongo/db/repl/replication_coordinator_external_state_impl.cpp @@ -240,8 +240,11 @@ void ReplicationCoordinatorExternalStateImpl::startSteadyStateReplication( _applierThread->startup(); log() << "Starting replication reporter thread"; invariant(!_syncSourceFeedbackThread); - _syncSourceFeedbackThread.reset(new stdx::thread(stdx::bind( - &SyncSourceFeedback::run, &_syncSourceFeedback, _taskExecutor.get(), _bgSync.get()))); + _syncSourceFeedbackThread.reset(new stdx::thread(stdx::bind(&SyncSourceFeedback::run, + &_syncSourceFeedback, + _taskExecutor.get(), + _bgSync.get(), + replCoord))); } void ReplicationCoordinatorExternalStateImpl::stopDataReplication(OperationContext* txn) { diff --git a/src/mongo/db/repl/replication_coordinator_impl.cpp b/src/mongo/db/repl/replication_coordinator_impl.cpp index 3de3d35bb61..ad1a9fd93ba 100644 --- a/src/mongo/db/repl/replication_coordinator_impl.cpp +++ b/src/mongo/db/repl/replication_coordinator_impl.cpp @@ -2922,6 +2922,26 @@ ReplicationCoordinatorImpl::_setCurrentRSConfig_inlock(const ReplSetConfig& newC const ReplSetConfig oldConfig = _rsConfig; _rsConfig = newConfig; _protVersion.store(_rsConfig.getProtocolVersion()); + + // Warn if running --nojournal and writeConcernMajorityJournalDefault = false + StorageEngine* storageEngine = getGlobalServiceContext()->getGlobalStorageEngine(); + if (storageEngine && !storageEngine->isDurable() && + (newConfig.getWriteConcernMajorityShouldJournal() && + (!oldConfig.isInitialized() || !oldConfig.getWriteConcernMajorityShouldJournal()))) { + log() << startupWarningsLog; + log() << "** WARNING: This replica set is running without journaling enabled but the " + << startupWarningsLog; + log() << "** writeConcernMajorityJournalDefault option to the replica set config " + << startupWarningsLog; + log() << "** is set to true. The writeConcernMajorityJournalDefault " + << startupWarningsLog; + log() << "** option to the replica set config must be set to false " + << startupWarningsLog; + log() << "** or w:majority write concerns will never complete." + << startupWarningsLog; + log() << startupWarningsLog; + } + log() << "New replica set config in use: " << _rsConfig.toBSON() << rsLog; _selfIndex = myIndex; if (_selfIndex >= 0) { diff --git a/src/mongo/db/repl/replication_coordinator_impl_elect_v1.cpp b/src/mongo/db/repl/replication_coordinator_impl_elect_v1.cpp index fbd976ec6b7..1dd9bf5b7c3 100644 --- a/src/mongo/db/repl/replication_coordinator_impl_elect_v1.cpp +++ b/src/mongo/db/repl/replication_coordinator_impl_elect_v1.cpp @@ -133,7 +133,9 @@ void ReplicationCoordinatorImpl::_startElectSelfV1() { return; } - log() << "conducting a dry run election to see if we could be elected"; + long long term = _topCoord->getTerm(); + + log() << "conducting a dry run election to see if we could be elected. current term: " << term; _voteRequester.reset(new VoteRequester); // This is necessary because the voteRequester may call directly into winning an @@ -141,12 +143,11 @@ void ReplicationCoordinatorImpl::_startElectSelfV1() { // _mutex again. lk.unlock(); - long long term = _topCoord->getTerm(); StatusWith<ReplicationExecutor::EventHandle> nextPhaseEvh = _voteRequester->start(&_replExecutor, _rsConfig, _selfIndex, - _topCoord->getTerm(), + term, true, // dry run lastOpTime); if (nextPhaseEvh.getStatus() == ErrorCodes::ShutdownInProgress) { @@ -165,7 +166,8 @@ void ReplicationCoordinatorImpl::_onDryRunComplete(long long originalTerm) { LockGuard lk(_topoMutex); if (_topCoord->getTerm() != originalTerm) { - log() << "not running for primary, we have been superceded already"; + log() << "not running for primary, we have been superseded already during dry run. " + << "original term: " << originalTerm << ", current term: " << _topCoord->getTerm(); return; } @@ -175,23 +177,24 @@ void ReplicationCoordinatorImpl::_onDryRunComplete(long long originalTerm) { log() << "not running for primary, we received insufficient votes"; return; } else if (endResult == VoteRequester::Result::kStaleTerm) { - log() << "not running for primary, we have been superceded already"; + log() << "not running for primary, we have been superseded already"; return; } else if (endResult != VoteRequester::Result::kSuccessfullyElected) { log() << "not running for primary, we received an unexpected problem"; return; } - log() << "dry election run succeeded, running for election"; + long long newTerm = originalTerm + 1; + log() << "dry election run succeeded, running for election in term " << newTerm; // Stepdown is impossible from this term update. TopologyCoordinator::UpdateTermResult updateTermResult; - _updateTerm_incallback(originalTerm + 1, &updateTermResult); + _updateTerm_incallback(newTerm, &updateTermResult); invariant(updateTermResult == TopologyCoordinator::UpdateTermResult::kUpdatedTerm); // Secure our vote for ourself first _topCoord->voteForMyselfV1(); // Store the vote in persistent storage. - LastVote lastVote{originalTerm + 1, _selfIndex}; + LastVote lastVote{newTerm, _selfIndex}; auto cbStatus = _replExecutor.scheduleDBWork( [this, lastVote](const ReplicationExecutor::CallbackArgs& cbData) { @@ -232,12 +235,19 @@ void ReplicationCoordinatorImpl::_startVoteRequester(long long newTerm) { LockGuard lk(_topoMutex); + if (_topCoord->getTerm() != newTerm) { + log() << "not running for primary, we have been superseded already while writing our last " + "vote. election term: " + << newTerm << ", current term: " << _topCoord->getTerm(); + return; + } + const auto lastOpTime = _isDurableStorageEngine() ? getMyLastDurableOpTime() : getMyLastAppliedOpTime(); _voteRequester.reset(new VoteRequester); - StatusWith<ReplicationExecutor::EventHandle> nextPhaseEvh = _voteRequester->start( - &_replExecutor, _rsConfig, _selfIndex, _topCoord->getTerm(), false, lastOpTime); + StatusWith<ReplicationExecutor::EventHandle> nextPhaseEvh = + _voteRequester->start(&_replExecutor, _rsConfig, _selfIndex, newTerm, false, lastOpTime); if (nextPhaseEvh.getStatus() == ErrorCodes::ShutdownInProgress) { return; } @@ -249,14 +259,15 @@ void ReplicationCoordinatorImpl::_startVoteRequester(long long newTerm) { lossGuard.dismiss(); } -void ReplicationCoordinatorImpl::_onVoteRequestComplete(long long originalTerm) { +void ReplicationCoordinatorImpl::_onVoteRequestComplete(long long newTerm) { invariant(_voteRequester); LoseElectionGuardV1 lossGuard(this); LockGuard lk(_topoMutex); - if (_topCoord->getTerm() != originalTerm) { - log() << "not becoming primary, we have been superceded already"; + if (_topCoord->getTerm() != newTerm) { + log() << "not becoming primary, we have been superseded already during election. " + << "election term: " << newTerm << ", current term: " << _topCoord->getTerm(); return; } @@ -267,7 +278,7 @@ void ReplicationCoordinatorImpl::_onVoteRequestComplete(long long originalTerm) log() << "not becoming primary, we received insufficient votes"; return; case VoteRequester::Result::kStaleTerm: - log() << "not becoming primary, we have been superceded already"; + log() << "not becoming primary, we have been superseded already"; return; case VoteRequester::Result::kSuccessfullyElected: log() << "election succeeded, assuming primary role in term " << _topCoord->getTerm(); diff --git a/src/mongo/db/repl/replication_coordinator_impl_elect_v1_test.cpp b/src/mongo/db/repl/replication_coordinator_impl_elect_v1_test.cpp index 6e87f882b48..0f550869546 100644 --- a/src/mongo/db/repl/replication_coordinator_impl_elect_v1_test.cpp +++ b/src/mongo/db/repl/replication_coordinator_impl_elect_v1_test.cpp @@ -431,7 +431,7 @@ TEST_F(ReplCoordTest, ElectionFailsWhenDryRunResponseContainsANewerTerm) { getReplCoord()->waitForElectionFinish_forTest(); stopCapturingLogMessages(); ASSERT_EQUALS( - 1, countLogLinesContaining("not running for primary, we have been superceded already")); + 1, countLogLinesContaining("not running for primary, we have been superseded already")); } TEST_F(ReplCoordTest, NodeWillNotStandForElectionDuringHeartbeatReconfig) { @@ -689,7 +689,7 @@ TEST_F(ReplCoordTest, ElectionFailsWhenVoteRequestResponseContainsANewerTerm) { getReplCoord()->waitForElectionFinish_forTest(); stopCapturingLogMessages(); ASSERT_EQUALS(1, - countLogLinesContaining("not becoming primary, we have been superceded already")); + countLogLinesContaining("not becoming primary, we have been superseded already")); } TEST_F(ReplCoordTest, ElectionFailsWhenTermChangesDuringDryRun) { @@ -728,8 +728,9 @@ TEST_F(ReplCoordTest, ElectionFailsWhenTermChangesDuringDryRun) { simulateSuccessfulDryRun(onDryRunRequest); stopCapturingLogMessages(); - ASSERT_EQUALS( - 1, countLogLinesContaining("not running for primary, we have been superceded already")); + ASSERT_EQUALS(1, + countLogLinesContaining( + "not running for primary, we have been superseded already during dry run")); } TEST_F(ReplCoordTest, ElectionFailsWhenTermChangesDuringActualElection) { @@ -784,7 +785,7 @@ TEST_F(ReplCoordTest, ElectionFailsWhenTermChangesDuringActualElection) { getReplCoord()->waitForElectionFinish_forTest(); stopCapturingLogMessages(); ASSERT_EQUALS(1, - countLogLinesContaining("not becoming primary, we have been superceded already")); + countLogLinesContaining("not becoming primary, we have been superseded already")); } class PriorityTakeoverTest : public ReplCoordTest { diff --git a/src/mongo/db/repl/replication_coordinator_impl_heartbeat.cpp b/src/mongo/db/repl/replication_coordinator_impl_heartbeat.cpp index b72d7e7dbcc..66c9799029d 100644 --- a/src/mongo/db/repl/replication_coordinator_impl_heartbeat.cpp +++ b/src/mongo/db/repl/replication_coordinator_impl_heartbeat.cpp @@ -778,18 +778,24 @@ void ReplicationCoordinatorImpl::_scheduleNextLivenessUpdate_inlock() { } auto nextTimeout = earliestDate + _rsConfig.getElectionTimeoutPeriod(); - if (nextTimeout > _replExecutor.now()) { - LOG(3) << "scheduling next check at " << nextTimeout; - auto cbh = _scheduleWorkAt(nextTimeout, - stdx::bind(&ReplicationCoordinatorImpl::_handleLivenessTimeout, - this, - stdx::placeholders::_1)); - if (!cbh) { - return; - } - _handleLivenessTimeoutCbh = cbh; - _earliestMemberId = earliestMemberId; + LOG(3) << "scheduling next check at " << nextTimeout; + + // It is possible we will schedule the next timeout in the past. + // ReplicationExecutor::_scheduleWorkAt() schedules its work immediately if it's given a + // time <= now(). + // If we missed the timeout, it means that on our last check the earliest live member was + // just barely fresh and it has become stale since then. We must schedule another liveness + // check to continue conducting liveness checks and be able to step down from primary if we + // lose contact with a majority of nodes. + auto cbh = _scheduleWorkAt(nextTimeout, + stdx::bind(&ReplicationCoordinatorImpl::_handleLivenessTimeout, + this, + stdx::placeholders::_1)); + if (!cbh) { + return; } + _handleLivenessTimeoutCbh = cbh; + _earliestMemberId = earliestMemberId; } void ReplicationCoordinatorImpl::_cancelAndRescheduleLivenessUpdate_inlock(int updatedMemberId) { diff --git a/src/mongo/db/repl/replication_executor_test.cpp b/src/mongo/db/repl/replication_executor_test.cpp index ba41df46769..2e8a66465de 100644 --- a/src/mongo/db/repl/replication_executor_test.cpp +++ b/src/mongo/db/repl/replication_executor_test.cpp @@ -249,6 +249,21 @@ TEST_F(ReplicationExecutorTest, ScheduleCallbackAtNow) { executor.waitForEvent(finishEvent); } +TEST_F(ReplicationExecutorTest, ScheduleCallbackInPast) { + launchExecutorThread(); + getNet()->exitNetwork(); + + ReplicationExecutor& executor = getReplExecutor(); + auto finishEvent = assertGet(executor.makeEvent()); + auto fn = [&executor, finishEvent](const ReplicationExecutor::CallbackArgs& cbData) { + ASSERT_OK(cbData.status); + executor.signalEvent(finishEvent); + }; + + auto cb = executor.scheduleWorkAt(getNet()->now() - Milliseconds(1000), fn); + executor.waitForEvent(finishEvent); +} + TEST_F(ReplicationExecutorTest, ScheduleCallbackAtAFutureTime) { launchExecutorThread(); getNet()->exitNetwork(); diff --git a/src/mongo/db/repl/sync_source_feedback.cpp b/src/mongo/db/repl/sync_source_feedback.cpp index 4d395e376e6..a82b1553212 100644 --- a/src/mongo/db/repl/sync_source_feedback.cpp +++ b/src/mongo/db/repl/sync_source_feedback.cpp @@ -50,27 +50,23 @@ namespace repl { namespace { /** - * Calculates the keep alive interval based on the current configuration in the replication - * coordinator. + * Calculates the keep alive interval based on the given ReplSetConfig. */ -Milliseconds calculateKeepAliveInterval(OperationContext* txn, stdx::mutex& mtx) { - stdx::lock_guard<stdx::mutex> lock(mtx); - auto replCoord = repl::ReplicationCoordinator::get(txn); - auto rsConfig = replCoord->getConfig(); - auto keepAliveInterval = rsConfig.getElectionTimeoutPeriod() / 2; - return keepAliveInterval; +Milliseconds calculateKeepAliveInterval(const ReplSetConfig& rsConfig) { + return rsConfig.getElectionTimeoutPeriod() / 2; } /** * Returns function to prepare update command */ Reporter::PrepareReplSetUpdatePositionCommandFn makePrepareReplSetUpdatePositionCommandFn( - OperationContext* txn, + ReplicationCoordinator* replCoord, stdx::mutex& mtx, const HostAndPort& syncTarget, BackgroundSync* bgsync) { - return [&mtx, syncTarget, txn, bgsync](ReplicationCoordinator::ReplSetUpdatePositionCommandStyle - commandStyle) -> StatusWith<BSONObj> { + return [&mtx, syncTarget, replCoord, bgsync]( + ReplicationCoordinator::ReplSetUpdatePositionCommandStyle + commandStyle) -> StatusWith<BSONObj> { auto currentSyncTarget = bgsync->getSyncTarget(); if (currentSyncTarget != syncTarget) { if (currentSyncTarget.empty()) { @@ -86,7 +82,6 @@ Reporter::PrepareReplSetUpdatePositionCommandFn makePrepareReplSetUpdatePosition } } - auto replCoord = repl::ReplicationCoordinator::get(txn); if (replCoord->getMemberState().primary()) { // Primary has no one to send updates to. return Status(ErrorCodes::InvalidSyncSource, @@ -114,7 +109,7 @@ void SyncSourceFeedback::forwardSlaveProgress() { } } -Status SyncSourceFeedback::_updateUpstream(OperationContext* txn, +Status SyncSourceFeedback::_updateUpstream(ReplicationCoordinator* replCoord, BackgroundSync* bgsync, Reporter* reporter) { auto syncTarget = reporter->getTarget(); @@ -139,7 +134,6 @@ Status SyncSourceFeedback::_updateUpstream(OperationContext* txn, } else { // Blacklist sync target for .5 seconds and find a new one. stdx::lock_guard<stdx::mutex> lock(_mtx); - auto replCoord = repl::ReplicationCoordinator::get(txn); const auto blacklistDuration = Milliseconds{500}; const auto until = Date_t::now() + blacklistDuration; log() << "Blacklisting " << syncTarget << " due to error: '" << status << "' for " @@ -161,7 +155,9 @@ void SyncSourceFeedback::shutdown() { _cond.notify_all(); } -void SyncSourceFeedback::run(executor::TaskExecutor* executor, BackgroundSync* bgsync) { +void SyncSourceFeedback::run(executor::TaskExecutor* executor, + BackgroundSync* bgsync, + ReplicationCoordinator* replCoord) { Client::initThread("SyncSourceFeedback"); HostAndPort syncTarget; @@ -170,10 +166,9 @@ void SyncSourceFeedback::run(executor::TaskExecutor* executor, BackgroundSync* b Milliseconds keepAliveInterval(0); while (true) { // breaks once _shutdownSignaled is true - auto txn = cc().makeOperationContext(); if (keepAliveInterval == Milliseconds(0)) { - keepAliveInterval = calculateKeepAliveInterval(txn.get(), _mtx); + keepAliveInterval = calculateKeepAliveInterval(replCoord->getConfig()); } { @@ -189,7 +184,7 @@ void SyncSourceFeedback::run(executor::TaskExecutor* executor, BackgroundSync* b continue; } } - MemberState state = ReplicationCoordinator::get(txn.get())->getMemberState(); + MemberState state = replCoord->getMemberState(); if (!(state.primary() || state.startup())) { break; } @@ -204,7 +199,7 @@ void SyncSourceFeedback::run(executor::TaskExecutor* executor, BackgroundSync* b { stdx::lock_guard<stdx::mutex> lock(_mtx); - MemberState state = ReplicationCoordinator::get(txn.get())->getMemberState(); + MemberState state = replCoord->getMemberState(); if (state.primary() || state.startup()) { continue; } @@ -226,7 +221,7 @@ void SyncSourceFeedback::run(executor::TaskExecutor* executor, BackgroundSync* b // Update keepalive value from config. auto oldKeepAliveInterval = keepAliveInterval; - keepAliveInterval = calculateKeepAliveInterval(txn.get(), _mtx); + keepAliveInterval = calculateKeepAliveInterval(replCoord->getConfig()); if (oldKeepAliveInterval != keepAliveInterval) { LOG(1) << "new syncSourceFeedback keep alive duration = " << keepAliveInterval << " (previously " << oldKeepAliveInterval << ")"; @@ -235,7 +230,7 @@ void SyncSourceFeedback::run(executor::TaskExecutor* executor, BackgroundSync* b Reporter reporter( executor, - makePrepareReplSetUpdatePositionCommandFn(txn.get(), _mtx, syncTarget, bgsync), + makePrepareReplSetUpdatePositionCommandFn(replCoord, _mtx, syncTarget, bgsync), syncTarget, keepAliveInterval); { @@ -250,7 +245,7 @@ void SyncSourceFeedback::run(executor::TaskExecutor* executor, BackgroundSync* b _reporter = nullptr; }); - auto status = _updateUpstream(txn.get(), bgsync, &reporter); + auto status = _updateUpstream(replCoord, bgsync, &reporter); if (!status.isOK()) { LOG(1) << "The replication progress command (replSetUpdatePosition) failed and will be " "retried: " diff --git a/src/mongo/db/repl/sync_source_feedback.h b/src/mongo/db/repl/sync_source_feedback.h index 40c29dc172a..35ed470fc26 100644 --- a/src/mongo/db/repl/sync_source_feedback.h +++ b/src/mongo/db/repl/sync_source_feedback.h @@ -31,6 +31,7 @@ #include "mongo/base/disallow_copying.h" #include "mongo/base/status.h" +#include "mongo/db/repl/replication_coordinator.h" #include "mongo/stdx/condition_variable.h" #include "mongo/stdx/mutex.h" @@ -63,7 +64,9 @@ public: * * Task executor is used to run replSetUpdatePosition command on sync source. */ - void run(executor::TaskExecutor* executor, BackgroundSync* bgsync); + void run(executor::TaskExecutor* executor, + BackgroundSync* bgsync, + ReplicationCoordinator* replCoord); /// Signals the run() method to terminate. void shutdown(); @@ -72,7 +75,9 @@ private: /* Inform the sync target of our current position in the oplog, as well as the positions * of all secondaries chained through us. */ - Status _updateUpstream(OperationContext* txn, BackgroundSync* bgsync, Reporter* reporter); + Status _updateUpstream(ReplicationCoordinator* replCoord, + BackgroundSync* bgsync, + Reporter* reporter); // protects cond, _shutdownSignaled, _keepAliveInterval, and _positionChanged. stdx::mutex _mtx; diff --git a/src/mongo/db/repl/sync_source_resolver.cpp b/src/mongo/db/repl/sync_source_resolver.cpp index 948a471b245..17a706f1b61 100644 --- a/src/mongo/db/repl/sync_source_resolver.cpp +++ b/src/mongo/db/repl/sync_source_resolver.cpp @@ -54,6 +54,7 @@ const Seconds SyncSourceResolver::kFirstOplogEntryEmptyBlacklistDuration(10); const Seconds SyncSourceResolver::kFirstOplogEntryNullTimestampBlacklistDuration(10); const Minutes SyncSourceResolver::kTooStaleBlacklistDuration(1); const Seconds SyncSourceResolver::kNoRequiredOpTimeBlacklistDuration(60); +const int SyncSourceResolver::kUninitializedRollbackId(-1); SyncSourceResolver::SyncSourceResolver(executor::TaskExecutor* taskExecutor, SyncSourceSelector* syncSourceSelector, @@ -175,11 +176,13 @@ std::unique_ptr<Fetcher> SyncSourceResolver::_makeFirstOplogEntryFetcher( candidate, earliestOpTimeSeen), rpc::ServerSelectionMetadata(true, boost::none).toBSON(), - kFetcherTimeout); + kFetcherTimeout /* find network timeout */, + kFetcherTimeout /* getMore network timeout */); } std::unique_ptr<Fetcher> SyncSourceResolver::_makeRequiredOpTimeFetcher(HostAndPort candidate, - OpTime earliestOpTimeSeen) { + OpTime earliestOpTimeSeen, + int rbid) { // This query is structured so that it is executed on the sync source using the oplog // start hack (oplogReplay=true and $gt/$gte predicate over "ts"). return stdx::make_unique<Fetcher>( @@ -193,9 +196,11 @@ std::unique_ptr<Fetcher> SyncSourceResolver::_makeRequiredOpTimeFetcher(HostAndP this, stdx::placeholders::_1, candidate, - earliestOpTimeSeen), + earliestOpTimeSeen, + rbid), rpc::ServerSelectionMetadata(true, boost::none).toBSON(), - kFetcherTimeout); + kFetcherTimeout /* find network timeout */, + kFetcherTimeout /* getMore network timeout */); } Status SyncSourceResolver::_scheduleFetcher(std::unique_ptr<Fetcher> fetcher) { @@ -205,6 +210,9 @@ Status SyncSourceResolver::_scheduleFetcher(std::unique_ptr<Fetcher> fetcher) { // executor. auto status = fetcher->schedule(); if (status.isOK()) { + // Fetcher destruction blocks on all outstanding callbacks. If we are currently in a + // Fetcher-related callback, we can't destroy the Fetcher just yet, so we assign it to a + // temporary unique pointer to allow the destruction to run to completion. _shuttingDownFetcher = std::move(_fetcher); _fetcher = std::move(fetcher); } else { @@ -311,10 +319,26 @@ void SyncSourceResolver::_firstOplogEntryFetcherCallback( return; } - _scheduleRBIDRequest(candidate, earliestOpTimeSeen); + auto status = _scheduleRBIDRequest(candidate, earliestOpTimeSeen); + if (!status.isOK()) { + _finishCallback(status); + } } -void SyncSourceResolver::_scheduleRBIDRequest(HostAndPort candidate, OpTime earliestOpTimeSeen) { +Status SyncSourceResolver::_scheduleRBIDRequest(HostAndPort candidate, OpTime earliestOpTimeSeen) { + // Once a work is scheduled, nothing prevents it finishing. We need the mutex to protect the + // access of member variables after scheduling, because otherwise the scheduled callback could + // finish and allow the destructor to fire before we access the member variables. + stdx::lock_guard<stdx::mutex> lk(_mutex); + if (_state == State::kShuttingDown) { + return Status( + ErrorCodes::CallbackCanceled, + str::stream() + << "sync source resolver shut down while checking rollbackId on candidate: " + << candidate); + } + + invariant(_state == State::kRunning); auto handle = _taskExecutor->scheduleRemoteCommand( {candidate, "admin", BSON("replSetGetRBID" << 1), nullptr, kFetcherTimeout}, stdx::bind(&SyncSourceResolver::_rbidRequestCallback, @@ -324,15 +348,11 @@ void SyncSourceResolver::_scheduleRBIDRequest(HostAndPort candidate, OpTime earl stdx::placeholders::_1)); if (!handle.isOK()) { - _finishCallback(handle.getStatus()); - return; + return handle.getStatus(); } - stdx::lock_guard<stdx::mutex> lk(_mutex); _rbidCommandHandle = std::move(handle.getValue()); - if (_state == State::kShuttingDown) { - _taskExecutor->cancel(_rbidCommandHandle); - } + return Status::OK(); } void SyncSourceResolver::_rbidRequestCallback( @@ -344,10 +364,11 @@ void SyncSourceResolver::_rbidRequestCallback( return; } + int rbid = kUninitializedRollbackId; try { uassertStatusOK(rbidReply.response.status); uassertStatusOK(getStatusFromCommandResult(rbidReply.response.data)); - _rbid = rbidReply.response.data["rbid"].Int(); + rbid = rbidReply.response.data["rbid"].Int(); } catch (const DBException& ex) { const auto until = _taskExecutor->now() + kFetcherErrorBlacklistDuration; log() << "Blacklisting " << candidate << " due to error: '" << ex << "' for " @@ -360,13 +381,15 @@ void SyncSourceResolver::_rbidRequestCallback( if (!_requiredOpTime.isNull()) { // Schedule fetcher to look for '_requiredOpTime' in the remote oplog. // Unittest requires that this kind of failure be handled specially. - auto status = _scheduleFetcher(_makeRequiredOpTimeFetcher(candidate, earliestOpTimeSeen)); + auto status = + _scheduleFetcher(_makeRequiredOpTimeFetcher(candidate, earliestOpTimeSeen, rbid)); if (!status.isOK()) { _finishCallback(status); } return; } - _finishCallback(candidate); + + _finishCallback(candidate, rbid); } Status SyncSourceResolver::_compareRequiredOpTimeWithQueryResponse( @@ -399,7 +422,8 @@ Status SyncSourceResolver::_compareRequiredOpTimeWithQueryResponse( void SyncSourceResolver::_requiredOpTimeFetcherCallback( const StatusWith<Fetcher::QueryResponse>& queryResult, HostAndPort candidate, - OpTime earliestOpTimeSeen) { + OpTime earliestOpTimeSeen, + int rbid) { if (_isShuttingDown()) { _finishCallback(Status(ErrorCodes::CallbackCanceled, str::stream() << "sync source resolver shut down while looking for " @@ -444,18 +468,18 @@ void SyncSourceResolver::_requiredOpTimeFetcherCallback( return; } - _finishCallback(candidate); + _finishCallback(candidate, rbid); } Status SyncSourceResolver::_chooseAndProbeNextSyncSource(OpTime earliestOpTimeSeen) { auto candidateResult = _chooseNewSyncSource(); if (!candidateResult.isOK()) { - return _finishCallback(candidateResult); + return _finishCallback(candidateResult.getStatus()); } if (candidateResult.getValue().empty()) { if (earliestOpTimeSeen.isNull()) { - return _finishCallback(candidateResult); + return _finishCallback(candidateResult.getValue(), kUninitializedRollbackId); } SyncSourceResolverResponse response; @@ -473,16 +497,22 @@ Status SyncSourceResolver::_chooseAndProbeNextSyncSource(OpTime earliestOpTimeSe return Status::OK(); } -Status SyncSourceResolver::_finishCallback(StatusWith<HostAndPort> result) { +Status SyncSourceResolver::_finishCallback(HostAndPort hostAndPort, int rbid) { SyncSourceResolverResponse response; - response.syncSourceStatus = std::move(result); - if (response.isOK() && !response.getSyncSource().empty()) { - invariant(_requiredOpTime.isNull() || _rbid); - response.rbid = _rbid; + response.syncSourceStatus = std::move(hostAndPort); + if (rbid != kUninitializedRollbackId) { + response.rbid = rbid; } return _finishCallback(response); } +Status SyncSourceResolver::_finishCallback(Status status) { + invariant(!status.isOK()); + SyncSourceResolverResponse response; + response.syncSourceStatus = std::move(status); + return _finishCallback(response); +} + Status SyncSourceResolver::_finishCallback(const SyncSourceResolverResponse& response) { try { _onCompletion(response); diff --git a/src/mongo/db/repl/sync_source_resolver.h b/src/mongo/db/repl/sync_source_resolver.h index 201658caacf..0f9343cfcdb 100644 --- a/src/mongo/db/repl/sync_source_resolver.h +++ b/src/mongo/db/repl/sync_source_resolver.h @@ -102,6 +102,7 @@ public: static const Seconds kFirstOplogEntryNullTimestampBlacklistDuration; static const Minutes kTooStaleBlacklistDuration; static const Seconds kNoRequiredOpTimeBlacklistDuration; + static const int kUninitializedRollbackId; /** * Callback function to report final status of resolving sync source. @@ -154,7 +155,8 @@ private: * Creates fetcher to check the remote oplog for '_requiredOpTime'. */ std::unique_ptr<Fetcher> _makeRequiredOpTimeFetcher(HostAndPort candidate, - OpTime earliestOpTimeSeen); + OpTime earliestOpTimeSeen, + int rbid); /** * Schedules fetcher to read oplog on sync source. @@ -179,7 +181,7 @@ private: /** * Schedules a replSetGetRBID command against the candidate to fetch its current rollback id. */ - void _scheduleRBIDRequest(HostAndPort candidate, OpTime earliestOpTimeSeen); + Status _scheduleRBIDRequest(HostAndPort candidate, OpTime earliestOpTimeSeen); void _rbidRequestCallback(HostAndPort candidate, OpTime earliestOpTimeSeen, const executor::TaskExecutor::RemoteCommandCallbackArgs& rbidReply); @@ -194,7 +196,8 @@ private: */ void _requiredOpTimeFetcherCallback(const StatusWith<Fetcher::QueryResponse>& queryResult, HostAndPort candidate, - OpTime earliestOpTimeSeen); + OpTime earliestOpTimeSeen, + int rbid); /** * Obtains new sync source candidate and schedules remote command to fetcher first oplog entry. @@ -207,7 +210,8 @@ private: * Invokes completion callback and transitions state to State::kComplete. * Returns result.getStatus(). */ - Status _finishCallback(StatusWith<HostAndPort> result); + Status _finishCallback(HostAndPort hostAndPort, int rbid); + Status _finishCallback(Status status); Status _finishCallback(const SyncSourceResolverResponse& response); // Executor used to send remote commands to sync source candidates. @@ -229,9 +233,6 @@ private: // resolver via this callback in a SyncSourceResolverResponse struct when the resolver finishes. const OnCompletionFn _onCompletion; - // The rbid we will return to our caller. - int _rbid; - // Protects members of this sync source resolver defined below. mutable stdx::mutex _mutex; mutable stdx::condition_variable _condition; diff --git a/src/mongo/db/repl/sync_tail.cpp b/src/mongo/db/repl/sync_tail.cpp index 30f1e28b619..3f65e3fc6b7 100644 --- a/src/mongo/db/repl/sync_tail.cpp +++ b/src/mongo/db/repl/sync_tail.cpp @@ -448,7 +448,6 @@ void applyOps(std::vector<MultiApplier::OperationPtrs>& writerVectors, const MultiApplier::ApplyOperationFn& func, std::vector<Status>* statusVector) { invariant(writerVectors.size() == statusVector->size()); - TimerHolder timer(&applyBatchStats); for (size_t i = 0; i < writerVectors.size(); i++) { if (!writerVectors[i].empty()) { writerPool->schedule([&func, &writerVectors, statusVector, i] { @@ -695,32 +694,53 @@ public: } private: + /** + * Calculates batch limit size (in bytes) using the maximum capped collection size of the oplog + * size. + * Batches are limited to 10% of the oplog. + */ + std::size_t _calculateBatchLimitBytes() { + auto opCtx = cc().makeOperationContext(); + auto storageInterface = StorageInterface::get(opCtx.get()); + auto oplogMaxSizeResult = + storageInterface->getOplogMaxSize(opCtx.get(), NamespaceString(rsOplogName)); + auto oplogMaxSize = fassertStatusOK(40301, oplogMaxSizeResult); + return std::min(oplogMaxSize / 10, std::size_t(replBatchLimitBytes)); + } + + /** + * If slaveDelay is enabled, this function calculates the most recent timestamp of any oplog + * entries that can be be returned in a batch. + */ + boost::optional<Date_t> _calculateSlaveDelayLatestTimestamp() { + auto service = cc().getServiceContext(); + auto replCoord = ReplicationCoordinator::get(service); + auto slaveDelay = replCoord->getSlaveDelaySecs(); + if (slaveDelay <= Seconds(0)) { + return {}; + } + auto fastClockSource = service->getFastClockSource(); + return fastClockSource->now() - slaveDelay; + } + void run() { Client::initThread("ReplBatcher"); - const ServiceContext::UniqueOperationContext txnPtr = cc().makeOperationContext(); - OperationContext& txn = *txnPtr; - const auto replCoord = ReplicationCoordinator::get(&txn); - const auto fastClockSource = txn.getServiceContext()->getFastClockSource(); - const auto oplogMaxSize = fassertStatusOK( - 40301, - StorageInterface::get(&txn)->getOplogMaxSize(&txn, NamespaceString(rsOplogName))); - // Batches are limited to 10% of the oplog. BatchLimits batchLimits; - batchLimits.bytes = std::min(oplogMaxSize / 10, size_t(replBatchLimitBytes)); + batchLimits.bytes = _calculateBatchLimitBytes(); while (true) { - const auto slaveDelay = replCoord->getSlaveDelaySecs(); - batchLimits.slaveDelayLatestTimestamp = (slaveDelay > Seconds(0)) - ? (fastClockSource->now() - slaveDelay) - : boost::optional<Date_t>(); + batchLimits.slaveDelayLatestTimestamp = _calculateSlaveDelayLatestTimestamp(); // Check this once per batch since users can change it at runtime. batchLimits.ops = replBatchLimitOperations.load(); OpQueue ops; // tryPopAndWaitForMore adds to ops and returns true when we need to end a batch early. - while (!_syncTail->tryPopAndWaitForMore(&txn, &ops, batchLimits)) { + { + auto opCtx = cc().makeOperationContext(); + while (!_syncTail->tryPopAndWaitForMore(opCtx.get(), &ops, batchLimits)) { + } } if (ops.empty() && !ops.mustShutdown()) { @@ -755,14 +775,15 @@ private: void SyncTail::oplogApplication(ReplicationCoordinator* replCoord) { OpQueueBatcher batcher(this); - const ServiceContext::UniqueOperationContext txnPtr = cc().makeOperationContext(); - OperationContext& txn = *txnPtr; std::unique_ptr<ApplyBatchFinalizer> finalizer{ getGlobalServiceContext()->getGlobalStorageEngine()->isDurable() ? new ApplyBatchFinalizerForJournal(replCoord) : new ApplyBatchFinalizer(replCoord)}; while (true) { // Exits on message from OpQueueBatcher. + const ServiceContext::UniqueOperationContext txnPtr = cc().makeOperationContext(); + OperationContext& txn = *txnPtr; + // For pausing replication in tests. while (MONGO_FAIL_POINT(rsSyncApplyStop)) { // Tests should not trigger clean shutdown while that failpoint is active. If we @@ -1301,6 +1322,9 @@ StatusWith<OpTime> multiApply(OperationContext* txn, std::vector<Status> statusVector(workerPool->getNumThreads(), Status::OK()); { + // Each node records cumulative batch application stats for itself using this timer. + TimerHolder timer(&applyBatchStats); + // We must wait for the all work we've dispatched to complete before leaving this block // because the spawned threads refer to objects on our stack, including writerVectors. std::vector<MultiApplier::OperationPtrs> writerVectors(workerPool->getNumThreads()); diff --git a/src/mongo/db/repl/topology_coordinator_impl.cpp b/src/mongo/db/repl/topology_coordinator_impl.cpp index 88337b6a500..f5302008206 100644 --- a/src/mongo/db/repl/topology_coordinator_impl.cpp +++ b/src/mongo/db/repl/topology_coordinator_impl.cpp @@ -36,6 +36,7 @@ #include "mongo/db/audit.h" #include "mongo/db/client.h" +#include "mongo/db/mongod_options.h" #include "mongo/db/operation_context.h" #include "mongo/db/repl/heartbeat_response_action.h" #include "mongo/db/repl/is_master_response.h" @@ -1411,7 +1412,7 @@ bool TopologyCoordinatorImpl::_aMajoritySeemsToBeUp() const { return vUp * 2 > _rsConfig.getTotalVotingMembers(); } -bool TopologyCoordinatorImpl::_canSeeHealthyPrimaryOfEqualOrGreaterPriority( +int TopologyCoordinatorImpl::_findHealthyPrimaryOfEqualOrGreaterPriority( const int candidateIndex) const { const double candidatePriority = _rsConfig.getMemberAt(candidateIndex).getPriority(); for (auto it = _hbdata.begin(); it != _hbdata.end(); ++it) { @@ -1421,11 +1422,11 @@ bool TopologyCoordinatorImpl::_canSeeHealthyPrimaryOfEqualOrGreaterPriority( const int itIndex = indexOfIterator(_hbdata, it); const double priority = _rsConfig.getMemberAt(itIndex).getPriority(); if (itIndex != candidateIndex && priority >= candidatePriority) { - return true; + return itIndex; } } - return false; + return -1; } bool TopologyCoordinatorImpl::_isOpTimeCloseEnoughToLatestToElect( @@ -2232,7 +2233,7 @@ MemberState TopologyCoordinatorImpl::getMemberState() const { } if (_rsConfig.isConfigServer()) { - if (_options.clusterRole != ClusterRole::ConfigServer) { + if (_options.clusterRole != ClusterRole::ConfigServer && !skipShardingConfigurationChecks) { return MemberState::RS_REMOVED; } else { invariant(_storageEngineSupportsReadCommitted != ReadCommittedSupport::kUnknown); @@ -2241,7 +2242,7 @@ MemberState TopologyCoordinatorImpl::getMemberState() const { } } } else { - if (_options.clusterRole == ClusterRole::ConfigServer) { + if (_options.clusterRole == ClusterRole::ConfigServer && !skipShardingConfigurationChecks) { return MemberState::RS_REMOVED; } } @@ -2571,29 +2572,54 @@ void TopologyCoordinatorImpl::processReplSetRequestVotes(const ReplSetRequestVot if (args.getTerm() < _term) { response->setVoteGranted(false); - response->setReason("candidate's term is lower than mine"); + response->setReason(str::stream() << "candidate's term (" << args.getTerm() + << ") is lower than mine (" + << _term + << ")"); } else if (args.getConfigVersion() != _rsConfig.getConfigVersion()) { response->setVoteGranted(false); - response->setReason("candidate's config version differs from mine"); + response->setReason(str::stream() << "candidate's config version (" + << args.getConfigVersion() + << ") differs from mine (" + << _rsConfig.getConfigVersion() + << ")"); } else if (args.getSetName() != _rsConfig.getReplSetName()) { response->setVoteGranted(false); - response->setReason("candidate's set name differs from mine"); + response->setReason(str::stream() << "candidate's set name (" << args.getSetName() + << ") differs from mine (" + << _rsConfig.getReplSetName() + << ")"); } else if (args.getLastDurableOpTime() < lastAppliedOpTime) { response->setVoteGranted(false); - response->setReason("candidate's data is staler than mine"); + response + ->setReason(str::stream() + << "candidate's data is staler than mine. candidate's last applied OpTime: " + << args.getLastDurableOpTime().toString() + << ", my last applied OpTime: " + << lastAppliedOpTime.toString()); } else if (!args.isADryRun() && _lastVote.getTerm() == args.getTerm()) { response->setVoteGranted(false); - response->setReason("already voted for another candidate this term"); - } else if (_selfConfig().isArbiter() && - _canSeeHealthyPrimaryOfEqualOrGreaterPriority(args.getCandidateIndex())) { - response->setVoteGranted(false); - response->setReason("can see a healthy primary of equal or greater priority"); + response->setReason(str::stream() + << "already voted for another candidate (" + << _rsConfig.getMemberAt(_lastVote.getCandidateIndex()).getHostAndPort() + << ") this term (" + << _lastVote.getTerm() + << ")"); } else { - if (!args.isADryRun()) { - _lastVote.setTerm(args.getTerm()); - _lastVote.setCandidateIndex(args.getCandidateIndex()); + int betterPrimary = _findHealthyPrimaryOfEqualOrGreaterPriority(args.getCandidateIndex()); + if (_selfConfig().isArbiter() && betterPrimary >= 0) { + response->setVoteGranted(false); + response->setReason(str::stream() + << "can see a healthy primary (" + << _rsConfig.getMemberAt(betterPrimary).getHostAndPort() + << ") of equal or greater priority"); + } else { + if (!args.isADryRun()) { + _lastVote.setTerm(args.getTerm()); + _lastVote.setCandidateIndex(args.getCandidateIndex()); + } + response->setVoteGranted(true); } - response->setVoteGranted(true); } } diff --git a/src/mongo/db/repl/topology_coordinator_impl.h b/src/mongo/db/repl/topology_coordinator_impl.h index d4b0476cde1..1a03338e3c4 100644 --- a/src/mongo/db/repl/topology_coordinator_impl.h +++ b/src/mongo/db/repl/topology_coordinator_impl.h @@ -310,9 +310,9 @@ private: // Sees if a majority number of votes are held by members who are currently "up" bool _aMajoritySeemsToBeUp() const; - // Returns true if the node can see a healthy primary of equal or greater priority to the - // candidate. - bool _canSeeHealthyPrimaryOfEqualOrGreaterPriority(const int candidateIndex) const; + // Checks if the node can see a healthy primary of equal or greater priority to the + // candidate. If so, returns the index of that node. Otherwise returns -1. + int _findHealthyPrimaryOfEqualOrGreaterPriority(const int candidateIndex) const; // Is otherOpTime close enough (within 10 seconds) to the latest known optime to qualify // for an election diff --git a/src/mongo/db/repl/topology_coordinator_impl_test.cpp b/src/mongo/db/repl/topology_coordinator_impl_test.cpp index e3708a9bf54..1f8d4b56c40 100644 --- a/src/mongo/db/repl/topology_coordinator_impl_test.cpp +++ b/src/mongo/db/repl/topology_coordinator_impl_test.cpp @@ -5912,7 +5912,8 @@ TEST_F(TopoCoordTest, NodeDoesNotGrantVotesToTwoDifferentNodesInTheSameTerm) { // different candidate same term, should be a problem getTopoCoord().processReplSetRequestVotes(args2, &response2, lastAppliedOpTime); - ASSERT_EQUALS("already voted for another candidate this term", response2.getReason()); + ASSERT_EQUALS("already voted for another candidate (hself:27017) this term (1)", + response2.getReason()); ASSERT_FALSE(response2.getVoteGranted()); } @@ -6028,7 +6029,8 @@ TEST_F(TopoCoordTest, VoteRequestShouldNotPreventDryRunsForThatTerm) { ReplSetRequestVotesResponse response2; getTopoCoord().processReplSetRequestVotes(args2, &response2, lastAppliedOpTime); - ASSERT_EQUALS("already voted for another candidate this term", response2.getReason()); + ASSERT_EQUALS("already voted for another candidate (hself:27017) this term (1)", + response2.getReason()); ASSERT_FALSE(response2.getVoteGranted()); } @@ -6063,7 +6065,7 @@ TEST_F(TopoCoordTest, NodeDoesNotGrantVoteWhenReplSetNameDoesNotMatch) { OpTime lastAppliedOpTime; getTopoCoord().processReplSetRequestVotes(args, &response, lastAppliedOpTime); - ASSERT_EQUALS("candidate's set name differs from mine", response.getReason()); + ASSERT_EQUALS("candidate's set name (wrongName) differs from mine (rs0)", response.getReason()); ASSERT_FALSE(response.getVoteGranted()); } @@ -6098,7 +6100,7 @@ TEST_F(TopoCoordTest, NodeDoesNotGrantVoteWhenConfigVersionDoesNotMatch) { OpTime lastAppliedOpTime; getTopoCoord().processReplSetRequestVotes(args, &response, lastAppliedOpTime); - ASSERT_EQUALS("candidate's config version differs from mine", response.getReason()); + ASSERT_EQUALS("candidate's config version (0) differs from mine (1)", response.getReason()); ASSERT_FALSE(response.getVoteGranted()); } @@ -6145,7 +6147,8 @@ TEST_F(TopoCoordTest, ArbiterDoesNotGrantVoteWhenItCanSeeAHealthyPrimaryOfEqualO OpTime lastAppliedOpTime; getTopoCoord().processReplSetRequestVotes(args, &response, lastAppliedOpTime); - ASSERT_EQUALS("can see a healthy primary of equal or greater priority", response.getReason()); + ASSERT_EQUALS("can see a healthy primary (h2:27017) of equal or greater priority", + response.getReason()); ASSERT_FALSE(response.getVoteGranted()); } @@ -6184,7 +6187,7 @@ TEST_F(TopoCoordTest, NodeDoesNotGrantVoteWhenTermIsStale) { OpTime lastAppliedOpTime; getTopoCoord().processReplSetRequestVotes(args, &response, lastAppliedOpTime); - ASSERT_EQUALS("candidate's term is lower than mine", response.getReason()); + ASSERT_EQUALS("candidate's term (1) is lower than mine (2)", response.getReason()); ASSERT_EQUALS(2, response.getTerm()); ASSERT_FALSE(response.getVoteGranted()); } @@ -6221,7 +6224,12 @@ TEST_F(TopoCoordTest, NodeDoesNotGrantVoteWhenOpTimeIsStale) { OpTime lastAppliedOpTime2 = {Timestamp(20, 0), 0}; getTopoCoord().processReplSetRequestVotes(args, &response, lastAppliedOpTime2); - ASSERT_EQUALS("candidate's data is staler than mine", response.getReason()); + ASSERT_EQUALS( + str::stream() << "candidate's data is staler than mine. candidate's last applied OpTime: " + << OpTime().toString() + << ", my last applied OpTime: " + << OpTime(Timestamp(20, 0), 0).toString(), + response.getReason()); ASSERT_FALSE(response.getVoteGranted()); } @@ -6281,7 +6289,7 @@ TEST_F(TopoCoordTest, NodeDoesNotGrantDryRunVoteWhenReplSetNameDoesNotMatch) { ReplSetRequestVotesResponse response; getTopoCoord().processReplSetRequestVotes(args, &response, lastAppliedOpTime); - ASSERT_EQUALS("candidate's set name differs from mine", response.getReason()); + ASSERT_EQUALS("candidate's set name (wrongName) differs from mine (rs0)", response.getReason()); ASSERT_EQUALS(1, response.getTerm()); ASSERT_FALSE(response.getVoteGranted()); } @@ -6342,7 +6350,7 @@ TEST_F(TopoCoordTest, NodeDoesNotGrantDryRunVoteWhenConfigVersionDoesNotMatch) { ReplSetRequestVotesResponse response; getTopoCoord().processReplSetRequestVotes(args, &response, lastAppliedOpTime); - ASSERT_EQUALS("candidate's config version differs from mine", response.getReason()); + ASSERT_EQUALS("candidate's config version (0) differs from mine (1)", response.getReason()); ASSERT_EQUALS(1, response.getTerm()); ASSERT_FALSE(response.getVoteGranted()); } @@ -6402,7 +6410,7 @@ TEST_F(TopoCoordTest, NodeDoesNotGrantDryRunVoteWhenTermIsStale) { ReplSetRequestVotesResponse response; getTopoCoord().processReplSetRequestVotes(args, &response, lastAppliedOpTime); - ASSERT_EQUALS("candidate's term is lower than mine", response.getReason()); + ASSERT_EQUALS("candidate's term (0) is lower than mine (1)", response.getReason()); ASSERT_EQUALS(1, response.getTerm()); ASSERT_FALSE(response.getVoteGranted()); } @@ -6525,7 +6533,12 @@ TEST_F(TopoCoordTest, DoNotGrantDryRunVoteWhenOpTimeIsStale) { OpTime lastAppliedOpTime2 = {Timestamp(20, 0), 0}; getTopoCoord().processReplSetRequestVotes(args, &response, lastAppliedOpTime2); - ASSERT_EQUALS("candidate's data is staler than mine", response.getReason()); + ASSERT_EQUALS( + str::stream() << "candidate's data is staler than mine. candidate's last applied OpTime: " + << OpTime().toString() + << ", my last applied OpTime: " + << OpTime(Timestamp(20, 0), 0).toString(), + response.getReason()); ASSERT_EQUALS(1, response.getTerm()); ASSERT_FALSE(response.getVoteGranted()); } diff --git a/src/mongo/db/repl/topology_coordinator_impl_v1_test.cpp b/src/mongo/db/repl/topology_coordinator_impl_v1_test.cpp index 8062208949b..8f60269cd71 100644 --- a/src/mongo/db/repl/topology_coordinator_impl_v1_test.cpp +++ b/src/mongo/db/repl/topology_coordinator_impl_v1_test.cpp @@ -2574,7 +2574,8 @@ TEST_F(TopoCoordTest, NodeDoesNotGrantVotesToTwoDifferentNodesInTheSameTerm) { // different candidate same term, should be a problem getTopoCoord().processReplSetRequestVotes(args2, &response2, lastAppliedOpTime); - ASSERT_EQUALS("already voted for another candidate this term", response2.getReason()); + ASSERT_EQUALS("already voted for another candidate (hself:27017) this term (1)", + response2.getReason()); ASSERT_FALSE(response2.getVoteGranted()); } @@ -2674,7 +2675,8 @@ TEST_F(TopoCoordTest, DryRunVoteRequestShouldNotPreventSubsequentDryRunsForThatT ReplSetRequestVotesResponse response4; getTopoCoord().processReplSetRequestVotes(args4, &response4, lastAppliedOpTime); - ASSERT_EQUALS("already voted for another candidate this term", response4.getReason()); + ASSERT_EQUALS("already voted for another candidate (hself:27017) this term (1)", + response4.getReason()); ASSERT_FALSE(response4.getVoteGranted()); } @@ -2732,7 +2734,8 @@ TEST_F(TopoCoordTest, VoteRequestShouldNotPreventDryRunsForThatTerm) { ReplSetRequestVotesResponse response2; getTopoCoord().processReplSetRequestVotes(args2, &response2, lastAppliedOpTime); - ASSERT_EQUALS("already voted for another candidate this term", response2.getReason()); + ASSERT_EQUALS("already voted for another candidate (hself:27017) this term (1)", + response2.getReason()); ASSERT_FALSE(response2.getVoteGranted()); } @@ -2767,7 +2770,7 @@ TEST_F(TopoCoordTest, NodeDoesNotGrantVoteWhenReplSetNameDoesNotMatch) { OpTime lastAppliedOpTime; getTopoCoord().processReplSetRequestVotes(args, &response, lastAppliedOpTime); - ASSERT_EQUALS("candidate's set name differs from mine", response.getReason()); + ASSERT_EQUALS("candidate's set name (wrongName) differs from mine (rs0)", response.getReason()); ASSERT_FALSE(response.getVoteGranted()); } @@ -2802,7 +2805,7 @@ TEST_F(TopoCoordTest, NodeDoesNotGrantVoteWhenConfigVersionDoesNotMatch) { OpTime lastAppliedOpTime; getTopoCoord().processReplSetRequestVotes(args, &response, lastAppliedOpTime); - ASSERT_EQUALS("candidate's config version differs from mine", response.getReason()); + ASSERT_EQUALS("candidate's config version (0) differs from mine (1)", response.getReason()); ASSERT_FALSE(response.getVoteGranted()); } @@ -2841,7 +2844,7 @@ TEST_F(TopoCoordTest, NodeDoesNotGrantVoteWhenTermIsStale) { OpTime lastAppliedOpTime; getTopoCoord().processReplSetRequestVotes(args, &response, lastAppliedOpTime); - ASSERT_EQUALS("candidate's term is lower than mine", response.getReason()); + ASSERT_EQUALS("candidate's term (1) is lower than mine (2)", response.getReason()); ASSERT_EQUALS(2, response.getTerm()); ASSERT_FALSE(response.getVoteGranted()); } @@ -2878,7 +2881,12 @@ TEST_F(TopoCoordTest, NodeDoesNotGrantVoteWhenOpTimeIsStale) { OpTime lastAppliedOpTime2 = {Timestamp(20, 0), 0}; getTopoCoord().processReplSetRequestVotes(args, &response, lastAppliedOpTime2); - ASSERT_EQUALS("candidate's data is staler than mine", response.getReason()); + ASSERT_EQUALS( + str::stream() << "candidate's data is staler than mine. candidate's last applied OpTime: " + << OpTime().toString() + << ", my last applied OpTime: " + << OpTime(Timestamp(20, 0), 0).toString(), + response.getReason()); ASSERT_FALSE(response.getVoteGranted()); } @@ -2938,7 +2946,7 @@ TEST_F(TopoCoordTest, NodeDoesNotGrantDryRunVoteWhenReplSetNameDoesNotMatch) { ReplSetRequestVotesResponse response; getTopoCoord().processReplSetRequestVotes(args, &response, lastAppliedOpTime); - ASSERT_EQUALS("candidate's set name differs from mine", response.getReason()); + ASSERT_EQUALS("candidate's set name (wrongName) differs from mine (rs0)", response.getReason()); ASSERT_EQUALS(1, response.getTerm()); ASSERT_FALSE(response.getVoteGranted()); } @@ -2999,7 +3007,7 @@ TEST_F(TopoCoordTest, NodeDoesNotGrantDryRunVoteWhenConfigVersionDoesNotMatch) { ReplSetRequestVotesResponse response; getTopoCoord().processReplSetRequestVotes(args, &response, lastAppliedOpTime); - ASSERT_EQUALS("candidate's config version differs from mine", response.getReason()); + ASSERT_EQUALS("candidate's config version (0) differs from mine (1)", response.getReason()); ASSERT_EQUALS(1, response.getTerm()); ASSERT_FALSE(response.getVoteGranted()); } @@ -3059,7 +3067,7 @@ TEST_F(TopoCoordTest, NodeDoesNotGrantDryRunVoteWhenTermIsStale) { ReplSetRequestVotesResponse response; getTopoCoord().processReplSetRequestVotes(args, &response, lastAppliedOpTime); - ASSERT_EQUALS("candidate's term is lower than mine", response.getReason()); + ASSERT_EQUALS("candidate's term (0) is lower than mine (1)", response.getReason()); ASSERT_EQUALS(1, response.getTerm()); ASSERT_FALSE(response.getVoteGranted()); } @@ -3182,7 +3190,12 @@ TEST_F(TopoCoordTest, DoNotGrantDryRunVoteWhenOpTimeIsStale) { OpTime lastAppliedOpTime2 = {Timestamp(20, 0), 0}; getTopoCoord().processReplSetRequestVotes(args, &response, lastAppliedOpTime2); - ASSERT_EQUALS("candidate's data is staler than mine", response.getReason()); + ASSERT_EQUALS( + str::stream() << "candidate's data is staler than mine. candidate's last applied OpTime: " + << OpTime().toString() + << ", my last applied OpTime: " + << OpTime(Timestamp(20, 0), 0).toString(), + response.getReason()); ASSERT_EQUALS(1, response.getTerm()); ASSERT_FALSE(response.getVoteGranted()); } |