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 2013/08/07 04:58:04 UTC
[3/5] git commit: Fixed slave to not recover completed executors.
Fixed slave to not recover completed executors.
Review: https://reviews.apache.org/r/13253
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/acc24164
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/acc24164
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/acc24164
Branch: refs/heads/master
Commit: acc24164dd12d74a8776c85ef99dc4e62d7b40d4
Parents: febd337
Author: Vinod Kone <vi...@twitter.com>
Authored: Fri Aug 2 17:35:09 2013 -0700
Committer: Vinod Kone <vi...@twitter.com>
Committed: Tue Aug 6 19:57:26 2013 -0700
----------------------------------------------------------------------
src/slave/cgroups_isolator.cpp | 7 +++
src/slave/paths.hpp | 65 +++++++++++++++++------
src/slave/process_isolator.cpp | 7 +++
src/slave/slave.cpp | 90 +++++++++++++++++---------------
src/slave/state.cpp | 7 +++
src/slave/state.hpp | 1 +
src/slave/status_update_manager.cpp | 8 +++
src/tests/slave_recovery_tests.cpp | 2 +-
8 files changed, 129 insertions(+), 58 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/acc24164/src/slave/cgroups_isolator.cpp
----------------------------------------------------------------------
diff --git a/src/slave/cgroups_isolator.cpp b/src/slave/cgroups_isolator.cpp
index 7f6d13e..3427c62 100644
--- a/src/slave/cgroups_isolator.cpp
+++ b/src/slave/cgroups_isolator.cpp
@@ -768,6 +768,13 @@ Future<Nothing> CgroupsIsolator::recover(
CHECK(executor.runs.contains(uuid));
const RunState& run = executor.runs.get(uuid).get();
+ if (run.completed) {
+ VLOG(1) << "Skipping recovery of executor '" << executor.id
+ << "' of framework " << framework.id
+ << " because its latest run " << uuid << " is completed";
+ continue;
+ }
+
// TODO(vinod): Currently, we assume that the cgroups
// information (e.g., hierarchy, root) used while recovering
// is same as the one that was used by the previous slave
http://git-wip-us.apache.org/repos/asf/mesos/blob/acc24164/src/slave/paths.hpp
----------------------------------------------------------------------
diff --git a/src/slave/paths.hpp b/src/slave/paths.hpp
index 9d2a2a4..3a02a0d 100644
--- a/src/slave/paths.hpp
+++ b/src/slave/paths.hpp
@@ -49,31 +49,47 @@ const std::string FRAMEWORK_PID_FILE = "framework.pid";
const std::string FRAMEWORK_INFO_FILE = "framework.info";
const std::string LIBPROCESS_PID_FILE = "libprocess.pid";
const std::string EXECUTOR_INFO_FILE = "executor.info";
+const std::string EXECUTOR_SENTINEL_FILE = "executor.sentinel";
const std::string FORKED_PID_FILE = "forked.pid";
const std::string TASK_INFO_FILE = "task.info";
const std::string TASK_UPDATES_FILE = "task.updates";
// Path layout templates.
const std::string ROOT_PATH = "%s";
-const std::string LATEST_SLAVE_PATH = ROOT_PATH + "/slaves/" + LATEST_SYMLINK;
-const std::string SLAVE_PATH = ROOT_PATH + "/slaves/%s";
-const std::string SLAVE_INFO_PATH = SLAVE_PATH + "/" + SLAVE_INFO_FILE;
-const std::string FRAMEWORK_PATH = SLAVE_PATH + "/frameworks/%s";
+const std::string LATEST_SLAVE_PATH =
+ path::join(ROOT_PATH, "slaves", LATEST_SYMLINK);
+const std::string SLAVE_PATH =
+ path::join(ROOT_PATH, "slaves", "%s");
+const std::string SLAVE_INFO_PATH =
+ path::join(SLAVE_PATH, SLAVE_INFO_FILE);
+const std::string FRAMEWORK_PATH =
+ path::join(SLAVE_PATH, "frameworks", "%s");
const std::string FRAMEWORK_PID_PATH =
- FRAMEWORK_PATH + "/" + FRAMEWORK_PID_FILE;
+ path::join(FRAMEWORK_PATH, FRAMEWORK_PID_FILE);
const std::string FRAMEWORK_INFO_PATH =
- FRAMEWORK_PATH + "/" + FRAMEWORK_INFO_FILE;
-const std::string EXECUTOR_PATH = FRAMEWORK_PATH + "/executors/%s";
-const std::string EXECUTOR_INFO_PATH = EXECUTOR_PATH + "/" + EXECUTOR_INFO_FILE;
-const std::string EXECUTOR_RUN_PATH = EXECUTOR_PATH + "/runs/%s";
+ path::join(FRAMEWORK_PATH, FRAMEWORK_INFO_FILE);
+const std::string EXECUTOR_PATH =
+ path::join(FRAMEWORK_PATH, "executors", "%s");
+const std::string EXECUTOR_INFO_PATH =
+ path::join(EXECUTOR_PATH, EXECUTOR_INFO_FILE);
+const std::string EXECUTOR_RUN_PATH =
+ path::join(EXECUTOR_PATH, "runs", "%s");
+const std::string EXECUTOR_SENTINEL_PATH =
+ path::join(EXECUTOR_RUN_PATH, EXECUTOR_SENTINEL_FILE);
const std::string EXECUTOR_LATEST_RUN_PATH =
- EXECUTOR_PATH + "/runs/" + LATEST_SYMLINK;
-const std::string PIDS_PATH = EXECUTOR_RUN_PATH + "/pids";
-const std::string LIBPROCESS_PID_PATH = PIDS_PATH + "/" + LIBPROCESS_PID_FILE;
-const std::string FORKED_PID_PATH = PIDS_PATH + "/" + FORKED_PID_FILE;
-const std::string TASK_PATH = EXECUTOR_RUN_PATH + "/tasks/%s";
-const std::string TASK_INFO_PATH = TASK_PATH + "/" + TASK_INFO_FILE;
-const std::string TASK_UPDATES_PATH = TASK_PATH + "/" + TASK_UPDATES_FILE;
+ path::join(EXECUTOR_PATH, "runs", LATEST_SYMLINK);
+const std::string PIDS_PATH =
+ path::join(EXECUTOR_RUN_PATH, "pids");
+const std::string LIBPROCESS_PID_PATH =
+ path::join(PIDS_PATH, LIBPROCESS_PID_FILE);
+const std::string FORKED_PID_PATH =
+ path::join(PIDS_PATH, FORKED_PID_FILE);
+const std::string TASK_PATH =
+ path::join(EXECUTOR_RUN_PATH, "tasks", "%s");
+const std::string TASK_INFO_PATH =
+ path::join(TASK_PATH, TASK_INFO_FILE);
+const std::string TASK_UPDATES_PATH =
+ path::join(TASK_PATH, TASK_UPDATES_FILE);
inline std::string getMetaRootDir(const std::string rootDir)
@@ -179,6 +195,23 @@ inline std::string getExecutorRunPath(
}
+inline std::string getExecutorSentinelPath(
+ const std::string& rootDir,
+ const SlaveID& slaveId,
+ const FrameworkID& frameworkId,
+ const ExecutorID& executorId,
+ const UUID& executorUUID)
+{
+ return strings::format(
+ EXECUTOR_SENTINEL_PATH,
+ rootDir,
+ slaveId,
+ frameworkId,
+ executorId,
+ executorUUID.toString()).get();
+}
+
+
inline std::string getExecutorLatestRunPath(
const std::string& rootDir,
const SlaveID& slaveId,
http://git-wip-us.apache.org/repos/asf/mesos/blob/acc24164/src/slave/process_isolator.cpp
----------------------------------------------------------------------
diff --git a/src/slave/process_isolator.cpp b/src/slave/process_isolator.cpp
index cb07448..a80b047 100644
--- a/src/slave/process_isolator.cpp
+++ b/src/slave/process_isolator.cpp
@@ -342,6 +342,13 @@ Future<Nothing> ProcessIsolator::recover(
CHECK(executor.runs.contains(uuid));
const RunState& run = executor.runs.get(uuid).get();
+ if (run.completed) {
+ VLOG(1) << "Skipping recovery of executor '" << executor.id
+ << "' of framework " << framework.id
+ << " because its latest run " << uuid << " is completed";
+ continue;
+ }
+
ProcessInfo* info =
new ProcessInfo(framework.id, executor.id, run.forkedPid);
http://git-wip-us.apache.org/repos/asf/mesos/blob/acc24164/src/slave/slave.cpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.cpp b/src/slave/slave.cpp
index 9cd7754..7378e01 100644
--- a/src/slave/slave.cpp
+++ b/src/slave/slave.cpp
@@ -602,11 +602,10 @@ void Slave::registered(const SlaveID& slaveId)
if (flags.checkpoint) {
// Create the slave meta directory.
- paths::createSlaveDirectory(paths::getMetaRootDir(flags.work_dir), slaveId);
+ paths::createSlaveDirectory(metaDir, slaveId);
// Checkpoint slave info.
- const string& path = paths::getSlaveInfoPath(
- paths::getMetaRootDir(flags.work_dir), slaveId);
+ const string& path = paths::getSlaveInfoPath(metaDir, slaveId);
CHECK_SOME(state::checkpoint(path, info));
}
@@ -1308,9 +1307,7 @@ void Slave::updateFramework(const FrameworkID& frameworkId, const string& pid)
if (framework->info.checkpoint()) {
// Checkpoint the framework pid.
const string& path = paths::getFrameworkPidPath(
- paths::getMetaRootDir(flags.work_dir),
- info.id(),
- frameworkId);
+ metaDir, info.id(), frameworkId);
CHECK_SOME(state::checkpoint(path, framework->pid));
}
@@ -1489,7 +1486,7 @@ void Slave::registerExecutor(
// Checkpoint the libprocess pid.
string path = paths::getLibprocessPidPath(
- paths::getMetaRootDir(flags.work_dir),
+ metaDir,
info.id(),
executor->frameworkId,
executor->id,
@@ -2219,6 +2216,14 @@ void Slave::removeExecutor(Framework* framework, Executor* executor)
CHECK(framework->state == Framework::TERMINATING ||
!executor->incompleteTasks());
+ // Write a sentinel file to indicate that this executor
+ // is completed.
+ if (executor->checkpoint) {
+ const string& path = paths::getExecutorSentinelPath(
+ metaDir, info.id(), framework->id, executor->id, executor->uuid);
+ CHECK_SOME(os::touch(path));
+ }
+
// TODO(vinod): Move the responsibility of gc'ing to the
// Executor struct.
@@ -2463,13 +2468,9 @@ void Slave::registerExecutorTimeout(
switch (executor->state) {
case Executor::RUNNING:
- // Executor has registered. Ignore the registration timeout.
- break;
case Executor::TERMINATING:
case Executor::TERMINATED:
- LOG(INFO) << "Ignoring registration timeout for executor '" << executorId
- << "' of framework " << frameworkId
- << " because the executor is terminating/terminated";
+ // Ignore the registration timeout.
break;
case Executor::REGISTERING:
LOG(INFO) << "Terminating executor " << executor->id
@@ -2539,8 +2540,6 @@ void Slave::_checkDiskUsage(const Future<Try<double> >& usage)
Future<Nothing> Slave::recover(bool reconnect, bool strict)
{
- const string& metaDir = paths::getMetaRootDir(flags.work_dir);
-
// We consider the absence of 'metaDir' to mean that this is either
// the first time this slave was started with checkpointing enabled
// or this slave was started after an upgrade (--recover=cleanup).
@@ -2717,17 +2716,13 @@ Framework::Framework(
if (info.checkpoint() && slave->state != slave->RECOVERING) {
// Checkpoint the framework info.
string path = paths::getFrameworkInfoPath(
- paths::getMetaRootDir(slave->flags.work_dir),
- slave->info.id(),
- id);
+ slave->metaDir, slave->info.id(), id);
CHECK_SOME(state::checkpoint(path, info));
// Checkpoint the framework pid.
path = paths::getFrameworkPidPath(
- paths::getMetaRootDir(slave->flags.work_dir),
- slave->info.id(),
- id);
+ slave->metaDir, slave->info.id(), id);
CHECK_SOME(state::checkpoint(path, pid));
}
@@ -2876,6 +2871,36 @@ Executor* Framework::recoverExecutor(const ExecutorState& state)
}
}
+ CHECK(state.runs.contains(uuid));
+ const RunState& run = state.runs.get(uuid).get();
+
+ // If the latest run of the executor was completed (i.e., terminated
+ // and all updates are acknowledged) in the previous run, we don't
+ // recover it.
+ if (run.completed) {
+ VLOG(1) << "Skipping recovery of executor " << state.id
+ << " of framework " << id
+ << " because its latest run " << uuid << " is completed";
+
+ // GC the executor run's work directory.
+ slave->gc.schedule(slave->flags.gc_delay, paths::getExecutorRunPath(
+ slave->flags.work_dir, slave->info.id(), id, state.id, run.id.get()));
+
+ // GC the executor run's meta directory.
+ slave->gc.schedule(slave->flags.gc_delay, paths::getExecutorRunPath(
+ slave->metaDir, slave->info.id(), id, state.id, run.id.get()));
+
+ // GC the top level executor work directory.
+ slave->gc.schedule(slave->flags.gc_delay, paths::getExecutorPath(
+ slave->flags.work_dir, slave->info.id(), id, state.id));
+
+ // GC the top level executor meta directory.
+ slave->gc.schedule(slave->flags.gc_delay, paths::getExecutorPath(
+ slave->metaDir, slave->info.id(), id, state.id));
+
+ return NULL;
+ }
+
// Create executor.
const string& directory = paths::getExecutorRunPath(
slave->flags.work_dir, slave->info.id(), id, state.id, uuid);
@@ -2883,9 +2908,6 @@ Executor* Framework::recoverExecutor(const ExecutorState& state)
Executor* executor = new Executor(
slave, id, state.info.get(), uuid, directory, info.checkpoint());
- CHECK(state.runs.contains(uuid));
- const RunState& run = state.runs.get(uuid).get();
-
// Recover the libprocess PID if possible.
if (run.libprocessPid.isSome()) {
// When recovering in non-strict mode, the assumption is that the
@@ -2935,21 +2957,14 @@ Executor::Executor(
if (checkpoint && slave->state != slave->RECOVERING) {
// Checkpoint the executor info.
const string& path = paths::getExecutorInfoPath(
- paths::getMetaRootDir(slave->flags.work_dir),
- slave->info.id(),
- frameworkId,
- id);
+ slave->metaDir, slave->info.id(), frameworkId, id);
CHECK_SOME(state::checkpoint(path, info));
// Create the meta executor directory.
// NOTE: This creates the 'latest' symlink in the meta directory.
paths::createExecutorDirectory(
- paths::getMetaRootDir(slave->flags.work_dir),
- slave->info.id(),
- frameworkId,
- id,
- uuid);
+ slave->metaDir, slave->info.id(), frameworkId, id, uuid);
}
}
@@ -3022,16 +3037,9 @@ void Executor::checkpointTask(const TaskInfo& task)
if (checkpoint) {
CHECK_NOTNULL(slave);
+ const Task& t = protobuf::createTask(task, TASK_STAGING, id, frameworkId);
const string& path = paths::getTaskInfoPath(
- paths::getMetaRootDir(slave->flags.work_dir),
- slave->info.id(),
- frameworkId,
- id,
- uuid,
- task.task_id());
-
- const Task& t = protobuf::createTask(
- task, TASK_STAGING, id, frameworkId);
+ slave->metaDir, slave->info.id(), frameworkId, id, uuid, t.task_id());
CHECK_SOME(state::checkpoint(path, t));
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/acc24164/src/slave/state.cpp
----------------------------------------------------------------------
diff --git a/src/slave/state.cpp b/src/slave/state.cpp
index e910ab7..77b29dc 100644
--- a/src/slave/state.cpp
+++ b/src/slave/state.cpp
@@ -293,6 +293,7 @@ Try<RunState> RunState::recover(
{
RunState state;
state.id = uuid;
+ state.completed = false;
string message;
// Find the tasks.
@@ -372,6 +373,12 @@ Try<RunState> RunState::recover(
state.libprocessPid = process::UPID(pid.get());
+ // See if the sentinel file exists.
+ path = paths::getExecutorSentinelPath(
+ rootDir, slaveId, frameworkId, executorId, uuid);
+
+ state.completed = os::exists(path);
+
return state;
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/acc24164/src/slave/state.hpp
----------------------------------------------------------------------
diff --git a/src/slave/state.hpp b/src/slave/state.hpp
index 08e3617..7e88705 100644
--- a/src/slave/state.hpp
+++ b/src/slave/state.hpp
@@ -117,6 +117,7 @@ struct RunState
hashmap<TaskID, TaskState> tasks;
Option<pid_t> forkedPid;
Option<process::UPID> libprocessPid;
+ bool completed; // Executor terminated and all its updates acknowledged.
};
http://git-wip-us.apache.org/repos/asf/mesos/blob/acc24164/src/slave/status_update_manager.cpp
----------------------------------------------------------------------
diff --git a/src/slave/status_update_manager.cpp b/src/slave/status_update_manager.cpp
index e17ecf4..ffd4736 100644
--- a/src/slave/status_update_manager.cpp
+++ b/src/slave/status_update_manager.cpp
@@ -204,6 +204,14 @@ Future<Nothing> StatusUpdateManagerProcess::recover(
CHECK(executor.runs.contains(uuid));
const RunState& run = executor.runs.get(uuid).get();
+ if (run.completed) {
+ VLOG(1) << "Skipping recovering updates of"
+ << " executor '" << executor.id
+ << "' of framework " << framework.id
+ << " because its latest run " << uuid << " is completed";
+ continue;
+ }
+
foreachvalue (const TaskState& task, run.tasks) {
// No updates were ever received for this task!
// This means either:
http://git-wip-us.apache.org/repos/asf/mesos/blob/acc24164/src/tests/slave_recovery_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/slave_recovery_tests.cpp b/src/tests/slave_recovery_tests.cpp
index c451e0f..9bfce72 100644
--- a/src/tests/slave_recovery_tests.cpp
+++ b/src/tests/slave_recovery_tests.cpp
@@ -686,7 +686,7 @@ TYPED_TEST(SlaveRecoveryTest, RecoverTerminatedExecutor)
// terminated and all its updates have been acknowledged).
// When it comes back up with recovery=reconnect, make
// sure the recovery successfully completes.
-TYPED_TEST(SlaveRecoveryTest, RecoverPreviouslyTerminatedExecutor)
+TYPED_TEST(SlaveRecoveryTest, RecoverCompletedExecutor)
{
Try<PID<Master> > master = this->StartMaster();
ASSERT_SOME(master);