SERVER-24222 Update current known primary from command metadata

visualzhou · visualzhou · commit 504c299109fd · 2016-06-01T15:56:27.000-04:00
This reverts commit ed3f25c. Fixed replication legacy test suite.
diff --git a/src/mongo/db/repl/bgsync.cpp b/src/mongo/db/repl/bgsync.cpp
@@ -85,8 +85,7 @@ class DataReplicatorExternalStateBackgroundSync : public DataReplicatorExternalS
         ReplicationCoordinatorExternalState* replicationCoordinatorExternalState,
         BackgroundSync* bgsync);
     bool shouldStopFetching(const HostAndPort& source,
-                            const OpTime& sourceOpTime,
-                            bool sourceHasSyncSource) override;
+                            const rpc::ReplSetMetadata& metadata) override;
 
 private:
     BackgroundSync* _bgsync;
@@ -99,15 +98,13 @@ DataReplicatorExternalStateBackgroundSync::DataReplicatorExternalStateBackground
     : DataReplicatorExternalStateImpl(replicationCoordinator, replicationCoordinatorExternalState),
       _bgsync(bgsync) {}
 
-bool DataReplicatorExternalStateBackgroundSync::shouldStopFetching(const HostAndPort& source,
-                                                                   const OpTime& sourceOpTime,
-                                                                   bool sourceHasSyncSource) {
+bool DataReplicatorExternalStateBackgroundSync::shouldStopFetching(
+    const HostAndPort& source, const rpc::ReplSetMetadata& metadata) {
     if (_bgsync->shouldStopFetching()) {
         return true;
     }
 
-    return DataReplicatorExternalStateImpl::shouldStopFetching(
-        source, sourceOpTime, sourceHasSyncSource);
+    return DataReplicatorExternalStateImpl::shouldStopFetching(source, metadata);
 }
 
 /**
diff --git a/src/mongo/db/repl/data_replicator_external_state.h b/src/mongo/db/repl/data_replicator_external_state.h
@@ -77,8 +77,7 @@ class DataReplicatorExternalState {
      * metadata).
      */
     virtual bool shouldStopFetching(const HostAndPort& source,
-                                    const OpTime& sourceOpTime,
-                                    bool sourceHasSyncSource) = 0;
+                                    const rpc::ReplSetMetadata& metadata) = 0;
 
 private:
     /**
diff --git a/src/mongo/db/repl/data_replicator_external_state_impl.cpp b/src/mongo/db/repl/data_replicator_external_state_impl.cpp
@@ -60,14 +60,12 @@ void DataReplicatorExternalStateImpl::processMetadata(const rpc::ReplSetMetadata
 }
 
 bool DataReplicatorExternalStateImpl::shouldStopFetching(const HostAndPort& source,
-                                                         const OpTime& sourceOpTime,
-                                                         bool sourceHasSyncSource) {
+                                                         const rpc::ReplSetMetadata& metadata) {
     // Re-evaluate quality of sync target.
-    if (_replicationCoordinator->shouldChangeSyncSource(
-            source, sourceOpTime, sourceHasSyncSource)) {
+    if (_replicationCoordinator->shouldChangeSyncSource(source, metadata)) {
         LOG(1) << "Canceling oplog query because we have to choose a sync source. Current source: "
-               << source << ", OpTime " << sourceOpTime
-               << ", hasSyncSource:" << sourceHasSyncSource;
+               << source << ", OpTime " << metadata.getLastOpVisible()
+               << ", its sync source index:" << metadata.getSyncSourceIndex();
         return true;
     }
     return false;
diff --git a/src/mongo/db/repl/data_replicator_external_state_impl.h b/src/mongo/db/repl/data_replicator_external_state_impl.h
@@ -51,8 +51,7 @@ class DataReplicatorExternalStateImpl : public DataReplicatorExternalState {
     void processMetadata(const rpc::ReplSetMetadata& metadata) override;
 
     bool shouldStopFetching(const HostAndPort& source,
-                            const OpTime& sourceOpTime,
-                            bool sourceHasSyncSource) override;
+                            const rpc::ReplSetMetadata& metadata) override;
 
 private:
     StatusWith<OpTime> _multiApply(OperationContext* txn,
diff --git a/src/mongo/db/repl/data_replicator_external_state_mock.cpp b/src/mongo/db/repl/data_replicator_external_state_mock.cpp
@@ -47,11 +47,10 @@ void DataReplicatorExternalStateMock::processMetadata(const rpc::ReplSetMetadata
 }
 
 bool DataReplicatorExternalStateMock::shouldStopFetching(const HostAndPort& source,
-                                                         const OpTime& sourceOpTime,
-                                                         bool sourceHasSyncSource) {
+                                                         const rpc::ReplSetMetadata& metadata) {
     lastSyncSourceChecked = source;
-    syncSourceLastOpTime = sourceOpTime;
-    syncSourceHasSyncSource = sourceHasSyncSource;
+    syncSourceLastOpTime = metadata.getLastOpVisible();
+    syncSourceHasSyncSource = metadata.getSyncSourceIndex() != -1;
     return shouldStopFetchingResult;
 }
 
diff --git a/src/mongo/db/repl/data_replicator_external_state_mock.h b/src/mongo/db/repl/data_replicator_external_state_mock.h
@@ -48,8 +48,7 @@ class DataReplicatorExternalStateMock : public DataReplicatorExternalState {
     void processMetadata(const rpc::ReplSetMetadata& metadata) override;
 
     bool shouldStopFetching(const HostAndPort& source,
-                            const OpTime& sourceOpTime,
-                            bool sourceHasSyncSource) override;
+                            const rpc::ReplSetMetadata& metadata) override;
 
     // Returned by getCurrentTermAndLastCommittedOpTime.
     long long currentTerm = OpTime::kUninitializedTerm;
diff --git a/src/mongo/db/repl/data_replicator_test.cpp b/src/mongo/db/repl/data_replicator_test.cpp
@@ -83,8 +83,7 @@ class SyncSourceSelectorMock : public SyncSourceSelector {
         _blacklistedSource = host;
     }
     bool shouldChangeSyncSource(const HostAndPort& currentSource,
-                                const OpTime& sourcesOpTime,
-                                bool syncSourceHasSyncSource) override {
+                                const rpc::ReplSetMetadata& metadata) override {
         return false;
     }
     SyncSourceResolverResponse selectSyncSource(OperationContext* txn,
@@ -126,10 +125,8 @@ class DataReplicatorTest : public ReplicationExecutorTest, public SyncSourceSele
         _syncSourceSelector->blacklistSyncSource(host, until);
     }
     bool shouldChangeSyncSource(const HostAndPort& currentSource,
-                                const OpTime& sourcesOpTime,
-                                bool syncSourceHasSyncSource) override {
-        return _syncSourceSelector->shouldChangeSyncSource(
-            currentSource, sourcesOpTime, syncSourceHasSyncSource);
+                                const rpc::ReplSetMetadata& metadata) override {
+        return _syncSourceSelector->shouldChangeSyncSource(currentSource, metadata);
     }
     SyncSourceResolverResponse selectSyncSource(OperationContext* txn,
                                                 const OpTime& lastOpTimeFetched) override {
@@ -718,8 +715,7 @@ class TestSyncSourceSelector2 : public SyncSourceSelector {
         _blacklistedSource = host;
     }
     bool shouldChangeSyncSource(const HostAndPort& currentSource,
-                                const OpTime& sourcesOpTime,
-                                bool syncSourceHasSyncSource) override {
+                                const rpc::ReplSetMetadata& metadata) override {
         return false;
     }
     SyncSourceResolverResponse selectSyncSource(OperationContext* txn,
@@ -850,8 +846,7 @@ class ShutdownExecutorSyncSourceSelector : public SyncSourceSelector {
     }
     void blacklistSyncSource(const HostAndPort& host, Date_t until) override {}
     bool shouldChangeSyncSource(const HostAndPort& currentSource,
-                                const OpTime& sourcesOpTime,
-                                bool syncSourceHasSyncSource) override {
+                                const rpc::ReplSetMetadata& metadata) override {
         return false;
     }
     SyncSourceResolverResponse selectSyncSource(OperationContext* txn,
diff --git a/src/mongo/db/repl/oplog_fetcher.cpp b/src/mongo/db/repl/oplog_fetcher.cpp
@@ -50,10 +50,8 @@ namespace {
  * Calculates await data timeout based on the current replica set configuration.
  */
 Milliseconds calculateAwaitDataTimeout(const ReplicaSetConfig& config) {
-    // Under protocol version 1, make the awaitData timeout (maxTimeMS) dependent
-    // on the election
-    // timeout. This enables the sync source to communicate liveness of the
-    // primary to secondaries.
+    // Under protocol version 1, make the awaitData timeout (maxTimeMS) dependent on the election
+    // timeout. This enables the sync source to communicate liveness of the primary to secondaries.
     // Under protocol version 0, use a default timeout of 2 seconds for awaitData.
     if (config.getProtocolVersion() == 1LL) {
         return config.getElectionTimeoutPeriod() / 2;
@@ -111,19 +109,15 @@ StatusWith<BSONObj> makeMetadataObject(bool isV1ElectionProtocol) {
 
 /**
  * Checks the first batch of results from query.
- * 'documents' are the first batch of results returned from tailing the remote
- * oplog.
- * 'lastFetched' optime and hash should be consistent with the predicate in the
- * query.
+ * 'documents' are the first batch of results returned from tailing the remote oplog.
+ * 'lastFetched' optime and hash should be consistent with the predicate in the query.
  * Returns RemoteOplogStale if the oplog query has no results.
- * Returns OplogStartMissing if we cannot find the optime of the last fetched
- * operation in
+ * Returns OplogStartMissing if we cannot find the optime of the last fetched operation in
  * the remote oplog.
  */
 Status checkRemoteOplogStart(const Fetcher::Documents& documents, OpTimeWithHash lastFetched) {
     if (documents.empty()) {
-        // The GTE query from upstream returns nothing, so we're ahead of the
-        // upstream.
+        // The GTE query from upstream returns nothing, so we're ahead of the upstream.
         return Status(ErrorCodes::RemoteOplogStale,
                       str::stream() << "We are ahead of the sync source. Our last op time fetched: "
                                     << lastFetched.opTime.toString());
@@ -176,8 +170,7 @@ StatusWith<OplogFetcher::DocumentsInfo> OplogFetcher::validateDocuments(
         info.networkDocumentBytes += doc.objsize();
         ++info.networkDocumentCount;
 
-        // If this is the first response (to the $gte query) then we already applied
-        // the first doc.
+        // If this is the first response (to the $gte query) then we already applied the first doc.
         if (first && info.networkDocumentCount == 1U) {
             continue;
         }
@@ -208,8 +201,7 @@ StatusWith<OplogFetcher::DocumentsInfo> OplogFetcher::validateDocuments(
     info.toApplyDocumentCount = documents.size();
     info.toApplyDocumentBytes = info.networkDocumentBytes;
     if (first) {
-        // The count is one less since the first document found was already applied
-        // ($gte $ts query)
+        // The count is one less since the first document found was already applied ($gte $ts query)
         // and we will not apply it again.
         --info.toApplyDocumentCount;
         auto alreadyAppliedDocument = documents.cbegin();
@@ -302,11 +294,9 @@ void OplogFetcher::_callback(const Fetcher::QueryResponseStatus& result,
     }
 
     const auto& queryResponse = result.getValue();
-    OpTime sourcesLastOpTime;
-    bool syncSourceHasSyncSource = false;
+    rpc::ReplSetMetadata metadata;
 
-    // Forward metadata (containing liveness information) to data replicator
-    // external state.
+    // Forward metadata (containing liveness information) to data replicator external state.
     bool receivedMetadata =
         queryResponse.otherFields.metadata.hasElement(rpc::kReplSetMetadataFieldName);
     if (receivedMetadata) {
@@ -318,10 +308,8 @@ void OplogFetcher::_callback(const Fetcher::QueryResponseStatus& result,
             _onShutdown(metadataResult.getStatus());
             return;
         }
-        auto metadata = metadataResult.getValue();
+        metadata = metadataResult.getValue();
         _dataReplicatorExternalState->processMetadata(metadata);
-        sourcesLastOpTime = metadata.getLastOpVisible();
-        syncSourceHasSyncSource = metadata.getSyncSourceIndex() != -1;
     }
 
     const auto& documents = queryResponse.documents;
@@ -337,8 +325,7 @@ void OplogFetcher::_callback(const Fetcher::QueryResponseStatus& result,
 
     auto opTimeWithHash = getLastOpTimeWithHashFetched();
 
-    // Check start of remote oplog and, if necessary, stop fetcher to execute
-    // rollback.
+    // Check start of remote oplog and, if necessary, stop fetcher to execute rollback.
     if (queryResponse.first) {
         auto status = checkRemoteOplogStart(documents, opTimeWithHash);
         if (!status.isOK()) {
@@ -347,8 +334,7 @@ void OplogFetcher::_callback(const Fetcher::QueryResponseStatus& result,
             return;
         }
 
-        // If this is the first batch and no rollback is needed, skip the first
-        // document.
+        // If this is the first batch and no rollback is needed, skip the first document.
         firstDocToApply++;
     }
 
@@ -373,14 +359,15 @@ void OplogFetcher::_callback(const Fetcher::QueryResponseStatus& result,
         _lastFetched = opTimeWithHash;
     }
 
-    if (_dataReplicatorExternalState->shouldStopFetching(
-            _fetcher.getSource(), sourcesLastOpTime, syncSourceHasSyncSource)) {
+    if (_dataReplicatorExternalState->shouldStopFetching(_fetcher.getSource(), metadata)) {
         _onShutdown(Status(ErrorCodes::InvalidSyncSource,
                            str::stream() << "sync source " << _fetcher.getSource().toString()
                                          << " (last optime: "
-                                         << sourcesLastOpTime.toString()
-                                         << "; has sync source: "
-                                         << syncSourceHasSyncSource
+                                         << metadata.getLastOpVisible().toString()
+                                         << "; sync source index: "
+                                         << metadata.getSyncSourceIndex()
+                                         << "; primary index: "
+                                         << metadata.getPrimaryIndex()
                                          << ") is no longer valid"),
                     opTimeWithHash);
         return;
@@ -407,5 +394,6 @@ void OplogFetcher::_onShutdown(Status status, OpTimeWithHash opTimeWithHash) {
     _onShutdownCallbackFn(status, opTimeWithHash);
 }
 
+
 }  // namespace repl
 }  // namespace mongo
diff --git a/src/mongo/db/repl/replication_coordinator_impl.cpp b/src/mongo/db/repl/replication_coordinator_impl.cpp
@@ -2901,14 +2901,10 @@ void ReplicationCoordinatorImpl::resetLastOpTimesFromOplog(OperationContext* txn
 }
 
 bool ReplicationCoordinatorImpl::shouldChangeSyncSource(const HostAndPort& currentSource,
-                                                        const OpTime& syncSourceLastOpTime,
-                                                        bool syncSourceHasSyncSource) {
+                                                        const rpc::ReplSetMetadata& metadata) {
     LockGuard topoLock(_topoMutex);
-    return _topCoord->shouldChangeSyncSource(currentSource,
-                                             getMyLastAppliedOpTime(),
-                                             syncSourceLastOpTime,
-                                             syncSourceHasSyncSource,
-                                             _replExecutor.now());
+    return _topCoord->shouldChangeSyncSource(
+        currentSource, getMyLastAppliedOpTime(), metadata, _replExecutor.now());
 }
 
 SyncSourceResolverResponse ReplicationCoordinatorImpl::selectSyncSource(
diff --git a/src/mongo/db/repl/replication_coordinator_impl.h b/src/mongo/db/repl/replication_coordinator_impl.h
@@ -275,8 +275,7 @@ class ReplicationCoordinatorImpl : public ReplicationCoordinator, public KillOpL
     virtual void resetLastOpTimesFromOplog(OperationContext* txn) override;
 
     virtual bool shouldChangeSyncSource(const HostAndPort& currentSource,
-                                        const OpTime& syncSourceLastOpTime,
-                                        bool syncSourceHasSyncSource) override;
+                                        const rpc::ReplSetMetadata& metadata) override;
 
     virtual SyncSourceResolverResponse selectSyncSource(OperationContext* txn,
                                                         const OpTime& lastOpTimeFetched) override;
diff --git a/src/mongo/db/repl/replication_coordinator_mock.cpp b/src/mongo/db/repl/replication_coordinator_mock.cpp
@@ -357,8 +357,7 @@ void ReplicationCoordinatorMock::resetLastOpTimesFromOplog(OperationContext* txn
 }
 
 bool ReplicationCoordinatorMock::shouldChangeSyncSource(const HostAndPort& currentSource,
-                                                        const OpTime& syncSourceLastOpTime,
-                                                        bool syncSourceHasSyncSource) {
+                                                        const rpc::ReplSetMetadata& metadata) {
     invariant(false);
 }
 
diff --git a/src/mongo/db/repl/replication_coordinator_mock.h b/src/mongo/db/repl/replication_coordinator_mock.h
@@ -208,8 +208,7 @@ class ReplicationCoordinatorMock : public ReplicationCoordinator {
     virtual void resetLastOpTimesFromOplog(OperationContext* txn);
 
     virtual bool shouldChangeSyncSource(const HostAndPort& currentSource,
-                                        const OpTime& syncSourceLastOpTime,
-                                        bool syncSourceHasSyncSource);
+                                        const rpc::ReplSetMetadata& metadata);
 
     virtual OpTime getLastCommittedOpTime() const;
 
diff --git a/src/mongo/db/repl/sync_source_selector.h b/src/mongo/db/repl/sync_source_selector.h
@@ -37,6 +37,10 @@ namespace mongo {
 class OperationContext;
 class Timestamp;
 
+namespace rpc {
+class ReplSetMetadata;
+}
+
 namespace repl {
 
 class OpTime;
@@ -69,17 +73,16 @@ class SyncSourceSelector {
 
     /**
      * Determines if a new sync source should be chosen, if a better candidate sync source is
-     * available.  If the current sync source's last optime ("syncSourceLastOpTime" under
+     * available.  If the current sync source's last optime (visibleOpTime of metadata under
      * protocolVersion 1, but pulled from the MemberHeartbeatData in protocolVersion 0) is more than
      * _maxSyncSourceLagSecs behind any syncable source, this function returns true. If we are
      * running in ProtocolVersion 1, our current sync source is not primary, has no sync source
-     * ("syncSourceHasSyncSource" is false), and only has data up to "myLastOpTime", returns true.
+     * and only has data up to "myLastOpTime", returns true.
      *
      * "now" is used to skip over currently blacklisted sync sources.
      */
     virtual bool shouldChangeSyncSource(const HostAndPort& currentSource,
-                                        const OpTime& syncSourceLastOpTime,
-                                        bool syncSourceHasSyncSource) = 0;
+                                        const rpc::ReplSetMetadata& metadata) = 0;
 
     /**
      * Returns a SyncSourceResolverResponse containing the syncSource or a new MinValid boundry as
diff --git a/src/mongo/db/repl/topology_coordinator.h b/src/mongo/db/repl/topology_coordinator.h
@@ -165,8 +165,7 @@ class TopologyCoordinator {
      */
     virtual bool shouldChangeSyncSource(const HostAndPort& currentSource,
                                         const OpTime& myLastOpTime,
-                                        const OpTime& syncSourceLastOpTime,
-                                        bool syncSourceHasSyncSource,
+                                        const rpc::ReplSetMetadata& metadata,
                                         Date_t now) const = 0;
 
     /**
diff --git a/src/mongo/db/repl/topology_coordinator_impl.cpp b/src/mongo/db/repl/topology_coordinator_impl.cpp
diff --git a/src/mongo/db/repl/topology_coordinator_impl.h b/src/mongo/db/repl/topology_coordinator_impl.h
diff --git a/src/mongo/db/repl/topology_coordinator_impl_test.cpp b/src/mongo/db/repl/topology_coordinator_impl_test.cpp
diff --git a/src/mongo/db/repl/topology_coordinator_impl_v1_test.cpp b/src/mongo/db/repl/topology_coordinator_impl_v1_test.cpp
diff --git a/src/mongo/db/repl/update_position_args.h b/src/mongo/db/repl/update_position_args.h
diff --git a/src/mongo/rpc/metadata/repl_set_metadata.h b/src/mongo/rpc/metadata/repl_set_metadata.h

Original file line number	Diff line number	Diff line change
`@@ -47,11 +47,10 @@ void DataReplicatorExternalStateMock::processMetadata(const rpc::ReplSetMetadata`
`47`	`47`	`}`
`48`	`48`
`49`	`49`	`bool DataReplicatorExternalStateMock::shouldStopFetching(const HostAndPort& source,`
`50`		`- const OpTime& sourceOpTime,`
`51`		`- bool sourceHasSyncSource) {`
	`50`	`+ const rpc::ReplSetMetadata& metadata) {`
`52`	`51`	`lastSyncSourceChecked = source;`
`53`		`- syncSourceLastOpTime = sourceOpTime;`
`54`		`- syncSourceHasSyncSource = sourceHasSyncSource;`
	`52`	`+ syncSourceLastOpTime = metadata.getLastOpVisible();`
	`53`	`+ syncSourceHasSyncSource = metadata.getSyncSourceIndex() != -1;`
`55`	`54`	`return shouldStopFetchingResult;`
`56`	`55`	`}`
`57`	`56`
Original file line number	Diff line number	Diff line change
`@@ -83,8 +83,7 @@ class SyncSourceSelectorMock : public SyncSourceSelector {`
`83`	`83`	`_blacklistedSource = host;`
`84`	`84`	`}`
`85`	`85`	`bool shouldChangeSyncSource(const HostAndPort& currentSource,`
`86`		`- const OpTime& sourcesOpTime,`
`87`		`- bool syncSourceHasSyncSource) override {`
	`86`	`+ const rpc::ReplSetMetadata& metadata) override {`
`88`	`87`	`return false;`
`89`	`88`	`}`
`90`	`89`	`SyncSourceResolverResponse selectSyncSource(OperationContext* txn,`
`@@ -126,10 +125,8 @@ class DataReplicatorTest : public ReplicationExecutorTest, public SyncSourceSele`
`126`	`125`	`_syncSourceSelector->blacklistSyncSource(host, until);`
`127`	`126`	`}`
`128`	`127`	`bool shouldChangeSyncSource(const HostAndPort& currentSource,`
`129`		`- const OpTime& sourcesOpTime,`
`130`		`- bool syncSourceHasSyncSource) override {`
`131`		`- return _syncSourceSelector->shouldChangeSyncSource(`
`132`		`- currentSource, sourcesOpTime, syncSourceHasSyncSource);`
	`128`	`+ const rpc::ReplSetMetadata& metadata) override {`
	`129`	`+ return _syncSourceSelector->shouldChangeSyncSource(currentSource, metadata);`
`133`	`130`	`}`
`134`	`131`	`SyncSourceResolverResponse selectSyncSource(OperationContext* txn,`
`135`	`132`	`const OpTime& lastOpTimeFetched) override {`
`@@ -718,8 +715,7 @@ class TestSyncSourceSelector2 : public SyncSourceSelector {`
`718`	`715`	`_blacklistedSource = host;`
`719`	`716`	`}`
`720`	`717`	`bool shouldChangeSyncSource(const HostAndPort& currentSource,`
`721`		`- const OpTime& sourcesOpTime,`
`722`		`- bool syncSourceHasSyncSource) override {`
	`718`	`+ const rpc::ReplSetMetadata& metadata) override {`
`723`	`719`	`return false;`
`724`	`720`	`}`
`725`	`721`	`SyncSourceResolverResponse selectSyncSource(OperationContext* txn,`
`@@ -850,8 +846,7 @@ class ShutdownExecutorSyncSourceSelector : public SyncSourceSelector {`
`850`	`846`	`}`
`851`	`847`	`void blacklistSyncSource(const HostAndPort& host, Date_t until) override {}`
`852`	`848`	`bool shouldChangeSyncSource(const HostAndPort& currentSource,`
`853`		`- const OpTime& sourcesOpTime,`
`854`		`- bool syncSourceHasSyncSource) override {`
	`849`	`+ const rpc::ReplSetMetadata& metadata) override {`
`855`	`850`	`return false;`
`856`	`851`	`}`
`857`	`852`	`SyncSourceResolverResponse selectSyncSource(OperationContext* txn,`