You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mesos.apache.org by gr...@apache.org on 2017/11/21 00:54:16 UTC
[1/6] mesos git commit: Added a missing protobuf field to the
unversioned definitions.
Repository: mesos
Updated Branches:
refs/heads/master f145a4df6 -> 62d117334
Added a missing protobuf field to the unversioned definitions.
The new `OfferOperationUpdate` event type was added to the unversioned
scheduler API protobuf definitions, but the corresponding field was not
added to the `Event` message.
Review: https://reviews.apache.org/r/63767/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/056349f8
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/056349f8
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/056349f8
Branch: refs/heads/master
Commit: 056349f88f8f0933907443868845c4910edb2173
Parents: f145a4d
Author: Greg Mann <gr...@mesosphere.io>
Authored: Mon Nov 20 16:42:09 2017 -0800
Committer: Greg Mann <gr...@gmail.com>
Committed: Mon Nov 20 16:48:49 2017 -0800
----------------------------------------------------------------------
include/mesos/scheduler/scheduler.proto | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/056349f8/include/mesos/scheduler/scheduler.proto
----------------------------------------------------------------------
diff --git a/include/mesos/scheduler/scheduler.proto b/include/mesos/scheduler/scheduler.proto
index f5dda10..112f496 100644
--- a/include/mesos/scheduler/scheduler.proto
+++ b/include/mesos/scheduler/scheduler.proto
@@ -200,6 +200,7 @@ message Event {
optional Rescind rescind = 4;
optional RescindInverseOffer rescind_inverse_offer = 10;
optional Update update = 5;
+ optional OfferOperationUpdate offer_operation_update = 11;
optional Message message = 6;
optional Failure failure = 7;
optional Error error = 8;
[4/6] mesos git commit: Added "Task" prefix to status update manager
related classes/methods.
Posted by gr...@apache.org.
Added "Task" prefix to status update manager related classes/methods.
To prepare for the addition of an offer operation status update
manager and a generic `StatusUpdateManagerProcess` class, this patch
renames existing objects which would conflict with the new
classes/methods/variables.
Review: https://reviews.apache.org/r/63852/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/2377cd8f
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/2377cd8f
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/2377cd8f
Branch: refs/heads/master
Commit: 2377cd8f02705994f27925659445634359d305c2
Parents: d3a3896
Author: Gaston Kleiman <ga...@mesosphere.io>
Authored: Mon Nov 20 16:42:51 2017 -0800
Committer: Greg Mann <gr...@gmail.com>
Committed: Mon Nov 20 16:49:47 2017 -0800
----------------------------------------------------------------------
src/local/local.cpp | 21 +-
src/messages/messages.proto | 9 +-
src/slave/main.cpp | 9 +-
src/slave/slave.cpp | 136 ++++++-------
src/slave/slave.hpp | 14 +-
src/slave/status_update_manager.cpp | 196 ++++++++++---------
src/slave/status_update_manager.hpp | 22 +--
src/tests/api_tests.cpp | 2 +-
src/tests/cluster.cpp | 11 +-
src/tests/cluster.hpp | 5 +-
src/tests/master_slave_reconciliation_tests.cpp | 8 +-
src/tests/master_tests.cpp | 8 +-
src/tests/mock_slave.cpp | 4 +-
src/tests/mock_slave.hpp | 2 +-
src/tests/reconciliation_tests.cpp | 4 +-
src/tests/slave_recovery_tests.cpp | 5 +-
src/tests/slave_tests.cpp | 6 +-
src/tests/status_update_manager_tests.cpp | 19 +-
18 files changed, 250 insertions(+), 231 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/2377cd8f/src/local/local.cpp
----------------------------------------------------------------------
diff --git a/src/local/local.cpp b/src/local/local.cpp
index 99c1b08..63d9822 100644
--- a/src/local/local.cpp
+++ b/src/local/local.cpp
@@ -104,7 +104,7 @@ using mesos::internal::slave::Containerizer;
using mesos::internal::slave::Fetcher;
using mesos::internal::slave::GarbageCollector;
using mesos::internal::slave::Slave;
-using mesos::internal::slave::StatusUpdateManager;
+using mesos::internal::slave::TaskStatusUpdateManager;
using mesos::modules::Anonymous;
using mesos::modules::ModuleManager;
@@ -143,7 +143,7 @@ static MasterContender* contender = nullptr;
static Option<Authorizer*> authorizer_ = None();
static Files* files = nullptr;
static vector<GarbageCollector*>* garbageCollectors = nullptr;
-static vector<StatusUpdateManager*>* statusUpdateManagers = nullptr;
+static vector<TaskStatusUpdateManager*>* taskStatusUpdateManagers = nullptr;
static vector<Fetcher*>* fetchers = nullptr;
static vector<ResourceEstimator*>* resourceEstimators = nullptr;
static vector<QoSController*>* qosControllers = nullptr;
@@ -351,7 +351,7 @@ PID<Master> launch(const Flags& flags, Allocator* _allocator)
PID<Master> pid = process::spawn(master);
garbageCollectors = new vector<GarbageCollector*>();
- statusUpdateManagers = new vector<StatusUpdateManager*>();
+ taskStatusUpdateManagers = new vector<TaskStatusUpdateManager*>();
fetchers = new vector<Fetcher*>();
resourceEstimators = new vector<ResourceEstimator*>();
qosControllers = new vector<QoSController*>();
@@ -408,7 +408,8 @@ PID<Master> launch(const Flags& flags, Allocator* _allocator)
}
garbageCollectors->push_back(new GarbageCollector());
- statusUpdateManagers->push_back(new StatusUpdateManager(slaveFlags));
+ taskStatusUpdateManagers->push_back(
+ new TaskStatusUpdateManager(slaveFlags));
fetchers->push_back(new Fetcher(slaveFlags));
Try<ResourceEstimator*> resourceEstimator =
@@ -472,7 +473,7 @@ PID<Master> launch(const Flags& flags, Allocator* _allocator)
containerizer.get(),
files,
garbageCollectors->back(),
- statusUpdateManagers->back(),
+ taskStatusUpdateManagers->back(),
resourceEstimators->back(),
qosControllers->back(),
authorizer_); // Same authorizer as master.
@@ -531,12 +532,14 @@ void shutdown()
delete garbageCollectors;
garbageCollectors = nullptr;
- foreach (StatusUpdateManager* statusUpdateManager, *statusUpdateManagers) {
- delete statusUpdateManager;
+ foreach (
+ TaskStatusUpdateManager* taskStatusUpdateManager,
+ *taskStatusUpdateManagers) {
+ delete taskStatusUpdateManager;
}
- delete statusUpdateManagers;
- statusUpdateManagers = nullptr;
+ delete taskStatusUpdateManagers;
+ taskStatusUpdateManagers = nullptr;
foreach (Fetcher* fetcher, *fetchers) {
delete fetcher;
http://git-wip-us.apache.org/repos/asf/mesos/blob/2377cd8f/src/messages/messages.proto
----------------------------------------------------------------------
diff --git a/src/messages/messages.proto b/src/messages/messages.proto
index 33732e2..d7ef68f 100644
--- a/src/messages/messages.proto
+++ b/src/messages/messages.proto
@@ -59,9 +59,10 @@ message StatusUpdate {
// This corresponds to the latest state of the task according to the
// agent. Note that this state might be different than the state in
- // 'status' because status update manager queues updates. In other
- // words, 'status' corresponds to the update at top of the queue and
- // 'latest_state' corresponds to the update at bottom of the queue.
+ // 'status' because task status update manager queues updates. In
+ // other words, 'status' corresponds to the update at top of the
+ // queue and 'latest_state' corresponds to the update at bottom of
+ // the queue.
optional TaskState latest_state = 7;
}
@@ -69,7 +70,7 @@ message StatusUpdate {
/**
* Encapsulates how we checkpoint a `StatusUpdate` to disk.
*
- * See the StatusUpdateManager and slave/state.cpp.
+ * See the `TaskStatusUpdateManager` and slave/state.cpp.
*/
message StatusUpdateRecord {
enum Type {
http://git-wip-us.apache.org/repos/asf/mesos/blob/2377cd8f/src/slave/main.cpp
----------------------------------------------------------------------
diff --git a/src/slave/main.cpp b/src/slave/main.cpp
index 94431b0..de87553 100644
--- a/src/slave/main.cpp
+++ b/src/slave/main.cpp
@@ -239,7 +239,7 @@ int main(int argc, char** argv)
// * Master detector.
// * Authorizer.
// * Garbage collector.
- // * Status update manager.
+ // * Task status update manager.
// * Resource estimator.
// * QoS controller.
// * `Agent` process.
@@ -533,7 +533,8 @@ int main(int argc, char** argv)
Files* files = new Files(READONLY_HTTP_AUTHENTICATION_REALM, authorizer_);
GarbageCollector* gc = new GarbageCollector();
- StatusUpdateManager* statusUpdateManager = new StatusUpdateManager(flags);
+ TaskStatusUpdateManager* taskStatusUpdateManager =
+ new TaskStatusUpdateManager(flags);
Try<ResourceEstimator*> resourceEstimator =
ResourceEstimator::create(flags.resource_estimator);
@@ -558,7 +559,7 @@ int main(int argc, char** argv)
containerizer.get(),
files,
gc,
- statusUpdateManager,
+ taskStatusUpdateManager,
resourceEstimator.get(),
qosController.get(),
authorizer_);
@@ -572,7 +573,7 @@ int main(int argc, char** argv)
delete resourceEstimator.get();
- delete statusUpdateManager;
+ delete taskStatusUpdateManager;
delete gc;
http://git-wip-us.apache.org/repos/asf/mesos/blob/2377cd8f/src/slave/slave.cpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.cpp b/src/slave/slave.cpp
index d8edc5e..c3a4088 100644
--- a/src/slave/slave.cpp
+++ b/src/slave/slave.cpp
@@ -190,7 +190,7 @@ Slave::Slave(const string& id,
Containerizer* _containerizer,
Files* _files,
GarbageCollector* _gc,
- StatusUpdateManager* _statusUpdateManager,
+ TaskStatusUpdateManager* _taskStatusUpdateManager,
ResourceEstimator* _resourceEstimator,
QoSController* _qosController,
const Option<Authorizer*>& _authorizer)
@@ -209,7 +209,7 @@ Slave::Slave(const string& id,
files(_files),
metrics(*this),
gc(_gc),
- statusUpdateManager(_statusUpdateManager),
+ taskStatusUpdateManager(_taskStatusUpdateManager),
masterPingTimeout(DEFAULT_MASTER_PING_TIMEOUT()),
metaDir(paths::getMetaRootDir(flags.work_dir)),
recoveryErrors(0),
@@ -609,13 +609,13 @@ void Slave::initialize()
LOG(INFO) << "Agent hostname: " << info.hostname();
- statusUpdateManager->initialize(defer(self(), &Slave::forward, lambda::_1)
+ taskStatusUpdateManager->initialize(defer(self(), &Slave::forward, lambda::_1)
.operator std::function<void(StatusUpdate)>());
- // We pause the status update manager so that it doesn't forward any updates
- // while the slave is still recovering. It is unpaused/resumed when the slave
- // (re-)registers with the master.
- statusUpdateManager->pause();
+ // We pause the task status update manager so that it doesn't forward any
+ // updates while the slave is still recovering. It is unpaused/resumed when
+ // the slave (re-)registers with the master.
+ taskStatusUpdateManager->pause();
// Start disk monitoring.
// NOTE: We send a delayed message here instead of directly calling
@@ -984,7 +984,7 @@ void Slave::detected(const Future<Option<MasterInfo>>& _master)
}
// Pause the status updates.
- statusUpdateManager->pause();
+ taskStatusUpdateManager->pause();
if (_master.isFailed()) {
EXIT(EXIT_FAILURE) << "Failed to detect a master: " << _master.failure();
@@ -1214,7 +1214,7 @@ void Slave::registered(
// is safe even if no timer is active or pending.
Clock::cancel(agentRegistrationTimer);
- statusUpdateManager->resume(); // Resume status updates.
+ taskStatusUpdateManager->resume(); // Resume status updates.
info.mutable_id()->CopyFrom(slaveId); // Store the slave id.
@@ -1339,7 +1339,7 @@ void Slave::reregistered(
case DISCONNECTED:
LOG(INFO) << "Re-registered with master " << master.get();
state = RUNNING;
- statusUpdateManager->resume(); // Resume status updates.
+ taskStatusUpdateManager->resume(); // Resume status updates.
// Setup a timer so that the agent attempts to re-register if it
// doesn't receive a ping from the master for an extended period
@@ -1461,7 +1461,7 @@ void Slave::reregistered(
// NOTE: We can't use statusUpdate() here because it drops
// updates for unknown frameworks.
- statusUpdateManager->update(update, info.id())
+ taskStatusUpdateManager->update(update, info.id())
.onAny(defer(self(),
&Slave::___statusUpdate,
lambda::_1,
@@ -2049,10 +2049,10 @@ void Slave::_run(
" directories scheduled for gc",
TaskStatus::REASON_GC_ERROR);
- // TODO(vinod): Ensure that the status update manager reliably
- // delivers this update. Currently, we don't guarantee this
- // because removal of the framework causes the status update
- // manager to stop retrying for its un-acked updates.
+ // TODO(vinod): Ensure that the task status update manager
+ // reliably delivers this update. Currently, we don't guarantee
+ // this because removal of the framework causes the status
+ // update manager to stop retrying for its un-acked updates.
statusUpdate(update, UPID());
}
@@ -3456,9 +3456,9 @@ void Slave::updateFramework(
framework->checkpointFramework();
}
- // Inform status update manager to immediately resend any pending
+ // Inform task status update manager to immediately resend any pending
// updates.
- statusUpdateManager->resume();
+ taskStatusUpdateManager->resume();
break;
}
@@ -3766,7 +3766,7 @@ void Slave::statusUpdateAcknowledgement(
}
}
- statusUpdateManager->acknowledgement(
+ taskStatusUpdateManager->acknowledgement(
taskId, frameworkId, UUID::fromBytes(uuid).get())
.onAny(defer(self(),
&Slave::_statusUpdateAcknowledgement,
@@ -3798,7 +3798,7 @@ void Slave::_statusUpdateAcknowledgement(
return;
}
- VLOG(1) << "Status update manager successfully handled status update"
+ VLOG(1) << "Task status update manager successfully handled status update"
<< " acknowledgement (UUID: " << uuid
<< ") for task " << taskId
<< " of framework " << frameworkId;
@@ -3964,10 +3964,10 @@ void Slave::subscribe(
executor->send(event);
// Handle all the pending updates.
- // The status update manager might have already checkpointed some
- // of these pending updates (for example, if the slave died right
- // after it checkpointed the update but before it could send the
- // ACK to the executor). This is ok because the status update
+ // The task status update manager might have already checkpointed
+ // some of these pending updates (for example, if the slave died
+ // right after it checkpointed the update but before it could send
+ // the ACK to the executor). This is ok because the status update
// manager correctly handles duplicate updates.
foreach (const Call::Update& update, subscribe.unacknowledged_updates()) {
// NOTE: This also updates the executor's resources!
@@ -4320,10 +4320,10 @@ void Slave::reregisterExecutor(
send(executor->pid.get(), message);
// Handle all the pending updates.
- // The status update manager might have already checkpointed some
- // of these pending updates (for example, if the slave died right
- // after it checkpointed the update but before it could send the
- // ACK to the executor). This is ok because the status update
+ // The task status update manager might have already checkpointed
+ // some of these pending updates (for example, if the slave died
+ // right after it checkpointed the update but before it could send
+ // the ACK to the executor). This is ok because the status update
// manager correctly handles duplicate updates.
foreach (const StatusUpdate& update, updates) {
// NOTE: This also updates the executor's resources!
@@ -4623,7 +4623,7 @@ void Slave::statusUpdate(StatusUpdate update, const Option<UPID>& pid)
metrics.valid_status_updates++;
- statusUpdateManager->update(update, info.id())
+ taskStatusUpdateManager->update(update, info.id())
.onAny(defer(self(), &Slave::___statusUpdate, lambda::_1, update, pid));
}
@@ -4651,7 +4651,7 @@ void Slave::statusUpdate(StatusUpdate update, const Option<UPID>& pid)
// because the container is unknown. We cannot use the slave IP
// address here (for the `NetworkInfo`) since we do not know the
// type of network isolation used for this container.
- statusUpdateManager->update(update, info.id())
+ taskStatusUpdateManager->update(update, info.id())
.onAny(defer(self(), &Slave::___statusUpdate, lambda::_1, update, pid));
return;
@@ -4778,14 +4778,14 @@ void Slave::_statusUpdate(
// We set the latest state of the task here so that the slave can
// inform the master about the latest state (via status update or
- // ReregisterSlaveMessage message) as soon as possible. Master can
- // use this information, for example, to release resources as soon
- // as the latest state of the task reaches a terminal state. This
- // is important because status update manager queues updates and
+ // ReregisterSlaveMessage message) as soon as possible. Master can use
+ // this information, for example, to release resources as soon as the
+ // latest state of the task reaches a terminal state. This is
+ // important because task status update manager queues updates and
// only sends one update per task at a time; the next update for a
- // task is sent only after the acknowledgement for the previous one
- // is received, which could take a long time if the framework is
- // backed up or is down.
+ // task is sent only after the acknowledgement for the previous one is
+ // received, which could take a long time if the framework is backed
+ // up or is down.
Try<Nothing> updated = executor->updateTaskState(status);
// If we fail to update the task state, drop the update. Note that
@@ -4794,9 +4794,9 @@ void Slave::_statusUpdate(
LOG(ERROR) << "Failed to update state of task '" << status.task_id() << "'"
<< " to " << status.state() << ": " << updated.error();
- // NOTE: This may lead to out-of-order acknowledgements since
- // other update acknowledgements may be waiting for the
- // containerizer or status update manager.
+ // NOTE: This may lead to out-of-order acknowledgements since other
+ // update acknowledgements may be waiting for the containerizer or
+ // task status update manager.
___statusUpdate(Nothing(), update, pid);
return;
}
@@ -4871,12 +4871,13 @@ void Slave::__statusUpdate(
}
if (checkpoint) {
- // Ask the status update manager to checkpoint and reliably send the update.
- statusUpdateManager->update(update, info.id(), executorId, containerId)
+ // Ask the task status update manager to checkpoint and reliably send the
+ // update.
+ taskStatusUpdateManager->update(update, info.id(), executorId, containerId)
.onAny(defer(self(), &Slave::___statusUpdate, lambda::_1, update, pid));
} else {
- // Ask the status update manager to just retry the update.
- statusUpdateManager->update(update, info.id())
+ // Ask the task status update manager to just retry the update.
+ taskStatusUpdateManager->update(update, info.id())
.onAny(defer(self(), &Slave::___statusUpdate, lambda::_1, update, pid));
}
}
@@ -4889,7 +4890,7 @@ void Slave::___statusUpdate(
{
CHECK_READY(future) << "Failed to handle status update " << update;
- VLOG(1) << "Status update manager successfully handled status update "
+ VLOG(1) << "Task status update manager successfully handled status update "
<< update;
if (pid == UPID()) {
@@ -4902,7 +4903,7 @@ void Slave::___statusUpdate(
message.mutable_task_id()->MergeFrom(update.status().task_id());
message.set_uuid(update.uuid());
- // Status update manager successfully handled the status update.
+ // Task status update manager successfully handled the status update.
// Acknowledge the executor, if we have a valid pid.
if (pid.isSome()) {
LOG(INFO) << "Sending acknowledgement for status update " << update
@@ -4933,7 +4934,7 @@ void Slave::___statusUpdate(
// NOTE: An acknowledgement for this update might have already been
-// processed by the slave but not the status update manager.
+// processed by the slave but not the task status update manager.
void Slave::forward(StatusUpdate update)
{
CHECK(state == RECOVERING || state == DISCONNECTED ||
@@ -4942,15 +4943,16 @@ void Slave::forward(StatusUpdate update)
if (state != RUNNING) {
LOG(WARNING) << "Dropping status update " << update
- << " sent by status update manager because the agent"
+ << " sent by task status update manager because the agent"
<< " is in " << state << " state";
return;
}
- // Ensure that task status uuid is set even if this update was sent by the
- // status update manager after recovering a pre 0.23.x slave/executor driver's
- // updates. This allows us to simplify the master code (in >= 0.27.0) to
- // assume the uuid is always set for retryable updates.
+ // Ensure that task status uuid is set even if this update was sent by
+ // the task status update manager after recovering a pre 0.23.x
+ // slave/executor driver's updates. This allows us to simplify the
+ // master code (in >= 0.27.0) to assume the uuid is always set for
+ // retryable updates.
CHECK(update.has_uuid())
<< "Expecting updates without 'uuid' to have been rejected";
@@ -4979,8 +4981,8 @@ void Slave::forward(StatusUpdate update)
// steady state master updates the status update state of the
// task when it receives this update. If the master fails over,
// slave re-registers with this task in this status update
- // state. Note that an acknowledgement for this update might
- // be enqueued on status update manager when we are here. But
+ // state. Note that an acknowledgement for this update might be
+ // enqueued on task status update manager when we are here. But
// that is ok because the status update state will be updated
// when the next update is forwarded to the slave.
task->set_status_update_state(update.status().state());
@@ -4997,12 +4999,12 @@ void Slave::forward(StatusUpdate update)
CHECK_SOME(master);
LOG(INFO) << "Forwarding the update " << update << " to " << master.get();
- // NOTE: We forward the update even if framework/executor/task
- // doesn't exist because the status update manager will be expecting
- // an acknowledgement for the update. This could happen for example
- // if this is a retried terminal update and before we are here the
- // slave has already processed the acknowledgement of the original
- // update and removed the framework/executor/task. Also, slave
+ // NOTE: We forward the update even if framework/executor/task doesn't
+ // exist because the task status update manager will be expecting an
+ // acknowledgement for the update. This could happen for example if
+ // this is a retried terminal update and before we are here the slave
+ // has already processed the acknowledgement of the original update
+ // and removed the framework/executor/task. Also, slave
// re-registration can generate updates when framework/executor/task
// are unknown.
@@ -5573,11 +5575,11 @@ void Slave::executorTerminated(
// If the containerizer killed the executor (e.g., due to OOM event)
// or if this is a command executor, we send TASK_FAILED status updates
// instead of TASK_GONE.
- // NOTE: We don't send updates if the framework is terminating
- // because we don't want the status update manager to keep retrying
- // these updates since it won't receive ACKs from the scheduler. Also,
- // the status update manager should have already cleaned up all the
- // status update streams for a framework that is terminating.
+ // NOTE: We don't send updates if the framework is terminating because we
+ // don't want the task status update manager to keep retrying these
+ // updates since it won't receive ACKs from the scheduler. Also, the task
+ // status update manager should have already cleaned up all the status
+ // update streams for a framework that is terminating.
if (framework->state != Framework::TERMINATING) {
// Transition all live launched tasks. Note that the map is
// removed from within the loop due terminal status updates.
@@ -5755,8 +5757,8 @@ void Slave::removeFramework(Framework* framework)
// We only remove frameworks once they become idle.
CHECK(framework->idle());
- // Close all status update streams for this framework.
- statusUpdateManager->cleanup(framework->id());
+ // Close all task status update streams for this framework.
+ taskStatusUpdateManager->cleanup(framework->id());
// Schedule the framework work and meta directories for garbage
// collection.
@@ -6286,7 +6288,7 @@ Future<Nothing> Slave::recover(const Try<state::State>& state)
}
}
- return statusUpdateManager->recover(metaDir, slaveState)
+ return taskStatusUpdateManager->recover(metaDir, slaveState)
.then(defer(self(), &Slave::_recoverContainerizer, slaveState));
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/2377cd8f/src/slave/slave.hpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.hpp b/src/slave/slave.hpp
index 40442f2..f86ebd6 100644
--- a/src/slave/slave.hpp
+++ b/src/slave/slave.hpp
@@ -103,7 +103,7 @@ namespace internal {
namespace slave {
// Some forward declarations.
-class StatusUpdateManager;
+class TaskStatusUpdateManager;
class Executor;
class Framework;
@@ -119,7 +119,7 @@ public:
Containerizer* containerizer,
Files* files,
GarbageCollector* gc,
- StatusUpdateManager* statusUpdateManager,
+ TaskStatusUpdateManager* taskStatusUpdateManager,
mesos::slave::ResourceEstimator* resourceEstimator,
mesos::slave::QoSController* qosController,
const Option<Authorizer*>& authorizer);
@@ -235,7 +235,7 @@ public:
void ping(const process::UPID& from, bool connected);
- // Handles the status update.
+ // Handles the task status update.
// NOTE: If 'pid' is a valid UPID an ACK is sent to this pid
// after the update is successfully handled. If pid == UPID()
// no ACK is sent. The latter is used by the slave to send
@@ -264,7 +264,7 @@ public:
const ContainerID& containerId,
bool checkpoint);
- // This is called when the status update manager finishes
+ // This is called when the task status update manager finishes
// handling the update. If the handling is successful, an
// acknowledgment is sent to the executor.
void ___statusUpdate(
@@ -272,7 +272,7 @@ public:
const StatusUpdate& update,
const Option<process::UPID>& pid);
- // This is called by status update manager to forward a status
+ // This is called by task status update manager to forward a status
// update to the master. Note that the latest state of the task is
// added to the update before forwarding.
void forward(StatusUpdate update);
@@ -424,7 +424,7 @@ public:
// Checks the current disk usage and schedules for gc as necessary.
void checkDiskUsage();
- // Recovers the slave, status update manager and isolator.
+ // Recovers the slave, task status update manager and isolator.
process::Future<Nothing> recover(const Try<state::State>& state);
// This is called after 'recover()'. If 'flags.reconnect' is
@@ -623,7 +623,7 @@ private:
GarbageCollector* gc;
- StatusUpdateManager* statusUpdateManager;
+ TaskStatusUpdateManager* taskStatusUpdateManager;
// Master detection future.
process::Future<Option<MasterInfo>> detection;
http://git-wip-us.apache.org/repos/asf/mesos/blob/2377cd8f/src/slave/status_update_manager.cpp
----------------------------------------------------------------------
diff --git a/src/slave/status_update_manager.cpp b/src/slave/status_update_manager.cpp
index e0d029b..fed0903 100644
--- a/src/slave/status_update_manager.cpp
+++ b/src/slave/status_update_manager.cpp
@@ -64,17 +64,17 @@ using state::RunState;
using state::TaskState;
-class StatusUpdateManagerProcess
- : public ProtobufProcess<StatusUpdateManagerProcess>
+class TaskStatusUpdateManagerProcess
+ : public ProtobufProcess<TaskStatusUpdateManagerProcess>
{
public:
- StatusUpdateManagerProcess(const Flags& flags);
- virtual ~StatusUpdateManagerProcess();
+ TaskStatusUpdateManagerProcess(const Flags& flags);
+ virtual ~TaskStatusUpdateManagerProcess();
// Explicitly use 'initialize' since we're overloading below.
using process::ProcessBase::initialize;
- // StatusUpdateManager implementation.
+ // TaskStatusUpdateManager implementation.
void initialize(const function<void(StatusUpdate)>& forward);
Future<Nothing> update(
@@ -123,7 +123,7 @@ private:
// Creates a new status update stream (opening the updates file, if path is
// present) and adds it to streams.
- StatusUpdateStream* createStatusUpdateStream(
+ TaskStatusUpdateStream* createStatusUpdateStream(
const TaskID& taskId,
const FrameworkID& frameworkId,
const SlaveID& slaveId,
@@ -131,7 +131,7 @@ private:
const Option<ExecutorID>& executorId,
const Option<ContainerID>& containerId);
- StatusUpdateStream* getStatusUpdateStream(
+ TaskStatusUpdateStream* getStatusUpdateStream(
const TaskID& taskId,
const FrameworkID& frameworkId);
@@ -144,20 +144,23 @@ private:
function<void(StatusUpdate)> forward_;
- hashmap<FrameworkID, hashmap<TaskID, StatusUpdateStream*>> streams;
+ hashmap<FrameworkID, hashmap<TaskID, TaskStatusUpdateStream*>> streams;
};
-StatusUpdateManagerProcess::StatusUpdateManagerProcess(const Flags& _flags)
- : ProcessBase(process::ID::generate("status-update-manager")),
+TaskStatusUpdateManagerProcess::TaskStatusUpdateManagerProcess(
+ const Flags& _flags)
+ : ProcessBase(process::ID::generate("task-status-update-manager")),
flags(_flags),
- paused(false) {}
+ paused(false)
+{
+}
-StatusUpdateManagerProcess::~StatusUpdateManagerProcess()
+TaskStatusUpdateManagerProcess::~TaskStatusUpdateManagerProcess()
{
foreachkey (const FrameworkID& frameworkId, streams) {
- foreachvalue (StatusUpdateStream* stream, streams[frameworkId]) {
+ foreachvalue (TaskStatusUpdateStream* stream, streams[frameworkId]) {
delete stream;
}
}
@@ -165,30 +168,30 @@ StatusUpdateManagerProcess::~StatusUpdateManagerProcess()
}
-void StatusUpdateManagerProcess::initialize(
+void TaskStatusUpdateManagerProcess::initialize(
const function<void(StatusUpdate)>& forward)
{
forward_ = forward;
}
-void StatusUpdateManagerProcess::pause()
+void TaskStatusUpdateManagerProcess::pause()
{
- LOG(INFO) << "Pausing sending status updates";
+ LOG(INFO) << "Pausing sending task status updates";
paused = true;
}
-void StatusUpdateManagerProcess::resume()
+void TaskStatusUpdateManagerProcess::resume()
{
- LOG(INFO) << "Resuming sending status updates";
+ LOG(INFO) << "Resuming sending task status updates";
paused = false;
foreachkey (const FrameworkID& frameworkId, streams) {
- foreachvalue (StatusUpdateStream* stream, streams[frameworkId]) {
+ foreachvalue (TaskStatusUpdateStream* stream, streams[frameworkId]) {
if (!stream->pending.empty()) {
const StatusUpdate& update = stream->pending.front();
- LOG(WARNING) << "Resending status update " << update;
+ LOG(WARNING) << "Resending task status update " << update;
stream->timeout = forward(update, STATUS_UPDATE_RETRY_INTERVAL_MIN);
}
}
@@ -196,11 +199,11 @@ void StatusUpdateManagerProcess::resume()
}
-Future<Nothing> StatusUpdateManagerProcess::recover(
+Future<Nothing> TaskStatusUpdateManagerProcess::recover(
const string& rootDir,
const Option<SlaveState>& state)
{
- LOG(INFO) << "Recovering status update manager";
+ LOG(INFO) << "Recovering task status update manager";
if (state.isNone()) {
return Nothing();
@@ -212,7 +215,7 @@ Future<Nothing> StatusUpdateManagerProcess::recover(
<< "' of framework " << framework.id;
if (executor.info.isNone()) {
- LOG(WARNING) << "Skipping recovering updates of"
+ LOG(WARNING) << "Skipping recovering task status updates of"
<< " executor '" << executor.id
<< "' of framework " << framework.id
<< " because its info cannot be recovered";
@@ -220,7 +223,7 @@ Future<Nothing> StatusUpdateManagerProcess::recover(
}
if (executor.latest.isNone()) {
- LOG(WARNING) << "Skipping recovering updates of"
+ LOG(WARNING) << "Skipping recovering task status updates of"
<< " executor '" << executor.id
<< "' of framework " << framework.id
<< " because its latest run cannot be recovered";
@@ -233,7 +236,7 @@ Future<Nothing> StatusUpdateManagerProcess::recover(
CHECK_SOME(run);
if (run->completed) {
- VLOG(1) << "Skipping recovering updates of"
+ VLOG(1) << "Skipping recovering task status updates of"
<< " executor '" << executor.id
<< "' of framework " << framework.id
<< " because its latest run " << latest.value()
@@ -247,13 +250,13 @@ Future<Nothing> StatusUpdateManagerProcess::recover(
// 1) the executor never received this task or
// 2) executor launched it but the slave died before it got any updates.
if (task.updates.empty()) {
- LOG(WARNING) << "No updates found for task " << task.id
+ LOG(WARNING) << "No status updates found for task " << task.id
<< " of framework " << framework.id;
continue;
}
// Create a new status update stream.
- StatusUpdateStream* stream = createStatusUpdateStream(
+ TaskStatusUpdateStream* stream = createStatusUpdateStream(
task.id, framework.id, state->id, true, executor.id, latest);
// Replay the stream.
@@ -280,9 +283,10 @@ Future<Nothing> StatusUpdateManagerProcess::recover(
}
-void StatusUpdateManagerProcess::cleanup(const FrameworkID& frameworkId)
+void TaskStatusUpdateManagerProcess::cleanup(const FrameworkID& frameworkId)
{
- LOG(INFO) << "Closing status update streams for framework " << frameworkId;
+ LOG(INFO) << "Closing task status update streams for framework "
+ << frameworkId;
if (streams.contains(frameworkId)) {
foreachkey (const TaskID& taskId, utils::copy(streams[frameworkId])) {
@@ -292,7 +296,7 @@ void StatusUpdateManagerProcess::cleanup(const FrameworkID& frameworkId)
}
-Future<Nothing> StatusUpdateManagerProcess::update(
+Future<Nothing> TaskStatusUpdateManagerProcess::update(
const StatusUpdate& update,
const SlaveID& slaveId,
const ExecutorID& executorId,
@@ -302,7 +306,7 @@ Future<Nothing> StatusUpdateManagerProcess::update(
}
-Future<Nothing> StatusUpdateManagerProcess::update(
+Future<Nothing> TaskStatusUpdateManagerProcess::update(
const StatusUpdate& update,
const SlaveID& slaveId)
{
@@ -310,7 +314,7 @@ Future<Nothing> StatusUpdateManagerProcess::update(
}
-Future<Nothing> StatusUpdateManagerProcess::_update(
+Future<Nothing> TaskStatusUpdateManagerProcess::_update(
const StatusUpdate& update,
const SlaveID& slaveId,
bool checkpoint,
@@ -320,11 +324,11 @@ Future<Nothing> StatusUpdateManagerProcess::_update(
const TaskID& taskId = update.status().task_id();
const FrameworkID& frameworkId = update.framework_id();
- LOG(INFO) << "Received status update " << update;
+ LOG(INFO) << "Received task status update " << update;
// Write the status update to disk and enqueue it to send it to the master.
// Create/Get the status update stream for this task.
- StatusUpdateStream* stream = getStatusUpdateStream(taskId, frameworkId);
+ TaskStatusUpdateStream* stream = getStatusUpdateStream(taskId, frameworkId);
if (stream == nullptr) {
stream = createStatusUpdateStream(
taskId, frameworkId, slaveId, checkpoint, executorId, containerId);
@@ -334,9 +338,10 @@ Future<Nothing> StatusUpdateManagerProcess::_update(
// stream that is checkpointable, and vice-versa.
if (stream->checkpoint != checkpoint) {
return Failure(
- "Mismatched checkpoint value for status update " + stringify(update) +
- " (expected checkpoint=" + stringify(stream->checkpoint) +
- " actual checkpoint=" + stringify(checkpoint) + ")");
+ "Mismatched checkpoint value for task status update " +
+ stringify(update) + " (expected checkpoint=" +
+ stringify(stream->checkpoint) + " actual checkpoint=" +
+ stringify(checkpoint) + ")");
}
// Handle the status update.
@@ -368,13 +373,13 @@ Future<Nothing> StatusUpdateManagerProcess::_update(
}
-Timeout StatusUpdateManagerProcess::forward(
+Timeout TaskStatusUpdateManagerProcess::forward(
const StatusUpdate& update,
const Duration& duration)
{
CHECK(!paused);
- VLOG(1) << "Forwarding update " << update << " to the agent";
+ VLOG(1) << "Forwarding task status update " << update << " to the agent";
// Forward the update.
forward_(update);
@@ -382,28 +387,28 @@ Timeout StatusUpdateManagerProcess::forward(
// Send a message to self to resend after some delay if no ACK is received.
return delay(duration,
self(),
- &StatusUpdateManagerProcess::timeout,
+ &TaskStatusUpdateManagerProcess::timeout,
duration).timeout();
}
-Future<bool> StatusUpdateManagerProcess::acknowledgement(
+Future<bool> TaskStatusUpdateManagerProcess::acknowledgement(
const TaskID& taskId,
const FrameworkID& frameworkId,
const UUID& uuid)
{
- LOG(INFO) << "Received status update acknowledgement (UUID: " << uuid
+ LOG(INFO) << "Received task status update acknowledgement (UUID: " << uuid
<< ") for task " << taskId
<< " of framework " << frameworkId;
- StatusUpdateStream* stream = getStatusUpdateStream(taskId, frameworkId);
+ TaskStatusUpdateStream* stream = getStatusUpdateStream(taskId, frameworkId);
// This might happen if we haven't completed recovery yet or if the
// acknowledgement is for a stream that has been cleaned up.
if (stream == nullptr) {
return Failure(
- "Cannot find the status update stream for task " + stringify(taskId) +
- " of framework " + stringify(frameworkId));
+ "Cannot find the task status update stream for task " +
+ stringify(taskId) + " of framework " + stringify(frameworkId));
}
// Get the corresponding update for this ACK.
@@ -416,9 +421,9 @@ Future<bool> StatusUpdateManagerProcess::acknowledgement(
// acknowledgments for both the original and the retried update.
if (update.isNone()) {
return Failure(
- "Unexpected status update acknowledgment (UUID: " + uuid.toString() +
- ") for task " + stringify(taskId) +
- " of framework " + stringify(frameworkId));
+ "Unexpected task status update acknowledgment (UUID: " +
+ uuid.toString() + ") for task " + stringify(taskId) + " of framework " +
+ stringify(frameworkId));
}
// Handle the acknowledgement.
@@ -430,7 +435,7 @@ Future<bool> StatusUpdateManagerProcess::acknowledgement(
}
if (!result.get()) {
- return Failure("Duplicate acknowledgement");
+ return Failure("Duplicate task status acknowledgement");
}
// Reset the timeout.
@@ -447,7 +452,7 @@ Future<bool> StatusUpdateManagerProcess::acknowledgement(
if (terminated) {
if (next.isSome()) {
LOG(WARNING) << "Acknowledged a terminal"
- << " status update " << update.get()
+ << " task status update " << update.get()
<< " but updates are still pending";
}
cleanupStatusUpdateStream(taskId, frameworkId);
@@ -461,7 +466,7 @@ Future<bool> StatusUpdateManagerProcess::acknowledgement(
// TODO(vinod): There should be a limit on the retries.
-void StatusUpdateManagerProcess::timeout(const Duration& duration)
+void TaskStatusUpdateManagerProcess::timeout(const Duration& duration)
{
if (paused) {
return;
@@ -469,13 +474,13 @@ void StatusUpdateManagerProcess::timeout(const Duration& duration)
// Check and see if we should resend any status updates.
foreachkey (const FrameworkID& frameworkId, streams) {
- foreachvalue (StatusUpdateStream* stream, streams[frameworkId]) {
+ foreachvalue (TaskStatusUpdateStream* stream, streams[frameworkId]) {
CHECK_NOTNULL(stream);
if (!stream->pending.empty()) {
CHECK_SOME(stream->timeout);
if (stream->timeout->expired()) {
const StatusUpdate& update = stream->pending.front();
- LOG(WARNING) << "Resending status update " << update;
+ LOG(WARNING) << "Resending task status update " << update;
// Bounded exponential backoff.
Duration duration_ =
@@ -489,7 +494,8 @@ void StatusUpdateManagerProcess::timeout(const Duration& duration)
}
-StatusUpdateStream* StatusUpdateManagerProcess::createStatusUpdateStream(
+TaskStatusUpdateStream*
+TaskStatusUpdateManagerProcess::createStatusUpdateStream(
const TaskID& taskId,
const FrameworkID& frameworkId,
const SlaveID& slaveId,
@@ -500,7 +506,7 @@ StatusUpdateStream* StatusUpdateManagerProcess::createStatusUpdateStream(
VLOG(1) << "Creating StatusUpdate stream for task " << taskId
<< " of framework " << frameworkId;
- StatusUpdateStream* stream = new StatusUpdateStream(
+ TaskStatusUpdateStream* stream = new TaskStatusUpdateStream(
taskId, frameworkId, slaveId, flags, checkpoint, executorId, containerId);
streams[frameworkId][taskId] = stream;
@@ -508,7 +514,7 @@ StatusUpdateStream* StatusUpdateManagerProcess::createStatusUpdateStream(
}
-StatusUpdateStream* StatusUpdateManagerProcess::getStatusUpdateStream(
+TaskStatusUpdateStream* TaskStatusUpdateManagerProcess::getStatusUpdateStream(
const TaskID& taskId,
const FrameworkID& frameworkId)
{
@@ -524,7 +530,7 @@ StatusUpdateStream* StatusUpdateManagerProcess::getStatusUpdateStream(
}
-void StatusUpdateManagerProcess::cleanupStatusUpdateStream(
+void TaskStatusUpdateManagerProcess::cleanupStatusUpdateStream(
const TaskID& taskId,
const FrameworkID& frameworkId)
{
@@ -533,12 +539,13 @@ void StatusUpdateManagerProcess::cleanupStatusUpdateStream(
<< " of framework " << frameworkId;
CHECK(streams.contains(frameworkId))
- << "Cannot find the status update streams for framework " << frameworkId;
+ << "Cannot find the task status update streams for framework "
+ << frameworkId;
CHECK(streams[frameworkId].contains(taskId))
<< "Cannot find the status update streams for task " << taskId;
- StatusUpdateStream* stream = streams[frameworkId][taskId];
+ TaskStatusUpdateStream* stream = streams[frameworkId][taskId];
streams[frameworkId].erase(taskId);
if (streams[frameworkId].empty()) {
@@ -549,14 +556,14 @@ void StatusUpdateManagerProcess::cleanupStatusUpdateStream(
}
-StatusUpdateManager::StatusUpdateManager(const Flags& flags)
+TaskStatusUpdateManager::TaskStatusUpdateManager(const Flags& flags)
{
- process = new StatusUpdateManagerProcess(flags);
+ process = new TaskStatusUpdateManagerProcess(flags);
spawn(process);
}
-StatusUpdateManager::~StatusUpdateManager()
+TaskStatusUpdateManager::~TaskStatusUpdateManager()
{
terminate(process);
wait(process);
@@ -564,14 +571,14 @@ StatusUpdateManager::~StatusUpdateManager()
}
-void StatusUpdateManager::initialize(
+void TaskStatusUpdateManager::initialize(
const function<void(StatusUpdate)>& forward)
{
- dispatch(process, &StatusUpdateManagerProcess::initialize, forward);
+ dispatch(process, &TaskStatusUpdateManagerProcess::initialize, forward);
}
-Future<Nothing> StatusUpdateManager::update(
+Future<Nothing> TaskStatusUpdateManager::update(
const StatusUpdate& update,
const SlaveID& slaveId,
const ExecutorID& executorId,
@@ -579,7 +586,7 @@ Future<Nothing> StatusUpdateManager::update(
{
return dispatch(
process,
- &StatusUpdateManagerProcess::update,
+ &TaskStatusUpdateManagerProcess::update,
update,
slaveId,
executorId,
@@ -587,60 +594,60 @@ Future<Nothing> StatusUpdateManager::update(
}
-Future<Nothing> StatusUpdateManager::update(
+Future<Nothing> TaskStatusUpdateManager::update(
const StatusUpdate& update,
const SlaveID& slaveId)
{
return dispatch(
process,
- &StatusUpdateManagerProcess::update,
+ &TaskStatusUpdateManagerProcess::update,
update,
slaveId);
}
-Future<bool> StatusUpdateManager::acknowledgement(
+Future<bool> TaskStatusUpdateManager::acknowledgement(
const TaskID& taskId,
const FrameworkID& frameworkId,
const UUID& uuid)
{
return dispatch(
process,
- &StatusUpdateManagerProcess::acknowledgement,
+ &TaskStatusUpdateManagerProcess::acknowledgement,
taskId,
frameworkId,
uuid);
}
-Future<Nothing> StatusUpdateManager::recover(
+Future<Nothing> TaskStatusUpdateManager::recover(
const string& rootDir,
const Option<SlaveState>& state)
{
return dispatch(
- process, &StatusUpdateManagerProcess::recover, rootDir, state);
+ process, &TaskStatusUpdateManagerProcess::recover, rootDir, state);
}
-void StatusUpdateManager::pause()
+void TaskStatusUpdateManager::pause()
{
- dispatch(process, &StatusUpdateManagerProcess::pause);
+ dispatch(process, &TaskStatusUpdateManagerProcess::pause);
}
-void StatusUpdateManager::resume()
+void TaskStatusUpdateManager::resume()
{
- dispatch(process, &StatusUpdateManagerProcess::resume);
+ dispatch(process, &TaskStatusUpdateManagerProcess::resume);
}
-void StatusUpdateManager::cleanup(const FrameworkID& frameworkId)
+void TaskStatusUpdateManager::cleanup(const FrameworkID& frameworkId)
{
- dispatch(process, &StatusUpdateManagerProcess::cleanup, frameworkId);
+ dispatch(process, &TaskStatusUpdateManagerProcess::cleanup, frameworkId);
}
-StatusUpdateStream::StatusUpdateStream(
+TaskStatusUpdateStream::TaskStatusUpdateStream(
const TaskID& _taskId,
const FrameworkID& _frameworkId,
const SlaveID& _slaveId,
@@ -699,7 +706,7 @@ StatusUpdateStream::StatusUpdateStream(
}
-StatusUpdateStream::~StatusUpdateStream()
+TaskStatusUpdateStream::~TaskStatusUpdateStream()
{
if (fd.isSome()) {
Try<Nothing> close = os::close(fd.get());
@@ -712,21 +719,21 @@ StatusUpdateStream::~StatusUpdateStream()
}
-Try<bool> StatusUpdateStream::update(const StatusUpdate& update)
+Try<bool> TaskStatusUpdateStream::update(const StatusUpdate& update)
{
if (error.isSome()) {
return Error(error.get());
}
if (!update.has_uuid()) {
- return Error("Status update is missing 'uuid'");
+ return Error("Task status update is missing 'uuid'");
}
// Check that this status update has not already been acknowledged.
// This could happen in the rare case when the slave received the ACK
// from the framework, died, but slave's ACK to the executor never made it!
if (acknowledged.contains(UUID::fromBytes(update.uuid()).get())) {
- LOG(WARNING) << "Ignoring status update " << update
+ LOG(WARNING) << "Ignoring task status update " << update
<< " that has already been acknowledged by the framework!";
return false;
}
@@ -735,7 +742,7 @@ Try<bool> StatusUpdateStream::update(const StatusUpdate& update)
// This could happen if the slave receives a status update from an executor,
// then crashes after it writes it to disk but before it sends an ack.
if (received.contains(UUID::fromBytes(update.uuid()).get())) {
- LOG(WARNING) << "Ignoring duplicate status update " << update;
+ LOG(WARNING) << "Ignoring duplicate task status update " << update;
return false;
}
@@ -749,7 +756,7 @@ Try<bool> StatusUpdateStream::update(const StatusUpdate& update)
}
-Try<bool> StatusUpdateStream::acknowledgement(
+Try<bool> TaskStatusUpdateStream::acknowledgement(
const TaskID& taskId,
const FrameworkID& frameworkId,
const UUID& uuid,
@@ -760,7 +767,7 @@ Try<bool> StatusUpdateStream::acknowledgement(
}
if (acknowledged.contains(uuid)) {
- LOG(WARNING) << "Duplicate status update acknowledgment (UUID: "
+ LOG(WARNING) << "Duplicate task status update acknowledgment (UUID: "
<< uuid << ") for update " << update;
return false;
}
@@ -768,7 +775,7 @@ Try<bool> StatusUpdateStream::acknowledgement(
// This might happen if we retried a status update and got back
// acknowledgments for both the original and the retried update.
if (uuid != UUID::fromBytes(update.uuid()).get()) {
- LOG(WARNING) << "Unexpected status update acknowledgement (received "
+ LOG(WARNING) << "Unexpected task status update acknowledgement (received "
<< uuid << ", expecting "
<< UUID::fromBytes(update.uuid()).get()
<< ") for update " << update;
@@ -785,7 +792,7 @@ Try<bool> StatusUpdateStream::acknowledgement(
}
-Result<StatusUpdate> StatusUpdateStream::next()
+Result<StatusUpdate> TaskStatusUpdateStream::next()
{
if (error.isSome()) {
return Error(error.get());
@@ -799,7 +806,7 @@ Result<StatusUpdate> StatusUpdateStream::next()
}
-Try<Nothing> StatusUpdateStream::replay(
+Try<Nothing> TaskStatusUpdateStream::replay(
const std::vector<StatusUpdate>& updates,
const hashset<UUID>& acks)
{
@@ -807,7 +814,7 @@ Try<Nothing> StatusUpdateStream::replay(
return Error(error.get());
}
- VLOG(1) << "Replaying status update stream for task " << taskId;
+ VLOG(1) << "Replaying task status update stream for task " << taskId;
foreach (const StatusUpdate& update, updates) {
// Handle the update.
@@ -823,7 +830,7 @@ Try<Nothing> StatusUpdateStream::replay(
}
-Try<Nothing> StatusUpdateStream::handle(
+Try<Nothing> TaskStatusUpdateStream::handle(
const StatusUpdate& update,
const StatusUpdateRecord::Type& type)
{
@@ -831,7 +838,8 @@ Try<Nothing> StatusUpdateStream::handle(
// Checkpoint the update if necessary.
if (checkpoint) {
- LOG(INFO) << "Checkpointing " << type << " for status update " << update;
+ LOG(INFO) << "Checkpointing " << type << " for task status update "
+ << update;
CHECK_SOME(fd);
@@ -846,7 +854,7 @@ Try<Nothing> StatusUpdateStream::handle(
Try<Nothing> write = ::protobuf::write(fd.get(), record);
if (write.isError()) {
- error = "Failed to write status update " + stringify(update) +
+ error = "Failed to write task status update " + stringify(update) +
" to '" + path.get() + "': " + write.error();
return Error(error.get());
}
@@ -859,7 +867,7 @@ Try<Nothing> StatusUpdateStream::handle(
}
-void StatusUpdateStream::_handle(
+void TaskStatusUpdateStream::_handle(
const StatusUpdate& update,
const StatusUpdateRecord::Type& type)
{
http://git-wip-us.apache.org/repos/asf/mesos/blob/2377cd8f/src/slave/status_update_manager.hpp
----------------------------------------------------------------------
diff --git a/src/slave/status_update_manager.hpp b/src/slave/status_update_manager.hpp
index 14ead42..4f7d45d 100644
--- a/src/slave/status_update_manager.hpp
+++ b/src/slave/status_update_manager.hpp
@@ -46,20 +46,20 @@ namespace state {
struct SlaveState;
}
-class StatusUpdateManagerProcess;
-struct StatusUpdateStream;
+class TaskStatusUpdateManagerProcess;
+struct TaskStatusUpdateStream;
-// StatusUpdateManager is responsible for
+// TaskStatusUpdateManager is responsible for
// 1) Reliably sending status updates to the master.
// 2) Checkpointing the update to disk (optional).
// 3) Sending ACKs to the executor (optional).
// 4) Receiving ACKs from the scheduler.
-class StatusUpdateManager
+class TaskStatusUpdateManager
{
public:
- StatusUpdateManager(const Flags& flags);
- virtual ~StatusUpdateManager();
+ TaskStatusUpdateManager(const Flags& flags);
+ virtual ~TaskStatusUpdateManager();
// Expects a callback 'forward' which gets called whenever there is
// a new status update that needs to be forwarded to the master.
@@ -119,19 +119,19 @@ public:
void cleanup(const FrameworkID& frameworkId);
private:
- StatusUpdateManagerProcess* process;
+ TaskStatusUpdateManagerProcess* process;
};
-// StatusUpdateStream handles the status updates and acknowledgements
+// TaskStatusUpdateStream handles the status updates and acknowledgements
// of a task, checkpointing them if necessary. It also holds the information
// about received, acknowledged and pending status updates.
// NOTE: A task is expected to have a globally unique ID across the lifetime
// of a framework. In other words the tuple (taskId, frameworkId) should be
// always unique.
-struct StatusUpdateStream
+struct TaskStatusUpdateStream
{
- StatusUpdateStream(const TaskID& _taskId,
+ TaskStatusUpdateStream(const TaskID& _taskId,
const FrameworkID& _frameworkId,
const SlaveID& _slaveId,
const Flags& _flags,
@@ -139,7 +139,7 @@ struct StatusUpdateStream
const Option<ExecutorID>& executorId,
const Option<ContainerID>& containerId);
- ~StatusUpdateStream();
+ ~TaskStatusUpdateStream();
// This function handles the update, checkpointing if necessary.
// @return True if the update is successfully handled.
http://git-wip-us.apache.org/repos/asf/mesos/blob/2377cd8f/src/tests/api_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/api_tests.cpp b/src/tests/api_tests.cpp
index c9c50b9..66cb059 100644
--- a/src/tests/api_tests.cpp
+++ b/src/tests/api_tests.cpp
@@ -2167,7 +2167,7 @@ TEST_P(MasterAPITest, Subscribe)
EXPECT_CALL(*scheduler, update(_, _))
.WillOnce(FutureSatisfy(&update2));
- // After we advance the clock, the status update manager would retry the
+ // After we advance the clock, the task status update manager would retry the
// `TASK_RUNNING` update. Since, the state of the task is not changed, this
// should not result in another `TASK_UPDATED` event.
Clock::pause();
http://git-wip-us.apache.org/repos/asf/mesos/blob/2377cd8f/src/tests/cluster.cpp
----------------------------------------------------------------------
diff --git a/src/tests/cluster.cpp b/src/tests/cluster.cpp
index d657da6..6111be4 100644
--- a/src/tests/cluster.cpp
+++ b/src/tests/cluster.cpp
@@ -396,7 +396,7 @@ Try<process::Owned<Slave>> Slave::start(
const Option<string>& id,
const Option<slave::Containerizer*>& containerizer,
const Option<slave::GarbageCollector*>& gc,
- const Option<slave::StatusUpdateManager*>& statusUpdateManager,
+ const Option<slave::TaskStatusUpdateManager*>& taskStatusUpdateManager,
const Option<mesos::slave::ResourceEstimator*>& resourceEstimator,
const Option<mesos::slave::QoSController*>& qosController,
const Option<Authorizer*>& providedAuthorizer)
@@ -503,9 +503,10 @@ Try<process::Owned<Slave>> Slave::start(
slave->qosController.reset(_qosController.get());
}
- // If the status update manager is not provided, create a default one.
- if (statusUpdateManager.isNone()) {
- slave->statusUpdateManager.reset(new slave::StatusUpdateManager(flags));
+ // If the task status update manager is not provided, create a default one.
+ if (taskStatusUpdateManager.isNone()) {
+ slave->taskStatusUpdateManager.reset(
+ new slave::TaskStatusUpdateManager(flags));
}
// Inject all the dependencies.
@@ -516,7 +517,7 @@ Try<process::Owned<Slave>> Slave::start(
slave->containerizer,
&slave->files,
gc.getOrElse(slave->gc.get()),
- statusUpdateManager.getOrElse(slave->statusUpdateManager.get()),
+ taskStatusUpdateManager.getOrElse(slave->taskStatusUpdateManager.get()),
resourceEstimator.getOrElse(slave->resourceEstimator.get()),
qosController.getOrElse(slave->qosController.get()),
authorizer));
http://git-wip-us.apache.org/repos/asf/mesos/blob/2377cd8f/src/tests/cluster.hpp
----------------------------------------------------------------------
diff --git a/src/tests/cluster.hpp b/src/tests/cluster.hpp
index 6563412..b3b1348 100644
--- a/src/tests/cluster.hpp
+++ b/src/tests/cluster.hpp
@@ -160,7 +160,8 @@ public:
const Option<std::string>& id = None(),
const Option<slave::Containerizer*>& containerizer = None(),
const Option<slave::GarbageCollector*>& gc = None(),
- const Option<slave::StatusUpdateManager*>& statusUpdateManager = None(),
+ const Option<slave::TaskStatusUpdateManager*>& taskStatusUpdateManager =
+ None(),
const Option<mesos::slave::ResourceEstimator*>& resourceEstimator =
None(),
const Option<mesos::slave::QoSController*>& qosController = None(),
@@ -220,7 +221,7 @@ private:
process::Owned<slave::GarbageCollector> gc;
process::Owned<mesos::slave::QoSController> qosController;
process::Owned<mesos::slave::ResourceEstimator> resourceEstimator;
- process::Owned<slave::StatusUpdateManager> statusUpdateManager;
+ process::Owned<slave::TaskStatusUpdateManager> taskStatusUpdateManager;
// Indicates whether or not authorization callbacks were set when this agent
// was constructed.
http://git-wip-us.apache.org/repos/asf/mesos/blob/2377cd8f/src/tests/master_slave_reconciliation_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/master_slave_reconciliation_tests.cpp b/src/tests/master_slave_reconciliation_tests.cpp
index 6acf694..d2501f5 100644
--- a/src/tests/master_slave_reconciliation_tests.cpp
+++ b/src/tests/master_slave_reconciliation_tests.cpp
@@ -152,8 +152,8 @@ TEST_F(MasterSlaveReconciliationTest, SlaveReregisterTerminatedExecutor)
.WillOnce(FutureArg<1>(&status2));
// We drop the 'UpdateFrameworkMessage' from the master to slave to
- // stop the status update manager from retrying the update that was
- // already sent due to the new master detection.
+ // stop the task status update manager from retrying the update that
+ // was already sent due to the new master detection.
DROP_PROTOBUFS(UpdateFrameworkMessage(), _, _);
detector.appoint(master.get()->pid);
@@ -623,8 +623,8 @@ TEST_F(MasterSlaveReconciliationTest, SlaveReregisterTerminalTask)
// re-registers. We check this by calling Clock::settle() so that
// the only update the scheduler receives is the retried
// TASK_FINISHED update.
- // NOTE: The status update manager resends the status update when
- // it detects a new master.
+ // NOTE: The task status update manager resends the status update
+ // when it detects a new master.
Clock::pause();
Clock::settle();
http://git-wip-us.apache.org/repos/asf/mesos/blob/2377cd8f/src/tests/master_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/master_tests.cpp b/src/tests/master_tests.cpp
index 9c450b9..139125b 100644
--- a/src/tests/master_tests.cpp
+++ b/src/tests/master_tests.cpp
@@ -4388,7 +4388,7 @@ TEST_F(MasterTest, ReleaseResourcesForTerminalTaskWithPendingUpdates)
// Wait until TASK_RUNNING is sent to the master.
AWAIT_READY(statusUpdateMessage);
- // Ensure status update manager handles TASK_RUNNING update.
+ // Ensure task status update manager handles TASK_RUNNING update.
AWAIT_READY(___statusUpdate);
Future<Nothing> ___statusUpdate2 =
@@ -4400,13 +4400,13 @@ TEST_F(MasterTest, ReleaseResourcesForTerminalTaskWithPendingUpdates)
finishedStatus.set_state(TASK_FINISHED);
execDriver->sendStatusUpdate(finishedStatus);
- // Ensure status update manager handles TASK_FINISHED update.
+ // Ensure task status update manager handles TASK_FINISHED update.
AWAIT_READY(___statusUpdate2);
Future<Nothing> recoverResources = FUTURE_DISPATCH(
_, &MesosAllocatorProcess::recoverResources);
- // Advance the clock so that the status update manager resends
+ // Advance the clock so that the task status update manager resends
// TASK_RUNNING update with 'latest_state' as TASK_FINISHED.
Clock::pause();
Clock::advance(slave::STATUS_UPDATE_RETRY_INTERVAL_MIN);
@@ -6758,7 +6758,7 @@ TEST_F(MasterTest, DISABLED_RecoverResourcesOrphanedTask)
Future<Nothing> recoverResources =
FUTURE_DISPATCH(_, &MesosAllocatorProcess::recoverResources);
- // Advance the clock for the status update manager to retry with the
+ // Advance the clock for the task status update manager to retry with the
// latest state of the task.
Clock::advance(slave::STATUS_UPDATE_RETRY_INTERVAL_MIN);
Clock::settle();
http://git-wip-us.apache.org/repos/asf/mesos/blob/2377cd8f/src/tests/mock_slave.cpp
----------------------------------------------------------------------
diff --git a/src/tests/mock_slave.cpp b/src/tests/mock_slave.cpp
index db24f9e..5dd2188 100644
--- a/src/tests/mock_slave.cpp
+++ b/src/tests/mock_slave.cpp
@@ -117,7 +117,7 @@ MockSlave::MockSlave(
containerizer,
&files,
&gc,
- statusUpdateManager = new slave::StatusUpdateManager(flags),
+ taskStatusUpdateManager = new slave::TaskStatusUpdateManager(flags),
&resourceEstimator,
_qosController.isSome() ? _qosController.get() : &qosController,
authorizer),
@@ -150,7 +150,7 @@ MockSlave::MockSlave(
MockSlave::~MockSlave()
{
- delete statusUpdateManager;
+ delete taskStatusUpdateManager;
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/2377cd8f/src/tests/mock_slave.hpp
----------------------------------------------------------------------
diff --git a/src/tests/mock_slave.hpp b/src/tests/mock_slave.hpp
index 57189ce..7c4d43a 100644
--- a/src/tests/mock_slave.hpp
+++ b/src/tests/mock_slave.hpp
@@ -216,7 +216,7 @@ private:
MockGarbageCollector gc;
MockResourceEstimator resourceEstimator;
MockQoSController qosController;
- slave::StatusUpdateManager* statusUpdateManager;
+ slave::TaskStatusUpdateManager* taskStatusUpdateManager;
// Set to the base class `secretGenerator` in `initialize()`. After
// `initialize()` has executed, this will be `None()`.
http://git-wip-us.apache.org/repos/asf/mesos/blob/2377cd8f/src/tests/reconciliation_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/reconciliation_tests.cpp b/src/tests/reconciliation_tests.cpp
index 8c43ffd..18551ba 100644
--- a/src/tests/reconciliation_tests.cpp
+++ b/src/tests/reconciliation_tests.cpp
@@ -1067,7 +1067,7 @@ TEST_F(ReconciliationTest, ReconcileStatusUpdateTaskState)
// Wait until TASK_RUNNING is sent to the master.
AWAIT_READY(statusUpdateMessage);
- // Ensure status update manager handles TASK_RUNNING update.
+ // Ensure task status update manager handles TASK_RUNNING update.
AWAIT_READY(___statusUpdate);
Future<Nothing> ___statusUpdate2 =
@@ -1079,7 +1079,7 @@ TEST_F(ReconciliationTest, ReconcileStatusUpdateTaskState)
finishedStatus.set_state(TASK_FINISHED);
execDriver->sendStatusUpdate(finishedStatus);
- // Ensure status update manager handles TASK_FINISHED update.
+ // Ensure task status update manager handles TASK_FINISHED update.
AWAIT_READY(___statusUpdate2);
EXPECT_CALL(sched, disconnected(&driver))
http://git-wip-us.apache.org/repos/asf/mesos/blob/2377cd8f/src/tests/slave_recovery_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/slave_recovery_tests.cpp b/src/tests/slave_recovery_tests.cpp
index c864aa9..f14c6ef 100644
--- a/src/tests/slave_recovery_tests.cpp
+++ b/src/tests/slave_recovery_tests.cpp
@@ -353,7 +353,7 @@ TYPED_TEST(SlaveRecoveryTest, RecoverSlaveState)
// The slave is killed before the update reaches the scheduler.
// When the slave comes back up it resends the unacknowledged update.
-TYPED_TEST(SlaveRecoveryTest, RecoverStatusUpdateManager)
+TYPED_TEST(SlaveRecoveryTest, RecoverTaskStatusUpdateManager)
{
Try<Owned<cluster::Master>> master = this->StartMaster();
ASSERT_SOME(master);
@@ -4601,8 +4601,9 @@ TYPED_TEST(SlaveRecoveryTest, RestartBeforeContainerizerLaunch)
slave.get()->terminate();
Future<TaskStatus> status;
+
// There is a race here where the Slave may reregister before we
- // shut down. If it does, it causes the StatusUpdateManager to
+ // shut down. If it does, it causes the TaskStatusUpdateManager to
// flush which will cause a duplicate status update to be sent. As
// such, we ignore any subsequent updates.
EXPECT_CALL(sched, statusUpdate(_, _))
http://git-wip-us.apache.org/repos/asf/mesos/blob/2377cd8f/src/tests/slave_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/slave_tests.cpp b/src/tests/slave_tests.cpp
index a75bb26..f5befc2 100644
--- a/src/tests/slave_tests.cpp
+++ b/src/tests/slave_tests.cpp
@@ -287,7 +287,7 @@ TEST_F(SlaveTest, DuplicateTerminalUpdateBeforeAck)
AWAIT_READY(statusUpdateAckMessage);
- // At this point the status update manager has enqueued
+ // At this point the task status update manager has enqueued
// TASK_FINISHED update.
AWAIT_READY(___statusUpdate);
@@ -4525,7 +4525,7 @@ TEST_F(SlaveTest, ReregisterWithStatusUpdateTaskState)
// Wait until TASK_RUNNING is sent to the master.
AWAIT_READY(statusUpdateMessage);
- // Ensure status update manager handles TASK_RUNNING update.
+ // Ensure task status update manager handles TASK_RUNNING update.
AWAIT_READY(___statusUpdate);
Future<Nothing> ___statusUpdate2 =
@@ -4537,7 +4537,7 @@ TEST_F(SlaveTest, ReregisterWithStatusUpdateTaskState)
finishedStatus.set_state(TASK_FINISHED);
execDriver->sendStatusUpdate(finishedStatus);
- // Ensure status update manager handles TASK_FINISHED update.
+ // Ensure task status update manager handles TASK_FINISHED update.
AWAIT_READY(___statusUpdate2);
Future<ReregisterSlaveMessage> reregisterSlaveMessage =
http://git-wip-us.apache.org/repos/asf/mesos/blob/2377cd8f/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 24d6a99..d180e8c 100644
--- a/src/tests/status_update_manager_tests.cpp
+++ b/src/tests/status_update_manager_tests.cpp
@@ -86,10 +86,11 @@ vector<TaskInfo> createTasks(const Offer& offer)
}
-class StatusUpdateManagerTest: public MesosTest {};
+class TaskStatusUpdateManagerTest: public MesosTest {};
-TEST_F_TEMP_DISABLED_ON_WINDOWS(StatusUpdateManagerTest, CheckpointStatusUpdate)
+TEST_F_TEMP_DISABLED_ON_WINDOWS(
+ TaskStatusUpdateManagerTest, CheckpointStatusUpdate)
{
Try<Owned<cluster::Master>> master = StartMaster();
ASSERT_SOME(master);
@@ -183,7 +184,7 @@ TEST_F_TEMP_DISABLED_ON_WINDOWS(StatusUpdateManagerTest, CheckpointStatusUpdate)
}
-TEST_F(StatusUpdateManagerTest, RetryStatusUpdate)
+TEST_F(TaskStatusUpdateManagerTest, RetryStatusUpdate)
{
Try<Owned<cluster::Master>> master = StartMaster();
ASSERT_SOME(master);
@@ -255,7 +256,7 @@ TEST_F(StatusUpdateManagerTest, RetryStatusUpdate)
// duplicate ACK for an earlier update when it is waiting
// for an ACK for a later update. This could happen when the
// duplicate ACK is for a retried update.
-TEST_F(StatusUpdateManagerTest, IgnoreDuplicateStatusUpdateAck)
+TEST_F(TaskStatusUpdateManagerTest, IgnoreDuplicateStatusUpdateAck)
{
Try<Owned<cluster::Master>> master = StartMaster();
ASSERT_SOME(master);
@@ -367,7 +368,7 @@ TEST_F(StatusUpdateManagerTest, IgnoreDuplicateStatusUpdateAck)
// unexpected ACK for an earlier update when it is waiting
// for an ACK for another update. We do this by dropping ACKs
// for the original update and sending a random ACK to the slave.
-TEST_F(StatusUpdateManagerTest, IgnoreUnexpectedStatusUpdateAck)
+TEST_F(TaskStatusUpdateManagerTest, IgnoreUnexpectedStatusUpdateAck)
{
Try<Owned<cluster::Master>> master = StartMaster();
ASSERT_SOME(master);
@@ -458,7 +459,7 @@ TEST_F(StatusUpdateManagerTest, IgnoreUnexpectedStatusUpdateAck)
// second update is received after the ACK for the first update.
// The proper behavior here is for the status update manager to
// forward the duplicate update to the scheduler.
-TEST_F(StatusUpdateManagerTest, DuplicateTerminalUpdateAfterAck)
+TEST_F(TaskStatusUpdateManagerTest, DuplicateTerminalUpdateAfterAck)
{
Try<Owned<cluster::Master>> master = StartMaster();
ASSERT_SOME(master);
@@ -556,7 +557,7 @@ TEST_F(StatusUpdateManagerTest, DuplicateTerminalUpdateAfterAck)
// update with the same UUID is received before the ACK for the
// first update. The proper behavior here is for the status update
// manager to drop the duplicate update.
-TEST_F(StatusUpdateManagerTest, DuplicateUpdateBeforeAck)
+TEST_F(TaskStatusUpdateManagerTest, DuplicateUpdateBeforeAck)
{
Try<Owned<cluster::Master>> master = StartMaster();
ASSERT_SOME(master);
@@ -656,7 +657,7 @@ TEST_F(StatusUpdateManagerTest, DuplicateUpdateBeforeAck)
// This test verifies that the status update manager correctly includes
// the latest state of the task in status update.
-TEST_F(StatusUpdateManagerTest, LatestTaskState)
+TEST_F(TaskStatusUpdateManagerTest, LatestTaskState)
{
Try<Owned<cluster::Master>> master = StartMaster();
ASSERT_SOME(master);
@@ -741,7 +742,7 @@ TEST_F(StatusUpdateManagerTest, LatestTaskState)
// This test verifies that if master receives a status update
// for an already terminated task it forwards it without
// changing the state of the task.
-TEST_F(StatusUpdateManagerTest, DuplicatedTerminalStatusUpdate)
+TEST_F(TaskStatusUpdateManagerTest, DuplicatedTerminalStatusUpdate)
{
Try<Owned<cluster::Master>> master = StartMaster();
ASSERT_SOME(master);
[2/6] mesos git commit: Added 'ReconcileOfferOperations' response to
the scheduler API.
Posted by gr...@apache.org.
Added 'ReconcileOfferOperations' response to the scheduler API.
This patch adds a first-class `Response` message to the scheduler API
along with its first member, the `ReconcileOfferOperations` response
message.
Review: https://reviews.apache.org/r/63768/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/f6359bd3
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/f6359bd3
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/f6359bd3
Branch: refs/heads/master
Commit: f6359bd3e660e6fa883bc5895ac8366715fb79f0
Parents: 056349f
Author: Greg Mann <gr...@mesosphere.io>
Authored: Mon Nov 20 16:42:28 2017 -0800
Committer: Greg Mann <gr...@gmail.com>
Committed: Mon Nov 20 16:49:10 2017 -0800
----------------------------------------------------------------------
include/mesos/scheduler/scheduler.proto | 19 +++++++++++++++++++
include/mesos/v1/scheduler/scheduler.proto | 19 +++++++++++++++++++
src/scheduler/scheduler.cpp | 5 ++---
3 files changed, 40 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/f6359bd3/include/mesos/scheduler/scheduler.proto
----------------------------------------------------------------------
diff --git a/include/mesos/scheduler/scheduler.proto b/include/mesos/scheduler/scheduler.proto
index 112f496..a7907e2 100644
--- a/include/mesos/scheduler/scheduler.proto
+++ b/include/mesos/scheduler/scheduler.proto
@@ -208,6 +208,25 @@ message Event {
/**
+ * Synchronous responses for calls made to the scheduler API.
+ */
+message Response {
+ // Each of the responses of type `FOO` corresponds to `Foo` message below.
+ enum Type {
+ UNKNOWN = 0;
+
+ RECONCILE_OFFER_OPERATIONS = 1; // See 'ReconcileOfferOperations' below.
+ }
+
+ message ReconcileOfferOperations {
+ repeated OfferOperationStatus offer_operation_statuses = 1;
+ }
+
+ optional ReconcileOfferOperations reconcile_offer_operations = 1;
+}
+
+
+/**
* Scheduler call API.
*
* Like Event, a Call is described using the standard protocol buffer
http://git-wip-us.apache.org/repos/asf/mesos/blob/f6359bd3/include/mesos/v1/scheduler/scheduler.proto
----------------------------------------------------------------------
diff --git a/include/mesos/v1/scheduler/scheduler.proto b/include/mesos/v1/scheduler/scheduler.proto
index 4fa6606..e115bca 100644
--- a/include/mesos/v1/scheduler/scheduler.proto
+++ b/include/mesos/v1/scheduler/scheduler.proto
@@ -206,6 +206,25 @@ message Event {
/**
+ * Synchronous responses for calls made to the scheduler API.
+ */
+message Response {
+ // Each of the responses of type `FOO` corresponds to `Foo` message below.
+ enum Type {
+ UNKNOWN = 0;
+
+ RECONCILE_OFFER_OPERATIONS = 1; // See 'ReconcileOfferOperations' below.
+ }
+
+ message ReconcileOfferOperations {
+ repeated OfferOperationStatus offer_operation_statuses = 1;
+ }
+
+ optional ReconcileOfferOperations reconcile_offer_operations = 1;
+}
+
+
+/**
* Scheduler call API.
*
* Like Event, a Call is described using the standard protocol buffer
http://git-wip-us.apache.org/repos/asf/mesos/blob/f6359bd3/src/scheduler/scheduler.cpp
----------------------------------------------------------------------
diff --git a/src/scheduler/scheduler.cpp b/src/scheduler/scheduler.cpp
index 286a760..7294f23 100644
--- a/src/scheduler/scheduler.cpp
+++ b/src/scheduler/scheduler.cpp
@@ -121,7 +121,6 @@ using process::http::Connection;
using process::http::Pipe;
using process::http::post;
using process::http::Request;
-using process::http::Response;
using process::http::URL;
using ::recordio::Decoder;
@@ -555,7 +554,7 @@ protected:
return;
}
- Future<Response> response;
+ Future<process::http::Response> response;
if (call.type() == Call::SUBSCRIBE) {
state = SUBSCRIBING;
@@ -581,7 +580,7 @@ protected:
void __send(
const UUID& _connectionId,
const Call& call,
- const Future<Response>& response)
+ const Future<process::http::Response>& response)
{
// It is possible that we detected a new master before a response could
// be received.
[5/6] mesos git commit: Added the "task" prefix to the name of the
status update manager files.
Posted by gr...@apache.org.
http://git-wip-us.apache.org/repos/asf/mesos/blob/62d11733/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
deleted file mode 100644
index d180e8c..0000000
--- a/src/tests/status_update_manager_tests.cpp
+++ /dev/null
@@ -1,852 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-#include <string>
-#include <vector>
-
-#include <gmock/gmock.h>
-
-#include <mesos/executor.hpp>
-#include <mesos/scheduler.hpp>
-
-#include <process/clock.hpp>
-#include <process/future.hpp>
-#include <process/gmock.hpp>
-#include <process/owned.hpp>
-#include <process/pid.hpp>
-
-#include <stout/none.hpp>
-#include <stout/result.hpp>
-#include <stout/try.hpp>
-
-#include "master/master.hpp"
-
-#include "slave/constants.hpp"
-#include "slave/paths.hpp"
-#include "slave/slave.hpp"
-#include "slave/state.hpp"
-
-#include "messages/messages.hpp"
-
-#include "tests/containerizer.hpp"
-#include "tests/mesos.hpp"
-
-using mesos::internal::master::Master;
-
-using mesos::internal::slave::Slave;
-
-using mesos::master::detector::MasterDetector;
-
-using process::Clock;
-using process::Future;
-using process::Owned;
-using process::PID;
-
-using std::string;
-using std::vector;
-
-using testing::_;
-using testing::AtMost;
-using testing::Return;
-using testing::SaveArg;
-
-namespace mesos {
-namespace internal {
-namespace tests {
-
-
-// TODO(benh): Move this into utils, make more generic, and use in
-// other tests.
-vector<TaskInfo> createTasks(const Offer& offer)
-{
- TaskInfo task;
- task.set_name("test-task");
- task.mutable_task_id()->set_value("1");
- task.mutable_slave_id()->MergeFrom(offer.slave_id());
- task.mutable_resources()->MergeFrom(offer.resources());
- task.mutable_executor()->MergeFrom(DEFAULT_EXECUTOR_INFO);
-
- vector<TaskInfo> tasks;
- tasks.push_back(task);
-
- return tasks;
-}
-
-
-class TaskStatusUpdateManagerTest: public MesosTest {};
-
-
-TEST_F_TEMP_DISABLED_ON_WINDOWS(
- TaskStatusUpdateManagerTest, CheckpointStatusUpdate)
-{
- Try<Owned<cluster::Master>> master = StartMaster();
- ASSERT_SOME(master);
-
- MockExecutor exec(DEFAULT_EXECUTOR_ID);
- TestContainerizer containerizer(&exec);
-
- // Require flags to retrieve work_dir when recovering
- // the checkpointed data.
- slave::Flags flags = CreateSlaveFlags();
-
- Owned<MasterDetector> detector = master.get()->createDetector();
-
- Try<Owned<cluster::Slave>> slave =
- StartSlave(detector.get(), &containerizer, flags);
- ASSERT_SOME(slave);
- FrameworkInfo frameworkInfo = DEFAULT_FRAMEWORK_INFO;
- frameworkInfo.set_checkpoint(true); // Enable checkpointing.
-
- MockScheduler sched;
- MesosSchedulerDriver driver(
- &sched, frameworkInfo, master.get()->pid, DEFAULT_CREDENTIAL);
-
- Future<FrameworkID> frameworkId;
- EXPECT_CALL(sched, registered(_, _, _))
- .WillOnce(FutureArg<1>(&frameworkId));
-
- Future<vector<Offer>> offers;
- EXPECT_CALL(sched, resourceOffers(_, _))
- .WillOnce(FutureArg<1>(&offers))
- .WillRepeatedly(Return()); // Ignore subsequent offers.
-
- driver.start();
-
- AWAIT_READY(frameworkId);
- AWAIT_READY(offers);
- ASSERT_FALSE(offers->empty());
-
- EXPECT_CALL(exec, registered(_, _, _, _));
-
- EXPECT_CALL(exec, launchTask(_, _))
- .WillOnce(SendStatusUpdateFromTask(TASK_RUNNING));
-
- Future<TaskStatus> status;
- EXPECT_CALL(sched, statusUpdate(_, _))
- .WillOnce(FutureArg<1>(&status));
-
- Future<Nothing> _statusUpdateAcknowledgement =
- FUTURE_DISPATCH(slave.get()->pid, &Slave::_statusUpdateAcknowledgement);
-
- driver.launchTasks(offers.get()[0].id(), createTasks(offers.get()[0]));
-
- AWAIT_READY(status);
- EXPECT_EQ(TASK_RUNNING, status->state());
-
- AWAIT_READY(_statusUpdateAcknowledgement);
-
- // Ensure that both the status update and its acknowledgement are
- // correctly checkpointed.
- Result<slave::state::State> state =
- slave::state::recover(slave::paths::getMetaRootDir(flags.work_dir), true);
-
- ASSERT_SOME(state);
- ASSERT_SOME(state->slave);
- ASSERT_TRUE(state->slave->frameworks.contains(frameworkId.get()));
-
- slave::state::FrameworkState frameworkState =
- state->slave->frameworks.get(frameworkId.get()).get();
-
- ASSERT_EQ(1u, frameworkState.executors.size());
-
- slave::state::ExecutorState executorState =
- frameworkState.executors.begin()->second;
-
- ASSERT_EQ(1u, executorState.runs.size());
-
- slave::state::RunState runState = executorState.runs.begin()->second;
-
- ASSERT_EQ(1u, runState.tasks.size());
-
- slave::state::TaskState taskState = runState.tasks.begin()->second;
-
- EXPECT_EQ(1u, taskState.updates.size());
- EXPECT_EQ(1u, taskState.acks.size());
-
- EXPECT_CALL(exec, shutdown(_))
- .Times(AtMost(1));
-
- driver.stop();
- driver.join();
-}
-
-
-TEST_F(TaskStatusUpdateManagerTest, RetryStatusUpdate)
-{
- Try<Owned<cluster::Master>> master = StartMaster();
- ASSERT_SOME(master);
-
- MockExecutor exec(DEFAULT_EXECUTOR_ID);
- TestContainerizer containerizer(&exec);
-
- slave::Flags flags = CreateSlaveFlags();
-
- Owned<MasterDetector> detector = master.get()->createDetector();
-
- Try<Owned<cluster::Slave>> slave =
- StartSlave(detector.get(), &containerizer, flags);
- ASSERT_SOME(slave);
- FrameworkInfo frameworkInfo = DEFAULT_FRAMEWORK_INFO;
- frameworkInfo.set_checkpoint(true); // Enable checkpointing.
-
- MockScheduler sched;
- MesosSchedulerDriver driver(
- &sched, frameworkInfo, master.get()->pid, DEFAULT_CREDENTIAL);
-
- EXPECT_CALL(sched, registered(_, _, _));
-
- Future<vector<Offer>> offers;
- EXPECT_CALL(sched, resourceOffers(_, _))
- .WillOnce(FutureArg<1>(&offers))
- .WillRepeatedly(Return()); // Ignore subsequent offers.
-
- driver.start();
-
- AWAIT_READY(offers);
- ASSERT_FALSE(offers->empty());
-
- EXPECT_CALL(exec, registered(_, _, _, _));
-
- EXPECT_CALL(exec, launchTask(_, _))
- .WillOnce(SendStatusUpdateFromTask(TASK_RUNNING));
-
- Future<StatusUpdateMessage> statusUpdateMessage =
- DROP_PROTOBUF(StatusUpdateMessage(), master.get()->pid, _);
-
- Clock::pause();
-
- driver.launchTasks(offers.get()[0].id(), createTasks(offers.get()[0]));
-
- AWAIT_READY(statusUpdateMessage);
-
- Future<TaskStatus> status;
- EXPECT_CALL(sched, statusUpdate(_, _))
- .WillOnce(FutureArg<1>(&status));
-
- Clock::advance(slave::STATUS_UPDATE_RETRY_INTERVAL_MIN);
-
- AWAIT_READY(status);
-
- EXPECT_EQ(TASK_RUNNING, status->state());
-
- Clock::resume();
-
- EXPECT_CALL(exec, shutdown(_))
- .Times(AtMost(1));
-
- driver.stop();
- driver.join();
-}
-
-
-// This test verifies that status update manager ignores
-// duplicate ACK for an earlier update when it is waiting
-// for an ACK for a later update. This could happen when the
-// duplicate ACK is for a retried update.
-TEST_F(TaskStatusUpdateManagerTest, IgnoreDuplicateStatusUpdateAck)
-{
- Try<Owned<cluster::Master>> master = StartMaster();
- ASSERT_SOME(master);
-
- MockExecutor exec(DEFAULT_EXECUTOR_ID);
- TestContainerizer containerizer(&exec);
-
- Owned<MasterDetector> detector = master.get()->createDetector();
- Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), &containerizer);
- ASSERT_SOME(slave);
-
- FrameworkInfo frameworkInfo = DEFAULT_FRAMEWORK_INFO;
- frameworkInfo.set_checkpoint(true); // Enable checkpointing.
-
- MockScheduler sched;
- MesosSchedulerDriver driver(
- &sched, frameworkInfo, master.get()->pid, DEFAULT_CREDENTIAL);
-
- FrameworkID frameworkId;
- EXPECT_CALL(sched, registered(_, _, _))
- .WillOnce(SaveArg<1>(&frameworkId));
-
- Future<vector<Offer>> offers;
- EXPECT_CALL(sched, resourceOffers(_, _))
- .WillOnce(FutureArg<1>(&offers))
- .WillRepeatedly(Return()); // Ignore subsequent offers.
-
- driver.start();
-
- AWAIT_READY(offers);
- ASSERT_FALSE(offers->empty());
-
- ExecutorDriver* execDriver;
- EXPECT_CALL(exec, registered(_, _, _, _))
- .WillOnce(SaveArg<0>(&execDriver));
-
- EXPECT_CALL(exec, launchTask(_, _))
- .WillOnce(SendStatusUpdateFromTask(TASK_RUNNING));
-
- // Drop the first update, so that status update manager
- // resends the update.
- Future<StatusUpdateMessage> statusUpdateMessage =
- DROP_PROTOBUF(StatusUpdateMessage(), master.get()->pid, _);
-
- Clock::pause();
-
- driver.launchTasks(offers.get()[0].id(), createTasks(offers.get()[0]));
-
- AWAIT_READY(statusUpdateMessage);
- StatusUpdate update = statusUpdateMessage->update();
-
- Future<TaskStatus> status;
- EXPECT_CALL(sched, statusUpdate(_, _))
- .WillOnce(FutureArg<1>(&status));
-
- // This is the ACK for the retried update.
- Future<Nothing> ack =
- FUTURE_DISPATCH(_, &Slave::_statusUpdateAcknowledgement);
-
- Clock::advance(slave::STATUS_UPDATE_RETRY_INTERVAL_MIN);
-
- AWAIT_READY(status);
-
- EXPECT_EQ(TASK_RUNNING, status->state());
-
- AWAIT_READY(ack);
-
- // Now send TASK_FINISHED update so that the status update manager
- // is waiting for its ACK, which it never gets because we drop the
- // update.
- DROP_PROTOBUFS(StatusUpdateMessage(), master.get()->pid, _);
-
- Future<Nothing> update2 = FUTURE_DISPATCH(_, &Slave::_statusUpdate);
-
- TaskStatus status2 = status.get();
- status2.set_state(TASK_FINISHED);
-
- execDriver->sendStatusUpdate(status2);
-
- AWAIT_READY(update2);
-
- // This is to catch the duplicate ack for TASK_RUNNING.
- Future<Nothing> duplicateAck =
- FUTURE_DISPATCH(_, &Slave::_statusUpdateAcknowledgement);
-
- // Now send a duplicate ACK for the TASK_RUNNING update.
- process::dispatch(
- slave.get()->pid,
- &Slave::statusUpdateAcknowledgement,
- master.get()->pid,
- update.slave_id(),
- frameworkId,
- update.status().task_id(),
- update.uuid());
-
- AWAIT_READY(duplicateAck);
-
- Clock::resume();
-
- EXPECT_CALL(exec, shutdown(_))
- .Times(AtMost(1));
-
- driver.stop();
- driver.join();
-}
-
-
-// This test verifies that status update manager ignores
-// unexpected ACK for an earlier update when it is waiting
-// for an ACK for another update. We do this by dropping ACKs
-// for the original update and sending a random ACK to the slave.
-TEST_F(TaskStatusUpdateManagerTest, IgnoreUnexpectedStatusUpdateAck)
-{
- Try<Owned<cluster::Master>> master = StartMaster();
- ASSERT_SOME(master);
-
- MockExecutor exec(DEFAULT_EXECUTOR_ID);
- TestContainerizer containerizer(&exec);
-
- Owned<MasterDetector> detector = master.get()->createDetector();
- Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), &containerizer);
- ASSERT_SOME(slave);
-
- FrameworkInfo frameworkInfo = DEFAULT_FRAMEWORK_INFO;
- frameworkInfo.set_checkpoint(true); // Enable checkpointing.
-
- MockScheduler sched;
- MesosSchedulerDriver driver(
- &sched, frameworkInfo, master.get()->pid, DEFAULT_CREDENTIAL);
-
- FrameworkID frameworkId;
- EXPECT_CALL(sched, registered(_, _, _))
- .WillOnce(SaveArg<1>(&frameworkId));
-
- Future<vector<Offer>> offers;
- EXPECT_CALL(sched, resourceOffers(_, _))
- .WillOnce(FutureArg<1>(&offers))
- .WillRepeatedly(Return()); // Ignore subsequent offers.
-
- Future<TaskStatus> status;
- EXPECT_CALL(sched, statusUpdate(_, _))
- .WillOnce(FutureArg<1>(&status));
-
- driver.start();
-
- AWAIT_READY(offers);
- ASSERT_FALSE(offers->empty());
-
- ExecutorDriver* execDriver;
- EXPECT_CALL(exec, registered(_, _, _, _))
- .WillOnce(SaveArg<0>(&execDriver));
-
- EXPECT_CALL(exec, launchTask(_, _))
- .WillOnce(SendStatusUpdateFromTask(TASK_RUNNING));
-
- Future<StatusUpdateMessage> statusUpdateMessage =
- FUTURE_PROTOBUF(StatusUpdateMessage(), master.get()->pid, _);
-
- // Drop the ACKs, so that status update manager
- // retries the update.
- DROP_CALLS(mesos::scheduler::Call(),
- mesos::scheduler::Call::ACKNOWLEDGE,
- _,
- master.get()->pid);
-
- driver.launchTasks(offers.get()[0].id(), createTasks(offers.get()[0]));
-
- AWAIT_READY(statusUpdateMessage);
- StatusUpdate update = statusUpdateMessage->update();
-
- AWAIT_READY(status);
-
- EXPECT_EQ(TASK_RUNNING, status->state());
-
- Future<Nothing> unexpectedAck =
- FUTURE_DISPATCH(_, &Slave::_statusUpdateAcknowledgement);
-
- // Now send an ACK with a random UUID.
- process::dispatch(
- slave.get()->pid,
- &Slave::statusUpdateAcknowledgement,
- master.get()->pid,
- update.slave_id(),
- frameworkId,
- update.status().task_id(),
- UUID::random().toBytes());
-
- AWAIT_READY(unexpectedAck);
-
- EXPECT_CALL(exec, shutdown(_))
- .Times(AtMost(1));
-
- driver.stop();
- driver.join();
-}
-
-
-// This test verifies that the slave and status update manager
-// properly handle duplicate terminal status updates, when the
-// second update is received after the ACK for the first update.
-// The proper behavior here is for the status update manager to
-// forward the duplicate update to the scheduler.
-TEST_F(TaskStatusUpdateManagerTest, DuplicateTerminalUpdateAfterAck)
-{
- Try<Owned<cluster::Master>> master = StartMaster();
- ASSERT_SOME(master);
-
- MockExecutor exec(DEFAULT_EXECUTOR_ID);
- TestContainerizer containerizer(&exec);
-
- slave::Flags flags = CreateSlaveFlags();
-
- Owned<MasterDetector> detector = master.get()->createDetector();
-
- Try<Owned<cluster::Slave>> slave =
- StartSlave(detector.get(), &containerizer, flags);
- ASSERT_SOME(slave);
- FrameworkInfo frameworkInfo = DEFAULT_FRAMEWORK_INFO;
- frameworkInfo.set_checkpoint(true); // Enable checkpointing.
-
- MockScheduler sched;
- MesosSchedulerDriver driver(
- &sched, frameworkInfo, master.get()->pid, DEFAULT_CREDENTIAL);
-
- FrameworkID frameworkId;
- EXPECT_CALL(sched, registered(_, _, _))
- .WillOnce(SaveArg<1>(&frameworkId));
-
- Future<vector<Offer>> offers;
- EXPECT_CALL(sched, resourceOffers(_, _))
- .WillOnce(FutureArg<1>(&offers))
- .WillRepeatedly(Return()); // Ignore subsequent offers.
-
- driver.start();
-
- AWAIT_READY(offers);
- ASSERT_FALSE(offers->empty());
-
- ExecutorDriver* execDriver;
- EXPECT_CALL(exec, registered(_, _, _, _))
- .WillOnce(SaveArg<0>(&execDriver));
-
- // Send a terminal update right away.
- EXPECT_CALL(exec, launchTask(_, _))
- .WillOnce(SendStatusUpdateFromTask(TASK_FINISHED));
-
- Future<TaskStatus> status;
- EXPECT_CALL(sched, statusUpdate(_, _))
- .WillOnce(FutureArg<1>(&status));
-
- Future<Nothing> _statusUpdateAcknowledgement =
- FUTURE_DISPATCH(slave.get()->pid, &Slave::_statusUpdateAcknowledgement);
-
- driver.launchTasks(offers.get()[0].id(), createTasks(offers.get()[0]));
-
- AWAIT_READY(status);
-
- EXPECT_EQ(TASK_FINISHED, status->state());
-
- AWAIT_READY(_statusUpdateAcknowledgement);
-
- Future<TaskStatus> update;
- EXPECT_CALL(sched, statusUpdate(_, _))
- .WillOnce(FutureArg<1>(&update));
-
- Future<Nothing> _statusUpdateAcknowledgement2 =
- FUTURE_DISPATCH(slave.get()->pid, &Slave::_statusUpdateAcknowledgement);
-
- Clock::pause();
-
- // Now send a TASK_KILLED update for the same task.
- TaskStatus status2 = status.get();
- status2.set_state(TASK_KILLED);
- execDriver->sendStatusUpdate(status2);
-
- // Ensure the scheduler receives TASK_KILLED.
- AWAIT_READY(update);
- EXPECT_EQ(TASK_KILLED, update->state());
-
- // Ensure the slave properly handles the ACK.
- // Clock::settle() ensures that the slave successfully
- // executes Slave::_statusUpdateAcknowledgement().
- AWAIT_READY(_statusUpdateAcknowledgement2);
- Clock::settle();
-
- Clock::resume();
-
- EXPECT_CALL(exec, shutdown(_))
- .Times(AtMost(1));
-
- driver.stop();
- driver.join();
-}
-
-
-// This test verifies that the slave and status update manager
-// properly handle duplicate status updates, when the second
-// update with the same UUID is received before the ACK for the
-// first update. The proper behavior here is for the status update
-// manager to drop the duplicate update.
-TEST_F(TaskStatusUpdateManagerTest, DuplicateUpdateBeforeAck)
-{
- Try<Owned<cluster::Master>> master = StartMaster();
- ASSERT_SOME(master);
-
- MockExecutor exec(DEFAULT_EXECUTOR_ID);
- TestContainerizer containerizer(&exec);
-
- Owned<MasterDetector> detector = master.get()->createDetector();
- Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), &containerizer);
- ASSERT_SOME(slave);
-
- FrameworkInfo frameworkInfo = DEFAULT_FRAMEWORK_INFO;
- frameworkInfo.set_checkpoint(true); // Enable checkpointing.
-
- MockScheduler sched;
- MesosSchedulerDriver driver(
- &sched, frameworkInfo, master.get()->pid, DEFAULT_CREDENTIAL);
-
- FrameworkID frameworkId;
- EXPECT_CALL(sched, registered(_, _, _))
- .WillOnce(SaveArg<1>(&frameworkId));
-
- Future<vector<Offer>> offers;
- EXPECT_CALL(sched, resourceOffers(_, _))
- .WillOnce(FutureArg<1>(&offers))
- .WillRepeatedly(Return()); // Ignore subsequent offers.
-
- driver.start();
-
- AWAIT_READY(offers);
- ASSERT_FALSE(offers->empty());
-
- ExecutorDriver* execDriver;
- EXPECT_CALL(exec, registered(_, _, _, _))
- .WillOnce(SaveArg<0>(&execDriver));
-
- EXPECT_CALL(exec, launchTask(_, _))
- .WillOnce(SendStatusUpdateFromTask(TASK_RUNNING));
-
- // Capture the first status update message.
- Future<StatusUpdateMessage> statusUpdateMessage =
- FUTURE_PROTOBUF(StatusUpdateMessage(), _, _);
-
- Future<TaskStatus> status;
- EXPECT_CALL(sched, statusUpdate(_, _))
- .WillOnce(FutureArg<1>(&status));
-
- // Drop the first ACK from the scheduler to the slave.
- Future<StatusUpdateAcknowledgementMessage> statusUpdateAckMessage =
- DROP_PROTOBUF(StatusUpdateAcknowledgementMessage(), _, slave.get()->pid);
-
- Clock::pause();
-
- driver.launchTasks(offers.get()[0].id(), createTasks(offers.get()[0]));
-
- AWAIT_READY(statusUpdateMessage);
-
- AWAIT_READY(status);
-
- EXPECT_EQ(TASK_RUNNING, status->state());
-
- AWAIT_READY(statusUpdateAckMessage);
-
- Future<Nothing> ___statusUpdate =
- FUTURE_DISPATCH(slave.get()->pid, &Slave::___statusUpdate);
-
- // Now resend the TASK_RUNNING update.
- process::post(slave.get()->pid, statusUpdateMessage.get());
-
- // At this point the status update manager has handled
- // the duplicate status update.
- AWAIT_READY(___statusUpdate);
-
- // After we advance the clock, the status update manager should
- // retry the TASK_RUNNING update and the scheduler should receive
- // and acknowledge it.
- Future<TaskStatus> update;
- EXPECT_CALL(sched, statusUpdate(_, _))
- .WillOnce(FutureArg<1>(&update));
-
- Clock::advance(slave::STATUS_UPDATE_RETRY_INTERVAL_MIN);
- Clock::settle();
-
- // Ensure the scheduler receives TASK_FINISHED.
- AWAIT_READY(update);
- EXPECT_EQ(TASK_RUNNING, update->state());
-
- EXPECT_CALL(exec, shutdown(_))
- .Times(AtMost(1));
-
- Clock::resume();
-
- driver.stop();
- driver.join();
-}
-
-
-// This test verifies that the status update manager correctly includes
-// the latest state of the task in status update.
-TEST_F(TaskStatusUpdateManagerTest, LatestTaskState)
-{
- Try<Owned<cluster::Master>> master = StartMaster();
- ASSERT_SOME(master);
-
- MockExecutor exec(DEFAULT_EXECUTOR_ID);
- TestContainerizer containerizer(&exec);
-
- Owned<MasterDetector> detector = master.get()->createDetector();
- Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), &containerizer);
- ASSERT_SOME(slave);
-
- MockScheduler sched;
- MesosSchedulerDriver driver(
- &sched, DEFAULT_FRAMEWORK_INFO, master.get()->pid, DEFAULT_CREDENTIAL);
-
- EXPECT_CALL(sched, registered(_, _, _));
-
- EXPECT_CALL(sched, resourceOffers(_, _))
- .WillOnce(LaunchTasks(DEFAULT_EXECUTOR_INFO, 1, 1, 512, "*"))
- .WillRepeatedly(Return()); // Ignore subsequent offers.
-
- ExecutorDriver* execDriver;
- EXPECT_CALL(exec, registered(_, _, _, _))
- .WillOnce(SaveArg<0>(&execDriver));
-
- EXPECT_CALL(exec, launchTask(_, _))
- .WillOnce(SendStatusUpdateFromTask(TASK_RUNNING));
-
- // Signal when the first update is dropped.
- Future<StatusUpdateMessage> statusUpdateMessage =
- DROP_PROTOBUF(StatusUpdateMessage(), _, master.get()->pid);
-
- Future<Nothing> ___statusUpdate = FUTURE_DISPATCH(_, &Slave::___statusUpdate);
-
- driver.start();
-
- // Wait until TASK_RUNNING is sent to the master.
- AWAIT_READY(statusUpdateMessage);
-
- // Ensure the status update manager handles the TASK_RUNNING update.
- AWAIT_READY(___statusUpdate);
-
- // Pause the clock to avoid status update manager from retrying.
- Clock::pause();
-
- Future<Nothing> ___statusUpdate2 =
- FUTURE_DISPATCH(_, &Slave::___statusUpdate);
-
- // Now send TASK_FINISHED update.
- TaskStatus finishedStatus;
- finishedStatus = statusUpdateMessage->update().status();
- finishedStatus.set_state(TASK_FINISHED);
- execDriver->sendStatusUpdate(finishedStatus);
-
- // Ensure the status update manager handles the TASK_FINISHED update.
- AWAIT_READY(___statusUpdate2);
-
- // Signal when the second update is dropped.
- Future<StatusUpdateMessage> statusUpdateMessage2 =
- DROP_PROTOBUF(StatusUpdateMessage(), _, master.get()->pid);
-
- // Advance the clock for the status update manager to send a retry.
- Clock::advance(slave::STATUS_UPDATE_RETRY_INTERVAL_MIN);
-
- AWAIT_READY(statusUpdateMessage2);
-
- // The update should correspond to TASK_RUNNING.
- ASSERT_EQ(TASK_RUNNING, statusUpdateMessage2->update().status().state());
-
- // The update should include TASK_FINISHED as the latest state.
- ASSERT_EQ(TASK_FINISHED,
- statusUpdateMessage2->update().latest_state());
-
- EXPECT_CALL(exec, shutdown(_))
- .Times(AtMost(1));
-
- driver.stop();
- driver.join();
-}
-
-
-// This test verifies that if master receives a status update
-// for an already terminated task it forwards it without
-// changing the state of the task.
-TEST_F(TaskStatusUpdateManagerTest, DuplicatedTerminalStatusUpdate)
-{
- Try<Owned<cluster::Master>> master = StartMaster();
- ASSERT_SOME(master);
-
- MockExecutor exec(DEFAULT_EXECUTOR_ID);
- TestContainerizer containerizer(&exec);
-
- Owned<MasterDetector> detector = master.get()->createDetector();
- Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), &containerizer);
- ASSERT_SOME(slave);
-
- FrameworkInfo frameworkInfo = DEFAULT_FRAMEWORK_INFO;
- frameworkInfo.set_checkpoint(true); // Enable checkpointing.
-
- MockScheduler sched;
- MesosSchedulerDriver driver(
- &sched, frameworkInfo, master.get()->pid, DEFAULT_CREDENTIAL);
-
- FrameworkID frameworkId;
- EXPECT_CALL(sched, registered(_, _, _))
- .WillOnce(SaveArg<1>(&frameworkId));
-
- Future<vector<Offer>> offers;
- EXPECT_CALL(sched, resourceOffers(_, _))
- .WillOnce(FutureArg<1>(&offers))
- .WillRepeatedly(Return()); // Ignore subsequent offers.
-
- driver.start();
-
- AWAIT_READY(offers);
- ASSERT_FALSE(offers->empty());
-
- ExecutorDriver* execDriver;
- EXPECT_CALL(exec, registered(_, _, _, _))
- .WillOnce(SaveArg<0>(&execDriver));
-
- // Send a terminal update right away.
- EXPECT_CALL(exec, launchTask(_, _))
- .WillOnce(SendStatusUpdateFromTask(TASK_FINISHED));
-
- Future<TaskStatus> status;
- EXPECT_CALL(sched, statusUpdate(_, _))
- .WillOnce(FutureArg<1>(&status));
-
- Future<Nothing> _statusUpdateAcknowledgement =
- FUTURE_DISPATCH(slave.get()->pid, &Slave::_statusUpdateAcknowledgement);
-
- driver.launchTasks(offers.get()[0].id(), createTasks(offers.get()[0]));
-
- AWAIT_READY(status);
-
- EXPECT_EQ(TASK_FINISHED, status->state());
-
- AWAIT_READY(_statusUpdateAcknowledgement);
-
- Future<TaskStatus> update;
- EXPECT_CALL(sched, statusUpdate(_, _))
- .WillOnce(FutureArg<1>(&update));
-
- Future<Nothing> _statusUpdateAcknowledgement2 =
- FUTURE_DISPATCH(slave.get()->pid, &Slave::_statusUpdateAcknowledgement);
-
- Clock::pause();
-
- // Now send a TASK_KILLED update for the same task.
- TaskStatus status2 = status.get();
- status2.set_state(TASK_KILLED);
- execDriver->sendStatusUpdate(status2);
-
- // Ensure the scheduler receives TASK_KILLED.
- AWAIT_READY(update);
- EXPECT_EQ(TASK_KILLED, update->state());
-
- // Ensure the slave properly handles the ACK.
- // Clock::settle() ensures that the slave successfully
- // executes Slave::_statusUpdateAcknowledgement().
- AWAIT_READY(_statusUpdateAcknowledgement2);
-
- // Verify the latest task status.
- Future<process::http::Response> tasks = process::http::get(
- master.get()->pid,
- "tasks",
- None(),
- createBasicAuthHeaders(DEFAULT_CREDENTIAL));
-
- AWAIT_EXPECT_RESPONSE_STATUS_EQ(process::http::OK().status, tasks);
- AWAIT_EXPECT_RESPONSE_HEADER_EQ(APPLICATION_JSON, "Content-Type", tasks);
-
- Try<JSON::Object> parse = JSON::parse<JSON::Object>(tasks->body);
- ASSERT_SOME(parse);
-
- Result<JSON::String> state = parse->find<JSON::String>("tasks[0].state");
-
- ASSERT_SOME_EQ(JSON::String("TASK_FINISHED"), state);
-
- Clock::resume();
-
- EXPECT_CALL(exec, shutdown(_))
- .Times(AtMost(1));
-
- driver.stop();
- driver.join();
-}
-
-} // namespace tests {
-} // namespace internal {
-} // namespace mesos {
http://git-wip-us.apache.org/repos/asf/mesos/blob/62d11733/src/tests/task_status_update_manager_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/task_status_update_manager_tests.cpp b/src/tests/task_status_update_manager_tests.cpp
new file mode 100644
index 0000000..d180e8c
--- /dev/null
+++ b/src/tests/task_status_update_manager_tests.cpp
@@ -0,0 +1,852 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include <string>
+#include <vector>
+
+#include <gmock/gmock.h>
+
+#include <mesos/executor.hpp>
+#include <mesos/scheduler.hpp>
+
+#include <process/clock.hpp>
+#include <process/future.hpp>
+#include <process/gmock.hpp>
+#include <process/owned.hpp>
+#include <process/pid.hpp>
+
+#include <stout/none.hpp>
+#include <stout/result.hpp>
+#include <stout/try.hpp>
+
+#include "master/master.hpp"
+
+#include "slave/constants.hpp"
+#include "slave/paths.hpp"
+#include "slave/slave.hpp"
+#include "slave/state.hpp"
+
+#include "messages/messages.hpp"
+
+#include "tests/containerizer.hpp"
+#include "tests/mesos.hpp"
+
+using mesos::internal::master::Master;
+
+using mesos::internal::slave::Slave;
+
+using mesos::master::detector::MasterDetector;
+
+using process::Clock;
+using process::Future;
+using process::Owned;
+using process::PID;
+
+using std::string;
+using std::vector;
+
+using testing::_;
+using testing::AtMost;
+using testing::Return;
+using testing::SaveArg;
+
+namespace mesos {
+namespace internal {
+namespace tests {
+
+
+// TODO(benh): Move this into utils, make more generic, and use in
+// other tests.
+vector<TaskInfo> createTasks(const Offer& offer)
+{
+ TaskInfo task;
+ task.set_name("test-task");
+ task.mutable_task_id()->set_value("1");
+ task.mutable_slave_id()->MergeFrom(offer.slave_id());
+ task.mutable_resources()->MergeFrom(offer.resources());
+ task.mutable_executor()->MergeFrom(DEFAULT_EXECUTOR_INFO);
+
+ vector<TaskInfo> tasks;
+ tasks.push_back(task);
+
+ return tasks;
+}
+
+
+class TaskStatusUpdateManagerTest: public MesosTest {};
+
+
+TEST_F_TEMP_DISABLED_ON_WINDOWS(
+ TaskStatusUpdateManagerTest, CheckpointStatusUpdate)
+{
+ Try<Owned<cluster::Master>> master = StartMaster();
+ ASSERT_SOME(master);
+
+ MockExecutor exec(DEFAULT_EXECUTOR_ID);
+ TestContainerizer containerizer(&exec);
+
+ // Require flags to retrieve work_dir when recovering
+ // the checkpointed data.
+ slave::Flags flags = CreateSlaveFlags();
+
+ Owned<MasterDetector> detector = master.get()->createDetector();
+
+ Try<Owned<cluster::Slave>> slave =
+ StartSlave(detector.get(), &containerizer, flags);
+ ASSERT_SOME(slave);
+ FrameworkInfo frameworkInfo = DEFAULT_FRAMEWORK_INFO;
+ frameworkInfo.set_checkpoint(true); // Enable checkpointing.
+
+ MockScheduler sched;
+ MesosSchedulerDriver driver(
+ &sched, frameworkInfo, master.get()->pid, DEFAULT_CREDENTIAL);
+
+ Future<FrameworkID> frameworkId;
+ EXPECT_CALL(sched, registered(_, _, _))
+ .WillOnce(FutureArg<1>(&frameworkId));
+
+ Future<vector<Offer>> offers;
+ EXPECT_CALL(sched, resourceOffers(_, _))
+ .WillOnce(FutureArg<1>(&offers))
+ .WillRepeatedly(Return()); // Ignore subsequent offers.
+
+ driver.start();
+
+ AWAIT_READY(frameworkId);
+ AWAIT_READY(offers);
+ ASSERT_FALSE(offers->empty());
+
+ EXPECT_CALL(exec, registered(_, _, _, _));
+
+ EXPECT_CALL(exec, launchTask(_, _))
+ .WillOnce(SendStatusUpdateFromTask(TASK_RUNNING));
+
+ Future<TaskStatus> status;
+ EXPECT_CALL(sched, statusUpdate(_, _))
+ .WillOnce(FutureArg<1>(&status));
+
+ Future<Nothing> _statusUpdateAcknowledgement =
+ FUTURE_DISPATCH(slave.get()->pid, &Slave::_statusUpdateAcknowledgement);
+
+ driver.launchTasks(offers.get()[0].id(), createTasks(offers.get()[0]));
+
+ AWAIT_READY(status);
+ EXPECT_EQ(TASK_RUNNING, status->state());
+
+ AWAIT_READY(_statusUpdateAcknowledgement);
+
+ // Ensure that both the status update and its acknowledgement are
+ // correctly checkpointed.
+ Result<slave::state::State> state =
+ slave::state::recover(slave::paths::getMetaRootDir(flags.work_dir), true);
+
+ ASSERT_SOME(state);
+ ASSERT_SOME(state->slave);
+ ASSERT_TRUE(state->slave->frameworks.contains(frameworkId.get()));
+
+ slave::state::FrameworkState frameworkState =
+ state->slave->frameworks.get(frameworkId.get()).get();
+
+ ASSERT_EQ(1u, frameworkState.executors.size());
+
+ slave::state::ExecutorState executorState =
+ frameworkState.executors.begin()->second;
+
+ ASSERT_EQ(1u, executorState.runs.size());
+
+ slave::state::RunState runState = executorState.runs.begin()->second;
+
+ ASSERT_EQ(1u, runState.tasks.size());
+
+ slave::state::TaskState taskState = runState.tasks.begin()->second;
+
+ EXPECT_EQ(1u, taskState.updates.size());
+ EXPECT_EQ(1u, taskState.acks.size());
+
+ EXPECT_CALL(exec, shutdown(_))
+ .Times(AtMost(1));
+
+ driver.stop();
+ driver.join();
+}
+
+
+TEST_F(TaskStatusUpdateManagerTest, RetryStatusUpdate)
+{
+ Try<Owned<cluster::Master>> master = StartMaster();
+ ASSERT_SOME(master);
+
+ MockExecutor exec(DEFAULT_EXECUTOR_ID);
+ TestContainerizer containerizer(&exec);
+
+ slave::Flags flags = CreateSlaveFlags();
+
+ Owned<MasterDetector> detector = master.get()->createDetector();
+
+ Try<Owned<cluster::Slave>> slave =
+ StartSlave(detector.get(), &containerizer, flags);
+ ASSERT_SOME(slave);
+ FrameworkInfo frameworkInfo = DEFAULT_FRAMEWORK_INFO;
+ frameworkInfo.set_checkpoint(true); // Enable checkpointing.
+
+ MockScheduler sched;
+ MesosSchedulerDriver driver(
+ &sched, frameworkInfo, master.get()->pid, DEFAULT_CREDENTIAL);
+
+ EXPECT_CALL(sched, registered(_, _, _));
+
+ Future<vector<Offer>> offers;
+ EXPECT_CALL(sched, resourceOffers(_, _))
+ .WillOnce(FutureArg<1>(&offers))
+ .WillRepeatedly(Return()); // Ignore subsequent offers.
+
+ driver.start();
+
+ AWAIT_READY(offers);
+ ASSERT_FALSE(offers->empty());
+
+ EXPECT_CALL(exec, registered(_, _, _, _));
+
+ EXPECT_CALL(exec, launchTask(_, _))
+ .WillOnce(SendStatusUpdateFromTask(TASK_RUNNING));
+
+ Future<StatusUpdateMessage> statusUpdateMessage =
+ DROP_PROTOBUF(StatusUpdateMessage(), master.get()->pid, _);
+
+ Clock::pause();
+
+ driver.launchTasks(offers.get()[0].id(), createTasks(offers.get()[0]));
+
+ AWAIT_READY(statusUpdateMessage);
+
+ Future<TaskStatus> status;
+ EXPECT_CALL(sched, statusUpdate(_, _))
+ .WillOnce(FutureArg<1>(&status));
+
+ Clock::advance(slave::STATUS_UPDATE_RETRY_INTERVAL_MIN);
+
+ AWAIT_READY(status);
+
+ EXPECT_EQ(TASK_RUNNING, status->state());
+
+ Clock::resume();
+
+ EXPECT_CALL(exec, shutdown(_))
+ .Times(AtMost(1));
+
+ driver.stop();
+ driver.join();
+}
+
+
+// This test verifies that status update manager ignores
+// duplicate ACK for an earlier update when it is waiting
+// for an ACK for a later update. This could happen when the
+// duplicate ACK is for a retried update.
+TEST_F(TaskStatusUpdateManagerTest, IgnoreDuplicateStatusUpdateAck)
+{
+ Try<Owned<cluster::Master>> master = StartMaster();
+ ASSERT_SOME(master);
+
+ MockExecutor exec(DEFAULT_EXECUTOR_ID);
+ TestContainerizer containerizer(&exec);
+
+ Owned<MasterDetector> detector = master.get()->createDetector();
+ Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), &containerizer);
+ ASSERT_SOME(slave);
+
+ FrameworkInfo frameworkInfo = DEFAULT_FRAMEWORK_INFO;
+ frameworkInfo.set_checkpoint(true); // Enable checkpointing.
+
+ MockScheduler sched;
+ MesosSchedulerDriver driver(
+ &sched, frameworkInfo, master.get()->pid, DEFAULT_CREDENTIAL);
+
+ FrameworkID frameworkId;
+ EXPECT_CALL(sched, registered(_, _, _))
+ .WillOnce(SaveArg<1>(&frameworkId));
+
+ Future<vector<Offer>> offers;
+ EXPECT_CALL(sched, resourceOffers(_, _))
+ .WillOnce(FutureArg<1>(&offers))
+ .WillRepeatedly(Return()); // Ignore subsequent offers.
+
+ driver.start();
+
+ AWAIT_READY(offers);
+ ASSERT_FALSE(offers->empty());
+
+ ExecutorDriver* execDriver;
+ EXPECT_CALL(exec, registered(_, _, _, _))
+ .WillOnce(SaveArg<0>(&execDriver));
+
+ EXPECT_CALL(exec, launchTask(_, _))
+ .WillOnce(SendStatusUpdateFromTask(TASK_RUNNING));
+
+ // Drop the first update, so that status update manager
+ // resends the update.
+ Future<StatusUpdateMessage> statusUpdateMessage =
+ DROP_PROTOBUF(StatusUpdateMessage(), master.get()->pid, _);
+
+ Clock::pause();
+
+ driver.launchTasks(offers.get()[0].id(), createTasks(offers.get()[0]));
+
+ AWAIT_READY(statusUpdateMessage);
+ StatusUpdate update = statusUpdateMessage->update();
+
+ Future<TaskStatus> status;
+ EXPECT_CALL(sched, statusUpdate(_, _))
+ .WillOnce(FutureArg<1>(&status));
+
+ // This is the ACK for the retried update.
+ Future<Nothing> ack =
+ FUTURE_DISPATCH(_, &Slave::_statusUpdateAcknowledgement);
+
+ Clock::advance(slave::STATUS_UPDATE_RETRY_INTERVAL_MIN);
+
+ AWAIT_READY(status);
+
+ EXPECT_EQ(TASK_RUNNING, status->state());
+
+ AWAIT_READY(ack);
+
+ // Now send TASK_FINISHED update so that the status update manager
+ // is waiting for its ACK, which it never gets because we drop the
+ // update.
+ DROP_PROTOBUFS(StatusUpdateMessage(), master.get()->pid, _);
+
+ Future<Nothing> update2 = FUTURE_DISPATCH(_, &Slave::_statusUpdate);
+
+ TaskStatus status2 = status.get();
+ status2.set_state(TASK_FINISHED);
+
+ execDriver->sendStatusUpdate(status2);
+
+ AWAIT_READY(update2);
+
+ // This is to catch the duplicate ack for TASK_RUNNING.
+ Future<Nothing> duplicateAck =
+ FUTURE_DISPATCH(_, &Slave::_statusUpdateAcknowledgement);
+
+ // Now send a duplicate ACK for the TASK_RUNNING update.
+ process::dispatch(
+ slave.get()->pid,
+ &Slave::statusUpdateAcknowledgement,
+ master.get()->pid,
+ update.slave_id(),
+ frameworkId,
+ update.status().task_id(),
+ update.uuid());
+
+ AWAIT_READY(duplicateAck);
+
+ Clock::resume();
+
+ EXPECT_CALL(exec, shutdown(_))
+ .Times(AtMost(1));
+
+ driver.stop();
+ driver.join();
+}
+
+
+// This test verifies that status update manager ignores
+// unexpected ACK for an earlier update when it is waiting
+// for an ACK for another update. We do this by dropping ACKs
+// for the original update and sending a random ACK to the slave.
+TEST_F(TaskStatusUpdateManagerTest, IgnoreUnexpectedStatusUpdateAck)
+{
+ Try<Owned<cluster::Master>> master = StartMaster();
+ ASSERT_SOME(master);
+
+ MockExecutor exec(DEFAULT_EXECUTOR_ID);
+ TestContainerizer containerizer(&exec);
+
+ Owned<MasterDetector> detector = master.get()->createDetector();
+ Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), &containerizer);
+ ASSERT_SOME(slave);
+
+ FrameworkInfo frameworkInfo = DEFAULT_FRAMEWORK_INFO;
+ frameworkInfo.set_checkpoint(true); // Enable checkpointing.
+
+ MockScheduler sched;
+ MesosSchedulerDriver driver(
+ &sched, frameworkInfo, master.get()->pid, DEFAULT_CREDENTIAL);
+
+ FrameworkID frameworkId;
+ EXPECT_CALL(sched, registered(_, _, _))
+ .WillOnce(SaveArg<1>(&frameworkId));
+
+ Future<vector<Offer>> offers;
+ EXPECT_CALL(sched, resourceOffers(_, _))
+ .WillOnce(FutureArg<1>(&offers))
+ .WillRepeatedly(Return()); // Ignore subsequent offers.
+
+ Future<TaskStatus> status;
+ EXPECT_CALL(sched, statusUpdate(_, _))
+ .WillOnce(FutureArg<1>(&status));
+
+ driver.start();
+
+ AWAIT_READY(offers);
+ ASSERT_FALSE(offers->empty());
+
+ ExecutorDriver* execDriver;
+ EXPECT_CALL(exec, registered(_, _, _, _))
+ .WillOnce(SaveArg<0>(&execDriver));
+
+ EXPECT_CALL(exec, launchTask(_, _))
+ .WillOnce(SendStatusUpdateFromTask(TASK_RUNNING));
+
+ Future<StatusUpdateMessage> statusUpdateMessage =
+ FUTURE_PROTOBUF(StatusUpdateMessage(), master.get()->pid, _);
+
+ // Drop the ACKs, so that status update manager
+ // retries the update.
+ DROP_CALLS(mesos::scheduler::Call(),
+ mesos::scheduler::Call::ACKNOWLEDGE,
+ _,
+ master.get()->pid);
+
+ driver.launchTasks(offers.get()[0].id(), createTasks(offers.get()[0]));
+
+ AWAIT_READY(statusUpdateMessage);
+ StatusUpdate update = statusUpdateMessage->update();
+
+ AWAIT_READY(status);
+
+ EXPECT_EQ(TASK_RUNNING, status->state());
+
+ Future<Nothing> unexpectedAck =
+ FUTURE_DISPATCH(_, &Slave::_statusUpdateAcknowledgement);
+
+ // Now send an ACK with a random UUID.
+ process::dispatch(
+ slave.get()->pid,
+ &Slave::statusUpdateAcknowledgement,
+ master.get()->pid,
+ update.slave_id(),
+ frameworkId,
+ update.status().task_id(),
+ UUID::random().toBytes());
+
+ AWAIT_READY(unexpectedAck);
+
+ EXPECT_CALL(exec, shutdown(_))
+ .Times(AtMost(1));
+
+ driver.stop();
+ driver.join();
+}
+
+
+// This test verifies that the slave and status update manager
+// properly handle duplicate terminal status updates, when the
+// second update is received after the ACK for the first update.
+// The proper behavior here is for the status update manager to
+// forward the duplicate update to the scheduler.
+TEST_F(TaskStatusUpdateManagerTest, DuplicateTerminalUpdateAfterAck)
+{
+ Try<Owned<cluster::Master>> master = StartMaster();
+ ASSERT_SOME(master);
+
+ MockExecutor exec(DEFAULT_EXECUTOR_ID);
+ TestContainerizer containerizer(&exec);
+
+ slave::Flags flags = CreateSlaveFlags();
+
+ Owned<MasterDetector> detector = master.get()->createDetector();
+
+ Try<Owned<cluster::Slave>> slave =
+ StartSlave(detector.get(), &containerizer, flags);
+ ASSERT_SOME(slave);
+ FrameworkInfo frameworkInfo = DEFAULT_FRAMEWORK_INFO;
+ frameworkInfo.set_checkpoint(true); // Enable checkpointing.
+
+ MockScheduler sched;
+ MesosSchedulerDriver driver(
+ &sched, frameworkInfo, master.get()->pid, DEFAULT_CREDENTIAL);
+
+ FrameworkID frameworkId;
+ EXPECT_CALL(sched, registered(_, _, _))
+ .WillOnce(SaveArg<1>(&frameworkId));
+
+ Future<vector<Offer>> offers;
+ EXPECT_CALL(sched, resourceOffers(_, _))
+ .WillOnce(FutureArg<1>(&offers))
+ .WillRepeatedly(Return()); // Ignore subsequent offers.
+
+ driver.start();
+
+ AWAIT_READY(offers);
+ ASSERT_FALSE(offers->empty());
+
+ ExecutorDriver* execDriver;
+ EXPECT_CALL(exec, registered(_, _, _, _))
+ .WillOnce(SaveArg<0>(&execDriver));
+
+ // Send a terminal update right away.
+ EXPECT_CALL(exec, launchTask(_, _))
+ .WillOnce(SendStatusUpdateFromTask(TASK_FINISHED));
+
+ Future<TaskStatus> status;
+ EXPECT_CALL(sched, statusUpdate(_, _))
+ .WillOnce(FutureArg<1>(&status));
+
+ Future<Nothing> _statusUpdateAcknowledgement =
+ FUTURE_DISPATCH(slave.get()->pid, &Slave::_statusUpdateAcknowledgement);
+
+ driver.launchTasks(offers.get()[0].id(), createTasks(offers.get()[0]));
+
+ AWAIT_READY(status);
+
+ EXPECT_EQ(TASK_FINISHED, status->state());
+
+ AWAIT_READY(_statusUpdateAcknowledgement);
+
+ Future<TaskStatus> update;
+ EXPECT_CALL(sched, statusUpdate(_, _))
+ .WillOnce(FutureArg<1>(&update));
+
+ Future<Nothing> _statusUpdateAcknowledgement2 =
+ FUTURE_DISPATCH(slave.get()->pid, &Slave::_statusUpdateAcknowledgement);
+
+ Clock::pause();
+
+ // Now send a TASK_KILLED update for the same task.
+ TaskStatus status2 = status.get();
+ status2.set_state(TASK_KILLED);
+ execDriver->sendStatusUpdate(status2);
+
+ // Ensure the scheduler receives TASK_KILLED.
+ AWAIT_READY(update);
+ EXPECT_EQ(TASK_KILLED, update->state());
+
+ // Ensure the slave properly handles the ACK.
+ // Clock::settle() ensures that the slave successfully
+ // executes Slave::_statusUpdateAcknowledgement().
+ AWAIT_READY(_statusUpdateAcknowledgement2);
+ Clock::settle();
+
+ Clock::resume();
+
+ EXPECT_CALL(exec, shutdown(_))
+ .Times(AtMost(1));
+
+ driver.stop();
+ driver.join();
+}
+
+
+// This test verifies that the slave and status update manager
+// properly handle duplicate status updates, when the second
+// update with the same UUID is received before the ACK for the
+// first update. The proper behavior here is for the status update
+// manager to drop the duplicate update.
+TEST_F(TaskStatusUpdateManagerTest, DuplicateUpdateBeforeAck)
+{
+ Try<Owned<cluster::Master>> master = StartMaster();
+ ASSERT_SOME(master);
+
+ MockExecutor exec(DEFAULT_EXECUTOR_ID);
+ TestContainerizer containerizer(&exec);
+
+ Owned<MasterDetector> detector = master.get()->createDetector();
+ Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), &containerizer);
+ ASSERT_SOME(slave);
+
+ FrameworkInfo frameworkInfo = DEFAULT_FRAMEWORK_INFO;
+ frameworkInfo.set_checkpoint(true); // Enable checkpointing.
+
+ MockScheduler sched;
+ MesosSchedulerDriver driver(
+ &sched, frameworkInfo, master.get()->pid, DEFAULT_CREDENTIAL);
+
+ FrameworkID frameworkId;
+ EXPECT_CALL(sched, registered(_, _, _))
+ .WillOnce(SaveArg<1>(&frameworkId));
+
+ Future<vector<Offer>> offers;
+ EXPECT_CALL(sched, resourceOffers(_, _))
+ .WillOnce(FutureArg<1>(&offers))
+ .WillRepeatedly(Return()); // Ignore subsequent offers.
+
+ driver.start();
+
+ AWAIT_READY(offers);
+ ASSERT_FALSE(offers->empty());
+
+ ExecutorDriver* execDriver;
+ EXPECT_CALL(exec, registered(_, _, _, _))
+ .WillOnce(SaveArg<0>(&execDriver));
+
+ EXPECT_CALL(exec, launchTask(_, _))
+ .WillOnce(SendStatusUpdateFromTask(TASK_RUNNING));
+
+ // Capture the first status update message.
+ Future<StatusUpdateMessage> statusUpdateMessage =
+ FUTURE_PROTOBUF(StatusUpdateMessage(), _, _);
+
+ Future<TaskStatus> status;
+ EXPECT_CALL(sched, statusUpdate(_, _))
+ .WillOnce(FutureArg<1>(&status));
+
+ // Drop the first ACK from the scheduler to the slave.
+ Future<StatusUpdateAcknowledgementMessage> statusUpdateAckMessage =
+ DROP_PROTOBUF(StatusUpdateAcknowledgementMessage(), _, slave.get()->pid);
+
+ Clock::pause();
+
+ driver.launchTasks(offers.get()[0].id(), createTasks(offers.get()[0]));
+
+ AWAIT_READY(statusUpdateMessage);
+
+ AWAIT_READY(status);
+
+ EXPECT_EQ(TASK_RUNNING, status->state());
+
+ AWAIT_READY(statusUpdateAckMessage);
+
+ Future<Nothing> ___statusUpdate =
+ FUTURE_DISPATCH(slave.get()->pid, &Slave::___statusUpdate);
+
+ // Now resend the TASK_RUNNING update.
+ process::post(slave.get()->pid, statusUpdateMessage.get());
+
+ // At this point the status update manager has handled
+ // the duplicate status update.
+ AWAIT_READY(___statusUpdate);
+
+ // After we advance the clock, the status update manager should
+ // retry the TASK_RUNNING update and the scheduler should receive
+ // and acknowledge it.
+ Future<TaskStatus> update;
+ EXPECT_CALL(sched, statusUpdate(_, _))
+ .WillOnce(FutureArg<1>(&update));
+
+ Clock::advance(slave::STATUS_UPDATE_RETRY_INTERVAL_MIN);
+ Clock::settle();
+
+ // Ensure the scheduler receives TASK_FINISHED.
+ AWAIT_READY(update);
+ EXPECT_EQ(TASK_RUNNING, update->state());
+
+ EXPECT_CALL(exec, shutdown(_))
+ .Times(AtMost(1));
+
+ Clock::resume();
+
+ driver.stop();
+ driver.join();
+}
+
+
+// This test verifies that the status update manager correctly includes
+// the latest state of the task in status update.
+TEST_F(TaskStatusUpdateManagerTest, LatestTaskState)
+{
+ Try<Owned<cluster::Master>> master = StartMaster();
+ ASSERT_SOME(master);
+
+ MockExecutor exec(DEFAULT_EXECUTOR_ID);
+ TestContainerizer containerizer(&exec);
+
+ Owned<MasterDetector> detector = master.get()->createDetector();
+ Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), &containerizer);
+ ASSERT_SOME(slave);
+
+ MockScheduler sched;
+ MesosSchedulerDriver driver(
+ &sched, DEFAULT_FRAMEWORK_INFO, master.get()->pid, DEFAULT_CREDENTIAL);
+
+ EXPECT_CALL(sched, registered(_, _, _));
+
+ EXPECT_CALL(sched, resourceOffers(_, _))
+ .WillOnce(LaunchTasks(DEFAULT_EXECUTOR_INFO, 1, 1, 512, "*"))
+ .WillRepeatedly(Return()); // Ignore subsequent offers.
+
+ ExecutorDriver* execDriver;
+ EXPECT_CALL(exec, registered(_, _, _, _))
+ .WillOnce(SaveArg<0>(&execDriver));
+
+ EXPECT_CALL(exec, launchTask(_, _))
+ .WillOnce(SendStatusUpdateFromTask(TASK_RUNNING));
+
+ // Signal when the first update is dropped.
+ Future<StatusUpdateMessage> statusUpdateMessage =
+ DROP_PROTOBUF(StatusUpdateMessage(), _, master.get()->pid);
+
+ Future<Nothing> ___statusUpdate = FUTURE_DISPATCH(_, &Slave::___statusUpdate);
+
+ driver.start();
+
+ // Wait until TASK_RUNNING is sent to the master.
+ AWAIT_READY(statusUpdateMessage);
+
+ // Ensure the status update manager handles the TASK_RUNNING update.
+ AWAIT_READY(___statusUpdate);
+
+ // Pause the clock to avoid status update manager from retrying.
+ Clock::pause();
+
+ Future<Nothing> ___statusUpdate2 =
+ FUTURE_DISPATCH(_, &Slave::___statusUpdate);
+
+ // Now send TASK_FINISHED update.
+ TaskStatus finishedStatus;
+ finishedStatus = statusUpdateMessage->update().status();
+ finishedStatus.set_state(TASK_FINISHED);
+ execDriver->sendStatusUpdate(finishedStatus);
+
+ // Ensure the status update manager handles the TASK_FINISHED update.
+ AWAIT_READY(___statusUpdate2);
+
+ // Signal when the second update is dropped.
+ Future<StatusUpdateMessage> statusUpdateMessage2 =
+ DROP_PROTOBUF(StatusUpdateMessage(), _, master.get()->pid);
+
+ // Advance the clock for the status update manager to send a retry.
+ Clock::advance(slave::STATUS_UPDATE_RETRY_INTERVAL_MIN);
+
+ AWAIT_READY(statusUpdateMessage2);
+
+ // The update should correspond to TASK_RUNNING.
+ ASSERT_EQ(TASK_RUNNING, statusUpdateMessage2->update().status().state());
+
+ // The update should include TASK_FINISHED as the latest state.
+ ASSERT_EQ(TASK_FINISHED,
+ statusUpdateMessage2->update().latest_state());
+
+ EXPECT_CALL(exec, shutdown(_))
+ .Times(AtMost(1));
+
+ driver.stop();
+ driver.join();
+}
+
+
+// This test verifies that if master receives a status update
+// for an already terminated task it forwards it without
+// changing the state of the task.
+TEST_F(TaskStatusUpdateManagerTest, DuplicatedTerminalStatusUpdate)
+{
+ Try<Owned<cluster::Master>> master = StartMaster();
+ ASSERT_SOME(master);
+
+ MockExecutor exec(DEFAULT_EXECUTOR_ID);
+ TestContainerizer containerizer(&exec);
+
+ Owned<MasterDetector> detector = master.get()->createDetector();
+ Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), &containerizer);
+ ASSERT_SOME(slave);
+
+ FrameworkInfo frameworkInfo = DEFAULT_FRAMEWORK_INFO;
+ frameworkInfo.set_checkpoint(true); // Enable checkpointing.
+
+ MockScheduler sched;
+ MesosSchedulerDriver driver(
+ &sched, frameworkInfo, master.get()->pid, DEFAULT_CREDENTIAL);
+
+ FrameworkID frameworkId;
+ EXPECT_CALL(sched, registered(_, _, _))
+ .WillOnce(SaveArg<1>(&frameworkId));
+
+ Future<vector<Offer>> offers;
+ EXPECT_CALL(sched, resourceOffers(_, _))
+ .WillOnce(FutureArg<1>(&offers))
+ .WillRepeatedly(Return()); // Ignore subsequent offers.
+
+ driver.start();
+
+ AWAIT_READY(offers);
+ ASSERT_FALSE(offers->empty());
+
+ ExecutorDriver* execDriver;
+ EXPECT_CALL(exec, registered(_, _, _, _))
+ .WillOnce(SaveArg<0>(&execDriver));
+
+ // Send a terminal update right away.
+ EXPECT_CALL(exec, launchTask(_, _))
+ .WillOnce(SendStatusUpdateFromTask(TASK_FINISHED));
+
+ Future<TaskStatus> status;
+ EXPECT_CALL(sched, statusUpdate(_, _))
+ .WillOnce(FutureArg<1>(&status));
+
+ Future<Nothing> _statusUpdateAcknowledgement =
+ FUTURE_DISPATCH(slave.get()->pid, &Slave::_statusUpdateAcknowledgement);
+
+ driver.launchTasks(offers.get()[0].id(), createTasks(offers.get()[0]));
+
+ AWAIT_READY(status);
+
+ EXPECT_EQ(TASK_FINISHED, status->state());
+
+ AWAIT_READY(_statusUpdateAcknowledgement);
+
+ Future<TaskStatus> update;
+ EXPECT_CALL(sched, statusUpdate(_, _))
+ .WillOnce(FutureArg<1>(&update));
+
+ Future<Nothing> _statusUpdateAcknowledgement2 =
+ FUTURE_DISPATCH(slave.get()->pid, &Slave::_statusUpdateAcknowledgement);
+
+ Clock::pause();
+
+ // Now send a TASK_KILLED update for the same task.
+ TaskStatus status2 = status.get();
+ status2.set_state(TASK_KILLED);
+ execDriver->sendStatusUpdate(status2);
+
+ // Ensure the scheduler receives TASK_KILLED.
+ AWAIT_READY(update);
+ EXPECT_EQ(TASK_KILLED, update->state());
+
+ // Ensure the slave properly handles the ACK.
+ // Clock::settle() ensures that the slave successfully
+ // executes Slave::_statusUpdateAcknowledgement().
+ AWAIT_READY(_statusUpdateAcknowledgement2);
+
+ // Verify the latest task status.
+ Future<process::http::Response> tasks = process::http::get(
+ master.get()->pid,
+ "tasks",
+ None(),
+ createBasicAuthHeaders(DEFAULT_CREDENTIAL));
+
+ AWAIT_EXPECT_RESPONSE_STATUS_EQ(process::http::OK().status, tasks);
+ AWAIT_EXPECT_RESPONSE_HEADER_EQ(APPLICATION_JSON, "Content-Type", tasks);
+
+ Try<JSON::Object> parse = JSON::parse<JSON::Object>(tasks->body);
+ ASSERT_SOME(parse);
+
+ Result<JSON::String> state = parse->find<JSON::String>("tasks[0].state");
+
+ ASSERT_SOME_EQ(JSON::String("TASK_FINISHED"), state);
+
+ Clock::resume();
+
+ EXPECT_CALL(exec, shutdown(_))
+ .Times(AtMost(1));
+
+ driver.stop();
+ driver.join();
+}
+
+} // namespace tests {
+} // namespace internal {
+} // namespace mesos {
[3/6] mesos git commit: Used fully qualified namespace in the
scheduler library for consistency.
Posted by gr...@apache.org.
Used fully qualified namespace in the scheduler library for consistency.
Review: https://reviews.apache.org/r/63850/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/d3a38962
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/d3a38962
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/d3a38962
Branch: refs/heads/master
Commit: d3a3896201b7e8f4e44e9db57d245b18ab903454
Parents: f6359bd
Author: Greg Mann <gr...@mesosphere.io>
Authored: Mon Nov 20 16:42:42 2017 -0800
Committer: Greg Mann <gr...@gmail.com>
Committed: Mon Nov 20 16:49:27 2017 -0800
----------------------------------------------------------------------
src/scheduler/scheduler.cpp | 7 +++----
1 file changed, 3 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/d3a38962/src/scheduler/scheduler.cpp
----------------------------------------------------------------------
diff --git a/src/scheduler/scheduler.cpp b/src/scheduler/scheduler.cpp
index 7294f23..e354709 100644
--- a/src/scheduler/scheduler.cpp
+++ b/src/scheduler/scheduler.cpp
@@ -120,7 +120,6 @@ using process::wait; // Necessary on some OS's to disambiguate.
using process::http::Connection;
using process::http::Pipe;
using process::http::post;
-using process::http::Request;
using process::http::URL;
using ::recordio::Decoder;
@@ -251,7 +250,7 @@ public:
// to the slave, instead of relaying it through the master, as
// the scheduler driver does.
- ::Request request;
+ process::http::Request request;
request.method = "POST";
request.url = master.get();
request.body = serialize(contentType, call);
@@ -538,7 +537,7 @@ protected:
LOG(WARNING) << "Dropping " << call.type() << ": " << message;
}
- void _send(const Call& call, const Future<::Request>& future)
+ void _send(const Call& call, const Future<process::http::Request>& future)
{
if (!future.isReady()) {
LOG(ERROR) << "HTTP authenticatee "
@@ -547,7 +546,7 @@ protected:
return;
}
- ::Request request = future.get();
+ process::http::Request request = future.get();
if (connections.isNone()) {
drop(call, "Connection to master interrupted");
[6/6] mesos git commit: Added the "task" prefix to the name of the
status update manager files.
Posted by gr...@apache.org.
Added the "task" prefix to the name of the status update manager files.
Review: https://reviews.apache.org/r/63853/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/62d11733
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/62d11733
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/62d11733
Branch: refs/heads/master
Commit: 62d11733446cf54beb02295e7cd94d419541f152
Parents: 2377cd8
Author: Gaston Kleiman <ga...@mesosphere.io>
Authored: Mon Nov 20 16:43:12 2017 -0800
Committer: Greg Mann <gr...@gmail.com>
Committed: Mon Nov 20 16:53:05 2017 -0800
----------------------------------------------------------------------
src/CMakeLists.txt | 2 +-
src/Makefile.am | 6 +-
src/local/local.cpp | 2 +-
src/slave/main.cpp | 2 +-
src/slave/slave.cpp | 2 +-
src/slave/status_update_manager.cpp | 897 -------------------
src/slave/status_update_manager.hpp | 210 -----
src/slave/task_status_update_manager.cpp | 898 ++++++++++++++++++++
src/slave/task_status_update_manager.hpp | 210 +++++
src/tests/CMakeLists.txt | 2 +-
src/tests/cluster.cpp | 2 +-
src/tests/cluster.hpp | 2 +-
src/tests/mock_slave.cpp | 2 +-
src/tests/status_update_manager_tests.cpp | 852 -------------------
src/tests/task_status_update_manager_tests.cpp | 852 +++++++++++++++++++
15 files changed, 1971 insertions(+), 1970 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/62d11733/src/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index 4f11418..be212d9 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -257,7 +257,7 @@ set(AGENT_SRC
slave/resource_estimator.cpp
slave/slave.cpp
slave/state.cpp
- slave/status_update_manager.cpp
+ slave/task_status_update_manager.cpp
slave/validation.cpp
slave/container_loggers/sandbox.cpp
slave/containerizer/composing.cpp
http://git-wip-us.apache.org/repos/asf/mesos/blob/62d11733/src/Makefile.am
----------------------------------------------------------------------
diff --git a/src/Makefile.am b/src/Makefile.am
index 49dec55..9641ad4 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -1022,7 +1022,7 @@ libmesos_no_3rdparty_la_SOURCES += \
slave/resource_estimator.cpp \
slave/slave.cpp \
slave/state.cpp \
- slave/status_update_manager.cpp \
+ slave/task_status_update_manager.cpp \
slave/validation.cpp \
slave/container_loggers/sandbox.cpp \
slave/containerizer/composing.cpp \
@@ -1167,7 +1167,7 @@ libmesos_no_3rdparty_la_SOURCES += \
slave/posix_signalhandler.hpp \
slave/slave.hpp \
slave/state.hpp \
- slave/status_update_manager.hpp \
+ slave/task_status_update_manager.hpp \
slave/validation.hpp \
slave/windows_ctrlhandler.hpp \
slave/container_loggers/sandbox.hpp \
@@ -2469,7 +2469,7 @@ mesos_tests_SOURCES = \
tests/slave_tests.cpp \
tests/sorter_tests.cpp \
tests/state_tests.cpp \
- tests/status_update_manager_tests.cpp \
+ tests/task_status_update_manager_tests.cpp \
tests/teardown_tests.cpp \
tests/upgrade_tests.cpp \
tests/uri_tests.cpp \
http://git-wip-us.apache.org/repos/asf/mesos/blob/62d11733/src/local/local.cpp
----------------------------------------------------------------------
diff --git a/src/local/local.cpp b/src/local/local.cpp
index 63d9822..2e141c6 100644
--- a/src/local/local.cpp
+++ b/src/local/local.cpp
@@ -75,7 +75,7 @@
#include "slave/gc.hpp"
#include "slave/slave.hpp"
-#include "slave/status_update_manager.hpp"
+#include "slave/task_status_update_manager.hpp"
#include "slave/containerizer/containerizer.hpp"
#include "slave/containerizer/fetcher.hpp"
http://git-wip-us.apache.org/repos/asf/mesos/blob/62d11733/src/slave/main.cpp
----------------------------------------------------------------------
diff --git a/src/slave/main.cpp b/src/slave/main.cpp
index de87553..f0716fb 100644
--- a/src/slave/main.cpp
+++ b/src/slave/main.cpp
@@ -69,7 +69,7 @@
#include "slave/gc.hpp"
#include "slave/slave.hpp"
-#include "slave/status_update_manager.hpp"
+#include "slave/task_status_update_manager.hpp"
#include "version/version.hpp"
http://git-wip-us.apache.org/repos/asf/mesos/blob/62d11733/src/slave/slave.cpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.cpp b/src/slave/slave.cpp
index c3a4088..6e9adc6 100644
--- a/src/slave/slave.cpp
+++ b/src/slave/slave.cpp
@@ -98,7 +98,7 @@
#include "slave/flags.hpp"
#include "slave/paths.hpp"
#include "slave/slave.hpp"
-#include "slave/status_update_manager.hpp"
+#include "slave/task_status_update_manager.hpp"
#ifdef __WINDOWS__
// Used to install a Windows console ctrl handler.
http://git-wip-us.apache.org/repos/asf/mesos/blob/62d11733/src/slave/status_update_manager.cpp
----------------------------------------------------------------------
diff --git a/src/slave/status_update_manager.cpp b/src/slave/status_update_manager.cpp
deleted file mode 100644
index fed0903..0000000
--- a/src/slave/status_update_manager.cpp
+++ /dev/null
@@ -1,897 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-#include <process/delay.hpp>
-#include <process/id.hpp>
-#include <process/process.hpp>
-#include <process/timer.hpp>
-
-#include <stout/check.hpp>
-#include <stout/foreach.hpp>
-#include <stout/hashmap.hpp>
-#include <stout/hashset.hpp>
-#include <stout/lambda.hpp>
-#include <stout/option.hpp>
-#include <stout/os.hpp>
-#include <stout/path.hpp>
-#include <stout/protobuf.hpp>
-#include <stout/stringify.hpp>
-#include <stout/utils.hpp>
-#include <stout/uuid.hpp>
-
-#include "common/protobuf_utils.hpp"
-
-#include "logging/logging.hpp"
-
-#include "slave/constants.hpp"
-#include "slave/flags.hpp"
-#include "slave/slave.hpp"
-#include "slave/state.hpp"
-#include "slave/status_update_manager.hpp"
-
-using lambda::function;
-
-using std::string;
-
-using process::wait; // Necessary on some OS's to disambiguate.
-using process::Failure;
-using process::Future;
-using process::PID;
-using process::Timeout;
-using process::UPID;
-
-namespace mesos {
-namespace internal {
-namespace slave {
-
-using state::SlaveState;
-using state::FrameworkState;
-using state::ExecutorState;
-using state::RunState;
-using state::TaskState;
-
-
-class TaskStatusUpdateManagerProcess
- : public ProtobufProcess<TaskStatusUpdateManagerProcess>
-{
-public:
- TaskStatusUpdateManagerProcess(const Flags& flags);
- virtual ~TaskStatusUpdateManagerProcess();
-
- // Explicitly use 'initialize' since we're overloading below.
- using process::ProcessBase::initialize;
-
- // TaskStatusUpdateManager implementation.
- void initialize(const function<void(StatusUpdate)>& forward);
-
- Future<Nothing> update(
- const StatusUpdate& update,
- const SlaveID& slaveId,
- const ExecutorID& executorId,
- const ContainerID& containerId);
-
- Future<Nothing> update(
- const StatusUpdate& update,
- const SlaveID& slaveId);
-
- Future<bool> acknowledgement(
- const TaskID& taskId,
- const FrameworkID& frameworkId,
- const UUID& uuid);
-
- Future<Nothing> recover(
- const string& rootDir,
- const Option<SlaveState>& state);
-
- void pause();
- void resume();
-
- void cleanup(const FrameworkID& frameworkId);
-
-private:
- // Helper function to handle update.
- Future<Nothing> _update(
- const StatusUpdate& update,
- const SlaveID& slaveId,
- bool checkpoint,
- const Option<ExecutorID>& executorId,
- const Option<ContainerID>& containerId);
-
- // Status update timeout.
- void timeout(const Duration& duration);
-
- // Forwards the status update to the master and starts a timer based
- // on the 'duration' to check for ACK from the scheduler.
- // NOTE: This should only be used for those messages that expect an
- // ACK (e.g updates from the executor).
- Timeout forward(const StatusUpdate& update, const Duration& duration);
-
- // Helper functions.
-
- // Creates a new status update stream (opening the updates file, if path is
- // present) and adds it to streams.
- TaskStatusUpdateStream* createStatusUpdateStream(
- const TaskID& taskId,
- const FrameworkID& frameworkId,
- const SlaveID& slaveId,
- bool checkpoint,
- const Option<ExecutorID>& executorId,
- const Option<ContainerID>& containerId);
-
- TaskStatusUpdateStream* getStatusUpdateStream(
- const TaskID& taskId,
- const FrameworkID& frameworkId);
-
- void cleanupStatusUpdateStream(
- const TaskID& taskId,
- const FrameworkID& frameworkId);
-
- const Flags flags;
- bool paused;
-
- function<void(StatusUpdate)> forward_;
-
- hashmap<FrameworkID, hashmap<TaskID, TaskStatusUpdateStream*>> streams;
-};
-
-
-TaskStatusUpdateManagerProcess::TaskStatusUpdateManagerProcess(
- const Flags& _flags)
- : ProcessBase(process::ID::generate("task-status-update-manager")),
- flags(_flags),
- paused(false)
-{
-}
-
-
-TaskStatusUpdateManagerProcess::~TaskStatusUpdateManagerProcess()
-{
- foreachkey (const FrameworkID& frameworkId, streams) {
- foreachvalue (TaskStatusUpdateStream* stream, streams[frameworkId]) {
- delete stream;
- }
- }
- streams.clear();
-}
-
-
-void TaskStatusUpdateManagerProcess::initialize(
- const function<void(StatusUpdate)>& forward)
-{
- forward_ = forward;
-}
-
-
-void TaskStatusUpdateManagerProcess::pause()
-{
- LOG(INFO) << "Pausing sending task status updates";
- paused = true;
-}
-
-
-void TaskStatusUpdateManagerProcess::resume()
-{
- LOG(INFO) << "Resuming sending task status updates";
- paused = false;
-
- foreachkey (const FrameworkID& frameworkId, streams) {
- foreachvalue (TaskStatusUpdateStream* stream, streams[frameworkId]) {
- if (!stream->pending.empty()) {
- const StatusUpdate& update = stream->pending.front();
- LOG(WARNING) << "Resending task status update " << update;
- stream->timeout = forward(update, STATUS_UPDATE_RETRY_INTERVAL_MIN);
- }
- }
- }
-}
-
-
-Future<Nothing> TaskStatusUpdateManagerProcess::recover(
- const string& rootDir,
- const Option<SlaveState>& state)
-{
- LOG(INFO) << "Recovering task status update manager";
-
- if (state.isNone()) {
- return Nothing();
- }
-
- foreachvalue (const FrameworkState& framework, state->frameworks) {
- foreachvalue (const ExecutorState& executor, framework.executors) {
- LOG(INFO) << "Recovering executor '" << executor.id
- << "' of framework " << framework.id;
-
- if (executor.info.isNone()) {
- LOG(WARNING) << "Skipping recovering task status updates of"
- << " executor '" << executor.id
- << "' of framework " << framework.id
- << " because its info cannot be recovered";
- continue;
- }
-
- if (executor.latest.isNone()) {
- LOG(WARNING) << "Skipping recovering task status updates of"
- << " executor '" << executor.id
- << "' of framework " << framework.id
- << " because its latest run cannot be recovered";
- continue;
- }
-
- // We are only interested in the latest run of the executor!
- const ContainerID& latest = executor.latest.get();
- Option<RunState> run = executor.runs.get(latest);
- CHECK_SOME(run);
-
- if (run->completed) {
- VLOG(1) << "Skipping recovering task status updates of"
- << " executor '" << executor.id
- << "' of framework " << framework.id
- << " because its latest run " << latest.value()
- << " is completed";
- continue;
- }
-
- foreachvalue (const TaskState& task, run->tasks) {
- // No updates were ever received for this task!
- // This means either:
- // 1) the executor never received this task or
- // 2) executor launched it but the slave died before it got any updates.
- if (task.updates.empty()) {
- LOG(WARNING) << "No status updates found for task " << task.id
- << " of framework " << framework.id;
- continue;
- }
-
- // Create a new status update stream.
- TaskStatusUpdateStream* stream = createStatusUpdateStream(
- task.id, framework.id, state->id, true, executor.id, latest);
-
- // Replay the stream.
- Try<Nothing> replay = stream->replay(task.updates, task.acks);
- if (replay.isError()) {
- return Failure(
- "Failed to replay status updates for task " + stringify(task.id) +
- " of framework " + stringify(framework.id) +
- ": " + replay.error());
- }
-
- // At the end of the replay, the stream is either terminated or
- // contains only unacknowledged, if any, pending updates. The
- // pending updates will be flushed after the slave
- // re-registers with the master.
- if (stream->terminated) {
- cleanupStatusUpdateStream(task.id, framework.id);
- }
- }
- }
- }
-
- return Nothing();
-}
-
-
-void TaskStatusUpdateManagerProcess::cleanup(const FrameworkID& frameworkId)
-{
- LOG(INFO) << "Closing task status update streams for framework "
- << frameworkId;
-
- if (streams.contains(frameworkId)) {
- foreachkey (const TaskID& taskId, utils::copy(streams[frameworkId])) {
- cleanupStatusUpdateStream(taskId, frameworkId);
- }
- }
-}
-
-
-Future<Nothing> TaskStatusUpdateManagerProcess::update(
- const StatusUpdate& update,
- const SlaveID& slaveId,
- const ExecutorID& executorId,
- const ContainerID& containerId)
-{
- return _update(update, slaveId, true, executorId, containerId);
-}
-
-
-Future<Nothing> TaskStatusUpdateManagerProcess::update(
- const StatusUpdate& update,
- const SlaveID& slaveId)
-{
- return _update(update, slaveId, false, None(), None());
-}
-
-
-Future<Nothing> TaskStatusUpdateManagerProcess::_update(
- const StatusUpdate& update,
- const SlaveID& slaveId,
- bool checkpoint,
- const Option<ExecutorID>& executorId,
- const Option<ContainerID>& containerId)
-{
- const TaskID& taskId = update.status().task_id();
- const FrameworkID& frameworkId = update.framework_id();
-
- LOG(INFO) << "Received task status update " << update;
-
- // Write the status update to disk and enqueue it to send it to the master.
- // Create/Get the status update stream for this task.
- TaskStatusUpdateStream* stream = getStatusUpdateStream(taskId, frameworkId);
- if (stream == nullptr) {
- stream = createStatusUpdateStream(
- taskId, frameworkId, slaveId, checkpoint, executorId, containerId);
- }
-
- // Verify that we didn't get a non-checkpointable update for a
- // stream that is checkpointable, and vice-versa.
- if (stream->checkpoint != checkpoint) {
- return Failure(
- "Mismatched checkpoint value for task status update " +
- stringify(update) + " (expected checkpoint=" +
- stringify(stream->checkpoint) + " actual checkpoint=" +
- stringify(checkpoint) + ")");
- }
-
- // Handle the status update.
- Try<bool> result = stream->update(update);
- if (result.isError()) {
- return Failure(result.error());
- }
-
- // We don't return a failed future here so that the slave can re-ack
- // the duplicate update.
- if (!result.get()) {
- return Nothing();
- }
-
- // Forward the status update to the master if this is the first in the stream.
- // Subsequent status updates will get sent in 'acknowledgement()'.
- if (!paused && stream->pending.size() == 1) {
- CHECK_NONE(stream->timeout);
- const Result<StatusUpdate>& next = stream->next();
- if (next.isError()) {
- return Failure(next.error());
- }
-
- CHECK_SOME(next);
- stream->timeout = forward(next.get(), STATUS_UPDATE_RETRY_INTERVAL_MIN);
- }
-
- return Nothing();
-}
-
-
-Timeout TaskStatusUpdateManagerProcess::forward(
- const StatusUpdate& update,
- const Duration& duration)
-{
- CHECK(!paused);
-
- VLOG(1) << "Forwarding task status update " << update << " to the agent";
-
- // Forward the update.
- forward_(update);
-
- // Send a message to self to resend after some delay if no ACK is received.
- return delay(duration,
- self(),
- &TaskStatusUpdateManagerProcess::timeout,
- duration).timeout();
-}
-
-
-Future<bool> TaskStatusUpdateManagerProcess::acknowledgement(
- const TaskID& taskId,
- const FrameworkID& frameworkId,
- const UUID& uuid)
-{
- LOG(INFO) << "Received task status update acknowledgement (UUID: " << uuid
- << ") for task " << taskId
- << " of framework " << frameworkId;
-
- TaskStatusUpdateStream* stream = getStatusUpdateStream(taskId, frameworkId);
-
- // This might happen if we haven't completed recovery yet or if the
- // acknowledgement is for a stream that has been cleaned up.
- if (stream == nullptr) {
- return Failure(
- "Cannot find the task status update stream for task " +
- stringify(taskId) + " of framework " + stringify(frameworkId));
- }
-
- // Get the corresponding update for this ACK.
- const Result<StatusUpdate>& update = stream->next();
- if (update.isError()) {
- return Failure(update.error());
- }
-
- // This might happen if we retried a status update and got back
- // acknowledgments for both the original and the retried update.
- if (update.isNone()) {
- return Failure(
- "Unexpected task status update acknowledgment (UUID: " +
- uuid.toString() + ") for task " + stringify(taskId) + " of framework " +
- stringify(frameworkId));
- }
-
- // Handle the acknowledgement.
- Try<bool> result =
- stream->acknowledgement(taskId, frameworkId, uuid, update.get());
-
- if (result.isError()) {
- return Failure(result.error());
- }
-
- if (!result.get()) {
- return Failure("Duplicate task status acknowledgement");
- }
-
- // Reset the timeout.
- stream->timeout = None();
-
- // Get the next update in the queue.
- const Result<StatusUpdate>& next = stream->next();
- if (next.isError()) {
- return Failure(next.error());
- }
-
- bool terminated = stream->terminated;
-
- if (terminated) {
- if (next.isSome()) {
- LOG(WARNING) << "Acknowledged a terminal"
- << " task status update " << update.get()
- << " but updates are still pending";
- }
- cleanupStatusUpdateStream(taskId, frameworkId);
- } else if (!paused && next.isSome()) {
- // Forward the next queued status update.
- stream->timeout = forward(next.get(), STATUS_UPDATE_RETRY_INTERVAL_MIN);
- }
-
- return !terminated;
-}
-
-
-// TODO(vinod): There should be a limit on the retries.
-void TaskStatusUpdateManagerProcess::timeout(const Duration& duration)
-{
- if (paused) {
- return;
- }
-
- // Check and see if we should resend any status updates.
- foreachkey (const FrameworkID& frameworkId, streams) {
- foreachvalue (TaskStatusUpdateStream* stream, streams[frameworkId]) {
- CHECK_NOTNULL(stream);
- if (!stream->pending.empty()) {
- CHECK_SOME(stream->timeout);
- if (stream->timeout->expired()) {
- const StatusUpdate& update = stream->pending.front();
- LOG(WARNING) << "Resending task status update " << update;
-
- // Bounded exponential backoff.
- Duration duration_ =
- std::min(duration * 2, STATUS_UPDATE_RETRY_INTERVAL_MAX);
-
- stream->timeout = forward(update, duration_);
- }
- }
- }
- }
-}
-
-
-TaskStatusUpdateStream*
-TaskStatusUpdateManagerProcess::createStatusUpdateStream(
- const TaskID& taskId,
- const FrameworkID& frameworkId,
- const SlaveID& slaveId,
- bool checkpoint,
- const Option<ExecutorID>& executorId,
- const Option<ContainerID>& containerId)
-{
- VLOG(1) << "Creating StatusUpdate stream for task " << taskId
- << " of framework " << frameworkId;
-
- TaskStatusUpdateStream* stream = new TaskStatusUpdateStream(
- taskId, frameworkId, slaveId, flags, checkpoint, executorId, containerId);
-
- streams[frameworkId][taskId] = stream;
- return stream;
-}
-
-
-TaskStatusUpdateStream* TaskStatusUpdateManagerProcess::getStatusUpdateStream(
- const TaskID& taskId,
- const FrameworkID& frameworkId)
-{
- if (!streams.contains(frameworkId)) {
- return nullptr;
- }
-
- if (!streams[frameworkId].contains(taskId)) {
- return nullptr;
- }
-
- return streams[frameworkId][taskId];
-}
-
-
-void TaskStatusUpdateManagerProcess::cleanupStatusUpdateStream(
- const TaskID& taskId,
- const FrameworkID& frameworkId)
-{
- VLOG(1) << "Cleaning up status update stream"
- << " for task " << taskId
- << " of framework " << frameworkId;
-
- CHECK(streams.contains(frameworkId))
- << "Cannot find the task status update streams for framework "
- << frameworkId;
-
- CHECK(streams[frameworkId].contains(taskId))
- << "Cannot find the status update streams for task " << taskId;
-
- TaskStatusUpdateStream* stream = streams[frameworkId][taskId];
-
- streams[frameworkId].erase(taskId);
- if (streams[frameworkId].empty()) {
- streams.erase(frameworkId);
- }
-
- delete stream;
-}
-
-
-TaskStatusUpdateManager::TaskStatusUpdateManager(const Flags& flags)
-{
- process = new TaskStatusUpdateManagerProcess(flags);
- spawn(process);
-}
-
-
-TaskStatusUpdateManager::~TaskStatusUpdateManager()
-{
- terminate(process);
- wait(process);
- delete process;
-}
-
-
-void TaskStatusUpdateManager::initialize(
- const function<void(StatusUpdate)>& forward)
-{
- dispatch(process, &TaskStatusUpdateManagerProcess::initialize, forward);
-}
-
-
-Future<Nothing> TaskStatusUpdateManager::update(
- const StatusUpdate& update,
- const SlaveID& slaveId,
- const ExecutorID& executorId,
- const ContainerID& containerId)
-{
- return dispatch(
- process,
- &TaskStatusUpdateManagerProcess::update,
- update,
- slaveId,
- executorId,
- containerId);
-}
-
-
-Future<Nothing> TaskStatusUpdateManager::update(
- const StatusUpdate& update,
- const SlaveID& slaveId)
-{
- return dispatch(
- process,
- &TaskStatusUpdateManagerProcess::update,
- update,
- slaveId);
-}
-
-
-Future<bool> TaskStatusUpdateManager::acknowledgement(
- const TaskID& taskId,
- const FrameworkID& frameworkId,
- const UUID& uuid)
-{
- return dispatch(
- process,
- &TaskStatusUpdateManagerProcess::acknowledgement,
- taskId,
- frameworkId,
- uuid);
-}
-
-
-Future<Nothing> TaskStatusUpdateManager::recover(
- const string& rootDir,
- const Option<SlaveState>& state)
-{
- return dispatch(
- process, &TaskStatusUpdateManagerProcess::recover, rootDir, state);
-}
-
-
-void TaskStatusUpdateManager::pause()
-{
- dispatch(process, &TaskStatusUpdateManagerProcess::pause);
-}
-
-
-void TaskStatusUpdateManager::resume()
-{
- dispatch(process, &TaskStatusUpdateManagerProcess::resume);
-}
-
-
-void TaskStatusUpdateManager::cleanup(const FrameworkID& frameworkId)
-{
- dispatch(process, &TaskStatusUpdateManagerProcess::cleanup, frameworkId);
-}
-
-
-TaskStatusUpdateStream::TaskStatusUpdateStream(
- const TaskID& _taskId,
- const FrameworkID& _frameworkId,
- const SlaveID& _slaveId,
- const Flags& _flags,
- bool _checkpoint,
- const Option<ExecutorID>& executorId,
- const Option<ContainerID>& containerId)
- : checkpoint(_checkpoint),
- terminated(false),
- taskId(_taskId),
- frameworkId(_frameworkId),
- slaveId(_slaveId),
- flags(_flags),
- error(None())
-{
- if (checkpoint) {
- CHECK_SOME(executorId);
- CHECK_SOME(containerId);
-
- path = paths::getTaskUpdatesPath(
- paths::getMetaRootDir(flags.work_dir),
- slaveId,
- frameworkId,
- executorId.get(),
- containerId.get(),
- taskId);
-
- // Create the base updates directory, if it doesn't exist.
- const string& dirName = Path(path.get()).dirname();
- Try<Nothing> directory = os::mkdir(dirName);
- if (directory.isError()) {
- error = "Failed to create '" + dirName + "': " + directory.error();
- return;
- }
-
- // Open the updates file.
- // NOTE: We don't use `O_SYNC` here because we only read this file
- // if the host did not crash. `os::write` success implies the kernel
- // will have flushed our data to the page cache. This is sufficient
- // for the recovery scenarios we use this data for.
- Try<int_fd> result = os::open(
- path.get(),
- O_CREAT | O_WRONLY | O_APPEND | O_CLOEXEC,
- S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH);
-
- if (result.isError()) {
- error = "Failed to open '" + path.get() +
- "' for status updates: " + result.error();
- return;
- }
-
- // We keep the file open through the lifetime of the task, because it
- // makes it easy to append status update records to the file.
- fd = result.get();
- }
-}
-
-
-TaskStatusUpdateStream::~TaskStatusUpdateStream()
-{
- if (fd.isSome()) {
- Try<Nothing> close = os::close(fd.get());
- if (close.isError()) {
- CHECK_SOME(path);
- LOG(ERROR) << "Failed to close file '" << path.get() << "': "
- << close.error();
- }
- }
-}
-
-
-Try<bool> TaskStatusUpdateStream::update(const StatusUpdate& update)
-{
- if (error.isSome()) {
- return Error(error.get());
- }
-
- if (!update.has_uuid()) {
- return Error("Task status update is missing 'uuid'");
- }
-
- // Check that this status update has not already been acknowledged.
- // This could happen in the rare case when the slave received the ACK
- // from the framework, died, but slave's ACK to the executor never made it!
- if (acknowledged.contains(UUID::fromBytes(update.uuid()).get())) {
- LOG(WARNING) << "Ignoring task status update " << update
- << " that has already been acknowledged by the framework!";
- return false;
- }
-
- // Check that this update hasn't already been received.
- // This could happen if the slave receives a status update from an executor,
- // then crashes after it writes it to disk but before it sends an ack.
- if (received.contains(UUID::fromBytes(update.uuid()).get())) {
- LOG(WARNING) << "Ignoring duplicate task status update " << update;
- return false;
- }
-
- // Handle the update, checkpointing if necessary.
- Try<Nothing> result = handle(update, StatusUpdateRecord::UPDATE);
- if (result.isError()) {
- return Error(result.error());
- }
-
- return true;
-}
-
-
-Try<bool> TaskStatusUpdateStream::acknowledgement(
- const TaskID& taskId,
- const FrameworkID& frameworkId,
- const UUID& uuid,
- const StatusUpdate& update)
-{
- if (error.isSome()) {
- return Error(error.get());
- }
-
- if (acknowledged.contains(uuid)) {
- LOG(WARNING) << "Duplicate task status update acknowledgment (UUID: "
- << uuid << ") for update " << update;
- return false;
- }
-
- // This might happen if we retried a status update and got back
- // acknowledgments for both the original and the retried update.
- if (uuid != UUID::fromBytes(update.uuid()).get()) {
- LOG(WARNING) << "Unexpected task status update acknowledgement (received "
- << uuid << ", expecting "
- << UUID::fromBytes(update.uuid()).get()
- << ") for update " << update;
- return false;
- }
-
- // Handle the ACK, checkpointing if necessary.
- Try<Nothing> result = handle(update, StatusUpdateRecord::ACK);
- if (result.isError()) {
- return Error(result.error());
- }
-
- return true;
-}
-
-
-Result<StatusUpdate> TaskStatusUpdateStream::next()
-{
- if (error.isSome()) {
- return Error(error.get());
- }
-
- if (!pending.empty()) {
- return pending.front();
- }
-
- return None();
-}
-
-
-Try<Nothing> TaskStatusUpdateStream::replay(
- const std::vector<StatusUpdate>& updates,
- const hashset<UUID>& acks)
-{
- if (error.isSome()) {
- return Error(error.get());
- }
-
- VLOG(1) << "Replaying task status update stream for task " << taskId;
-
- foreach (const StatusUpdate& update, updates) {
- // Handle the update.
- _handle(update, StatusUpdateRecord::UPDATE);
-
- // Check if the update has an ACK too.
- if (acks.contains(UUID::fromBytes(update.uuid()).get())) {
- _handle(update, StatusUpdateRecord::ACK);
- }
- }
-
- return Nothing();
-}
-
-
-Try<Nothing> TaskStatusUpdateStream::handle(
- const StatusUpdate& update,
- const StatusUpdateRecord::Type& type)
-{
- CHECK_NONE(error);
-
- // Checkpoint the update if necessary.
- if (checkpoint) {
- LOG(INFO) << "Checkpointing " << type << " for task status update "
- << update;
-
- CHECK_SOME(fd);
-
- StatusUpdateRecord record;
- record.set_type(type);
-
- if (type == StatusUpdateRecord::UPDATE) {
- record.mutable_update()->CopyFrom(update);
- } else {
- record.set_uuid(update.uuid());
- }
-
- Try<Nothing> write = ::protobuf::write(fd.get(), record);
- if (write.isError()) {
- error = "Failed to write task status update " + stringify(update) +
- " to '" + path.get() + "': " + write.error();
- return Error(error.get());
- }
- }
-
- // Now actually handle the update.
- _handle(update, type);
-
- return Nothing();
-}
-
-
-void TaskStatusUpdateStream::_handle(
- const StatusUpdate& update,
- const StatusUpdateRecord::Type& type)
-{
- CHECK_NONE(error);
-
- if (type == StatusUpdateRecord::UPDATE) {
- // Record this update.
- received.insert(UUID::fromBytes(update.uuid()).get());
-
- // Add it to the pending updates queue.
- pending.push(update);
- } else {
- // Record this ACK.
- acknowledged.insert(UUID::fromBytes(update.uuid()).get());
-
- // Remove the corresponding update from the pending queue.
- pending.pop();
-
- if (!terminated) {
- terminated = protobuf::isTerminalState(update.status().state());
- }
- }
-}
-
-} // namespace slave {
-} // namespace internal {
-} // namespace mesos {
http://git-wip-us.apache.org/repos/asf/mesos/blob/62d11733/src/slave/status_update_manager.hpp
----------------------------------------------------------------------
diff --git a/src/slave/status_update_manager.hpp b/src/slave/status_update_manager.hpp
deleted file mode 100644
index 4f7d45d..0000000
--- a/src/slave/status_update_manager.hpp
+++ /dev/null
@@ -1,210 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-#ifndef __STATUS_UPDATE_MANAGER_HPP__
-#define __STATUS_UPDATE_MANAGER_HPP__
-
-#include <queue>
-#include <string>
-
-#include <mesos/mesos.hpp>
-
-#include <process/future.hpp>
-#include <process/pid.hpp>
-#include <process/timeout.hpp>
-
-#include <stout/hashset.hpp>
-#include <stout/lambda.hpp>
-#include <stout/option.hpp>
-#include <stout/try.hpp>
-#include <stout/uuid.hpp>
-
-#include "messages/messages.hpp"
-
-#include "slave/flags.hpp"
-
-namespace mesos {
-namespace internal {
-namespace slave {
-
-// Forward declarations.
-
-namespace state {
-struct SlaveState;
-}
-
-class TaskStatusUpdateManagerProcess;
-struct TaskStatusUpdateStream;
-
-
-// TaskStatusUpdateManager is responsible for
-// 1) Reliably sending status updates to the master.
-// 2) Checkpointing the update to disk (optional).
-// 3) Sending ACKs to the executor (optional).
-// 4) Receiving ACKs from the scheduler.
-class TaskStatusUpdateManager
-{
-public:
- TaskStatusUpdateManager(const Flags& flags);
- virtual ~TaskStatusUpdateManager();
-
- // Expects a callback 'forward' which gets called whenever there is
- // a new status update that needs to be forwarded to the master.
- void initialize(const lambda::function<void(StatusUpdate)>& forward);
-
- // TODO(vinod): Come up with better names/signatures for the
- // checkpointing and non-checkpointing 'update()' functions.
- // Currently, it is not obvious that one version of 'update()'
- // does checkpointing while the other doesn't.
-
- // Checkpoints the status update and reliably sends the
- // update to the master (and hence the scheduler).
- // @return Whether the update is handled successfully
- // (e.g. checkpointed).
- process::Future<Nothing> update(
- const StatusUpdate& update,
- const SlaveID& slaveId,
- const ExecutorID& executorId,
- const ContainerID& containerId);
-
- // Retries the update to the master (as long as the slave is
- // alive), but does not checkpoint the update.
- // @return Whether the update is handled successfully.
- process::Future<Nothing> update(
- const StatusUpdate& update,
- const SlaveID& slaveId);
-
- // Checkpoints the status update to disk if necessary.
- // Also, sends the next pending status update, if any.
- // @return True if the ACK is handled successfully (e.g., checkpointed)
- // and the task's status update stream is not terminated.
- // False same as above except the status update stream is terminated.
- // Failed if there are any errors (e.g., duplicate, checkpointing).
- process::Future<bool> acknowledgement(
- const TaskID& taskId,
- const FrameworkID& frameworkId,
- const UUID& uuid);
-
- // Recover status updates.
- process::Future<Nothing> recover(
- const std::string& rootDir,
- const Option<state::SlaveState>& state);
-
-
- // Pause sending updates.
- // This is useful when the slave is disconnected because a
- // disconnected slave will drop the updates.
- void pause();
-
- // Unpause and resend all the pending updates right away.
- // This is useful when the updates were pending because there was
- // no master elected (e.g., during recovery) or framework failed over.
- void resume();
-
- // Closes all the status update streams corresponding to this framework.
- // NOTE: This stops retrying any pending status updates for this framework.
- void cleanup(const FrameworkID& frameworkId);
-
-private:
- TaskStatusUpdateManagerProcess* process;
-};
-
-
-// TaskStatusUpdateStream handles the status updates and acknowledgements
-// of a task, checkpointing them if necessary. It also holds the information
-// about received, acknowledged and pending status updates.
-// NOTE: A task is expected to have a globally unique ID across the lifetime
-// of a framework. In other words the tuple (taskId, frameworkId) should be
-// always unique.
-struct TaskStatusUpdateStream
-{
- TaskStatusUpdateStream(const TaskID& _taskId,
- const FrameworkID& _frameworkId,
- const SlaveID& _slaveId,
- const Flags& _flags,
- bool _checkpoint,
- const Option<ExecutorID>& executorId,
- const Option<ContainerID>& containerId);
-
- ~TaskStatusUpdateStream();
-
- // This function handles the update, checkpointing if necessary.
- // @return True if the update is successfully handled.
- // False if the update is a duplicate.
- // Error Any errors (e.g., checkpointing).
- Try<bool> update(const StatusUpdate& update);
-
- // This function handles the ACK, checkpointing if necessary.
- // @return True if the acknowledgement is successfully handled.
- // False if the acknowledgement is a duplicate.
- // Error Any errors (e.g., checkpointing).
- Try<bool> acknowledgement(
- const TaskID& taskId,
- const FrameworkID& frameworkId,
- const UUID& uuid,
- const StatusUpdate& update);
-
- // Returns the next update (or none, if empty) in the queue.
- Result<StatusUpdate> next();
-
- // Replays the stream by sequentially handling an update and its
- // corresponding ACK, if present.
- Try<Nothing> replay(
- const std::vector<StatusUpdate>& updates,
- const hashset<UUID>& acks);
-
- // TODO(vinod): Explore semantics to make these private.
- const bool checkpoint;
- bool terminated;
- Option<process::Timeout> timeout; // Timeout for resending status update.
- std::queue<StatusUpdate> pending;
-
-private:
- // Handles the status update and writes it to disk, if necessary.
- // TODO(vinod): The write has to be asynchronous to avoid status updates that
- // are being checkpointed, blocking the processing of other updates.
- // One solution is to wrap the protobuf::write inside async, but its probably
- // too much of an overhead to spin up a new libprocess per status update?
- // A better solution might be to be have async write capability for file io.
- Try<Nothing> handle(
- const StatusUpdate& update,
- const StatusUpdateRecord::Type& type);
-
- void _handle(
- const StatusUpdate& update,
- const StatusUpdateRecord::Type& type);
-
- const TaskID taskId;
- const FrameworkID frameworkId;
- const SlaveID slaveId;
-
- const Flags flags;
-
- hashset<UUID> received;
- hashset<UUID> acknowledged;
-
- Option<std::string> path; // File path of the update stream.
- Option<int_fd> fd; // File descriptor to the update stream.
-
- Option<std::string> error; // Potential non-retryable error.
-};
-
-} // namespace slave {
-} // namespace internal {
-} // namespace mesos {
-
-
-#endif // __STATUS_UPDATE_MANAGER_HPP__
http://git-wip-us.apache.org/repos/asf/mesos/blob/62d11733/src/slave/task_status_update_manager.cpp
----------------------------------------------------------------------
diff --git a/src/slave/task_status_update_manager.cpp b/src/slave/task_status_update_manager.cpp
new file mode 100644
index 0000000..1ec6be7
--- /dev/null
+++ b/src/slave/task_status_update_manager.cpp
@@ -0,0 +1,898 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "slave/task_status_update_manager.hpp"
+
+#include <process/delay.hpp>
+#include <process/id.hpp>
+#include <process/process.hpp>
+#include <process/timer.hpp>
+
+#include <stout/check.hpp>
+#include <stout/foreach.hpp>
+#include <stout/hashmap.hpp>
+#include <stout/hashset.hpp>
+#include <stout/lambda.hpp>
+#include <stout/option.hpp>
+#include <stout/os.hpp>
+#include <stout/path.hpp>
+#include <stout/protobuf.hpp>
+#include <stout/stringify.hpp>
+#include <stout/utils.hpp>
+#include <stout/uuid.hpp>
+
+#include "common/protobuf_utils.hpp"
+
+#include "logging/logging.hpp"
+
+#include "slave/constants.hpp"
+#include "slave/flags.hpp"
+#include "slave/slave.hpp"
+#include "slave/state.hpp"
+
+using lambda::function;
+
+using std::string;
+
+using process::wait; // Necessary on some OS's to disambiguate.
+using process::Failure;
+using process::Future;
+using process::PID;
+using process::Timeout;
+using process::UPID;
+
+namespace mesos {
+namespace internal {
+namespace slave {
+
+using state::SlaveState;
+using state::FrameworkState;
+using state::ExecutorState;
+using state::RunState;
+using state::TaskState;
+
+
+class TaskStatusUpdateManagerProcess
+ : public ProtobufProcess<TaskStatusUpdateManagerProcess>
+{
+public:
+ TaskStatusUpdateManagerProcess(const Flags& flags);
+ virtual ~TaskStatusUpdateManagerProcess();
+
+ // Explicitly use 'initialize' since we're overloading below.
+ using process::ProcessBase::initialize;
+
+ // TaskStatusUpdateManager implementation.
+ void initialize(const function<void(StatusUpdate)>& forward);
+
+ Future<Nothing> update(
+ const StatusUpdate& update,
+ const SlaveID& slaveId,
+ const ExecutorID& executorId,
+ const ContainerID& containerId);
+
+ Future<Nothing> update(
+ const StatusUpdate& update,
+ const SlaveID& slaveId);
+
+ Future<bool> acknowledgement(
+ const TaskID& taskId,
+ const FrameworkID& frameworkId,
+ const UUID& uuid);
+
+ Future<Nothing> recover(
+ const string& rootDir,
+ const Option<SlaveState>& state);
+
+ void pause();
+ void resume();
+
+ void cleanup(const FrameworkID& frameworkId);
+
+private:
+ // Helper function to handle update.
+ Future<Nothing> _update(
+ const StatusUpdate& update,
+ const SlaveID& slaveId,
+ bool checkpoint,
+ const Option<ExecutorID>& executorId,
+ const Option<ContainerID>& containerId);
+
+ // Status update timeout.
+ void timeout(const Duration& duration);
+
+ // Forwards the status update to the master and starts a timer based
+ // on the 'duration' to check for ACK from the scheduler.
+ // NOTE: This should only be used for those messages that expect an
+ // ACK (e.g updates from the executor).
+ Timeout forward(const StatusUpdate& update, const Duration& duration);
+
+ // Helper functions.
+
+ // Creates a new status update stream (opening the updates file, if path is
+ // present) and adds it to streams.
+ TaskStatusUpdateStream* createStatusUpdateStream(
+ const TaskID& taskId,
+ const FrameworkID& frameworkId,
+ const SlaveID& slaveId,
+ bool checkpoint,
+ const Option<ExecutorID>& executorId,
+ const Option<ContainerID>& containerId);
+
+ TaskStatusUpdateStream* getStatusUpdateStream(
+ const TaskID& taskId,
+ const FrameworkID& frameworkId);
+
+ void cleanupStatusUpdateStream(
+ const TaskID& taskId,
+ const FrameworkID& frameworkId);
+
+ const Flags flags;
+ bool paused;
+
+ function<void(StatusUpdate)> forward_;
+
+ hashmap<FrameworkID, hashmap<TaskID, TaskStatusUpdateStream*>> streams;
+};
+
+
+TaskStatusUpdateManagerProcess::TaskStatusUpdateManagerProcess(
+ const Flags& _flags)
+ : ProcessBase(process::ID::generate("task-status-update-manager")),
+ flags(_flags),
+ paused(false)
+{
+}
+
+
+TaskStatusUpdateManagerProcess::~TaskStatusUpdateManagerProcess()
+{
+ foreachkey (const FrameworkID& frameworkId, streams) {
+ foreachvalue (TaskStatusUpdateStream* stream, streams[frameworkId]) {
+ delete stream;
+ }
+ }
+ streams.clear();
+}
+
+
+void TaskStatusUpdateManagerProcess::initialize(
+ const function<void(StatusUpdate)>& forward)
+{
+ forward_ = forward;
+}
+
+
+void TaskStatusUpdateManagerProcess::pause()
+{
+ LOG(INFO) << "Pausing sending task status updates";
+ paused = true;
+}
+
+
+void TaskStatusUpdateManagerProcess::resume()
+{
+ LOG(INFO) << "Resuming sending task status updates";
+ paused = false;
+
+ foreachkey (const FrameworkID& frameworkId, streams) {
+ foreachvalue (TaskStatusUpdateStream* stream, streams[frameworkId]) {
+ if (!stream->pending.empty()) {
+ const StatusUpdate& update = stream->pending.front();
+ LOG(WARNING) << "Resending task status update " << update;
+ stream->timeout = forward(update, STATUS_UPDATE_RETRY_INTERVAL_MIN);
+ }
+ }
+ }
+}
+
+
+Future<Nothing> TaskStatusUpdateManagerProcess::recover(
+ const string& rootDir,
+ const Option<SlaveState>& state)
+{
+ LOG(INFO) << "Recovering task status update manager";
+
+ if (state.isNone()) {
+ return Nothing();
+ }
+
+ foreachvalue (const FrameworkState& framework, state->frameworks) {
+ foreachvalue (const ExecutorState& executor, framework.executors) {
+ LOG(INFO) << "Recovering executor '" << executor.id
+ << "' of framework " << framework.id;
+
+ if (executor.info.isNone()) {
+ LOG(WARNING) << "Skipping recovering task status updates of"
+ << " executor '" << executor.id
+ << "' of framework " << framework.id
+ << " because its info cannot be recovered";
+ continue;
+ }
+
+ if (executor.latest.isNone()) {
+ LOG(WARNING) << "Skipping recovering task status updates of"
+ << " executor '" << executor.id
+ << "' of framework " << framework.id
+ << " because its latest run cannot be recovered";
+ continue;
+ }
+
+ // We are only interested in the latest run of the executor!
+ const ContainerID& latest = executor.latest.get();
+ Option<RunState> run = executor.runs.get(latest);
+ CHECK_SOME(run);
+
+ if (run->completed) {
+ VLOG(1) << "Skipping recovering task status updates of"
+ << " executor '" << executor.id
+ << "' of framework " << framework.id
+ << " because its latest run " << latest.value()
+ << " is completed";
+ continue;
+ }
+
+ foreachvalue (const TaskState& task, run->tasks) {
+ // No updates were ever received for this task!
+ // This means either:
+ // 1) the executor never received this task or
+ // 2) executor launched it but the slave died before it got any updates.
+ if (task.updates.empty()) {
+ LOG(WARNING) << "No status updates found for task " << task.id
+ << " of framework " << framework.id;
+ continue;
+ }
+
+ // Create a new status update stream.
+ TaskStatusUpdateStream* stream = createStatusUpdateStream(
+ task.id, framework.id, state->id, true, executor.id, latest);
+
+ // Replay the stream.
+ Try<Nothing> replay = stream->replay(task.updates, task.acks);
+ if (replay.isError()) {
+ return Failure(
+ "Failed to replay status updates for task " + stringify(task.id) +
+ " of framework " + stringify(framework.id) +
+ ": " + replay.error());
+ }
+
+ // At the end of the replay, the stream is either terminated or
+ // contains only unacknowledged, if any, pending updates. The
+ // pending updates will be flushed after the slave
+ // re-registers with the master.
+ if (stream->terminated) {
+ cleanupStatusUpdateStream(task.id, framework.id);
+ }
+ }
+ }
+ }
+
+ return Nothing();
+}
+
+
+void TaskStatusUpdateManagerProcess::cleanup(const FrameworkID& frameworkId)
+{
+ LOG(INFO) << "Closing task status update streams for framework "
+ << frameworkId;
+
+ if (streams.contains(frameworkId)) {
+ foreachkey (const TaskID& taskId, utils::copy(streams[frameworkId])) {
+ cleanupStatusUpdateStream(taskId, frameworkId);
+ }
+ }
+}
+
+
+Future<Nothing> TaskStatusUpdateManagerProcess::update(
+ const StatusUpdate& update,
+ const SlaveID& slaveId,
+ const ExecutorID& executorId,
+ const ContainerID& containerId)
+{
+ return _update(update, slaveId, true, executorId, containerId);
+}
+
+
+Future<Nothing> TaskStatusUpdateManagerProcess::update(
+ const StatusUpdate& update,
+ const SlaveID& slaveId)
+{
+ return _update(update, slaveId, false, None(), None());
+}
+
+
+Future<Nothing> TaskStatusUpdateManagerProcess::_update(
+ const StatusUpdate& update,
+ const SlaveID& slaveId,
+ bool checkpoint,
+ const Option<ExecutorID>& executorId,
+ const Option<ContainerID>& containerId)
+{
+ const TaskID& taskId = update.status().task_id();
+ const FrameworkID& frameworkId = update.framework_id();
+
+ LOG(INFO) << "Received task status update " << update;
+
+ // Write the status update to disk and enqueue it to send it to the master.
+ // Create/Get the status update stream for this task.
+ TaskStatusUpdateStream* stream = getStatusUpdateStream(taskId, frameworkId);
+ if (stream == nullptr) {
+ stream = createStatusUpdateStream(
+ taskId, frameworkId, slaveId, checkpoint, executorId, containerId);
+ }
+
+ // Verify that we didn't get a non-checkpointable update for a
+ // stream that is checkpointable, and vice-versa.
+ if (stream->checkpoint != checkpoint) {
+ return Failure(
+ "Mismatched checkpoint value for task status update " +
+ stringify(update) + " (expected checkpoint=" +
+ stringify(stream->checkpoint) + " actual checkpoint=" +
+ stringify(checkpoint) + ")");
+ }
+
+ // Handle the status update.
+ Try<bool> result = stream->update(update);
+ if (result.isError()) {
+ return Failure(result.error());
+ }
+
+ // We don't return a failed future here so that the slave can re-ack
+ // the duplicate update.
+ if (!result.get()) {
+ return Nothing();
+ }
+
+ // Forward the status update to the master if this is the first in the stream.
+ // Subsequent status updates will get sent in 'acknowledgement()'.
+ if (!paused && stream->pending.size() == 1) {
+ CHECK_NONE(stream->timeout);
+ const Result<StatusUpdate>& next = stream->next();
+ if (next.isError()) {
+ return Failure(next.error());
+ }
+
+ CHECK_SOME(next);
+ stream->timeout = forward(next.get(), STATUS_UPDATE_RETRY_INTERVAL_MIN);
+ }
+
+ return Nothing();
+}
+
+
+Timeout TaskStatusUpdateManagerProcess::forward(
+ const StatusUpdate& update,
+ const Duration& duration)
+{
+ CHECK(!paused);
+
+ VLOG(1) << "Forwarding task status update " << update << " to the agent";
+
+ // Forward the update.
+ forward_(update);
+
+ // Send a message to self to resend after some delay if no ACK is received.
+ return delay(duration,
+ self(),
+ &TaskStatusUpdateManagerProcess::timeout,
+ duration).timeout();
+}
+
+
+Future<bool> TaskStatusUpdateManagerProcess::acknowledgement(
+ const TaskID& taskId,
+ const FrameworkID& frameworkId,
+ const UUID& uuid)
+{
+ LOG(INFO) << "Received task status update acknowledgement (UUID: " << uuid
+ << ") for task " << taskId
+ << " of framework " << frameworkId;
+
+ TaskStatusUpdateStream* stream = getStatusUpdateStream(taskId, frameworkId);
+
+ // This might happen if we haven't completed recovery yet or if the
+ // acknowledgement is for a stream that has been cleaned up.
+ if (stream == nullptr) {
+ return Failure(
+ "Cannot find the task status update stream for task " +
+ stringify(taskId) + " of framework " + stringify(frameworkId));
+ }
+
+ // Get the corresponding update for this ACK.
+ const Result<StatusUpdate>& update = stream->next();
+ if (update.isError()) {
+ return Failure(update.error());
+ }
+
+ // This might happen if we retried a status update and got back
+ // acknowledgments for both the original and the retried update.
+ if (update.isNone()) {
+ return Failure(
+ "Unexpected task status update acknowledgment (UUID: " +
+ uuid.toString() + ") for task " + stringify(taskId) + " of framework " +
+ stringify(frameworkId));
+ }
+
+ // Handle the acknowledgement.
+ Try<bool> result =
+ stream->acknowledgement(taskId, frameworkId, uuid, update.get());
+
+ if (result.isError()) {
+ return Failure(result.error());
+ }
+
+ if (!result.get()) {
+ return Failure("Duplicate task status acknowledgement");
+ }
+
+ // Reset the timeout.
+ stream->timeout = None();
+
+ // Get the next update in the queue.
+ const Result<StatusUpdate>& next = stream->next();
+ if (next.isError()) {
+ return Failure(next.error());
+ }
+
+ bool terminated = stream->terminated;
+
+ if (terminated) {
+ if (next.isSome()) {
+ LOG(WARNING) << "Acknowledged a terminal"
+ << " task status update " << update.get()
+ << " but updates are still pending";
+ }
+ cleanupStatusUpdateStream(taskId, frameworkId);
+ } else if (!paused && next.isSome()) {
+ // Forward the next queued status update.
+ stream->timeout = forward(next.get(), STATUS_UPDATE_RETRY_INTERVAL_MIN);
+ }
+
+ return !terminated;
+}
+
+
+// TODO(vinod): There should be a limit on the retries.
+void TaskStatusUpdateManagerProcess::timeout(const Duration& duration)
+{
+ if (paused) {
+ return;
+ }
+
+ // Check and see if we should resend any status updates.
+ foreachkey (const FrameworkID& frameworkId, streams) {
+ foreachvalue (TaskStatusUpdateStream* stream, streams[frameworkId]) {
+ CHECK_NOTNULL(stream);
+ if (!stream->pending.empty()) {
+ CHECK_SOME(stream->timeout);
+ if (stream->timeout->expired()) {
+ const StatusUpdate& update = stream->pending.front();
+ LOG(WARNING) << "Resending task status update " << update;
+
+ // Bounded exponential backoff.
+ Duration duration_ =
+ std::min(duration * 2, STATUS_UPDATE_RETRY_INTERVAL_MAX);
+
+ stream->timeout = forward(update, duration_);
+ }
+ }
+ }
+ }
+}
+
+
+TaskStatusUpdateStream*
+TaskStatusUpdateManagerProcess::createStatusUpdateStream(
+ const TaskID& taskId,
+ const FrameworkID& frameworkId,
+ const SlaveID& slaveId,
+ bool checkpoint,
+ const Option<ExecutorID>& executorId,
+ const Option<ContainerID>& containerId)
+{
+ VLOG(1) << "Creating StatusUpdate stream for task " << taskId
+ << " of framework " << frameworkId;
+
+ TaskStatusUpdateStream* stream = new TaskStatusUpdateStream(
+ taskId, frameworkId, slaveId, flags, checkpoint, executorId, containerId);
+
+ streams[frameworkId][taskId] = stream;
+ return stream;
+}
+
+
+TaskStatusUpdateStream* TaskStatusUpdateManagerProcess::getStatusUpdateStream(
+ const TaskID& taskId,
+ const FrameworkID& frameworkId)
+{
+ if (!streams.contains(frameworkId)) {
+ return nullptr;
+ }
+
+ if (!streams[frameworkId].contains(taskId)) {
+ return nullptr;
+ }
+
+ return streams[frameworkId][taskId];
+}
+
+
+void TaskStatusUpdateManagerProcess::cleanupStatusUpdateStream(
+ const TaskID& taskId,
+ const FrameworkID& frameworkId)
+{
+ VLOG(1) << "Cleaning up status update stream"
+ << " for task " << taskId
+ << " of framework " << frameworkId;
+
+ CHECK(streams.contains(frameworkId))
+ << "Cannot find the task status update streams for framework "
+ << frameworkId;
+
+ CHECK(streams[frameworkId].contains(taskId))
+ << "Cannot find the status update streams for task " << taskId;
+
+ TaskStatusUpdateStream* stream = streams[frameworkId][taskId];
+
+ streams[frameworkId].erase(taskId);
+ if (streams[frameworkId].empty()) {
+ streams.erase(frameworkId);
+ }
+
+ delete stream;
+}
+
+
+TaskStatusUpdateManager::TaskStatusUpdateManager(const Flags& flags)
+{
+ process = new TaskStatusUpdateManagerProcess(flags);
+ spawn(process);
+}
+
+
+TaskStatusUpdateManager::~TaskStatusUpdateManager()
+{
+ terminate(process);
+ wait(process);
+ delete process;
+}
+
+
+void TaskStatusUpdateManager::initialize(
+ const function<void(StatusUpdate)>& forward)
+{
+ dispatch(process, &TaskStatusUpdateManagerProcess::initialize, forward);
+}
+
+
+Future<Nothing> TaskStatusUpdateManager::update(
+ const StatusUpdate& update,
+ const SlaveID& slaveId,
+ const ExecutorID& executorId,
+ const ContainerID& containerId)
+{
+ return dispatch(
+ process,
+ &TaskStatusUpdateManagerProcess::update,
+ update,
+ slaveId,
+ executorId,
+ containerId);
+}
+
+
+Future<Nothing> TaskStatusUpdateManager::update(
+ const StatusUpdate& update,
+ const SlaveID& slaveId)
+{
+ return dispatch(
+ process,
+ &TaskStatusUpdateManagerProcess::update,
+ update,
+ slaveId);
+}
+
+
+Future<bool> TaskStatusUpdateManager::acknowledgement(
+ const TaskID& taskId,
+ const FrameworkID& frameworkId,
+ const UUID& uuid)
+{
+ return dispatch(
+ process,
+ &TaskStatusUpdateManagerProcess::acknowledgement,
+ taskId,
+ frameworkId,
+ uuid);
+}
+
+
+Future<Nothing> TaskStatusUpdateManager::recover(
+ const string& rootDir,
+ const Option<SlaveState>& state)
+{
+ return dispatch(
+ process, &TaskStatusUpdateManagerProcess::recover, rootDir, state);
+}
+
+
+void TaskStatusUpdateManager::pause()
+{
+ dispatch(process, &TaskStatusUpdateManagerProcess::pause);
+}
+
+
+void TaskStatusUpdateManager::resume()
+{
+ dispatch(process, &TaskStatusUpdateManagerProcess::resume);
+}
+
+
+void TaskStatusUpdateManager::cleanup(const FrameworkID& frameworkId)
+{
+ dispatch(process, &TaskStatusUpdateManagerProcess::cleanup, frameworkId);
+}
+
+
+TaskStatusUpdateStream::TaskStatusUpdateStream(
+ const TaskID& _taskId,
+ const FrameworkID& _frameworkId,
+ const SlaveID& _slaveId,
+ const Flags& _flags,
+ bool _checkpoint,
+ const Option<ExecutorID>& executorId,
+ const Option<ContainerID>& containerId)
+ : checkpoint(_checkpoint),
+ terminated(false),
+ taskId(_taskId),
+ frameworkId(_frameworkId),
+ slaveId(_slaveId),
+ flags(_flags),
+ error(None())
+{
+ if (checkpoint) {
+ CHECK_SOME(executorId);
+ CHECK_SOME(containerId);
+
+ path = paths::getTaskUpdatesPath(
+ paths::getMetaRootDir(flags.work_dir),
+ slaveId,
+ frameworkId,
+ executorId.get(),
+ containerId.get(),
+ taskId);
+
+ // Create the base updates directory, if it doesn't exist.
+ const string& dirName = Path(path.get()).dirname();
+ Try<Nothing> directory = os::mkdir(dirName);
+ if (directory.isError()) {
+ error = "Failed to create '" + dirName + "': " + directory.error();
+ return;
+ }
+
+ // Open the updates file.
+ // NOTE: We don't use `O_SYNC` here because we only read this file
+ // if the host did not crash. `os::write` success implies the kernel
+ // will have flushed our data to the page cache. This is sufficient
+ // for the recovery scenarios we use this data for.
+ Try<int_fd> result = os::open(
+ path.get(),
+ O_CREAT | O_WRONLY | O_APPEND | O_CLOEXEC,
+ S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH);
+
+ if (result.isError()) {
+ error = "Failed to open '" + path.get() +
+ "' for status updates: " + result.error();
+ return;
+ }
+
+ // We keep the file open through the lifetime of the task, because it
+ // makes it easy to append status update records to the file.
+ fd = result.get();
+ }
+}
+
+
+TaskStatusUpdateStream::~TaskStatusUpdateStream()
+{
+ if (fd.isSome()) {
+ Try<Nothing> close = os::close(fd.get());
+ if (close.isError()) {
+ CHECK_SOME(path);
+ LOG(ERROR) << "Failed to close file '" << path.get() << "': "
+ << close.error();
+ }
+ }
+}
+
+
+Try<bool> TaskStatusUpdateStream::update(const StatusUpdate& update)
+{
+ if (error.isSome()) {
+ return Error(error.get());
+ }
+
+ if (!update.has_uuid()) {
+ return Error("Task status update is missing 'uuid'");
+ }
+
+ // Check that this status update has not already been acknowledged.
+ // This could happen in the rare case when the slave received the ACK
+ // from the framework, died, but slave's ACK to the executor never made it!
+ if (acknowledged.contains(UUID::fromBytes(update.uuid()).get())) {
+ LOG(WARNING) << "Ignoring task status update " << update
+ << " that has already been acknowledged by the framework!";
+ return false;
+ }
+
+ // Check that this update hasn't already been received.
+ // This could happen if the slave receives a status update from an executor,
+ // then crashes after it writes it to disk but before it sends an ack.
+ if (received.contains(UUID::fromBytes(update.uuid()).get())) {
+ LOG(WARNING) << "Ignoring duplicate task status update " << update;
+ return false;
+ }
+
+ // Handle the update, checkpointing if necessary.
+ Try<Nothing> result = handle(update, StatusUpdateRecord::UPDATE);
+ if (result.isError()) {
+ return Error(result.error());
+ }
+
+ return true;
+}
+
+
+Try<bool> TaskStatusUpdateStream::acknowledgement(
+ const TaskID& taskId,
+ const FrameworkID& frameworkId,
+ const UUID& uuid,
+ const StatusUpdate& update)
+{
+ if (error.isSome()) {
+ return Error(error.get());
+ }
+
+ if (acknowledged.contains(uuid)) {
+ LOG(WARNING) << "Duplicate task status update acknowledgment (UUID: "
+ << uuid << ") for update " << update;
+ return false;
+ }
+
+ // This might happen if we retried a status update and got back
+ // acknowledgments for both the original and the retried update.
+ if (uuid != UUID::fromBytes(update.uuid()).get()) {
+ LOG(WARNING) << "Unexpected task status update acknowledgement (received "
+ << uuid << ", expecting "
+ << UUID::fromBytes(update.uuid()).get()
+ << ") for update " << update;
+ return false;
+ }
+
+ // Handle the ACK, checkpointing if necessary.
+ Try<Nothing> result = handle(update, StatusUpdateRecord::ACK);
+ if (result.isError()) {
+ return Error(result.error());
+ }
+
+ return true;
+}
+
+
+Result<StatusUpdate> TaskStatusUpdateStream::next()
+{
+ if (error.isSome()) {
+ return Error(error.get());
+ }
+
+ if (!pending.empty()) {
+ return pending.front();
+ }
+
+ return None();
+}
+
+
+Try<Nothing> TaskStatusUpdateStream::replay(
+ const std::vector<StatusUpdate>& updates,
+ const hashset<UUID>& acks)
+{
+ if (error.isSome()) {
+ return Error(error.get());
+ }
+
+ VLOG(1) << "Replaying task status update stream for task " << taskId;
+
+ foreach (const StatusUpdate& update, updates) {
+ // Handle the update.
+ _handle(update, StatusUpdateRecord::UPDATE);
+
+ // Check if the update has an ACK too.
+ if (acks.contains(UUID::fromBytes(update.uuid()).get())) {
+ _handle(update, StatusUpdateRecord::ACK);
+ }
+ }
+
+ return Nothing();
+}
+
+
+Try<Nothing> TaskStatusUpdateStream::handle(
+ const StatusUpdate& update,
+ const StatusUpdateRecord::Type& type)
+{
+ CHECK_NONE(error);
+
+ // Checkpoint the update if necessary.
+ if (checkpoint) {
+ LOG(INFO) << "Checkpointing " << type << " for task status update "
+ << update;
+
+ CHECK_SOME(fd);
+
+ StatusUpdateRecord record;
+ record.set_type(type);
+
+ if (type == StatusUpdateRecord::UPDATE) {
+ record.mutable_update()->CopyFrom(update);
+ } else {
+ record.set_uuid(update.uuid());
+ }
+
+ Try<Nothing> write = ::protobuf::write(fd.get(), record);
+ if (write.isError()) {
+ error = "Failed to write task status update " + stringify(update) +
+ " to '" + path.get() + "': " + write.error();
+ return Error(error.get());
+ }
+ }
+
+ // Now actually handle the update.
+ _handle(update, type);
+
+ return Nothing();
+}
+
+
+void TaskStatusUpdateStream::_handle(
+ const StatusUpdate& update,
+ const StatusUpdateRecord::Type& type)
+{
+ CHECK_NONE(error);
+
+ if (type == StatusUpdateRecord::UPDATE) {
+ // Record this update.
+ received.insert(UUID::fromBytes(update.uuid()).get());
+
+ // Add it to the pending updates queue.
+ pending.push(update);
+ } else {
+ // Record this ACK.
+ acknowledged.insert(UUID::fromBytes(update.uuid()).get());
+
+ // Remove the corresponding update from the pending queue.
+ pending.pop();
+
+ if (!terminated) {
+ terminated = protobuf::isTerminalState(update.status().state());
+ }
+ }
+}
+
+} // namespace slave {
+} // namespace internal {
+} // namespace mesos {
http://git-wip-us.apache.org/repos/asf/mesos/blob/62d11733/src/slave/task_status_update_manager.hpp
----------------------------------------------------------------------
diff --git a/src/slave/task_status_update_manager.hpp b/src/slave/task_status_update_manager.hpp
new file mode 100644
index 0000000..6bdb468
--- /dev/null
+++ b/src/slave/task_status_update_manager.hpp
@@ -0,0 +1,210 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef __TASK_STATUS_UPDATE_MANAGER_HPP__
+#define __TASK_STATUS_UPDATE_MANAGER_HPP__
+
+#include <queue>
+#include <string>
+
+#include <mesos/mesos.hpp>
+
+#include <process/future.hpp>
+#include <process/pid.hpp>
+#include <process/timeout.hpp>
+
+#include <stout/hashset.hpp>
+#include <stout/lambda.hpp>
+#include <stout/option.hpp>
+#include <stout/try.hpp>
+#include <stout/uuid.hpp>
+
+#include "messages/messages.hpp"
+
+#include "slave/flags.hpp"
+
+namespace mesos {
+namespace internal {
+namespace slave {
+
+// Forward declarations.
+
+namespace state {
+struct SlaveState;
+}
+
+class TaskStatusUpdateManagerProcess;
+struct TaskStatusUpdateStream;
+
+
+// TaskStatusUpdateManager is responsible for
+// 1) Reliably sending status updates to the master.
+// 2) Checkpointing the update to disk (optional).
+// 3) Sending ACKs to the executor (optional).
+// 4) Receiving ACKs from the scheduler.
+class TaskStatusUpdateManager
+{
+public:
+ TaskStatusUpdateManager(const Flags& flags);
+ virtual ~TaskStatusUpdateManager();
+
+ // Expects a callback 'forward' which gets called whenever there is
+ // a new status update that needs to be forwarded to the master.
+ void initialize(const lambda::function<void(StatusUpdate)>& forward);
+
+ // TODO(vinod): Come up with better names/signatures for the
+ // checkpointing and non-checkpointing 'update()' functions.
+ // Currently, it is not obvious that one version of 'update()'
+ // does checkpointing while the other doesn't.
+
+ // Checkpoints the status update and reliably sends the
+ // update to the master (and hence the scheduler).
+ // @return Whether the update is handled successfully
+ // (e.g. checkpointed).
+ process::Future<Nothing> update(
+ const StatusUpdate& update,
+ const SlaveID& slaveId,
+ const ExecutorID& executorId,
+ const ContainerID& containerId);
+
+ // Retries the update to the master (as long as the slave is
+ // alive), but does not checkpoint the update.
+ // @return Whether the update is handled successfully.
+ process::Future<Nothing> update(
+ const StatusUpdate& update,
+ const SlaveID& slaveId);
+
+ // Checkpoints the status update to disk if necessary.
+ // Also, sends the next pending status update, if any.
+ // @return True if the ACK is handled successfully (e.g., checkpointed)
+ // and the task's status update stream is not terminated.
+ // False same as above except the status update stream is terminated.
+ // Failed if there are any errors (e.g., duplicate, checkpointing).
+ process::Future<bool> acknowledgement(
+ const TaskID& taskId,
+ const FrameworkID& frameworkId,
+ const UUID& uuid);
+
+ // Recover status updates.
+ process::Future<Nothing> recover(
+ const std::string& rootDir,
+ const Option<state::SlaveState>& state);
+
+
+ // Pause sending updates.
+ // This is useful when the slave is disconnected because a
+ // disconnected slave will drop the updates.
+ void pause();
+
+ // Unpause and resend all the pending updates right away.
+ // This is useful when the updates were pending because there was
+ // no master elected (e.g., during recovery) or framework failed over.
+ void resume();
+
+ // Closes all the status update streams corresponding to this framework.
+ // NOTE: This stops retrying any pending status updates for this framework.
+ void cleanup(const FrameworkID& frameworkId);
+
+private:
+ TaskStatusUpdateManagerProcess* process;
+};
+
+
+// TaskStatusUpdateStream handles the status updates and acknowledgements
+// of a task, checkpointing them if necessary. It also holds the information
+// about received, acknowledged and pending status updates.
+// NOTE: A task is expected to have a globally unique ID across the lifetime
+// of a framework. In other words the tuple (taskId, frameworkId) should be
+// always unique.
+struct TaskStatusUpdateStream
+{
+ TaskStatusUpdateStream(const TaskID& _taskId,
+ const FrameworkID& _frameworkId,
+ const SlaveID& _slaveId,
+ const Flags& _flags,
+ bool _checkpoint,
+ const Option<ExecutorID>& executorId,
+ const Option<ContainerID>& containerId);
+
+ ~TaskStatusUpdateStream();
+
+ // This function handles the update, checkpointing if necessary.
+ // @return True if the update is successfully handled.
+ // False if the update is a duplicate.
+ // Error Any errors (e.g., checkpointing).
+ Try<bool> update(const StatusUpdate& update);
+
+ // This function handles the ACK, checkpointing if necessary.
+ // @return True if the acknowledgement is successfully handled.
+ // False if the acknowledgement is a duplicate.
+ // Error Any errors (e.g., checkpointing).
+ Try<bool> acknowledgement(
+ const TaskID& taskId,
+ const FrameworkID& frameworkId,
+ const UUID& uuid,
+ const StatusUpdate& update);
+
+ // Returns the next update (or none, if empty) in the queue.
+ Result<StatusUpdate> next();
+
+ // Replays the stream by sequentially handling an update and its
+ // corresponding ACK, if present.
+ Try<Nothing> replay(
+ const std::vector<StatusUpdate>& updates,
+ const hashset<UUID>& acks);
+
+ // TODO(vinod): Explore semantics to make these private.
+ const bool checkpoint;
+ bool terminated;
+ Option<process::Timeout> timeout; // Timeout for resending status update.
+ std::queue<StatusUpdate> pending;
+
+private:
+ // Handles the status update and writes it to disk, if necessary.
+ // TODO(vinod): The write has to be asynchronous to avoid status updates that
+ // are being checkpointed, blocking the processing of other updates.
+ // One solution is to wrap the protobuf::write inside async, but its probably
+ // too much of an overhead to spin up a new libprocess per status update?
+ // A better solution might be to be have async write capability for file io.
+ Try<Nothing> handle(
+ const StatusUpdate& update,
+ const StatusUpdateRecord::Type& type);
+
+ void _handle(
+ const StatusUpdate& update,
+ const StatusUpdateRecord::Type& type);
+
+ const TaskID taskId;
+ const FrameworkID frameworkId;
+ const SlaveID slaveId;
+
+ const Flags flags;
+
+ hashset<UUID> received;
+ hashset<UUID> acknowledged;
+
+ Option<std::string> path; // File path of the update stream.
+ Option<int_fd> fd; // File descriptor to the update stream.
+
+ Option<std::string> error; // Potential non-retryable error.
+};
+
+} // namespace slave {
+} // namespace internal {
+} // namespace mesos {
+
+
+#endif // __TASK_STATUS_UPDATE_MANAGER_HPP__
http://git-wip-us.apache.org/repos/asf/mesos/blob/62d11733/src/tests/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/src/tests/CMakeLists.txt b/src/tests/CMakeLists.txt
index db5e531..8997cc0 100644
--- a/src/tests/CMakeLists.txt
+++ b/src/tests/CMakeLists.txt
@@ -117,7 +117,7 @@ set(MESOS_TESTS_SRC
slave_tests.cpp
slave_validation_tests.cpp
sorter_tests.cpp
- status_update_manager_tests.cpp
+ task_status_update_manager_tests.cpp
uri_tests.cpp
uri_fetcher_tests.cpp
values_tests.cpp
http://git-wip-us.apache.org/repos/asf/mesos/blob/62d11733/src/tests/cluster.cpp
----------------------------------------------------------------------
diff --git a/src/tests/cluster.cpp b/src/tests/cluster.cpp
index 6111be4..b854904 100644
--- a/src/tests/cluster.cpp
+++ b/src/tests/cluster.cpp
@@ -84,7 +84,7 @@
#include "slave/flags.hpp"
#include "slave/gc.hpp"
#include "slave/slave.hpp"
-#include "slave/status_update_manager.hpp"
+#include "slave/task_status_update_manager.hpp"
#include "slave/containerizer/containerizer.hpp"
#include "slave/containerizer/fetcher.hpp"
http://git-wip-us.apache.org/repos/asf/mesos/blob/62d11733/src/tests/cluster.hpp
----------------------------------------------------------------------
diff --git a/src/tests/cluster.hpp b/src/tests/cluster.hpp
index b3b1348..d572a09 100644
--- a/src/tests/cluster.hpp
+++ b/src/tests/cluster.hpp
@@ -58,7 +58,7 @@
#include "slave/flags.hpp"
#include "slave/gc.hpp"
#include "slave/slave.hpp"
-#include "slave/status_update_manager.hpp"
+#include "slave/task_status_update_manager.hpp"
#include "slave/containerizer/containerizer.hpp"
#include "slave/containerizer/fetcher.hpp"
http://git-wip-us.apache.org/repos/asf/mesos/blob/62d11733/src/tests/mock_slave.cpp
----------------------------------------------------------------------
diff --git a/src/tests/mock_slave.cpp b/src/tests/mock_slave.cpp
index 5dd2188..90c4369 100644
--- a/src/tests/mock_slave.cpp
+++ b/src/tests/mock_slave.cpp
@@ -29,7 +29,7 @@
#include <stout/option.hpp>
#include "slave/slave.hpp"
-#include "slave/status_update_manager.hpp"
+#include "slave/task_status_update_manager.hpp"
#include "tests/mock_slave.hpp"