SERVER-18043 migrated Reporter to use new replSetUpdatePosition command

7 files changed, 45 insertions, 33 deletions

diff --git a/src/mongo/db/repl/data_replicator.cpp b/src/mongo/db/repl/data_replicator.cpp
index aed481df509..3b8580fbdaa 100644
--- a/src/mongo/db/repl/data_replicator.cpp
+++ b/src/mongo/db/repl/data_replicator.cpp
@@ -525,7 +525,7 @@ DataReplicator::DataReplicator(DataReplicatorOptions opts, ReplicationExecutor*
     uassert(ErrorCodes::BadValue, "invalid rollback function", _opts.rollbackFn);
     uassert(ErrorCodes::BadValue,
             "invalid replSetUpdatePosition command object creation function",
-            _opts.prepareOldReplSetUpdatePositionCommandFn);
+            _opts.prepareReplSetUpdatePositionCommandFn);
     uassert(ErrorCodes::BadValue, "invalid getMyLastOptime function", _opts.getMyLastOptime);
     uassert(ErrorCodes::BadValue, "invalid setMyLastOptime function", _opts.setMyLastOptime);
     uassert(ErrorCodes::BadValue, "invalid setFollowerMode function", _opts.setFollowerMode);
@@ -1018,9 +1018,8 @@ void DataReplicator::_doNextActions_Steady_inlock() {
     }
 
     if (!_reporterPaused && (!_reporter || !_reporter->getStatus().isOK())) {
-        // TODO get reporter in good shape
         _reporter.reset(
-            new Reporter(_exec, _opts.prepareOldReplSetUpdatePositionCommandFn, _syncSource));
+            new Reporter(_exec, _opts.prepareReplSetUpdatePositionCommandFn, _syncSource));
     }
 }
 
diff --git a/src/mongo/db/repl/data_replicator.h b/src/mongo/db/repl/data_replicator.h
index 992aeb71d58..6d7c45d2286 100644
--- a/src/mongo/db/repl/data_replicator.h
+++ b/src/mongo/db/repl/data_replicator.h
@@ -127,7 +127,7 @@ struct DataReplicatorOptions {
 
     Applier::ApplyOperationFn applierFn;
     RollbackFn rollbackFn;
-    Reporter::PrepareReplSetUpdatePositionCommandFn prepareOldReplSetUpdatePositionCommandFn;
+    Reporter::PrepareReplSetUpdatePositionCommandFn prepareReplSetUpdatePositionCommandFn;
     GetMyLastOptimeFn getMyLastOptime;
     SetMyLastOptimeFn setMyLastOptime;
     SetFollowerModeFn setFollowerMode;
diff --git a/src/mongo/db/repl/data_replicator_test.cpp b/src/mongo/db/repl/data_replicator_test.cpp
index 046e73956df..adff8d96782 100644
--- a/src/mongo/db/repl/data_replicator_test.cpp
+++ b/src/mongo/db/repl/data_replicator_test.cpp
@@ -179,7 +179,7 @@ protected:
             return _rollbackFn(txn, lastOpTimeWritten, syncSource);
         };
 
-        options.prepareOldReplSetUpdatePositionCommandFn =
+        options.prepareReplSetUpdatePositionCommandFn =
             []() -> StatusWith<BSONObj> { return BSON("replSetUpdatePosition" << 1); };
         options.getMyLastOptime = [this]() { return _myLastOpTime; };
         options.setMyLastOptime = [this](const OpTime& opTime) { _setMyLastOptime(opTime); };
