SERVER-15698 remove RID from replica set replication progress tracking

Conflicts:
	src/mongo/db/repl/sync_source_feedback.cpp

Author: matt dannenberg

src/mongo/db/repl/SConscript

@@ -173,6 +173,7 @@ env.Library('replmocks',
             LIBDEPS=[
                 '$BUILD_DIR/mongo/db/concurrency/lock_manager',
                 'repl_coordinator_interface',
+                'replica_set_messages',
                 'replication_executor',
             ])
 

src/mongo/db/repl/replication_coordinator.h

@@ -53,6 +53,7 @@ namespace repl {
     class HandshakeArgs;
     class IsMasterResponse;
     class OplogReader;
+    class ReplicaSetConfig;
     class ReplSetHeartbeatArgs;
     class ReplSetHeartbeatResponse;
     class UpdatePositionArgs;
@@ -337,15 +338,6 @@ namespace repl {
          */
         virtual bool prepareReplSetUpdatePositionCommand(BSONObjBuilder* cmdBuilder) = 0;
 
-        /**
-         * For ourself and each secondary chaining off of us, adds a BSONObj to "handshakes"
-         * describing an invocation of the replSetUpdateCommand that can be sent to this node's
-         * sync source to handshake us and our chained secondaries, informing the sync source that
-         * we are replicating off of it.
-         */
-        virtual void prepareReplSetUpdatePositionCommandHandshakes(
-                std::vector<BSONObj>* handshakes) = 0;
-
         /**
          * Handles an incoming replSetGetStatus command. Adds BSON to 'result'.
          */
@@ -363,6 +355,11 @@ namespace repl {
          */
         virtual void appendSlaveInfoData(BSONObjBuilder* result) = 0;
 
+        /**
+         * Returns a copy of the current ReplicaSetConfig.
+         */
+        virtual ReplicaSetConfig getConfig() const = 0;
+
         /**
          * Handles an incoming replSetGetConfig command. Adds BSON to 'result'.
          */
@@ -477,24 +474,25 @@ namespace repl {
          * Handles an incoming replSetUpdatePosition command, updating each node's oplog progress.
          * Returns Status::OK() if all updates are processed correctly, NodeNotFound
          * if any updating node cannot be found in the config, InvalidReplicaSetConfig if the
-         * "cfgver" sent in any of the updates doesn't match our config version, or
+         * "configVersion" sent in any of the updates doesn't match our config version, or
          * NotMasterOrSecondaryCode if we are in state REMOVED or otherwise don't have a valid
          * replica set config.
          * If a non-OK status is returned, it is unspecified whether none or some of the updates
          * were applied.
+         * "configVersion" will be populated with our config version if and only if we return
+         * InvalidReplicaSetConfig.
          */
-        virtual Status processReplSetUpdatePosition(const UpdatePositionArgs& updates) = 0;
+        virtual Status processReplSetUpdatePosition(const UpdatePositionArgs& updates,
+                                                    long long* configVersion) = 0;
 
-        /**
-         * Handles an incoming Handshake command (or a handshake from replSetUpdatePosition).
-         * Associates the node's 'remoteID' with its 'handshake' object. This association is used
-         * to update local.slaves and to forward the node's replication progress upstream when this
-         * node is being chained through.
+        /** 
+         * Handles an incoming Handshake command. Associates the node's 'remoteID' with its
+         * 'handshake' object. This association is used to update internal representation of 
+         * replication progress and to forward the node's replication progress upstream when this
+         * node is being chained through in master/slave replication.
          *
-         * Returns ErrorCodes::NodeNotFound if no replica set member exists with the given member ID
-         * and ErrorCodes::NotMasterOrSecondaryCode if we're in state REMOVED or otherwise don't
-         * have a valid config.
-         */
+         * Returns ErrorCodes::IllegalOperation if we're not running with master/slave replication.  
+         */  
         virtual Status processHandshake(OperationContext* txn, const HandshakeArgs& handshake) = 0;
 
         /**

src/mongo/db/repl/replication_coordinator_external_state.h

@@ -81,13 +81,6 @@ namespace repl {
          */
         virtual void initiateOplog(OperationContext* txn) = 0;
 
-        /**
-         * Simple wrapper around SyncSourceFeedback::forwardSlaveHandshake.  Signals to the
-         * SyncSourceFeedback thread that it needs to wake up and send a replication handshake
-         * upstream.
-         */
-        virtual void forwardSlaveHandshake() = 0;
-
         /**
          * Simple wrapper around SyncSourceFeedback::forwardSlaveProgress.  Signals to the
          * SyncSourceFeedback thread that it needs to wake up and send a replSetUpdatePosition

src/mongo/db/repl/replication_coordinator_external_state_impl.cpp

@@ -113,10 +113,6 @@ namespace {
         logOpInitiate(txn, BSON("msg" << "initiating set"));
     }
 
-    void ReplicationCoordinatorExternalStateImpl::forwardSlaveHandshake() {
-        _syncSourceFeedback.forwardSlaveHandshake();
-    }
-
     void ReplicationCoordinatorExternalStateImpl::forwardSlaveProgress() {
         _syncSourceFeedback.forwardSlaveProgress();
     }

src/mongo/db/repl/replication_coordinator_external_state_impl.h

@@ -49,7 +49,6 @@ namespace repl {
         virtual void startMasterSlave(OperationContext* txn);
         virtual void shutdown();
         virtual void initiateOplog(OperationContext* txn);
-        virtual void forwardSlaveHandshake();
         virtual void forwardSlaveProgress();
         virtual OID ensureMe(OperationContext* txn);
         virtual bool isSelf(const HostAndPort& host);

src/mongo/db/repl/replication_coordinator_external_state_mock.cpp

@@ -57,7 +57,6 @@ namespace repl {
     void ReplicationCoordinatorExternalStateMock::startMasterSlave(OperationContext*) {}
     void ReplicationCoordinatorExternalStateMock::initiateOplog(OperationContext* txn) {}
     void ReplicationCoordinatorExternalStateMock::shutdown() {}
-    void ReplicationCoordinatorExternalStateMock::forwardSlaveHandshake() {}
     void ReplicationCoordinatorExternalStateMock::forwardSlaveProgress() {}
 
     OID ReplicationCoordinatorExternalStateMock::ensureMe(OperationContext*) {

src/mongo/db/repl/replication_coordinator_external_state_mock.h

@@ -52,7 +52,6 @@ namespace repl {
         virtual void startMasterSlave(OperationContext*);
         virtual void shutdown();
         virtual void initiateOplog(OperationContext* txn);
-        virtual void forwardSlaveHandshake();
         virtual void forwardSlaveProgress();
         virtual OID ensureMe(OperationContext*);
         virtual bool isSelf(const HostAndPort& host);

src/mongo/db/repl/replication_coordinator_impl.cpp

@@ -541,7 +541,7 @@ namespace {
     }
 
     void ReplicationCoordinatorImpl::signalUpstreamUpdater() {
-        _externalState->forwardSlaveHandshake();
+        _externalState->forwardSlaveProgress();
     }
 
     ReplicationCoordinatorImpl::SlaveInfo*
@@ -717,16 +717,19 @@ namespace {
         return _slaveInfo[_getMyIndexInSlaveInfo_inlock()].opTime;
     }
 
-    Status ReplicationCoordinatorImpl::setLastOptime_forTest(const OID& rid, const OpTime& ts) {
+    Status ReplicationCoordinatorImpl::setLastOptime_forTest(long long cfgVer,
+                                                             long long memberId,
+                                                             const OpTime& ts) {
         boost::lock_guard<boost::mutex> lock(_mutex);
         invariant(_getReplicationMode_inlock() == modeReplSet);
 
-        const UpdatePositionArgs::UpdateInfo update(rid, ts, -1, -1);
-        return _setLastOptime_inlock(update);
+        const UpdatePositionArgs::UpdateInfo update(OID(), ts, cfgVer, memberId);
+        long long configVersion;
+        return _setLastOptime_inlock(update, &configVersion);
     }
 
     Status ReplicationCoordinatorImpl::_setLastOptime_inlock(
-            const UpdatePositionArgs::UpdateInfo& args) {
+            const UpdatePositionArgs::UpdateInfo& args, long long* configVersion) {
 
         if (_selfIndex == -1) {
             // Ignore updates when we're in state REMOVED
@@ -735,57 +738,50 @@ namespace {
         }
         invariant(_getReplicationMode_inlock() == modeReplSet);
 
+        if (args.memberId < 0) {
+            std::string errmsg = str::stream()
+                      << "Received replSetUpdatePosition for node with memberId "
+                      << args.memberId << " which is negative and therefore invalid";
+            LOG(1) << errmsg;
+            return Status(ErrorCodes::NodeNotFound, errmsg);
+        }
+
         if (args.rid == _getMyRID_inlock() ||
                 args.memberId == _rsConfig.getMemberAt(_selfIndex).getId()) {
             // Do not let remote nodes tell us what our optime is.
             return Status::OK();
         }
 
-        LOG(2) << "received notification that node with RID " << args.rid <<
-                " has reached optime: " << args.ts;
+        LOG(2) << "received notification that node with memberID " << args.memberId <<
+                  " in config with version " << args.cfgver << " has reached optime: " << args.ts;
 
         SlaveInfo* slaveInfo = NULL;
-        if (args.memberId >= 0) {
-            if (args.cfgver != _rsConfig.getConfigVersion()) {
-                std::string errmsg = str::stream()
-                          << "Received replSetUpdatePosition for node with memberId "
-                          << args.memberId << " whose config version of " << args.cfgver
-                          << " doesn't match our config version of "
-                          << _rsConfig.getConfigVersion();
-                LOG(1) << errmsg;
-                return Status(ErrorCodes::InvalidReplicaSetConfig, errmsg);
-            }
+        if (args.cfgver != _rsConfig.getConfigVersion()) {
+            std::string errmsg = str::stream()
+                      << "Received replSetUpdatePosition for node with memberId "
+                      << args.memberId << " whose config version of " << args.cfgver
+                      << " doesn't match our config version of "
+                      << _rsConfig.getConfigVersion();
+            LOG(1) << errmsg;
+            *configVersion = _rsConfig.getConfigVersion();
+            return Status(ErrorCodes::InvalidReplicaSetConfig, errmsg);
+        }
 
-            slaveInfo = _findSlaveInfoByMemberID_inlock(args.memberId);
-            if (!slaveInfo) {
-                invariant(!_rsConfig.findMemberByID(args.memberId));
+        slaveInfo = _findSlaveInfoByMemberID_inlock(args.memberId);
+        if (!slaveInfo) {
+            invariant(!_rsConfig.findMemberByID(args.memberId));
 
-                std::string errmsg = str::stream()
-                          << "Received replSetUpdatePosition for node with memberId "
-                          << args.memberId << " which doesn't exist in our config";
-                LOG(1) << errmsg;
-                return Status(ErrorCodes::NodeNotFound, errmsg);
-            }
-        }
-        else {
-            // The command we received didn't contain a memberId, most likely this is because it
-            // came from a member running something prior to 3.0.
-            // Fall back to finding the node by RID.
-            slaveInfo = _findSlaveInfoByRID_inlock(args.rid);
-            if (!slaveInfo) {
-                std::string errmsg = str::stream()
-                          << "Received replSetUpdatePosition for node with RID " << args.rid
-                          << ", but we haven't yet received a handshake for that node.";
-                LOG(1) << errmsg;
-                return Status(ErrorCodes::NodeNotFound, errmsg);
-            }
-            invariant(slaveInfo->memberId >= 0);
+            std::string errmsg = str::stream()
+                      << "Received replSetUpdatePosition for node with memberId "
+                      << args.memberId << " which doesn't exist in our config";
+            LOG(1) << errmsg;
+            return Status(ErrorCodes::NodeNotFound, errmsg);
         }
-        invariant(slaveInfo);
-        invariant(args.memberId < 0 || args.memberId == slaveInfo->memberId);
 
-        LOG(3) << "Node with RID " << args.rid << " and memberId " << slaveInfo->memberId
-               << " currently has optime " << slaveInfo->opTime << "; updating to " << args.ts;
+        invariant(args.memberId == slaveInfo->memberId);
+
+        LOG(3) << "Node with memberID " << args.memberId << " currently has optime " <<
+                  slaveInfo->opTime << "; updating to " << args.ts;
 
         // Only update remote optimes if they increase.
         if (slaveInfo->opTime < args.ts) {
@@ -1345,41 +1341,13 @@ namespace {
                 // we need to keep sending it for 2.6 compatibility.
                 // TODO(spencer): Remove this after 3.0 is released.
                 const MemberConfig* member = _rsConfig.findMemberByID(itr->memberId);
-                fassert(18651, member); // We ensured the member existed in processHandshake.
+                fassert(18651, member);
                 entry.append("config", member->toBSON(_rsConfig.getTagConfig()));
             }
         }
         return true;
     }
 
-    void ReplicationCoordinatorImpl::prepareReplSetUpdatePositionCommandHandshakes(
-            std::vector<BSONObj>* handshakes) {
-        boost::lock_guard<boost::mutex> lock(_mutex);
-        // handshake objs for ourself and all chained members
-        for (SlaveInfoVector::const_iterator itr = _slaveInfo.begin();
-                itr != _slaveInfo.end(); ++itr) {
-            if (!itr->rid.isSet()) {
-                // Don't include info on members we haven't heard from yet.
-                continue;
-            }
-
-            BSONObjBuilder cmd;
-            cmd.append("replSetUpdatePosition", 1);
-            {
-                BSONObjBuilder subCmd (cmd.subobjStart("handshake"));
-                subCmd.append("handshake", itr->rid);
-                subCmd.append("member", itr->memberId);
-                // SERVER-14550 Even though the "config" field isn't used on the other end in 3.0,
-                // we need to keep sending it for 2.6 compatibility.
-                // TODO(spencer): Remove this after 3.0 is released.
-                const MemberConfig* member = _rsConfig.findMemberByID(itr->memberId);
-                fassert(18650, member); // We ensured the member existed in processHandshake.
-                subCmd.append("config", member->toBSON(_rsConfig.getTagConfig()));
-            }
-            handshakes->push_back(cmd.obj());
-        }
-    }
-
     Status ReplicationCoordinatorImpl::processReplSetGetStatus(BSONObjBuilder* response) {
         Status result(ErrorCodes::InternalError, "didn't set status in prepareStatusResponse");
         CBHStatus cbh = _replExecutor.scheduleWork(
@@ -1452,6 +1420,11 @@ namespace {
         }
     }
 
+    ReplicaSetConfig ReplicationCoordinatorImpl::getConfig() const {
+        boost::lock_guard<boost::mutex> lock(_mutex);
+        return _rsConfig;
+    }
+
     void ReplicationCoordinatorImpl::processReplSetGetConfig(BSONObjBuilder* result) {
         boost::lock_guard<boost::mutex> lock(_mutex);
         result->append("config", _rsConfig.toBSON());
@@ -2109,15 +2082,15 @@ namespace {
     }
 
     Status ReplicationCoordinatorImpl::processReplSetUpdatePosition(
-            const UpdatePositionArgs& updates) {
+            const UpdatePositionArgs& updates, long long* configVersion) {
 
         boost::unique_lock<boost::mutex> lock(_mutex);
         Status status = Status::OK();
         bool somethingChanged = false;
         for (UpdatePositionArgs::UpdateIterator update = updates.updatesBegin();
                 update != updates.updatesEnd();
                 ++update) {
-            status = _setLastOptime_inlock(*update);
+            status = _setLastOptime_inlock(*update, configVersion);
             if (!status.isOK()) {
                 break;
             }
@@ -2136,36 +2109,12 @@ namespace {
         LOG(2) << "Received handshake " << handshake.toBSON();
 
         boost::unique_lock<boost::mutex> lock(_mutex);
-        if (_getReplicationMode_inlock() == modeReplSet) {
-            if (_selfIndex == -1) {
-                // Ignore updates when we're in state REMOVED
-                return Status(ErrorCodes::NotMasterOrSecondaryCode,
-                              "Received replSetUpdatePosition command but we are in state REMOVED");
-            }
 
-            int memberId = handshake.getMemberId();
-            const MemberConfig* member = _rsConfig.findMemberByID(memberId);
-            if (!member) {
-                return Status(ErrorCodes::NodeNotFound,
-                              str::stream() << "Node with replica set memberId " << memberId <<
-                                      " could not be found in replica set config while attempting"
-                                      " to associate it with RID " << handshake.getRid() <<
-                                      " in replication handshake.  ReplSet Config: " <<
-                                      _rsConfig.toBSON().toString());
-            }
-            SlaveInfo* slaveInfo = _findSlaveInfoByMemberID_inlock(handshake.getMemberId());
-            invariant(slaveInfo); // If it's in the config it must be in _slaveInfo
-            slaveInfo->rid = handshake.getRid();
-            slaveInfo->hostAndPort = member->getHostAndPort();
-
-            if (!_getMemberState_inlock().primary()) {
-                lock.unlock();
-                _externalState->forwardSlaveHandshake(); // must do outside _mutex
-            }
-            return Status::OK();
+        if (_getReplicationMode_inlock() != modeMasterSlave) {
+            return Status(ErrorCodes::IllegalOperation,
+                          "The handshake command is only used for master/slave replication");
         }
 
-        // master-slave from here down
         SlaveInfo* slaveInfo = _findSlaveInfoByRID_inlock(handshake.getRid());
         if (slaveInfo) {
             return Status::OK(); // nothing to do