DockerContainerExecutor extends ContainerExecutor { @Override public int launchContainer(Container container, Path nmPrivateContainerScriptPath, Path nmPrivateTokensPath, String userName, String appId, Path containerWorkDir, List<String> localDirs, List<String> logDirs) throws IOException { String containerImageName = container.getLaunchContext().getEnvironment() .get(YarnConfiguration.NM_DOCKER_CONTAINER_EXECUTOR_IMAGE_NAME); if (LOG.isDebugEnabled()) { LOG.debug("containerImageName from launchContext: " + containerImageName); } Preconditions.checkArgument(!Strings.isNullOrEmpty(containerImageName), "Container image must not be null"); containerImageName = containerImageName.replaceAll("['\"]", ""); Preconditions.checkArgument(saneDockerImage(containerImageName), "Image: " + containerImageName + " is not a proper docker image"); String dockerExecutor = getConf().get(YarnConfiguration.NM_DOCKER_CONTAINER_EXECUTOR_EXEC_NAME, YarnConfiguration.NM_DEFAULT_DOCKER_CONTAINER_EXECUTOR_EXEC_NAME); FsPermission dirPerm = new FsPermission(APPDIR_PERM); ContainerId containerId = container.getContainerId(); String containerIdStr = ConverterUtils.toString(containerId); String appIdStr = ConverterUtils.toString( containerId.getApplicationAttemptId(). getApplicationId()); for (String sLocalDir : localDirs) { Path usersdir = new Path(sLocalDir, ContainerLocalizer.USERCACHE); Path userdir = new Path(usersdir, userName); Path appCacheDir = new Path(userdir, ContainerLocalizer.APPCACHE); Path appDir = new Path(appCacheDir, appIdStr); Path containerDir = new Path(appDir, containerIdStr); createDir(containerDir, dirPerm, true, userName); } createContainerLogDirs(appIdStr, containerIdStr, logDirs, userName); Path tmpDir = new Path(containerWorkDir, YarnConfiguration.DEFAULT_CONTAINER_TEMP_DIR); createDir(tmpDir, dirPerm, false, userName); Path launchDst = new Path(containerWorkDir, ContainerLaunch.CONTAINER_SCRIPT); lfs.util().copy(nmPrivateContainerScriptPath, launchDst); Path tokenDst = new Path(containerWorkDir, ContainerLaunch.FINAL_CONTAINER_TOKENS_FILE); lfs.util().copy(nmPrivateTokensPath, tokenDst); String localDirMount = toMount(localDirs); String logDirMount = toMount(logDirs); String containerWorkDirMount = toMount(Collections.singletonList(containerWorkDir.toUri().getPath())); StringBuilder commands = new StringBuilder(); String commandStr = commands.append(dockerExecutor) .append(" ") .append("run") .append(" ") .append("--rm --net=host") .append(" ") .append(" --name " + containerIdStr) .append(localDirMount) .append(logDirMount) .append(containerWorkDirMount) .append(" ") .append(containerImageName) .toString(); String dockerPidScript = "`" + dockerExecutor + " inspect --format {{.State.Pid}} " + containerIdStr + "`"; LocalWrapperScriptBuilder sb = new UnixLocalWrapperScriptBuilder(containerWorkDir, commandStr, dockerPidScript); Path pidFile = getPidFilePath(containerId); if (pidFile != null) { sb.writeLocalWrapperScript(launchDst, pidFile); } else { LOG.info("Container " + containerIdStr + " was marked as inactive. Returning terminated error"); return ExitCode.TERMINATED.getExitCode(); } ShellCommandExecutor shExec = null; try { lfs.setPermission(launchDst, ContainerExecutor.TASK_LAUNCH_SCRIPT_PERMISSION); lfs.setPermission(sb.getWrapperScriptPath(), ContainerExecutor.TASK_LAUNCH_SCRIPT_PERMISSION); String[] command = getRunCommand(sb.getWrapperScriptPath().toString(), containerIdStr, userName, pidFile, this.getConf()); if (LOG.isDebugEnabled()) { LOG.debug("launchContainer: " + commandStr + " " + Joiner.on(" ").join(command)); } shExec = new ShellCommandExecutor( command, new File(containerWorkDir.toUri().getPath()), container.getLaunchContext().getEnvironment(), 0L, false); if (isContainerActive(containerId)) { shExec.execute(); } else { LOG.info("Container " + containerIdStr + " was marked as inactive. Returning terminated error"); return ExitCode.TERMINATED.getExitCode(); } } catch (IOException e) { if (null == shExec) { return -1; } int exitCode = shExec.getExitCode(); LOG.warn("Exit code from container " + containerId + " is : " + exitCode); if (exitCode != ExitCode.FORCE_KILLED.getExitCode() && exitCode != ExitCode.TERMINATED.getExitCode()) { LOG.warn("Exception from container-launch with container ID: " + containerId + " and exit code: " + exitCode, e); logOutput(shExec.getOutput()); String diagnostics = "Exception from container-launch: \n" + StringUtils.stringifyException(e) + "\n" + shExec.getOutput(); container.handle(new ContainerDiagnosticsUpdateEvent(containerId, diagnostics)); } else { container.handle(new ContainerDiagnosticsUpdateEvent(containerId, "Container killed on request. Exit code is " + exitCode)); } return exitCode; } finally { if (shExec != null) { shExec.close(); } } return 0; } DockerContainerExecutor(); @Override void init(); @Override synchronized void startLocalizer(Path nmPrivateContainerTokensPath,
                                          InetSocketAddress nmAddr, String user, String appId, String locId,
                                          LocalDirsHandlerService dirsHandler); @Override int launchContainer(Container container,
                             Path nmPrivateContainerScriptPath, Path nmPrivateTokensPath,
                             String userName, String appId, Path containerWorkDir,
                             List<String> localDirs, List<String> logDirs); @Override void writeLaunchEnv(OutputStream out, Map<String, String> environment, Map<Path, List<String>> resources, List<String> command); @Override boolean signalContainer(String user, String pid, Signal signal); @Override boolean isContainerProcessAlive(String user, String pid); @VisibleForTesting static boolean containerIsAlive(String pid); @Override void deleteAsUser(String user, Path subDir, Path... baseDirs); static final String DOCKER_CONTAINER_EXECUTOR_SCRIPT; static final String DOCKER_CONTAINER_EXECUTOR_SESSION_SCRIPT; static final String DOCKER_IMAGE_PATTERN; }
@Test public void testLaunchContainer() throws IOException { if (!shouldRun()) { LOG.warn("Docker not installed, aborting test."); return; } Map<String, String> env = new HashMap<String, String>(); env.put(YarnConfiguration.NM_DOCKER_CONTAINER_EXECUTOR_IMAGE_NAME, testImage); String touchFileName = "touch-file-" + System.currentTimeMillis(); File touchFile = new File(dirsHandler.getLocalDirs().get(0), touchFileName); ContainerId cId = getNextContainerId(); int ret = runAndBlock( cId, env, "touch", touchFile.getAbsolutePath(), "&&", "cp", touchFile.getAbsolutePath(), "/"); assertEquals(0, ret); }