diff options
author | Tess Avitabile <tess.avitabile@mongodb.com> | 2018-08-14 10:50:21 -0400 |
---|---|---|
committer | Tess Avitabile <tess.avitabile@mongodb.com> | 2018-08-22 10:41:55 -0400 |
commit | f5e7c8f3e81fe0cd34d4952ed2b547f3c29e06a4 (patch) | |
tree | f6db97b38a429e65abe042b367afcd79cb08ce1e /src/mongo/db/repl/topology_coordinator_v1_test.cpp | |
parent | 76b46c9e1080dc2f80091f06591eb4775d66a340 (diff) | |
download | mongo-f5e7c8f3e81fe0cd34d4952ed2b547f3c29e06a4.tar.gz |
SERVER-33248 Allow choosing a sync source that we are up to date with if it has a higher lastOpCommitted
Diffstat (limited to 'src/mongo/db/repl/topology_coordinator_v1_test.cpp')
-rw-r--r-- | src/mongo/db/repl/topology_coordinator_v1_test.cpp | 196 |
1 files changed, 162 insertions, 34 deletions
diff --git a/src/mongo/db/repl/topology_coordinator_v1_test.cpp b/src/mongo/db/repl/topology_coordinator_v1_test.cpp index 89a8af0d74e..be14cb543bb 100644 --- a/src/mongo/db/repl/topology_coordinator_v1_test.cpp +++ b/src/mongo/db/repl/topology_coordinator_v1_test.cpp @@ -169,9 +169,10 @@ protected: // Only set visibleOpTime, primaryIndex and syncSourceIndex ReplSetMetadata makeReplSetMetadata(OpTime visibleOpTime = OpTime(), int primaryIndex = -1, - int syncSourceIndex = -1) { + int syncSourceIndex = -1, + OpTime committedOpTime = OpTime()) { return ReplSetMetadata(_topo->getTerm(), - OpTime(), + committedOpTime, visibleOpTime, _currentConfig.getConfigVersion(), OID(), @@ -183,8 +184,10 @@ protected: // Only set lastAppliedOpTime, primaryIndex and syncSourceIndex OplogQueryMetadata makeOplogQueryMetadata(OpTime lastAppliedOpTime = OpTime(), int primaryIndex = -1, - int syncSourceIndex = -1) { - return OplogQueryMetadata(OpTime(), lastAppliedOpTime, -1, primaryIndex, syncSourceIndex); + int syncSourceIndex = -1, + OpTime committedOpTime = OpTime()) { + return OplogQueryMetadata( + committedOpTime, lastAppliedOpTime, -1, primaryIndex, syncSourceIndex); } HeartbeatResponseAction receiveUpHeartbeat(const HostAndPort& member, @@ -192,13 +195,15 @@ protected: MemberState memberState, const OpTime& electionTime, const OpTime& lastOpTimeSender, - const HostAndPort& syncingTo = HostAndPort()) { + const HostAndPort& syncingTo = HostAndPort(), + const OpTime& lastOpCommittedSender = OpTime()) { return _receiveHeartbeatHelper(Status::OK(), member, setName, memberState, electionTime.getTimestamp(), lastOpTimeSender, + lastOpCommittedSender, Milliseconds(1), syncingTo); } @@ -216,6 +221,7 @@ protected: MemberState::RS_UNKNOWN, Timestamp(), OpTime(), + OpTime(), roundTripTime, HostAndPort()); } @@ -224,13 +230,15 @@ protected: const std::string& setName, MemberState memberState, const OpTime& lastOpTimeSender, - Milliseconds roundTripTime = Milliseconds(1)) { + Milliseconds roundTripTime = Milliseconds(1), + const OpTime& lastOpCommittedSender = OpTime()) { return _receiveHeartbeatHelper(Status::OK(), member, setName, memberState, Timestamp(), lastOpTimeSender, + lastOpCommittedSender, roundTripTime, HostAndPort()); } @@ -242,6 +250,7 @@ private: MemberState memberState, Timestamp electionTime, const OpTime& lastOpTimeSender, + const OpTime& lastOpCommittedSender, Milliseconds roundTripTime, const HostAndPort& syncingTo) { ReplSetHeartbeatResponse hb; @@ -259,7 +268,8 @@ private: getTopoCoord().prepareHeartbeatRequestV1(now(), setName, member); now() += roundTripTime; - return getTopoCoord().processHeartbeatResponse(now(), roundTripTime, member, hbResponse); + return getTopoCoord().processHeartbeatResponse( + now(), roundTripTime, member, hbResponse, lastOpCommittedSender); } private: @@ -574,6 +584,80 @@ TEST_F(TopoCoordTest, NodeWontChooseSyncSourceFromOlderTerm) { ASSERT(getTopoCoord().getSyncSourceAddress().empty()); } +TEST_F(TopoCoordTest, NodeCanChooseSyncSourceWithSameLastAppliedAndHigherLastOpCommitted) { + updateConfig(BSON("_id" + << "rs0" + << "version" + << 1 + << "members" + << BSON_ARRAY(BSON("_id" << 1 << "host" + << "hself") + << BSON("_id" << 10 << "host" + << "h1"))), + 0); + + setSelfMemberState(MemberState::RS_SECONDARY); + OpTime lastApplied = OpTime(Timestamp(100, 3), 3); + OpTime ourLastOpCommitted = OpTime(Timestamp(100, 1), 3); + OpTime lastOpCommittedSyncSource = OpTime(Timestamp(100, 2), 3); + + heartbeatFromMember(HostAndPort("h1"), + "rs0", + MemberState::RS_SECONDARY, + lastApplied, + Milliseconds(100), + lastOpCommittedSyncSource); + + // Record 2nd round of pings to allow choosing a new sync source. + heartbeatFromMember(HostAndPort("h1"), + "rs0", + MemberState::RS_SECONDARY, + lastApplied, + Milliseconds(100), + lastOpCommittedSyncSource); + + ASSERT(getTopoCoord().advanceLastCommittedOpTime(ourLastOpCommitted)); + getTopoCoord().chooseNewSyncSource( + now()++, lastApplied, TopologyCoordinator::ChainingPreference::kUseConfiguration); + ASSERT_EQUALS(HostAndPort("h1"), getTopoCoord().getSyncSourceAddress()); +} + +TEST_F(TopoCoordTest, NodeCannotChooseSyncSourceWithSameLastAppliedAndSameLastOpCommitted) { + updateConfig(BSON("_id" + << "rs0" + << "version" + << 1 + << "members" + << BSON_ARRAY(BSON("_id" << 1 << "host" + << "hself") + << BSON("_id" << 10 << "host" + << "h1"))), + 0); + + setSelfMemberState(MemberState::RS_SECONDARY); + OpTime lastApplied = OpTime(Timestamp(100, 3), 3); + OpTime lastOpCommitted = OpTime(Timestamp(100, 1), 3); + + heartbeatFromMember(HostAndPort("h1"), + "rs0", + MemberState::RS_SECONDARY, + lastApplied, + Milliseconds(100), + lastOpCommitted); + + // Record 2nd round of pings to allow choosing a new sync source. + heartbeatFromMember(HostAndPort("h1"), + "rs0", + MemberState::RS_SECONDARY, + lastApplied, + Milliseconds(100), + lastOpCommitted); + + ASSERT(getTopoCoord().advanceLastCommittedOpTime(lastOpCommitted)); + getTopoCoord().chooseNewSyncSource( + now()++, lastApplied, TopologyCoordinator::ChainingPreference::kUseConfiguration); + ASSERT(getTopoCoord().getSyncSourceAddress().empty()); +} TEST_F(TopoCoordTest, ChooseOnlyPrimaryAsSyncSourceWhenChainingIsDisallowed) { updateConfig(BSON("_id" @@ -1613,7 +1697,7 @@ TEST_F(TopoCoordTest, ReplSetGetStatus) { HostAndPort member = HostAndPort("test0:1234"); getTopoCoord().prepareHeartbeatRequestV1(startupTime + Milliseconds(1), setName, member); getTopoCoord().processHeartbeatResponse( - startupTime + Milliseconds(2), Milliseconds(1), member, hbResponseGood); + startupTime + Milliseconds(2), Milliseconds(1), member, hbResponseGood, OpTime()); getTopoCoord().prepareHeartbeatRequestV1(startupTime + Milliseconds(3), setName, member); Date_t timeoutTime = startupTime + Milliseconds(3) + ReplSetConfig::kDefaultHeartbeatTimeoutPeriod; @@ -1622,12 +1706,12 @@ TEST_F(TopoCoordTest, ReplSetGetStatus) { StatusWith<ReplSetHeartbeatResponse>(Status(ErrorCodes::HostUnreachable, "")); getTopoCoord().processHeartbeatResponse( - timeoutTime, Milliseconds(5000), member, hbResponseDown); + timeoutTime, Milliseconds(5000), member, hbResponseDown, OpTime()); member = HostAndPort("test1:1234"); getTopoCoord().prepareHeartbeatRequestV1(startupTime + Milliseconds(2), setName, member); getTopoCoord().processHeartbeatResponse( - heartbeatTime, Milliseconds(4000), member, hbResponseGood); + heartbeatTime, Milliseconds(4000), member, hbResponseGood, OpTime()); makeSelfPrimary(electionTime); getTopoCoord().setMyLastAppliedOpTime(oplogProgress, startupTime, false); getTopoCoord().setMyLastDurableOpTime(oplogDurable, startupTime, false); @@ -1814,7 +1898,7 @@ TEST_F(TopoCoordTest, HeartbeatFrequencyShouldBeHalfElectionTimeoutWhenArbiter) std::pair<ReplSetHeartbeatArgsV1, Milliseconds> uppingRequest = getTopoCoord().prepareHeartbeatRequestV1(requestDate, "myset", target); auto action = getTopoCoord().processHeartbeatResponse( - requestDate, Milliseconds(0), target, makeStatusWith<ReplSetHeartbeatResponse>()); + requestDate, Milliseconds(0), target, makeStatusWith<ReplSetHeartbeatResponse>(), OpTime()); Date_t expected(now() + Milliseconds(2500)); ASSERT_EQUALS(expected, action.getNextHeartbeatStartDate()); } @@ -3397,6 +3481,31 @@ TEST_F(HeartbeatResponseTestV1, now())); ASSERT_FALSE(getTopoCoord().shouldChangeSyncSource( HostAndPort("host2"), makeReplSetMetadata(newerThanLastOpTimeApplied), boost::none, now())); + + // If we are as up-to-date as this sync source, but it has a higher lastOpCommitted, we will not + // change sync sources. + OpTime lastOpCommitted = OpTime(Timestamp(100, 0), 0); + OpTime newerLastOpCommitted = OpTime(Timestamp(200, 0), 0); + ASSERT(getTopoCoord().advanceLastCommittedOpTime(lastOpCommitted)); + nextAction = receiveUpHeartbeat(HostAndPort("host2"), + "rs0", + MemberState::RS_SECONDARY, + election, + lastOpTimeApplied, + HostAndPort(), + newerLastOpCommitted); + ASSERT_NO_ACTION(nextAction.getAction()); + ASSERT_FALSE(getTopoCoord().shouldChangeSyncSource( + HostAndPort("host2"), + makeReplSetMetadata(), + makeOplogQueryMetadata(lastOpTimeApplied, -1, -1, newerLastOpCommitted), + now())); + + ASSERT_FALSE(getTopoCoord().shouldChangeSyncSource( + HostAndPort("host2"), + makeReplSetMetadata(lastOpTimeApplied, -1, -1, newerLastOpCommitted), + boost::none, + now())); } TEST_F(HeartbeatResponseTestV1, ShouldNotChangeSyncSourceWhenFresherMemberIsDown) { @@ -3653,7 +3762,7 @@ TEST_F(HeartbeatResponseTestV1, ReconfigNodeRemovedBetweenHeartbeatRequestAndRep hb.setElectionTime(election.getTimestamp()); StatusWith<ReplSetHeartbeatResponse> hbResponse = StatusWith<ReplSetHeartbeatResponse>(hb); HeartbeatResponseAction action = getTopoCoord().processHeartbeatResponse( - now()++, Milliseconds(0), HostAndPort("host3"), hbResponse); + now()++, Milliseconds(0), HostAndPort("host3"), hbResponse, OpTime()); // primary should not be set and we should perform NoAction in response ASSERT_EQUALS(-1, getCurrentPrimaryIndex()); @@ -3708,7 +3817,7 @@ TEST_F(HeartbeatResponseTestV1, ReconfigBetweenHeartbeatRequestAndRepsonse) { StatusWith<ReplSetHeartbeatResponse> hbResponse = StatusWith<ReplSetHeartbeatResponse>(hb); getTopoCoord().setMyLastAppliedOpTime(lastOpTimeApplied, Date_t(), false); HeartbeatResponseAction action = getTopoCoord().processHeartbeatResponse( - now()++, Milliseconds(0), HostAndPort("host3"), hbResponse); + now()++, Milliseconds(0), HostAndPort("host3"), hbResponse, OpTime()); // now primary should be host3, index 1, and we should perform NoAction in response ASSERT_EQUALS(1, getCurrentPrimaryIndex()); @@ -3884,13 +3993,15 @@ TEST_F(TopoCoordTest, FreshestNodeDoesCatchupTakeover) { getTopoCoord().processHeartbeatResponse(firstRequestDate + Milliseconds(1000), Milliseconds(999), HostAndPort("host3:27017"), - StatusWith<ReplSetHeartbeatResponse>(hbResp)); + StatusWith<ReplSetHeartbeatResponse>(hbResp), + OpTime()); hbResp.setAppliedOpTime(behindOptime); hbResp.setState(MemberState::RS_PRIMARY); getTopoCoord().processHeartbeatResponse(firstRequestDate + Milliseconds(1000), Milliseconds(999), HostAndPort("host2:27017"), - StatusWith<ReplSetHeartbeatResponse>(hbResp)); + StatusWith<ReplSetHeartbeatResponse>(hbResp), + OpTime()); getTopoCoord().updateTerm(1, Date_t()); ASSERT_OK(getTopoCoord().becomeCandidateIfElectable( @@ -3937,13 +4048,15 @@ TEST_F(TopoCoordTest, StaleNodeDoesntDoCatchupTakeover) { getTopoCoord().processHeartbeatResponse(firstRequestDate + Milliseconds(1000), Milliseconds(999), HostAndPort("host3:27017"), - StatusWith<ReplSetHeartbeatResponse>(hbResp)); + StatusWith<ReplSetHeartbeatResponse>(hbResp), + OpTime()); hbResp.setAppliedOpTime(behindOptime); hbResp.setState(MemberState::RS_PRIMARY); getTopoCoord().processHeartbeatResponse(firstRequestDate + Milliseconds(1000), Milliseconds(999), HostAndPort("host2:27017"), - StatusWith<ReplSetHeartbeatResponse>(hbResp)); + StatusWith<ReplSetHeartbeatResponse>(hbResp), + OpTime()); getTopoCoord().updateTerm(1, Date_t()); Status result = getTopoCoord().becomeCandidateIfElectable( @@ -3993,12 +4106,14 @@ TEST_F(TopoCoordTest, NodeDoesntDoCatchupTakeoverHeartbeatSaysPrimaryCaughtUp) { getTopoCoord().processHeartbeatResponse(firstRequestDate + Milliseconds(1000), Milliseconds(999), HostAndPort("host3:27017"), - StatusWith<ReplSetHeartbeatResponse>(hbResp)); + StatusWith<ReplSetHeartbeatResponse>(hbResp), + OpTime()); hbResp.setState(MemberState::RS_PRIMARY); getTopoCoord().processHeartbeatResponse(firstRequestDate + Milliseconds(1000), Milliseconds(999), HostAndPort("host2:27017"), - StatusWith<ReplSetHeartbeatResponse>(hbResp)); + StatusWith<ReplSetHeartbeatResponse>(hbResp), + OpTime()); getTopoCoord().updateTerm(1, Date_t()); Status result = getTopoCoord().becomeCandidateIfElectable( @@ -4051,13 +4166,15 @@ TEST_F(TopoCoordTest, NodeDoesntDoCatchupTakeoverIfTermNumbersSayPrimaryCaughtUp getTopoCoord().processHeartbeatResponse(firstRequestDate + Milliseconds(1000), Milliseconds(999), HostAndPort("host3:27017"), - StatusWith<ReplSetHeartbeatResponse>(hbResp)); + StatusWith<ReplSetHeartbeatResponse>(hbResp), + OpTime()); hbResp.setAppliedOpTime(behindOptime); hbResp.setState(MemberState::RS_PRIMARY); getTopoCoord().processHeartbeatResponse(firstRequestDate + Milliseconds(1000), Milliseconds(999), HostAndPort("host2:27017"), - StatusWith<ReplSetHeartbeatResponse>(hbResp)); + StatusWith<ReplSetHeartbeatResponse>(hbResp), + OpTime()); getTopoCoord().updateTerm(1, Date_t()); Status result = getTopoCoord().becomeCandidateIfElectable( @@ -5526,7 +5643,8 @@ TEST_F(HeartbeatResponseTestV1, NodeDoesNotRetryHeartbeatIfTheFirstFailureTakesT // no retry allowed. Milliseconds(4990), // Spent 4.99 of the 5 seconds in the network. target, - StatusWith<ReplSetHeartbeatResponse>(ErrorCodes::ExceededTimeLimit, "Took too long")); + StatusWith<ReplSetHeartbeatResponse>(ErrorCodes::ExceededTimeLimit, "Took too long"), + OpTime()); ASSERT_EQUALS(HeartbeatResponseAction::NoAction, action.getAction()); ASSERT_TRUE(TopologyCoordinator::Role::kFollower == getTopoCoord().getRole()); @@ -5626,8 +5744,12 @@ public: Date_t _upRequestDate = unittest::assertGet(dateFromISOString("2014-08-29T12:55Z")); std::pair<ReplSetHeartbeatArgsV1, Milliseconds> uppingRequest = getTopoCoord().prepareHeartbeatRequestV1(_upRequestDate, "rs0", _target); - HeartbeatResponseAction upAction = getTopoCoord().processHeartbeatResponse( - _upRequestDate, Milliseconds(0), _target, makeStatusWith<ReplSetHeartbeatResponse>()); + HeartbeatResponseAction upAction = + getTopoCoord().processHeartbeatResponse(_upRequestDate, + Milliseconds(0), + _target, + makeStatusWith<ReplSetHeartbeatResponse>(), + OpTime()); ASSERT_EQUALS(HeartbeatResponseAction::NoAction, upAction.getAction()); ASSERT_TRUE(TopologyCoordinator::Role::kFollower == getTopoCoord().getRole()); @@ -5646,8 +5768,8 @@ public: _firstRequestDate + Seconds(4), // 4 seconds elapsed, retry allowed. Milliseconds(3990), // Spent 3.99 of the 4 seconds in the network. _target, - StatusWith<ReplSetHeartbeatResponse>( - ErrorCodes::ExceededTimeLimit, "Took too long")); // We've never applied anything. + StatusWith<ReplSetHeartbeatResponse>(ErrorCodes::ExceededTimeLimit, "Took too long"), + OpTime()); // We've never applied anything. ASSERT_EQUALS(HeartbeatResponseAction::NoAction, action.getAction()); ASSERT_TRUE(TopologyCoordinator::Role::kFollower == getTopoCoord().getRole()); @@ -5704,8 +5826,8 @@ TEST_F(HeartbeatResponseTestOneRetryV1, // no retry allowed. Milliseconds(1000), // Spent 1 of the 1.01 seconds in the network. target(), - StatusWith<ReplSetHeartbeatResponse>(ErrorCodes::ExceededTimeLimit, - "Took too long")); // We've never applied anything. + StatusWith<ReplSetHeartbeatResponse>(ErrorCodes::ExceededTimeLimit, "Took too long"), + OpTime()); // We've never applied anything. ASSERT_EQUALS(HeartbeatResponseAction::NoAction, action.getAction()); ASSERT_TRUE(TopologyCoordinator::Role::kFollower == getTopoCoord().getRole()); @@ -5725,7 +5847,8 @@ public: // could retry. Milliseconds(400), // Spent 0.4 of the 0.5 seconds in the network. target(), - StatusWith<ReplSetHeartbeatResponse>(ErrorCodes::NodeNotFound, "Bad DNS?")); + StatusWith<ReplSetHeartbeatResponse>(ErrorCodes::NodeNotFound, "Bad DNS?"), + OpTime()); ASSERT_EQUALS(HeartbeatResponseAction::NoAction, action.getAction()); ASSERT_TRUE(TopologyCoordinator::Role::kFollower == getTopoCoord().getRole()); // Because the first retry failed without timing out, we expect to retry immediately. @@ -5772,7 +5895,8 @@ TEST_F(HeartbeatResponseTestTwoRetriesV1, NodeDoesNotRetryHeartbeatsAfterFailing // could still retry. Milliseconds(100), // Spent 0.1 of the 0.3 seconds in the network. target(), - StatusWith<ReplSetHeartbeatResponse>(ErrorCodes::NodeNotFound, "Bad DNS?")); + StatusWith<ReplSetHeartbeatResponse>(ErrorCodes::NodeNotFound, "Bad DNS?"), + OpTime()); ASSERT_EQUALS(HeartbeatResponseAction::NoAction, action.getAction()); ASSERT_TRUE(TopologyCoordinator::Role::kFollower == getTopoCoord().getRole()); // Because this is the second retry, rather than retry again, we expect to wait for a quarter @@ -5811,7 +5935,8 @@ TEST_F(HeartbeatResponseTestTwoRetriesV1, HeartbeatThreeNonconsecutiveFailures) getTopoCoord().processHeartbeatResponse(firstRequestDate() + Milliseconds(4500), Milliseconds(400), target(), - StatusWith<ReplSetHeartbeatResponse>(response)); + StatusWith<ReplSetHeartbeatResponse>(response), + OpTime()); ASSERT_EQUALS(HeartbeatResponseAction::NoAction, action.getAction()); ASSERT_TRUE(TopologyCoordinator::Role::kFollower == getTopoCoord().getRole()); @@ -5828,7 +5953,8 @@ TEST_F(HeartbeatResponseTestTwoRetriesV1, HeartbeatThreeNonconsecutiveFailures) firstRequestDate() + Milliseconds(7100), Milliseconds(400), target(), - StatusWith<ReplSetHeartbeatResponse>(Status{ErrorCodes::HostUnreachable, ""})); + StatusWith<ReplSetHeartbeatResponse>(Status{ErrorCodes::HostUnreachable, ""}), + OpTime()); ASSERT_EQUALS(HeartbeatResponseAction::NoAction, action.getAction()); ASSERT_TRUE(TopologyCoordinator::Role::kFollower == getTopoCoord().getRole()); @@ -5904,7 +6030,8 @@ TEST_F(HeartbeatResponseHighVerbosityTestV1, UpdateHeartbeatDataSameConfig) { now()++, // Time is left. Milliseconds(400), // Spent 0.4 of the 0.5 second in the network. HostAndPort("host2"), - StatusWith<ReplSetHeartbeatResponse>(sameConfigResponse)); + StatusWith<ReplSetHeartbeatResponse>(sameConfigResponse), + OpTime()); stopCapturingLogMessages(); ASSERT_NO_ACTION(action.getAction()); ASSERT_EQUALS(1, @@ -5929,7 +6056,8 @@ TEST_F(HeartbeatResponseHighVerbosityTestV1, now()++, // Time is left. Milliseconds(400), // Spent 0.4 of the 0.5 second in the network. HostAndPort("host5"), - StatusWith<ReplSetHeartbeatResponse>(memberMissingResponse)); + StatusWith<ReplSetHeartbeatResponse>(memberMissingResponse), + OpTime()); stopCapturingLogMessages(); ASSERT_NO_ACTION(action.getAction()); ASSERT_EQUALS(1, countLogLinesContaining("Could not find host5:27017 in current config")); |