You are viewing a plain text version of this content. The canonical link for it is here.
Posted to yarn-commits@hadoop.apache.org by ji...@apache.org on 2014/08/12 21:25:59 UTC
svn commit: r1617566 [1/2] - in
/hadoop/common/branches/HDFS-6584/hadoop-yarn-project: ./
hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/webapp/util/
hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/webapp/util/
...
Author: jing9
Date: Tue Aug 12 19:25:57 2014
New Revision: 1617566
URL: http://svn.apache.org/r1617566
Log:
Merging r1617377 through r1617565 from trunk.
Added:
hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/webapp/util/
- copied from r1617565, hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/webapp/util/
hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/RecoveredContainerLaunch.java
- copied unchanged from r1617565, hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/RecoveredContainerLaunch.java
Modified:
hadoop/common/branches/HDFS-6584/hadoop-yarn-project/CHANGES.txt
hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/webapp/util/WebAppUtils.java
hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/ContainerExecutor.java
hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/DefaultContainerExecutor.java
hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/LinuxContainerExecutor.java
hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeStatusUpdater.java
hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeStatusUpdaterImpl.java
hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/ContainerManagerImpl.java
hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/container/ContainerImpl.java
hadoop/common/branches/HDFS-6584/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
hadoop/common/branches/HDFS-6584/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
hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/ContainersLauncherEventType.java
hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/AppLogAggregator.java
hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/AppLogAggregatorImpl.java
hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/LogAggregationService.java
hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/recovery/NMLeveldbStateStoreService.java
hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/recovery/NMNullStateStoreService.java
hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/recovery/NMStateStoreService.java
hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/container-executor.c
hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/TestNodeStatusUpdater.java
hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/BaseContainerManagerTest.java
hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/TestAuxServices.java
hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/TestContainerManagerRecovery.java
hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/container/TestContainer.java
hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/recovery/NMMemoryStateStoreService.java
hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/recovery/TestNMLeveldbStateStoreService.java
hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/webapp/TestNMWebServer.java
hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmnode/RMNodeImpl.java
hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerUtils.java
hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacityScheduler.java
hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestResourceTrackerService.java
Modified: hadoop/common/branches/HDFS-6584/hadoop-yarn-project/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-6584/hadoop-yarn-project/CHANGES.txt?rev=1617566&r1=1617565&r2=1617566&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-6584/hadoop-yarn-project/CHANGES.txt (original)
+++ hadoop/common/branches/HDFS-6584/hadoop-yarn-project/CHANGES.txt Tue Aug 12 19:25:57 2014
@@ -38,6 +38,9 @@ Release 2.6.0 - UNRELEASED
YARN-1354. Recover applications upon nodemanager restart. (Jason Lowe via
junping_du)
+ YARN-1337. Recover containers upon nodemanager restart. (Jason Lowe via
+ junping_du)
+
IMPROVEMENTS
YARN-2242. Improve exception information on AM launch crashes. (Li Lu
@@ -113,6 +116,9 @@ Release 2.6.0 - UNRELEASED
YARN-2138. Cleaned up notifyDone* APIs in RMStateStore. (Varun Saxena via
jianhe)
+ YARN-2373. Changed WebAppUtils to use Configuration#getPassword for
+ accessing SSL passwords. (Larry McCay via jianhe)
+
OPTIMIZATIONS
BUG FIXES
Modified: hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/webapp/util/WebAppUtils.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/webapp/util/WebAppUtils.java?rev=1617566&r1=1617565&r2=1617566&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/webapp/util/WebAppUtils.java (original)
+++ hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/webapp/util/WebAppUtils.java Tue Aug 12 19:25:57 2014
@@ -19,12 +19,12 @@ package org.apache.hadoop.yarn.webapp.ut
import static org.apache.hadoop.yarn.util.StringHelper.PATH_JOINER;
+import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.List;
-import java.util.Map;
import org.apache.hadoop.classification.InterfaceAudience.Private;
import org.apache.hadoop.classification.InterfaceStability.Evolving;
@@ -40,6 +40,12 @@ import org.apache.hadoop.yarn.util.RMHAU
@Private
@Evolving
public class WebAppUtils {
+ public static final String WEB_APP_TRUSTSTORE_PASSWORD_KEY =
+ "ssl.server.truststore.password";
+ public static final String WEB_APP_KEYSTORE_PASSWORD_KEY =
+ "ssl.server.keystore.password";
+ public static final String WEB_APP_KEY_PASSWORD_KEY =
+ "ssl.server.keystore.keypassword";
public static final String HTTPS_PREFIX = "https://";
public static final String HTTP_PREFIX = "http://";
@@ -274,21 +280,56 @@ public class WebAppUtils {
/**
* Load the SSL keystore / truststore into the HttpServer builder.
+ * @param builder the HttpServer2.Builder to populate with ssl config
*/
public static HttpServer2.Builder loadSslConfiguration(
HttpServer2.Builder builder) {
- Configuration sslConf = new Configuration(false);
+ return loadSslConfiguration(builder, null);
+ }
+
+ /**
+ * Load the SSL keystore / truststore into the HttpServer builder.
+ * @param builder the HttpServer2.Builder to populate with ssl config
+ * @param sslConf the Configuration instance to use during loading of SSL conf
+ */
+ public static HttpServer2.Builder loadSslConfiguration(
+ HttpServer2.Builder builder, Configuration sslConf) {
+ if (sslConf == null) {
+ sslConf = new Configuration(false);
+ }
boolean needsClientAuth = YarnConfiguration.YARN_SSL_CLIENT_HTTPS_NEED_AUTH_DEFAULT;
sslConf.addResource(YarnConfiguration.YARN_SSL_SERVER_RESOURCE_DEFAULT);
return builder
.needsClientAuth(needsClientAuth)
- .keyPassword(sslConf.get("ssl.server.keystore.keypassword"))
+ .keyPassword(getPassword(sslConf, WEB_APP_KEY_PASSWORD_KEY))
.keyStore(sslConf.get("ssl.server.keystore.location"),
- sslConf.get("ssl.server.keystore.password"),
+ getPassword(sslConf, WEB_APP_KEYSTORE_PASSWORD_KEY),
sslConf.get("ssl.server.keystore.type", "jks"))
.trustStore(sslConf.get("ssl.server.truststore.location"),
- sslConf.get("ssl.server.truststore.password"),
+ getPassword(sslConf, WEB_APP_TRUSTSTORE_PASSWORD_KEY),
sslConf.get("ssl.server.truststore.type", "jks"));
}
+
+ /**
+ * Leverages the Configuration.getPassword method to attempt to get
+ * passwords from the CredentialProvider API before falling back to
+ * clear text in config - if falling back is allowed.
+ * @param conf Configuration instance
+ * @param alias name of the credential to retreive
+ * @return String credential value or null
+ */
+ static String getPassword(Configuration conf, String alias) {
+ String password = null;
+ try {
+ char[] passchars = conf.getPassword(alias);
+ if (passchars != null) {
+ password = new String(passchars);
+ }
+ }
+ catch (IOException ioe) {
+ password = null;
+ }
+ return password;
+ }
}
Modified: hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/ContainerExecutor.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/ContainerExecutor.java?rev=1617566&r1=1617565&r2=1617566&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/ContainerExecutor.java (original)
+++ hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/ContainerExecutor.java Tue Aug 12 19:25:57 2014
@@ -18,6 +18,7 @@
package org.apache.hadoop.yarn.server.nodemanager;
+import java.io.File;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.ArrayList;
@@ -29,17 +30,18 @@ import java.util.concurrent.locks.Reentr
import java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock;
import java.util.concurrent.locks.ReentrantReadWriteLock.WriteLock;
+import org.apache.commons.io.FileUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configurable;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
-import org.apache.hadoop.util.Shell.ShellCommandExecutor;
import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
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.launcher.ContainerLaunch;
import org.apache.hadoop.yarn.server.nodemanager.util.ProcessIdFileReader;
import org.apache.hadoop.util.Shell;
import org.apache.hadoop.util.StringUtils;
@@ -126,9 +128,76 @@ public abstract class ContainerExecutor
public abstract void deleteAsUser(String user, Path subDir, Path... basedirs)
throws IOException, InterruptedException;
+ public abstract boolean isContainerProcessAlive(String user, String pid)
+ throws IOException;
+
+ /**
+ * Recover an already existing container. This is a blocking call and returns
+ * only when the container exits. Note that the container must have been
+ * activated prior to this call.
+ * @param user the user of the container
+ * @param containerId The ID of the container to reacquire
+ * @return The exit code of the pre-existing container
+ * @throws IOException
+ */
+ public int reacquireContainer(String user, ContainerId containerId)
+ throws IOException {
+ Path pidPath = getPidFilePath(containerId);
+ if (pidPath == null) {
+ LOG.warn(containerId + " is not active, returning terminated error");
+ return ExitCode.TERMINATED.getExitCode();
+ }
+
+ String pid = null;
+ pid = ProcessIdFileReader.getProcessId(pidPath);
+ if (pid == null) {
+ throw new IOException("Unable to determine pid for " + containerId);
+ }
+
+ LOG.info("Reacquiring " + containerId + " with pid " + pid);
+ try {
+ while(isContainerProcessAlive(user, pid)) {
+ Thread.sleep(1000);
+ }
+ } catch (InterruptedException e) {
+ throw new IOException("Interrupted while waiting for process " + pid
+ + " to exit", e);
+ }
+
+ // wait for exit code file to appear
+ String exitCodeFile = ContainerLaunch.getExitCodeFile(pidPath.toString());
+ File file = new File(exitCodeFile);
+ final int sleepMsec = 100;
+ int msecLeft = 2000;
+ while (!file.exists() && msecLeft >= 0) {
+ if (!isContainerActive(containerId)) {
+ LOG.info(containerId + " was deactivated");
+ return ExitCode.TERMINATED.getExitCode();
+ }
+ try {
+ Thread.sleep(sleepMsec);
+ } catch (InterruptedException e) {
+ throw new IOException(
+ "Interrupted while waiting for exit code from " + containerId, e);
+ }
+ msecLeft -= sleepMsec;
+ }
+ if (msecLeft < 0) {
+ throw new IOException("Timeout while waiting for exit code from "
+ + containerId);
+ }
+
+ try {
+ return Integer.parseInt(FileUtils.readFileToString(file).trim());
+ } catch (NumberFormatException e) {
+ throw new IOException("Error parsing exit code from pid " + pid, e);
+ }
+ }
+
public enum ExitCode {
FORCE_KILLED(137),
- TERMINATED(143);
+ TERMINATED(143),
+ LOST(154);
private final int code;
private ExitCode(int exitCode) {
Modified: hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/DefaultContainerExecutor.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/DefaultContainerExecutor.java?rev=1617566&r1=1617565&r2=1617566&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/DefaultContainerExecutor.java (original)
+++ hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/DefaultContainerExecutor.java Tue Aug 12 19:25:57 2014
@@ -273,25 +273,57 @@ public class DefaultContainerExecutor ex
private final class UnixLocalWrapperScriptBuilder
extends LocalWrapperScriptBuilder {
+ private final Path sessionScriptPath;
public UnixLocalWrapperScriptBuilder(Path containerWorkDir) {
super(containerWorkDir);
+ this.sessionScriptPath = new Path(containerWorkDir,
+ Shell.appendScriptExtension("default_container_executor_session"));
+ }
+
+ @Override
+ public void writeLocalWrapperScript(Path launchDst, Path pidFile)
+ throws IOException {
+ writeSessionScript(launchDst, pidFile);
+ super.writeLocalWrapperScript(launchDst, pidFile);
}
@Override
public void writeLocalWrapperScript(Path launchDst, Path pidFile,
PrintStream pout) {
-
- // We need to do a move as writing to a file is not atomic
- // Process reading a file being written to may get garbled data
- // hence write pid to tmp file first followed by a mv
+ String exitCodeFile = ContainerLaunch.getExitCodeFile(
+ pidFile.toString());
+ String tmpFile = exitCodeFile + ".tmp";
pout.println("#!/bin/bash");
- pout.println();
- pout.println("echo $$ > " + pidFile.toString() + ".tmp");
- pout.println("/bin/mv -f " + pidFile.toString() + ".tmp " + pidFile);
- String exec = Shell.isSetsidAvailable? "exec setsid" : "exec";
- pout.println(exec + " /bin/bash \"" +
- launchDst.toUri().getPath().toString() + "\"");
+ pout.println("/bin/bash \"" + sessionScriptPath.toString() + "\"");
+ pout.println("rc=$?");
+ pout.println("echo $rc > \"" + tmpFile + "\"");
+ pout.println("/bin/mv -f \"" + tmpFile + "\" \"" + exitCodeFile + "\"");
+ pout.println("exit $rc");
+ }
+
+ private void writeSessionScript(Path launchDst, Path pidFile)
+ throws IOException {
+ DataOutputStream out = null;
+ PrintStream pout = null;
+ try {
+ out = lfs.create(sessionScriptPath, EnumSet.of(CREATE, OVERWRITE));
+ pout = new PrintStream(out);
+ // We need to do a move as writing to a file is not atomic
+ // Process reading a file being written to may get garbled data
+ // hence write pid to tmp file first followed by a mv
+ pout.println("#!/bin/bash");
+ pout.println();
+ pout.println("echo $$ > " + pidFile.toString() + ".tmp");
+ pout.println("/bin/mv -f " + pidFile.toString() + ".tmp " + pidFile);
+ String exec = Shell.isSetsidAvailable? "exec setsid" : "exec";
+ pout.println(exec + " /bin/bash \"" +
+ launchDst.toUri().getPath().toString() + "\"");
+ } finally {
+ IOUtils.cleanup(LOG, pout, out);
+ }
+ lfs.setPermission(sessionScriptPath,
+ ContainerExecutor.TASK_LAUNCH_SCRIPT_PERMISSION);
}
}
@@ -310,6 +342,7 @@ public class DefaultContainerExecutor ex
@Override
public void writeLocalWrapperScript(Path launchDst, Path pidFile,
PrintStream pout) {
+ // TODO: exit code script for Windows
// On Windows, the pid is the container ID, so that it can also serve as
// the name of the job object created by winutils for task management.
@@ -342,6 +375,12 @@ public class DefaultContainerExecutor ex
return true;
}
+ @Override
+ public boolean isContainerProcessAlive(String user, String pid)
+ throws IOException {
+ return containerIsAlive(pid);
+ }
+
/**
* Returns true if the process with the specified pid is alive.
*
Modified: hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/LinuxContainerExecutor.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/LinuxContainerExecutor.java?rev=1617566&r1=1617565&r2=1617566&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/LinuxContainerExecutor.java (original)
+++ hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/LinuxContainerExecutor.java Tue Aug 12 19:25:57 2014
@@ -403,6 +403,13 @@ public class LinuxContainerExecutor exte
}
}
+ @Override
+ public boolean isContainerProcessAlive(String user, String pid)
+ throws IOException {
+ // Send a test signal to the process as the user to see if it's alive
+ return signalContainer(user, pid, Signal.NULL);
+ }
+
public void mountCgroups(List<String> cgroupKVs, String hierarchy)
throws IOException {
List<String> command = new ArrayList<String>(
Modified: hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeStatusUpdater.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeStatusUpdater.java?rev=1617566&r1=1617565&r2=1617566&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeStatusUpdater.java (original)
+++ hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeStatusUpdater.java Tue Aug 12 19:25:57 2014
@@ -23,11 +23,34 @@ import org.apache.hadoop.yarn.api.record
public interface NodeStatusUpdater extends Service {
+ /**
+ * Schedule a heartbeat to the ResourceManager outside of the normal,
+ * periodic heartbeating process. This is typically called when the state
+ * of containers on the node has changed to notify the RM sooner.
+ */
void sendOutofBandHeartBeat();
+ /**
+ * Get the ResourceManager identifier received during registration
+ * @return the ResourceManager ID
+ */
long getRMIdentifier();
+ /**
+ * Query if a container has recently completed
+ * @param containerId the container ID
+ * @return true if the container has recently completed
+ */
public boolean isContainerRecentlyStopped(ContainerId containerId);
+ /**
+ * Add a container to the list of containers that have recently completed
+ * @param containerId the ID of the completed container
+ */
+ public void addCompletedContainer(ContainerId containerId);
+
+ /**
+ * Clear the list of recently completed containers
+ */
public void clearFinishedContainersFromCache();
}
Modified: hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeStatusUpdaterImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeStatusUpdaterImpl.java?rev=1617566&r1=1617565&r2=1617566&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeStatusUpdaterImpl.java (original)
+++ hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeStatusUpdaterImpl.java Tue Aug 12 19:25:57 2014
@@ -364,8 +364,7 @@ public class NodeStatusUpdaterImpl exten
// Adding to finished containers cache. Cache will keep it around at
// least for #durationToTrackStoppedContainers duration. In the
// subsequent call to stop container it will get removed from cache.
- updateStoppedContainersInCache(container.getContainerId());
- addCompletedContainer(container);
+ addCompletedContainer(container.getContainerId());
}
}
if (LOG.isDebugEnabled()) {
@@ -393,8 +392,7 @@ public class NodeStatusUpdaterImpl exten
// Adding to finished containers cache. Cache will keep it around at
// least for #durationToTrackStoppedContainers duration. In the
// subsequent call to stop container it will get removed from cache.
- updateStoppedContainersInCache(container.getContainerId());
- addCompletedContainer(container);
+ addCompletedContainer(container.getContainerId());
}
}
LOG.info("Sending out " + containerStatuses.size()
@@ -402,9 +400,15 @@ public class NodeStatusUpdaterImpl exten
return containerStatuses;
}
- private void addCompletedContainer(Container container) {
+ @Override
+ public void addCompletedContainer(ContainerId containerId) {
synchronized (previousCompletedContainers) {
- previousCompletedContainers.add(container.getContainerId());
+ previousCompletedContainers.add(containerId);
+ }
+ synchronized (recentlyStoppedContainers) {
+ removeVeryOldStoppedContainersFromCache();
+ recentlyStoppedContainers.put(containerId,
+ System.currentTimeMillis() + durationToTrackStoppedContainers);
}
}
@@ -451,16 +455,6 @@ public class NodeStatusUpdaterImpl exten
}
}
- @Private
- @VisibleForTesting
- public void updateStoppedContainersInCache(ContainerId containerId) {
- synchronized (recentlyStoppedContainers) {
- removeVeryOldStoppedContainersFromCache();
- recentlyStoppedContainers.put(containerId,
- System.currentTimeMillis() + durationToTrackStoppedContainers);
- }
- }
-
@Override
public void clearFinishedContainersFromCache() {
synchronized (recentlyStoppedContainers) {
@@ -476,8 +470,14 @@ public class NodeStatusUpdaterImpl exten
Iterator<ContainerId> i =
recentlyStoppedContainers.keySet().iterator();
while (i.hasNext()) {
- if (recentlyStoppedContainers.get(i.next()) < currentTime) {
+ ContainerId cid = i.next();
+ if (recentlyStoppedContainers.get(cid) < currentTime) {
i.remove();
+ try {
+ context.getNMStateStore().removeContainer(cid);
+ } catch (IOException e) {
+ LOG.error("Unable to remove container " + cid + " in store", e);
+ }
} else {
break;
}
Modified: hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/ContainerManagerImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/ContainerManagerImpl.java?rev=1617566&r1=1617565&r2=1617566&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/ContainerManagerImpl.java (original)
+++ hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/ContainerManagerImpl.java Tue Aug 12 19:25:57 2014
@@ -127,6 +127,8 @@ import org.apache.hadoop.yarn.server.nod
import org.apache.hadoop.yarn.server.nodemanager.metrics.NodeManagerMetrics;
import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService;
import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService.RecoveredApplicationsState;
+import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService.RecoveredContainerState;
+import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService.RecoveredContainerStatus;
import org.apache.hadoop.yarn.server.nodemanager.security.authorize.NMPolicyProvider;
import org.apache.hadoop.yarn.server.security.ApplicationACLsManager;
import org.apache.hadoop.yarn.server.utils.BuilderUtils;
@@ -246,6 +248,10 @@ public class ContainerManagerImpl extend
recoverApplication(proto);
}
+ for (RecoveredContainerState rcs : stateStore.loadContainersState()) {
+ recoverContainer(rcs);
+ }
+
String diagnostic = "Application marked finished during recovery";
for (ApplicationId appId : appsState.getFinishedApplications()) {
dispatcher.getEventHandler().handle(
@@ -276,6 +282,60 @@ public class ContainerManagerImpl extend
app.handle(new ApplicationInitEvent(appId, acls));
}
+ @SuppressWarnings("unchecked")
+ private void recoverContainer(RecoveredContainerState rcs)
+ throws IOException {
+ StartContainerRequest req = rcs.getStartRequest();
+ ContainerLaunchContext launchContext = req.getContainerLaunchContext();
+ ContainerTokenIdentifier token =
+ BuilderUtils.newContainerTokenIdentifier(req.getContainerToken());
+ ContainerId containerId = token.getContainerID();
+ ApplicationId appId =
+ containerId.getApplicationAttemptId().getApplicationId();
+
+ LOG.info("Recovering " + containerId + " in state " + rcs.getStatus()
+ + " with exit code " + rcs.getExitCode());
+
+ if (context.getApplications().containsKey(appId)) {
+ Credentials credentials = parseCredentials(launchContext);
+ Container container = new ContainerImpl(getConfig(), dispatcher,
+ context.getNMStateStore(), req.getContainerLaunchContext(),
+ credentials, metrics, token, rcs.getStatus(), rcs.getExitCode(),
+ rcs.getDiagnostics(), rcs.getKilled());
+ context.getContainers().put(containerId, container);
+ dispatcher.getEventHandler().handle(
+ new ApplicationContainerInitEvent(container));
+ } else {
+ if (rcs.getStatus() != RecoveredContainerStatus.COMPLETED) {
+ LOG.warn(containerId + " has no corresponding application!");
+ }
+ LOG.info("Adding " + containerId + " to recently stopped containers");
+ nodeStatusUpdater.addCompletedContainer(containerId);
+ }
+ }
+
+ private void waitForRecoveredContainers() throws InterruptedException {
+ final int sleepMsec = 100;
+ int waitIterations = 100;
+ List<ContainerId> newContainers = new ArrayList<ContainerId>();
+ while (--waitIterations >= 0) {
+ newContainers.clear();
+ for (Container container : context.getContainers().values()) {
+ if (container.getContainerState() == org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerState.NEW) {
+ newContainers.add(container.getContainerId());
+ }
+ }
+ if (newContainers.isEmpty()) {
+ break;
+ }
+ LOG.info("Waiting for containers: " + newContainers);
+ Thread.sleep(sleepMsec);
+ }
+ if (waitIterations < 0) {
+ LOG.warn("Timeout waiting for recovered containers");
+ }
+ }
+
protected LogHandler createLogHandler(Configuration conf, Context context,
DeletionService deletionService) {
if (conf.getBoolean(YarnConfiguration.LOG_AGGREGATION_ENABLED,
@@ -309,6 +369,23 @@ public class ContainerManagerImpl extend
// Enqueue user dirs in deletion context
Configuration conf = getConfig();
+ final InetSocketAddress initialAddress = conf.getSocketAddr(
+ YarnConfiguration.NM_BIND_HOST,
+ YarnConfiguration.NM_ADDRESS,
+ YarnConfiguration.DEFAULT_NM_ADDRESS,
+ YarnConfiguration.DEFAULT_NM_PORT);
+ boolean usingEphemeralPort = (initialAddress.getPort() == 0);
+ if (context.getNMStateStore().canRecover() && usingEphemeralPort) {
+ throw new IllegalArgumentException("Cannot support recovery with an "
+ + "ephemeral server port. Check the setting of "
+ + YarnConfiguration.NM_ADDRESS);
+ }
+ // If recovering then delay opening the RPC service until the recovery
+ // of resources and containers have completed, otherwise requests from
+ // clients during recovery can interfere with the recovery process.
+ final boolean delayedRpcServerStart =
+ context.getNMStateStore().canRecover();
+
Configuration serverConf = new Configuration(conf);
// always enforce it to be token-based.
@@ -318,12 +395,6 @@ public class ContainerManagerImpl extend
YarnRPC rpc = YarnRPC.create(conf);
- InetSocketAddress initialAddress = conf.getSocketAddr(
- YarnConfiguration.NM_BIND_HOST,
- YarnConfiguration.NM_ADDRESS,
- YarnConfiguration.DEFAULT_NM_ADDRESS,
- YarnConfiguration.DEFAULT_NM_PORT);
-
server =
rpc.getServer(ContainerManagementProtocol.class, this, initialAddress,
serverConf, this.context.getNMTokenSecretManager(),
@@ -340,32 +411,61 @@ public class ContainerManagerImpl extend
LOG.info("Blocking new container-requests as container manager rpc" +
" server is still starting.");
this.setBlockNewContainerRequests(true);
- server.start();
-
- InetSocketAddress connectAddress;
+
String bindHost = conf.get(YarnConfiguration.NM_BIND_HOST);
String nmAddress = conf.getTrimmed(YarnConfiguration.NM_ADDRESS);
- if (bindHost == null || bindHost.isEmpty() ||
- nmAddress == null || nmAddress.isEmpty()) {
- connectAddress = NetUtils.getConnectAddress(server);
- } else {
- //a bind-host case with an address, to support overriding the first hostname
- //found when querying for our hostname with the specified address, combine
- //the specified address with the actual port listened on by the server
- connectAddress = NetUtils.getConnectAddress(
- new InetSocketAddress(nmAddress.split(":")[0],
- server.getListenerAddress().getPort()));
+ String hostOverride = null;
+ if (bindHost != null && !bindHost.isEmpty()
+ && nmAddress != null && !nmAddress.isEmpty()) {
+ //a bind-host case with an address, to support overriding the first
+ //hostname found when querying for our hostname with the specified
+ //address, combine the specified address with the actual port listened
+ //on by the server
+ hostOverride = nmAddress.split(":")[0];
}
- NodeId nodeId = NodeId.newInstance(
- connectAddress.getAddress().getCanonicalHostName(),
- connectAddress.getPort());
+ // setup node ID
+ InetSocketAddress connectAddress;
+ if (delayedRpcServerStart) {
+ connectAddress = NetUtils.getConnectAddress(initialAddress);
+ } else {
+ server.start();
+ connectAddress = NetUtils.getConnectAddress(server);
+ }
+ NodeId nodeId = buildNodeId(connectAddress, hostOverride);
((NodeManager.NMContext)context).setNodeId(nodeId);
this.context.getNMTokenSecretManager().setNodeId(nodeId);
this.context.getContainerTokenSecretManager().setNodeId(nodeId);
+
+ // start remaining services
+ super.serviceStart();
+
+ if (delayedRpcServerStart) {
+ waitForRecoveredContainers();
+ server.start();
+
+ // check that the node ID is as previously advertised
+ connectAddress = NetUtils.getConnectAddress(server);
+ NodeId serverNode = buildNodeId(connectAddress, hostOverride);
+ if (!serverNode.equals(nodeId)) {
+ throw new IOException("Node mismatch after server started, expected '"
+ + nodeId + "' but found '" + serverNode + "'");
+ }
+ }
+
LOG.info("ContainerManager started at " + connectAddress);
LOG.info("ContainerManager bound to " + initialAddress);
- super.serviceStart();
+ }
+
+ private NodeId buildNodeId(InetSocketAddress connectAddress,
+ String hostOverride) {
+ if (hostOverride != null) {
+ connectAddress = NetUtils.getConnectAddress(
+ new InetSocketAddress(hostOverride, connectAddress.getPort()));
+ }
+ return NodeId.newInstance(
+ connectAddress.getAddress().getCanonicalHostName(),
+ connectAddress.getPort());
}
void refreshServiceAcls(Configuration configuration,
@@ -704,7 +804,8 @@ public class ContainerManagerImpl extend
Credentials credentials = parseCredentials(launchContext);
Container container =
- new ContainerImpl(getConfig(), this.dispatcher, launchContext,
+ new ContainerImpl(getConfig(), this.dispatcher,
+ context.getNMStateStore(), launchContext,
credentials, metrics, containerTokenIdentifier);
ApplicationId applicationID =
containerId.getApplicationAttemptId().getApplicationId();
@@ -733,6 +834,7 @@ public class ContainerManagerImpl extend
new ApplicationInitEvent(applicationID, appAcls));
}
+ this.context.getNMStateStore().storeContainer(containerId, request);
dispatcher.getEventHandler().handle(
new ApplicationContainerInitEvent(container));
@@ -780,7 +882,7 @@ public class ContainerManagerImpl extend
}
private Credentials parseCredentials(ContainerLaunchContext launchContext)
- throws YarnException {
+ throws IOException {
Credentials credentials = new Credentials();
// //////////// Parse credentials
ByteBuffer tokens = launchContext.getTokens();
@@ -789,15 +891,11 @@ public class ContainerManagerImpl extend
DataInputByteBuffer buf = new DataInputByteBuffer();
tokens.rewind();
buf.reset(tokens);
- try {
- credentials.readTokenStorageStream(buf);
- if (LOG.isDebugEnabled()) {
- for (Token<? extends TokenIdentifier> tk : credentials.getAllTokens()) {
- LOG.debug(tk.getService() + " = " + tk.toString());
- }
+ credentials.readTokenStorageStream(buf);
+ if (LOG.isDebugEnabled()) {
+ for (Token<? extends TokenIdentifier> tk : credentials.getAllTokens()) {
+ LOG.debug(tk.getService() + " = " + tk.toString());
}
- } catch (IOException e) {
- throw RPCUtil.getRemoteException(e);
}
}
// //////////// End of parsing credentials
@@ -830,7 +928,7 @@ public class ContainerManagerImpl extend
@SuppressWarnings("unchecked")
private void stopContainerInternal(NMTokenIdentifier nmTokenIdentifier,
- ContainerId containerID) throws YarnException {
+ ContainerId containerID) throws YarnException, IOException {
String containerIDStr = containerID.toString();
Container container = this.context.getContainers().get(containerID);
LOG.info("Stopping container with container Id: " + containerIDStr);
@@ -843,6 +941,7 @@ public class ContainerManagerImpl extend
+ " is not handled by this NodeManager");
}
} else {
+ context.getNMStateStore().storeContainerKilled(containerID);
dispatcher.getEventHandler().handle(
new ContainerKillEvent(containerID,
ContainerExitStatus.KILLED_BY_APPMASTER,
Modified: hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/container/ContainerImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/container/ContainerImpl.java?rev=1617566&r1=1617565&r2=1617566&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/container/ContainerImpl.java (original)
+++ hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/container/ContainerImpl.java Tue Aug 12 19:25:57 2014
@@ -18,6 +18,7 @@
package org.apache.hadoop.yarn.server.nodemanager.containermanager.container;
+import java.io.IOException;
import java.net.URISyntaxException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
@@ -62,6 +63,8 @@ import org.apache.hadoop.yarn.server.nod
import org.apache.hadoop.yarn.server.nodemanager.containermanager.monitor.ContainerStartMonitoringEvent;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.monitor.ContainerStopMonitoringEvent;
import org.apache.hadoop.yarn.server.nodemanager.metrics.NodeManagerMetrics;
+import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService;
+import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService.RecoveredContainerStatus;
import org.apache.hadoop.yarn.server.utils.BuilderUtils;
import org.apache.hadoop.yarn.state.InvalidStateTransitonException;
import org.apache.hadoop.yarn.state.MultipleArcTransition;
@@ -75,6 +78,7 @@ public class ContainerImpl implements Co
private final Lock readLock;
private final Lock writeLock;
private final Dispatcher dispatcher;
+ private final NMStateStoreService stateStore;
private final Credentials credentials;
private final NodeManagerMetrics metrics;
private final ContainerLaunchContext launchContext;
@@ -101,12 +105,19 @@ public class ContainerImpl implements Co
private final List<LocalResourceRequest> appRsrcs =
new ArrayList<LocalResourceRequest>();
+ // whether container has been recovered after a restart
+ private RecoveredContainerStatus recoveredStatus =
+ RecoveredContainerStatus.REQUESTED;
+ // whether container was marked as killed after recovery
+ private boolean recoveredAsKilled = false;
+
public ContainerImpl(Configuration conf, Dispatcher dispatcher,
- ContainerLaunchContext launchContext, Credentials creds,
- NodeManagerMetrics metrics,
+ NMStateStoreService stateStore, ContainerLaunchContext launchContext,
+ Credentials creds, NodeManagerMetrics metrics,
ContainerTokenIdentifier containerTokenIdentifier) {
this.daemonConf = conf;
this.dispatcher = dispatcher;
+ this.stateStore = stateStore;
this.launchContext = launchContext;
this.containerTokenIdentifier = containerTokenIdentifier;
this.containerId = containerTokenIdentifier.getContainerID();
@@ -122,6 +133,21 @@ public class ContainerImpl implements Co
stateMachine = stateMachineFactory.make(this);
}
+ // constructor for a recovered container
+ public ContainerImpl(Configuration conf, Dispatcher dispatcher,
+ NMStateStoreService stateStore, ContainerLaunchContext launchContext,
+ Credentials creds, NodeManagerMetrics metrics,
+ ContainerTokenIdentifier containerTokenIdentifier,
+ RecoveredContainerStatus recoveredStatus, int exitCode,
+ String diagnostics, boolean wasKilled) {
+ this(conf, dispatcher, stateStore, launchContext, creds, metrics,
+ containerTokenIdentifier);
+ this.recoveredStatus = recoveredStatus;
+ this.exitCode = exitCode;
+ this.recoveredAsKilled = wasKilled;
+ this.diagnostics.append(diagnostics);
+ }
+
private static final ContainerDoneTransition CONTAINER_DONE_TRANSITION =
new ContainerDoneTransition();
@@ -135,8 +161,10 @@ public class ContainerImpl implements Co
new StateMachineFactory<ContainerImpl, ContainerState, ContainerEventType, ContainerEvent>(ContainerState.NEW)
// From NEW State
.addTransition(ContainerState.NEW,
- EnumSet.of(ContainerState.LOCALIZING, ContainerState.LOCALIZED,
- ContainerState.LOCALIZATION_FAILED),
+ EnumSet.of(ContainerState.LOCALIZING,
+ ContainerState.LOCALIZED,
+ ContainerState.LOCALIZATION_FAILED,
+ ContainerState.DONE),
ContainerEventType.INIT_CONTAINER, new RequestResourcesTransition())
.addTransition(ContainerState.NEW, ContainerState.NEW,
ContainerEventType.UPDATE_DIAGNOSTICS_MSG,
@@ -281,7 +309,9 @@ public class ContainerImpl implements Co
UPDATE_DIAGNOSTICS_TRANSITION)
.addTransition(ContainerState.CONTAINER_CLEANEDUP_AFTER_KILL,
ContainerState.CONTAINER_CLEANEDUP_AFTER_KILL,
- ContainerEventType.KILL_CONTAINER)
+ EnumSet.of(ContainerEventType.KILL_CONTAINER,
+ ContainerEventType.CONTAINER_EXITED_WITH_SUCCESS,
+ ContainerEventType.CONTAINER_EXITED_WITH_FAILURE))
// From DONE
.addTransition(ContainerState.DONE, ContainerState.DONE,
@@ -295,7 +325,9 @@ public class ContainerImpl implements Co
// we notify container of failed localization if localizer thread (for
// that container) fails for some reason
.addTransition(ContainerState.DONE, ContainerState.DONE,
- ContainerEventType.RESOURCE_FAILED)
+ EnumSet.of(ContainerEventType.RESOURCE_FAILED,
+ ContainerEventType.CONTAINER_EXITED_WITH_SUCCESS,
+ ContainerEventType.CONTAINER_EXITED_WITH_FAILURE))
// create the topology tables
.installTopology();
@@ -420,7 +452,7 @@ public class ContainerImpl implements Co
}
}
- @SuppressWarnings({"fallthrough", "unchecked"})
+ @SuppressWarnings("fallthrough")
private void finished() {
ApplicationId applicationId =
containerId.getApplicationAttemptId().getApplicationId();
@@ -458,7 +490,11 @@ public class ContainerImpl implements Co
}
metrics.releaseContainer(this.resource);
+ sendFinishedEvents();
+ }
+ @SuppressWarnings("unchecked")
+ private void sendFinishedEvents() {
// Inform the application
@SuppressWarnings("rawtypes")
EventHandler eventHandler = dispatcher.getEventHandler();
@@ -471,6 +507,45 @@ public class ContainerImpl implements Co
}
@SuppressWarnings("unchecked") // dispatcher not typed
+ private void sendLaunchEvent() {
+ ContainersLauncherEventType launcherEvent =
+ ContainersLauncherEventType.LAUNCH_CONTAINER;
+ if (recoveredStatus == RecoveredContainerStatus.LAUNCHED) {
+ // try to recover a container that was previously launched
+ launcherEvent = ContainersLauncherEventType.RECOVER_CONTAINER;
+ }
+ dispatcher.getEventHandler().handle(
+ new ContainersLauncherEvent(this, launcherEvent));
+ }
+
+ // Inform the ContainersMonitor to start monitoring the container's
+ // resource usage.
+ @SuppressWarnings("unchecked") // dispatcher not typed
+ private void sendContainerMonitorStartEvent() {
+ long pmemBytes = getResource().getMemory() * 1024 * 1024L;
+ float pmemRatio = daemonConf.getFloat(
+ YarnConfiguration.NM_VMEM_PMEM_RATIO,
+ YarnConfiguration.DEFAULT_NM_VMEM_PMEM_RATIO);
+ long vmemBytes = (long) (pmemRatio * pmemBytes);
+
+ dispatcher.getEventHandler().handle(
+ new ContainerStartMonitoringEvent(containerId,
+ vmemBytes, pmemBytes));
+ }
+
+ private void addDiagnostics(String... diags) {
+ for (String s : diags) {
+ this.diagnostics.append(s);
+ }
+ try {
+ stateStore.storeContainerDiagnostics(containerId, diagnostics);
+ } catch (IOException e) {
+ LOG.warn("Unable to update diagnostics in state store for "
+ + containerId, e);
+ }
+ }
+
+ @SuppressWarnings("unchecked") // dispatcher not typed
public void cleanup() {
Map<LocalResourceVisibility, Collection<LocalResourceRequest>> rsrc =
new HashMap<LocalResourceVisibility,
@@ -518,6 +593,16 @@ public class ContainerImpl implements Co
@Override
public ContainerState transition(ContainerImpl container,
ContainerEvent event) {
+ if (container.recoveredStatus == RecoveredContainerStatus.COMPLETED) {
+ container.sendFinishedEvents();
+ return ContainerState.DONE;
+ } else if (container.recoveredAsKilled &&
+ container.recoveredStatus == RecoveredContainerStatus.REQUESTED) {
+ // container was killed but never launched
+ container.finished();
+ return ContainerState.DONE;
+ }
+
final ContainerLaunchContext ctxt = container.launchContext;
container.metrics.initingContainer();
@@ -593,9 +678,7 @@ public class ContainerImpl implements Co
new ContainerLocalizationRequestEvent(container, req));
return ContainerState.LOCALIZING;
} else {
- container.dispatcher.getEventHandler().handle(
- new ContainersLauncherEvent(container,
- ContainersLauncherEventType.LAUNCH_CONTAINER));
+ container.sendLaunchEvent();
container.metrics.endInitingContainer();
return ContainerState.LOCALIZED;
}
@@ -606,7 +689,6 @@ public class ContainerImpl implements Co
* Transition when one of the requested resources for this container
* has been successfully localized.
*/
- @SuppressWarnings("unchecked") // dispatcher not typed
static class LocalizedTransition implements
MultipleArcTransition<ContainerImpl,ContainerEvent,ContainerState> {
@Override
@@ -626,9 +708,8 @@ public class ContainerImpl implements Co
if (!container.pendingResources.isEmpty()) {
return ContainerState.LOCALIZING;
}
- container.dispatcher.getEventHandler().handle(
- new ContainersLauncherEvent(container,
- ContainersLauncherEventType.LAUNCH_CONTAINER));
+
+ container.sendLaunchEvent();
container.metrics.endInitingContainer();
return ContainerState.LOCALIZED;
}
@@ -638,24 +719,22 @@ public class ContainerImpl implements Co
* Transition from LOCALIZED state to RUNNING state upon receiving
* a CONTAINER_LAUNCHED event
*/
- @SuppressWarnings("unchecked") // dispatcher not typed
static class LaunchTransition extends ContainerTransition {
+ @SuppressWarnings("unchecked")
@Override
public void transition(ContainerImpl container, ContainerEvent event) {
- // Inform the ContainersMonitor to start monitoring the container's
- // resource usage.
- long pmemBytes =
- container.getResource().getMemory() * 1024 * 1024L;
- float pmemRatio = container.daemonConf.getFloat(
- YarnConfiguration.NM_VMEM_PMEM_RATIO,
- YarnConfiguration.DEFAULT_NM_VMEM_PMEM_RATIO);
- long vmemBytes = (long) (pmemRatio * pmemBytes);
-
- container.dispatcher.getEventHandler().handle(
- new ContainerStartMonitoringEvent(container.containerId,
- vmemBytes, pmemBytes));
+ container.sendContainerMonitorStartEvent();
container.metrics.runningContainer();
container.wasLaunched = true;
+
+ if (container.recoveredAsKilled) {
+ LOG.info("Killing " + container.containerId
+ + " due to recovered as killed");
+ container.addDiagnostics("Container recovered as killed.\n");
+ container.dispatcher.getEventHandler().handle(
+ new ContainersLauncherEvent(container,
+ ContainersLauncherEventType.CLEANUP_CONTAINER));
+ }
}
}
@@ -707,8 +786,7 @@ public class ContainerImpl implements Co
ContainerExitEvent exitEvent = (ContainerExitEvent) event;
container.exitCode = exitEvent.getExitCode();
if (exitEvent.getDiagnosticInfo() != null) {
- container.diagnostics.append(exitEvent.getDiagnosticInfo())
- .append('\n');
+ container.addDiagnostics(exitEvent.getDiagnosticInfo(), "\n");
}
// TODO: Add containerWorkDir to the deletion service.
@@ -735,7 +813,7 @@ public class ContainerImpl implements Co
@Override
public void transition(ContainerImpl container, ContainerEvent event) {
super.transition(container, event);
- container.diagnostics.append("Killed by external signal\n");
+ container.addDiagnostics("Killed by external signal\n");
}
}
@@ -750,9 +828,7 @@ public class ContainerImpl implements Co
ContainerResourceFailedEvent rsrcFailedEvent =
(ContainerResourceFailedEvent) event;
- container.diagnostics.append(rsrcFailedEvent.getDiagnosticMessage()
- + "\n");
-
+ container.addDiagnostics(rsrcFailedEvent.getDiagnosticMessage(), "\n");
// Inform the localizer to decrement reference counts and cleanup
// resources.
@@ -775,8 +851,8 @@ public class ContainerImpl implements Co
container.metrics.endInitingContainer();
ContainerKillEvent killEvent = (ContainerKillEvent) event;
container.exitCode = killEvent.getContainerExitStatus();
- container.diagnostics.append(killEvent.getDiagnostic()).append("\n");
- container.diagnostics.append("Container is killed before being launched.\n");
+ container.addDiagnostics(killEvent.getDiagnostic(), "\n");
+ container.addDiagnostics("Container is killed before being launched.\n");
}
}
@@ -817,7 +893,7 @@ public class ContainerImpl implements Co
new ContainersLauncherEvent(container,
ContainersLauncherEventType.CLEANUP_CONTAINER));
ContainerKillEvent killEvent = (ContainerKillEvent) event;
- container.diagnostics.append(killEvent.getDiagnostic()).append("\n");
+ container.addDiagnostics(killEvent.getDiagnostic(), "\n");
container.exitCode = killEvent.getContainerExitStatus();
}
}
@@ -836,8 +912,7 @@ public class ContainerImpl implements Co
}
if (exitEvent.getDiagnosticInfo() != null) {
- container.diagnostics.append(exitEvent.getDiagnosticInfo())
- .append('\n');
+ container.addDiagnostics(exitEvent.getDiagnosticInfo(), "\n");
}
// The process/process-grp is killed. Decrement reference counts and
@@ -877,8 +952,8 @@ public class ContainerImpl implements Co
public void transition(ContainerImpl container, ContainerEvent event) {
ContainerKillEvent killEvent = (ContainerKillEvent) event;
container.exitCode = killEvent.getContainerExitStatus();
- container.diagnostics.append(killEvent.getDiagnostic()).append("\n");
- container.diagnostics.append("Container is killed before being launched.\n");
+ container.addDiagnostics(killEvent.getDiagnostic(), "\n");
+ container.addDiagnostics("Container is killed before being launched.\n");
super.transition(container, event);
}
}
@@ -892,8 +967,14 @@ public class ContainerImpl implements Co
public void transition(ContainerImpl container, ContainerEvent event) {
ContainerDiagnosticsUpdateEvent updateEvent =
(ContainerDiagnosticsUpdateEvent) event;
- container.diagnostics.append(updateEvent.getDiagnosticsUpdate())
- .append("\n");
+ container.addDiagnostics(updateEvent.getDiagnosticsUpdate(), "\n");
+ try {
+ container.stateStore.storeContainerDiagnostics(container.containerId,
+ container.diagnostics);
+ } catch (IOException e) {
+ LOG.warn("Unable to update state store diagnostics for "
+ + container.containerId, e);
+ }
}
}
Modified: hadoop/common/branches/HDFS-6584/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
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-6584/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?rev=1617566&r1=1617565&r2=1617566&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-6584/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 (original)
+++ hadoop/common/branches/HDFS-6584/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 Tue Aug 12 19:25:57 2014
@@ -87,22 +87,23 @@ public class ContainerLaunch implements
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";
- private final Dispatcher dispatcher;
- private final ContainerExecutor exec;
+ protected final Dispatcher dispatcher;
+ protected final ContainerExecutor exec;
private final Application app;
- private final Container container;
+ protected final Container container;
private final Configuration conf;
private final Context context;
private final ContainerManagerImpl containerManager;
- private volatile AtomicBoolean shouldLaunchContainer = new AtomicBoolean(false);
- private volatile AtomicBoolean completed = new AtomicBoolean(false);
+ protected AtomicBoolean shouldLaunchContainer = new AtomicBoolean(false);
+ protected AtomicBoolean completed = new AtomicBoolean(false);
private long sleepDelayBeforeSigKill = 250;
private long maxKillWaitTime = 2000;
- private Path pidFilePath = null;
+ protected Path pidFilePath = null;
private final LocalDirsHandlerService dirsHandler;
@@ -223,14 +224,11 @@ public class ContainerLaunch implements
+ Path.SEPARATOR + containerIdStr,
LocalDirAllocator.SIZE_UNKNOWN, false);
- String pidFileSuffix = String.format(ContainerLaunch.PID_FILE_NAME_FMT,
- containerIdStr);
+ String pidFileSubpath = getPidFileSubpath(appIdStr, containerIdStr);
// pid file should be in nm private dir so that it is not
// accessible by users
- pidFilePath = dirsHandler.getLocalPathForWrite(
- ResourceLocalizationService.NM_PRIVATE_DIR + Path.SEPARATOR
- + pidFileSuffix);
+ pidFilePath = dirsHandler.getLocalPathForWrite(pidFileSubpath);
List<String> localDirs = dirsHandler.getLocalDirs();
List<String> logDirs = dirsHandler.getLogDirs();
@@ -288,6 +286,7 @@ public class ContainerLaunch implements
dispatcher.getEventHandler().handle(new ContainerEvent(
containerID,
ContainerEventType.CONTAINER_LAUNCHED));
+ context.getNMStateStore().storeContainerLaunched(containerID);
// Check if the container is signalled to be killed.
if (!shouldLaunchContainer.compareAndSet(false, true)) {
@@ -310,6 +309,11 @@ public class ContainerLaunch implements
} finally {
completed.set(true);
exec.deactivateContainer(containerID);
+ try {
+ context.getNMStateStore().storeContainerCompleted(containerID, ret);
+ } catch (IOException e) {
+ LOG.error("Unable to set exit code for container " + containerID);
+ }
}
if (LOG.isDebugEnabled()) {
@@ -342,6 +346,11 @@ public class ContainerLaunch implements
ContainerEventType.CONTAINER_EXITED_WITH_SUCCESS));
return 0;
}
+
+ protected String getPidFileSubpath(String appIdStr, String containerIdStr) {
+ return getContainerPrivateDir(appIdStr, containerIdStr) + Path.SEPARATOR
+ + String.format(ContainerLaunch.PID_FILE_NAME_FMT, containerIdStr);
+ }
/**
* Cleanup the container.
@@ -357,6 +366,13 @@ public class ContainerLaunch implements
String containerIdStr = ConverterUtils.toString(containerId);
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 = !shouldLaunchContainer.compareAndSet(false, true);
if (!alreadyLaunched) {
@@ -421,6 +437,7 @@ public class ContainerLaunch implements
if (pidFilePath != null) {
FileContext lfs = FileContext.getLocalFSFileContext();
lfs.delete(pidFilePath, false);
+ lfs.delete(pidFilePath.suffix(EXIT_CODE_FILE_SUFFIX), false);
}
}
}
@@ -479,6 +496,10 @@ public class ContainerLaunch implements
+ appIdStr;
}
+ Context getContext() {
+ return context;
+ }
+
@VisibleForTesting
static abstract class ShellScriptBuilder {
public static ShellScriptBuilder create() {
@@ -787,4 +808,7 @@ public class ContainerLaunch implements
}
}
+ public static String getExitCodeFile(String pidFile) {
+ return pidFile + EXIT_CODE_FILE_SUFFIX;
+ }
}
Modified: hadoop/common/branches/HDFS-6584/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
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-6584/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?rev=1617566&r1=1617565&r2=1617566&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-6584/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 (original)
+++ hadoop/common/branches/HDFS-6584/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 Tue Aug 12 19:25:57 2014
@@ -24,7 +24,6 @@ import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -32,21 +31,16 @@ import org.apache.hadoop.conf.Configurat
import org.apache.hadoop.fs.FileContext;
import org.apache.hadoop.fs.UnsupportedFileSystemException;
import org.apache.hadoop.service.AbstractService;
-import org.apache.hadoop.util.Shell;
import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.event.Dispatcher;
import org.apache.hadoop.yarn.event.EventHandler;
import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
import org.apache.hadoop.yarn.server.nodemanager.ContainerExecutor;
-import org.apache.hadoop.yarn.server.nodemanager.ContainerExecutor.ExitCode;
import org.apache.hadoop.yarn.server.nodemanager.Context;
import org.apache.hadoop.yarn.server.nodemanager.LocalDirsHandlerService;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.ContainerManagerImpl;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.application.Application;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.Container;
-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.container.ContainerState;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.localizer.ResourceLocalizationService;
import com.google.common.annotations.VisibleForTesting;
@@ -107,7 +101,6 @@ public class ContainersLauncher extends
super.serviceStop();
}
- @SuppressWarnings("unchecked")
@Override
public void handle(ContainersLauncherEvent event) {
// TODO: ContainersLauncher launches containers one by one!!
@@ -125,6 +118,14 @@ public class ContainersLauncher extends
containerLauncher.submit(launch);
running.put(containerId, launch);
break;
+ case RECOVER_CONTAINER:
+ app = context.getApplications().get(
+ containerId.getApplicationAttemptId().getApplicationId());
+ launch = new RecoveredContainerLaunch(context, getConfig(), dispatcher,
+ exec, app, event.getContainer(), dirsHandler, containerManager);
+ containerLauncher.submit(launch);
+ running.put(containerId, launch);
+ break;
case CLEANUP_CONTAINER:
ContainerLaunch launcher = running.remove(containerId);
if (launcher == null) {
Modified: hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/ContainersLauncherEventType.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/ContainersLauncherEventType.java?rev=1617566&r1=1617565&r2=1617566&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/ContainersLauncherEventType.java (original)
+++ hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/ContainersLauncherEventType.java Tue Aug 12 19:25:57 2014
@@ -20,5 +20,6 @@ package org.apache.hadoop.yarn.server.no
public enum ContainersLauncherEventType {
LAUNCH_CONTAINER,
+ RECOVER_CONTAINER,
CLEANUP_CONTAINER, // The process(grp) itself.
}
Modified: hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/AppLogAggregator.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/AppLogAggregator.java?rev=1617566&r1=1617565&r2=1617566&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/AppLogAggregator.java (original)
+++ hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/AppLogAggregator.java Tue Aug 12 19:25:57 2014
@@ -25,5 +25,7 @@ public interface AppLogAggregator extend
void startContainerLogAggregation(ContainerId containerId,
boolean wasContainerSuccessful);
+ void abortLogAggregation();
+
void finishLogAggregation();
}
Modified: hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/AppLogAggregatorImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/AppLogAggregatorImpl.java?rev=1617566&r1=1617565&r2=1617566&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/AppLogAggregatorImpl.java (original)
+++ hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/AppLogAggregatorImpl.java Tue Aug 12 19:25:57 2014
@@ -70,6 +70,7 @@ public class AppLogAggregatorImpl implem
private final BlockingQueue<ContainerId> pendingContainers;
private final AtomicBoolean appFinishing = new AtomicBoolean();
private final AtomicBoolean appAggregationFinished = new AtomicBoolean();
+ private final AtomicBoolean aborted = new AtomicBoolean();
private final Map<ApplicationAccessType, String> appAcls;
private LogWriter writer = null;
@@ -150,7 +151,7 @@ public class AppLogAggregatorImpl implem
private void doAppLogAggregation() {
ContainerId containerId;
- while (!this.appFinishing.get()) {
+ while (!this.appFinishing.get() && !this.aborted.get()) {
synchronized(this) {
try {
wait(THREAD_SLEEP_TIME);
@@ -161,6 +162,10 @@ public class AppLogAggregatorImpl implem
}
}
+ if (this.aborted.get()) {
+ return;
+ }
+
// Application is finished. Finish pending-containers
while ((containerId = this.pendingContainers.poll()) != null) {
uploadLogsForContainer(containerId);
@@ -255,4 +260,11 @@ public class AppLogAggregatorImpl implem
this.appFinishing.set(true);
this.notifyAll();
}
+
+ @Override
+ public synchronized void abortLogAggregation() {
+ LOG.info("Aborting log aggregation for " + this.applicationId);
+ this.aborted.set(true);
+ this.notifyAll();
+ }
}
Modified: hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/LogAggregationService.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/LogAggregationService.java?rev=1617566&r1=1617565&r2=1617566&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/LogAggregationService.java (original)
+++ hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/LogAggregationService.java Tue Aug 12 19:25:57 2014
@@ -142,9 +142,17 @@ public class LogAggregationService exten
private void stopAggregators() {
threadPool.shutdown();
+ // if recovery on restart is supported then leave outstanding aggregations
+ // to the next restart
+ boolean shouldAbort = context.getNMStateStore().canRecover()
+ && !context.getDecommissioned();
// politely ask to finish
for (AppLogAggregator aggregator : appLogAggregators.values()) {
- aggregator.finishLogAggregation();
+ if (shouldAbort) {
+ aggregator.abortLogAggregation();
+ } else {
+ aggregator.finishLogAggregation();
+ }
}
while (!threadPool.isTerminated()) { // wait for all threads to finish
for (ApplicationId appId : appLogAggregators.keySet()) {
Modified: hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/recovery/NMLeveldbStateStoreService.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/recovery/NMLeveldbStateStoreService.java?rev=1617566&r1=1617565&r2=1617566&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/recovery/NMLeveldbStateStoreService.java (original)
+++ hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/recovery/NMLeveldbStateStoreService.java Tue Aug 12 19:25:57 2014
@@ -35,6 +35,8 @@ import org.apache.hadoop.conf.Configurat
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.yarn.api.protocolrecords.StartContainerRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.StartContainerRequestPBImpl;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.ContainerId;
@@ -45,6 +47,7 @@ import org.apache.hadoop.yarn.proto.Yarn
import org.apache.hadoop.yarn.proto.YarnServerNodemanagerRecoveryProtos.ContainerManagerApplicationProto;
import org.apache.hadoop.yarn.proto.YarnServerNodemanagerRecoveryProtos.DeletionServiceDeleteTaskProto;
import org.apache.hadoop.yarn.proto.YarnServerNodemanagerRecoveryProtos.LocalizedResourceProto;
+import org.apache.hadoop.yarn.proto.YarnServiceProtos.StartContainerRequestProto;
import org.apache.hadoop.yarn.server.api.records.MasterKey;
import org.apache.hadoop.yarn.server.api.records.impl.pb.MasterKeyPBImpl;
import org.apache.hadoop.yarn.server.records.Version;
@@ -90,6 +93,14 @@ public class NMLeveldbStateStoreService
private static final String LOCALIZATION_FILECACHE_SUFFIX = "filecache/";
private static final String LOCALIZATION_APPCACHE_SUFFIX = "appcache/";
+ private static final String CONTAINERS_KEY_PREFIX =
+ "ContainerManager/containers/";
+ private static final String CONTAINER_REQUEST_KEY_SUFFIX = "/request";
+ private static final String CONTAINER_DIAGS_KEY_SUFFIX = "/diagnostics";
+ private static final String CONTAINER_LAUNCHED_KEY_SUFFIX = "/launched";
+ private static final String CONTAINER_KILLED_KEY_SUFFIX = "/killed";
+ private static final String CONTAINER_EXIT_CODE_KEY_SUFFIX = "/exitcode";
+
private static final String CURRENT_MASTER_KEY_SUFFIX = "CurrentMasterKey";
private static final String PREV_MASTER_KEY_SUFFIX = "PreviousMasterKey";
private static final String NM_TOKENS_KEY_PREFIX = "NMTokens/";
@@ -104,6 +115,8 @@ public class NMLeveldbStateStoreService
private static final String CONTAINER_TOKENS_PREV_MASTER_KEY =
CONTAINER_TOKENS_KEY_PREFIX + PREV_MASTER_KEY_SUFFIX;
+ private static final byte[] EMPTY_VALUE = new byte[0];
+
private DB db;
public NMLeveldbStateStoreService() {
@@ -123,6 +136,160 @@ public class NMLeveldbStateStoreService
@Override
+ public List<RecoveredContainerState> loadContainersState()
+ throws IOException {
+ ArrayList<RecoveredContainerState> containers =
+ new ArrayList<RecoveredContainerState>();
+ LeveldbIterator iter = null;
+ try {
+ iter = new LeveldbIterator(db);
+ iter.seek(bytes(CONTAINERS_KEY_PREFIX));
+
+ while (iter.hasNext()) {
+ Entry<byte[],byte[]> entry = iter.peekNext();
+ String key = asString(entry.getKey());
+ if (!key.startsWith(CONTAINERS_KEY_PREFIX)) {
+ break;
+ }
+
+ int idEndPos = key.indexOf('/', CONTAINERS_KEY_PREFIX.length());
+ if (idEndPos < 0) {
+ throw new IOException("Unable to determine container in key: " + key);
+ }
+ ContainerId containerId = ConverterUtils.toContainerId(
+ key.substring(CONTAINERS_KEY_PREFIX.length(), idEndPos));
+ String keyPrefix = key.substring(0, idEndPos+1);
+ containers.add(loadContainerState(containerId, iter, keyPrefix));
+ }
+ } catch (DBException e) {
+ throw new IOException(e);
+ } finally {
+ if (iter != null) {
+ iter.close();
+ }
+ }
+
+ return containers;
+ }
+
+ private RecoveredContainerState loadContainerState(ContainerId containerId,
+ LeveldbIterator iter, String keyPrefix) throws IOException {
+ RecoveredContainerState rcs = new RecoveredContainerState();
+ rcs.status = RecoveredContainerStatus.REQUESTED;
+ while (iter.hasNext()) {
+ Entry<byte[],byte[]> entry = iter.peekNext();
+ String key = asString(entry.getKey());
+ if (!key.startsWith(keyPrefix)) {
+ break;
+ }
+ iter.next();
+
+ String suffix = key.substring(keyPrefix.length()-1); // start with '/'
+ if (suffix.equals(CONTAINER_REQUEST_KEY_SUFFIX)) {
+ rcs.startRequest = new StartContainerRequestPBImpl(
+ StartContainerRequestProto.parseFrom(entry.getValue()));
+ } else if (suffix.equals(CONTAINER_DIAGS_KEY_SUFFIX)) {
+ rcs.diagnostics = asString(entry.getValue());
+ } else if (suffix.equals(CONTAINER_LAUNCHED_KEY_SUFFIX)) {
+ if (rcs.status == RecoveredContainerStatus.REQUESTED) {
+ rcs.status = RecoveredContainerStatus.LAUNCHED;
+ }
+ } else if (suffix.equals(CONTAINER_KILLED_KEY_SUFFIX)) {
+ rcs.killed = true;
+ } else if (suffix.equals(CONTAINER_EXIT_CODE_KEY_SUFFIX)) {
+ rcs.status = RecoveredContainerStatus.COMPLETED;
+ rcs.exitCode = Integer.parseInt(asString(entry.getValue()));
+ } else {
+ throw new IOException("Unexpected container state key: " + key);
+ }
+ }
+ return rcs;
+ }
+
+ @Override
+ public void storeContainer(ContainerId containerId,
+ StartContainerRequest startRequest) throws IOException {
+ String key = CONTAINERS_KEY_PREFIX + containerId.toString()
+ + CONTAINER_REQUEST_KEY_SUFFIX;
+ try {
+ db.put(bytes(key),
+ ((StartContainerRequestPBImpl) startRequest).getProto().toByteArray());
+ } catch (DBException e) {
+ throw new IOException(e);
+ }
+ }
+
+ @Override
+ public void storeContainerDiagnostics(ContainerId containerId,
+ StringBuilder diagnostics) throws IOException {
+ String key = CONTAINERS_KEY_PREFIX + containerId.toString()
+ + CONTAINER_DIAGS_KEY_SUFFIX;
+ try {
+ db.put(bytes(key), bytes(diagnostics.toString()));
+ } catch (DBException e) {
+ throw new IOException(e);
+ }
+ }
+
+ @Override
+ public void storeContainerLaunched(ContainerId containerId)
+ throws IOException {
+ String key = CONTAINERS_KEY_PREFIX + containerId.toString()
+ + CONTAINER_LAUNCHED_KEY_SUFFIX;
+ try {
+ db.put(bytes(key), EMPTY_VALUE);
+ } catch (DBException e) {
+ throw new IOException(e);
+ }
+ }
+
+ @Override
+ public void storeContainerKilled(ContainerId containerId)
+ throws IOException {
+ String key = CONTAINERS_KEY_PREFIX + containerId.toString()
+ + CONTAINER_KILLED_KEY_SUFFIX;
+ try {
+ db.put(bytes(key), EMPTY_VALUE);
+ } catch (DBException e) {
+ throw new IOException(e);
+ }
+ }
+
+ @Override
+ public void storeContainerCompleted(ContainerId containerId,
+ int exitCode) throws IOException {
+ String key = CONTAINERS_KEY_PREFIX + containerId.toString()
+ + CONTAINER_EXIT_CODE_KEY_SUFFIX;
+ try {
+ db.put(bytes(key), bytes(Integer.toString(exitCode)));
+ } catch (DBException e) {
+ throw new IOException(e);
+ }
+ }
+
+ @Override
+ public void removeContainer(ContainerId containerId)
+ throws IOException {
+ String keyPrefix = CONTAINERS_KEY_PREFIX + containerId.toString();
+ try {
+ WriteBatch batch = db.createWriteBatch();
+ try {
+ batch.delete(bytes(keyPrefix + CONTAINER_REQUEST_KEY_SUFFIX));
+ batch.delete(bytes(keyPrefix + CONTAINER_DIAGS_KEY_SUFFIX));
+ batch.delete(bytes(keyPrefix + CONTAINER_LAUNCHED_KEY_SUFFIX));
+ batch.delete(bytes(keyPrefix + CONTAINER_KILLED_KEY_SUFFIX));
+ batch.delete(bytes(keyPrefix + CONTAINER_EXIT_CODE_KEY_SUFFIX));
+ db.write(batch);
+ } finally {
+ batch.close();
+ }
+ } catch (DBException e) {
+ throw new IOException(e);
+ }
+ }
+
+
+ @Override
public RecoveredApplicationsState loadApplicationsState()
throws IOException {
RecoveredApplicationsState state = new RecoveredApplicationsState();
Modified: hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/recovery/NMNullStateStoreService.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/recovery/NMNullStateStoreService.java?rev=1617566&r1=1617565&r2=1617566&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/recovery/NMNullStateStoreService.java (original)
+++ hadoop/common/branches/HDFS-6584/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/recovery/NMNullStateStoreService.java Tue Aug 12 19:25:57 2014
@@ -19,9 +19,11 @@
package org.apache.hadoop.yarn.server.nodemanager.recovery;
import java.io.IOException;
+import java.util.List;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.yarn.api.protocolrecords.StartContainerRequest;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.ContainerId;
@@ -63,6 +65,42 @@ public class NMNullStateStoreService ext
}
@Override
+ public List<RecoveredContainerState> loadContainersState()
+ throws IOException {
+ throw new UnsupportedOperationException(
+ "Recovery not supported by this state store");
+ }
+
+ @Override
+ public void storeContainer(ContainerId containerId,
+ StartContainerRequest startRequest) throws IOException {
+ }
+
+ @Override
+ public void storeContainerDiagnostics(ContainerId containerId,
+ StringBuilder diagnostics) throws IOException {
+ }
+
+ @Override
+ public void storeContainerLaunched(ContainerId containerId)
+ throws IOException {
+ }
+
+ @Override
+ public void storeContainerKilled(ContainerId containerId)
+ throws IOException {
+ }
+
+ @Override
+ public void storeContainerCompleted(ContainerId containerId, int exitCode)
+ throws IOException {
+ }
+
+ @Override
+ public void removeContainer(ContainerId containerId) throws IOException {
+ }
+
+ @Override
public RecoveredLocalizationState loadLocalizationState()
throws IOException {
throw new UnsupportedOperationException(