You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mesos.apache.org by al...@apache.org on 2017/03/15 21:21:02 UTC
[01/13] mesos git commit: Added comment regarding exit status on
Windows vs. Posix in libprocess.
Repository: mesos
Updated Branches:
refs/heads/master 3ce0aa161 -> 9393c2321
Added comment regarding exit status on Windows vs. Posix in libprocess.
Review: https://reviews.apache.org/r/57598
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/9393c232
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/9393c232
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/9393c232
Branch: refs/heads/master
Commit: 9393c2321eeabcdd31e3df851d2b901a86e39761
Parents: 4f00918
Author: Alexander Rukletsov <al...@apache.org>
Authored: Tue Mar 14 14:29:27 2017 +0100
Committer: Alexander Rukletsov <al...@apache.org>
Committed: Wed Mar 15 22:20:20 2017 +0100
----------------------------------------------------------------------
.../include/process/subprocess_base.hpp | 18 +++++++++++++-----
1 file changed, 13 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/9393c232/3rdparty/libprocess/include/process/subprocess_base.hpp
----------------------------------------------------------------------
diff --git a/3rdparty/libprocess/include/process/subprocess_base.hpp b/3rdparty/libprocess/include/process/subprocess_base.hpp
index 7117a27..a3edae6 100644
--- a/3rdparty/libprocess/include/process/subprocess_base.hpp
+++ b/3rdparty/libprocess/include/process/subprocess_base.hpp
@@ -278,14 +278,22 @@ public:
* Exit status of this subprocess captured as a Future (completed
* when the subprocess exits).
*
- * The exit status is propagated from an underlying call to
- * 'waitpid' and can be used with macros defined in wait.h, i.e.,
- * 'WIFEXITED(status)'.
+ * On Posix, the exit status is propagated from an underlying call
+ * to `waitpid` and can be used with macros defined in wait.h, i.e.,
+ * `WIFEXITED(status)`.
+ *
+ * On Windows, the exit status contains the exit code from an
+ * underlying call to `GetExitCodeProcess()`.
+ *
+ * TODO(alexr): Ensure the code working with `status` is portable by
+ * either making `WIFEXITED` family macros no-op on Windows or
+ * converting `status` to a tuple <termination status, exit code>,
+ * see MESOS-7242.
*
* NOTE: Discarding this future has no effect on the subprocess!
*
- * @return Future from doing a process::reap of this subprocess.
- * Note that process::reap never fails or discards this future.
+ * @return Future from doing a `process::reap()` of this subprocess.
+ * Note that `process::reap()` never fails or discards this future.
*/
Future<Option<int>> status() const { return data->status; }
[03/13] mesos git commit: Reused previous task status for health
updates in command executor.
Posted by al...@apache.org.
Reused previous task status for health updates in command executor.
When a new health task status update is generated in the executor, we
have to make sure specific data is duplicated from the previous one to
avoid shadowing of those data during reconciliation.
Review: https://reviews.apache.org/r/56210
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/a352bbfb
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/a352bbfb
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/a352bbfb
Branch: refs/heads/master
Commit: a352bbfb057725429c9851a4d71f9aef07850816
Parents: 702562c
Author: Alexander Rukletsov <al...@apache.org>
Authored: Fri Jan 27 17:48:55 2017 +0100
Committer: Alexander Rukletsov <al...@apache.org>
Committed: Wed Mar 15 22:20:20 2017 +0100
----------------------------------------------------------------------
src/launcher/executor.cpp | 78 ++++++++++++++++++++++++++++++++----------
1 file changed, 60 insertions(+), 18 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/a352bbfb/src/launcher/executor.cpp
----------------------------------------------------------------------
diff --git a/src/launcher/executor.cpp b/src/launcher/executor.cpp
index 8b13b41..c9243a4 100644
--- a/src/launcher/executor.cpp
+++ b/src/launcher/executor.cpp
@@ -143,7 +143,8 @@ public:
capabilities(_capabilities),
frameworkId(_frameworkId),
executorId(_executorId),
- unacknowledgedTask(None())
+ unacknowledgedTask(None()),
+ lastTaskStatus(None())
{
#ifdef __WINDOWS__
processHandle = INVALID_HANDLE_VALUE;
@@ -291,8 +292,11 @@ protected:
void taskHealthUpdated(const TaskHealthStatus& healthStatus)
{
- // This check prevents us from sending `TASK_RUNNING` updates
- // after the task has been transitioned to `TASK_KILLING`.
+ // This prevents us from sending health updates after a terminal
+ // status update, because we may receive an update from a health
+ // check scheduled before the task has been reaped.
+ //
+ // TODO(alexr): Consider sending health updates after TASK_KILLING.
if (killed || terminated) {
return;
}
@@ -300,7 +304,20 @@ protected:
cout << "Received task health update, healthy: "
<< stringify(healthStatus.healthy()) << endl;
- update(healthStatus.task_id(), TASK_RUNNING, healthStatus.healthy());
+ // Use the previous task status to preserve all attached information.
+ CHECK_SOME(lastTaskStatus);
+ TaskStatus status = protobuf::createTaskStatus(
+ lastTaskStatus.get(),
+ UUID::random(),
+ Clock::now().secs(),
+ None(),
+ None(),
+ None(),
+ None(),
+ None(),
+ healthStatus.healthy());
+
+ forward(status);
if (healthStatus.kill_task()) {
killedByHealthCheck = true;
@@ -343,11 +360,13 @@ protected:
CHECK_EQ(SUBSCRIBED, state);
if (launched) {
- update(
+ TaskStatus status = createTaskStatus(
_task.task_id(),
TASK_FAILED,
None(),
"Attempted to run multiple tasks using a \"command\" executor");
+
+ forward(status);
return;
}
@@ -488,8 +507,10 @@ protected:
process::reap(pid)
.onAny(defer(self(), &Self::reaped, pid, lambda::_1));
- update(unacknowledgedTask->task_id(), TASK_RUNNING);
+ TaskStatus status =
+ createTaskStatus(unacknowledgedTask->task_id(), TASK_RUNNING);
+ forward(status);
launched = true;
}
@@ -607,7 +628,8 @@ private:
if (protobuf::frameworkHasCapability(
frameworkInfo.get(),
FrameworkInfo::Capability::TASK_KILLING_STATE)) {
- update(taskId.get(), TASK_KILLING);
+ TaskStatus status = createTaskStatus(taskId.get(), TASK_KILLING);
+ forward(status);
}
// Stop health checking the task.
@@ -692,11 +714,14 @@ private:
CHECK_SOME(taskId);
- if (killed && killedByHealthCheck) {
- update(taskId.get(), taskState, false, message);
- } else {
- update(taskId.get(), taskState, None(), message);
- }
+ TaskStatus status = createTaskStatus(
+ taskId.get(),
+ taskState,
+ None(),
+ message,
+ (killed && killedByHealthCheck) ? Option<bool>(false) : None());
+
+ forward(status);
Option<string> value = os::getenv("MESOS_HTTP_COMMAND_EXECUTOR");
if (value.isSome() && value.get() == "1") {
@@ -742,11 +767,14 @@ private:
}
}
- void update(
+ // Use this helper to create a status update from scratch, i.e., without
+ // previously attached extra information like `data` or `check_status`.
+ TaskStatus createTaskStatus(
const TaskID& _taskId,
const TaskState& state,
- const Option<bool>& healthy = None(),
- const Option<string>& message = None())
+ const Option<TaskStatus::Reason>& reason = None(),
+ const Option<string>& message = None(),
+ const Option<bool>& healthy = None())
{
UUID uuid = UUID::random();
@@ -759,14 +787,23 @@ private:
status.set_uuid(uuid.toBytes());
status.set_timestamp(Clock::now().secs());
- if (healthy.isSome()) {
- status.set_healthy(healthy.get());
+ if (reason.isSome()) {
+ status.set_reason(reason.get());
}
if (message.isSome()) {
status.set_message(message.get());
}
+ if (healthy.isSome()) {
+ status.set_healthy(healthy.get());
+ }
+
+ return status;
+ }
+
+ void forward(const TaskStatus& status)
+ {
Call call;
call.set_type(Call::UPDATE);
@@ -776,7 +813,10 @@ private:
call.mutable_update()->mutable_status()->CopyFrom(status);
// Capture the status update.
- unacknowledgedUpdates[uuid] = call.update();
+ unacknowledgedUpdates[UUID::fromBytes(status.uuid()).get()] = call.update();
+
+ // Overwrite the last task status.
+ lastTaskStatus = status;
mesos->send(evolve(call));
}
@@ -841,6 +881,8 @@ private:
// update acknowledgements have been received yet.
Option<TaskInfo> unacknowledgedTask;
+ Option<TaskStatus> lastTaskStatus;
+
Owned<checks::HealthChecker> checker;
};
[04/13] mesos git commit: Added a note about task status updates in
scheduler and internal API.
Posted by al...@apache.org.
Added a note about task status updates in scheduler and internal API.
Review: https://reviews.apache.org/r/56016
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/517607aa
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/517607aa
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/517607aa
Branch: refs/heads/master
Commit: 517607aaea83f8a009f9aa5f5120417764e893c7
Parents: 3ce0aa1
Author: Alexander Rukletsov <al...@apache.org>
Authored: Tue Feb 28 11:39:00 2017 +0100
Committer: Alexander Rukletsov <al...@apache.org>
Committed: Wed Mar 15 22:20:20 2017 +0100
----------------------------------------------------------------------
include/mesos/scheduler/scheduler.proto | 5 +++++
include/mesos/v1/scheduler/scheduler.proto | 5 +++++
src/messages/messages.proto | 5 +++++
3 files changed, 15 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/517607aa/include/mesos/scheduler/scheduler.proto
----------------------------------------------------------------------
diff --git a/include/mesos/scheduler/scheduler.proto b/include/mesos/scheduler/scheduler.proto
index d04924a..0f6803a 100644
--- a/include/mesos/scheduler/scheduler.proto
+++ b/include/mesos/scheduler/scheduler.proto
@@ -118,6 +118,11 @@ message Event {
// to the task. It is also the responsibility of the scheduler to
// explicitly acknowledge the receipt of a status update. See
// 'Acknowledge' in the 'Call' section below for the semantics.
+ //
+ // A task status update may be used for guaranteed delivery of some
+ // task-related information, e.g., task's health update. Such
+ // information may be shadowed by subsequent task status updates, that
+ // do not preserve fields of the previously sent message.
message Update {
required TaskStatus status = 1;
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/517607aa/include/mesos/v1/scheduler/scheduler.proto
----------------------------------------------------------------------
diff --git a/include/mesos/v1/scheduler/scheduler.proto b/include/mesos/v1/scheduler/scheduler.proto
index 6e8246d..2155864 100644
--- a/include/mesos/v1/scheduler/scheduler.proto
+++ b/include/mesos/v1/scheduler/scheduler.proto
@@ -118,6 +118,11 @@ message Event {
// to the task. It is also the responsibility of the scheduler to
// explicitly acknowledge the receipt of a status update. See
// 'Acknowledge' in the 'Call' section below for the semantics.
+ //
+ // A task status update may be used for guaranteed delivery of some
+ // task-related information, e.g., task's health update. Such
+ // information may be shadowed by subsequent task status updates, that
+ // do not preserve fields of the previously sent message.
message Update {
required TaskStatus status = 1;
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/517607aa/src/messages/messages.proto
----------------------------------------------------------------------
diff --git a/src/messages/messages.proto b/src/messages/messages.proto
index 508ff59..6cc06eb 100644
--- a/src/messages/messages.proto
+++ b/src/messages/messages.proto
@@ -34,6 +34,11 @@ package mesos.internal;
* passing it from the agent to the framework that spawned the task.
*
* See protobuf::createStatusUpdate.
+ *
+ * A task status update may be used for guaranteed delivery of some
+ * task-related information, e.g., task's health update. Such information
+ * may be shadowed by subsequent task status updates, that do not preserve
+ * fields of the previously sent message.
*/
message StatusUpdate {
required FrameworkID framework_id = 1;
[02/13] mesos git commit: Added support for general checks to command
executor.
Posted by al...@apache.org.
Added support for general checks to command executor.
Review: https://reviews.apache.org/r/56212
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/11160d67
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/11160d67
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/11160d67
Branch: refs/heads/master
Commit: 11160d6716d03865d2a74bad341d7ed33849385f
Parents: e60b57b
Author: Alexander Rukletsov <al...@apache.org>
Authored: Tue Feb 28 16:14:14 2017 +0100
Committer: Alexander Rukletsov <al...@apache.org>
Committed: Wed Mar 15 22:20:20 2017 +0100
----------------------------------------------------------------------
src/launcher/executor.cpp | 129 ++++++++++++++++++++++++++++++++++++-----
1 file changed, 116 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/11160d67/src/launcher/executor.cpp
----------------------------------------------------------------------
diff --git a/src/launcher/executor.cpp b/src/launcher/executor.cpp
index a32ec5b..bd3c0cf 100644
--- a/src/launcher/executor.cpp
+++ b/src/launcher/executor.cpp
@@ -62,6 +62,7 @@
#include <stout/os/kill.hpp>
#include <stout/os/killtree.hpp>
+#include "checks/checker.hpp"
#include "checks/health_checker.hpp"
#include "common/http.hpp"
@@ -292,6 +293,41 @@ protected:
}
}
+ void taskCheckUpdated(
+ const TaskID& _taskId,
+ const CheckStatusInfo& checkStatus)
+ {
+ CHECK_SOME(taskId);
+ CHECK_EQ(taskId.get(), _taskId);
+
+ // This prevents us from sending check updates after a terminal
+ // status update, because we may receive an update from a check
+ // scheduled before the task has been reaped.
+ //
+ // TODO(alexr): Consider sending check updates after TASK_KILLING.
+ if (killed || terminated) {
+ return;
+ }
+
+ cout << "Received check update" << endl;
+
+ // Use the previous task status to preserve all attached information.
+ CHECK_SOME(lastTaskStatus);
+ TaskStatus status = protobuf::createTaskStatus(
+ lastTaskStatus.get(),
+ UUID::random(),
+ Clock::now().secs(),
+ None(),
+ None(),
+ None(),
+ TaskStatus::REASON_TASK_CHECK_STATUS_UPDATED,
+ None(),
+ None(),
+ checkStatus);
+
+ forward(status);
+ }
+
void taskHealthUpdated(const TaskHealthStatus& healthStatus)
{
CHECK_SOME(taskId);
@@ -473,6 +509,35 @@ protected:
cout << "Forked command at " << pid << endl;
+ if (task.has_check()) {
+ vector<string> namespaces;
+ if (rootfs.isSome() &&
+ task.check().type() == CheckInfo::COMMAND) {
+ // Make sure command checks are run from the task's mount namespace.
+ // Otherwise if rootfs is specified the command binary may not be
+ // available in the executor.
+ //
+ // NOTE: The command executor shares the network namespace
+ // with its task, hence no need to enter it explicitly.
+ namespaces.push_back("mnt");
+ }
+
+ Try<Owned<checks::Checker>> _checker =
+ checks::Checker::create(
+ task.check(),
+ defer(self(), &Self::taskCheckUpdated, taskId.get(), lambda::_1),
+ taskId.get(),
+ pid,
+ namespaces);
+
+ if (_checker.isError()) {
+ // TODO(alexr): Consider ABORT and return a TASK_FAILED here.
+ cerr << "Failed to create checker: " << _checker.error() << endl;
+ } else {
+ checker = _checker.get();
+ }
+ }
+
if (task.has_health_check()) {
vector<string> namespaces;
if (rootfs.isSome() &&
@@ -632,6 +697,11 @@ private:
forward(status);
}
+ // Stop checking the task.
+ if (checker.get() != nullptr) {
+ checker->stop();
+ }
+
// Stop health checking the task.
if (healthChecker.get() != nullptr) {
healthChecker->stop();
@@ -672,6 +742,11 @@ private:
{
terminated = true;
+ // Stop checking the task.
+ if (checker.get() != nullptr) {
+ checker->stop();
+ }
+
// Stop health checking the task.
if (healthChecker.get() != nullptr) {
healthChecker->stop();
@@ -718,8 +793,12 @@ private:
taskId.get(),
taskState,
None(),
- message,
- (killed && killedByHealthCheck) ? Option<bool>(false) : None());
+ message);
+
+ // Indicate that a kill occured due to a failing health check.
+ if (killed && killedByHealthCheck) {
+ status.set_healthy(false);
+ }
forward(status);
@@ -773,19 +852,16 @@ private:
const TaskID& _taskId,
const TaskState& state,
const Option<TaskStatus::Reason>& reason = None(),
- const Option<string>& message = None(),
- const Option<bool>& healthy = None())
+ const Option<string>& message = None())
{
- UUID uuid = UUID::random();
+ TaskStatus status = protobuf::createTaskStatus(
+ _taskId,
+ state,
+ UUID::random(),
+ Clock::now().secs());
- TaskStatus status;
- status.mutable_task_id()->CopyFrom(_taskId);
status.mutable_executor_id()->CopyFrom(executorId);
-
- status.set_state(state);
status.set_source(TaskStatus::SOURCE_EXECUTOR);
- status.set_uuid(uuid.toBytes());
- status.set_timestamp(Clock::now().secs());
if (reason.isSome()) {
status.set_reason(reason.get());
@@ -795,8 +871,34 @@ private:
status.set_message(message.get());
}
- if (healthy.isSome()) {
- status.set_healthy(healthy.get());
+ // TODO(alexr): Augment health information in a way similar to
+ // `CheckStatusInfo`. See MESOS-6417 for more details.
+
+ // If a check for the task has been defined, `check_status` field in each
+ // task status must be set to a valid `CheckStatusInfo` message even if
+ // there is no check status available yet.
+ CHECK(taskData.isSome());
+ if (taskData->taskInfo.has_check()) {
+ CheckStatusInfo checkStatusInfo;
+ checkStatusInfo.set_type(taskData->taskInfo.check().type());
+ switch (taskData->taskInfo.check().type()) {
+ case CheckInfo::COMMAND: {
+ checkStatusInfo.mutable_command();
+ break;
+ }
+
+ case CheckInfo::HTTP: {
+ checkStatusInfo.mutable_http();
+ break;
+ }
+
+ case CheckInfo::UNKNOWN: {
+ CHECK_NE(CheckInfo::UNKNOWN, taskData->taskInfo.check().type());
+ break;
+ }
+ }
+
+ status.mutable_check_status()->CopyFrom(checkStatusInfo);
}
return status;
@@ -894,6 +996,7 @@ private:
Option<TaskStatus> lastTaskStatus;
+ Owned<checks::Checker> checker;
Owned<checks::HealthChecker> healthChecker;
};
[12/13] mesos git commit: Added comments regarding exit status on
Windows vs. Posix.
Posted by al...@apache.org.
Added comments regarding exit status on Windows vs. Posix.
Review: https://reviews.apache.org/r/57597
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/4f009180
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/4f009180
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/4f009180
Branch: refs/heads/master
Commit: 4f009180730babe782d15bd0bbc289c91e5a5b2e
Parents: 11160d6
Author: Alexander Rukletsov <al...@apache.org>
Authored: Tue Mar 14 14:29:05 2017 +0100
Committer: Alexander Rukletsov <al...@apache.org>
Committed: Wed Mar 15 22:20:20 2017 +0100
----------------------------------------------------------------------
include/mesos/mesos.proto | 4 +++-
include/mesos/scheduler/scheduler.proto | 10 ++++++++++
include/mesos/v1/mesos.proto | 4 +++-
include/mesos/v1/scheduler/scheduler.proto | 10 ++++++++++
src/checks/checker.cpp | 6 ++++++
5 files changed, 32 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/4f009180/include/mesos/mesos.proto
----------------------------------------------------------------------
diff --git a/include/mesos/mesos.proto b/include/mesos/mesos.proto
index a432ead..9a66967 100644
--- a/include/mesos/mesos.proto
+++ b/include/mesos/mesos.proto
@@ -1736,7 +1736,9 @@ enum TaskState {
*/
message CheckStatusInfo {
message Command {
- // Exit code of a command check.
+ // Exit code of a command check. It is the result of calling
+ // `WEXITSTATUS()` on `waitpid()` termination information on
+ // Posix and calling `GetExitCodeProcess()` on Windows.
optional int32 exit_code = 1;
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/4f009180/include/mesos/scheduler/scheduler.proto
----------------------------------------------------------------------
diff --git a/include/mesos/scheduler/scheduler.proto b/include/mesos/scheduler/scheduler.proto
index 0f6803a..cee0a17 100644
--- a/include/mesos/scheduler/scheduler.proto
+++ b/include/mesos/scheduler/scheduler.proto
@@ -155,6 +155,16 @@ message Event {
// 'executor_id' will be set and possibly 'status' (if we were
// able to determine the exit status).
optional ExecutorID executor_id = 2;
+
+ // On Posix, `status` corresponds to termination information in the
+ // `stat_loc` area returned from a `waitpid` call. On Windows, `status`
+ // is obtained via calling the `GetExitCodeProcess()` function. For
+ // messages coming from Posix agents, schedulers need to apply
+ // `WEXITSTATUS` family macros or equivalent transformations to obtain
+ // exit codes.
+ //
+ // TODO(alexr): Consider unifying Windows and Posix behavior by returning
+ // exit code here, see MESOS-7241.
optional int32 status = 3;
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/4f009180/include/mesos/v1/mesos.proto
----------------------------------------------------------------------
diff --git a/include/mesos/v1/mesos.proto b/include/mesos/v1/mesos.proto
index 8a733d9..115f1b4 100644
--- a/include/mesos/v1/mesos.proto
+++ b/include/mesos/v1/mesos.proto
@@ -1735,7 +1735,9 @@ enum TaskState {
*/
message CheckStatusInfo {
message Command {
- // Exit code of a command check.
+ // Exit code of a command check. It is the result of calling
+ // `WEXITSTATUS()` on `waitpid()` termination information on
+ // Posix and calling `GetExitCodeProcess()` on Windows.
optional int32 exit_code = 1;
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/4f009180/include/mesos/v1/scheduler/scheduler.proto
----------------------------------------------------------------------
diff --git a/include/mesos/v1/scheduler/scheduler.proto b/include/mesos/v1/scheduler/scheduler.proto
index 2155864..00a54ec 100644
--- a/include/mesos/v1/scheduler/scheduler.proto
+++ b/include/mesos/v1/scheduler/scheduler.proto
@@ -155,6 +155,16 @@ message Event {
// 'executor_id' will be set and possibly 'status' (if we were
// able to determine the exit status).
optional ExecutorID executor_id = 2;
+
+ // On Posix, `status` corresponds to termination information in the
+ // `stat_loc` area returned from a `waitpid` call. On Windows, `status`
+ // is obtained via calling the `GetExitCodeProcess()` function. For
+ // messages coming from Posix agents, schedulers need to apply
+ // `WEXITSTATUS` family macros or equivalent transformations to obtain
+ // exit codes.
+ //
+ // TODO(alexr): Consider unifying Windows and Posix behavior by returning
+ // exit code here, see MESOS-7241.
optional int32 status = 3;
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/4f009180/src/checks/checker.cpp
----------------------------------------------------------------------
diff --git a/src/checks/checker.cpp b/src/checks/checker.cpp
index 94315d7..5aae957 100644
--- a/src/checks/checker.cpp
+++ b/src/checks/checker.cpp
@@ -433,6 +433,12 @@ void CheckerProcess::processCommandCheckResult(
CheckStatusInfo checkStatusInfo;
checkStatusInfo.set_type(check.type());
+ // On Posix, `result` corresponds to termination information in the
+ // `stat_loc` area. On Windows, `status` is obtained via calling the
+ // `GetExitCodeProcess()` function.
+ //
+ // TODO(alexr): Ensure `WEXITSTATUS` family macros are no-op on Windows,
+ // see MESOS-7242.
if (result.isReady() && WIFEXITED(result.get())) {
const int exitCode = WEXITSTATUS(result.get());
VLOG(1) << check.type() << " check for task "
[05/13] mesos git commit: Overloaded `<<` for `CheckInfo::Type`.
Posted by al...@apache.org.
Overloaded `<<` for `CheckInfo::Type`.
Review: https://reviews.apache.org/r/57394
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/cc6e063a
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/cc6e063a
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/cc6e063a
Branch: refs/heads/master
Commit: cc6e063a6907a282dac875133e9f328d5a6fd364
Parents: 517607a
Author: Alexander Rukletsov <al...@apache.org>
Authored: Tue Mar 7 17:41:45 2017 +0100
Committer: Alexander Rukletsov <al...@apache.org>
Committed: Wed Mar 15 22:20:20 2017 +0100
----------------------------------------------------------------------
include/mesos/type_utils.hpp | 3 +++
include/mesos/v1/mesos.hpp | 3 +++
src/common/type_utils.cpp | 6 ++++++
src/v1/mesos.cpp | 6 ++++++
4 files changed, 18 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/cc6e063a/include/mesos/type_utils.hpp
----------------------------------------------------------------------
diff --git a/include/mesos/type_utils.hpp b/include/mesos/type_utils.hpp
index c7f86ac..f22e9ea 100644
--- a/include/mesos/type_utils.hpp
+++ b/include/mesos/type_utils.hpp
@@ -304,6 +304,9 @@ std::ostream& operator<<(
const std::vector<TaskID>& taskIds);
+std::ostream& operator<<(std::ostream& stream, const CheckInfo::Type& type);
+
+
std::ostream& operator<<(
std::ostream& stream,
const FrameworkInfo::Capability& capability);
http://git-wip-us.apache.org/repos/asf/mesos/blob/cc6e063a/include/mesos/v1/mesos.hpp
----------------------------------------------------------------------
diff --git a/include/mesos/v1/mesos.hpp b/include/mesos/v1/mesos.hpp
index 0700742..f5e0879 100644
--- a/include/mesos/v1/mesos.hpp
+++ b/include/mesos/v1/mesos.hpp
@@ -306,6 +306,9 @@ std::ostream& operator<<(
const std::vector<TaskID>& taskIds);
+std::ostream& operator<<(std::ostream& stream, const CheckInfo::Type& type);
+
+
std::ostream& operator<<(
std::ostream& stream,
const FrameworkInfo::Capability& capability);
http://git-wip-us.apache.org/repos/asf/mesos/blob/cc6e063a/src/common/type_utils.cpp
----------------------------------------------------------------------
diff --git a/src/common/type_utils.cpp b/src/common/type_utils.cpp
index d86d56d..ef26335 100644
--- a/src/common/type_utils.cpp
+++ b/src/common/type_utils.cpp
@@ -595,6 +595,12 @@ ostream& operator<<(ostream& stream, const vector<TaskID>& taskIds)
}
+ostream& operator<<(ostream& stream, const CheckInfo::Type& type)
+{
+ return stream << CheckInfo::Type_Name(type);
+}
+
+
ostream& operator<<(
ostream& stream,
const FrameworkInfo::Capability& capability)
http://git-wip-us.apache.org/repos/asf/mesos/blob/cc6e063a/src/v1/mesos.cpp
----------------------------------------------------------------------
diff --git a/src/v1/mesos.cpp b/src/v1/mesos.cpp
index 5b89603..3d9281d 100644
--- a/src/v1/mesos.cpp
+++ b/src/v1/mesos.cpp
@@ -509,6 +509,12 @@ ostream& operator<<(ostream& stream, const vector<TaskID>& taskIds)
}
+ostream& operator<<(ostream& stream, const CheckInfo::Type& type)
+{
+ return stream << CheckInfo::Type_Name(type);
+}
+
+
ostream& operator<<(
ostream& stream,
const FrameworkInfo::Capability& capability)
[09/13] mesos git commit: Renamed health checker in command executor
for clarity.
Posted by al...@apache.org.
Renamed health checker in command executor for clarity.
Review: https://reviews.apache.org/r/56211
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/6b332ade
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/6b332ade
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/6b332ade
Branch: refs/heads/master
Commit: 6b332ade085de92ba0ae29856e2b1990e4ea14b8
Parents: a352bbf
Author: Alexander Rukletsov <al...@apache.org>
Authored: Fri Jan 27 18:23:37 2017 +0100
Committer: Alexander Rukletsov <al...@apache.org>
Committed: Wed Mar 15 22:20:20 2017 +0100
----------------------------------------------------------------------
src/launcher/executor.cpp | 21 ++++++++++++---------
1 file changed, 12 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/6b332ade/src/launcher/executor.cpp
----------------------------------------------------------------------
diff --git a/src/launcher/executor.cpp b/src/launcher/executor.cpp
index c9243a4..57e4a32 100644
--- a/src/launcher/executor.cpp
+++ b/src/launcher/executor.cpp
@@ -292,6 +292,9 @@ protected:
void taskHealthUpdated(const TaskHealthStatus& healthStatus)
{
+ CHECK_SOME(taskId);
+ CHECK_EQ(taskId.get(), healthStatus.task_id());
+
// This prevents us from sending health updates after a terminal
// status update, because we may receive an update from a health
// check scheduled before the task has been reaped.
@@ -485,7 +488,7 @@ protected:
namespaces.push_back("mnt");
}
- Try<Owned<checks::HealthChecker>> _checker =
+ Try<Owned<checks::HealthChecker>> _healthChecker =
checks::HealthChecker::create(
unacknowledgedTask->health_check(),
launcherDir,
@@ -494,12 +497,12 @@ protected:
pid,
namespaces);
- if (_checker.isError()) {
+ if (_healthChecker.isError()) {
// TODO(gilbert): Consider ABORT and return a TASK_FAILED here.
cerr << "Failed to create health checker: "
- << _checker.error() << endl;
+ << _healthChecker.error() << endl;
} else {
- checker = _checker.get();
+ healthChecker = _healthChecker.get();
}
}
@@ -633,8 +636,8 @@ private:
}
// Stop health checking the task.
- if (checker.get() != nullptr) {
- checker->stop();
+ if (healthChecker.get() != nullptr) {
+ healthChecker->stop();
}
// Now perform signal escalation to begin killing the task.
@@ -673,8 +676,8 @@ private:
terminated = true;
// Stop health checking the task.
- if (checker.get() != nullptr) {
- checker->stop();
+ if (healthChecker.get() != nullptr) {
+ healthChecker->stop();
}
TaskState taskState;
@@ -883,7 +886,7 @@ private:
Option<TaskStatus> lastTaskStatus;
- Owned<checks::HealthChecker> checker;
+ Owned<checks::HealthChecker> healthChecker;
};
} // namespace internal {
[07/13] mesos git commit: Added a helper for building a task status
from scratch.
Posted by al...@apache.org.
Added a helper for building a task status from scratch.
Review: https://reviews.apache.org/r/57594
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/bf196f02
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/bf196f02
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/bf196f02
Branch: refs/heads/master
Commit: bf196f02d65199e69407af834c1a1d0fb0f35266
Parents: 85fbf8f
Author: Alexander Rukletsov <al...@apache.org>
Authored: Tue Mar 14 12:06:07 2017 +0100
Committer: Alexander Rukletsov <al...@apache.org>
Committed: Wed Mar 15 22:20:20 2017 +0100
----------------------------------------------------------------------
src/common/protobuf_utils.cpp | 19 +++++++++++++++++++
src/common/protobuf_utils.hpp | 9 +++++++++
src/docker/executor.cpp | 12 ++++++++++++
src/launcher/default_executor.cpp | 13 +++++++------
4 files changed, 47 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/bf196f02/src/common/protobuf_utils.cpp
----------------------------------------------------------------------
diff --git a/src/common/protobuf_utils.cpp b/src/common/protobuf_utils.cpp
index d568d41..138e288 100644
--- a/src/common/protobuf_utils.cpp
+++ b/src/common/protobuf_utils.cpp
@@ -122,6 +122,8 @@ StatusUpdate createStatusUpdate(
update.mutable_executor_id()->MergeFrom(executorId.get());
}
+ // TODO(alexr): Use `createTaskStatus()` instead
+ // once `UUID` is required in this function.
TaskStatus* status = update.mutable_status();
status->mutable_task_id()->MergeFrom(taskId);
@@ -207,6 +209,23 @@ StatusUpdate createStatusUpdate(
TaskStatus createTaskStatus(
+ const TaskID& taskId,
+ const TaskState& state,
+ const UUID& uuid,
+ double timestamp)
+{
+ TaskStatus status;
+
+ status.set_uuid(uuid.toBytes());
+ status.set_timestamp(timestamp);
+ status.mutable_task_id()->CopyFrom(taskId);
+ status.set_state(state);
+
+ return status;
+}
+
+
+TaskStatus createTaskStatus(
TaskStatus status,
const UUID& uuid,
double timestamp,
http://git-wip-us.apache.org/repos/asf/mesos/blob/bf196f02/src/common/protobuf_utils.hpp
----------------------------------------------------------------------
diff --git a/src/common/protobuf_utils.hpp b/src/common/protobuf_utils.hpp
index 3b157a0..ddfac24 100644
--- a/src/common/protobuf_utils.hpp
+++ b/src/common/protobuf_utils.hpp
@@ -93,6 +93,15 @@ StatusUpdate createStatusUpdate(
const Option<SlaveID>& slaveId);
+// Helper function that creates a new task status from scratch with
+// obligatory fields set.
+TaskStatus createTaskStatus(
+ const TaskID& taskId,
+ const TaskState& state,
+ const UUID& uuid,
+ double timestamp);
+
+
// Helper function that creates a new task status from the given task
// status. Specific fields in `status` can be overridden in the new
// status by specifying the appropriate argument. Fields `task_id`,
http://git-wip-us.apache.org/repos/asf/mesos/blob/bf196f02/src/docker/executor.cpp
----------------------------------------------------------------------
diff --git a/src/docker/executor.cpp b/src/docker/executor.cpp
index 6c11320..b652a99 100644
--- a/src/docker/executor.cpp
+++ b/src/docker/executor.cpp
@@ -131,6 +131,8 @@ public:
void launchTask(ExecutorDriver* driver, const TaskInfo& task)
{
if (run.isSome()) {
+ // TODO(alexr): Use `protobuf::createTaskStatus()`
+ // instead of manually setting fields.
TaskStatus status;
status.mutable_task_id()->CopyFrom(task.task_id());
status.set_state(TASK_FAILED);
@@ -169,6 +171,8 @@ public:
);
if (runOptions.isError()) {
+ // TODO(alexr): Use `protobuf::createTaskStatus()`
+ // instead of manually setting fields.
TaskStatus status;
status.mutable_task_id()->CopyFrom(task.task_id());
status.set_state(TASK_FAILED);
@@ -204,6 +208,8 @@ public:
if (!killed) {
containerPid = container.pid;
+ // TODO(alexr): Use `protobuf::createTaskStatus()`
+ // instead of manually setting fields.
TaskStatus status;
status.mutable_task_id()->CopyFrom(taskId.get());
status.set_state(TASK_RUNNING);
@@ -313,6 +319,8 @@ protected:
cout << "Received task health update, healthy: "
<< stringify(healthStatus.healthy()) << endl;
+ // TODO(alexr): Use `protobuf::createTaskStatus()`
+ // instead of manually setting fields.
TaskStatus status;
status.mutable_task_id()->CopyFrom(healthStatus.task_id());
status.set_healthy(healthStatus.healthy());
@@ -379,6 +387,8 @@ private:
if (protobuf::frameworkHasCapability(
frameworkInfo.get(),
FrameworkInfo::Capability::TASK_KILLING_STATE)) {
+ // TODO(alexr): Use `protobuf::createTaskStatus()`
+ // instead of manually setting fields.
TaskStatus status;
status.mutable_task_id()->CopyFrom(taskId.get());
status.set_state(TASK_KILLING);
@@ -463,6 +473,8 @@ private:
CHECK_SOME(taskId);
+ // TODO(alexr): Use `protobuf::createTaskStatus()`
+ // instead of manually setting fields.
TaskStatus taskStatus;
taskStatus.mutable_task_id()->CopyFrom(taskId.get());
taskStatus.set_state(state);
http://git-wip-us.apache.org/repos/asf/mesos/blob/bf196f02/src/launcher/default_executor.cpp
----------------------------------------------------------------------
diff --git a/src/launcher/default_executor.cpp b/src/launcher/default_executor.cpp
index 0ed436f..cbd4f7e 100644
--- a/src/launcher/default_executor.cpp
+++ b/src/launcher/default_executor.cpp
@@ -46,6 +46,7 @@
#include "checks/health_checker.hpp"
#include "common/http.hpp"
+#include "common/protobuf_utils.hpp"
#include "common/status_utils.hpp"
#include "internal/devolve.hpp"
@@ -890,14 +891,14 @@ private:
{
UUID uuid = UUID::random();
- TaskStatus status;
- status.mutable_task_id()->CopyFrom(taskId);
- status.mutable_executor_id()->CopyFrom(executorId);
+ TaskStatus status = protobuf::createTaskStatus(
+ taskId,
+ state,
+ uuid,
+ Clock::now().secs());
- status.set_state(state);
+ status.mutable_executor_id()->CopyFrom(executorId);
status.set_source(TaskStatus::SOURCE_EXECUTOR);
- status.set_uuid(uuid.toBytes());
- status.set_timestamp(Clock::now().secs());
if (message.isSome()) {
status.set_message(message.get());
[13/13] mesos git commit: Simplified task id procurement in command
executor.
Posted by al...@apache.org.
Simplified task id procurement in command executor.
Review: https://reviews.apache.org/r/57150
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/e60b57b4
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/e60b57b4
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/e60b57b4
Branch: refs/heads/master
Commit: e60b57b438289323f8ab4c91b83c113b066f320b
Parents: a67231b
Author: Alexander Rukletsov <al...@apache.org>
Authored: Tue Feb 28 16:24:42 2017 +0100
Committer: Alexander Rukletsov <al...@apache.org>
Committed: Wed Mar 15 22:20:20 2017 +0100
----------------------------------------------------------------------
src/launcher/executor.cpp | 20 +++++++-------------
1 file changed, 7 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/e60b57b4/src/launcher/executor.cpp
----------------------------------------------------------------------
diff --git a/src/launcher/executor.cpp b/src/launcher/executor.cpp
index 02834c9..a32ec5b 100644
--- a/src/launcher/executor.cpp
+++ b/src/launcher/executor.cpp
@@ -375,11 +375,9 @@ protected:
return;
}
- // Capture the task.
+ // Capture `TaskInfo` and `TaskID` of the task.
CHECK(taskData.isNone());
taskData = TaskData(task);
-
- // Capture the TaskID.
taskId = task.task_id();
// Capture the kill policy.
@@ -406,8 +404,7 @@ protected:
} else if (task.has_command()) {
command = task.command();
} else {
- LOG(FATAL) << "Expecting task '" << taskData.taskInfo->task_id()
- << "' to have a command";
+ LOG(FATAL) << "Expecting task '" << taskId.get() << "' to have a command";
}
// TODO(jieyu): For now, we just fail the executor if the task's
@@ -417,12 +414,10 @@ protected:
// correct solution is to perform this validation at master side.
if (command.shell()) {
CHECK(command.has_value())
- << "Shell command of task '" << taskData.taskInfo->task_id()
- << "' is not specified!";
+ << "Shell command of task '" << taskId.get() << "' is not specified";
} else {
CHECK(command.has_value())
- << "Executable of task '" << taskData.taskInfo->task_id()
- << "' is not specified!";
+ << "Executable of task '" << taskId.get() << "' is not specified";
}
// Determine the environment for the command to be launched.
@@ -450,7 +445,7 @@ protected:
launchEnvironment.MergeFrom(command.environment());
}
- cout << "Starting task " << taskData.taskInfo->task_id() << endl;
+ cout << "Starting task " << taskId.get() << endl;
#ifndef __WINDOWS__
pid = launchTaskPosix(
@@ -496,7 +491,7 @@ protected:
task.health_check(),
launcherDir,
defer(self(), &Self::taskHealthUpdated, lambda::_1),
- taskData.taskInfo->task_id(),
+ taskId.get(),
pid,
namespaces);
@@ -513,8 +508,7 @@ protected:
process::reap(pid)
.onAny(defer(self(), &Self::reaped, pid, lambda::_1));
- TaskStatus status =
- createTaskStatus(taskData.taskInfo->task_id(), TASK_RUNNING);
+ TaskStatus status = createTaskStatus(taskId.get(), TASK_RUNNING);
forward(status);
launched = true;
[08/13] mesos git commit: Updated checks library with general check
support.
Posted by al...@apache.org.
Updated checks library with general check support.
Add support for general checks, i.e. defined by CheckInfo, in
checking library. A general check can be either an command or
an HTTP request. The library performs the requested check at
the specified interval and sends the result to the framework
via a task status update. If the current result is the same as
the previous result, no status update is sent.
Review: https://reviews.apache.org/r/56208
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/f042bf60
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/f042bf60
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/f042bf60
Branch: refs/heads/master
Commit: f042bf607f9bdcf4d350df892b6985760a7ea7e8
Parents: bf196f0
Author: Alexander Rukletsov <al...@apache.org>
Authored: Sun Feb 5 21:53:51 2017 +0100
Committer: Alexander Rukletsov <al...@apache.org>
Committed: Wed Mar 15 22:20:20 2017 +0100
----------------------------------------------------------------------
src/checks/checker.cpp | 578 ++++++++++++++++++++++++++++++++++++-
src/checks/checker.hpp | 53 ++++
src/checks/health_checker.cpp | 2 +-
src/checks/health_checker.hpp | 3 +
4 files changed, 634 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/f042bf60/src/checks/checker.cpp
----------------------------------------------------------------------
diff --git a/src/checks/checker.cpp b/src/checks/checker.cpp
index 8716e4c..94315d7 100644
--- a/src/checks/checker.cpp
+++ b/src/checks/checker.cpp
@@ -16,20 +16,596 @@
#include "checks/checker.hpp"
+#include <cstdint>
+#include <iterator>
+#include <map>
#include <string>
+#include <tuple>
+#include <vector>
-#include <mesos/mesos.hpp>
+#include <glog/logging.h>
+#include <mesos/mesos.hpp>
+#include <mesos/type_utils.hpp>
+
+#include <process/collect.hpp>
+#include <process/defer.hpp>
+#include <process/delay.hpp>
+#include <process/future.hpp>
+#include <process/io.hpp>
+#include <process/protobuf.hpp>
+#include <process/subprocess.hpp>
+#include <process/time.hpp>
+
+#include <stout/check.hpp>
+#include <stout/duration.hpp>
+#include <stout/exit.hpp>
+#include <stout/foreach.hpp>
+#include <stout/jsonify.hpp>
+#include <stout/nothing.hpp>
+#include <stout/option.hpp>
+#include <stout/protobuf.hpp>
+#include <stout/stopwatch.hpp>
#include <stout/strings.hpp>
+#include <stout/try.hpp>
+#include <stout/unreachable.hpp>
+
+#include <stout/os/environment.hpp>
+#include <stout/os/killtree.hpp>
+#include "common/protobuf_utils.hpp"
+#include "common/status_utils.hpp"
#include "common/validation.hpp"
+#ifdef __linux__
+#include "linux/ns.hpp"
+#endif
+
+using process::Failure;
+using process::Future;
+using process::Owned;
+using process::Subprocess;
+
+using std::map;
using std::string;
+using std::tuple;
+using std::vector;
namespace mesos {
namespace internal {
namespace checks {
+#ifndef __WINDOWS__
+constexpr char HTTP_CHECK_COMMAND[] = "curl";
+#else
+constexpr char HTTP_CHECK_COMMAND[] = "curl.exe";
+#endif // __WINDOWS__
+
+static const string DEFAULT_HTTP_SCHEME = "http";
+
+// Use '127.0.0.1' instead of 'localhost', because the host
+// file in some container images may not contain 'localhost'.
+static const string DEFAULT_DOMAIN = "127.0.0.1";
+
+
+#ifdef __linux__
+// TODO(alexr): Instead of defining this ad-hoc clone function, provide a
+// general solution for entring namespace in child processes, see MESOS-6184.
+static pid_t cloneWithSetns(
+ const lambda::function<int()>& func,
+ const Option<pid_t>& taskPid,
+ const vector<string>& namespaces)
+{
+ return process::defaultClone([=]() -> int {
+ if (taskPid.isSome()) {
+ foreach (const string& ns, namespaces) {
+ Try<Nothing> setns = ns::setns(taskPid.get(), ns);
+ if (setns.isError()) {
+ // This effectively aborts the check.
+ LOG(FATAL) << "Failed to enter the " << ns << " namespace of "
+ << "task (pid: '" << taskPid.get() << "'): "
+ << setns.error();
+ }
+
+ VLOG(1) << "Entered the " << ns << " namespace of "
+ << "task (pid: '" << taskPid.get() << "') successfully";
+ }
+ }
+
+ return func();
+ });
+}
+#endif
+
+
+class CheckerProcess : public ProtobufProcess<CheckerProcess>
+{
+public:
+ CheckerProcess(
+ const CheckInfo& _check,
+ const lambda::function<void(const CheckStatusInfo&)>& _callback,
+ const TaskID& _taskId,
+ const Option<pid_t>& _taskPid,
+ const std::vector<std::string>& _namespaces);
+
+ virtual ~CheckerProcess() {}
+
+protected:
+ void initialize() override;
+ void finalize() override;
+
+private:
+ void performCheck();
+ void scheduleNext(const Duration& duration);
+ void processCheckResult(
+ const Stopwatch& stopwatch,
+ const CheckStatusInfo& result);
+
+ process::Future<int> commandCheck();
+ void processCommandCheckResult(
+ const Stopwatch& stopwatch,
+ const process::Future<int>& result);
+
+ process::Future<int> httpCheck();
+ process::Future<int> _httpCheck(
+ const std::tuple<
+ process::Future<Option<int>>,
+ process::Future<std::string>,
+ process::Future<std::string>>& t);
+ void processHttpCheckResult(
+ const Stopwatch& stopwatch,
+ const process::Future<int>& result);
+
+ const CheckInfo check;
+ Duration checkDelay;
+ Duration checkInterval;
+ Duration checkTimeout;
+
+ const lambda::function<void(const CheckStatusInfo&)> updateCallback;
+ const TaskID taskId;
+ const Option<pid_t> taskPid;
+ const std::vector<std::string> namespaces;
+ Option<lambda::function<pid_t(const lambda::function<int()>&)>> clone;
+
+ CheckStatusInfo previousCheckStatus;
+};
+
+
+Try<Owned<Checker>> Checker::create(
+ const CheckInfo& check,
+ const lambda::function<void(const CheckStatusInfo&)>& callback,
+ const TaskID& taskId,
+ const Option<pid_t>& taskPid,
+ const vector<string>& namespaces)
+{
+ // Validate the `CheckInfo` protobuf.
+ Option<Error> error = validation::checkInfo(check);
+ if (error.isSome()) {
+ return error.get();
+ }
+
+ Owned<CheckerProcess> process(new CheckerProcess(
+ check,
+ callback,
+ taskId,
+ taskPid,
+ namespaces));
+
+ return Owned<Checker>(new Checker(process));
+}
+
+
+Checker::Checker(Owned<CheckerProcess> _process)
+ : process(_process)
+{
+ spawn(CHECK_NOTNULL(process.get()));
+}
+
+
+Checker::~Checker()
+{
+ terminate(process.get());
+ wait(process.get());
+}
+
+
+void Checker::stop()
+{
+ terminate(process.get(), true);
+}
+
+
+CheckerProcess::CheckerProcess(
+ const CheckInfo& _check,
+ const lambda::function<void(const CheckStatusInfo&)>& _callback,
+ const TaskID& _taskId,
+ const Option<pid_t>& _taskPid,
+ const vector<string>& _namespaces)
+ : ProcessBase(process::ID::generate("checker")),
+ check(_check),
+ updateCallback(_callback),
+ taskId(_taskId),
+ taskPid(_taskPid),
+ namespaces(_namespaces)
+{
+ Try<Duration> create = Duration::create(check.delay_seconds());
+ CHECK_SOME(create);
+ checkDelay = create.get();
+
+ create = Duration::create(check.interval_seconds());
+ CHECK_SOME(create);
+ checkInterval = create.get();
+
+ // Zero value means infinite timeout.
+ create = Duration::create(check.timeout_seconds());
+ CHECK_SOME(create);
+ checkTimeout =
+ (create.get() > Duration::zero()) ? create.get() : Duration::max();
+
+ // The first check update should be sent only when a check succeeds,
+ // hence we should deduplicate against a corresponding "empty" result.
+ previousCheckStatus.set_type(check.type());
+ switch (check.type()) {
+ case CheckInfo::COMMAND: {
+ previousCheckStatus.mutable_command();
+ break;
+ }
+
+ case CheckInfo::HTTP: {
+ previousCheckStatus.mutable_http();
+ break;
+ }
+
+ case CheckInfo::UNKNOWN: {
+ LOG(FATAL) << "Received UNKNOWN check type";
+ break;
+ }
+ }
+
+#ifdef __linux__
+ if (!namespaces.empty()) {
+ clone = lambda::bind(&cloneWithSetns, lambda::_1, taskPid, namespaces);
+ }
+#endif
+}
+
+
+void CheckerProcess::initialize()
+{
+ VLOG(1) << "Check configuration for task " << taskId << ":"
+ << " '" << jsonify(JSON::Protobuf(check)) << "'";
+
+ scheduleNext(checkDelay);
+}
+
+
+void CheckerProcess::finalize()
+{
+ LOG(INFO) << "Checking for task " << taskId << " stopped";
+}
+
+
+void CheckerProcess::performCheck()
+{
+ Stopwatch stopwatch;
+ stopwatch.start();
+
+ switch (check.type()) {
+ case CheckInfo::COMMAND: {
+ commandCheck().onAny(defer(
+ self(),
+ &Self::processCommandCheckResult, stopwatch, lambda::_1));
+ break;
+ }
+
+ case CheckInfo::HTTP: {
+ httpCheck().onAny(defer(
+ self(),
+ &Self::processHttpCheckResult, stopwatch, lambda::_1));
+ break;
+ }
+
+ case CheckInfo::UNKNOWN: {
+ LOG(FATAL) << "Received UNKNOWN check type";
+ break;
+ }
+ }
+}
+
+
+void CheckerProcess::scheduleNext(const Duration& duration)
+{
+ VLOG(1) << "Scheduling check for task " << taskId << " in " << duration;
+
+ delay(duration, self(), &Self::performCheck);
+}
+
+
+void CheckerProcess::processCheckResult(
+ const Stopwatch& stopwatch,
+ const CheckStatusInfo& result)
+{
+ VLOG(1) << "Performed " << check.type() << " check for task " << taskId
+ << " in " << stopwatch.elapsed();
+
+ // Trigger the callback if check info changes.
+ if (result != previousCheckStatus) {
+ // We assume this is a local send, i.e., the checker library is not used
+ // in a binary external to the executor and hence can not exit before
+ // the data is sent to the executor.
+ updateCallback(result);
+ previousCheckStatus = result;
+ }
+
+ scheduleNext(checkInterval);
+}
+
+
+Future<int> CheckerProcess::commandCheck()
+{
+ CHECK_EQ(CheckInfo::COMMAND, check.type());
+ CHECK(check.has_command());
+
+ const CommandInfo& command = check.command().command();
+
+ map<string, string> environment = os::environment();
+
+ foreach (const Environment::Variable& variable,
+ command.environment().variables()) {
+ environment[variable.name()] = variable.value();
+ }
+
+ // Launch the subprocess.
+ Try<Subprocess> s = Error("Not launched");
+
+ if (command.shell()) {
+ // Use the shell variant.
+ VLOG(1) << "Launching command check '" << command.value() << "'"
+ << " for task " << taskId;
+
+ s = process::subprocess(
+ command.value(),
+ Subprocess::PATH("/dev/null"),
+ Subprocess::FD(STDERR_FILENO),
+ Subprocess::FD(STDERR_FILENO),
+ environment,
+ clone);
+ } else {
+ // Use the exec variant.
+ vector<string> argv(
+ std::begin(command.arguments()), std::end(command.arguments()));
+
+ VLOG(1) << "Launching command check [" << command.value() << ", "
+ << strings::join(", ", argv) << "] for task " << taskId;
+
+ s = process::subprocess(
+ command.value(),
+ argv,
+ Subprocess::PATH("/dev/null"),
+ Subprocess::FD(STDERR_FILENO),
+ Subprocess::FD(STDERR_FILENO),
+ nullptr,
+ environment,
+ clone);
+ }
+
+ if (s.isError()) {
+ return Failure("Failed to create subprocess: " + s.error());
+ }
+
+ // TODO(alexr): Use lambda named captures for
+ // these cached values once it is available.
+ const pid_t commandPid = s->pid();
+ const Duration timeout = checkTimeout;
+ const TaskID _taskId = taskId;
+
+ return s->status()
+ .after(
+ timeout,
+ [timeout, commandPid, _taskId](Future<Option<int>> future) {
+ future.discard();
+
+ if (commandPid != -1) {
+ // Cleanup the external command process.
+ VLOG(1) << "Killing the command check process " << commandPid
+ << " for task " << _taskId;
+
+ os::killtree(commandPid, SIGKILL);
+ }
+
+ return Failure(
+ "Command timed out after " + stringify(timeout) + "; aborting");
+ })
+ .then([](const Option<int>& exitCode) -> Future<int> {
+ if (exitCode.isNone()) {
+ return Failure("Failed to reap the command process");
+ }
+
+ return exitCode.get();
+ });
+}
+
+
+void CheckerProcess::processCommandCheckResult(
+ const Stopwatch& stopwatch,
+ const Future<int>& result)
+{
+ CheckStatusInfo checkStatusInfo;
+ checkStatusInfo.set_type(check.type());
+
+ if (result.isReady() && WIFEXITED(result.get())) {
+ const int exitCode = WEXITSTATUS(result.get());
+ VLOG(1) << check.type() << " check for task "
+ << taskId << " returned " << exitCode;
+
+ checkStatusInfo.mutable_command()->set_exit_code(
+ static_cast<int32_t>(exitCode));
+ } else {
+ // Check's status is currently not available, which may indicate a change
+ // that should be reported as an empty `CheckStatusInfo.Command` message.
+ LOG(WARNING) << "Check for task " << taskId << " failed: "
+ << (result.isFailed() ? result.failure() : "discarded");
+
+ checkStatusInfo.mutable_command();
+ }
+
+ processCheckResult(stopwatch, checkStatusInfo);
+}
+
+
+Future<int> CheckerProcess::httpCheck()
+{
+ CHECK_EQ(CheckInfo::HTTP, check.type());
+ CHECK(check.has_http());
+
+ const CheckInfo::Http& http = check.http();
+
+ const string scheme = DEFAULT_HTTP_SCHEME;
+ const string path = http.has_path() ? http.path() : "";
+ const string url = scheme + "://" + DEFAULT_DOMAIN + ":" +
+ stringify(http.port()) + path;
+
+ VLOG(1) << "Launching HTTP check '" << url << "' for task " << taskId;
+
+ const vector<string> argv = {
+ HTTP_CHECK_COMMAND,
+ "-s", // Don't show progress meter or error messages.
+ "-S", // Makes curl show an error message if it fails.
+ "-L", // Follows HTTP 3xx redirects.
+ "-k", // Ignores SSL validation when scheme is https.
+ "-w", "%{http_code}", // Displays HTTP response code on stdout.
+ "-o", "/dev/null", // Ignores output.
+ url
+ };
+
+ // TODO(alexr): Consider launching the helper binary once per task lifetime,
+ // see MESOS-6766.
+ Try<Subprocess> s = process::subprocess(
+ HTTP_CHECK_COMMAND,
+ argv,
+ Subprocess::PATH("/dev/null"),
+ Subprocess::PIPE(),
+ Subprocess::PIPE(),
+ nullptr,
+ None(),
+ clone);
+
+ if (s.isError()) {
+ return Failure(
+ "Failed to create the " + string(HTTP_CHECK_COMMAND) +
+ " subprocess: " + s.error());
+ }
+
+ // TODO(alexr): Use lambda named captures for
+ // these cached values once it is available.
+ const pid_t curlPid = s->pid();
+ const Duration timeout = checkTimeout;
+ const TaskID _taskId = taskId;
+
+ return await(
+ s->status(),
+ process::io::read(s->out().get()),
+ process::io::read(s->err().get()))
+ .after(
+ timeout,
+ [timeout, curlPid, _taskId](Future<tuple<Future<Option<int>>,
+ Future<string>,
+ Future<string>>> future) {
+ future.discard();
+
+ if (curlPid != -1) {
+ // Cleanup the HTTP_CHECK_COMMAND process.
+ VLOG(1) << "Killing the HTTP check process " << curlPid
+ << " for task " << _taskId;
+
+ os::killtree(curlPid, SIGKILL);
+ }
+
+ return Failure(
+ string(HTTP_CHECK_COMMAND) + " timed out after " +
+ stringify(timeout) + "; aborting");
+ })
+ .then(defer(self(), &Self::_httpCheck, lambda::_1));
+}
+
+
+Future<int> CheckerProcess::_httpCheck(
+ const tuple<
+ Future<Option<int>>,
+ Future<string>,
+ Future<string>>& t)
+{
+ Future<Option<int>> status = std::get<0>(t);
+ if (!status.isReady()) {
+ return Failure(
+ "Failed to get the exit status of the " + string(HTTP_CHECK_COMMAND) +
+ " process: " + (status.isFailed() ? status.failure() : "discarded"));
+ }
+
+ if (status->isNone()) {
+ return Failure(
+ "Failed to reap the " + string(HTTP_CHECK_COMMAND) + " process");
+ }
+
+ int exitCode = status->get();
+ if (exitCode != 0) {
+ Future<string> error = std::get<2>(t);
+ if (!error.isReady()) {
+ return Failure(
+ string(HTTP_CHECK_COMMAND) + " returned " +
+ WSTRINGIFY(exitCode) + "; reading stderr failed: " +
+ (error.isFailed() ? error.failure() : "discarded"));
+ }
+
+ return Failure(
+ string(HTTP_CHECK_COMMAND) + " returned " +
+ WSTRINGIFY(exitCode) + ": " + error.get());
+ }
+
+ Future<string> output = std::get<1>(t);
+ if (!output.isReady()) {
+ return Failure(
+ "Failed to read stdout from " + string(HTTP_CHECK_COMMAND) + ": " +
+ (output.isFailed() ? output.failure() : "discarded"));
+ }
+
+ // Parse the output and get the HTTP status code.
+ Try<int> statusCode = numify<int>(output.get());
+ if (statusCode.isError()) {
+ return Failure(
+ "Unexpected output from " + string(HTTP_CHECK_COMMAND) + ": " +
+ output.get());
+ }
+
+ return statusCode.get();
+}
+
+
+void CheckerProcess::processHttpCheckResult(
+ const Stopwatch& stopwatch,
+ const process::Future<int>& result)
+{
+ CheckStatusInfo checkStatusInfo;
+ checkStatusInfo.set_type(check.type());
+
+ if (result.isReady()) {
+ VLOG(1) << check.type() << " check for task "
+ << taskId << " returned " << result.get();
+
+ checkStatusInfo.mutable_http()->set_status_code(
+ static_cast<uint32_t>(result.get()));
+ } else {
+ // Check's status is currently not available, which may indicate a change
+ // that should be reported as an empty `CheckStatusInfo.Http` message.
+ LOG(WARNING) << "Check for task " << taskId << " failed: "
+ << (result.isFailed() ? result.failure() : "discarded");
+
+ checkStatusInfo.mutable_http();
+ }
+
+ processCheckResult(stopwatch, checkStatusInfo);
+}
+
namespace validation {
Option<Error> checkInfo(const CheckInfo& checkInfo)
http://git-wip-us.apache.org/repos/asf/mesos/blob/f042bf60/src/checks/checker.hpp
----------------------------------------------------------------------
diff --git a/src/checks/checker.hpp b/src/checks/checker.hpp
index dc293f3..e8af316 100644
--- a/src/checks/checker.hpp
+++ b/src/checks/checker.hpp
@@ -17,15 +17,68 @@
#ifndef __CHECKER_HPP__
#define __CHECKER_HPP__
+#include <string>
+#include <vector>
+
#include <mesos/mesos.hpp>
+#include <process/owned.hpp>
+
#include <stout/error.hpp>
+#include <stout/lambda.hpp>
#include <stout/option.hpp>
+#include <stout/try.hpp>
namespace mesos {
namespace internal {
namespace checks {
+// Forward declarations.
+class CheckerProcess;
+
+class Checker
+{
+public:
+ /**
+ * Attempts to create a `Checker` object. In case of success, checking
+ * starts immediately after initialization.
+ *
+ * @param check The protobuf message definition of a check.
+ * @param callback A callback `Checker` uses to send check status updates
+ * to its owner (usually an executor).
+ * @param taskId The TaskID of the target task.
+ * @param taskPid The target task's pid used to enter the specified
+ * namespaces.
+ * @param namespaces The namespaces to enter prior to performing the check.
+ * @return A `Checker` object or an error if `create` fails.
+ *
+ * @todo A better approach would be to return a stream of updates, e.g.,
+ * `process::Stream<CheckStatusInfo>` rather than invoking a callback.
+ */
+ static Try<process::Owned<Checker>> create(
+ const CheckInfo& checkInfo,
+ const lambda::function<void(const CheckStatusInfo&)>& callback,
+ const TaskID& taskId,
+ const Option<pid_t>& taskPid,
+ const std::vector<std::string>& namespaces);
+
+ ~Checker();
+
+ // Not copyable, not assignable.
+ Checker(const Checker&) = delete;
+ Checker& operator=(const Checker&) = delete;
+
+ /**
+ * Immediately stops checking. Any in-flight checks are dropped.
+ */
+ void stop();
+
+private:
+ explicit Checker(process::Owned<CheckerProcess> process);
+
+ process::Owned<CheckerProcess> process;
+};
+
namespace validation {
// TODO(alexr): A better place for these functions would be something like
http://git-wip-us.apache.org/repos/asf/mesos/blob/f042bf60/src/checks/health_checker.cpp
----------------------------------------------------------------------
diff --git a/src/checks/health_checker.cpp b/src/checks/health_checker.cpp
index 6c97369..7efefe9 100644
--- a/src/checks/health_checker.cpp
+++ b/src/checks/health_checker.cpp
@@ -92,7 +92,7 @@ static const string DEFAULT_DOMAIN = "127.0.0.1";
#ifdef __linux__
// TODO(alexr): Instead of defining this ad-hoc clone function, provide a
// general solution for entring namespace in child processes, see MESOS-6184.
-pid_t cloneWithSetns(
+static pid_t cloneWithSetns(
const lambda::function<int()>& func,
const Option<pid_t>& taskPid,
const vector<string>& namespaces)
http://git-wip-us.apache.org/repos/asf/mesos/blob/f042bf60/src/checks/health_checker.hpp
----------------------------------------------------------------------
diff --git a/src/checks/health_checker.hpp b/src/checks/health_checker.hpp
index f1f2834..44df544 100644
--- a/src/checks/health_checker.hpp
+++ b/src/checks/health_checker.hpp
@@ -64,6 +64,9 @@ public:
*
* @todo A better approach would be to return a stream of updates, e.g.,
* `process::Stream<TaskHealthStatus>` rather than invoking a callback.
+ *
+ * @todo Consider leveraging `checks::Checker` for checking functionality.
+ * This class will then focus on interpreting and acting on the result.
*/
static Try<process::Owned<HealthChecker>> create(
const HealthCheck& check,
[11/13] mesos git commit: Kept TaskInfo beyond first scheduler ack in
command executor.
Posted by al...@apache.org.
Kept TaskInfo beyond first scheduler ack in command executor.
Prior to this patch, command executor wipes TaskInfo after receiving
a status update acknowledgement from the scheduler to indicate that
there are no unacknowledged tasks. Keeping original TaskInfo beyond
the ack can be beneficial, hence we introduce a struct TaskData that
holds TaskInfo and explicit ack flag.
Review: https://reviews.apache.org/r/57596
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/a67231b0
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/a67231b0
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/a67231b0
Branch: refs/heads/master
Commit: a67231b004202a196370d6e6ea83fd75f76c7beb
Parents: 6b332ad
Author: Alexander Rukletsov <al...@apache.org>
Authored: Tue Mar 14 12:36:09 2017 +0100
Committer: Alexander Rukletsov <al...@apache.org>
Committed: Wed Mar 15 22:20:20 2017 +0100
----------------------------------------------------------------------
src/launcher/executor.cpp | 66 +++++++++++++++++++++++++-----------------
1 file changed, 40 insertions(+), 26 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/a67231b0/src/launcher/executor.cpp
----------------------------------------------------------------------
diff --git a/src/launcher/executor.cpp b/src/launcher/executor.cpp
index 57e4a32..02834c9 100644
--- a/src/launcher/executor.cpp
+++ b/src/launcher/executor.cpp
@@ -125,6 +125,7 @@ public:
const Duration& _shutdownGracePeriod)
: ProcessBase(process::ID::generate("command-executor")),
state(DISCONNECTED),
+ taskData(None()),
launched(false),
killed(false),
killedByHealthCheck(false),
@@ -143,7 +144,6 @@ public:
capabilities(_capabilities),
frameworkId(_frameworkId),
executorId(_executorId),
- unacknowledgedTask(None()),
lastTaskStatus(None())
{
#ifdef __WINDOWS__
@@ -229,8 +229,10 @@ public:
// Remove the corresponding update.
unacknowledgedUpdates.erase(uuid);
- // Remove the corresponding task.
- unacknowledgedTask = None();
+ // Mark task as acknowledged.
+ CHECK(taskData.isSome());
+ taskData->acknowledged = true;
+
break;
}
@@ -348,9 +350,9 @@ protected:
}
// Send the unacknowledged task.
- if (unacknowledgedTask.isSome()) {
+ if (taskData.isSome() && !taskData->acknowledged) {
subscribe->add_unacknowledged_tasks()->MergeFrom(
- unacknowledgedTask.get());
+ taskData->taskInfo);
}
mesos->send(evolve(call));
@@ -358,13 +360,13 @@ protected:
delay(Seconds(1), self(), &Self::doReliableRegistration);
}
- void launch(const TaskInfo& _task)
+ void launch(const TaskInfo& task)
{
CHECK_EQ(SUBSCRIBED, state);
if (launched) {
TaskStatus status = createTaskStatus(
- _task.task_id(),
+ task.task_id(),
TASK_FAILED,
None(),
"Attempted to run multiple tasks using a \"command\" executor");
@@ -374,14 +376,15 @@ protected:
}
// Capture the task.
- unacknowledgedTask = _task;
+ CHECK(taskData.isNone());
+ taskData = TaskData(task);
// Capture the TaskID.
- taskId = unacknowledgedTask->task_id();
+ taskId = task.task_id();
// Capture the kill policy.
- if (unacknowledgedTask->has_kill_policy()) {
- killPolicy = unacknowledgedTask->kill_policy();
+ if (task.has_kill_policy()) {
+ killPolicy = task.kill_policy();
}
// Determine the command to launch the task.
@@ -400,10 +403,10 @@ protected:
}
command = parse.get();
- } else if (unacknowledgedTask->has_command()) {
- command = unacknowledgedTask->command();
+ } else if (task.has_command()) {
+ command = task.command();
} else {
- LOG(FATAL) << "Expecting task '" << unacknowledgedTask->task_id()
+ LOG(FATAL) << "Expecting task '" << taskData.taskInfo->task_id()
<< "' to have a command";
}
@@ -414,11 +417,11 @@ protected:
// correct solution is to perform this validation at master side.
if (command.shell()) {
CHECK(command.has_value())
- << "Shell command of task '" << unacknowledgedTask->task_id()
+ << "Shell command of task '" << taskData.taskInfo->task_id()
<< "' is not specified!";
} else {
CHECK(command.has_value())
- << "Executable of task '" << unacknowledgedTask->task_id()
+ << "Executable of task '" << taskData.taskInfo->task_id()
<< "' is not specified!";
}
@@ -447,7 +450,7 @@ protected:
launchEnvironment.MergeFrom(command.environment());
}
- cout << "Starting task " << unacknowledgedTask->task_id() << endl;
+ cout << "Starting task " << taskData.taskInfo->task_id() << endl;
#ifndef __WINDOWS__
pid = launchTaskPosix(
@@ -475,10 +478,10 @@ protected:
cout << "Forked command at " << pid << endl;
- if (unacknowledgedTask->has_health_check()) {
+ if (task.has_health_check()) {
vector<string> namespaces;
if (rootfs.isSome() &&
- unacknowledgedTask->health_check().type() == HealthCheck::COMMAND) {
+ task.health_check().type() == HealthCheck::COMMAND) {
// Make sure command health checks are run from the task's mount
// namespace. Otherwise if rootfs is specified the command binary
// may not be available in the executor.
@@ -490,10 +493,10 @@ protected:
Try<Owned<checks::HealthChecker>> _healthChecker =
checks::HealthChecker::create(
- unacknowledgedTask->health_check(),
+ task.health_check(),
launcherDir,
defer(self(), &Self::taskHealthUpdated, lambda::_1),
- unacknowledgedTask->task_id(),
+ taskData.taskInfo->task_id(),
pid,
namespaces);
@@ -511,7 +514,7 @@ protected:
.onAny(defer(self(), &Self::reaped, pid, lambda::_1));
TaskStatus status =
- createTaskStatus(unacknowledgedTask->task_id(), TASK_RUNNING);
+ createTaskStatus(taskData.taskInfo->task_id(), TASK_RUNNING);
forward(status);
launched = true;
@@ -848,6 +851,21 @@ private:
SUBSCRIBED
} state;
+ struct TaskData
+ {
+ explicit TaskData(const TaskInfo& _taskInfo)
+ : taskInfo(_taskInfo), acknowledged(false) {}
+
+ TaskInfo taskInfo;
+
+ // Indicates whether a status update acknowledgement
+ // has been received for any status update.
+ bool acknowledged;
+ };
+
+ // Once `TaskInfo` is received, it is cached for later access.
+ Option<TaskData> taskData;
+
// TODO(alexr): Introduce a state enum and document transitions,
// see MESOS-5252.
bool launched;
@@ -880,10 +898,6 @@ private:
LinkedHashMap<UUID, Call::Update> unacknowledgedUpdates;
- // `None` if there is either no task yet or no status
- // update acknowledgements have been received yet.
- Option<TaskInfo> unacknowledgedTask;
-
Option<TaskStatus> lastTaskStatus;
Owned<checks::HealthChecker> healthChecker;
[10/13] mesos git commit: Added a warning if command executor gets an
unknown acknowledgement.
Posted by al...@apache.org.
Added a warning if command executor gets an unknown acknowledgement.
Review: https://reviews.apache.org/r/57149
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/702562cb
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/702562cb
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/702562cb
Branch: refs/heads/master
Commit: 702562cb9cded3dc86e88f704d650a958b98396c
Parents: f042bf6
Author: Alexander Rukletsov <al...@apache.org>
Authored: Tue Feb 28 14:05:33 2017 +0100
Committer: Alexander Rukletsov <al...@apache.org>
Committed: Wed Mar 15 22:20:20 2017 +0100
----------------------------------------------------------------------
src/launcher/executor.cpp | 6 ++++++
1 file changed, 6 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/702562cb/src/launcher/executor.cpp
----------------------------------------------------------------------
diff --git a/src/launcher/executor.cpp b/src/launcher/executor.cpp
index adcc50f..8b13b41 100644
--- a/src/launcher/executor.cpp
+++ b/src/launcher/executor.cpp
@@ -211,6 +211,12 @@ public:
case Event::ACKNOWLEDGED: {
const UUID uuid = UUID::fromBytes(event.acknowledged().uuid()).get();
+ if (!unacknowledgedUpdates.contains(uuid)) {
+ LOG(WARNING) << "Received acknowledgement " << uuid
+ << " for unknown status update";
+ return;
+ }
+
// Terminate if we receive the ACK for the terminal status update.
// NOTE: The executor receives an ACK iff it uses the HTTP library.
// No ACK will be received if V0ToV1Adapter is used.
[06/13] mesos git commit: Added a helper for building a task status
from an existing one.
Posted by al...@apache.org.
Added a helper for building a task status from an existing one.
Review: https://reviews.apache.org/r/56017
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/85fbf8fc
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/85fbf8fc
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/85fbf8fc
Branch: refs/heads/master
Commit: 85fbf8fc6465d46b40696f4f1e72822765e8fc1e
Parents: cc6e063
Author: Alexander Rukletsov <al...@apache.org>
Authored: Fri Jan 20 19:26:45 2017 +0100
Committer: Alexander Rukletsov <al...@apache.org>
Committed: Wed Mar 15 22:20:20 2017 +0100
----------------------------------------------------------------------
src/common/protobuf_utils.cpp | 62 ++++++++++++++++++++++++++++++++++++++
src/common/protobuf_utils.hpp | 27 +++++++++++++++++
2 files changed, 89 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/85fbf8fc/src/common/protobuf_utils.cpp
----------------------------------------------------------------------
diff --git a/src/common/protobuf_utils.cpp b/src/common/protobuf_utils.cpp
index 34c14e8..d568d41 100644
--- a/src/common/protobuf_utils.cpp
+++ b/src/common/protobuf_utils.cpp
@@ -206,6 +206,68 @@ StatusUpdate createStatusUpdate(
}
+TaskStatus createTaskStatus(
+ TaskStatus status,
+ const UUID& uuid,
+ double timestamp,
+ const Option<TaskState>& state,
+ const Option<string>& message,
+ const Option<TaskStatus::Source>& source,
+ const Option<TaskStatus::Reason>& reason,
+ const Option<string>& data,
+ const Option<bool>& healthy,
+ const Option<CheckStatusInfo>& checkStatus,
+ const Option<Labels>& labels,
+ const Option<ContainerStatus>& containerStatus,
+ const Option<TimeInfo>& unreachableTime)
+{
+ status.set_uuid(uuid.toBytes());
+ status.set_timestamp(timestamp);
+
+ if (state.isSome()) {
+ status.set_state(state.get());
+ }
+
+ if (message.isSome()) {
+ status.set_message(message.get());
+ }
+
+ if (source.isSome()) {
+ status.set_source(source.get());
+ }
+
+ if (reason.isSome()) {
+ status.set_reason(reason.get());
+ }
+
+ if (data.isSome()) {
+ status.set_data(data.get());
+ }
+
+ if (healthy.isSome()) {
+ status.set_healthy(healthy.get());
+ }
+
+ if (checkStatus.isSome()) {
+ status.mutable_check_status()->CopyFrom(checkStatus.get());
+ }
+
+ if (labels.isSome()) {
+ status.mutable_labels()->CopyFrom(labels.get());
+ }
+
+ if (containerStatus.isSome()) {
+ status.mutable_container_status()->CopyFrom(containerStatus.get());
+ }
+
+ if (unreachableTime.isSome()) {
+ status.mutable_unreachable_time()->CopyFrom(unreachableTime.get());
+ }
+
+ return status;
+}
+
+
Task createTask(
const TaskInfo& task,
const TaskState& state,
http://git-wip-us.apache.org/repos/asf/mesos/blob/85fbf8fc/src/common/protobuf_utils.hpp
----------------------------------------------------------------------
diff --git a/src/common/protobuf_utils.hpp b/src/common/protobuf_utils.hpp
index 09e468c..3b157a0 100644
--- a/src/common/protobuf_utils.hpp
+++ b/src/common/protobuf_utils.hpp
@@ -93,6 +93,33 @@ StatusUpdate createStatusUpdate(
const Option<SlaveID>& slaveId);
+// Helper function that creates a new task status from the given task
+// status. Specific fields in `status` can be overridden in the new
+// status by specifying the appropriate argument. Fields `task_id`,
+// `slave_id`, `executor_id`, cannot be changed; while `timestamp`
+// and `uuid` cannot be preserved.
+//
+// NOTE: A task status update may be used for guaranteed delivery of
+// some task-related information, e.g., task's health update. In this
+// case, it is often desirable to preserve specific fields from the
+// previous status update to avoid shadowing information that was
+// delivered previously.
+TaskStatus createTaskStatus(
+ TaskStatus status,
+ const UUID& uuid,
+ double timestamp,
+ const Option<TaskState>& state = None(),
+ const Option<std::string>& message = None(),
+ const Option<TaskStatus::Source>& source = None(),
+ const Option<TaskStatus::Reason>& reason = None(),
+ const Option<std::string>& data = None(),
+ const Option<bool>& healthy = None(),
+ const Option<CheckStatusInfo>& checkStatus = None(),
+ const Option<Labels>& labels = None(),
+ const Option<ContainerStatus>& containerStatus = None(),
+ const Option<TimeInfo>& unreachableTime = None());
+
+
Task createTask(
const TaskInfo& task,
const TaskState& state,