You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by br...@apache.org on 2018/10/12 12:28:27 UTC
[10/50] [abbrv] hadoop git commit: YARN-7644. NM gets backed up
deleting docker containers. Contributed by Chandni Singh
YARN-7644. NM gets backed up deleting docker containers. Contributed by Chandni Singh
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/5ce70e12
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/5ce70e12
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/5ce70e12
Branch: refs/heads/HDFS-13891
Commit: 5ce70e1211e624d58e8bb1181aec00729ebdc085
Parents: cd28051
Author: Jason Lowe <jl...@apache.org>
Authored: Wed Oct 10 09:52:19 2018 -0500
Committer: Jason Lowe <jl...@apache.org>
Committed: Wed Oct 10 09:52:19 2018 -0500
----------------------------------------------------------------------
.../launcher/ContainerCleanup.java | 229 +++++++++++++++++++
.../launcher/ContainerLaunch.java | 226 +++++-------------
.../launcher/ContainersLauncher.java | 14 +-
.../launcher/TestContainerCleanup.java | 108 +++++++++
.../launcher/TestContainersLauncher.java | 12 +-
5 files changed, 401 insertions(+), 188 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5ce70e12/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/ContainerCleanup.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/ContainerCleanup.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/ContainerCleanup.java
new file mode 100644
index 0000000..963d28b
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/ContainerCleanup.java
@@ -0,0 +1,229 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.yarn.server.nodemanager.containermanager.launcher;
+
+import com.google.common.base.Preconditions;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileContext;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.util.Shell;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.event.Dispatcher;
+import org.apache.hadoop.yarn.server.nodemanager.ContainerExecutor;
+import org.apache.hadoop.yarn.server.nodemanager.Context;
+import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.Container;
+import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerDiagnosticsUpdateEvent;
+import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerEventType;
+import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerExitEvent;
+import org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.runtime.DockerLinuxContainerRuntime;
+import org.apache.hadoop.yarn.server.nodemanager.executor.ContainerReapContext;
+import org.apache.hadoop.yarn.server.nodemanager.executor.ContainerSignalContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+import static org.apache.hadoop.yarn.server.nodemanager.containermanager.launcher.ContainerLaunch.EXIT_CODE_FILE_SUFFIX;
+
+/**
+ * Cleanup the container.
+ * Cancels the launch if launch has not started yet or signals
+ * the executor to not execute the process if not already done so.
+ * Also, sends a SIGTERM followed by a SIGKILL to the process if
+ * the process id is available.
+ */
+public class ContainerCleanup implements Runnable {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(ContainerCleanup.class);
+
+ private final Context context;
+ private final Configuration conf;
+ private final Dispatcher dispatcher;
+ private final ContainerExecutor exec;
+ private final Container container;
+ private final ContainerLaunch launch;
+ private final long sleepDelayBeforeSigKill;
+
+
+ public ContainerCleanup(Context context, Configuration configuration,
+ Dispatcher dispatcher, ContainerExecutor exec,
+ Container container,
+ ContainerLaunch containerLaunch) {
+
+ this.context = Preconditions.checkNotNull(context, "context");
+ this.conf = Preconditions.checkNotNull(configuration, "config");
+ this.dispatcher = Preconditions.checkNotNull(dispatcher, "dispatcher");
+ this.exec = Preconditions.checkNotNull(exec, "exec");
+ this.container = Preconditions.checkNotNull(container, "container");
+ this.launch = Preconditions.checkNotNull(containerLaunch, "launch");
+ this.sleepDelayBeforeSigKill = conf.getLong(
+ YarnConfiguration.NM_SLEEP_DELAY_BEFORE_SIGKILL_MS,
+ YarnConfiguration.DEFAULT_NM_SLEEP_DELAY_BEFORE_SIGKILL_MS);
+ }
+
+ @Override
+ public void run() {
+ ContainerId containerId = container.getContainerId();
+ String containerIdStr = containerId.toString();
+ LOG.info("Cleaning up container " + containerIdStr);
+
+ try {
+ context.getNMStateStore().storeContainerKilled(containerId);
+ } catch (IOException e) {
+ LOG.error("Unable to mark container " + containerId
+ + " killed in store", e);
+ }
+
+ // launch flag will be set to true if process already launched
+ boolean alreadyLaunched = !launch.markLaunched();
+ if (!alreadyLaunched) {
+ LOG.info("Container " + containerIdStr + " not launched."
+ + " No cleanup needed to be done");
+ return;
+ }
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Marking container " + containerIdStr + " as inactive");
+ }
+ // this should ensure that if the container process has not launched
+ // by this time, it will never be launched
+ exec.deactivateContainer(containerId);
+ Path pidFilePath = launch.getPidFilePath();
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Getting pid for container {} to kill"
+ + " from pid file {}", containerIdStr, pidFilePath != null ?
+ pidFilePath : "null");
+ }
+
+ // however the container process may have already started
+ try {
+
+ // get process id from pid file if available
+ // else if shell is still active, get it from the shell
+ String processId = launch.getContainerPid();
+
+ // kill process
+ String user = container.getUser();
+ if (processId != null) {
+ signalProcess(processId, user, containerIdStr);
+ } else {
+ // Normally this means that the process was notified about
+ // deactivateContainer above and did not start.
+ // Since we already set the state to RUNNING or REINITIALIZING
+ // we have to send a killed event to continue.
+ if (!launch.isLaunchCompleted()) {
+ LOG.warn("Container clean up before pid file created "
+ + containerIdStr);
+ dispatcher.getEventHandler().handle(
+ new ContainerExitEvent(container.getContainerId(),
+ ContainerEventType.CONTAINER_KILLED_ON_REQUEST,
+ Shell.WINDOWS ?
+ ContainerExecutor.ExitCode.FORCE_KILLED.getExitCode() :
+ ContainerExecutor.ExitCode.TERMINATED.getExitCode(),
+ "Container terminated before pid file created."));
+ // There is a possibility that the launch grabbed the file name before
+ // the deactivateContainer above but it was slow enough to avoid
+ // getContainerPid.
+ // Increasing YarnConfiguration.NM_PROCESS_KILL_WAIT_MS
+ // reduces the likelihood of this race condition and process leak.
+ }
+ // The Docker container may not have fully started, reap the container.
+ if (DockerLinuxContainerRuntime.isDockerContainerRequested(conf,
+ container.getLaunchContext().getEnvironment())) {
+ reapDockerContainerNoPid(user);
+ }
+ }
+ } catch (Exception e) {
+ String message =
+ "Exception when trying to cleanup container " + containerIdStr
+ + ": " + StringUtils.stringifyException(e);
+ LOG.warn(message);
+ dispatcher.getEventHandler().handle(
+ new ContainerDiagnosticsUpdateEvent(containerId, message));
+ } finally {
+ // cleanup pid file if present
+ if (pidFilePath != null) {
+ try {
+ FileContext lfs = FileContext.getLocalFSFileContext();
+ lfs.delete(pidFilePath, false);
+ lfs.delete(pidFilePath.suffix(EXIT_CODE_FILE_SUFFIX), false);
+ } catch (IOException ioe) {
+ LOG.warn("{} exception trying to delete pid file {}. Ignoring.",
+ containerId, pidFilePath, ioe);
+ }
+ }
+ }
+
+ try {
+ // Reap the container
+ launch.reapContainer();
+ } catch (IOException ioe) {
+ LOG.warn("{} exception trying to reap container. Ignoring.", containerId,
+ ioe);
+ }
+ }
+
+ private void signalProcess(String processId, String user,
+ String containerIdStr) throws IOException {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Sending signal to pid " + processId + " as user " + user
+ + " for container " + containerIdStr);
+ }
+ final ContainerExecutor.Signal signal =
+ sleepDelayBeforeSigKill > 0 ? ContainerExecutor.Signal.TERM :
+ ContainerExecutor.Signal.KILL;
+
+ boolean result = sendSignal(user, processId, signal);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Sent signal " + signal + " to pid " + processId + " as user "
+ + user + " for container " + containerIdStr + ", result="
+ + (result ? "success" : "failed"));
+ }
+ if (sleepDelayBeforeSigKill > 0) {
+ new ContainerExecutor.DelayedProcessKiller(container, user, processId,
+ sleepDelayBeforeSigKill, ContainerExecutor.Signal.KILL, exec).start();
+ }
+ }
+
+ private boolean sendSignal(String user, String processId,
+ ContainerExecutor.Signal signal)
+ throws IOException {
+ return exec.signalContainer(
+ new ContainerSignalContext.Builder().setContainer(container)
+ .setUser(user).setPid(processId).setSignal(signal).build());
+ }
+
+ private void reapDockerContainerNoPid(String user) throws IOException {
+ String containerIdStr =
+ container.getContainerTokenIdentifier().getContainerID().toString();
+ LOG.info("Unable to obtain pid, but docker container request detected. "
+ + "Attempting to reap container " + containerIdStr);
+ boolean result = exec.reapContainer(
+ new ContainerReapContext.Builder()
+ .setContainer(container)
+ .setUser(container.getUser())
+ .build());
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Sent signal to docker container " + containerIdStr
+ + " as user " + user + ", result=" + (result ? "success" : "failed"));
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5ce70e12/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/ContainerLaunch.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/ContainerLaunch.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/ContainerLaunch.java
index 3fa73ec..6776836 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/ContainerLaunch.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/ContainerLaunch.java
@@ -70,7 +70,6 @@ import org.apache.hadoop.yarn.exceptions.ConfigurationException;
import org.apache.hadoop.yarn.exceptions.YarnException;
import org.apache.hadoop.yarn.ipc.RPCUtil;
import org.apache.hadoop.yarn.server.nodemanager.ContainerExecutor;
-import org.apache.hadoop.yarn.server.nodemanager.ContainerExecutor.DelayedProcessKiller;
import org.apache.hadoop.yarn.server.nodemanager.ContainerExecutor.ExitCode;
import org.apache.hadoop.yarn.server.nodemanager.ContainerExecutor.Signal;
import org.apache.hadoop.yarn.server.nodemanager.Context;
@@ -85,7 +84,6 @@ import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.Cont
import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerExitEvent;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerKillEvent;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerState;
-import org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.runtime.DockerLinuxContainerRuntime;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.localizer.ContainerLocalizer;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.localizer.ResourceLocalizationService;
import org.apache.hadoop.yarn.server.nodemanager.executor.ContainerPrepareContext;
@@ -115,7 +113,7 @@ public class ContainerLaunch implements Callable<Integer> {
public static final String FINAL_CONTAINER_TOKENS_FILE = "container_tokens";
private static final String PID_FILE_NAME_FMT = "%s.pid";
- private static final String EXIT_CODE_FILE_SUFFIX = ".exitcode";
+ static final String EXIT_CODE_FILE_SUFFIX = ".exitcode";
protected final Dispatcher dispatcher;
protected final ContainerExecutor exec;
@@ -131,7 +129,6 @@ public class ContainerLaunch implements Callable<Integer> {
protected AtomicBoolean completed = new AtomicBoolean(false);
private volatile boolean killedBeforeStart = false;
- private long sleepDelayBeforeSigKill = 250;
private long maxKillWaitTime = 2000;
protected Path pidFilePath = null;
@@ -152,9 +149,6 @@ public class ContainerLaunch implements Callable<Integer> {
this.dispatcher = dispatcher;
this.dirsHandler = dirsHandler;
this.containerManager = containerManager;
- this.sleepDelayBeforeSigKill =
- conf.getLong(YarnConfiguration.NM_SLEEP_DELAY_BEFORE_SIGKILL_MS,
- YarnConfiguration.DEFAULT_NM_SLEEP_DELAY_BEFORE_SIGKILL_MS);
this.maxKillWaitTime =
conf.getLong(YarnConfiguration.NM_PROCESS_KILL_WAIT_MS,
YarnConfiguration.DEFAULT_NM_PROCESS_KILL_WAIT_MS);
@@ -515,6 +509,25 @@ public class ContainerLaunch implements Callable<Integer> {
return launchPrep;
}
+ void reapContainer() throws IOException {
+ containerExecLock.lock();
+ try {
+ // Reap the container
+ boolean result = exec.reapContainer(
+ new ContainerReapContext.Builder()
+ .setContainer(container)
+ .setUser(container.getUser())
+ .build());
+ if (!result) {
+ throw new IOException("Reap container failed for container " +
+ container.getContainerId());
+ }
+ cleanupContainerFiles(getContainerWorkDir());
+ } finally {
+ containerExecLock.unlock();
+ }
+ }
+
protected int prepareForLaunch(ContainerStartContext ctx) throws IOException {
ContainerId containerId = container.getContainerId();
if (container.isMarkedForKilling()) {
@@ -721,121 +734,6 @@ public class ContainerLaunch implements Callable<Integer> {
return getContainerPrivateDir(appIdStr, containerIdStr) + Path.SEPARATOR
+ String.format(ContainerLaunch.PID_FILE_NAME_FMT, containerIdStr);
}
-
- /**
- * Cleanup the container.
- * Cancels the launch if launch has not started yet or signals
- * the executor to not execute the process if not already done so.
- * Also, sends a SIGTERM followed by a SIGKILL to the process if
- * the process id is available.
- * @throws IOException
- */
- public void cleanupContainer() throws IOException {
- ContainerId containerId = container.getContainerId();
- String containerIdStr = containerId.toString();
- LOG.info("Cleaning up container " + containerIdStr);
-
- try {
- context.getNMStateStore().storeContainerKilled(containerId);
- } catch (IOException e) {
- LOG.error("Unable to mark container " + containerId
- + " killed in store", e);
- }
-
- // launch flag will be set to true if process already launched
- boolean alreadyLaunched =
- !containerAlreadyLaunched.compareAndSet(false, true);
- if (!alreadyLaunched) {
- LOG.info("Container " + containerIdStr + " not launched."
- + " No cleanup needed to be done");
- return;
- }
- if (LOG.isDebugEnabled()) {
- LOG.debug("Marking container " + containerIdStr + " as inactive");
- }
- // this should ensure that if the container process has not launched
- // by this time, it will never be launched
- exec.deactivateContainer(containerId);
-
- if (LOG.isDebugEnabled()) {
- LOG.debug("Getting pid for container " + containerIdStr + " to kill"
- + " from pid file "
- + (pidFilePath != null ? pidFilePath.toString() : "null"));
- }
-
- // however the container process may have already started
- try {
-
- // get process id from pid file if available
- // else if shell is still active, get it from the shell
- String processId = null;
- if (pidFilePath != null) {
- processId = getContainerPid(pidFilePath);
- }
-
- // kill process
- String user = container.getUser();
- if (processId != null) {
- signalProcess(processId, user, containerIdStr);
- } else {
- // Normally this means that the process was notified about
- // deactivateContainer above and did not start.
- // Since we already set the state to RUNNING or REINITIALIZING
- // we have to send a killed event to continue.
- if (!completed.get()) {
- LOG.warn("Container clean up before pid file created "
- + containerIdStr);
- dispatcher.getEventHandler().handle(
- new ContainerExitEvent(container.getContainerId(),
- ContainerEventType.CONTAINER_KILLED_ON_REQUEST,
- Shell.WINDOWS ? ExitCode.FORCE_KILLED.getExitCode() :
- ExitCode.TERMINATED.getExitCode(),
- "Container terminated before pid file created."));
- // There is a possibility that the launch grabbed the file name before
- // the deactivateContainer above but it was slow enough to avoid
- // getContainerPid.
- // Increasing YarnConfiguration.NM_PROCESS_KILL_WAIT_MS
- // reduces the likelihood of this race condition and process leak.
- }
- // The Docker container may not have fully started, reap the container.
- if (DockerLinuxContainerRuntime.isDockerContainerRequested(
- conf,
- container.getLaunchContext().getEnvironment())) {
- reapDockerContainerNoPid(user);
- }
- }
- } catch (Exception e) {
- String message =
- "Exception when trying to cleanup container " + containerIdStr
- + ": " + StringUtils.stringifyException(e);
- LOG.warn(message);
- dispatcher.getEventHandler().handle(
- new ContainerDiagnosticsUpdateEvent(containerId, message));
- } finally {
- // cleanup pid file if present
- if (pidFilePath != null) {
- FileContext lfs = FileContext.getLocalFSFileContext();
- lfs.delete(pidFilePath, false);
- lfs.delete(pidFilePath.suffix(EXIT_CODE_FILE_SUFFIX), false);
- }
- }
- containerExecLock.lock();
- try {
- // Reap the container
- boolean result = exec.reapContainer(
- new ContainerReapContext.Builder()
- .setContainer(container)
- .setUser(container.getUser())
- .build());
- if (!result) {
- throw new IOException("Reap container failed for container "
- + containerIdStr);
- }
- cleanupContainerFiles(getContainerWorkDir());
- } finally {
- containerExecLock.unlock();
- }
- }
/**
* Send a signal to the container.
@@ -874,11 +772,7 @@ public class ContainerLaunch implements Callable<Integer> {
try {
// get process id from pid file if available
// else if shell is still active, get it from the shell
- String processId = null;
- if (pidFilePath != null) {
- processId = getContainerPid(pidFilePath);
- }
-
+ String processId = getContainerPid();
if (processId != null) {
if (LOG.isDebugEnabled()) {
LOG.debug("Sending signal to pid " + processId
@@ -912,50 +806,6 @@ public class ContainerLaunch implements Callable<Integer> {
}
}
- private boolean sendSignal(String user, String processId, Signal signal)
- throws IOException {
- return exec.signalContainer(
- new ContainerSignalContext.Builder().setContainer(container)
- .setUser(user).setPid(processId).setSignal(signal).build());
- }
-
- private void signalProcess(String processId, String user,
- String containerIdStr) throws IOException {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Sending signal to pid " + processId + " as user " + user
- + " for container " + containerIdStr);
- }
- final Signal signal =
- sleepDelayBeforeSigKill > 0 ? Signal.TERM : Signal.KILL;
-
- boolean result = sendSignal(user, processId, signal);
- if (LOG.isDebugEnabled()) {
- LOG.debug("Sent signal " + signal + " to pid " + processId + " as user "
- + user + " for container " + containerIdStr + ", result="
- + (result ? "success" : "failed"));
- }
- if (sleepDelayBeforeSigKill > 0) {
- new DelayedProcessKiller(container, user, processId,
- sleepDelayBeforeSigKill, Signal.KILL, exec).start();
- }
- }
-
- private void reapDockerContainerNoPid(String user) throws IOException {
- String containerIdStr =
- container.getContainerTokenIdentifier().getContainerID().toString();
- LOG.info("Unable to obtain pid, but docker container request detected. "
- + "Attempting to reap container " + containerIdStr);
- boolean result = exec.reapContainer(
- new ContainerReapContext.Builder()
- .setContainer(container)
- .setUser(container.getUser())
- .build());
- if (LOG.isDebugEnabled()) {
- LOG.debug("Sent signal to docker container " + containerIdStr
- + " as user " + user + ", result=" + (result ? "success" : "failed"));
- }
- }
-
@VisibleForTesting
public static Signal translateCommandToSignal(
SignalContainerCommand command) {
@@ -1076,14 +926,16 @@ public class ContainerLaunch implements Callable<Integer> {
/**
* Loop through for a time-bounded interval waiting to
* read the process id from a file generated by a running process.
- * @param pidFilePath File from which to read the process id
- * @return Process ID
+ * @return Process ID; null when pidFilePath is null
* @throws Exception
*/
- private String getContainerPid(Path pidFilePath) throws Exception {
+ String getContainerPid() throws Exception {
+ if (pidFilePath == null) {
+ return null;
+ }
String containerIdStr =
container.getContainerId().toString();
- String processId = null;
+ String processId;
if (LOG.isDebugEnabled()) {
LOG.debug("Accessing pid for container " + containerIdStr
+ " from pid file " + pidFilePath);
@@ -1889,4 +1741,28 @@ public class ContainerLaunch implements Callable<Integer> {
LOG.warn("Failed to delete " + path, e);
}
}
+
+ /**
+ * Returns the PID File Path.
+ */
+ Path getPidFilePath() {
+ return pidFilePath;
+ }
+
+ /**
+ * Marks the container to be launched only if it was not launched.
+ *
+ * @return true if successful; false otherwise.
+ */
+ boolean markLaunched() {
+ return containerAlreadyLaunched.compareAndSet(false, true);
+ }
+
+ /**
+ * Returns if the launch is completed or not.
+ */
+ boolean isLaunchCompleted() {
+ return completed.get();
+ }
+
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5ce70e12/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/ContainersLauncher.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/ContainersLauncher.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/ContainersLauncher.java
index 7870f86..fdfe5b1 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/ContainersLauncher.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/ContainersLauncher.java
@@ -25,7 +25,6 @@ import java.util.Map;
import java.util.concurrent.ExecutorService;
import org.apache.hadoop.util.Shell;
-import org.apache.hadoop.yarn.api.records.ContainerExitStatus;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerEventType;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerExitEvent;
import org.slf4j.Logger;
@@ -154,8 +153,8 @@ public class ContainersLauncher extends AbstractService
break;
case CLEANUP_CONTAINER:
case CLEANUP_CONTAINER_FOR_REINIT:
- ContainerLaunch launcher = running.remove(containerId);
- if (launcher == null) {
+ ContainerLaunch existingLaunch = running.remove(containerId);
+ if (existingLaunch == null) {
// Container not launched.
// triggering KILLING to CONTAINER_CLEANEDUP_AFTER_KILL transition.
dispatcher.getEventHandler().handle(
@@ -169,12 +168,9 @@ public class ContainersLauncher extends AbstractService
// Cleanup a container whether it is running/killed/completed, so that
// no sub-processes are alive.
- try {
- launcher.cleanupContainer();
- } catch (IOException e) {
- LOG.warn("Got exception while cleaning container " + containerId
- + ". Ignoring.");
- }
+ ContainerCleanup cleanup = new ContainerCleanup(context, getConfig(),
+ dispatcher, exec, event.getContainer(), existingLaunch);
+ containerLauncher.submit(cleanup);
break;
case SIGNAL_CONTAINER:
SignalContainersLauncherEvent signalEvent =
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5ce70e12/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/TestContainerCleanup.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/TestContainerCleanup.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/TestContainerCleanup.java
new file mode 100644
index 0000000..6c99379
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/TestContainerCleanup.java
@@ -0,0 +1,108 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.yarn.server.nodemanager.containermanager.launcher;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.api.records.SignalContainerCommand;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.event.Dispatcher;
+import org.apache.hadoop.yarn.event.InlineDispatcher;
+import org.apache.hadoop.yarn.server.nodemanager.ContainerExecutor;
+import org.apache.hadoop.yarn.server.nodemanager.Context;
+import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.Container;
+import org.apache.hadoop.yarn.server.nodemanager.executor.ContainerSignalContext;
+import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.apache.hadoop.yarn.conf.YarnConfiguration.NM_SLEEP_DELAY_BEFORE_SIGKILL_MS;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for {@link ContainerCleanup}.
+ */
+public class TestContainerCleanup {
+
+ private YarnConfiguration conf;
+ private ContainerId containerId;
+ private ContainerExecutor executor;
+ private ContainerLaunch launch;
+ private ContainerCleanup cleanup;
+
+ @Before
+ public void setup() throws Exception {
+ conf = new YarnConfiguration();
+ conf.setLong(NM_SLEEP_DELAY_BEFORE_SIGKILL_MS, 60000);
+ Context context = mock(Context.class);
+ NMStateStoreService storeService = mock(NMStateStoreService.class);
+ when(context.getNMStateStore()).thenReturn(storeService);
+
+ Dispatcher dispatcher = new InlineDispatcher();
+ executor = mock(ContainerExecutor.class);
+ when(executor.signalContainer(Mockito.any(
+ ContainerSignalContext.class))).thenReturn(true);
+
+ ApplicationId appId = ApplicationId.newInstance(System.currentTimeMillis(),
+ 1);
+ ApplicationAttemptId attemptId = ApplicationAttemptId.newInstance(appId, 1);
+ containerId = ContainerId.newContainerId(attemptId, 1);
+ Container container = mock(Container.class);
+
+ when(container.getContainerId()).thenReturn(containerId);
+
+ launch = mock(ContainerLaunch.class);
+ launch.containerAlreadyLaunched = new AtomicBoolean(false);
+
+ launch.pidFilePath = new Path("target/" + containerId.toString() + ".pid");
+ when(launch.getContainerPid()).thenReturn(containerId.toString());
+
+ cleanup = new ContainerCleanup(context, conf, dispatcher, executor,
+ container, launch);
+ }
+
+ @Test
+ public void testNoCleanupWhenContainerNotLaunched() throws IOException {
+ cleanup.run();
+ verify(launch, Mockito.times(0)).signalContainer(
+ Mockito.any(SignalContainerCommand.class));
+ }
+
+ @Test
+ public void testCleanup() throws Exception {
+ launch.containerAlreadyLaunched.set(true);
+ cleanup.run();
+ ArgumentCaptor<ContainerSignalContext> captor =
+ ArgumentCaptor.forClass(ContainerSignalContext.class);
+
+ verify(executor, Mockito.times(1)).signalContainer(captor.capture());
+ Assert.assertEquals("signal", ContainerExecutor.Signal.TERM,
+ captor.getValue().getSignal());
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5ce70e12/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/TestContainersLauncher.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/TestContainersLauncher.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/TestContainersLauncher.java
index 0234ac2..f2fafd2 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/TestContainersLauncher.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/TestContainersLauncher.java
@@ -192,12 +192,14 @@ public class TestContainersLauncher {
.synchronizedMap(new HashMap<ContainerId, ContainerLaunch>());
dummyMap.put(containerId, containerLaunch);
Whitebox.setInternalState(spy, "running", dummyMap);
+
when(event.getType())
.thenReturn(ContainersLauncherEventType.CLEANUP_CONTAINER);
- doNothing().when(containerLaunch).cleanupContainer();
+ assertEquals(1, dummyMap.size());
spy.handle(event);
assertEquals(0, dummyMap.size());
- Mockito.verify(containerLaunch, Mockito.times(1)).cleanupContainer();
+ Mockito.verify(containerLauncher, Mockito.times(1))
+ .submit(Mockito.any(ContainerCleanup.class));
}
@Test
@@ -207,12 +209,14 @@ public class TestContainersLauncher {
.synchronizedMap(new HashMap<ContainerId, ContainerLaunch>());
dummyMap.put(containerId, containerLaunch);
Whitebox.setInternalState(spy, "running", dummyMap);
+
when(event.getType())
.thenReturn(ContainersLauncherEventType.CLEANUP_CONTAINER_FOR_REINIT);
- doNothing().when(containerLaunch).cleanupContainer();
+ assertEquals(1, dummyMap.size());
spy.handle(event);
assertEquals(0, dummyMap.size());
- Mockito.verify(containerLaunch, Mockito.times(1)).cleanupContainer();
+ Mockito.verify(containerLauncher, Mockito.times(1))
+ .submit(Mockito.any(ContainerCleanup.class));
}
@Test
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org