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 2017/05/27 02:24:15 UTC
[5/9] mesos git commit: Introduced executor reconnect retries on the
agent.
Introduced executor reconnect retries on the agent.
PID-based v0 executors using Mesos libraries >= 1.1.2 always re-link
with the agent upon receiving the reconnect message. This avoids the
executor replying on a half-open TCP connection to the old agent
(possible if netfilter is dropping packets, see: MESOS-7057).
However, PID-based executors using Mesos libraries < 1.1.2 do not
re-link and are therefore prone to replying on a half-open connection
after the agent restarts. If we only send a single reconnect message,
these "old" executors will reply on their half-open connection,
receive a RST, and think the agent just died. To ensure these "old"
executors can reconnect in the presence of netfilter dropping packets,
we introduced optional retries of the reconnect message. This results
in "old" executors correctly establishing a link when processing the
second reconnect message.
Generally, users should not enable this flag unless they are affected
by this issue.
Review: https://reviews.apache.org/r/59584/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/d5e4ea14
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/d5e4ea14
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/d5e4ea14
Branch: refs/heads/1.1.x
Commit: d5e4ea1401cf72394cf0bbfb9950e93db821a842
Parents: 9567e6b
Author: Benjamin Mahler <bm...@apache.org>
Authored: Fri May 26 10:17:58 2017 -0400
Committer: Benjamin Mahler <bm...@apache.org>
Committed: Fri May 26 19:18:17 2017 -0700
----------------------------------------------------------------------
src/slave/flags.cpp | 22 +++++++++++
src/slave/flags.hpp | 1 +
src/slave/slave.cpp | 101 ++++++++++++++++++++++++++++++++++++++++++++---
src/slave/slave.hpp | 7 ++++
4 files changed, 126 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/d5e4ea14/src/slave/flags.cpp
----------------------------------------------------------------------
diff --git a/src/slave/flags.cpp b/src/slave/flags.cpp
index a23cca7..675b8ed 100644
--- a/src/slave/flags.cpp
+++ b/src/slave/flags.cpp
@@ -321,6 +321,28 @@ mesos::internal::slave::Flags::Flags()
return None();
});
+ // TODO(bmahler): Remove this once v0 executors are no longer supported.
+ add(&Flags::executor_reregistration_retry_interval,
+ "executor_reregistration_retry_interval",
+ "For PID-based executors, how long the agent waits before retrying\n"
+ "the reconnect message sent to the executor during recovery.\n"
+ "NOTE: Do not use this unless you understand the following\n"
+ "(see MESOS-5332): PID-based executors using Mesos libraries >= 1.1.2\n"
+ "always re-link with the agent upon receiving the reconnect message.\n"
+ "This avoids the executor replying on a half-open TCP connection to\n"
+ "the old agent (possible if netfilter is dropping packets,\n"
+ "see: MESOS-7057). However, PID-based executors using Mesos\n"
+ "libraries < 1.1.2 do not re-link and are therefore prone to\n"
+ "replying on a half-open connection after the agent restarts. If we\n"
+ "only send a single reconnect message, these \"old\" executors will\n"
+ "reply on their half-open connection and receive a RST; without any\n"
+ "retries, they will fail to reconnect and be killed by the agent once\n"
+ "the executor re-registration timeout elapses. To ensure these \"old\"\n"
+ "executors can reconnect in the presence of netfilter dropping\n"
+ "packets, we introduced optional retries of the reconnect message.\n"
+ "This results in \"old\" executors correctly establishing a link\n"
+ "when processing the second reconnect message.");
+
add(&Flags::executor_shutdown_grace_period,
"executor_shutdown_grace_period",
"Default amount of time to wait for an executor to shut down\n"
http://git-wip-us.apache.org/repos/asf/mesos/blob/d5e4ea14/src/slave/flags.hpp
----------------------------------------------------------------------
diff --git a/src/slave/flags.hpp b/src/slave/flags.hpp
index c5eb0aa..65b534d 100644
--- a/src/slave/flags.hpp
+++ b/src/slave/flags.hpp
@@ -76,6 +76,7 @@ public:
Option<JSON::Object> executor_environment_variables;
Duration executor_registration_timeout;
Duration executor_reregistration_timeout;
+ Option<Duration> executor_reregistration_retry_interval;
Duration executor_shutdown_grace_period;
Duration gc_delay;
double gc_disk_headroom;
http://git-wip-us.apache.org/repos/asf/mesos/blob/d5e4ea14/src/slave/slave.cpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.cpp b/src/slave/slave.cpp
index 8023843..90ec503 100644
--- a/src/slave/slave.cpp
+++ b/src/slave/slave.cpp
@@ -3410,10 +3410,11 @@ void Slave::reregisterExecutor(
state == RUNNING || state == TERMINATING)
<< state;
- if (state != RECOVERING) {
- LOG(WARNING) << "Shutting down executor '" << executorId
- << "' of framework " << frameworkId
- << " because the agent is not in recovery mode";
+ if (state == TERMINATING) {
+ LOG(WARNING) << "Shutting down executor '" << executorId << "'"
+ << " of framework " << frameworkId
+ << " because the agent is terminating";
+
reply(ShutdownExecutorMessage());
return;
}
@@ -3447,11 +3448,37 @@ void Slave::reregisterExecutor(
case Executor::TERMINATED:
// TERMINATED is possible if the executor forks, the parent process
// terminates and the child process (driver) tries to register!
- case Executor::RUNNING:
LOG(WARNING) << "Shutting down executor " << *executor
<< " because it is in unexpected state " << executor->state;
reply(ShutdownExecutorMessage());
break;
+
+ case Executor::RUNNING:
+ if (flags.executor_reregistration_retry_interval.isNone()) {
+ // Previously, when an executor sends a re-registration while
+ // in the RUNNING state, we would shut the executor down. We
+ // preserve that behavior when the optional reconnect retry
+ // is not enabled.
+ LOG(WARNING) << "Shutting down executor " << *executor
+ << " because it is in unexpected state "
+ << executor->state;
+ reply(ShutdownExecutorMessage());
+ } else {
+ // When the agent is configured to retry the reconnect requests
+ // to executors, we ignore any further re-registrations. This
+ // is because we can't easily handle re-registering libprocess
+ // based executors in the steady state, and we plan to move to
+ // only allowing v1 HTTP executors (where re-subscription in
+ // the steady state is supported). Also, ignoring this message
+ // ensures that any executors mimicking the libprocess protocol
+ // do not have any illusion of being able to re-register without
+ // an agent restart (hopefully they will commit suicide if they
+ // fail to re-register).
+ LOG(WARNING) << "Ignoring executor re-registration message from "
+ << *executor << " because it is already registered";
+ }
+ break;
+
case Executor::REGISTERING: {
executor->state = Executor::RUNNING;
@@ -5196,6 +5223,45 @@ Future<Nothing> Slave::_recoverContainerizer(
}
+void Slave::retryReconnectExecutor(
+ const FrameworkID& frameworkId,
+ const ExecutorID& executorId,
+ const ReconnectExecutorMessage& message)
+{
+ CHECK_SOME(flags.executor_reregistration_retry_interval);
+
+ if (state != RECOVERING) {
+ return;
+ }
+
+ Framework* framework = getFramework(frameworkId);
+ if (framework == nullptr) {
+ return;
+ }
+
+ Executor* executor = framework->getExecutor(executorId);
+ if (executor == nullptr) {
+ return;
+ }
+
+ if (executor->state != Executor::REGISTERING) {
+ return;
+ }
+
+ LOG(INFO) << "Re-sending reconnect request to executor "
+ << *executor;
+
+ send(executor->pid.get(), message);
+
+ delay(flags.executor_reregistration_retry_interval.get(),
+ self(),
+ &Slave::retryReconnectExecutor,
+ frameworkId,
+ executorId,
+ message);
+}
+
+
Future<Nothing> Slave::_recover()
{
// Alow HTTP based executors to subscribe after the
@@ -5226,6 +5292,31 @@ Future<Nothing> Slave::_recover()
ReconnectExecutorMessage message;
message.mutable_slave_id()->MergeFrom(info.id());
send(executor->pid.get(), message);
+
+ // PID-based executors using Mesos libraries >= 1.1.2 always
+ // re-link with the agent upon receiving the reconnect message.
+ // This avoids the executor replying on a half-open TCP
+ // connection to the old agent (possible if netfilter is
+ // dropping packets, see: MESOS-7057). However, PID-based
+ // executors using Mesos libraries < 1.1.2 do not re-link
+ // and are therefore prone to replying on a half-open connection
+ // after the agent restarts. If we only send a single reconnect
+ // message, these "old" executors will reply on their half-open
+ // connection and receive a RST; without any retries, they will
+ // fail to reconnect and be killed by the agent once the executor
+ // re-registration timeout elapses. To ensure these "old"
+ // executors can reconnect in the presence of netfilter dropping
+ // packets, we introduced optional retries of the reconnect
+ // message. This results in "old" executors correctly establishing
+ // a link when processing the second reconnect message.
+ if (flags.executor_reregistration_retry_interval.isSome()) {
+ delay(flags.executor_reregistration_retry_interval.get(),
+ self(),
+ &Slave::retryReconnectExecutor,
+ framework->id(),
+ executor->id,
+ message);
+ }
} else if (executor->pid.isNone()) {
LOG(INFO) << "Waiting for executor " << *executor
<< " to subscribe";
http://git-wip-us.apache.org/repos/asf/mesos/blob/d5e4ea14/src/slave/slave.hpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.hpp b/src/slave/slave.hpp
index d8c5873..7490e33 100644
--- a/src/slave/slave.hpp
+++ b/src/slave/slave.hpp
@@ -402,6 +402,13 @@ public:
process::Future<Nothing> _recoverContainerizer(
const Option<state::SlaveState>& state);
+ // Sends a `ReconnectExecutorMessage` to the specified executor,
+ // retrying if necessary.
+ void retryReconnectExecutor(
+ const FrameworkID& frameworkId,
+ const ExecutorID& executorId,
+ const ReconnectExecutorMessage& message);
+
// This is called when recovery finishes.
// Made 'virtual' for Slave mocking.
virtual void __recover(const process::Future<Nothing>& future);