SERVER-14449 Implement processReplSetSyncFrom in ReplicationCoordinatorImpl

stbrody · stbrody · commit 018b95b10a10 · 2014-08-19T12:56:55.000-04:00
diff --git a/src/mongo/db/repl/member_heartbeat_data.cpp b/src/mongo/db/repl/member_heartbeat_data.cpp
@@ -88,5 +88,11 @@ namespace repl {
         _authIssue = false;
     }
 
+    void MemberHeartbeatData::setAuthIssue() {
+        _state = MemberState::RS_UNKNOWN;
+        _health = 0; // set health to 0 so that this doesn't count towards majority.
+        _authIssue = true;
+    }
+
 } // namespace repl
 } // namespace mongo
diff --git a/src/mongo/db/repl/member_heartbeat_data.h b/src/mongo/db/repl/member_heartbeat_data.h
@@ -91,6 +91,12 @@ namespace repl {
          */
         void setDownValues(Date_t now, const std::string& heartbeatMessage);
 
+        /**
+         * Sets values in this object that indicate there was an auth issue on the last heartbeat
+         * command.
+         */
+        void setAuthIssue();
+
     private:
         // This member's index into the ReplicaSetConfig
         int _configIndex;
diff --git a/src/mongo/db/repl/repl_coordinator_impl.cpp b/src/mongo/db/repl/repl_coordinator_impl.cpp
@@ -706,8 +706,68 @@ namespace repl {
 
     Status ReplicationCoordinatorImpl::processReplSetSyncFrom(const std::string& target,
                                                               BSONObjBuilder* resultObj) {
-        // TODO
-        return Status::OK();
+        resultObj->append("syncFromRequested", target);
+
+        HostAndPort targetHostAndPort;
+        Status status = targetHostAndPort.initialize(target);
+        if (!status.isOK()) {
+            return status;
+        }
+
+        boost::lock_guard<boost::mutex> lock(_mutex);
+        const MemberConfig& selfConfig = _rsConfig.getMemberAt(_thisMembersConfigIndex);
+        if (selfConfig.isArbiter()) {
+            return Status(ErrorCodes::NotSecondary, "arbiters don't sync");
+        }
+        if (_getCurrentMemberState_inlock().primary()) {
+            return Status(ErrorCodes::NotSecondary, "primaries don't sync");
+        }
+
+        ReplicaSetConfig::MemberIterator targetConfig = _rsConfig.membersEnd();
+        int targetIndex = 0;
+        for (ReplicaSetConfig::MemberIterator it = _rsConfig.membersBegin();
+                it != _rsConfig.membersEnd(); ++it) {
+            if (it->getHostAndPort() == targetHostAndPort) {
+                targetConfig = it;
+                break;
+            }
+            ++targetIndex;
+        }
+        if (targetConfig == _rsConfig.membersEnd()) {
+            return Status(ErrorCodes::NodeNotFound,
+                          str::stream() << "Could not find member \"" << target <<
+                                  "\" in replica set");
+        }
+        if (targetIndex == _thisMembersConfigIndex) {
+            return Status(ErrorCodes::InvalidOptions, "I cannot sync from myself");
+        }
+        if (targetConfig->isArbiter()) {
+            return Status(ErrorCodes::InvalidOptions,
+                          str::stream() << "Cannot sync from \"" << target <<
+                                  "\" because it is an arbiter");
+        }
+        if (!targetConfig->shouldBuildIndexes() && selfConfig.shouldBuildIndexes()) {
+            // TODO(spencer): Is this check actually necessary?
+            return Status(ErrorCodes::InvalidOptions,
+                          str::stream() << "Cannot sync from \"" << target <<
+                                  "\" because it does not build indexes");
+        }
+
+        Status result(ErrorCodes::InternalError, "didn't set status in prepareSyncFromResponse");
+        CBHStatus cbh = _replExecutor.scheduleWork(
+            stdx::bind(&TopologyCoordinator::prepareSyncFromResponse,
+                       _topCoord.get(),
+                       stdx::placeholders::_1,
+                       targetIndex,
+                       _getLastOpApplied_inlock(),
+                       resultObj,
+                       &result));
+        if (cbh.getStatus() == ErrorCodes::ShutdownInProgress) {
+            return Status(ErrorCodes::ShutdownInProgress, "replication shutdown in progress");
+        }
+        fassert(18649, cbh.getStatus());
+        _replExecutor.wait(cbh.getValue());
+        return result;
     }
 
     Status ReplicationCoordinatorImpl::processReplSetMaintenance(OperationContext* txn,
diff --git a/src/mongo/db/repl/repl_coordinator_impl_test.cpp b/src/mongo/db/repl/repl_coordinator_impl_test.cpp
@@ -693,6 +693,71 @@ namespace {
         ASSERT_NOT_EQUALS(electionID1, electionID2);
     }
 
+    TEST_F(ReplCoordTest, TestProcessReplSetSyncFromSelfArbiter) {
+        init("mySet/test1:1234,test2:1234,test3:1234");
+        assertStartSuccess(
+                BSON("_id" << "mySet" <<
+                     "version" << 2 <<
+                     "members" << BSON_ARRAY(BSON("_id" << 0 <<
+                                                  "host" << "test0:1234" <<
+                                                  "arbiterOnly" << true) <<
+                                             BSON("_id" << 1 << "host" << "test1:1234"))),
+                HostAndPort("test0", 1234));
+
+        // Try to sync while we are an arbiter
+        BSONObjBuilder response;
+        Status result = getReplCoord()->processReplSetSyncFrom("test0:1234", &response);
+        ASSERT_EQUALS(ErrorCodes::NotSecondary, result);
+    }
+
+    TEST_F(ReplCoordTest, TestProcessReplSetSyncFrom) {
+        init("mySet/test1:1234,test2:1234,test3:1234");
+        assertStartSuccess(
+                BSON("_id" << "mySet" <<
+                     "version" << 2 <<
+                     "members" << BSON_ARRAY(BSON("_id" << 0 <<
+                                                  "host" << "test0:1234" <<
+                                                  "arbiterOnly" << true) <<
+                                             BSON("_id" << 1 << "host" << "test1:1234") <<
+                                             BSON("_id" << 2 <<
+                                                  "host" << "test2:1234" <<
+                                                  "priority" << 0 <<
+                                                  "buildIndexes" << false) <<
+                                             BSON("_id" << 3 << "host" << "test3:1234"))),
+                HostAndPort("test1", 1234));
+
+
+        // Try to sync from an invalid URL
+        BSONObjBuilder response;
+        Status result = getReplCoord()->processReplSetSyncFrom("", &response);
+        ASSERT_EQUALS(ErrorCodes::FailedToParse, result);
+
+        // Try to sync while in PRIMARY state
+        getTopoCoord()._changeMemberState(MemberState::RS_PRIMARY);
+        result = getReplCoord()->processReplSetSyncFrom("test1:1234", &response);
+        ASSERT_EQUALS(ErrorCodes::NotSecondary, result);
+
+        // Try to sync from self
+        getTopoCoord()._changeMemberState(MemberState::RS_SECONDARY);
+        result = getReplCoord()->processReplSetSyncFrom("test1:1234", &response);
+        ASSERT_EQUALS(ErrorCodes::InvalidOptions, result);
+
+        // Try to sync from non-existent member
+        result = getReplCoord()->processReplSetSyncFrom("fakemember:1234", &response);
+        ASSERT_EQUALS(ErrorCodes::NodeNotFound, result);
+
+        // Try to sync from an arbiter
+        result = getReplCoord()->processReplSetSyncFrom("test0:1234", &response);
+        ASSERT_EQUALS(ErrorCodes::InvalidOptions, result);
+
+        // Try to sync from a node that doesn't build indexes
+        result = getReplCoord()->processReplSetSyncFrom("test2:1234", &response);
+        ASSERT_EQUALS(ErrorCodes::InvalidOptions, result);
+
+        // Finally sync from someone valid
+        result = getReplCoord()->processReplSetSyncFrom("test3:1234", &response);
+        ASSERT_OK(result);
+    }
     // TODO(spencer): Unit test replSetFreeze
 
 }  // namespace
diff --git a/src/mongo/db/repl/topology_coordinator.h b/src/mongo/db/repl/topology_coordinator.h
@@ -91,6 +91,13 @@ namespace repl {
         // Applier calls this to notify that it's now safe to transition from SECONDARY to PRIMARY
         virtual void signalDrainComplete() = 0;
 
+        // produces a reply to a replSetSyncFrom command
+        virtual void prepareSyncFromResponse(const ReplicationExecutor::CallbackData& data,
+                                             int targetIndex,
+                                             const OpTime& lastOpApplied,
+                                             BSONObjBuilder* response,
+                                             Status* result) = 0;
+
         // produce a reply to a RAFT-style RequestVote RPC
         virtual void prepareRequestVoteResponse(const Date_t now,
                                                 const BSONObj& cmdObj,
diff --git a/src/mongo/db/repl/topology_coordinator_impl.cpp b/src/mongo/db/repl/topology_coordinator_impl.cpp
@@ -39,6 +39,7 @@
 #include "mongo/db/repl/repl_set_heartbeat_args.h"
 #include "mongo/db/repl/repl_set_heartbeat_response.h"
 #include "mongo/db/repl/replication_executor.h"
+#include "mongo/db/repl/rslog.h"
 #include "mongo/db/server_parameters.h"
 #include "mongo/util/log.h"
 #include "mongo/util/mongoutils/str.h"
@@ -283,6 +284,51 @@ namespace repl {
 */
     }
 
+    void TopologyCoordinatorImpl::prepareSyncFromResponse(
+            const ReplicationExecutor::CallbackData& data,
+            int targetIndex,
+            const OpTime& lastOpApplied,
+            BSONObjBuilder* response,
+            Status* result) {
+        if (data.status == ErrorCodes::CallbackCanceled) {
+            *result = Status(ErrorCodes::ShutdownInProgress, "replication system is shutting down");
+            return;
+        }
+
+        const MemberConfig& targetConfig = _currentConfig.getMemberAt(targetIndex);
+        const MemberHeartbeatData& hbdata = _hbdata[targetIndex];
+        const std::string targetHostAndPort = targetConfig.getHostAndPort().toString();
+        if (hbdata.hasAuthIssue()) {
+            *result = Status(ErrorCodes::Unauthorized,
+                             "not authorized to communicate with " + targetHostAndPort);
+            return;
+        }
+        if (hbdata.getHealth() == 0) {
+            *result = Status(ErrorCodes::HostUnreachable,
+                             str::stream() << "I cannot reach the requested member: \"" <<
+                                     targetHostAndPort << "\"");
+            return;
+        }
+        if (hbdata.getOpTime().getSecs()+10 < lastOpApplied.getSecs()) {
+            warning() << "attempting to sync from " << targetHostAndPort
+                      << ", but its latest opTime is " << hbdata.getOpTime().getSecs()
+                      << " and ours is " << lastOpApplied.getSecs() << " so this may not work"
+                      << rsLog;
+            response->append("warning",
+                             "requested member \"" + targetHostAndPort + "\" is more than 10 "
+                                     "seconds behind us");
+            // not returning bad Status, just warning
+        }
+
+        HostAndPort prevSyncSource = getSyncSourceAddress();
+        if (!prevSyncSource.empty()) {
+            response->append("prevSyncTarget", prevSyncSource.toString());
+        }
+
+        setForceSyncSourceIndex(targetIndex);
+        *result = Status::OK();
+    }
+
     // Produce a reply to a RAFT-style RequestVote RPC; this is MongoDB ReplSetFresh command
     // The caller should validate that the message is for the correct set, and has the required data
     void TopologyCoordinatorImpl::prepareRequestVoteResponse(const Date_t now,
diff --git a/src/mongo/db/repl/topology_coordinator_impl.h b/src/mongo/db/repl/topology_coordinator_impl.h
@@ -74,6 +74,7 @@ namespace repl {
 
         virtual void setCommitOkayThrough(const OpTime& optime);
         virtual void setLastReceived(const OpTime& optime);
+        // TODO(spencer): Can this be made private?
         virtual void setForceSyncSourceIndex(int index);
 
         // Looks up syncSource's address and returns it, for use by the Applier
@@ -94,6 +95,13 @@ namespace repl {
         // Applier calls this to notify that it's now safe to transition from SECONDARY to PRIMARY
         virtual void signalDrainComplete();
 
+        // produces a reply to a replSetSyncFrom command
+        virtual void prepareSyncFromResponse(const ReplicationExecutor::CallbackData& data,
+                                             int targetIndex,
+                                             const OpTime& lastOpApplied,
+                                             BSONObjBuilder* response,
+                                             Status* result);
+
         // produces a reply to a RAFT-style RequestVote RPC
         virtual void prepareRequestVoteResponse(const Date_t now,
                                                 const BSONObj& cmdObj,
@@ -208,7 +216,7 @@ namespace repl {
         // the member we currently believe is primary, if one exists
         int _currentPrimaryIndex;
         // the member we are currently syncing from
-        // NULL if no sync source (we are primary, or we cannot connect to anyone yet)
+        // -1 if no sync source (we are primary, or we cannot connect to anyone yet)
         int _syncSourceIndex; 
         // These members are not chosen as sync sources for a period of time, due to connection
         // issues with them
@@ -238,7 +246,10 @@ namespace repl {
         const MemberConfig& _selfConfig();  // Helper shortcut to self config
 
         ReplicaSetConfig _currentConfig; // The current config, including a vector of MemberConfigs
-        std::vector<MemberHeartbeatData> _hbdata; // heartbeat data for each member
+        // heartbeat data for each member.  It is guaranteed that this vector will be maintained
+        // in the same order as the MemberConfigs in _currentConfig, therefore the member config
+        // index can be used to index into this vector as well.
+        std::vector<MemberHeartbeatData> _hbdata;
 
         // Time when stepDown command expires
         Date_t _stepDownUntil;
diff --git a/src/mongo/db/repl/topology_coordinator_impl_test.cpp b/src/mongo/db/repl/topology_coordinator_impl_test.cpp
@@ -386,6 +386,72 @@ namespace {
         ASSERT_EQUALS(topocoord.getSyncSourceAddress(),HostAndPort("h3"));
     }
 
+    TEST(TopologyCoordinator, PrepareSyncFromResponse) {
+        ReplicationExecutor::CallbackHandle cbh;
+        ReplicationExecutor::CallbackData cbData(NULL,
+                                                 cbh,
+                                                 Status::OK());
+        ReplicaSetConfig config;
+
+        ASSERT_OK(config.initialize(BSON("_id" << "rs0" <<
+                                         "version" << 1 <<
+                                         "members" << BSON_ARRAY(
+                                             BSON("_id" << 10 << "host" << "hself") <<
+                                             BSON("_id" << 20 << "host" << "h1") <<
+                                             BSON("_id" << 30 << "host" << "h2") <<
+                                             BSON("_id" << 40 << "host" << "h3")))));
+
+        OpTime staleOpTime(1, 1);
+        OpTime ourOpTime(staleOpTime.getSecs() + 11, 1);
+
+
+        TopologyCoordinatorImpl topocoord((Seconds(999)));
+        Date_t now = 0;
+        topocoord.updateConfig(cbData, config, 0, now++, OpTime(0,0));
+
+        MemberHeartbeatData hselfInfo(0);
+        hselfInfo.setAuthIssue();
+        topocoord.updateHeartbeatData(now++, hselfInfo, 10, OpTime(0,0));
+
+        MemberHeartbeatData h1Info(1);
+        h1Info.setDownValues(now, "");
+        topocoord.updateHeartbeatData(now++, h1Info, 20, OpTime(0,0));
+
+        MemberHeartbeatData h2Info(2);
+        h2Info.setUpValues(now, MemberState::RS_SECONDARY, OpTime(0,0), staleOpTime, "", "");
+        topocoord.updateHeartbeatData(now++, h2Info, 30, OpTime(0,0));
+
+        MemberHeartbeatData h3Info(2);
+        h3Info.setUpValues(now, MemberState::RS_SECONDARY, OpTime(0,0), ourOpTime, "", "");
+        topocoord.updateHeartbeatData(now++, h3Info, 40, OpTime(0,0));
+
+        Status result = Status::OK();
+        BSONObjBuilder response0;
+        topocoord.prepareSyncFromResponse(cbData, 0, ourOpTime, &response0, &result);
+        ASSERT_EQUALS(ErrorCodes::Unauthorized, result);
+
+        BSONObjBuilder response1;
+        topocoord.prepareSyncFromResponse(cbData, 1, ourOpTime, &response1, &result);
+        ASSERT_EQUALS(ErrorCodes::HostUnreachable, result);
+
+        BSONObjBuilder response2;
+        topocoord.prepareSyncFromResponse(cbData, 2, ourOpTime, &response2, &result);
+        ASSERT_OK(result);
+        topocoord.chooseNewSyncSource(now++, ourOpTime);
+        ASSERT_EQUALS(topocoord.getSyncSourceAddress(), HostAndPort("h2"));
+        ASSERT_EQUALS("requested member \"h2:27017\" is more than 10 seconds behind us",
+                      response2.obj()["warning"].String());
+
+        BSONObjBuilder response3;
+        topocoord.prepareSyncFromResponse(cbData, 3, ourOpTime, &response3, &result);
+        ASSERT_OK(result);
+        topocoord.chooseNewSyncSource(now++, ourOpTime);
+        ASSERT_EQUALS(topocoord.getSyncSourceAddress(), HostAndPort("h3"));
+        BSONObj response3Obj = response3.obj();
+        ASSERT_FALSE(response3Obj.hasField("warning"));
+        ASSERT_EQUALS(HostAndPort("h2").toString(), response3Obj["prevSyncTarget"].String());
+    }
+
     TEST(TopologyCoordinator, ReplSetGetStatus) {
         // This test starts by configuring a TopologyCoordinator as a member of a 4 node replica
         // set, with each node in a different state.