diff --git a/src/mongo/db/repl/replication_coordinator_impl.cpp b/src/mongo/db/repl/replication_coordinator_impl.cpp
index 275f5948ca7..6013c29dd25 100644
--- a/src/mongo/db/repl/replication_coordinator_impl.cpp
+++ b/src/mongo/db/repl/replication_coordinator_impl.cpp
@@ -163,9 +163,9 @@ DataReplicatorOptions createDataReplicatorOptions(ReplicationCoordinator* replCo
     options.applierFn = [](OperationContext*, const BSONObj&) -> Status { return Status::OK(); };
     options.rollbackFn =
         [](OperationContext*, const OpTime&, const HostAndPort&) { return Status::OK(); };
-    options.prepareOldReplSetUpdatePositionCommandFn = [replCoord]() -> StatusWith<BSONObj> {
+    options.prepareReplSetUpdatePositionCommandFn = [replCoord]() -> StatusWith<BSONObj> {
         BSONObjBuilder bob;
-        if (replCoord->prepareOldReplSetUpdatePositionCommand(&bob)) {
+        if (replCoord->prepareReplSetUpdatePositionCommand(&bob)) {
             return bob.obj();
         }
         return Status(ErrorCodes::OperationFailed,
diff --git a/src/mongo/db/repl/reporter.cpp b/src/mongo/db/repl/reporter.cpp
index 2084daef505..4a0160d6211 100644
--- a/src/mongo/db/repl/reporter.cpp
+++ b/src/mongo/db/repl/reporter.cpp
@@ -39,10 +39,10 @@ namespace repl {
 using executor::RemoteCommandRequest;
 
 Reporter::Reporter(executor::TaskExecutor* executor,
-                   PrepareReplSetUpdatePositionCommandFn prepareOldReplSetUpdatePositionCommandFn,
+                   PrepareReplSetUpdatePositionCommandFn prepareReplSetUpdatePositionCommandFn,
                    const HostAndPort& target)
     : _executor(executor),
-      _prepareOldReplSetUpdatePositionCommandFn(prepareOldReplSetUpdatePositionCommandFn),
+      _prepareReplSetUpdatePositionCommandFn(prepareReplSetUpdatePositionCommandFn),
       _target(target),
       _status(Status::OK()),
       _willRunAgain(false),
@@ -50,7 +50,7 @@ Reporter::Reporter(executor::TaskExecutor* executor,
     uassert(ErrorCodes::BadValue, "null task executor", executor);
     uassert(ErrorCodes::BadValue,
             "null function to create replSetUpdatePosition command object",
-            prepareOldReplSetUpdatePositionCommandFn);
+            prepareReplSetUpdatePositionCommandFn);
     uassert(ErrorCodes::BadValue, "target name cannot be empty", !target.empty());
 }
 
@@ -103,11 +103,11 @@ Status Reporter::_schedule_inlock() {
 
     LOG(2) << "Reporter scheduling report to : " << _target;
 
-    auto prepareResult = _prepareOldReplSetUpdatePositionCommandFn();
+    auto prepareResult = _prepareReplSetUpdatePositionCommandFn();
 
     if (!prepareResult.isOK()) {
         // Returning NodeNotFound because currently this is the only way
-        // prepareOldReplSetUpdatePositionCommand() can fail in production.
+        // prepareReplSetUpdatePositionCommand() can fail in production.
         return Status(ErrorCodes::NodeNotFound,
                       "Reporter failed to create replSetUpdatePositionCommand command.");
     }
diff --git a/src/mongo/db/repl/reporter.h b/src/mongo/db/repl/reporter.h
index ad1b3be953f..e9cc9c06196 100644
--- a/src/mongo/db/repl/reporter.h
+++ b/src/mongo/db/repl/reporter.h
@@ -50,7 +50,7 @@ public:
     using PrepareReplSetUpdatePositionCommandFn = stdx::function<StatusWith<BSONObj>()>;
 
     Reporter(executor::TaskExecutor* executor,
-             PrepareReplSetUpdatePositionCommandFn prepareOldReplSetUpdatePositionCommandFn,
+             PrepareReplSetUpdatePositionCommandFn prepareReplSetUpdatePositionCommandFn,
              const HostAndPort& target);
     virtual ~Reporter();
 
@@ -105,7 +105,7 @@ private:
     executor::TaskExecutor* _executor;
 
     // Prepares update command object.
-    PrepareReplSetUpdatePositionCommandFn _prepareOldReplSetUpdatePositionCommandFn;
+    PrepareReplSetUpdatePositionCommandFn _prepareReplSetUpdatePositionCommandFn;
 
     // Host to whom the Reporter sends updates.
     HostAndPort _target;
diff --git a/src/mongo/db/repl/reporter_test.cpp b/src/mongo/db/repl/reporter_test.cpp
index eeb2527cdb0..448e70472b9 100644
--- a/src/mongo/db/repl/reporter_test.cpp
+++ b/src/mongo/db/repl/reporter_test.cpp
@@ -28,6 +28,7 @@
 
 #include "mongo/platform/basic.h"
 
+#include "mongo/db/repl/optime.h"
 #include "mongo/db/repl/reporter.h"
 #include "mongo/executor/thread_pool_task_executor_test_fixture.h"
 #include "mongo/executor/network_interface_mock.h"
@@ -44,31 +45,43 @@ using executor::RemoteCommandResponse;
 
 class MockProgressManager {
 public:
-    void updateMap(int memberId, const Timestamp& ts) {
-        progressMap[memberId] = ts;
+    void updateMap(int memberId, const OpTime& lastDurableOpTime, const OpTime& lastAppliedOpTime) {
+        progressMap[memberId] = ProgressInfo(lastDurableOpTime, lastAppliedOpTime);
     }
 
     void setResult(bool newResult) {
         _result = newResult;
     }
 
-    bool prepareOldReplSetUpdatePositionCommand(BSONObjBuilder* cmdBuilder) {
+    bool prepareReplSetUpdatePositionCommand(BSONObjBuilder* cmdBuilder) {
         if (!_result) {
             return _result;
         }
         cmdBuilder->append("replSetUpdatePosition", 1);
         BSONArrayBuilder arrayBuilder(cmdBuilder->subarrayStart("optimes"));
-        for (auto itr = progressMap.begin(); itr != progressMap.end(); ++itr) {
+        for (auto&& itr : progressMap) {
             BSONObjBuilder entry(arrayBuilder.subobjStart());
-            entry.append("optime", itr->second);
-            entry.append("memberId", itr->first);
+            itr.second.lastDurableOpTime.append(&entry, "durableOpTime");
+            itr.second.lastAppliedOpTime.append(&entry, "appliedOpTime");
+            entry.append("memberId", itr.first);
             entry.append("cfgver", 1);
         }
         return true;
     }
 
 private:
-    std::map<int, Timestamp> progressMap;
+    struct ProgressInfo {
+        ProgressInfo() = default;
+        ProgressInfo(const OpTime& lastDurableOpTime, const OpTime& lastAppliedOpTime)
+            : lastDurableOpTime(lastDurableOpTime), lastAppliedOpTime(lastAppliedOpTime) {}
+
+        // Our last known OpTime that this slave has applied and journaled to.
+        OpTime lastDurableOpTime;
+        // Our last known OpTime that this slave has applied, whether journaled or unjournaled.
+        OpTime lastAppliedOpTime;
+    };
+
+    std::map<int, ProgressInfo> progressMap;
     bool _result = true;
 };
 
@@ -84,7 +97,7 @@ protected:
 
     std::unique_ptr<Reporter> reporter;
     std::unique_ptr<MockProgressManager> posUpdater;
-    Reporter::PrepareReplSetUpdatePositionCommandFn prepareOldReplSetUpdatePositionCommandFn;
+    Reporter::PrepareReplSetUpdatePositionCommandFn prepareReplSetUpdatePositionCommandFn;
 };
 
 ReporterTest::ReporterTest() {}
@@ -92,16 +105,16 @@ ReporterTest::ReporterTest() {}
 void ReporterTest::setUp() {
     executor::ThreadPoolExecutorTest::setUp();
     posUpdater.reset(new MockProgressManager());
-    prepareOldReplSetUpdatePositionCommandFn = [this]() -> StatusWith<BSONObj> {
+    prepareReplSetUpdatePositionCommandFn = [this]() -> StatusWith<BSONObj> {
         BSONObjBuilder bob;
-        if (posUpdater->prepareOldReplSetUpdatePositionCommand(&bob)) {
+        if (posUpdater->prepareReplSetUpdatePositionCommand(&bob)) {
             return bob.obj();
         }
         return Status(ErrorCodes::OperationFailed,
                       "unable to prepare replSetUpdatePosition command object");
     };
     reporter.reset(new Reporter(&getExecutor(),
-                                [this]() { return prepareOldReplSetUpdatePositionCommandFn(); },
+                                [this]() { return prepareReplSetUpdatePositionCommandFn(); },
                                 HostAndPort("h1")));
     launchExecutorThread();
 }
@@ -137,11 +150,11 @@ TEST_F(ReporterTest, InvalidConstruction) {
                   UserException);
 
     // null TaskExecutor
-    ASSERT_THROWS(Reporter(nullptr, prepareOldReplSetUpdatePositionCommandFn, HostAndPort("h1")),
+    ASSERT_THROWS(Reporter(nullptr, prepareReplSetUpdatePositionCommandFn, HostAndPort("h1")),
                   UserException);
 
     // empty HostAndPort
-    ASSERT_THROWS(Reporter(&getExecutor(), prepareOldReplSetUpdatePositionCommandFn, HostAndPort()),
+    ASSERT_THROWS(Reporter(&getExecutor(), prepareReplSetUpdatePositionCommandFn, HostAndPort()),
                   UserException);
 }
 
@@ -165,7 +178,7 @@ TEST_F(ReporterTest, ShutdownBeforeSchedule) {
 
 // If an error is returned, it should be recorded in the Reporter and be returned when triggered
 TEST_F(ReporterTest, ErrorsAreStoredInTheReporter) {
-    posUpdater->updateMap(0, Timestamp(3, 0));
+    posUpdater->updateMap(0, OpTime({3, 0}, 1), OpTime({3, 0}, 1));
     ASSERT_OK(reporter->trigger());
     ASSERT_TRUE(reporter->isActive());
 
@@ -183,7 +196,7 @@ TEST_F(ReporterTest, ErrorsAreStoredInTheReporter) {
 
 // If an error is returned, it should be recorded in the Reporter and not run again.
 TEST_F(ReporterTest, ErrorsStopTheReporter) {
-    posUpdater->updateMap(0, Timestamp(3, 0));
+    posUpdater->updateMap(0, OpTime({3, 0}, 1), OpTime({3, 0}, 1));
     ASSERT_OK(reporter->trigger());
     ASSERT_TRUE(reporter->isActive());
     ASSERT_FALSE(reporter->willRunAgain());
@@ -206,7 +219,7 @@ TEST_F(ReporterTest, ErrorsStopTheReporter) {
 // Schedule while we are already scheduled, it should set willRunAgain, then automatically
 // schedule itself after finishing.
 TEST_F(ReporterTest, DoubleScheduleShouldCauseRescheduleImmediatelyAfterRespondedTo) {
-    posUpdater->updateMap(0, Timestamp(3, 0));
+    posUpdater->updateMap(0, OpTime({3, 0}, 1), OpTime({3, 0}, 1));
     ASSERT_OK(reporter->trigger());
     ASSERT_TRUE(reporter->isActive());
     ASSERT_FALSE(reporter->willRunAgain());
@@ -238,7 +251,7 @@ TEST_F(ReporterTest, DoubleScheduleShouldCauseRescheduleImmediatelyAfterResponde
 // then automatically schedule itself after finishing, but not a third time since the latter
 // two will contain the same batch of updates.
 TEST_F(ReporterTest, TripleScheduleShouldCauseRescheduleImmediatelyAfterRespondedToOnlyOnce) {
-    posUpdater->updateMap(0, Timestamp(3, 0));
+    posUpdater->updateMap(0, OpTime({3, 0}, 1), OpTime({3, 0}, 1));
     ASSERT_OK(reporter->trigger());
     ASSERT_TRUE(reporter->isActive());
     ASSERT_FALSE(reporter->willRunAgain());
@@ -270,7 +283,7 @@ TEST_F(ReporterTest, TripleScheduleShouldCauseRescheduleImmediatelyAfterResponde
 }
 
 TEST_F(ReporterTest, CancelWhileScheduled) {
-    posUpdater->updateMap(0, Timestamp(3, 0));
+    posUpdater->updateMap(0, OpTime({3, 0}, 1), OpTime({3, 0}, 1));
     ASSERT_OK(reporter->trigger());
     ASSERT_TRUE(reporter->isActive());
     ASSERT_FALSE(reporter->willRunAgain());
@@ -294,7 +307,7 @@ TEST_F(ReporterTest, CancelWhileScheduled) {
 }
 
 TEST_F(ReporterTest, CancelAfterFirstReturns) {
-    posUpdater->updateMap(0, Timestamp(3, 0));
+    posUpdater->updateMap(0, OpTime({3, 0}, 1), OpTime({3, 0}, 1));
     ASSERT_OK(reporter->trigger());
     ASSERT_TRUE(reporter->isActive());
     ASSERT_FALSE(reporter->willRunAgain());