You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mesos.apache.org by vi...@apache.org on 2016/12/08 01:06:19 UTC
[4/9] mesos git commit: Changed how master represents "recovered"
frameworks.
Changed how master represents "recovered" frameworks.
After master failover, the new master doesn't know which frameworks were
registered with the previous master (because this information is not
currently stored in the registry). In the period after the master fails
over but before the framework scheduler has re-registered, the master
learns about the frameworks in the cluster when agents re-register (an
agent reports the FrameworkInfo for all of the frameworks it is running
when it re-registers).
Such frameworks were previously represented separately from the normal
list of frameworks in the master: the master kept a collection of
`FrameworkInfo` for these "recovered" frameworks.
This commit removes this separate collection of "recovered" frameworks.
Instead, the master now treats recovered frameworks very similarly to
frameworks that are registered but currently disconnected. For example,
recovered frameworks will now have a `Framework` object which tracks the
tasks/executors running under that framework; recovered frameworks will
be reported via the normal "frameworks" key when querying HTTP
endpoints. Similarly, "teardown" operations on recovered frameworks will
now work correctly (MESOS-6419).
This means there is no longer a concept of "orphan tasks" [1]: if the
master knows about a task, the task will be running under a framework
(albeit the framework might be recovered or disconnected). A new
"recovered" key has been added to various HTTP endpoints/APIs to
determine if a framework hasn't yet re-registered after master failover.
Another behavioral change is that, previously, a framework
re-registering after master failover was allowed to make arbitrary
changes to its `FrameworkInfo`. The master now only applies updates to
FrameworkInfo fields that can be safely changed (as was already done
when a framework fails over to a master that hasn't failed over). The
previous behavior qualifies as a bug.
[1] The exception here is if the cluster contains Mesos agents older
than 1.0, because old Mesos agents don't report `FrameworkInfo`s when
they re-register.
Review: https://reviews.apache.org/r/53897/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/0dbceafa
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/0dbceafa
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/0dbceafa
Branch: refs/heads/master
Commit: 0dbceafa3b7caba9e541cd64bce3d5421e3b1262
Parents: b9b6e9f
Author: Neil Conway <ne...@gmail.com>
Authored: Wed Dec 7 17:04:22 2016 -0800
Committer: Vinod Kone <vi...@gmail.com>
Committed: Wed Dec 7 17:05:44 2016 -0800
----------------------------------------------------------------------
CHANGELOG | 7 +
include/mesos/master/master.proto | 28 +-
include/mesos/v1/master/master.proto | 28 +-
src/master/http.cpp | 130 +++-----
src/master/master.cpp | 423 +++++++++++++++-----------
src/master/master.hpp | 57 +++-
src/tests/api_tests.cpp | 1 +
src/tests/fault_tolerance_tests.cpp | 122 ++++++++
src/tests/master_allocator_tests.cpp | 15 +-
src/tests/master_authorization_tests.cpp | 32 +-
src/tests/master_tests.cpp | 392 ++++++++++++++++++------
src/tests/partition_tests.cpp | 52 ++--
src/tests/teardown_tests.cpp | 126 ++++++++
13 files changed, 1017 insertions(+), 396 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/0dbceafa/CHANGELOG
----------------------------------------------------------------------
diff --git a/CHANGELOG b/CHANGELOG
index 7847fe6..fac4daa 100644
--- a/CHANGELOG
+++ b/CHANGELOG
@@ -5,6 +5,13 @@ Additional API Changes:
* [MESOS-6675] - Changed the allocator API to support adding inactive
frameworks. Custom allocator implementations will need to be updated.
+ * [MESOS-6419] - When a framework re-registers after master failover,
+ it is only allowed to change certain fields in its FrameworkInfo.
+ For example, changing "failover_timeout" is allowed, but changing
+ "role" is not. In previous Mesos releases, the same restrictions on
+ changes to FrameworkInfo were only enforced after framework
+ failover, not master failover.
+
Release Notes - Mesos - Version 1.1.1 (WIP)
-------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/0dbceafa/include/mesos/master/master.proto
----------------------------------------------------------------------
diff --git a/include/mesos/master/master.proto b/include/mesos/master/master.proto
index 966105c..0d33251 100644
--- a/include/mesos/master/master.proto
+++ b/include/mesos/master/master.proto
@@ -320,6 +320,13 @@ message Response {
required bool active = 2;
required bool connected = 3;
+ // If true, this framework was previously subscribed but hasn't
+ // yet re-subscribed after a master failover. Recovered frameworks
+ // are only reported if one or more agents running a task or
+ // executor for the framework have re-registered after master
+ // failover.
+ required bool recovered = 11;
+
optional TimeInfo registered_time = 4;
optional TimeInfo reregistered_time = 5;
optional TimeInfo unregistered_time = 6;
@@ -338,8 +345,14 @@ message Response {
// Frameworks that have been teared down.
repeated Framework completed_frameworks = 2;
- // Frameworks that have previously subscribed but haven't yet subscribed
- // after a master failover.
+ // This field previously contained frameworks that previously
+ // subscribed but haven't yet re-subscribed after a master failover.
+ // As of Mesos 1.2, this field will always be empty; recovered
+ // frameworks are now reported in the `frameworks` list with the
+ // `recovered` field set to true.
+ //
+ // TODO(neilc): Remove this field after a deprecation cycle starting
+ // in Mesos 1.2.
repeated FrameworkInfo recovered_frameworks = 3;
}
@@ -353,6 +366,11 @@ message Response {
}
repeated Executor executors = 1;
+
+ // As of Mesos 1.2, this field will always be empty.
+ //
+ // TODO(neilc): Remove this field after a deprecation cycle starting
+ // in Mesos 1.2.
repeated Executor orphan_executors = 2;
}
@@ -372,8 +390,10 @@ message Response {
// acknowledged by the scheduler.
repeated Task completed_tasks = 3;
- // Tasks belonging to frameworks that have not yet re-subscribed with
- // master (e.g., immediately after master failover).
+ // As of Mesos 1.2, this field will always be empty.
+ //
+ // TODO(neilc): Remove this field after a deprecation cycle starting
+ // in Mesos 1.2.
repeated Task orphan_tasks = 4;
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/0dbceafa/include/mesos/v1/master/master.proto
----------------------------------------------------------------------
diff --git a/include/mesos/v1/master/master.proto b/include/mesos/v1/master/master.proto
index 0e5a867..09a82af 100644
--- a/include/mesos/v1/master/master.proto
+++ b/include/mesos/v1/master/master.proto
@@ -321,6 +321,13 @@ message Response {
required bool active = 2;
required bool connected = 3;
+ // If true, this framework was previously subscribed but hasn't
+ // yet re-subscribed after a master failover. Recovered frameworks
+ // are only reported if one or more agents running a task or
+ // executor for the framework have re-registered after master
+ // failover.
+ required bool recovered = 11;
+
optional TimeInfo registered_time = 4;
optional TimeInfo reregistered_time = 5;
optional TimeInfo unregistered_time = 6;
@@ -339,8 +346,14 @@ message Response {
// Frameworks that have been teared down.
repeated Framework completed_frameworks = 2;
- // Frameworks that have previously subscribed but haven't yet subscribed
- // after a master failover.
+ // This field previously contained frameworks that previously
+ // subscribed but haven't yet re-subscribed after a master failover.
+ // As of Mesos 1.2, this field will always be empty; recovered
+ // frameworks are now reported in the `frameworks` list with the
+ // `recovered` field set to true.
+ //
+ // TODO(neilc): Remove this field after a deprecation cycle starting
+ // in Mesos 1.2.
repeated FrameworkInfo recovered_frameworks = 3;
}
@@ -354,6 +367,11 @@ message Response {
}
repeated Executor executors = 1;
+
+ // As of Mesos 1.2, this field will always be empty.
+ //
+ // TODO(neilc): Remove this field after a deprecation cycle starting
+ // in Mesos 1.2.
repeated Executor orphan_executors = 2;
}
@@ -373,8 +391,10 @@ message Response {
// acknowledged by the scheduler.
repeated Task completed_tasks = 3;
- // Tasks belonging to frameworks that have not yet re-subscribed with
- // master (e.g., immediately after master failover).
+ // As of Mesos 1.2, this field will always be empty.
+ //
+ // TODO(neilc): Remove this field after a deprecation cycle starting
+ // in Mesos 1.2.
repeated Task orphan_tasks = 4;
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/0dbceafa/src/master/http.cpp
----------------------------------------------------------------------
diff --git a/src/master/http.cpp b/src/master/http.cpp
index e415124..d410c2f 100644
--- a/src/master/http.cpp
+++ b/src/master/http.cpp
@@ -380,6 +380,8 @@ static void json(JSON::ObjectWriter* writer, const Summary<Framework>& summary)
writer->field("hostname", framework.info.hostname());
writer->field("webui_url", framework.info.webui_url());
writer->field("active", framework.active);
+ writer->field("connected", framework.connected());
+ writer->field("recovered", framework.recovered());
}
@@ -1343,8 +1345,12 @@ Future<Response> Master::Http::frameworks(
}
});
- // Model all currently unregistered frameworks. This can happen
- // when a framework has yet to re-register after master failover.
+ // Model all unregistered frameworks. Such frameworks are only
+ // possible if the cluster contains pre-1.0 agents.
+ //
+ // TODO(neilc): Remove this once we break compatibility with
+ // pre-1.0 agents.
+ //
// TODO(vinod): Need to filter these frameworks based on authorization!
// See TODO in `state()` for further details.
writer->field("unregistered_frameworks", [this](
@@ -1377,6 +1383,7 @@ mesos::master::Response::GetFrameworks::Framework model(
_framework.set_active(framework.active);
_framework.set_connected(framework.connected());
+ _framework.set_recovered(framework.recovered());
int64_t time = framework.registeredTime.duration().ns();
if (time != 0) {
@@ -1475,33 +1482,6 @@ mesos::master::Response::GetFrameworks Master::Http::_getFrameworks(
getFrameworks.add_completed_frameworks()->CopyFrom(model(*framework));
}
- foreachvalue (const Slave* slave, master->slaves.registered) {
- foreachkey (const FrameworkID& frameworkId, slave->tasks) {
- if (!master->frameworks.registered.contains(frameworkId)) {
- // TODO(haosdent): This logic should be simplified after
- // a deprecation cycle starting with 1.0 as after that
- // we can rely on `master->frameworks.recovered` containing
- // all FrameworkInfos.
- // Until then there are 3 cases:
- // - No authorization enabled: show all orphaned frameworks.
- // - Authorization enabled, but no FrameworkInfo present:
- // do not show orphaned frameworks.
- // - Authorization enabled, FrameworkInfo present: filter
- // based on `approveViewFrameworkInfo`.
- if (master->authorizer.isSome() &&
- (!master->frameworks.recovered.contains(frameworkId) ||
- !approveViewFrameworkInfo(
- frameworksApprover,
- master->frameworks.recovered[frameworkId]))) {
- continue;
- }
-
- getFrameworks.add_recovered_frameworks()->CopyFrom(
- master->frameworks.recovered[frameworkId]);
- }
- }
- }
-
return getFrameworks;
}
@@ -1602,7 +1582,11 @@ mesos::master::Response::GetExecutors Master::Http::_getExecutors(
}
}
- // Orphan executors.
+ // Orphan executors. Such executors are only possible if the cluster
+ // contains pre-1.0 agents.
+ //
+ // TODO(neilc): Remove this once we break compatibility with pre-1.0
+ // agents.
foreachvalue (const Slave* slave, master->slaves.registered) {
typedef hashmap<ExecutorID, ExecutorInfo> ExecutorMap;
foreachpair (const FrameworkID& frameworkId,
@@ -1610,22 +1594,11 @@ mesos::master::Response::GetExecutors Master::Http::_getExecutors(
slave->executors) {
foreachvalue (const ExecutorInfo& executorInfo, executors) {
if (!master->frameworks.registered.contains(frameworkId)) {
- // TODO(haosdent): This logic should be simplified after
- // a deprecation cycle starting with 1.0 as after that
- // we can rely on `master->frameworks.recovered` containing
- // all FrameworkInfos.
- // Until then there are 3 cases:
- // - No authorization enabled: show all orphaned executors.
- // - Authorization enabled, but no FrameworkInfo present:
- // do not show orphaned executors.
- // - Authorization enabled, FrameworkInfo present: filter
- // based on `approveViewExecutorInfo`.
- if (master->authorizer.isSome() &&
- (!master->frameworks.recovered.contains(frameworkId) ||
- !approveViewExecutorInfo(
- executorsApprover,
- executorInfo,
- master->frameworks.recovered[frameworkId]))) {
+ // If authorization is enabled, do not show any orphaned
+ // executors. We need the executor's FrameworkInfo to
+ // authorize it, but if we had its FrameworkInfo, it would
+ // not be orphaned.
+ if (master->authorizer.isSome()) {
continue;
}
@@ -2706,33 +2679,24 @@ Future<Response> Master::Http::state(
}
});
- // Model all of the orphan tasks.
+ // Model all of the orphan tasks. Such tasks are only possible
+ // if the cluster contains pre-1.0 agents.
+ //
+ // TODO(neilc): Remove this once we break compatibility with
+ // pre-1.0 agents.
writer->field("orphan_tasks", [this, &tasksApprover](
JSON::ArrayWriter* writer) {
- // Find those orphan tasks.
foreachvalue (const Slave* slave, master->slaves.registered) {
typedef hashmap<TaskID, Task*> TaskMap;
foreachvalue (const TaskMap& tasks, slave->tasks) {
foreachvalue (const Task* task, tasks) {
- CHECK_NOTNULL(task);
const FrameworkID& frameworkId = task->framework_id();
if (!master->frameworks.registered.contains(frameworkId)) {
- // TODO(joerg84): This logic should be simplified after
- // a deprecation cycle starting with 1.0 as after that
- // we can rely on 'master->frameworks.recovered' containing
- // all FrameworkInfos.
- // Until then there are 3 cases:
- // - No authorization enabled: show all orphaned tasks.
- // - Authorization enabled, but no FrameworkInfo present:
- // do not show orphaned tasks.
- // - Authorization enabled, FrameworkInfo present: filter
- // based on 'approveViewTask'.
- if (master->authorizer.isSome() &&
- (!master->frameworks.recovered.contains(frameworkId) ||
- !approveViewTask(
- tasksApprover,
- *task,
- master->frameworks.recovered[frameworkId]))) {
+ // If authorization is enabled, do not show any
+ // orphan tasks. We need the task's FrameworkInfo to
+ // authorize it, but if we had its FrameworkInfo, it
+ // would not be an orphan.
+ if (master->authorizer.isSome()) {
continue;
}
@@ -2743,8 +2707,12 @@ Future<Response> Master::Http::state(
}
});
- // Model all currently unregistered frameworks. This can happen
- // when a framework has yet to re-register after master failover.
+ // Model all unregistered frameworks. Such frameworks are only
+ // possible if the cluster contains pre-1.0 agents.
+ //
+ // TODO(neilc): Remove this once we break compatibility with
+ // pre-1.0 agents.
+ //
// TODO(vinod): Need to filter these frameworks based on authorization!
// See the TODO above for "orphan_tasks" for further details.
writer->field("unregistered_frameworks", [this](
@@ -3830,30 +3798,22 @@ mesos::master::Response::GetTasks Master::Http::_getTasks(
}
}
- // Orphan tasks.
+ // Orphan tasks. Such tasks are only possible if the cluster
+ // contains pre-1.0 agents.
+ //
+ // TODO(neilc): Remove this once we break compatibility with pre-1.0
+ // agents.
foreachvalue (const Slave* slave, master->slaves.registered) {
typedef hashmap<TaskID, Task*> TaskMap;
foreachvalue (const TaskMap& tasks, slave->tasks) {
foreachvalue (const Task* task, tasks) {
- CHECK_NOTNULL(task);
const FrameworkID& frameworkId = task->framework_id();
if (!master->frameworks.registered.contains(frameworkId)) {
- // TODO(joerg84): This logic should be simplified after
- // a deprecation cycle starting with 1.0 as after that
- // we can rely on `master->frameworks.recovered` containing
- // all FrameworkInfos.
- // Until then there are 3 cases:
- // - No authorization enabled: show all orphaned tasks.
- // - Authorization enabled, but no FrameworkInfo present:
- // do not show orphaned tasks.
- // - Authorization enabled, FrameworkInfo present: filter
- // based on `approveViewTask`.
- if (master->authorizer.isSome() &&
- (!master->frameworks.recovered.contains(frameworkId) ||
- !approveViewTask(
- tasksApprover,
- *task,
- master->frameworks.recovered[frameworkId]))) {
+ // If authorization is enabled, do not show any orphan
+ // tasks. We need the task's FrameworkInfo to authorize it,
+ // but if we had its FrameworkInfo, it would not be an
+ // orphan.
+ if (master->authorizer.isSome()) {
continue;
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/0dbceafa/src/master/master.cpp
----------------------------------------------------------------------
diff --git a/src/master/master.cpp b/src/master/master.cpp
index a3ad0b1..08a5733 100644
--- a/src/master/master.cpp
+++ b/src/master/master.cpp
@@ -1272,8 +1272,10 @@ void Master::exited(const UPID& pid)
// structs, its tasks transitioned to LOST and resources
// recovered.
//
- // NOTE: If the framework hasn't re-registered since the master
- // failed over, we assume the framework is checkpointing.
+ // NOTE: If the framework hasn't re-registered yet and no agents
+ // (>= Mesos 1.0) have re-registered that are running any of the
+ // framework's tasks, we can't determine if the framework has
+ // enabled checkpointing. In that case, we assume it has.
hashset<FrameworkID> frameworkIds =
slave->tasks.keys() | slave->executors.keys();
@@ -2477,16 +2479,26 @@ void Master::_subscribe(
return;
}
- // If we are here framework has already been assigned an id.
+ // If we are here the framework has already been assigned an id.
CHECK(!frameworkInfo.id().value().empty());
- if (frameworks.registered.contains(frameworkInfo.id())) {
- Framework* framework =
- CHECK_NOTNULL(frameworks.registered[frameworkInfo.id()]);
+ Framework* framework = getFramework(frameworkInfo.id());
- // It is now safe to update the framework fields since the request is now
- // guaranteed to be successful. We use the fields passed in during
- // subscription.
+ if (framework == nullptr) {
+ // The framework has not yet re-registered after master failover.
+ // Furthermore, no agents have re-registered running one of this
+ // framework's tasks. Reconstruct a `Framework` object from the
+ // supplied `FrameworkInfo`.
+ recoverFramework(frameworkInfo);
+
+ framework = getFramework(frameworkInfo.id());
+ }
+
+ CHECK_NOTNULL(framework);
+
+ if (!framework->recovered()) {
+ // The framework has previously been registered with this master;
+ // it may or may not currently be connected.
LOG(INFO) << "Updating info for framework " << framework->id();
framework->updateFrameworkInfo(frameworkInfo);
@@ -2497,46 +2509,10 @@ void Master::_subscribe(
// Always failover the old framework connection. See MESOS-4712 for details.
failoverFramework(framework, http);
} else {
- // We don't have a framework with this ID, so we must be a newly
- // elected Mesos master to which either an existing scheduler or a
- // failed-over one is connecting. Create a Framework object and add
- // any tasks it has that have been reported by reconnecting slaves.
- Framework* framework = new Framework(this, flags, frameworkInfo, http);
-
- // Add active tasks and executors to the framework.
- foreachvalue (Slave* slave, slaves.registered) {
- if (slave->tasks.contains(framework->id())) {
- foreachvalue (Task* task, slave->tasks.at(framework->id())) {
- framework->addTask(task);
- }
- }
-
- if (slave->executors.contains(framework->id())) {
- foreachvalue (const ExecutorInfo& executor,
- slave->executors.at(framework->id())) {
- framework->addExecutor(slave->id, executor);
- }
- }
- }
-
- // NOTE: Need to add the framework _after_ we add its tasks
- // (above) so that we can properly determine the resources it's
- // currently using!
- addFramework(framework);
-
- FrameworkReregisteredMessage message;
- message.mutable_framework_id()->MergeFrom(framework->id());
- message.mutable_master_info()->MergeFrom(info_);
- framework->send(message);
-
- // Start the heartbeat after sending SUBSCRIBED event.
- framework->heartbeat();
+ // The framework has not yet re-registered after master failover.
+ activateRecoveredFramework(framework, frameworkInfo, None(), http);
}
- CHECK(frameworks.registered.contains(frameworkInfo.id()))
- << "Unknown framework " << frameworkInfo.id()
- << " (" << frameworkInfo.name() << ")";
-
// Broadcast the new framework pid to all the slaves. We have to
// broadcast because an executor might be running on a slave but
// it currently isn't running any tasks.
@@ -2764,19 +2740,33 @@ void Master::_subscribe(
return;
}
- // If we are here framework has already been assigned an id.
+ // If we are here the framework has already been assigned an id.
CHECK(!frameworkInfo.id().value().empty());
- if (frameworks.registered.contains(frameworkInfo.id())) {
+ Framework* framework = getFramework(frameworkInfo.id());
+
+ if (framework == nullptr) {
+ // The framework has not yet re-registered after master failover.
+ // Furthermore, no agents have re-registered running one of this
+ // framework's tasks. Reconstruct a `Framework` object from the
+ // supplied `FrameworkInfo`.
+ recoverFramework(frameworkInfo);
+
+ framework = getFramework(frameworkInfo.id());
+ }
+
+ CHECK_NOTNULL(framework);
+
+ if (!framework->recovered()) {
+ // The framework has previously been registered with this master;
+ // it may or may not currently be connected.
+ //
// Using the "force" field of the scheduler allows us to keep a
// scheduler that got partitioned but didn't die (in ZooKeeper
// speak this means didn't lose their session) and then
// eventually tried to connect to this master even though
// another instance of their scheduler has reconnected.
- Framework* framework =
- CHECK_NOTNULL(frameworks.registered[frameworkInfo.id()]);
-
// Test for the error case first.
if ((framework->pid != from) && !force) {
LOG(ERROR) << "Disallowing subscription attempt of"
@@ -2858,47 +2848,10 @@ void Master::_subscribe(
return;
}
} else {
- // We don't have a framework with this ID, so we must be a newly
- // elected Mesos master to which either an existing scheduler or a
- // failed-over one is connecting. Create a Framework object and add
- // any tasks it has that have been reported by reconnecting slaves.
- Framework* framework = new Framework(this, flags, frameworkInfo, from);
-
- // Add active tasks and executors to the framework.
- foreachvalue (Slave* slave, slaves.registered) {
- if (slave->tasks.contains(framework->id())) {
- foreachvalue (Task* task, slave->tasks.at(framework->id())) {
- framework->addTask(task);
- }
- }
-
- if (slave->executors.contains(framework->id())) {
- foreachvalue (const ExecutorInfo& executor,
- slave->executors.at(framework->id())) {
- framework->addExecutor(slave->id, executor);
- }
- }
- }
-
- // N.B. Need to add the framework _after_ we add its tasks
- // (above) so that we can properly determine the resources it's
- // currently using!
- addFramework(framework);
-
- // TODO(bmahler): We have to send a registered message here for
- // the re-registering framework, per the API contract. Send
- // re-register here per MESOS-786; requires deprecation or it
- // will break frameworks.
- FrameworkRegisteredMessage message;
- message.mutable_framework_id()->MergeFrom(framework->id());
- message.mutable_master_info()->MergeFrom(info_);
- framework->send(message);
+ // The framework has not yet re-registered after master failover.
+ activateRecoveredFramework(framework, frameworkInfo, from, None());
}
- CHECK(frameworks.registered.contains(frameworkInfo.id()))
- << "Unknown framework " << frameworkInfo.id()
- << " (" << frameworkInfo.name() << ")";
-
// Broadcast the new framework pid to all the slaves. We have to
// broadcast because an executor might be running on a slave but
// it currently isn't running any tasks.
@@ -5561,33 +5514,25 @@ void Master::__reregisterSlave(
hashset<FrameworkID> ids;
// TODO(joerg84): Remove this after a deprecation cycle starting
- // with the 1.0 release. It is only required if an older
- // (pre 1.0 agent) reregisters with a newer master.
- // In that case the agent does not have the 'frameworks' message
- // set which is used below to retrieve the framework information.
+ // with the 1.0 release. It is only required if an older (pre 1.0)
+ // agent reregisters with a newer master. In that case the agent
+ // does not have the `frameworks` field set which is used below to
+ // retrieve the framework information.
foreach (const Task& task, tasks) {
- Framework* framework = getFramework(task.framework_id());
-
- if (framework != nullptr && !ids.contains(framework->id())) {
- UpdateFrameworkMessage message;
- message.mutable_framework_id()->MergeFrom(framework->id());
-
- // TODO(anand): We set 'pid' to UPID() for http frameworks
- // as 'pid' was made optional in 0.24.0. In 0.25.0, we
- // no longer have to set pid here for http frameworks.
- message.set_pid(framework->pid.getOrElse(UPID()));
-
- send(slave->pid, message);
-
- ids.insert(framework->id());
- }
+ ids.insert(task.framework_id());
}
foreach (const FrameworkInfo& frameworkInfo, frameworks) {
CHECK(frameworkInfo.has_id());
- Framework* framework = getFramework(frameworkInfo.id());
+ ids.insert(frameworkInfo.id());
+ }
- if (framework != nullptr && !ids.contains(framework->id())) {
+ foreach (const FrameworkID& frameworkId, ids) {
+ Framework* framework = getFramework(frameworkId);
+
+ // TODO(neilc): It doesn't seem useful to send the PIDs of
+ // frameworks that are not connected.
+ if (framework != nullptr) {
UpdateFrameworkMessage message;
message.mutable_framework_id()->MergeFrom(framework->id());
@@ -5597,15 +5542,25 @@ void Master::__reregisterSlave(
message.set_pid(framework->pid.getOrElse(UPID()));
send(slave->pid, message);
+ }
+ }
- ids.insert(framework->id());
- } else {
- // The framework hasn't yet reregistered with the master,
- // hence we store the 'FrameworkInfo' in the recovered list.
- // TODO(joerg84): Consider recovering this information from
- // registrar instead of from agents.
- this->frameworks.recovered[frameworkInfo.id()] = frameworkInfo;
+ // If the agent is running any frameworks the master doesn't know
+ // about, recover the framework using the `FrameworkInfo` supplied
+ // by the agent. We don't recover frameworks that have already
+ // completed at the master.
+ foreach (const FrameworkInfo& frameworkInfo, frameworks) {
+ if (getFramework(frameworkInfo.id()) != nullptr) {
+ continue;
}
+ if (isCompletedFramework(frameworkInfo.id())) {
+ continue;
+ }
+
+ LOG(INFO) << "Recovering framework " << frameworkInfo.id()
+ << " from re-registering agent " << *slave;
+
+ recoverFramework(frameworkInfo);
}
// NOTE: Here we always send the message. Slaves whose version are
@@ -5940,11 +5895,13 @@ void Master::exitedExecutor(
// TODO(vinod): Reliably forward this message to the scheduler.
Framework* framework = getFramework(frameworkId);
- if (framework == nullptr) {
+ if (framework == nullptr || !framework->connected()) {
+ string status = (framework == nullptr ? "unknown" : "disconnected");
+
LOG(WARNING)
<< "Not forwarding exited executor message for executor '" << executorId
<< "' of framework " << frameworkId << " on agent " << *slave
- << " because the framework is unknown";
+ << " because the framework is " << status;
return;
}
@@ -6088,24 +6045,13 @@ void Master::_markUnreachable(
foreachkey (const FrameworkID& frameworkId, utils::copy(slave->tasks)) {
Framework* framework = getFramework(frameworkId);
- // If the framework has not yet re-registered after master failover,
- // its FrameworkInfo will be in the `recovered` collection. Note that
- // if the master knows about a task, its FrameworkInfo must appear in
- // either the `registered` or `recovered` collections.
- //
- // NOTE: If the framework is only running tasks on pre-1.0 agents
- // and the framework hasn't yet re-registered, its FrameworkInfo
- // will not appear in `recovered`. We can't accurately determine
- // whether the framework is partition-aware; we assume it is NOT
- // partition-aware, since using TASK_LOST ensures compatibility
- // with the previous (and default) Mesos behavior.
- Option<FrameworkInfo> frameworkInfo;
-
- if (framework != nullptr) {
- frameworkInfo = framework->info;
- } else if (frameworks.recovered.contains(frameworkId)) {
- frameworkInfo = frameworks.recovered[frameworkId];
- } else {
+ if (framework == nullptr) {
+ // If the framework is only running tasks on pre-1.0 agents and
+ // the framework hasn't yet re-registered, the master doesn't
+ // have a FrameworkInfo for it, and hence we cannot accurately
+ // determine whether the framework is partition-aware. We assume
+ // it is NOT partition-aware, since using TASK_LOST ensures
+ // compatibility with the previous (and default) Mesos behavior.
LOG(WARNING) << "Unable to determine if framework " << frameworkId
<< " is partition-aware, because the cluster contains"
<< " agents running an old version of Mesos; upgrading"
@@ -6113,8 +6059,8 @@ void Master::_markUnreachable(
}
TaskState newTaskState = TASK_UNREACHABLE;
- if (frameworkInfo.isNone() || !protobuf::frameworkHasCapability(
- frameworkInfo.get(), FrameworkInfo::Capability::PARTITION_AWARE)) {
+ if (framework == nullptr || !protobuf::frameworkHasCapability(
+ framework->info, FrameworkInfo::Capability::PARTITION_AWARE)) {
newTaskState = TASK_LOST;
}
@@ -6138,9 +6084,12 @@ void Master::_markUnreachable(
updateTask(task, update);
removeTask(task);
- if (framework == nullptr) {
+ if (framework == nullptr || !framework->connected()) {
+ string status = (framework == nullptr ? "unknown" : "disconnected");
+
LOG(WARNING) << "Dropping update " << update
- << " for unknown framework " << frameworkId;
+ << " for " << status
+ << " framework " << frameworkId;
} else {
forward(update, UPID(), framework);
}
@@ -6917,8 +6866,8 @@ void Master::reconcileKnownSlave(
// master has a task that the agent doesn't know about, the
// framework must have reregistered with this master since the
// last master failover.
- Framework* framework = getFramework(frameworkId);
- CHECK_NOTNULL(framework);
+ Framework* framework = CHECK_NOTNULL(getFramework(frameworkId));
+ CHECK(!framework->recovered());
reconcile.mutable_framework_id()->CopyFrom(frameworkId);
reconcile.mutable_framework()->CopyFrom(framework->info);
@@ -6990,9 +6939,11 @@ void Master::reconcileKnownSlave(
// Send ShutdownFrameworkMessages for frameworks that are completed.
// This could happen if the message wasn't received by the slave
// (e.g., slave was down, partitioned).
+ //
// NOTE: This is a short-term hack because this information is lost
// when the master fails over. Also, we only store a limited number
// of completed frameworks.
+ //
// TODO(vinod): Revisit this when registrar is in place. It would
// likely involve storing this information in the registrar.
foreach (const shared_ptr<Framework>& framework, frameworks.completed) {
@@ -7018,18 +6969,17 @@ void Master::addFramework(Framework* framework)
frameworks.registered[framework->id()] = framework;
- // Remove from 'frameworks.recovered' if necessary.
- frameworks.recovered.erase(framework->id());
-
- if (framework->pid.isSome()) {
- link(framework->pid.get());
- } else {
- CHECK_SOME(framework->http);
+ if (framework->connected()) {
+ if (framework->pid.isSome()) {
+ link(framework->pid.get());
+ } else {
+ CHECK_SOME(framework->http);
- const HttpConnection& http = framework->http.get();
+ const HttpConnection& http = framework->http.get();
- http.closed()
- .onAny(defer(self(), &Self::exited, framework->id(), http));
+ http.closed()
+ .onAny(defer(self(), &Self::exited, framework->id(), http));
+ }
}
auto addFrameworkRole = [this](Framework* framework, const string& role) {
@@ -7086,6 +7036,118 @@ void Master::addFramework(Framework* framework)
}
+void Master::recoverFramework(const FrameworkInfo& info)
+{
+ CHECK(!frameworks.registered.contains(info.id()));
+
+ Framework* framework = new Framework(this, flags, info);
+
+ // Add active tasks and executors to the framework.
+ foreachvalue (Slave* slave, slaves.registered) {
+ if (slave->tasks.contains(framework->id())) {
+ foreachvalue (Task* task, slave->tasks.at(framework->id())) {
+ framework->addTask(task);
+ }
+ }
+
+ if (slave->executors.contains(framework->id())) {
+ foreachvalue (const ExecutorInfo& executor,
+ slave->executors.at(framework->id())) {
+ framework->addExecutor(slave->id, executor);
+ }
+ }
+ }
+
+ addFramework(framework);
+}
+
+
+void Master::activateRecoveredFramework(
+ Framework* framework,
+ const FrameworkInfo& frameworkInfo,
+ const Option<UPID>& pid,
+ const Option<HttpConnection>& http)
+{
+ // Exactly one of `pid` or `http` must be provided.
+ CHECK(pid.isSome() != http.isSome());
+
+ CHECK_NOTNULL(framework);
+ CHECK(framework->recovered());
+ CHECK(!framework->active);
+ CHECK(framework->offers.empty());
+ CHECK(framework->inverseOffers.empty());
+ CHECK(framework->pid.isNone());
+ CHECK(framework->http.isNone());
+
+ // The `FrameworkInfo` might have changed.
+ LOG(INFO) << "Updating info for framework " << framework->id();
+
+ framework->updateFrameworkInfo(frameworkInfo);
+ allocator->updateFramework(framework->id(), framework->info);
+
+ // Updating `registeredTime` here is debatable: ideally,
+ // `registeredTime` would be the time at which the framework first
+ // registered with the master. However, we cannot determine this
+ // because the time at which a framework first registered is not
+ // persisted across master failover.
+ framework->registeredTime = Clock::now();
+ framework->reregisteredTime = Clock::now();
+
+ // Update the framework's connection state.
+ framework->state = Framework::State::CONNECTED;
+
+ if (pid.isSome()) {
+ framework->updateConnection(pid.get());
+ link(pid.get());
+ } else {
+ framework->updateConnection(http.get());
+ http->closed()
+ .onAny(defer(self(), &Self::exited, framework->id(), http.get()));
+ }
+
+ // Activate the framework.
+ framework->active = true;
+ allocator->activateFramework(framework->id());
+
+ // Export framework metrics if a principal is specified in `FrameworkInfo`.
+ Option<string> principal = framework->info.has_principal()
+ ? Option<string>(framework->info.principal())
+ : None();
+
+ if (framework->pid.isSome()) {
+ CHECK(!frameworks.principals.contains(framework->pid.get()));
+ frameworks.principals.put(framework->pid.get(), principal);
+ }
+
+ // We expect the framework metrics for this principal to be created
+ // when the framework is recovered. This implies that the framework
+ // principal cannot change on re-registration, which is currently
+ // the case (MESOS-2842).
+ if (principal.isSome()) {
+ CHECK(metrics->frameworks.contains(principal.get()));
+ }
+
+ if (pid.isSome()) {
+ // TODO(bmahler): We have to send a registered message here for
+ // the re-registering framework, per the API contract. Send
+ // re-register here per MESOS-786; requires deprecation or it
+ // will break frameworks.
+ FrameworkRegisteredMessage message;
+ message.mutable_framework_id()->MergeFrom(framework->id());
+ message.mutable_master_info()->MergeFrom(info_);
+ framework->send(message);
+ } else {
+ FrameworkReregisteredMessage message;
+ message.mutable_framework_id()->MergeFrom(framework->id());
+ message.mutable_master_info()->MergeFrom(info_);
+ framework->send(message);
+
+ // Start the heartbeat after sending SUBSCRIBED event.
+ framework->heartbeat();
+ }
+}
+
+
void Master::failoverFramework(Framework* framework, const HttpConnection& http)
{
// Notify the old connected framework that it has failed over.
@@ -7380,9 +7442,6 @@ void Master::removeFramework(Framework* framework)
frameworks.registered.erase(framework->id());
allocator->removeFramework(framework->id());
- // Remove from 'frameworks.recovered' if necessary.
- frameworks.recovered.erase(framework->id());
-
// The completedFramework buffer now owns the framework pointer.
frameworks.completed.push_back(shared_ptr<Framework>(framework));
}
@@ -7419,7 +7478,10 @@ void Master::removeFramework(Slave* slave, Framework* framework)
updateTask(task, update);
removeTask(task);
- forward(update, UPID(), framework);
+
+ if (framework->connected()) {
+ forward(update, UPID(), framework);
+ }
}
}
@@ -7468,7 +7530,11 @@ void Master::addSlave(
foreachkey (const FrameworkID& frameworkId, slave->executors) {
Framework* framework = getFramework(frameworkId);
- // The framework might not be re-registered yet.
+ // If the framework has not re-registered yet and this is the
+ // first agent to re-register that is running the framework, we
+ // skip adding the framework's executors here. Instead, the
+ // framework will be recovered in `__reregisterSlave` and its
+ // executors will be added by `recoverFramework`.
if (framework == nullptr) {
continue;
}
@@ -7483,27 +7549,29 @@ void Master::addSlave(
foreachkey (const FrameworkID& frameworkId, slave->tasks) {
Framework* framework = getFramework(frameworkId);
+ // If the framework has not re-registered yet and this is the
+ // first agent to re-register that is running the framework, we
+ // skip adding the framework's tasks here. Instead, the framework
+ // will be recovered in `__reregisterSlave` and its tasks will be
+ // added by `recoverFramework`.
+ if (framework == nullptr) {
+ continue;
+ }
+
foreachvalue (Task* task, slave->tasks[frameworkId]) {
- // The framework might not be re-registered yet.
- if (framework != nullptr) {
- framework->addTask(task);
- } else {
- // TODO(benh): We should really put a timeout on how long we
- // keep tasks running on a slave that never have frameworks
- // reregister and claim them.
- LOG(WARNING) << "Possibly orphaned task " << task->task_id()
- << " of framework " << task->framework_id()
- << " running on agent " << *slave;
- }
+ framework->addTask(task);
}
}
// Re-add completed tasks reported by the slave.
+ //
// Note that a slave considers a framework completed when it has no
- // tasks/executors running for that framework. But a master considers a
- // framework completed when the framework is removed after a failover timeout.
- // TODO(vinod): Reconcile the notion of a completed framework across the
- // master and slave.
+ // tasks/executors running for that framework. But a master
+ // considers a framework completed when the framework is removed
+ // after a failover timeout.
+ //
+ // TODO(vinod): Reconcile the notion of a completed framework across
+ // the master and slave.
foreach (const Archive::Framework& completedFramework, completedFrameworks) {
Framework* framework = getFramework(
completedFramework.framework_info().id());
@@ -7515,7 +7583,8 @@ void Master::addSlave(
<< " that ran on agent " << *slave;
framework->addCompletedTask(task);
} else {
- // We could be here if the framework hasn't registered yet.
+ // The framework might not be re-registered yet.
+ //
// TODO(vinod): Revisit these semantics when we store frameworks'
// information in the registrar.
LOG(WARNING) << "Possibly orphaned completed task " << task.task_id()
@@ -7652,7 +7721,7 @@ void Master::_removeSlave(
updateTask(task, update);
removeTask(task);
- if (framework == nullptr) {
+ if (framework == nullptr || !framework->connected()) {
LOG(WARNING) << "Dropping update " << update
<< " for unknown framework " << frameworkId;
} else {
http://git-wip-us.apache.org/repos/asf/mesos/blob/0dbceafa/src/master/master.hpp
----------------------------------------------------------------------
diff --git a/src/master/master.hpp b/src/master/master.hpp
index 9f5929c..c0bdb80 100644
--- a/src/master/master.hpp
+++ b/src/master/master.hpp
@@ -643,9 +643,10 @@ protected:
Framework* framework,
const std::vector<TaskStatus>& statuses);
- // When a slave that is known to the master re-registers, we need to
- // reconcile the master's view of the slave's tasks and executors.
- // This function also sends the `ReregisterSlaveMessage`.
+ // When a slave that was previously registered with this master
+ // re-registers, we need to reconcile the master's view of the
+ // slave's tasks and executors. This function also sends the
+ // `ReregisterSlaveMessage`.
void reconcileKnownSlave(
Slave* slave,
const std::vector<ExecutorInfo>& executors,
@@ -654,6 +655,23 @@ protected:
// Add a framework.
void addFramework(Framework* framework);
+ // Recover a framework from its `FrameworkInfo`. This happens after
+ // master failover, when an agent running one of the framework's
+ // tasks re-registers or when the framework itself re-registers,
+ // whichever happens first. The result of this function is a
+ // registered, inactive framework with state `RECOVERED`.
+ void recoverFramework(const FrameworkInfo& info);
+
+ // Transition a framework from `RECOVERED` to `CONNECTED` state and
+ // activate it. This happens at most once after master failover, the
+ // first time that the framework re-registers with the new master.
+ // Exactly one of `newPid` or `http` must be provided.
+ void activateRecoveredFramework(
+ Framework* framework,
+ const FrameworkInfo& frameworkInfo,
+ const Option<process::UPID>& pid,
+ const Option<HttpConnection>& http);
+
// Replace the scheduler for a framework with a new process ID, in
// the event of a scheduler failover.
void failoverFramework(Framework* framework, const process::UPID& newPid);
@@ -1746,10 +1764,6 @@ private:
hashmap<FrameworkID, Framework*> registered;
- // `recovered` contains 'FrameworkInfo's for frameworks that have
- // not yet re-registered after master failover.
- hashmap<FrameworkID, FrameworkInfo> recovered;
-
boost::circular_buffer<std::shared_ptr<Framework>> completed;
// Principals of frameworks keyed by PID.
@@ -2190,7 +2204,13 @@ struct Framework
// Framework was previously connected to this master. A framework
// becomes disconnected when there is a socket error.
- DISCONNECTED
+ DISCONNECTED,
+
+ // Framework has never connected to this master. This implies the
+ // master failed over and the framework has not yet re-registered,
+ // but some framework state has been recovered from re-registering
+ // agents that are running tasks for the framework.
+ RECOVERED
};
Framework(Master* const _master,
@@ -2221,6 +2241,15 @@ struct Framework
reregisteredTime(time),
completedTasks(masterFlags.max_completed_tasks_per_framework) {}
+ Framework(Master* const _master,
+ const Flags& masterFlags,
+ const FrameworkInfo& _info)
+ : master(_master),
+ info(_info),
+ state(RECOVERED),
+ active(false),
+ completedTasks(masterFlags.max_completed_tasks_per_framework) {}
+
~Framework()
{
if (http.isSome()) {
@@ -2477,7 +2506,7 @@ struct Framework
// Wipe the PID if this is an upgrade from PID to HTTP.
// TODO(benh): unlink(oldPid);
pid = None();
- } else {
+ } else if (http.isSome()) {
// Cleanup the old HTTP connection.
// Note that master creates a new HTTP connection for every
// subscribe request, so 'newHttp' should always be different
@@ -2527,14 +2556,16 @@ struct Framework
bool connected() const { return state == CONNECTED; }
+ bool recovered() const { return state == RECOVERED; }
+
Master* const master;
FrameworkInfo info;
- // Frameworks can either be connected via HTTP or by message
- // passing (scheduler driver). Exactly one of 'http' and 'pid'
- // will be set according to the last connection made by the
- // framework.
+ // Frameworks can either be connected via HTTP or by message passing
+ // (scheduler driver). At most one of `http` and `pid` will be set
+ // according to the last connection made by the framework; neither
+ // field will be set if the framework is in state `RECOVERED`.
Option<HttpConnection> http;
Option<process::UPID> pid;
http://git-wip-us.apache.org/repos/asf/mesos/blob/0dbceafa/src/tests/api_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/api_tests.cpp b/src/tests/api_tests.cpp
index 21aefb2..4bc2c74 100644
--- a/src/tests/api_tests.cpp
+++ b/src/tests/api_tests.cpp
@@ -241,6 +241,7 @@ TEST_P(MasterAPITest, GetFrameworks)
ASSERT_FALSE(frameworks.frameworks(0).framework_info().checkpoint());
ASSERT_TRUE(frameworks.frameworks(0).active());
ASSERT_TRUE(frameworks.frameworks(0).connected());
+ ASSERT_FALSE(frameworks.frameworks(0).recovered());
driver.stop();
driver.join();
http://git-wip-us.apache.org/repos/asf/mesos/blob/0dbceafa/src/tests/fault_tolerance_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/fault_tolerance_tests.cpp b/src/tests/fault_tolerance_tests.cpp
index 59f68f6..277dcfa 100644
--- a/src/tests/fault_tolerance_tests.cpp
+++ b/src/tests/fault_tolerance_tests.cpp
@@ -873,6 +873,8 @@ TEST_F(FaultToleranceTest, FrameworkReregister)
JSON::Object framework = frameworks.values.front().as<JSON::Object>();
EXPECT_TRUE(framework.values["active"].as<JSON::Boolean>().value);
+ EXPECT_TRUE(framework.values["connected"].as<JSON::Boolean>().value);
+ EXPECT_FALSE(framework.values["recovered"].as<JSON::Boolean>().value);
EXPECT_EQ(
static_cast<int64_t>(registerTime.secs()),
@@ -2123,6 +2125,126 @@ TEST_F(FaultToleranceTest, UpdateFrameworkInfoOnSchedulerFailover)
EXPECT_EQ(DRIVER_STOPPED, driver1.join());
}
+
+// This test verifies that when a framework re-registers after master
+// failover with an updated FrameworkInfo, the updated FrameworkInfo
+// is reflected in the master.
+TEST_F(FaultToleranceTest, UpdateFrameworkInfoOnMasterFailover)
+{
+ Try<Owned<cluster::Master>> master = StartMaster();
+ ASSERT_SOME(master);
+
+ StandaloneMasterDetector detector(master.get()->pid);
+ Try<Owned<cluster::Slave>> slave = StartSlave(&detector);
+ ASSERT_SOME(slave);
+
+ FrameworkInfo frameworkInfo1 = DEFAULT_FRAMEWORK_INFO;
+ frameworkInfo1.set_name("Framework 1");
+ frameworkInfo1.set_failover_timeout(1000);
+ frameworkInfo1.set_checkpoint(true);
+
+ MockScheduler sched1;
+ MesosSchedulerDriver driver1(
+ &sched1, frameworkInfo1, master.get()->pid, DEFAULT_CREDENTIAL);
+
+ Future<FrameworkID> frameworkId;
+ EXPECT_CALL(sched1, registered(&driver1, _, _))
+ .WillOnce(FutureArg<1>(&frameworkId));
+
+ Future<vector<Offer>> offers;
+ EXPECT_CALL(sched1, resourceOffers(&driver1, _))
+ .WillOnce(FutureArg<1>(&offers));
+
+ driver1.start();
+
+ AWAIT_READY(frameworkId);
+
+ AWAIT_READY(offers);
+ ASSERT_FALSE(offers->empty());
+
+ Offer offer = offers.get()[0];
+
+ TaskInfo task = createTask(offer, "sleep 60");
+
+ Future<TaskStatus> runningStatus;
+ EXPECT_CALL(sched1, statusUpdate(&driver1, _))
+ .WillOnce(FutureArg<1>(&runningStatus));
+
+ Future<Nothing> statusUpdateAck = FUTURE_DISPATCH(
+ slave.get()->pid, &Slave::_statusUpdateAcknowledgement);
+
+ driver1.launchTasks(offer.id(), {task});
+
+ AWAIT_READY(runningStatus);
+ EXPECT_EQ(TASK_RUNNING, runningStatus->state());
+ EXPECT_EQ(task.task_id(), runningStatus->task_id());
+
+ AWAIT_READY(statusUpdateAck);
+
+ // Stop the master.
+ master->reset();
+
+ // While the master is down, shutdown the first framework.
+ driver1.stop();
+ driver1.join();
+
+ // Restart master; ensure the agent re-registers before the second
+ // scheduler connects. This ensures the first framework's
+ // FrameworkInfo is recovered from the re-registering agent.
+ Future<SlaveReregisteredMessage> slaveReregistered = FUTURE_PROTOBUF(
+ SlaveReregisteredMessage(), _, slave.get()->pid);
+
+ master = StartMaster();
+ ASSERT_SOME(master);
+
+ detector.appoint(master.get()->pid);
+
+ AWAIT_READY(slaveReregistered);
+
+ // Connect a new scheduler that uses the same framework ID but a
+ // different `FrameworkInfo`. Note that updating the `checkpoint`
+ // field is NOT supported, so we expect the master to ignore the
+ // updated version of this field.
+ FrameworkInfo frameworkInfo2 = DEFAULT_FRAMEWORK_INFO;
+ frameworkInfo2.mutable_id()->MergeFrom(frameworkId.get());
+ frameworkInfo2.set_name("Framework 2");
+ frameworkInfo2.set_failover_timeout(2000);
+ frameworkInfo2.set_checkpoint(false);
+
+ MockScheduler sched2;
+ MesosSchedulerDriver driver2(
+ &sched2, frameworkInfo2, master.get()->pid, DEFAULT_CREDENTIAL);
+
+ Future<Nothing> sched2Registered;
+ EXPECT_CALL(sched2, registered(&driver2, frameworkId.get(), _))
+ .WillOnce(FutureSatisfy(&sched2Registered));
+
+ driver2.start();
+
+ AWAIT_READY(sched2Registered);
+
+ Future<Response> response = process::http::get(
+ master.get()->pid,
+ "state",
+ None(),
+ createBasicAuthHeaders(DEFAULT_CREDENTIAL));
+
+ Try<JSON::Object> parse = JSON::parse<JSON::Object>(response->body);
+ ASSERT_SOME(parse);
+
+ JSON::Array frameworks = parse->values["frameworks"].as<JSON::Array>();
+ ASSERT_EQ(1u, frameworks.values.size());
+
+ JSON::Object framework = frameworks.values.front().as<JSON::Object>();
+ EXPECT_EQ(frameworkId.get(), framework.values["id"].as<JSON::String>().value);
+ EXPECT_EQ("Framework 2", framework.values["name"].as<JSON::String>().value);
+ EXPECT_EQ(2000, framework.values["failover_timeout"].as<JSON::Number>());
+ EXPECT_TRUE(framework.values["checkpoint"].as<JSON::Boolean>().value);
+
+ driver2.stop();
+ driver2.join();
+}
+
} // namespace tests {
} // namespace internal {
} // namespace mesos {
http://git-wip-us.apache.org/repos/asf/mesos/blob/0dbceafa/src/tests/master_allocator_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/master_allocator_tests.cpp b/src/tests/master_allocator_tests.cpp
index 266047b..34e2b49 100644
--- a/src/tests/master_allocator_tests.cpp
+++ b/src/tests/master_allocator_tests.cpp
@@ -1528,6 +1528,13 @@ TYPED_TEST(MasterAllocatorTest, SlaveReregistersFirst)
.WillOnce(DoAll(InvokeAddSlave(&allocator2),
FutureSatisfy(&addSlave)));
+ // Expect the framework to be added but to be inactive when the
+ // agent re-registers, until the framework re-registers below.
+ Future<Nothing> addFramework;
+ EXPECT_CALL(allocator2, addFramework(_, _, _, false))
+ .WillOnce(DoAll(InvokeAddFramework(&allocator2),
+ FutureSatisfy(&addFramework)));
+
Try<Owned<cluster::Master>> master2 = this->StartMaster(&allocator2);
ASSERT_SOME(master2);
@@ -1535,12 +1542,18 @@ TYPED_TEST(MasterAllocatorTest, SlaveReregistersFirst)
slaveDetector.appoint(master2.get()->pid);
AWAIT_READY(addSlave);
+ AWAIT_READY(addFramework);
EXPECT_CALL(sched, disconnected(_));
EXPECT_CALL(sched, registered(&driver, _, _));
- EXPECT_CALL(allocator2, addFramework(_, _, _, _));
+ // Because the framework was re-added above, we expect to only
+ // activate the framework when it re-registers.
+ EXPECT_CALL(allocator2, addFramework(_, _, _, _))
+ .Times(0);
+
+ EXPECT_CALL(allocator2, activateFramework(_));
Future<vector<Offer>> resourceOffers2;
EXPECT_CALL(sched, resourceOffers(&driver, _))
http://git-wip-us.apache.org/repos/asf/mesos/blob/0dbceafa/src/tests/master_authorization_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/master_authorization_tests.cpp b/src/tests/master_authorization_tests.cpp
index 06c6e47..f973cf4 100644
--- a/src/tests/master_authorization_tests.cpp
+++ b/src/tests/master_authorization_tests.cpp
@@ -2147,9 +2147,11 @@ TYPED_TEST(MasterAuthorizerTest, FilterRolesEndpoint)
// This test verifies that authorization based endpoint filtering
-// works correctly on the /state endpoint with orphaned tasks.
-// Both default users are allowed to to view high level frameworks, but only
-// one is allowed to view the tasks.
+// works correctly on the /state endpoint with orphaned tasks. Both
+// default users are allowed to to view high level frameworks, but
+// only one is allowed to view the tasks. Although the tasks are
+// "orphaned" (in the sense that their framework has not yet
+// re-registered), they are now reported under the "tasks" key.
TYPED_TEST(MasterAuthorizerTest, FilterOrphanedTasks)
{
ACLs acls;
@@ -2242,7 +2244,7 @@ TYPED_TEST(MasterAuthorizerTest, FilterOrphanedTasks)
EXPECT_CALL(exec, launchTask(_, _))
.WillOnce(SendStatusUpdateFromTask(TASK_RUNNING));
- // Wait until TASK_RUNNING of the task is received.
+ // Wait until TASK_RUNNING of the task is received.
AWAIT_READY(statusUpdate);
Future<SlaveReregisteredMessage> slaveReregisteredMessage =
@@ -2279,8 +2281,18 @@ TYPED_TEST(MasterAuthorizerTest, FilterOrphanedTasks)
ASSERT_SOME(parse);
JSON::Object tasks = parse.get();
+
+ // The "orphan_tasks" array should be empty, regardless of authorization.
ASSERT_TRUE(tasks.values["orphan_tasks"].is<JSON::Array>());
- EXPECT_EQ(1u, tasks.values["orphan_tasks"].as<JSON::Array>().values.size());
+ EXPECT_TRUE(tasks.values["orphan_tasks"].as<JSON::Array>().values.empty());
+
+ ASSERT_TRUE(tasks.values["frameworks"].is<JSON::Array>());
+
+ JSON::Array frameworks = tasks.values["frameworks"].as<JSON::Array>();
+ EXPECT_EQ(1u, frameworks.values.size());
+
+ JSON::Object framework = frameworks.values.front().as<JSON::Object>();
+ EXPECT_EQ(1u, framework.values["tasks"].as<JSON::Array>().values.size());
}
// Retrieve endpoint with the user allowed to view the framework,
@@ -2299,8 +2311,18 @@ TYPED_TEST(MasterAuthorizerTest, FilterOrphanedTasks)
ASSERT_SOME(parse);
JSON::Object tasks = parse.get();
+
+ // The "orphan_tasks" array should be empty, regardless of authorization.
ASSERT_TRUE(tasks.values["orphan_tasks"].is<JSON::Array>());
EXPECT_TRUE(tasks.values["orphan_tasks"].as<JSON::Array>().values.empty());
+
+ ASSERT_TRUE(tasks.values["frameworks"].is<JSON::Array>());
+
+ JSON::Array frameworks = tasks.values["frameworks"].as<JSON::Array>();
+ EXPECT_EQ(1u, frameworks.values.size());
+
+ JSON::Object framework = frameworks.values.front().as<JSON::Object>();
+ EXPECT_TRUE(framework.values["tasks"].as<JSON::Array>().values.empty());
}
EXPECT_CALL(exec, shutdown(_))
http://git-wip-us.apache.org/repos/asf/mesos/blob/0dbceafa/src/tests/master_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/master_tests.cpp b/src/tests/master_tests.cpp
index dfedbbd..5901614 100644
--- a/src/tests/master_tests.cpp
+++ b/src/tests/master_tests.cpp
@@ -2752,11 +2752,11 @@ TEST_F(MasterZooKeeperTest, MasterInfoAddress)
#endif // MESOS_HAS_JAVA
-// This test ensures that when a master fails over, those tasks that
-// belong to some currently unregistered frameworks will appear in the
-// "orphan_tasks" field in the state endpoint. And those unregistered
-// frameworks will appear in the "unregistered_frameworks" field.
-TEST_F(MasterTest, OrphanTasks)
+// This test ensures that when a master fails over, tasks that belong
+// to frameworks that have not re-registered will be reported in the
+// "/state" endpoint. The framework itself should have the "recovered"
+// field set to true.
+TEST_F(MasterTest, RecoveredFramework)
{
// Start a master.
Try<Owned<cluster::Master>> master = StartMaster();
@@ -2791,8 +2791,7 @@ TEST_F(MasterTest, OrphanTasks)
Future<TaskStatus> status;
EXPECT_CALL(sched, statusUpdate(&driver, _))
- .WillOnce(FutureArg<1>(&status))
- .WillRepeatedly(Return()); // Ignore subsequent updates.
+ .WillOnce(FutureArg<1>(&status));
driver.start();
@@ -2800,38 +2799,40 @@ TEST_F(MasterTest, OrphanTasks)
EXPECT_EQ(TASK_RUNNING, status.get().state());
// Get the master's state.
- Future<Response> response = process::http::get(
+ Future<Response> response1 = process::http::get(
master.get()->pid,
"state",
None(),
createBasicAuthHeaders(DEFAULT_CREDENTIAL));
- AWAIT_EXPECT_RESPONSE_STATUS_EQ(OK().status, response);
- AWAIT_EXPECT_RESPONSE_HEADER_EQ(APPLICATION_JSON, "Content-Type", response);
+ AWAIT_EXPECT_RESPONSE_STATUS_EQ(OK().status, response1);
+ AWAIT_EXPECT_RESPONSE_HEADER_EQ(APPLICATION_JSON, "Content-Type", response1);
- Try<JSON::Object> parse = JSON::parse<JSON::Object>(response.get().body);
- ASSERT_SOME(parse);
+ Try<JSON::Object> parse1 = JSON::parse<JSON::Object>(response1.get().body);
+ ASSERT_SOME(parse1);
- JSON::Object state = parse.get();
- // Record the original framework and task info.
- JSON::Array frameworks =
- state.values["frameworks"].as<JSON::Array>();
- JSON::Object activeFramework =
- frameworks.values.front().as<JSON::Object>();
- JSON::String activeFrameworkId =
- activeFramework.values["id"].as<JSON::String>();
- JSON::Array activeTasks =
- activeFramework.values["tasks"].as<JSON::Array>();
- JSON::Array orphanTasks =
- state.values["orphan_tasks"].as<JSON::Array>();
- JSON::Array unknownFrameworksArray =
- state.values["unregistered_frameworks"].as<JSON::Array>();
+ JSON::Array frameworks1 = parse1->values["frameworks"].as<JSON::Array>();
+ EXPECT_EQ(1u, frameworks1.values.size());
- EXPECT_EQ(1u, frameworks.values.size());
- EXPECT_EQ(1u, activeTasks.values.size());
- EXPECT_EQ(0u, orphanTasks.values.size());
- EXPECT_EQ(0u, unknownFrameworksArray.values.size());
- EXPECT_EQ(frameworkId.value(), activeFrameworkId.value);
+ JSON::Object activeFramework1 = frameworks1.values.front().as<JSON::Object>();
+
+ EXPECT_EQ(
+ frameworkId.value(),
+ activeFramework1.values["id"].as<JSON::String>().value);
+
+ EXPECT_TRUE(activeFramework1.values["active"].as<JSON::Boolean>().value);
+ EXPECT_TRUE(activeFramework1.values["connected"].as<JSON::Boolean>().value);
+ EXPECT_FALSE(activeFramework1.values["recovered"].as<JSON::Boolean>().value);
+
+ JSON::Array activeTasks1 = activeFramework1.values["tasks"].as<JSON::Array>();
+ EXPECT_EQ(1u, activeTasks1.values.size());
+
+ JSON::Array unregisteredFrameworks1 =
+ parse1->values["unregistered_frameworks"].as<JSON::Array>();
+
+ EXPECT_TRUE(unregisteredFrameworks1.values.empty());
+
+ EXPECT_TRUE(parse1->values["orphan_tasks"].as<JSON::Array>().values.empty());
EXPECT_CALL(sched, disconnected(&driver))
.Times(1);
@@ -2870,31 +2871,43 @@ TEST_F(MasterTest, OrphanTasks)
AWAIT_READY(subscribeCall);
// Get the master's state.
- response = process::http::get(
+ Future<Response> response2 = process::http::get(
master.get()->pid,
"state",
None(),
createBasicAuthHeaders(DEFAULT_CREDENTIAL));
- AWAIT_EXPECT_RESPONSE_STATUS_EQ(OK().status, response);
- AWAIT_EXPECT_RESPONSE_HEADER_EQ(APPLICATION_JSON, "Content-Type", response);
+ AWAIT_EXPECT_RESPONSE_STATUS_EQ(OK().status, response2);
+ AWAIT_EXPECT_RESPONSE_HEADER_EQ(APPLICATION_JSON, "Content-Type", response2);
- parse = JSON::parse<JSON::Object>(response.get().body);
- ASSERT_SOME(parse);
+ Try<JSON::Object> parse2 = JSON::parse<JSON::Object>(response2.get().body);
+ ASSERT_SOME(parse2);
- // Verify that we have some orphan tasks and unregistered
- // frameworks.
- state = parse.get();
- orphanTasks = state.values["orphan_tasks"].as<JSON::Array>();
- EXPECT_EQ(activeTasks, orphanTasks);
+ // Check that there is a single recovered framework, a single active
+ // task, and no orphan tasks.
- unknownFrameworksArray =
- state.values["unregistered_frameworks"].as<JSON::Array>();
- EXPECT_EQ(1u, unknownFrameworksArray.values.size());
+ JSON::Array frameworks2 = parse2->values["frameworks"].as<JSON::Array>();
+ EXPECT_EQ(1u, frameworks2.values.size());
+
+ JSON::Object activeFramework2 = frameworks2.values.front().as<JSON::Object>();
+
+ EXPECT_EQ(
+ frameworkId.value(),
+ activeFramework2.values["id"].as<JSON::String>().value);
+
+ EXPECT_FALSE(activeFramework2.values["active"].as<JSON::Boolean>().value);
+ EXPECT_FALSE(activeFramework2.values["connected"].as<JSON::Boolean>().value);
+ EXPECT_TRUE(activeFramework2.values["recovered"].as<JSON::Boolean>().value);
+
+ JSON::Array activeTasks2 = activeFramework2.values["tasks"].as<JSON::Array>();
+ EXPECT_EQ(activeTasks1, activeTasks2);
- JSON::String unknownFrameworkId =
- unknownFrameworksArray.values.front().as<JSON::String>();
- EXPECT_EQ(activeFrameworkId, unknownFrameworkId);
+ JSON::Array unregisteredFrameworks2 =
+ parse2->values["unregistered_frameworks"].as<JSON::Array>();
+
+ EXPECT_TRUE(unregisteredFrameworks2.values.empty());
+
+ EXPECT_TRUE(parse2->values["orphan_tasks"].as<JSON::Array>().values.empty());
Future<FrameworkRegisteredMessage> frameworkRegisteredMessage =
FUTURE_PROTOBUF(FrameworkRegisteredMessage(), _, _);
@@ -2907,40 +2920,55 @@ TEST_F(MasterTest, OrphanTasks)
AWAIT_READY(frameworkRegisteredMessage);
// Get the master's state.
- response = process::http::get(
+ Future<Response> response3 = process::http::get(
master.get()->pid,
"state",
None(),
createBasicAuthHeaders(DEFAULT_CREDENTIAL));
- AWAIT_EXPECT_RESPONSE_STATUS_EQ(OK().status, response);
- AWAIT_EXPECT_RESPONSE_HEADER_EQ(APPLICATION_JSON, "Content-Type", response);
+ AWAIT_EXPECT_RESPONSE_STATUS_EQ(OK().status, response3);
+ AWAIT_EXPECT_RESPONSE_HEADER_EQ(APPLICATION_JSON, "Content-Type", response3);
- parse = JSON::parse<JSON::Object>(response.get().body);
- ASSERT_SOME(parse);
+ Try<JSON::Object> parse3 = JSON::parse<JSON::Object>(response3.get().body);
+ ASSERT_SOME(parse3);
- // Verify the orphan tasks and unregistered frameworks are removed.
- state = parse.get();
- unknownFrameworksArray =
- state.values["unregistered_frameworks"].as<JSON::Array>();
- EXPECT_EQ(0u, unknownFrameworksArray.values.size());
+ // The framework should no longer be listed as recovered.
+
+ JSON::Array frameworks3 = parse3->values["frameworks"].as<JSON::Array>();
+ EXPECT_EQ(1u, frameworks3.values.size());
+
+ JSON::Object activeFramework3 = frameworks3.values.front().as<JSON::Object>();
+
+ EXPECT_EQ(
+ frameworkId.value(),
+ activeFramework3.values["id"].as<JSON::String>().value);
- orphanTasks = state.values["orphan_tasks"].as<JSON::Array>();
- EXPECT_EQ(0u, orphanTasks.values.size());
+ EXPECT_TRUE(activeFramework3.values["active"].as<JSON::Boolean>().value);
+ EXPECT_TRUE(activeFramework3.values["connected"].as<JSON::Boolean>().value);
+ EXPECT_FALSE(activeFramework3.values["recovered"].as<JSON::Boolean>().value);
+
+ JSON::Array activeTasks3 = activeFramework3.values["tasks"].as<JSON::Array>();
+ EXPECT_EQ(activeTasks1, activeTasks3);
+
+ JSON::Array unregisteredFrameworks3 =
+ parse3->values["unregistered_frameworks"].as<JSON::Array>();
+
+ EXPECT_TRUE(unregisteredFrameworks3.values.empty());
+
+ EXPECT_TRUE(parse3->values["orphan_tasks"].as<JSON::Array>().values.empty());
EXPECT_CALL(exec, shutdown(_))
.Times(AtMost(1));
- // Cleanup.
driver.stop();
driver.join();
}
-// This test verifies that a framework that has not yet re-registered after
-// a master failover doesn't show up multiple times in "unregistered_frameworks"
-// when quering "/state" or "/frameworks" endpoints. This is to catch
-// any regressions for MESOS-4973 and MESOS-6461.
+// This test verifies that a framework that has not yet re-registered
+// after a master failover doesn't show up multiple times in
+// "frameworks" when quering "/state" or "/frameworks" endpoints. This
+// is to catch any regressions for MESOS-4973 and MESOS-6461.
TEST_F(MasterTest, OrphanTasksMultipleAgents)
{
Try<Owned<cluster::Master>> master = StartMaster();
@@ -3047,8 +3075,8 @@ TEST_F(MasterTest, OrphanTasksMultipleAgents)
AWAIT_READY(slaveReregisteredMessage1);
AWAIT_READY(slaveReregisteredMessage2);
- // Ensure that there are 2 orphan tasks and 1 unregistered framework
- // in "/state" endpoint.
+ // Ensure that there are 2 tasks and 1 recovered framework in
+ // "/state" endpoint.
{
Future<Response> response = process::http::get(
master.get()->pid,
@@ -3059,24 +3087,26 @@ TEST_F(MasterTest, OrphanTasksMultipleAgents)
AWAIT_EXPECT_RESPONSE_STATUS_EQ(OK().status, response);
AWAIT_EXPECT_RESPONSE_HEADER_EQ(APPLICATION_JSON, "Content-Type", response);
- Try<JSON::Object> parse = JSON::parse<JSON::Object>(response.get().body);
+ Try<JSON::Object> parse = JSON::parse<JSON::Object>(response->body);
ASSERT_SOME(parse);
- JSON::Object state = parse.get();
-
JSON::Array frameworks =
- state.values["frameworks"].as<JSON::Array>();
+ parse->values["frameworks"].as<JSON::Array>();
JSON::Array orphanTasks =
- state.values["orphan_tasks"].as<JSON::Array>();
+ parse->values["orphan_tasks"].as<JSON::Array>();
JSON::Array unregisteredFrameworks =
- state.values["unregistered_frameworks"].as<JSON::Array>();
+ parse->values["unregistered_frameworks"].as<JSON::Array>();
+
+ ASSERT_EQ(1u, frameworks.values.size());
+ EXPECT_EQ(0u, orphanTasks.values.size());
+ EXPECT_EQ(0u, unregisteredFrameworks.values.size());
- EXPECT_EQ(0u, frameworks.values.size());
- EXPECT_EQ(2u, orphanTasks.values.size());
- EXPECT_EQ(1u, unregisteredFrameworks.values.size());
+ JSON::Object framework = frameworks.values.front().as<JSON::Object>();
+
+ EXPECT_TRUE(framework.values["recovered"].as<JSON::Boolean>().value);
}
- // Ensure that there is 1 unregistered framework in "/frameworks" endpoint.
+ // Ensure that there is 1 recovered framework in "/frameworks" endpoint.
{
Future<Response> response = process::http::get(
master.get()->pid,
@@ -3087,15 +3117,20 @@ TEST_F(MasterTest, OrphanTasksMultipleAgents)
AWAIT_EXPECT_RESPONSE_STATUS_EQ(OK().status, response);
AWAIT_EXPECT_RESPONSE_HEADER_EQ(APPLICATION_JSON, "Content-Type", response);
- Try<JSON::Object> parse = JSON::parse<JSON::Object>(response.get().body);
+ Try<JSON::Object> parse = JSON::parse<JSON::Object>(response->body);
ASSERT_SOME(parse);
- JSON::Object state = parse.get();
-
+ JSON::Array frameworks =
+ parse->values["frameworks"].as<JSON::Array>();
JSON::Array unregisteredFrameworks =
- state.values["unregistered_frameworks"].as<JSON::Array>();
+ parse->values["unregistered_frameworks"].as<JSON::Array>();
- EXPECT_EQ(1u, unregisteredFrameworks.values.size());
+ ASSERT_EQ(1u, frameworks.values.size());
+ EXPECT_EQ(0u, unregisteredFrameworks.values.size());
+
+ JSON::Object framework = frameworks.values.front().as<JSON::Object>();
+
+ EXPECT_TRUE(framework.values["recovered"].as<JSON::Boolean>().value);
}
EXPECT_CALL(exec1, shutdown(_))
@@ -5549,8 +5584,10 @@ TEST_F(MasterTest, DISABLED_RecoverResourcesOrphanedTask)
// Ensure that the resources are successfully recovered.
AWAIT_READY(recoverResources);
- // Ensure that the state of the task is updated to `TASK_FINISHED` on the
- // master.
+ // Ensure that the state of the task is updated to `TASK_FINISHED`
+ // on the master. We don't expect the task to be displayed as a
+ // "completed task", because the terminal status update has not yet
+ // been ack'ed by the scheduler.
{
v1::master::Call call;
call.set_type(v1::master::Call::GET_TASKS);
@@ -5571,8 +5608,10 @@ TEST_F(MasterTest, DISABLED_RecoverResourcesOrphanedTask)
contentType, response->body).get().get_tasks();
ASSERT_TRUE(tasks.IsInitialized());
- ASSERT_EQ(1, tasks.orphan_tasks().size());
- ASSERT_EQ(TASK_FINISHED, tasks.orphan_tasks(0).state());
+ ASSERT_EQ(1, tasks.tasks().size());
+ EXPECT_EQ(TASK_FINISHED, tasks.tasks(0).state());
+ EXPECT_TRUE(tasks.orphan_tasks().empty());
+ EXPECT_TRUE(tasks.completed_tasks().empty());
}
EXPECT_CALL(*executor, shutdown(_))
@@ -5582,6 +5621,189 @@ TEST_F(MasterTest, DISABLED_RecoverResourcesOrphanedTask)
.Times(AtMost(1));
}
+
+// This test checks that the "/state" endpoint displays the correct
+// information when the master fails over and an agent running one of
+// the framework's tasks re-registers before the framework does.
+TEST_F(MasterTest, FailoverAgentReregisterFirst)
+{
+ master::Flags masterFlags = CreateMasterFlags();
+ Try<Owned<cluster::Master>> master = StartMaster(masterFlags);
+ ASSERT_SOME(master);
+
+ StandaloneMasterDetector slaveDetector(master.get()->pid);
+ Try<Owned<cluster::Slave>> slave = StartSlave(&slaveDetector);
+ ASSERT_SOME(slave);
+
+ StandaloneMasterDetector schedDetector(master.get()->pid);
+ MockScheduler sched;
+ TestingMesosSchedulerDriver driver(
+ &sched, &schedDetector, DEFAULT_FRAMEWORK_INFO);
+
+ Future<FrameworkID> frameworkId;
+ EXPECT_CALL(sched, registered(&driver, _, _))
+ .WillOnce(FutureArg<1>(&frameworkId));
+
+ Future<vector<Offer>> offers;
+ EXPECT_CALL(sched, resourceOffers(&driver, _))
+ .WillOnce(FutureArg<1>(&offers))
+ .WillRepeatedly(Return()); // Ignore subsequent offers.
+
+ driver.start();
+
+ AWAIT_READY(frameworkId);
+
+ AWAIT_READY(offers);
+ EXPECT_FALSE(offers->empty());
+
+ TaskInfo task = createTask(offers.get()[0], "sleep 100");
+
+ Future<TaskStatus> runningStatus;
+ EXPECT_CALL(sched, statusUpdate(&driver, _))
+ .WillOnce(FutureArg<1>(&runningStatus));
+
+ Future<Nothing> statusUpdateAck = FUTURE_DISPATCH(
+ slave.get()->pid, &Slave::_statusUpdateAcknowledgement);
+
+ driver.launchTasks(offers.get()[0].id(), {task});
+
+ AWAIT_READY(runningStatus);
+ EXPECT_EQ(TASK_RUNNING, runningStatus.get().state());
+ EXPECT_EQ(task.task_id(), runningStatus.get().task_id());
+
+ AWAIT_READY(statusUpdateAck);
+
+ // Simulate master failover. We leave the scheduler without a master
+ // so it does not attempt to re-register yet.
+ EXPECT_CALL(sched, disconnected(&driver))
+ .Times(1);
+
+ schedDetector.appoint(None());
+ slaveDetector.appoint(None());
+
+ master->reset();
+ master = StartMaster(masterFlags);
+ ASSERT_SOME(master);
+
+ Future<SlaveReregisteredMessage> slaveReregisteredMessage =
+ FUTURE_PROTOBUF(SlaveReregisteredMessage(), _, _);
+
+ slaveDetector.appoint(master.get()->pid);
+
+ AWAIT_READY(slaveReregisteredMessage);
+
+ // Check the master's "/state" endpoint. Because the slave has
+ // re-registered, the master should know about the framework but
+ // view it as disconnected and inactive.
+ {
+ Future<Response> response = process::http::get(
+ master.get()->pid,
+ "state",
+ None(),
+ createBasicAuthHeaders(DEFAULT_CREDENTIAL));
+
+ AWAIT_EXPECT_RESPONSE_STATUS_EQ(OK().status, response);
+ AWAIT_EXPECT_RESPONSE_HEADER_EQ(APPLICATION_JSON, "Content-Type", response);
+
+ Try<JSON::Object> parse = JSON::parse<JSON::Object>(response.get().body);
+ ASSERT_SOME(parse);
+
+ JSON::Array frameworks = parse->values["frameworks"].as<JSON::Array>();
+
+ EXPECT_EQ(1u, frameworks.values.size());
+
+ JSON::Object framework = frameworks.values.front().as<JSON::Object>();
+
+ EXPECT_EQ(
+ frameworkId.get(),
+ framework.values["id"].as<JSON::String>().value);
+
+ EXPECT_FALSE(framework.values["active"].as<JSON::Boolean>().value);
+ EXPECT_FALSE(framework.values["connected"].as<JSON::Boolean>().value);
+ EXPECT_TRUE(framework.values["recovered"].as<JSON::Boolean>().value);
+ EXPECT_EQ(0u, framework.values["registered_time"].as<JSON::Number>());
+ EXPECT_EQ(0u, framework.values["unregistered_time"].as<JSON::Number>());
+ EXPECT_EQ(0u, framework.values.count("reregistered_time"));
+
+ JSON::Array unregisteredFrameworks =
+ parse->values["unregistered_frameworks"].as<JSON::Array>();
+
+ EXPECT_TRUE(unregisteredFrameworks.values.empty());
+
+ JSON::Array completedFrameworks =
+ parse->values["completed_frameworks"].as<JSON::Array>();
+
+ EXPECT_TRUE(completedFrameworks.values.empty());
+ }
+
+ // Cause the scheduler to re-register. We pause the clock to ensure
+ // the re-registration time is predictable. We get a "registered"
+ // callback in the scheduler driver because of MESOS-786.
+ Future<Nothing> registered;
+ EXPECT_CALL(sched, registered(&driver, _, _))
+ .WillOnce(FutureSatisfy(®istered));
+
+ Clock::pause();
+ process::Time reregisterTime = Clock::now();
+
+ schedDetector.appoint(master.get()->pid);
+ AWAIT_READY(registered);
+
+ Clock::resume();
+
+ // Check the master's "/state" endpoint. The framework should now be
+ // listed as connected and active.
+ {
+ Future<Response> response = process::http::get(
+ master.get()->pid,
+ "state",
+ None(),
+ createBasicAuthHeaders(DEFAULT_CREDENTIAL));
+
+ AWAIT_EXPECT_RESPONSE_STATUS_EQ(OK().status, response);
+ AWAIT_EXPECT_RESPONSE_HEADER_EQ(APPLICATION_JSON, "Content-Type", response);
+
+ Try<JSON::Object> parse = JSON::parse<JSON::Object>(response.get().body);
+ ASSERT_SOME(parse);
+
+ JSON::Array frameworks = parse->values["frameworks"].as<JSON::Array>();
+
+ EXPECT_EQ(1u, frameworks.values.size());
+
+ JSON::Object framework = frameworks.values.front().as<JSON::Object>();
+
+ EXPECT_EQ(
+ frameworkId.get(),
+ framework.values["id"].as<JSON::String>().value);
+
+ EXPECT_TRUE(framework.values["active"].as<JSON::Boolean>().value);
+ EXPECT_TRUE(framework.values["connected"].as<JSON::Boolean>().value);
+ EXPECT_FALSE(framework.values["recovered"].as<JSON::Boolean>().value);
+ EXPECT_EQ(0u, framework.values["unregistered_time"].as<JSON::Number>());
+
+ EXPECT_EQ(
+ static_cast<int64_t>(reregisterTime.secs()),
+ framework.values["registered_time"].as<JSON::Number>().as<int64_t>());
+
+ // The state endpoint does not return "reregistered_time" if it is
+ // the same as "registered_time".
+ EXPECT_EQ(0, framework.values.count("reregistered_time"));
+
+ JSON::Array unregisteredFrameworks =
+ parse->values["unregistered_frameworks"].as<JSON::Array>();
+
+ EXPECT_TRUE(unregisteredFrameworks.values.empty());
+
+ JSON::Array completedFrameworks =
+ parse->values["completed_frameworks"].as<JSON::Array>();
+
+ EXPECT_TRUE(completedFrameworks.values.empty());
+ }
+
+ driver.stop();
+ driver.join();
+}
+
} // namespace tests {
} // namespace internal {
} // namespace mesos {
http://git-wip-us.apache.org/repos/asf/mesos/blob/0dbceafa/src/tests/partition_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/partition_tests.cpp b/src/tests/partition_tests.cpp
index 5a0d4bd..e823672 100644
--- a/src/tests/partition_tests.cpp
+++ b/src/tests/partition_tests.cpp
@@ -718,9 +718,10 @@ TEST_F(PartitionTest, PartitionedSlaveReregistrationMasterFailover)
// This test causes a slave to be partitioned while it is running a
-// task for a PARTITION_AWARE scheduler. The scheduler disconnects
+// task for a PARTITION_AWARE framework. The scheduler is shutdown
// before the partition heals. Right now, the task is left running as
-// an orphan; once MESOS-4659 is fixed, the task should be shutdown.
+// an orphan; when MESOS-6602 is fixed, the task will be shutdown when
+// the agent re-registers.
TEST_F(PartitionTest, PartitionedSlaveOrphanedTask)
{
Clock::pause();
@@ -829,15 +830,11 @@ TEST_F(PartitionTest, PartitionedSlaveOrphanedTask)
AWAIT_READY(slaveLost);
// Disconnect the scheduler. The default `failover_timeout` is 0, so
- // the framework's tasks should be shutdown when the slave
- // reregisters, but this is currently not implemented (MESOS-4659).
+ // the framework's tasks should be shutdown when the slave reregisters.
driver.stop();
driver.join();
- // Simulate a master loss event at the slave and then cause the
- // slave to reregister with the master.
- detector.appoint(None());
-
+ // Cause the slave to re-register with the master.
Future<SlaveReregisteredMessage> slaveReregistered = FUTURE_PROTOBUF(
SlaveReregisteredMessage(), master.get()->pid, slave.get()->pid);
@@ -845,6 +842,8 @@ TEST_F(PartitionTest, PartitionedSlaveOrphanedTask)
AWAIT_READY(slaveReregistered);
+ Clock::resume();
+
// Check if `task` is still running by querying master's state endpoint.
Future<Response> response = process::http::get(
master.get()->pid,
@@ -855,33 +854,42 @@ TEST_F(PartitionTest, PartitionedSlaveOrphanedTask)
AWAIT_EXPECT_RESPONSE_STATUS_EQ(OK().status, response);
AWAIT_EXPECT_RESPONSE_HEADER_EQ(APPLICATION_JSON, "Content-Type", response);
- Try<JSON::Object> parse = JSON::parse<JSON::Object>(response.get().body);
+ Try<JSON::Object> parse = JSON::parse<JSON::Object>(response->body);
ASSERT_SOME(parse);
- JSON::Object state = parse.get();
+ JSON::Array unregisteredFrameworks =
+ parse->values["unregistered_frameworks"].as<JSON::Array>();
+
+ // TODO(neilc): When MESOS-6602 is fixed, this should be empty.
+ EXPECT_EQ(1u, unregisteredFrameworks.values.size());
+
+ EXPECT_TRUE(parse->values["frameworks"].as<JSON::Array>().values.empty());
+ EXPECT_TRUE(parse->values["orphan_tasks"].as<JSON::Array>().values.empty());
+
JSON::Array completedFrameworks =
- state.values["completed_frameworks"].as<JSON::Array>();
+ parse->values["completed_frameworks"].as<JSON::Array>();
- EXPECT_EQ(1u, completedFrameworks.values.size());
+ ASSERT_EQ(1u, completedFrameworks.values.size());
JSON::Object jsonFramework =
completedFrameworks.values.front().as<JSON::Object>();
- JSON::String jsonFrameworkId = jsonFramework.values["id"].as<JSON::String>();
-
- EXPECT_EQ(frameworkId.get(), jsonFrameworkId.value);
+ EXPECT_EQ(
+ frameworkId.get(),
+ jsonFramework.values["id"].as<JSON::String>().value);
- // TODO(neilc): Update this when MESOS-4659 is fixed.
- JSON::Array orphanTasks = state.values["orphan_tasks"].as<JSON::Array>();
+ EXPECT_TRUE(jsonFramework.values["tasks"].as<JSON::Array>().values.empty());
- EXPECT_EQ(1u, orphanTasks.values.size());
+ JSON::Array completedTasks =
+ jsonFramework.values["completed_tasks"].as<JSON::Array>();
- JSON::Object jsonTask = orphanTasks.values.front().as<JSON::Object>();
- JSON::String jsonTaskId = jsonTask.values["id"].as<JSON::String>();
+ ASSERT_EQ(1u, completedTasks.values.size());
- EXPECT_EQ(task.task_id(), jsonTaskId.value);
+ JSON::Object completedTask = completedTasks.values.front().as<JSON::Object>();
- Clock::resume();
+ EXPECT_EQ(
+ "TASK_UNREACHABLE",
+ completedTask.values["state"].as<JSON::String>().value);
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/0dbceafa/src/tests/teardown_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/teardown_tests.cpp b/src/tests/teardown_tests.cpp
index 125e180..30c99f9 100644
--- a/src/tests/teardown_tests.cpp
+++ b/src/tests/teardown_tests.cpp
@@ -30,12 +30,17 @@
#include "master/flags.hpp"
#include "master/master.hpp"
+#include "master/detector/standalone.hpp"
+
#include "tests/mesos.hpp"
#include "tests/utils.hpp"
using mesos::internal::master::Master;
+
using mesos::internal::slave::Slave;
+using mesos::master::detector::StandaloneMasterDetector;
+
using process::Future;
using process::Owned;
using process::PID;
@@ -354,6 +359,127 @@ TEST_F(TeardownTest, NoHeader)
driver.join();
}
+
+TEST_F(TeardownTest, DisconnectedFrameworkAfterFailover)
+{
+ master::Flags masterFlags = CreateMasterFlags();
+ Try<Owned<cluster::Master>> master = StartMaster(masterFlags);
+ ASSERT_SOME(master);
+
+ StandaloneMasterDetector slaveDetector(master.get()->pid);
+ Try<Owned<cluster::Slave>> slave = StartSlave(&slaveDetector);
+ ASSERT_SOME(slave);
+
+ StandaloneMasterDetector schedDetector(master.get()->pid);
+ MockScheduler sched;
+ TestingMesosSchedulerDriver driver(&sched, &schedDetector);
+
+ Future<FrameworkID> frameworkId;
+ EXPECT_CALL(sched, registered(&driver, _, _))
+ .WillOnce(FutureArg<1>(&frameworkId));
+
+ Future<vector<Offer>> offers;
+ EXPECT_CALL(sched, resourceOffers(&driver, _))
+ .WillOnce(FutureArg<1>(&offers))
+ .WillRepeatedly(Return()); // Ignore subsequent offers.
+
+ ASSERT_EQ(DRIVER_RUNNING, driver.start());
+
+ AWAIT_READY(frameworkId);
+
+ AWAIT_READY(offers);
+ EXPECT_FALSE(offers->empty());
+
+ TaskInfo task = createTask(offers.get()[0], "sleep 100");
+
+ Future<TaskStatus> runningStatus;
+ EXPECT_CALL(sched, statusUpdate(&driver, _))
+ .WillOnce(FutureArg<1>(&runningStatus));
+
+ Future<Nothing> statusUpdateAck = FUTURE_DISPATCH(
+ slave.get()->pid, &Slave::_statusUpdateAcknowledgement);
+
+ driver.launchTasks(offers.get()[0].id(), {task});
+
+ AWAIT_READY(runningStatus);
+ EXPECT_EQ(TASK_RUNNING, runningStatus.get().state());
+ EXPECT_EQ(task.task_id(), runningStatus.get().task_id());
+
+ AWAIT_READY(statusUpdateAck);
+
+ // Simulate master failover. We leave the scheduler without a master
+ // so it does not attempt to re-register.
+ EXPECT_CALL(sched, disconnected(&driver));
+
+ schedDetector.appoint(None());
+ slaveDetector.appoint(None());
+
+ master->reset();
+ master = StartMaster(masterFlags);
+ ASSERT_SOME(master);
+
+ Future<SlaveReregisteredMessage> slaveReregisteredMessage =
+ FUTURE_PROTOBUF(SlaveReregisteredMessage(), _, _);
+
+ slaveDetector.appoint(master.get()->pid);
+
+ AWAIT_READY(slaveReregisteredMessage);
+
+ // Teardown the framework.
+ {
+ Future<Response> response = process::http::post(
+ master.get()->pid,
+ "teardown",
+ createBasicAuthHeaders(DEFAULT_CREDENTIAL),
+ "frameworkId=" + frameworkId.get().value());
+
+ AWAIT_READY(response);
+ AWAIT_EXPECT_RESPONSE_STATUS_EQ(OK().status, response);
+ }
+
+ // Check the master's "/state" endpoint to confirm that the
+ // "frameworks" and "unregistered_frameworks" keys are empty, and
+ // that the framework that was shutdown appears in the
+ // "completed_frameworks" key.
+ {
+ Future<Response> response = process::http::get(
+ master.get()->pid,
+ "state",
+ None(),
+ createBasicAuthHeaders(DEFAULT_CREDENTIAL));
+
+ AWAIT_EXPECT_RESPONSE_STATUS_EQ(OK().status, response);
+ AWAIT_EXPECT_RESPONSE_HEADER_EQ(APPLICATION_JSON, "Content-Type", response);
+
+ Try<JSON::Object> parse = JSON::parse<JSON::Object>(response.get().body);
+ ASSERT_SOME(parse);
+
+ JSON::Array frameworks = parse->values["frameworks"].as<JSON::Array>();
+
+ EXPECT_TRUE(frameworks.values.empty());
+
+ JSON::Array unregisteredFrameworks =
+ parse->values["unregistered_frameworks"].as<JSON::Array>();
+
+ EXPECT_TRUE(unregisteredFrameworks.values.empty());
+
+ JSON::Array completedFrameworks =
+ parse->values["completed_frameworks"].as<JSON::Array>();
+
+ ASSERT_EQ(1u, completedFrameworks.values.size());
+
+ JSON::Object firstCompletedFramework =
+ completedFrameworks.values.front().as<JSON::Object>();
+
+ EXPECT_EQ(
+ frameworkId.get(),
+ firstCompletedFramework.values["id"].as<JSON::String>().value);
+ }
+
+ driver.stop();
+ driver.join();
+}
+
} // namespace tests {
} // namespace internal {
} // namespace mesos {