summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--src/mongo/db/repl/data_replicator.cpp5
-rw-r--r--src/mongo/db/repl/data_replicator.h2
-rw-r--r--src/mongo/db/repl/data_replicator_test.cpp2
-rw-r--r--src/mongo/db/repl/replication_coordinator_impl.cpp4
-rw-r--r--src/mongo/db/repl/reporter.cpp10
-rw-r--r--src/mongo/db/repl/reporter.h4
-rw-r--r--src/mongo/db/repl/reporter_test.cpp51
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());
View Lorry Controller Status