You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mesos.apache.org by bm...@apache.org on 2014/05/28 00:41:13 UTC
[1/3] git commit: Updated the master to handle status update
acknowledgement messages.
Repository: mesos
Updated Branches:
refs/heads/master 3dd88aa1c -> ac5059bbe
Updated the master to handle status update acknowledgement messages.
Review: https://reviews.apache.org/r/21838
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/e8d6781d
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/e8d6781d
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/e8d6781d
Branch: refs/heads/master
Commit: e8d6781d0345da2ac827f952267a897673efc8f8
Parents: 2986ac8
Author: Benjamin Mahler <bm...@twitter.com>
Authored: Thu May 22 14:49:49 2014 -0700
Committer: Benjamin Mahler <bm...@twitter.com>
Committed: Tue May 27 15:41:03 2014 -0700
----------------------------------------------------------------------
src/master/master.cpp | 85 +++++++++++++++++++++++++++
src/master/master.hpp | 11 ++++
src/tests/master_tests.cpp | 97 +++++++++++++++++++++++++++++++
src/tests/mesos.hpp | 2 +
src/tests/slave_tests.cpp | 124 ++++++++++++++++++++++++++++++++++++++++
5 files changed, 319 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/e8d6781d/src/master/master.cpp
----------------------------------------------------------------------
diff --git a/src/master/master.cpp b/src/master/master.cpp
index 21472e9..98647e5 100644
--- a/src/master/master.cpp
+++ b/src/master/master.cpp
@@ -464,6 +464,13 @@ void Master::initialize()
&KillTaskMessage::framework_id,
&KillTaskMessage::task_id);
+ install<StatusUpdateAcknowledgementMessage>(
+ &Master::statusUpdateAcknowledgement,
+ &StatusUpdateAcknowledgementMessage::slave_id,
+ &StatusUpdateAcknowledgementMessage::framework_id,
+ &StatusUpdateAcknowledgementMessage::task_id,
+ &StatusUpdateAcknowledgementMessage::uuid);
+
install<FrameworkToExecutorMessage>(
&Master::schedulerMessage,
&FrameworkToExecutorMessage::slave_id,
@@ -2080,6 +2087,84 @@ void Master::killTask(
}
+void Master::statusUpdateAcknowledgement(
+ const UPID& from,
+ const SlaveID& slaveId,
+ const FrameworkID& frameworkId,
+ const TaskID& taskId,
+ const string& uuid)
+{
+ // TODO(bmahler): Add message counter.
+
+ // TODO(bmahler): Consider adding a message validator abstraction
+ // for the master that takes care of all this boilerplate. Ideally
+ // by the time we process messages in the critical master code, we
+ // can assume that they are valid. This will become especially
+ // important as validation logic is moved out of the scheduler
+ // driver and into the master.
+
+ Framework* framework = getFramework(frameworkId);
+
+ if (framework == NULL) {
+ LOG(WARNING)
+ << "Ignoring status update acknowledgement message for task " << taskId
+ << " of framework " << frameworkId << " on slave " << slaveId
+ << " because the framework cannot be found";
+ // TODO(bmahler): Add dropped counter.
+ return;
+ }
+
+ if (from != framework->pid) {
+ LOG(WARNING)
+ << "Ignoring status update acknowledgement message for task " << taskId
+ << " of framework " << frameworkId << " on slave " << slaveId
+ << " from " << from << " because it is not from the registered framework "
+ << framework->pid;
+ // TODO(bmahler): Add dropped counter.
+ return;
+ }
+
+ Slave* slave = getSlave(slaveId);
+
+ if (slave == NULL) {
+ LOG(WARNING)
+ << "Cannot send status update acknowledgement message for task " << taskId
+ << " of framework " << frameworkId << " to slave " << slaveId
+ << " because slave is not activated";
+ // TODO(bmahler): Add dropped counter.
+ return;
+ }
+
+ if (slave->disconnected) {
+ LOG(WARNING)
+ << "Cannot send status update acknowledgement message for task " << taskId
+ << " of framework " << frameworkId << " to slave " << *slave
+ << " because slave is disconnected";
+ // TODO(bmahler): Add dropped counter.
+ return;
+ }
+
+ LOG(INFO) << "Forwarding status update acknowledgement "
+ << UUID::fromBytes(uuid) << " for task " << taskId
+ << " of framework " << frameworkId << " to slave " << *slave;
+
+ // TODO(bmahler): Once we store terminal unacknowledged updates in
+ // the master per MESOS-1410, this is where we'll find the
+ // unacknowledged task and remove it if present.
+ // Also, be sure to confirm Master::reconcile is still correct!
+
+ StatusUpdateAcknowledgementMessage message;
+ message.mutable_slave_id()->CopyFrom(slaveId);
+ message.mutable_framework_id()->CopyFrom(frameworkId);
+ message.mutable_task_id()->CopyFrom(taskId);
+ message.set_uuid(uuid);
+
+ send(slave->pid, message);
+
+ // TODO(bmahler): Add valid acknowledgement counter.
+}
+
+
void Master::schedulerMessage(
const UPID& from,
const SlaveID& slaveId,
http://git-wip-us.apache.org/repos/asf/mesos/blob/e8d6781d/src/master/master.hpp
----------------------------------------------------------------------
diff --git a/src/master/master.hpp b/src/master/master.hpp
index f0490ec..45d868d 100644
--- a/src/master/master.hpp
+++ b/src/master/master.hpp
@@ -137,6 +137,14 @@ public:
const process::UPID& from,
const FrameworkID& frameworkId,
const TaskID& taskId);
+
+ void statusUpdateAcknowledgement(
+ const process::UPID& from,
+ const SlaveID& slaveId,
+ const FrameworkID& frameworkId,
+ const TaskID& taskId,
+ const std::string& uuid);
+
void schedulerMessage(
const process::UPID& from,
const SlaveID& slaveId,
@@ -156,15 +164,18 @@ public:
void unregisterSlave(
const SlaveID& slaveId);
+
void statusUpdate(
const StatusUpdate& update,
const process::UPID& pid);
+
void exitedExecutor(
const process::UPID& from,
const SlaveID& slaveId,
const FrameworkID& frameworkId,
const ExecutorID& executorId,
int32_t status);
+
void shutdownSlave(
const SlaveID& slaveId,
const std::string& message);
http://git-wip-us.apache.org/repos/asf/mesos/blob/e8d6781d/src/tests/master_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/master_tests.cpp b/src/tests/master_tests.cpp
index f607d89..56557a5 100644
--- a/src/tests/master_tests.cpp
+++ b/src/tests/master_tests.cpp
@@ -68,6 +68,7 @@ using process::Clock;
using process::Future;
using process::Owned;
using process::PID;
+using process::UPID;
using std::string;
using std::vector;
@@ -1386,6 +1387,102 @@ TEST_F(MasterTest, LaunchDuplicateOfferTest)
}
+// This test runs a task but intercepts the scheduler driver's
+// acknowledgement messages to the slave and instead sends them to
+// the master. This test is necessary to test that the
+// acknowledgement handling in the master is correct, but once the
+// driver sends these messages we should remove/update this test!
+TEST_F(MasterTest, StatusUpdateAcknowledgementsThroughMaster)
+{
+ Try<PID<Master> > master = StartMaster();
+ ASSERT_SOME(master);
+
+ MockExecutor exec(DEFAULT_EXECUTOR_ID);
+
+ Try<PID<Slave> > slave = StartSlave(&exec);
+ ASSERT_SOME(slave);
+
+ MockScheduler sched;
+ MesosSchedulerDriver schedDriver(
+ &sched, DEFAULT_FRAMEWORK_INFO, master.get(), DEFAULT_CREDENTIAL);
+
+ EXPECT_CALL(sched, registered(&schedDriver, _, _))
+ .Times(1);
+
+ Future<vector<Offer> > offers;
+ EXPECT_CALL(sched, resourceOffers(&schedDriver, _))
+ .WillOnce(FutureArg<1>(&offers))
+ .WillRepeatedly(Return()); // Ignore subsequent offers.
+
+ // We need to grab this message to get the scheduler's pid.
+ Future<process::Message> frameworkRegisteredMessage = FUTURE_MESSAGE(
+ Eq(FrameworkRegisteredMessage().GetTypeName()), master.get(), _);
+
+ schedDriver.start();
+
+ AWAIT_READY(frameworkRegisteredMessage);
+ const process::UPID schedulerPid = frameworkRegisteredMessage.get().to;
+
+ AWAIT_READY(offers);
+ EXPECT_NE(0u, offers.get().size());
+
+ TaskInfo task = createTask(offers.get()[0], "", DEFAULT_EXECUTOR_ID);
+
+ vector<TaskInfo> tasks;
+ tasks.push_back(task);
+
+ Future<ExecutorDriver*> execDriver;
+ EXPECT_CALL(exec, registered(_, _, _, _))
+ .WillOnce(FutureArg<0>(&execDriver));
+
+ EXPECT_CALL(exec, launchTask(_, _))
+ .WillOnce(SendStatusUpdateFromTask(TASK_RUNNING));
+
+ Future<TaskStatus> update;
+ EXPECT_CALL(sched, statusUpdate(&schedDriver, _))
+ .WillOnce(FutureArg<1>(&update));
+
+ // Pause the clock to prevent status update retries on the slave.
+ Clock::pause();
+
+ // Intercept the status update acknowledgement and send it to the
+ // master instead!
+ Future<StatusUpdateAcknowledgementMessage> acknowledgementMessage =
+ DROP_PROTOBUF(StatusUpdateAcknowledgementMessage(),
+ schedulerPid,
+ slave.get());
+
+ schedDriver.launchTasks(offers.get()[0].id(), tasks);
+
+ AWAIT_READY(update);
+ EXPECT_EQ(TASK_RUNNING, update.get().state());
+
+ AWAIT_READY(acknowledgementMessage);
+
+ Future<Nothing> _statusUpdateAcknowledgement =
+ FUTURE_DISPATCH(slave.get(), &Slave::_statusUpdateAcknowledgement);
+
+ // Send the acknowledgement to the master.
+ process::post(schedulerPid, master.get(), acknowledgementMessage.get());
+
+ // Ensure the slave receives and properly handles the ACK.
+ // Clock::settle() ensures that the slave successfully
+ // executes Slave::_statusUpdateAcknowledgement().
+ AWAIT_READY(_statusUpdateAcknowledgement);
+ Clock::settle();
+
+ Clock::resume();
+
+ EXPECT_CALL(exec, shutdown(_))
+ .Times(AtMost(1));
+
+ schedDriver.stop();
+ schedDriver.join();
+
+ Shutdown();
+}
+
+
TEST_F(MasterTest, MetricsInStatsEndpoint)
{
Try<PID<Master> > master = StartMaster();
http://git-wip-us.apache.org/repos/asf/mesos/blob/e8d6781d/src/tests/mesos.hpp
----------------------------------------------------------------------
diff --git a/src/tests/mesos.hpp b/src/tests/mesos.hpp
index c31199a..1b0f358 100644
--- a/src/tests/mesos.hpp
+++ b/src/tests/mesos.hpp
@@ -287,6 +287,8 @@ protected:
commandInfo; })
+// TODO(bmahler): Refactor this to make the distinction between
+// command tasks and executor tasks clearer.
inline TaskInfo createTask(
const Offer& offer,
const std::string& command,
http://git-wip-us.apache.org/repos/asf/mesos/blob/e8d6781d/src/tests/slave_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/slave_tests.cpp b/src/tests/slave_tests.cpp
index bd0796a..80fe3cf 100644
--- a/src/tests/slave_tests.cpp
+++ b/src/tests/slave_tests.cpp
@@ -539,6 +539,130 @@ TEST_F(SlaveTest, DISABLED_ROOT_RunTaskWithCommandInfoWithUser)
}
+// This test ensures that a status update acknowledgement from a
+// non-leading master is ignored.
+// TODO(bmahler): This test will need to be updated once all
+// acknowledgements go through the master.
+TEST_F(SlaveTest, IgnoreNonLeaderStatusUpdateAcknowledgement)
+{
+ Try<PID<Master> > master = StartMaster();
+ ASSERT_SOME(master);
+
+ MockExecutor exec(DEFAULT_EXECUTOR_ID);
+
+ Try<PID<Slave> > slave = StartSlave(&exec);
+ ASSERT_SOME(slave);
+
+ MockScheduler sched;
+ MesosSchedulerDriver schedDriver(
+ &sched, DEFAULT_FRAMEWORK_INFO, master.get(), DEFAULT_CREDENTIAL);
+
+ EXPECT_CALL(sched, registered(&schedDriver, _, _))
+ .Times(1);
+
+ Future<vector<Offer> > offers;
+ EXPECT_CALL(sched, resourceOffers(&schedDriver, _))
+ .WillOnce(FutureArg<1>(&offers))
+ .WillRepeatedly(Return()); // Ignore subsequent offers.
+
+ // We need to grab this message to get the scheduler's pid.
+ Future<process::Message> frameworkRegisteredMessage = FUTURE_MESSAGE(
+ Eq(FrameworkRegisteredMessage().GetTypeName()), master.get(), _);
+
+ schedDriver.start();
+
+ AWAIT_READY(frameworkRegisteredMessage);
+ const process::UPID schedulerPid = frameworkRegisteredMessage.get().to;
+
+ AWAIT_READY(offers);
+ EXPECT_NE(0u, offers.get().size());
+
+ TaskInfo task = createTask(offers.get()[0], "", DEFAULT_EXECUTOR_ID);
+
+ vector<TaskInfo> tasks;
+ tasks.push_back(task);
+
+ Future<ExecutorDriver*> execDriver;
+ EXPECT_CALL(exec, registered(_, _, _, _))
+ .WillOnce(FutureArg<0>(&execDriver));
+
+ EXPECT_CALL(exec, launchTask(_, _))
+ .WillOnce(SendStatusUpdateFromTask(TASK_RUNNING));
+
+ Future<TaskStatus> update;
+ EXPECT_CALL(sched, statusUpdate(&schedDriver, _))
+ .WillOnce(FutureArg<1>(&update));
+
+ // Pause the clock to prevent status update retries on the slave.
+ Clock::pause();
+
+ // Intercept the status update acknowledgement and send it to the
+ // master instead!
+ Future<StatusUpdateAcknowledgementMessage> acknowledgementMessage =
+ DROP_PROTOBUF(StatusUpdateAcknowledgementMessage(),
+ schedulerPid,
+ slave.get());
+
+ schedDriver.launchTasks(offers.get()[0].id(), tasks);
+
+ AWAIT_READY(update);
+ EXPECT_EQ(TASK_RUNNING, update.get().state());
+
+ AWAIT_READY(acknowledgementMessage);
+
+ // Intercept the status update acknowledgement from the master
+ // to the slave so that we can spoof a non-leading master pid.
+ Future<StatusUpdateAcknowledgementMessage> acknowledgementMessage2 =
+ DROP_PROTOBUF(StatusUpdateAcknowledgementMessage(),
+ master.get(),
+ slave.get());
+
+ // Send the acknowledgment to the master.
+ process::post(schedulerPid, master.get(), acknowledgementMessage.get());
+
+ AWAIT_READY(acknowledgementMessage2);
+
+ Future<Nothing> _statusUpdateAcknowledgement =
+ FUTURE_DISPATCH(slave.get(), &Slave::_statusUpdateAcknowledgement);
+
+ // Send the acknowledgement to the slave with a non-leading master.
+ process::post(
+ schedulerPid,
+ process::UPID("master@localhost:1"),
+ acknowledgementMessage.get());
+
+ // Make sure the acknowledgement was ignored.
+ Clock::settle();
+ ASSERT_TRUE(_statusUpdateAcknowledgement.isPending());
+
+ // Make sure the status update gets retried because the slave
+ // ignored the acknowledgement.
+ Future<TaskStatus> retriedUpdate;
+ EXPECT_CALL(sched, statusUpdate(&schedDriver, _))
+ .WillOnce(FutureArg<1>(&retriedUpdate));
+
+ Clock::advance(slave::STATUS_UPDATE_RETRY_INTERVAL_MIN);
+
+ AWAIT_READY(retriedUpdate);
+
+ // Ensure the slave receives and properly handles the ACK.
+ // Clock::settle() ensures that the slave successfully
+ // executes Slave::_statusUpdateAcknowledgement().
+ AWAIT_READY(_statusUpdateAcknowledgement);
+ Clock::settle();
+
+ Clock::resume();
+
+ EXPECT_CALL(exec, shutdown(_))
+ .Times(AtMost(1));
+
+ schedDriver.stop();
+ schedDriver.join();
+
+ Shutdown();
+}
+
+
TEST_F(SlaveTest, MetricsInStatsEndpoint)
{
Try<PID<Master> > master = StartMaster();
[2/3] git commit: Updated the slave to handle status update
acknowledgements from the master.
Posted by bm...@apache.org.
Updated the slave to handle status update acknowledgements from the master.
Review: https://reviews.apache.org/r/21837
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/2986ac8b
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/2986ac8b
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/2986ac8b
Branch: refs/heads/master
Commit: 2986ac8be22fe1c295d8eeced714fef3344815f7
Parents: 3dd88aa
Author: Benjamin Mahler <bm...@twitter.com>
Authored: Thu May 22 11:40:16 2014 -0700
Committer: Benjamin Mahler <bm...@twitter.com>
Committed: Tue May 27 15:41:03 2014 -0700
----------------------------------------------------------------------
src/slave/slave.cpp | 31 ++++++++++++++++++++++++++
src/slave/slave.hpp | 1 +
src/tests/status_update_manager_tests.cpp | 2 ++
3 files changed, 34 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/2986ac8b/src/slave/slave.cpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.cpp b/src/slave/slave.cpp
index e8fc7c6..7fbedb1 100644
--- a/src/slave/slave.cpp
+++ b/src/slave/slave.cpp
@@ -1540,11 +1540,42 @@ void Slave::updateFramework(const FrameworkID& frameworkId, const string& pid)
void Slave::statusUpdateAcknowledgement(
+ const UPID& from,
const SlaveID& slaveId,
const FrameworkID& frameworkId,
const TaskID& taskId,
const string& uuid)
{
+ // Originally, all status update acknowledgements were sent from the
+ // scheduler driver. We'd like to have all acknowledgements sent by
+ // the master instead. See: MESOS-1389.
+ // For now, we handle acknowledgements from the leading master and
+ // from the scheduler driver, for backwards compatibility.
+ // TODO(bmahler): Aim to have the scheduler driver no longer
+ // sending acknowledgements in 0.20.0. Stop handling those messages
+ // here in 0.21.0.
+ // NOTE: We must reject those acknowledgements coming from
+ // non-leading masters because we may have already sent the terminal
+ // un-acknowledged task to the leading master! Unfortunately, the
+ // master's pid will not change across runs on the same machine, so
+ // we may process a message from the old master on the same machine,
+ // but this is a more general problem!
+ if (strings::startsWith(from.id, "master")) {
+ if (state != RUNNING) {
+ LOG(WARNING) << "Dropping status update acknowledgement message for "
+ << frameworkId << " because the slave is in "
+ << state << " state";
+ return;
+ }
+
+ if (master != from) {
+ LOG(WARNING) << "Ignoring status update acknowledgement message from "
+ << from << " because it is not the expected master: "
+ << (master.isSome() ? stringify(master.get()) : "None");
+ return;
+ }
+ }
+
statusUpdateManager->acknowledgement(
taskId, frameworkId, UUID::fromBytes(uuid))
.onAny(defer(self(),
http://git-wip-us.apache.org/repos/asf/mesos/blob/2986ac8b/src/slave/slave.hpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.hpp b/src/slave/slave.hpp
index 57f84f7..34687e5 100644
--- a/src/slave/slave.hpp
+++ b/src/slave/slave.hpp
@@ -173,6 +173,7 @@ public:
const process::UPID& pid);
void statusUpdateAcknowledgement(
+ const process::UPID& from,
const SlaveID& slaveId,
const FrameworkID& frameworkId,
const TaskID& taskId,
http://git-wip-us.apache.org/repos/asf/mesos/blob/2986ac8b/src/tests/status_update_manager_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/status_update_manager_tests.cpp b/src/tests/status_update_manager_tests.cpp
index 8cbfb83..c753921 100644
--- a/src/tests/status_update_manager_tests.cpp
+++ b/src/tests/status_update_manager_tests.cpp
@@ -355,6 +355,7 @@ TEST_F(StatusUpdateManagerTest, IgnoreDuplicateStatusUpdateAck)
process::dispatch(
slave.get(),
&Slave::statusUpdateAcknowledgement,
+ master.get(),
update.slave_id(),
frameworkId,
update.status().task_id(),
@@ -447,6 +448,7 @@ TEST_F(StatusUpdateManagerTest, IgnoreUnexpectedStatusUpdateAck)
process::dispatch(
slave.get(),
&Slave::statusUpdateAcknowledgement,
+ master.get(),
update.slave_id(),
frameworkId,
update.status().task_id(),
[3/3] git commit: Added metrics for acknowledgement messages in the
master.
Posted by bm...@apache.org.
Added metrics for acknowledgement messages in the master.
Review: https://reviews.apache.org/r/21839
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/ac5059bb
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/ac5059bb
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/ac5059bb
Branch: refs/heads/master
Commit: ac5059bbec6d07e5e9091b49f84bf5a3b875634f
Parents: e8d6781
Author: Benjamin Mahler <bm...@twitter.com>
Authored: Thu May 22 14:53:15 2014 -0700
Committer: Benjamin Mahler <bm...@twitter.com>
Committed: Tue May 27 15:41:04 2014 -0700
----------------------------------------------------------------------
src/master/master.cpp | 26 ++++++++++++++++++++------
src/master/master.hpp | 4 ++++
src/tests/master_tests.cpp | 17 +++++++++++------
3 files changed, 35 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/ac5059bb/src/master/master.cpp
----------------------------------------------------------------------
diff --git a/src/master/master.cpp b/src/master/master.cpp
index 98647e5..efb4de1 100644
--- a/src/master/master.cpp
+++ b/src/master/master.cpp
@@ -2094,7 +2094,7 @@ void Master::statusUpdateAcknowledgement(
const TaskID& taskId,
const string& uuid)
{
- // TODO(bmahler): Add message counter.
+ metrics.messages_status_update_acknowledgement++;
// TODO(bmahler): Consider adding a message validator abstraction
// for the master that takes care of all this boilerplate. Ideally
@@ -2110,7 +2110,7 @@ void Master::statusUpdateAcknowledgement(
<< "Ignoring status update acknowledgement message for task " << taskId
<< " of framework " << frameworkId << " on slave " << slaveId
<< " because the framework cannot be found";
- // TODO(bmahler): Add dropped counter.
+ metrics.invalid_status_update_acknowledgements++;
return;
}
@@ -2120,7 +2120,7 @@ void Master::statusUpdateAcknowledgement(
<< " of framework " << frameworkId << " on slave " << slaveId
<< " from " << from << " because it is not from the registered framework "
<< framework->pid;
- // TODO(bmahler): Add dropped counter.
+ metrics.invalid_status_update_acknowledgements++;
return;
}
@@ -2131,7 +2131,7 @@ void Master::statusUpdateAcknowledgement(
<< "Cannot send status update acknowledgement message for task " << taskId
<< " of framework " << frameworkId << " to slave " << slaveId
<< " because slave is not activated";
- // TODO(bmahler): Add dropped counter.
+ metrics.invalid_status_update_acknowledgements++;
return;
}
@@ -2140,7 +2140,7 @@ void Master::statusUpdateAcknowledgement(
<< "Cannot send status update acknowledgement message for task " << taskId
<< " of framework " << frameworkId << " to slave " << *slave
<< " because slave is disconnected";
- // TODO(bmahler): Add dropped counter.
+ metrics.invalid_status_update_acknowledgements++;
return;
}
@@ -2161,7 +2161,7 @@ void Master::statusUpdateAcknowledgement(
send(slave->pid, message);
- // TODO(bmahler): Add valid acknowledgement counter.
+ metrics.valid_status_update_acknowledgements++;
}
@@ -3983,6 +3983,8 @@ Master::Metrics::Metrics(const Master& master)
"master/messages_deactivate_framework"),
messages_kill_task(
"master/messages_kill_task"),
+ messages_status_update_acknowledgement(
+ "master/messages_status_update_acknowledgement"),
messages_resource_request(
"master/messages_resource_request"),
messages_launch_tasks(
@@ -4013,6 +4015,10 @@ Master::Metrics::Metrics(const Master& master)
"master/valid_status_updates"),
invalid_status_updates(
"master/invalid_status_updates"),
+ valid_status_update_acknowledgements(
+ "master/valid_status_update_acknowledgements"),
+ invalid_status_update_acknowledgements(
+ "master/invalid_status_update_acknowledgements"),
recovery_slave_removals(
"master/recovery_slave_removals"),
event_queue_size(
@@ -4053,6 +4059,7 @@ Master::Metrics::Metrics(const Master& master)
process::metrics::add(messages_unregister_framework);
process::metrics::add(messages_deactivate_framework);
process::metrics::add(messages_kill_task);
+ process::metrics::add(messages_status_update_acknowledgement);
process::metrics::add(messages_resource_request);
process::metrics::add(messages_launch_tasks);
process::metrics::add(messages_revive_offers);
@@ -4075,6 +4082,9 @@ Master::Metrics::Metrics(const Master& master)
process::metrics::add(valid_status_updates);
process::metrics::add(invalid_status_updates);
+ process::metrics::add(valid_status_update_acknowledgements);
+ process::metrics::add(invalid_status_update_acknowledgements);
+
process::metrics::add(recovery_slave_removals);
process::metrics::add(event_queue_size);
@@ -4140,6 +4150,7 @@ Master::Metrics::~Metrics()
process::metrics::remove(messages_unregister_framework);
process::metrics::remove(messages_deactivate_framework);
process::metrics::remove(messages_kill_task);
+ process::metrics::remove(messages_status_update_acknowledgement);
process::metrics::remove(messages_resource_request);
process::metrics::remove(messages_launch_tasks);
process::metrics::remove(messages_revive_offers);
@@ -4162,6 +4173,9 @@ Master::Metrics::~Metrics()
process::metrics::remove(valid_status_updates);
process::metrics::remove(invalid_status_updates);
+ process::metrics::remove(valid_status_update_acknowledgements);
+ process::metrics::remove(invalid_status_update_acknowledgements);
+
process::metrics::remove(recovery_slave_removals);
process::metrics::remove(event_queue_size);
http://git-wip-us.apache.org/repos/asf/mesos/blob/ac5059bb/src/master/master.hpp
----------------------------------------------------------------------
diff --git a/src/master/master.hpp b/src/master/master.hpp
index 45d868d..4c21d9e 100644
--- a/src/master/master.hpp
+++ b/src/master/master.hpp
@@ -522,6 +522,7 @@ private:
process::metrics::Counter messages_unregister_framework;
process::metrics::Counter messages_deactivate_framework;
process::metrics::Counter messages_kill_task;
+ process::metrics::Counter messages_status_update_acknowledgement;
process::metrics::Counter messages_resource_request;
process::metrics::Counter messages_launch_tasks;
process::metrics::Counter messages_revive_offers;
@@ -544,6 +545,9 @@ private:
process::metrics::Counter valid_status_updates;
process::metrics::Counter invalid_status_updates;
+ process::metrics::Counter valid_status_update_acknowledgements;
+ process::metrics::Counter invalid_status_update_acknowledgements;
+
// Recovery counters.
process::metrics::Counter recovery_slave_removals;
http://git-wip-us.apache.org/repos/asf/mesos/blob/ac5059bb/src/tests/master_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/master_tests.cpp b/src/tests/master_tests.cpp
index 56557a5..b0ff627 100644
--- a/src/tests/master_tests.cpp
+++ b/src/tests/master_tests.cpp
@@ -1535,6 +1535,8 @@ TEST_F(MasterTest, MetricsInStatsEndpoint)
EXPECT_EQ(1u, stats.values.count("master/messages_unregister_framework"));
EXPECT_EQ(1u, stats.values.count("master/messages_deactivate_framework"));
EXPECT_EQ(1u, stats.values.count("master/messages_kill_task"));
+ EXPECT_EQ(1u, stats.values.count(
+ "master/messages_status_update_acknowledgement"));
EXPECT_EQ(1u, stats.values.count("master/messages_resource_request"));
EXPECT_EQ(1u, stats.values.count("master/messages_launch_tasks"));
EXPECT_EQ(1u, stats.values.count("master/messages_revive_offers"));
@@ -1551,16 +1553,19 @@ TEST_F(MasterTest, MetricsInStatsEndpoint)
// Messages from both schedulers and slaves.
EXPECT_EQ(1u, stats.values.count("master/messages_authenticate"));
- EXPECT_EQ(
- 1u,
- stats.values.count("master/valid_framework_to_executor_messages"));
- EXPECT_EQ(
- 1u,
- stats.values.count("master/invalid_framework_to_executor_messages"));
+ EXPECT_EQ(1u, stats.values.count(
+ "master/valid_framework_to_executor_messages"));
+ EXPECT_EQ(1u, stats.values.count(
+ "master/invalid_framework_to_executor_messages"));
EXPECT_EQ(1u, stats.values.count("master/valid_status_updates"));
EXPECT_EQ(1u, stats.values.count("master/invalid_status_updates"));
+ EXPECT_EQ(1u, stats.values.count(
+ "master/valid_status_update_acknowledgements"));
+ EXPECT_EQ(1u, stats.values.count(
+ "master/invalid_status_update_acknowledgements"));
+
EXPECT_EQ(1u, stats.values.count("master/recovery_slave_removals"));
EXPECT_EQ(1u, stats.values.count("master/event_queue_size"));