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/04/17 09:08:57 UTC
svn commit: r1468773 - in /incubator/mesos/trunk/src: slave/slave.cpp
slave/slave.hpp slave/state.cpp slave/status_update_manager.cpp
slave/status_update_manager.hpp tests/gc_tests.cpp
tests/slave_recovery_tests.cpp
Author: vinodkone
Date: Wed Apr 17 07:08:56 2013
New Revision: 1468773
URL: http://svn.apache.org/r1468773
Log:
Fixed slave to properly schedule work and meta directories for
garbage collection.
Review: https://reviews.apache.org/r/9997
Modified:
incubator/mesos/trunk/src/slave/slave.cpp
incubator/mesos/trunk/src/slave/slave.hpp
incubator/mesos/trunk/src/slave/state.cpp
incubator/mesos/trunk/src/slave/status_update_manager.cpp
incubator/mesos/trunk/src/slave/status_update_manager.hpp
incubator/mesos/trunk/src/tests/gc_tests.cpp
incubator/mesos/trunk/src/tests/slave_recovery_tests.cpp
Modified: incubator/mesos/trunk/src/slave/slave.cpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/slave/slave.cpp?rev=1468773&r1=1468772&r2=1468773&view=diff
==============================================================================
--- incubator/mesos/trunk/src/slave/slave.cpp (original)
+++ incubator/mesos/trunk/src/slave/slave.cpp Wed Apr 17 07:08:56 2013
@@ -21,7 +21,9 @@
#include <algorithm>
#include <iomanip>
+#include <list>
#include <sstream>
+#include <string>
#include <vector>
#include <process/defer.hpp>
@@ -56,6 +58,7 @@
namespace params = std::tr1::placeholders;
+using std::list;
using std::string;
using std::vector;
@@ -83,7 +86,8 @@ Slave::Slave(const Resources& _resources
isolator(_isolator),
files(_files),
monitor(_isolator),
- statusUpdateManager(new StatusUpdateManager()) {}
+ statusUpdateManager(new StatusUpdateManager()),
+ metaDir(paths::getMetaRootDir(flags.work_dir)) {}
Slave::Slave(const slave::Flags& _flags,
@@ -97,7 +101,8 @@ Slave::Slave(const slave::Flags& _flags,
isolator(_isolator),
files(_files),
monitor(_isolator),
- statusUpdateManager(new StatusUpdateManager())
+ statusUpdateManager(new StatusUpdateManager()),
+ metaDir(paths::getMetaRootDir(flags.work_dir))
{
// TODO(benh): Move this computation into Flags as the "default".
@@ -386,6 +391,36 @@ void Slave::_initialize(const Future<Not
LOG(INFO) << "Finished recovery";
+ // Schedule all old slave directories for garbage collection.
+ // TODO(vinod): Do this as part of recovery. This needs a fix
+ // in the recovery code, to recover all slaves instead of only
+ // the latest slave.
+ const string& directory = path::join(flags.work_dir, "slaves");
+ foreach (const string& entry, os::ls(directory)) {
+ const string& path = path::join(directory, entry);
+ // Ignore non-directory entries.
+ if (!os::isdir(path)) {
+ continue;
+ }
+
+ // We garbage collect a directory if either the slave has not
+ // recovered its id (hence going to get a new id when it
+ // registers with the master) or if it is an old work directory.
+ SlaveID slaveId;
+ slaveId.set_value(entry);
+ if (!info.has_id() || !(slaveId == info.id())) {
+ LOG(INFO) << "Garbage collecting old slave " << slaveId;
+
+ // GC the slave work directory.
+ gc.schedule(flags.gc_delay, path);
+
+ if (os::exists(paths::getSlavePath(metaDir, slaveId))) {
+ // GC the slave meta directory.
+ gc.schedule(flags.gc_delay, paths::getSlavePath(metaDir, slaveId));
+ }
+ }
+ }
+
// Signal recovery.
recovered.set(Nothing());
}
@@ -508,20 +543,6 @@ void Slave::registered(const SlaveID& sl
CHECK_SOME(state::checkpoint(path, info));
}
-
- // Schedule all old slave directories to get garbage collected.
- // TODO(benh): It's unclear if we really need/want to
- // wait until the slave is registered to do this.
- const string& directory = path::join(flags.work_dir, "slaves");
-
- foreach (const string& file, os::ls(directory)) {
- const string& path = path::join(directory, file);
-
- // Check that this path is a directory but not our directory!
- if (os::isdir(path) && file != info.id().value()) {
- gc.schedule(flags.gc_delay, path);
- }
- }
}
@@ -617,8 +638,7 @@ void Slave::runTask(
Framework* framework = getFramework(frameworkId);
if (framework == NULL) {
- framework = new Framework(
- info.id(), frameworkId, frameworkInfo, pid, flags);
+ framework = new Framework(this, frameworkId, frameworkInfo, pid);
frameworks[frameworkId] = framework;
}
@@ -1777,17 +1797,46 @@ void Slave::cleanup(Framework* framework
(executor->updates.empty() ||
framework->state == Framework::TERMINATING)) {
- // Schedule the executor directory to get garbage collected.
+ // Schedule the executor run work directory to get garbage collected.
+ // TODO(vinod): Also schedule the top level executor work directory.
gc.schedule(flags.gc_delay, executor->directory)
.onAny(defer(self(), &Self::detachFile, params::_1, executor->directory));
+ if (executor->checkpoint) {
+ // Schedule the executor run meta directory to get garbage collected.
+ // TODO(vinod): Also schedule the top level executor meta directory.
+ const string& executorMetaDir = paths::getExecutorRunPath(
+ metaDir,
+ info.id(),
+ framework->id,
+ executor->id,
+ executor->uuid);
+
+ gc.schedule(flags.gc_delay, executorMetaDir)
+ .onAny(defer(self(), &Self::detachFile, params::_1, executorMetaDir));
+ }
+
framework->destroyExecutor(executor->id);
}
// Cleanup if this framework has no executors running.
- // TODO(vinod): If the framework is not being shutdown, remove
- // it after all its pending status updates are acknowledged.
if (framework->executors.empty()) {
+ // Schedule the framework work directory to get garbage collected.
+ const string& frameworkDir =
+ paths::getFrameworkPath(flags.work_dir, info.id(), framework->id);
+
+ gc.schedule(flags.gc_delay, frameworkDir)
+ .onAny(defer(self(), &Self::detachFile, params::_1, frameworkDir));
+
+ if (framework->info.checkpoint()) {
+ // Schedule the framework meta directory to get garbage collected.
+ const string& frameworkMetaDir = paths::getFrameworkPath(
+ metaDir, info.id(), framework->id);
+
+ gc.schedule(flags.gc_delay, frameworkMetaDir)
+ .onAny(defer(self(), &Self::detachFile, params::_1, frameworkMetaDir));
+ }
+
frameworks.erase(framework->id);
// Pass ownership of the framework pointer.
@@ -2082,8 +2131,8 @@ Future<Nothing> Slave::recover(bool reco
// the first time this slave was started with checkpointing enabled
// or this slave was started after an upgrade (--recover=cleanup).
if (!os::exists(metaDir)) {
- // NOTE: We recover the isolator here to cleanup any old executors
- // (e.g: orphaned cgroups).
+ // NOTE: We recover the isolator here to cleanup any old
+ // executors (e.g: orphaned cgroups).
return dispatch(isolator, &Isolator::recover, None());
}
@@ -2095,8 +2144,8 @@ Future<Nothing> Slave::recover(bool reco
if (state.isNone() || state.get().info.isNone()) {
// We are here if the slave died before checkpointing its info.
- // NOTE: We recover the isolator here to cleanup any old executors
- // (e.g: orphaned cgroups).
+ // NOTE: We recover the isolator here to cleanup any old
+ // executors (e.g: orphaned cgroups).
return dispatch(isolator, &Isolator::recover, None());
}
@@ -2152,13 +2201,19 @@ Future<Nothing> Slave::_recover(const Sl
void Slave::recoverFramework(const FrameworkState& state, bool reconnect)
{
if (state.executors.empty()) {
- // TODO(vinod): Garbage collect this framework.
+ // GC the framework work directory.
+ gc.schedule(flags.gc_delay,
+ paths::getFrameworkPath(flags.work_dir, info.id(), state.id));
+
+ // GC the framework meta directory.
+ gc.schedule(flags.gc_delay,
+ paths::getFrameworkPath(metaDir, info.id(), state.id));
return;
}
CHECK(!frameworks.contains(state.id));
Framework* framework = new Framework(
- info.id(), state.id, state.info.get(), state.pid.get(), flags);
+ this, state.id, state.info.get(), state.pid.get());
frameworks[framework->id] = framework;
@@ -2168,7 +2223,6 @@ void Slave::recoverFramework(const Frame
// Continue to next executor if this one couldn't be recovered.
if (executor == NULL) {
- // TODO(vinod): Garbage collect this executor.
continue;
}
@@ -2224,32 +2278,30 @@ void Slave::recoverFramework(const Frame
Framework::Framework(
- const SlaveID& _slaveId,
+ Slave* _slave,
const FrameworkID& _id,
const FrameworkInfo& _info,
- const UPID& _pid,
- const Flags& _flags)
+ const UPID& _pid)
: state(RUNNING), // TODO(benh): Skipping INITIALIZING for now.
- slaveId(_slaveId),
+ slave(_slave),
id(_id),
info(_info),
pid(_pid),
- flags(_flags),
completedExecutors(MAX_COMPLETED_EXECUTORS_PER_FRAMEWORK)
{
if (info.checkpoint()) {
// Checkpoint the framework info.
string path = paths::getFrameworkInfoPath(
- paths::getMetaRootDir(flags.work_dir),
- slaveId,
+ paths::getMetaRootDir(slave->flags.work_dir),
+ slave->info.id(),
id);
CHECK_SOME(state::checkpoint(path, info));
// Checkpoint the framework pid.
path = paths::getFrameworkPidPath(
- paths::getMetaRootDir(flags.work_dir),
- slaveId,
+ paths::getMetaRootDir(slave->flags.work_dir),
+ slave->info.id(),
id);
CHECK_SOME(state::checkpoint(path, pid));
@@ -2280,6 +2332,7 @@ ExecutorInfo Framework::getExecutorInfo(
// command being launched.
string name =
"(Task: " + task.task_id().value() + ") " + "(Command: sh -c '";
+
if (task.command().value().length() > 15) {
name += task.command().value().substr(0, 12) + "...')";
} else {
@@ -2296,7 +2349,7 @@ ExecutorInfo Framework::getExecutorInfo(
executor.mutable_command()->MergeFrom(task.command());
Try<string> path = os::realpath(
- path::join(flags.launcher_dir, "mesos-executor"));
+ path::join(slave->flags.launcher_dir, "mesos-executor"));
if (path.isSome()) {
executor.mutable_command()->set_value(path.get());
@@ -2329,10 +2382,10 @@ Executor* Framework::createExecutor(cons
// Create a directory for the executor.
const string& directory = paths::createExecutorDirectory(
- flags.work_dir, slaveId, id, executorInfo.executor_id(), uuid);
+ slave->flags.work_dir, slave->info.id(), id, executorInfo.executor_id(), uuid);
Executor* executor = new Executor(
- slaveId, id, executorInfo, uuid, directory, flags, info.checkpoint());
+ slave, id, executorInfo, uuid, directory, info.checkpoint());
CHECK(!executors.contains(executorInfo.executor_id()));
executors[executorInfo.executor_id()] = executor;
@@ -2380,30 +2433,48 @@ Executor* Framework::recoverExecutor(con
LOG(INFO) << "Recovering executor '" << state.id
<< "' of framework " << id;
- if (state.info.isNone()) {
- LOG(WARNING) << "Skipping recovery of executor '" << state.id
- << "' of framework " << id
- << " because its info cannot be recovered";
- return NULL;
- }
+ CHECK_NOTNULL(slave);
- if (state.latest.isNone()) {
+ if (state.runs.empty() || state.latest.isNone()) {
LOG(WARNING) << "Skipping recovery of executor '" << state.id
<< "' of framework " << id
<< " because its latest run cannot be recovered";
+
+ // GC the executor work directory.
+ slave->gc.schedule(slave->flags.gc_delay, paths::getExecutorPath(
+ slave->flags.work_dir, slave->info.id(), id, state.id));
+
+ // GC the executor meta directory.
+ slave->gc.schedule(slave->flags.gc_delay, paths::getExecutorPath(
+ slave->metaDir, slave->info.id(), id, state.id));
+
return NULL;
}
// We are only interested in the latest run of the executor!
- // TODO(vinod): Garbage collect older runs.
+ // So, we GC all the old runs.
const UUID& uuid = state.latest.get();
+ foreachvalue (const RunState& run, state.runs) {
+ CHECK_SOME(run.id);
+ if (uuid != run.id.get()) {
+ // GC the 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 run's meta directory.
+ slave->gc.schedule(slave->flags.gc_delay, paths::getExecutorRunPath(
+ slave->metaDir, slave->info.id(), id, state.id, run.id.get()));
+ }
+ }
+
+ CHECK_NOTNULL(slave);
// Create executor.
const string& directory = paths::getExecutorRunPath(
- flags.work_dir, slaveId, id, state.id, uuid);
+ slave->flags.work_dir, slave->info.id(), id, state.id, uuid);
Executor* executor = new Executor(
- slaveId, id, state.info.get(), uuid, directory, flags, info.checkpoint());
+ slave, id, state.info.get(), uuid, directory, info.checkpoint());
CHECK(state.runs.contains(uuid));
const RunState& run = state.runs.get(uuid).get();
@@ -2432,31 +2503,31 @@ Executor* Framework::recoverExecutor(con
Executor::Executor(
- const SlaveID& _slaveId,
+ Slave* _slave,
const FrameworkID& _frameworkId,
const ExecutorInfo& _info,
const UUID& _uuid,
const string& _directory,
- const Flags& _flags,
bool _checkpoint)
: state(REGISTERING), // TODO(benh): Skipping INITIALIZING for now.
- slaveId(_slaveId),
+ slave(_slave),
id(_info.executor_id()),
info(_info),
frameworkId(_frameworkId),
uuid(_uuid),
directory(_directory),
- flags(_flags),
checkpoint(_checkpoint),
pid(UPID()),
resources(_info.resources()),
completedTasks(MAX_COMPLETED_TASKS_PER_EXECUTOR)
{
if (checkpoint) {
+ CHECK_NOTNULL(slave);
+
// Checkpoint the executor info.
const string& path = paths::getExecutorInfoPath(
- paths::getMetaRootDir(flags.work_dir),
- slaveId,
+ paths::getMetaRootDir(slave->flags.work_dir),
+ slave->info.id(),
frameworkId,
id);
@@ -2465,8 +2536,8 @@ Executor::Executor(
// Create the meta executor directory.
// NOTE: This creates the 'latest' symlink in the meta directory.
paths::createExecutorDirectory(
- paths::getMetaRootDir(flags.work_dir),
- slaveId,
+ paths::getMetaRootDir(slave->flags.work_dir),
+ slave->info.id(),
frameworkId,
id,
uuid);
@@ -2521,9 +2592,11 @@ void Executor::removeTask(const TaskID&
void Executor::checkpointTask(const TaskInfo& task)
{
if (checkpoint) {
+ CHECK_NOTNULL(slave);
+
const string& path = paths::getTaskInfoPath(
- paths::getMetaRootDir(flags.work_dir),
- slaveId,
+ paths::getMetaRootDir(slave->flags.work_dir),
+ slave->info.id(),
frameworkId,
id,
uuid,
Modified: incubator/mesos/trunk/src/slave/slave.hpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/slave/slave.hpp?rev=1468773&r1=1468772&r2=1468773&view=diff
==============================================================================
--- incubator/mesos/trunk/src/slave/slave.hpp (original)
+++ incubator/mesos/trunk/src/slave/slave.hpp Wed Apr 17 07:08:56 2013
@@ -164,13 +164,22 @@ public:
bool destroyed,
const std::string& message);
- // NOTE: Pulled this to public to make it visible for testing.
+ // NOTE: Pulled these to public to make it visible for testing.
+ // TODO(vinod): Make tests friends to this class instead.
+
// Garbage collects the directories based on the current disk usage.
// TODO(vinod): Instead of making this function public, we need to
// mock both GarbageCollector (and pass it through slave's constructor)
// and os calls.
void _checkDiskUsage(const Future<Try<double> >& capacity);
+ // Shut down an executor. This is a two phase process. First, an
+ // executor receives a shut down message (shut down phase), then
+ // after a configurable timeout the slave actually forces a kill
+ // (kill phase, via the isolator) if the executor has not
+ // exited.
+ void shutdownExecutor(Framework* framework, Executor* executor);
+
protected:
virtual void initialize();
virtual void finalize();
@@ -186,13 +195,6 @@ protected:
// Helper routine to lookup a framework.
Framework* getFramework(const FrameworkID& frameworkId);
- // Shut down an executor. This is a two phase process. First, an
- // executor receives a shut down message (shut down phase), then
- // after a configurable timeout the slave actually forces a kill
- // (kill phase, via the isolator) if the executor has not
- // exited.
- void shutdownExecutor(Framework* framework, Executor* executor);
-
// Handle the second phase of shutting down an executor for those
// executors that have not properly shutdown within a timeout.
void shutdownExecutorTimeout(
@@ -255,6 +257,9 @@ private:
const Slave& slave,
const process::http::Request& request);
+ friend class Framework;
+ friend class Executor;
+
const Flags flags;
bool local;
@@ -298,6 +303,9 @@ private:
Promise<Nothing> recovered;
bool halting; // Flag to indicate if the slave is shutting down.
+
+ // Root meta directory containing checkpointed data.
+ const std::string metaDir;
};
@@ -305,13 +313,13 @@ private:
struct Executor
{
Executor(
- const SlaveID& _slaveId,
- const FrameworkID& _frameworkId,
- const ExecutorInfo& _info,
- const UUID& _uuid,
- const std::string& _directory,
- const Flags& _flags,
- bool _checkpoint);
+ Slave* slave,
+ const FrameworkID& frameworkId,
+ const ExecutorInfo& info,
+ const UUID& uuid,
+ const std::string& directory,
+ bool checkpoint);
+
~Executor();
Task* addTask(const TaskInfo& task);
@@ -327,7 +335,10 @@ struct Executor
TERMINATED, // Executor has terminated but there might be pending updates.
} state;
- const SlaveID slaveId;
+ // We store the pointer to 'Slave' to get access to its methods
+ // variables. One could imagine 'Executor' as being an inner class
+ // of the 'Slave' class.
+ Slave* slave;
const ExecutorID id;
const ExecutorInfo info;
@@ -338,8 +349,6 @@ struct Executor
const std::string directory;
- const Flags flags;
-
const bool checkpoint;
UPID pid;
@@ -363,11 +372,10 @@ private:
struct Framework
{
Framework(
- const SlaveID& _slaveId,
- const FrameworkID& _id,
- const FrameworkInfo& _info,
- const UPID& _pid,
- const Flags& _flags);
+ Slave* slave,
+ const FrameworkID& id,
+ const FrameworkInfo& info,
+ const UPID& pid);
~Framework();
@@ -385,15 +393,16 @@ struct Framework
TERMINATING, // Framework is shutting down in the cluster.
} state;
- const SlaveID slaveId;
+ // We store the pointer to 'Slave' to get access to its methods
+ // variables. One could imagine 'Framework' as being an inner class
+ // of the 'Slave' class.
+ Slave* slave;
const FrameworkID id;
const FrameworkInfo info;
UPID pid;
- const Flags flags;
-
std::vector<TaskInfo> pending; // Pending tasks (used while INITIALIZING).
// Current running executors.
Modified: incubator/mesos/trunk/src/slave/state.cpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/slave/state.cpp?rev=1468773&r1=1468772&r2=1468773&view=diff
==============================================================================
--- incubator/mesos/trunk/src/slave/state.cpp (original)
+++ incubator/mesos/trunk/src/slave/state.cpp Wed Apr 17 07:08:56 2013
@@ -40,6 +40,8 @@ Result<SlaveState> recover(const string&
if (safe) {
return Error(message);
} else {
+ // The slave likely died before it registered and had a chance
+ // to create the "latest" symlink.
LOG(WARNING) << message;
return None();
}
Modified: incubator/mesos/trunk/src/slave/status_update_manager.cpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/slave/status_update_manager.cpp?rev=1468773&r1=1468772&r2=1468773&view=diff
==============================================================================
--- incubator/mesos/trunk/src/slave/status_update_manager.cpp (original)
+++ incubator/mesos/trunk/src/slave/status_update_manager.cpp Wed Apr 17 07:08:56 2013
@@ -495,9 +495,6 @@ void StatusUpdateManagerProcess::cleanup
streams.erase(frameworkId);
}
- // Delete the task meta directory.
- stream->cleanup();
-
delete stream;
}
Modified: incubator/mesos/trunk/src/slave/status_update_manager.hpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/slave/status_update_manager.hpp?rev=1468773&r1=1468772&r2=1468773&view=diff
==============================================================================
--- incubator/mesos/trunk/src/slave/status_update_manager.hpp (original)
+++ incubator/mesos/trunk/src/slave/status_update_manager.hpp Wed Apr 17 07:08:56 2013
@@ -279,17 +279,6 @@ struct StatusUpdateStream
return Nothing();
}
- // Delete the task meta directory.
- // TODO(vinod): Archive it.
- void cleanup() {
- if (path.isSome()) {
- LOG(INFO) << "Deleting the meta directory for task " << taskId
- << " of framework " << frameworkId;
-
- os::rmdir(os::basename(path.get()).get());
- }
- }
-
// TODO(vinod): Explore semantics to make these private.
const bool checkpoint;
bool terminated;
Modified: incubator/mesos/trunk/src/tests/gc_tests.cpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/tests/gc_tests.cpp?rev=1468773&r1=1468772&r2=1468773&view=diff
==============================================================================
--- incubator/mesos/trunk/src/tests/gc_tests.cpp (original)
+++ incubator/mesos/trunk/src/tests/gc_tests.cpp Wed Apr 17 07:08:56 2013
@@ -18,6 +18,11 @@
#include <gmock/gmock.h>
+#include <list>
+#include <map>
+#include <string>
+#include <vector>
+
#include <mesos/executor.hpp>
#include <mesos/scheduler.hpp>
@@ -65,13 +70,15 @@ using process::Clock;
using process::Future;
using process::PID;
-using std::string;
+using std::list;
using std::map;
+using std::string;
using std::vector;
using testing::_;
using testing::AtMost;
using testing::Return;
+using testing::SaveArg;
class GarbageCollectorTest : public TemporaryDirectoryTest {};
@@ -241,6 +248,7 @@ TEST_F(GarbageCollectorTest, Prune)
class GarbageCollectorIntegrationTest : public MesosClusterTest {};
+
TEST_F(GarbageCollectorIntegrationTest, Restart)
{
ASSERT_TRUE(GTEST_IS_THREADSAFE);
@@ -342,6 +350,105 @@ TEST_F(GarbageCollectorIntegrationTest,
}
+TEST_F(GarbageCollectorIntegrationTest, ExitedFramework)
+{
+ ASSERT_TRUE(GTEST_IS_THREADSAFE);
+
+ Try<PID<Master> > master = cluster.masters.start();
+ ASSERT_SOME(master);
+
+ Future<SlaveRegisteredMessage> slaveRegisteredMessage =
+ FUTURE_PROTOBUF(SlaveRegisteredMessage(), _, _);
+
+ MockExecutor exec;
+
+ Try<PID<Slave> > slave = cluster.slaves.start(DEFAULT_EXECUTOR_ID, &exec);
+ ASSERT_SOME(slave);
+
+ AWAIT_READY(slaveRegisteredMessage);
+ SlaveID slaveId = slaveRegisteredMessage.get().slave_id();
+
+ MockScheduler sched;
+ MesosSchedulerDriver driver(&sched, DEFAULT_FRAMEWORK_INFO, master.get());
+
+ // Scheduler expectations.
+ FrameworkID frameworkId;
+ EXPECT_CALL(sched, registered(_, _, _))
+ .WillOnce(SaveArg<1>(&frameworkId));
+
+ Resources resources = Resources::parse(cluster.slaves.flags.resources.get());
+ double cpus = resources.get("cpus", Value::Scalar()).value();
+ double mem = resources.get("mem", Value::Scalar()).value();
+
+ EXPECT_CALL(sched, resourceOffers(_, _))
+ .WillOnce(LaunchTasks(1, cpus, mem))
+ .WillRepeatedly(Return());
+
+ // Executor expectations.
+ EXPECT_CALL(exec, registered(_, _, _, _))
+ .WillRepeatedly(Return());
+
+ EXPECT_CALL(exec, launchTask(_, _))
+ .WillRepeatedly(SendStatusUpdateFromTask(TASK_RUNNING));
+
+ Future<TaskStatus> status;
+ EXPECT_CALL(sched, statusUpdate(_, _))
+ .WillOnce(FutureArg<1>(&status))
+ .WillRepeatedly(Return()); // Ignore subsequent updates.
+
+ driver.start();
+
+ AWAIT_READY(status);
+
+ EXPECT_EQ(TASK_RUNNING, status.get().state());
+
+ Future<Nothing> shutdown;
+ EXPECT_CALL(exec, shutdown(_))
+ .WillOnce(FutureSatisfy(&shutdown));
+
+ // Shutdown the framework.
+ driver.stop();
+ driver.join();
+
+ Clock::pause();
+
+ AWAIT_READY(shutdown);
+
+ Clock::settle(); // Wait for Slave::shutdownExecutor to complete.
+
+ Future<Nothing> schedule =
+ FUTURE_DISPATCH(_, &GarbageCollectorProcess::schedule);
+
+ // Advance clock to kill executor via isolator.
+ Clock::advance(cluster.slaves.flags.executor_shutdown_grace_period);
+
+ Clock::settle();
+
+ AWAIT_READY(schedule);
+
+ Clock::settle(); // Wait for GarbageCollectorProcess::schedule to complete.
+
+ Clock::advance(cluster.slaves.flags.gc_delay);
+
+ Clock::settle();
+
+ // Framework's directory should be gc'ed by now.
+ const string& frameworkDir = slave::paths::getFrameworkPath(
+ cluster.slaves.flags.work_dir, slaveId, frameworkId);
+
+ ASSERT_FALSE(os::exists(frameworkDir));
+
+ process::UPID filesUpid("files", process::ip(), process::port());
+ EXPECT_RESPONSE_STATUS_WILL_EQ(
+ process::http::NotFound().status,
+ process::http::get(filesUpid, "browse.json", "path=" + frameworkDir));
+
+ Clock::resume();
+
+ cluster.shutdown(); // Must shutdown before 'isolator' gets deallocated.
+}
+
+
TEST_F(GarbageCollectorIntegrationTest, ExitedExecutor)
{
ASSERT_TRUE(GTEST_IS_THREADSAFE);
Modified: incubator/mesos/trunk/src/tests/slave_recovery_tests.cpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/tests/slave_recovery_tests.cpp?rev=1468773&r1=1468772&r2=1468773&view=diff
==============================================================================
--- incubator/mesos/trunk/src/tests/slave_recovery_tests.cpp (original)
+++ incubator/mesos/trunk/src/tests/slave_recovery_tests.cpp Wed Apr 17 07:08:56 2013
@@ -966,3 +966,110 @@ TYPED_TEST(SlaveRecoveryTest, KillTask)
driver.stop();
driver.join();
}
+
+
+// When the slave is down we remove the "latest" symlink in the
+// executor's run directory, to simulate a situation where the slave
+// cannot recover the executor and hence schedules it for gc.
+TYPED_TEST(SlaveRecoveryTest, GCExecutor)
+{
+ // Scheduler expectations.
+ MockScheduler sched;
+ EXPECT_CALL(sched, registered(_, _, _));
+
+ Future<vector<Offer> > offers;
+ EXPECT_CALL(sched, resourceOffers(_, _))
+ .WillOnce(FutureArg<1>(&offers));
+
+ // Enable checkpointing for the framework.
+ FrameworkInfo frameworkInfo;
+ frameworkInfo.CopyFrom(DEFAULT_FRAMEWORK_INFO);
+ frameworkInfo.set_checkpoint(true);
+
+ MesosSchedulerDriver driver(&sched, frameworkInfo, this->master);
+
+ driver.start();
+
+ AWAIT_READY(offers);
+ EXPECT_NE(0u, offers.get().size());
+
+ TaskInfo task = createTask(offers.get()[0], "sleep 1000");
+ vector<TaskInfo> tasks;
+ tasks.push_back(task); // Long-running task
+
+ // Capture the slave and framework ids.
+ SlaveID slaveId = offers.get()[0].slave_id();
+ FrameworkID frameworkId = offers.get()[0].framework_id();
+
+ Future<Message> registerExecutorMessage =
+ FUTURE_MESSAGE(Eq(RegisterExecutorMessage().GetTypeName()), _, _);
+
+ Future<Nothing> status;
+ EXPECT_CALL(sched, statusUpdate(_, _))
+ .WillOnce(FutureSatisfy(&status))
+ .WillRepeatedly(Return()); // Ignore subsequent updates.
+
+ driver.launchTasks(offers.get()[0].id(), tasks);
+
+ // Capture the executor id and pid.
+ AWAIT_READY(registerExecutorMessage);
+ RegisterExecutorMessage registerExecutor;
+ registerExecutor.ParseFromString(registerExecutorMessage.get().body);
+ ExecutorID executorId = registerExecutor.executor_id();
+ UPID executorPid = registerExecutorMessage.get().from;
+
+ // Wait for TASK_RUNNING update.
+ AWAIT_READY(status);
+
+ this->stopSlave();
+
+ // Now shut down the executor, when the slave is down.
+ process::post(executorPid, ShutdownExecutorMessage());
+
+ // Remove the symlink "latest" in the executor directory
+ // to simulate a non-recoverable executor.
+ ASSERT_SOME(os::rm(paths::getExecutorLatestRunPath(
+ paths::getMetaRootDir(this->slaveFlags.work_dir),
+ slaveId,
+ frameworkId,
+ executorId)));
+
+ Future<Nothing> recover = FUTURE_DISPATCH(_, &Slave::_recover);
+
+ Future<ReregisterSlaveMessage> reregisterSlave =
+ FUTURE_PROTOBUF(ReregisterSlaveMessage(), _, _);
+
+ // Restart the slave.
+ this->startSlave();
+
+ Clock::pause();
+
+ AWAIT_READY(recover);
+
+ Clock::settle(); // Wait for slave to schedule reregister timeout.
+
+ Clock::advance(EXECUTOR_REREGISTER_TIMEOUT);
+
+ Clock::settle();
+
+ AWAIT_READY(reregisterSlave);
+
+ Clock::advance(this->slaveFlags.gc_delay);
+
+ Clock::settle();
+
+ // Executor's work and meta directories should be gc'ed by now.
+ ASSERT_FALSE(os::exists(paths::getExecutorPath(
+ this->slaveFlags.work_dir, slaveId, frameworkId, executorId)));
+
+ ASSERT_FALSE(os::exists(paths::getExecutorPath(
+ paths::getMetaRootDir(this->slaveFlags.work_dir),
+ slaveId,
+ frameworkId,
+ executorId)));
+
+ Clock::resume();
+
+ driver.stop();
+ driver.join();
+}