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/03/20 23:53:26 UTC
svn commit: r1459093 - in /incubator/mesos/trunk: src/master/ src/slave/
src/tests/ third_party/libprocess/include/process/
Author: vinodkone
Date: Wed Mar 20 22:53:23 2013
New Revision: 1459093
URL: http://svn.apache.org/r1459093
Log:
Added registration timeout for executors.
Review: https://reviews.apache.org/r/8077
Modified:
incubator/mesos/trunk/src/master/hierarchical_allocator_process.hpp
incubator/mesos/trunk/src/master/master.cpp
incubator/mesos/trunk/src/slave/cgroups_isolator.cpp
incubator/mesos/trunk/src/slave/cgroups_isolator.hpp
incubator/mesos/trunk/src/slave/constants.cpp
incubator/mesos/trunk/src/slave/constants.hpp
incubator/mesos/trunk/src/slave/flags.hpp
incubator/mesos/trunk/src/slave/process_isolator.cpp
incubator/mesos/trunk/src/slave/slave.cpp
incubator/mesos/trunk/src/slave/slave.hpp
incubator/mesos/trunk/src/tests/master_tests.cpp
incubator/mesos/trunk/third_party/libprocess/include/process/delay.hpp
Modified: incubator/mesos/trunk/src/master/hierarchical_allocator_process.hpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/master/hierarchical_allocator_process.hpp?rev=1459093&r1=1459092&r2=1459093&view=diff
==============================================================================
--- incubator/mesos/trunk/src/master/hierarchical_allocator_process.hpp (original)
+++ incubator/mesos/trunk/src/master/hierarchical_allocator_process.hpp Wed Mar 20 22:53:23 2013
@@ -20,6 +20,7 @@
#define __HIERARCHICAL_ALLOCATOR_PROCESS_HPP__
#include <process/delay.hpp>
+#include <process/id.hpp>
#include <process/timeout.hpp>
#include <stout/duration.hpp>
@@ -203,7 +204,8 @@ public:
template <class UserSorter, class FrameworkSorter>
HierarchicalAllocatorProcess<UserSorter, FrameworkSorter>::HierarchicalAllocatorProcess()
- : initialized(false) {}
+ : ProcessBase(ID::generate("hierarchical-allocator")),
+ initialized(false) {}
template <class UserSorter, class FrameworkSorter>
Modified: incubator/mesos/trunk/src/master/master.cpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/master/master.cpp?rev=1459093&r1=1459092&r2=1459093&view=diff
==============================================================================
--- incubator/mesos/trunk/src/master/master.cpp (original)
+++ incubator/mesos/trunk/src/master/master.cpp Wed Mar 20 22:53:23 2013
@@ -63,7 +63,9 @@ namespace master {
class WhitelistWatcher : public Process<WhitelistWatcher> {
public:
WhitelistWatcher(const string& _path, Allocator* _allocator)
- : path(_path), allocator(_allocator) {}
+ : ProcessBase(ID::generate("whitelist")),
+ path(_path),
+ allocator(_allocator) {}
protected:
virtual void initialize()
Modified: incubator/mesos/trunk/src/slave/cgroups_isolator.cpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/slave/cgroups_isolator.cpp?rev=1459093&r1=1459092&r2=1459093&view=diff
==============================================================================
--- incubator/mesos/trunk/src/slave/cgroups_isolator.cpp (original)
+++ incubator/mesos/trunk/src/slave/cgroups_isolator.cpp Wed Mar 20 22:53:23 2013
@@ -988,10 +988,10 @@ void CgroupsIsolator::oom(
<< " of framework " << frameworkId
<< " with uuid " << uuid;
- // Construct a "reason" string to describe why the isolator
+ // Construct a "message" string to describe why the isolator
// destroyed the executor's cgroup (in order to assist in debugging).
- ostringstream reason;
- reason << "Memory limit exceeded: ";
+ ostringstream message;
+ message << "Memory limit exceeded: ";
Try<string> read = cgroups::read(
hierarchy, info->name(), "memory.limit_in_bytes");
@@ -1000,9 +1000,9 @@ void CgroupsIsolator::oom(
if (bytes.isError()) {
LOG(ERROR)
<< "Failed to numify 'memory.limit_in_bytes': " << bytes.error();
- reason << "Requested: " << strings::trim(read.get()) << " bytes ";
+ message << "Requested: " << strings::trim(read.get()) << " bytes ";
} else {
- reason << "Requested: " << Bytes(bytes.get()) << " ";
+ message << "Requested: " << Bytes(bytes.get()) << " ";
}
}
@@ -1013,22 +1013,22 @@ void CgroupsIsolator::oom(
if (bytes.isError()) {
LOG(ERROR)
<< "Failed to numify 'memory.usage_in_bytes': " << bytes.error();
- reason << "Used: " << strings::trim(read.get()) << " bytes\n";
+ message << "Used: " << strings::trim(read.get()) << " bytes\n";
} else {
- reason << "Used: " << Bytes(bytes.get()) << "\n";
+ message << "Used: " << Bytes(bytes.get()) << "\n";
}
}
// Output 'memory.stat' of the cgroup to help with debugging.
read = cgroups::read(hierarchy, info->name(), "memory.stat");
if (read.isSome()) {
- reason << "\nMEMORY STATISTICS: \n" << read.get() << "\n";
+ message << "\nMEMORY STATISTICS: \n" << read.get() << "\n";
}
- LOG(INFO) << strings::trim(reason.str()); // Trim the extra '\n' at the end.
+ LOG(INFO) << strings::trim(message.str()); // Trim the extra '\n' at the end.
info->destroyed = true;
- info->reason = reason.str();
+ info->message = message.str();
killExecutor(frameworkId, executorId);
}
@@ -1074,7 +1074,7 @@ void CgroupsIsolator::_killExecutor(
info->executorId,
info->status,
info->destroyed,
- info->reason);
+ info->message);
// We make a copy here because 'info' will be deleted when we unregister.
unregisterCgroupInfo(
@@ -1102,7 +1102,7 @@ CgroupsIsolator::CgroupInfo* CgroupsIsol
info->killed = false;
info->destroyed = false;
info->status = -1;
- info->reason = "";
+ info->message = "";
info->flags = flags;
if (subsystems.contains("cpuset")) {
info->cpuset = new Cpuset();
Modified: incubator/mesos/trunk/src/slave/cgroups_isolator.hpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/slave/cgroups_isolator.hpp?rev=1459093&r1=1459092&r2=1459093&view=diff
==============================================================================
--- incubator/mesos/trunk/src/slave/cgroups_isolator.hpp (original)
+++ incubator/mesos/trunk/src/slave/cgroups_isolator.hpp Wed Mar 20 22:53:23 2013
@@ -163,7 +163,7 @@ private:
// PID of the forked process of the executor.
Option<pid_t> pid;
- bool killed; // True if "killing" has been initiated via 'killExecutor'.
+ bool killed; // True if "killing" has been initiated via 'killExecutor()'.
// Indicates if this executor has been destroyed by the isolator.
// NOTE: An executor may have terminated due to reasons
@@ -171,7 +171,7 @@ private:
// slave, exited, etc.).
bool destroyed;
- std::string reason; // The reason behind the destruction.
+ std::string message; // The reason behind the destruction.
int status; // Exit status of the executor.
Modified: incubator/mesos/trunk/src/slave/constants.cpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/slave/constants.cpp?rev=1459093&r1=1459092&r2=1459093&view=diff
==============================================================================
--- incubator/mesos/trunk/src/slave/constants.cpp (original)
+++ incubator/mesos/trunk/src/slave/constants.cpp Wed Mar 20 22:53:23 2013
@@ -22,6 +22,7 @@ namespace mesos {
namespace internal {
namespace slave {
+const Duration EXECUTOR_REGISTRATION_TIMEOUT = Minutes(1.0);
const Duration EXECUTOR_SHUTDOWN_GRACE_PERIOD = Seconds(5.0);
const Duration EXECUTOR_REREGISTER_TIMEOUT = Seconds(2.0);
const Duration STATUS_UPDATE_RETRY_INTERVAL = Seconds(10.0);
Modified: incubator/mesos/trunk/src/slave/constants.hpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/slave/constants.hpp?rev=1459093&r1=1459092&r2=1459093&view=diff
==============================================================================
--- incubator/mesos/trunk/src/slave/constants.hpp (original)
+++ incubator/mesos/trunk/src/slave/constants.hpp Wed Mar 20 22:53:23 2013
@@ -33,6 +33,7 @@ namespace slave {
// So we've moved these to have external linkage but perhaps in the future
// we can revert this.
+extern const Duration EXECUTOR_REGISTRATION_TIMEOUT;
extern const Duration EXECUTOR_SHUTDOWN_GRACE_PERIOD;
extern const Duration EXECUTOR_REREGISTER_TIMEOUT;
extern const Duration STATUS_UPDATE_RETRY_INTERVAL;
Modified: incubator/mesos/trunk/src/slave/flags.hpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/slave/flags.hpp?rev=1459093&r1=1459092&r2=1459093&view=diff
==============================================================================
--- incubator/mesos/trunk/src/slave/flags.hpp (original)
+++ incubator/mesos/trunk/src/slave/flags.hpp Wed Mar 20 22:53:23 2013
@@ -83,6 +83,13 @@ public:
"Directory prepended to relative executor URIs",
"");
+ add(&Flags::executor_registration_timeout,
+ "executor_registration_timeout",
+ "Amount of time to wait for an executor\n"
+ "to register with the slave before considering it hung and\n"
+ "shutting it down (e.g., 60secs, 3mins, etc)",
+ EXECUTOR_REGISTRATION_TIMEOUT);
+
add(&Flags::executor_shutdown_grace_period,
"executor_shutdown_grace_period",
"Amount of time to wait for an executor\n"
@@ -163,6 +170,7 @@ public:
std::string hadoop_home; // TODO(benh): Make an Option.
bool switch_user;
std::string frameworks_home; // TODO(benh): Make an Option.
+ Duration executor_registration_timeout;
Duration executor_shutdown_grace_period;
Duration gc_delay;
Duration disk_watch_interval;
Modified: incubator/mesos/trunk/src/slave/process_isolator.cpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/slave/process_isolator.cpp?rev=1459093&r1=1459092&r2=1459093&view=diff
==============================================================================
--- incubator/mesos/trunk/src/slave/process_isolator.cpp (original)
+++ incubator/mesos/trunk/src/slave/process_isolator.cpp Wed Mar 20 22:53:23 2013
@@ -420,7 +420,7 @@ void ProcessIsolator::processExited(pid_
executorId,
status,
false,
- "Executor exited");
+ "Executor terminated");
if (!info->killed) {
// Try and cleanup after the executor.
Modified: incubator/mesos/trunk/src/slave/slave.cpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/slave/slave.cpp?rev=1459093&r1=1459092&r2=1459093&view=diff
==============================================================================
--- incubator/mesos/trunk/src/slave/slave.cpp (original)
+++ incubator/mesos/trunk/src/slave/slave.cpp Wed Mar 20 22:53:23 2013
@@ -48,6 +48,7 @@
#include "logging/logging.hpp"
+#include "slave/constants.hpp"
#include "slave/flags.hpp"
#include "slave/paths.hpp"
#include "slave/slave.hpp"
@@ -801,6 +802,16 @@ void Slave::runTask(
// Queue task until the executor starts up.
executor->queuedTasks[task.task_id()] = task;
+ // Make sure the executor registers within the given timeout.
+ // NOTE: We send this message before dispatching the launchExecutor to
+ // the isolator, to make writing tests easier.
+ delay(flags.executor_registration_timeout,
+ self(),
+ &Slave::registerExecutorTimeout,
+ framework->id,
+ executor->id,
+ executor->uuid);
+
// Tell the isolator to launch the executor. (TODO(benh): Make the
// isolator a process so that it can block while trying to launch
// the executor.)
@@ -1002,6 +1013,7 @@ void Slave::_statusUpdateAcknowledgement
<< " for task " << taskId
<< " of framework " << frameworkId
<< (future.isFailed() ? future.failure() : "future discarded");
+ return;
}
if (future.get().isError()) {
@@ -1557,6 +1569,8 @@ void Slave::executorTerminated(
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()) {
frameworks.erase(framework->id);
@@ -1654,15 +1668,78 @@ void Slave::shutdownExecutorTimeout(
{
Framework* framework = getFramework(frameworkId);
if (framework == NULL) {
+ LOG(INFO) << "Framework " << frameworkId
+ << " seems to have exited. Ignoring shutdown timeout"
+ << " for executor '" << executorId << "'";
return;
}
Executor* executor = framework->getExecutor(executorId);
+ if (executor == NULL) {
+ LOG(INFO) << "Executor '" << executorId
+ << "' of framework " << frameworkId
+ << " seems to have exited. Ignoring its shutdown timeout";
+ return;
+ }
+
// Make sure this timeout is valid.
- if (executor != NULL && executor->uuid == uuid) {
- LOG(INFO) << "Killing executor '" << executor->id
- << "' of framework " << framework->id;
+ if (executor->uuid != uuid ) {
+ LOG(INFO) << "A new executor '" << executorId
+ << "' of framework " << frameworkId
+ << " with run " << executor->uuid
+ << " seems to be active. Ignoring the shutdown timeout"
+ << " for the old executor run " << uuid;
+ return;
+ }
+
+ LOG(INFO) << "Killing executor '" << executor->id
+ << "' of framework " << framework->id;
+
+ dispatch(isolator, &Isolator::killExecutor, framework->id, executor->id);
+}
+
+
+void Slave::registerExecutorTimeout(
+ const FrameworkID& frameworkId,
+ const ExecutorID& executorId,
+ const UUID& uuid)
+{
+ Framework* framework = getFramework(frameworkId);
+ if (framework == NULL) {
+ LOG(INFO) << "Framework " << frameworkId
+ << " seems to have exited. Ignoring shutdown timeout"
+ << " for executor '" << executorId << "'";
+ return;
+ }
+
+ Executor* executor = framework->getExecutor(executorId);
+ if (executor == NULL) {
+ LOG(INFO) << "Executor '" << executorId
+ << "' of framework " << frameworkId
+ << " seems to have exited. Ignoring its shutdown timeout";
+ return;
+ }
+
+ // Make sure this timeout is valid.
+ if (executor->uuid != uuid ) {
+ LOG(INFO) << "A new executor '" << executorId
+ << "' of framework " << frameworkId
+ << " with run " << executor->uuid
+ << " seems to be active. Ignoring the shutdown timeout"
+ << " for the old executor run " << uuid;
+ return;
+ }
+
+ // Shutdown the executor if it has not registered yet.
+ if (!executor->pid) {
+ LOG(INFO) << "Shutting down executor " << executor->id
+ << " of framework " << framework->id
+ << " because it did not register within "
+ << flags.executor_registration_timeout;
+
+ executor->shutdown = true;
+ // Immediately kill the executor.
dispatch(isolator, &Isolator::killExecutor, framework->id, executor->id);
}
}
Modified: incubator/mesos/trunk/src/slave/slave.hpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/slave/slave.hpp?rev=1459093&r1=1459092&r2=1459093&view=diff
==============================================================================
--- incubator/mesos/trunk/src/slave/slave.hpp (original)
+++ incubator/mesos/trunk/src/slave/slave.hpp Wed Mar 20 22:53:23 2013
@@ -200,6 +200,12 @@ protected:
const ExecutorID& executorId,
const UUID& uuid);
+ // Shuts down the executor if it did not register yet.
+ void registerExecutorTimeout(
+ const FrameworkID& frameworkId,
+ const ExecutorID& executorId,
+ const UUID& uuid);
+
// Cleans up all un-reregistered executors during recovery.
void reregisterExecutorTimeout();
Modified: incubator/mesos/trunk/src/tests/master_tests.cpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/tests/master_tests.cpp?rev=1459093&r1=1459092&r2=1459093&view=diff
==============================================================================
--- incubator/mesos/trunk/src/tests/master_tests.cpp (original)
+++ incubator/mesos/trunk/src/tests/master_tests.cpp Wed Mar 20 22:53:23 2013
@@ -45,6 +45,8 @@
#include <process/dispatch.hpp>
#include <process/future.hpp>
+#include "slave/constants.hpp"
+#include "slave/process_isolator.hpp"
#include "slave/slave.hpp"
#include "tests/filter.hpp"
@@ -60,6 +62,7 @@ using mesos::internal::master::Framework
using mesos::internal::master::HierarchicalDRFAllocatorProcess;
using mesos::internal::master::Master;
+using mesos::internal::slave::ProcessIsolator;
using mesos::internal::slave::Slave;
using process::Clock;
@@ -847,6 +850,102 @@ TEST_F(MasterTest, MultipleExecutors)
}
+TEST_F(MasterTest, ShutdownUnregisteredExecutor)
+{
+ ASSERT_TRUE(GTEST_IS_THREADSAFE);
+
+ HierarchicalDRFAllocatorProcess allocator;
+ Allocator a(&allocator);
+ Files files;
+ Master m(&a, &files);
+ PID<Master> master = process::spawn(&m);
+
+ // Drop the registration message from the executor to the slave.
+ trigger registerExecutorMsg;
+ EXPECT_MESSAGE(Eq(RegisterExecutorMessage().GetTypeName()), _, _)
+ .WillOnce(DoAll(Trigger(®isterExecutorMsg),
+ Return(true)));
+
+ ProcessIsolator isolator;
+
+ Slave s(slaveFlags, true, &isolator, &files);
+ PID<Slave> slave = process::spawn(&s);
+
+ BasicMasterDetector detector(master, slave, true);
+
+ MockScheduler sched;
+ MesosSchedulerDriver driver(&sched, DEFAULT_FRAMEWORK_INFO, master);
+
+ vector<Offer> offers;
+ TaskStatus status;
+
+ EXPECT_CALL(sched, registered(&driver, _, _))
+ .Times(1);
+
+ trigger resourceOffersCall;
+ EXPECT_CALL(sched, resourceOffers(&driver, _))
+ .WillOnce(DoAll(SaveArg<1>(&offers),
+ Trigger(&resourceOffersCall)))
+ .WillRepeatedly(Return());
+
+ trigger statusUpdateCall;
+ EXPECT_CALL(sched, statusUpdate(&driver, _))
+ .WillOnce(DoAll(SaveArg<1>(&status),
+ Trigger(&statusUpdateCall)));
+
+ driver.start();
+
+ WAIT_UNTIL(resourceOffersCall);
+
+ EXPECT_NE(0u, offers.size());
+
+ // Launch a task with the command executor.
+ TaskInfo task;
+ task.set_name("");
+ task.mutable_task_id()->set_value("1");
+ task.mutable_slave_id()->MergeFrom(offers[0].slave_id());
+ task.mutable_resources()->MergeFrom(offers[0].resources());
+
+ CommandInfo command;
+ command.set_value("sleep 10");
+
+ task.mutable_command()->MergeFrom(command);
+
+ vector<TaskInfo> tasks;
+ tasks.push_back(task);
+
+ driver.launchTasks(offers[0].id(), tasks);
+
+ WAIT_UNTIL(registerExecutorMsg);
+
+ Clock::pause();
+
+ // Ensure that the slave times out and kills the executor.
+ Clock::advance(slaveFlags.executor_registration_timeout.secs());
+ Clock::settle();
+
+ // Ensure that the reaper reaps the executor.
+ Clock::advance(1.0);
+ Clock::settle();
+
+ WAIT_UNTIL(statusUpdateCall);
+
+ // This signals that the command executor has exited.
+ ASSERT_EQ(TASK_FAILED, status.state());
+
+ Clock::resume();
+
+ driver.stop();
+ driver.join();
+
+ process::terminate(slave);
+ process::wait(slave);
+
+ process::terminate(master);
+ process::wait(master);
+}
+
+
TEST_F(MasterTest, MasterInfo)
{
ASSERT_TRUE(GTEST_IS_THREADSAFE);
Modified: incubator/mesos/trunk/third_party/libprocess/include/process/delay.hpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/third_party/libprocess/include/process/delay.hpp?rev=1459093&r1=1459092&r2=1459093&view=diff
==============================================================================
--- incubator/mesos/trunk/third_party/libprocess/include/process/delay.hpp (original)
+++ incubator/mesos/trunk/third_party/libprocess/include/process/delay.hpp Wed Mar 20 22:53:23 2013
@@ -195,6 +195,55 @@ Timer delay(const Duration& duration,
return delay(duration, process->self(), method, a1, a2, a3);
}
+
+template <typename T,
+ typename P1, typename P2, typename P3, typename P4,
+ typename A1, typename A2, typename A3, typename A4>
+Timer delay(const Duration& duration,
+ const PID<T>& pid,
+ void (T::*method)(P1, P2, P3, P4),
+ A1 a1, A2 a2, A3 a3, A4 a4)
+{
+ std::tr1::shared_ptr<std::tr1::function<void(T*)> > thunk(
+ new std::tr1::function<void(T*)>(
+ std::tr1::bind(method, std::tr1::placeholders::_1, a1, a2, a3, a4)));
+
+ std::tr1::shared_ptr<std::tr1::function<void(ProcessBase*)> > dispatcher(
+ new std::tr1::function<void(ProcessBase*)>(
+ std::tr1::bind(&internal::vdispatcher<T>,
+ std::tr1::placeholders::_1,
+ thunk)));
+
+ std::tr1::function<void(void)> dispatch =
+ std::tr1::bind(internal::dispatch, pid, dispatcher);
+
+ return Timer::create(duration, dispatch);
+}
+
+
+template <typename T,
+ typename P1, typename P2, typename P3, typename P4,
+ typename A1, typename A2, typename A3, typename A4>
+Timer delay(const Duration& duration,
+ const Process<T>& process,
+ void (T::*method)(P1, P2, P3, P4),
+ A1 a1, A2 a2, A3 a3, A4 a4)
+{
+ return delay(duration, process.self(), method, a1, a2, a3, a4);
+}
+
+
+template <typename T,
+ typename P1, typename P2, typename P3, typename P4,
+ typename A1, typename A2, typename A3, typename A4>
+Timer delay(const Duration& duration,
+ const Process<T>* process,
+ void (T::*method)(P1, P2, P3, P4),
+ A1 a1, A2 a2, A3 a3, A4 a4)
+{
+ return delay(duration, process->self(), method, a1, a2, a3, a4);
+}
+
} // namespace process {
#endif // __PROCESS_DELAY_HPP__