You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mesos.apache.org by bm...@apache.org on 2013/12/20 01:31:59 UTC
[2/7] git commit: Updated the slave to not recover after a reboot.
Updated the slave to not recover after a reboot.
Review: https://reviews.apache.org/r/16238
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/1c123520
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/1c123520
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/1c123520
Branch: refs/heads/master
Commit: 1c123520c596d120efccd10b7b92335a559ef167
Parents: 8d485a6
Author: Benjamin Mahler <bm...@twitter.com>
Authored: Thu Dec 12 20:09:19 2013 -0800
Committer: Benjamin Mahler <bm...@twitter.com>
Committed: Thu Dec 19 14:26:27 2013 -0800
----------------------------------------------------------------------
src/slave/paths.hpp | 11 ++++
src/slave/slave.cpp | 9 +++
src/slave/state.cpp | 13 ++++
src/tests/slave_recovery_tests.cpp | 103 ++++++++++++++++++++++++++++++++
4 files changed, 136 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/1c123520/src/slave/paths.hpp
----------------------------------------------------------------------
diff --git a/src/slave/paths.hpp b/src/slave/paths.hpp
index 8f80b93..7a1a001 100644
--- a/src/slave/paths.hpp
+++ b/src/slave/paths.hpp
@@ -44,6 +44,7 @@ const std::string LATEST_SYMLINK = "latest";
// Helper functions to generate paths.
// File names.
+const std::string BOOT_ID_FILE = "boot_id";
const std::string SLAVE_INFO_FILE = "slave.info";
const std::string FRAMEWORK_PID_FILE = "framework.pid";
const std::string FRAMEWORK_INFO_FILE = "framework.info";
@@ -60,6 +61,8 @@ 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 BOOT_ID_PATH =
+ path::join(SLAVE_PATH, BOOT_ID_FILE);
const std::string SLAVE_INFO_PATH =
path::join(SLAVE_PATH, SLAVE_INFO_FILE);
const std::string FRAMEWORK_PATH =
@@ -110,6 +113,14 @@ inline std::string getLatestSlavePath(const std::string& rootDir)
}
+inline std::string getBootIdPath(
+ const std::string& rootDir,
+ const SlaveID& slaveId)
+{
+ return strings::format(BOOT_ID_PATH, rootDir, slaveId).get();
+}
+
+
inline std::string getSlaveInfoPath(
const std::string& rootDir,
const SlaveID& slaveId)
http://git-wip-us.apache.org/repos/asf/mesos/blob/1c123520/src/slave/slave.cpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.cpp b/src/slave/slave.cpp
index 6e6107e..47e679f 100644
--- a/src/slave/slave.cpp
+++ b/src/slave/slave.cpp
@@ -549,6 +549,15 @@ void Slave::registered(const UPID& from, const SlaveID& slaveId)
// Create the slave meta directory.
paths::createSlaveDirectory(metaDir, slaveId);
+ // Checkpoint boot ID.
+ Try<string> bootId = os::bootId();
+ if (bootId.isError()) {
+ LOG(ERROR) << "Could not retrieve boot id: " << bootId.error();
+ } else {
+ const string& path = paths::getBootIdPath(metaDir, slaveId);
+ CHECK_SOME(state::checkpoint(path, bootId.get()));
+ }
+
// Checkpoint slave info.
const string& path = paths::getSlaveInfoPath(metaDir, slaveId);
http://git-wip-us.apache.org/repos/asf/mesos/blob/1c123520/src/slave/state.cpp
----------------------------------------------------------------------
diff --git a/src/slave/state.cpp b/src/slave/state.cpp
index bf267b5..93cde0c 100644
--- a/src/slave/state.cpp
+++ b/src/slave/state.cpp
@@ -61,6 +61,19 @@ Result<SlaveState> recover(const string& rootDir, bool strict)
SlaveID slaveId;
slaveId.set_value(os::basename(directory.get()).get());
+ // Did the machine reboot?
+ if (os::exists(paths::getBootIdPath(rootDir, slaveId))) {
+ Try<string> read = os::read(paths::getBootIdPath(rootDir, slaveId));
+ if (read.isSome()) {
+ Try<string> id = os::bootId();
+ CHECK_SOME(id);
+
+ if (id.get() != strings::trim(read.get())) {
+ return None(); // Machine rebooted!
+ }
+ }
+ }
+
Try<SlaveState> state = SlaveState::recover(rootDir, slaveId, strict);
if (state.isError()) {
return Error(state.error());
http://git-wip-us.apache.org/repos/asf/mesos/blob/1c123520/src/tests/slave_recovery_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/slave_recovery_tests.cpp b/src/tests/slave_recovery_tests.cpp
index 250083d..5a4c4fc 100644
--- a/src/tests/slave_recovery_tests.cpp
+++ b/src/tests/slave_recovery_tests.cpp
@@ -1334,6 +1334,109 @@ TYPED_TEST(SlaveRecoveryTest, KillTask)
}
+// When the slave is down we modify the BOOT_ID_FILE to simulate a
+// reboot. The subsequent run of the slave should not recover.
+TYPED_TEST(SlaveRecoveryTest, Reboot)
+{
+ Try<PID<Master> > master = this->StartMaster();
+ ASSERT_SOME(master);
+
+ TypeParam isolator1;
+
+ slave::Flags flags = this->CreateSlaveFlags();
+ flags.strict = false;
+
+ Try<PID<Slave> > slave = this->StartSlave(&isolator1, flags);
+ ASSERT_SOME(slave);
+
+ MockScheduler sched;
+
+ // Enable checkpointing for the framework.
+ FrameworkInfo frameworkInfo;
+ frameworkInfo.CopyFrom(DEFAULT_FRAMEWORK_INFO);
+ frameworkInfo.set_checkpoint(true);
+
+ MesosSchedulerDriver driver(
+ &sched, frameworkInfo, master.get(), DEFAULT_CREDENTIAL);
+
+ EXPECT_CALL(sched, registered(_, _, _));
+
+ Future<vector<Offer> > offers1;
+ EXPECT_CALL(sched, resourceOffers(_, _))
+ .WillOnce(FutureArg<1>(&offers1));
+
+ driver.start();
+
+ AWAIT_READY(offers1);
+ EXPECT_NE(0u, offers1.get().size());
+
+ TaskInfo task = createTask(offers1.get()[0], "sleep 1000");
+ vector<TaskInfo> tasks;
+ tasks.push_back(task); // Long-running task
+
+ // Capture the slave and framework ids.
+ SlaveID slaveId = offers1.get()[0].slave_id();
+ FrameworkID frameworkId = offers1.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(offers1.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->Stop(slave.get());
+
+ // Shut down the executor manually so that it doesn't hang around
+ // after the test finishes.
+ process::post(executorPid, ShutdownExecutorMessage());
+
+ // Modify the boot ID to simulate a reboot.
+ ASSERT_SOME(os::write(
+ paths::getBootIdPath(paths::getMetaRootDir(flags.work_dir), slaveId),
+ "rebooted! ;)"));
+
+ Future<RegisterSlaveMessage> registerSlave =
+ FUTURE_PROTOBUF(RegisterSlaveMessage(), _, _);
+
+ // Restart the slave (use same flags) with a new isolator.
+ TypeParam isolator2;
+
+ Future<vector<Offer> > offers2;
+ EXPECT_CALL(sched, resourceOffers(_, _))
+ .WillOnce(FutureArg<1>(&offers2))
+ .WillRepeatedly(Return()); // Ignore subsequent offers.
+
+ slave = this->StartSlave(&isolator2, flags);
+ ASSERT_SOME(slave);
+
+ AWAIT_READY(registerSlave);
+
+ // Make sure all slave resources are reoffered.
+ AWAIT_READY(offers2);
+ ASSERT_EQ(Resources(offers1.get()[0].resources()),
+ Resources(offers2.get()[0].resources()));
+
+ driver.stop();
+ driver.join();
+
+ this->Shutdown(); // Shutdown before isolator(s) get deallocated.
+}
+
+
// When the slave is down we remove the "latest" symlink in the
// executor's run directory, to simulate a situation where the
// recovered slave (--no-strict) cannot recover the executor and