@SuppressWarnings("rawtypes") @Test (timeout = 10000) public void testCallFailureWithNullLocalizedResources() { Container container = mock(Container.class); when(container.getContainerId()).thenReturn(ContainerId.newContainerId( ApplicationAttemptId.newInstance(ApplicationId.newInstance( System.currentTimeMillis(), 1), 1), 1)); ContainerLaunchContext clc = mock(ContainerLaunchContext.class); when(clc.getCommands()).thenReturn(Collections.<String>emptyList()); when(container.getLaunchContext()).thenReturn(clc); when(container.getLocalizedResources()).thenReturn(null); Dispatcher dispatcher = mock(Dispatcher.class); EventHandler eventHandler = new EventHandler() { public void handle(Event event) { Assert.assertTrue(event instanceof ContainerExitEvent); ContainerExitEvent exitEvent = (ContainerExitEvent) event; Assert.assertEquals(ContainerEventType.CONTAINER_EXITED_WITH_FAILURE, exitEvent.getType()); } }; when(dispatcher.getEventHandler()).thenReturn(eventHandler); ContainerLaunch launch = new ContainerLaunch(context, new Configuration(), dispatcher, exec, null, container, dirsHandler, containerManager); launch.call(); }
@Override @SuppressWarnings("unchecked") public Integer call() { final ContainerLaunchContext launchContext = container.getLaunchContext(); Map<Path,List<String>> localResources = null; ContainerId containerID = container.getContainerId(); String containerIdStr = ConverterUtils.toString(containerID); final List<String> command = launchContext.getCommands(); int ret = -1; if (container.getContainerState() == ContainerState.KILLING) { dispatcher.getEventHandler().handle( new ContainerExitEvent(containerID, ContainerEventType.CONTAINER_KILLED_ON_REQUEST, Shell.WINDOWS ? ExitCode.FORCE_KILLED.getExitCode() : ExitCode.TERMINATED.getExitCode(), "Container terminated before launch.")); return 0; } Path containerLogDir; try { localResources = container.getLocalizedResources(); if (localResources == null) { throw RPCUtil.getRemoteException( "Unable to get local resources when Container " + containerID + " is at " + container.getContainerState()); } localResources = olr.filterLocalResources(localResources); final String user = container.getUser(); List<String> newCmds = new ArrayList<String>(command.size()); String appIdStr = app.getAppId().toString(); String relativeContainerLogDir = ContainerLaunch .getRelativeContainerLogDir(appIdStr, containerIdStr); containerLogDir = dirsHandler.getLogPathForWrite(relativeContainerLogDir, false); for (String str : command) { newCmds.add(expandEnvironment(str, containerLogDir)); } launchContext.setCommands(newCmds); Map<String, String> environment = launchContext.getEnvironment(); for (Entry<String, String> entry : environment.entrySet()) { String value = entry.getValue(); value = expandEnvironment(value, containerLogDir); entry.setValue(value); } FileContext lfs = FileContext.getLocalFSFileContext(); Path nmPrivateContainerScriptPath = dirsHandler.getLocalPathForWrite( getContainerPrivateDir(appIdStr, containerIdStr) + Path.SEPARATOR + CONTAINER_SCRIPT); Path nmPrivateTokensPath = dirsHandler.getLocalPathForWrite( getContainerPrivateDir(appIdStr, containerIdStr) + Path.SEPARATOR + String.format(ContainerLocalizer.TOKEN_FILE_NAME_FMT, containerIdStr)); Path nmPrivateClasspathJarDir = dirsHandler.getLocalPathForWrite( getContainerPrivateDir(appIdStr, containerIdStr)); DataOutputStream containerScriptOutStream = null; DataOutputStream tokensOutStream = null; Path containerWorkDir = dirsHandler.getLocalPathForWrite(ContainerLocalizer.USERCACHE + Path.SEPARATOR + user + Path.SEPARATOR + ContainerLocalizer.APPCACHE + Path.SEPARATOR + appIdStr + Path.SEPARATOR + containerIdStr, LocalDirAllocator.SIZE_UNKNOWN, false); String pidFileSubpath = getPidFileSubpath(appIdStr, containerIdStr); pidFilePath = dirsHandler.getLocalPathForWrite(pidFileSubpath); List<String> localDirs = dirsHandler.getLocalDirs(); List<String> logDirs = dirsHandler.getLogDirs(); List<String> filecacheDirs = getNMFilecacheDirs(localDirs); List<String> userLocalDirs = getUserLocalDirs(localDirs); List<String> containerLogDirs = new ArrayList<String>(); for( String logDir : logDirs) { containerLogDirs.add(logDir + Path.SEPARATOR + relativeContainerLogDir); } if (!dirsHandler.areDisksHealthy()) { ret = ContainerExitStatus.DISKS_FAILED; throw new IOException("Most of the disks failed. " + dirsHandler.getDisksHealthReport(false)); } List<String> containerLocalDirs = new ArrayList<>(localDirs.size()); try { List<Path> appDirs = new ArrayList<Path>(localDirs.size()); for (String localDir : localDirs) { Path usersdir = new Path(localDir, ContainerLocalizer.USERCACHE); Path userdir = new Path(usersdir, user); Path appsdir = new Path(userdir, ContainerLocalizer.APPCACHE); appDirs.add(new Path(appsdir, appIdStr)); String containerLocalDir = localDir + Path.SEPARATOR + ContainerLocalizer.USERCACHE + Path.SEPARATOR + user + Path.SEPARATOR + ContainerLocalizer.APPCACHE + Path.SEPARATOR + appIdStr + Path.SEPARATOR; containerLocalDirs.add(containerLocalDir); } containerScriptOutStream = lfs.create(nmPrivateContainerScriptPath, EnumSet.of(CREATE, OVERWRITE)); environment.put( ApplicationConstants.CONTAINER_TOKEN_FILE_ENV_NAME, new Path(containerWorkDir, FINAL_CONTAINER_TOKENS_FILE).toUri().getPath()); sanitizeEnv(environment, containerWorkDir, appDirs, userLocalDirs, containerLogDirs, localResources, nmPrivateClasspathJarDir); exec.writeLaunchEnv(containerScriptOutStream, environment, localResources, launchContext.getCommands()); tokensOutStream = lfs.create(nmPrivateTokensPath, EnumSet.of(CREATE, OVERWRITE)); Credentials creds = container.getCredentials(); creds.writeTokenStorageToStream(tokensOutStream); } finally { IOUtils.cleanup(LOG, containerScriptOutStream, tokensOutStream); } dispatcher.getEventHandler().handle(new ContainerEvent( containerID, ContainerEventType.CONTAINER_LAUNCHED)); context.getNMStateStore().storeContainerLaunched(containerID); if (!shouldLaunchContainer.compareAndSet(false, true)) { LOG.info("Container " + containerIdStr + " not launched as " + "cleanup already called"); ret = ExitCode.TERMINATED.getExitCode(); } else { ContainerStartContext csc = new ContainerStartContext.Builder() .setContainer(container) .setLocalizedResources(localResources) .setNmPrivateContainerScriptPath(nmPrivateContainerScriptPath) .setNmPrivateTokensPath(nmPrivateTokensPath) .setUser(user) .setAppId(appIdStr) .setContainerWorkDir(containerWorkDir) .setLocalDirs(localDirs) .setLogDirs(logDirs) .setFilecacheDirs(filecacheDirs) .setUserLocalDirs(userLocalDirs) .setContainerLocalDirs(containerLocalDirs) .setContainerLogDirs(containerLogDirs) .build(); ret = exec.prepareContainer(csc); if (ret == 0) { exec.activateContainer(containerID, pidFilePath); ret = exec.launchContainer(csc); } else { LOG.warn("Failed to prepare contaner." + ret); return ret; } } } catch (Throwable e) { LOG.warn("Failed to launch container.", e); dispatcher.getEventHandler().handle(new ContainerExitEvent( containerID, ContainerEventType.CONTAINER_EXITED_WITH_FAILURE, ret, e.getMessage())); return ret; } 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()) { LOG.debug("Container " + containerIdStr + " completed with exit code " + ret); } StringBuilder diagnosticInfo = new StringBuilder("Container exited with a non-zero exit code "); diagnosticInfo.append(ret); diagnosticInfo.append(". "); if (ret == ExitCode.FORCE_KILLED.getExitCode() || ret == ExitCode.TERMINATED.getExitCode()) { dispatcher.getEventHandler().handle( new ContainerExitEvent(containerID, ContainerEventType.CONTAINER_KILLED_ON_REQUEST, ret, diagnosticInfo.toString())); return ret; } if (ret != 0) { handleContainerExitWithFailure(containerID, ret, containerLogDir, diagnosticInfo); return ret; } LOG.info("Container " + containerIdStr + " succeeded "); dispatcher.getEventHandler().handle( new ContainerEvent(containerID, ContainerEventType.CONTAINER_EXITED_WITH_SUCCESS)); return 0; }
ContainerLaunch implements Callable<Integer> { @Override @SuppressWarnings("unchecked") public Integer call() { final ContainerLaunchContext launchContext = container.getLaunchContext(); Map<Path,List<String>> localResources = null; ContainerId containerID = container.getContainerId(); String containerIdStr = ConverterUtils.toString(containerID); final List<String> command = launchContext.getCommands(); int ret = -1; if (container.getContainerState() == ContainerState.KILLING) { dispatcher.getEventHandler().handle( new ContainerExitEvent(containerID, ContainerEventType.CONTAINER_KILLED_ON_REQUEST, Shell.WINDOWS ? ExitCode.FORCE_KILLED.getExitCode() : ExitCode.TERMINATED.getExitCode(), "Container terminated before launch.")); return 0; } Path containerLogDir; try { localResources = container.getLocalizedResources(); if (localResources == null) { throw RPCUtil.getRemoteException( "Unable to get local resources when Container " + containerID + " is at " + container.getContainerState()); } localResources = olr.filterLocalResources(localResources); final String user = container.getUser(); List<String> newCmds = new ArrayList<String>(command.size()); String appIdStr = app.getAppId().toString(); String relativeContainerLogDir = ContainerLaunch .getRelativeContainerLogDir(appIdStr, containerIdStr); containerLogDir = dirsHandler.getLogPathForWrite(relativeContainerLogDir, false); for (String str : command) { newCmds.add(expandEnvironment(str, containerLogDir)); } launchContext.setCommands(newCmds); Map<String, String> environment = launchContext.getEnvironment(); for (Entry<String, String> entry : environment.entrySet()) { String value = entry.getValue(); value = expandEnvironment(value, containerLogDir); entry.setValue(value); } FileContext lfs = FileContext.getLocalFSFileContext(); Path nmPrivateContainerScriptPath = dirsHandler.getLocalPathForWrite( getContainerPrivateDir(appIdStr, containerIdStr) + Path.SEPARATOR + CONTAINER_SCRIPT); Path nmPrivateTokensPath = dirsHandler.getLocalPathForWrite( getContainerPrivateDir(appIdStr, containerIdStr) + Path.SEPARATOR + String.format(ContainerLocalizer.TOKEN_FILE_NAME_FMT, containerIdStr)); Path nmPrivateClasspathJarDir = dirsHandler.getLocalPathForWrite( getContainerPrivateDir(appIdStr, containerIdStr)); DataOutputStream containerScriptOutStream = null; DataOutputStream tokensOutStream = null; Path containerWorkDir = dirsHandler.getLocalPathForWrite(ContainerLocalizer.USERCACHE + Path.SEPARATOR + user + Path.SEPARATOR + ContainerLocalizer.APPCACHE + Path.SEPARATOR + appIdStr + Path.SEPARATOR + containerIdStr, LocalDirAllocator.SIZE_UNKNOWN, false); String pidFileSubpath = getPidFileSubpath(appIdStr, containerIdStr); pidFilePath = dirsHandler.getLocalPathForWrite(pidFileSubpath); List<String> localDirs = dirsHandler.getLocalDirs(); List<String> logDirs = dirsHandler.getLogDirs(); List<String> filecacheDirs = getNMFilecacheDirs(localDirs); List<String> userLocalDirs = getUserLocalDirs(localDirs); List<String> containerLogDirs = new ArrayList<String>(); for( String logDir : logDirs) { containerLogDirs.add(logDir + Path.SEPARATOR + relativeContainerLogDir); } if (!dirsHandler.areDisksHealthy()) { ret = ContainerExitStatus.DISKS_FAILED; throw new IOException("Most of the disks failed. " + dirsHandler.getDisksHealthReport(false)); } List<String> containerLocalDirs = new ArrayList<>(localDirs.size()); try { List<Path> appDirs = new ArrayList<Path>(localDirs.size()); for (String localDir : localDirs) { Path usersdir = new Path(localDir, ContainerLocalizer.USERCACHE); Path userdir = new Path(usersdir, user); Path appsdir = new Path(userdir, ContainerLocalizer.APPCACHE); appDirs.add(new Path(appsdir, appIdStr)); String containerLocalDir = localDir + Path.SEPARATOR + ContainerLocalizer.USERCACHE + Path.SEPARATOR + user + Path.SEPARATOR + ContainerLocalizer.APPCACHE + Path.SEPARATOR + appIdStr + Path.SEPARATOR; containerLocalDirs.add(containerLocalDir); } containerScriptOutStream = lfs.create(nmPrivateContainerScriptPath, EnumSet.of(CREATE, OVERWRITE)); environment.put( ApplicationConstants.CONTAINER_TOKEN_FILE_ENV_NAME, new Path(containerWorkDir, FINAL_CONTAINER_TOKENS_FILE).toUri().getPath()); sanitizeEnv(environment, containerWorkDir, appDirs, userLocalDirs, containerLogDirs, localResources, nmPrivateClasspathJarDir); exec.writeLaunchEnv(containerScriptOutStream, environment, localResources, launchContext.getCommands()); tokensOutStream = lfs.create(nmPrivateTokensPath, EnumSet.of(CREATE, OVERWRITE)); Credentials creds = container.getCredentials(); creds.writeTokenStorageToStream(tokensOutStream); } finally { IOUtils.cleanup(LOG, containerScriptOutStream, tokensOutStream); } dispatcher.getEventHandler().handle(new ContainerEvent( containerID, ContainerEventType.CONTAINER_LAUNCHED)); context.getNMStateStore().storeContainerLaunched(containerID); if (!shouldLaunchContainer.compareAndSet(false, true)) { LOG.info("Container " + containerIdStr + " not launched as " + "cleanup already called"); ret = ExitCode.TERMINATED.getExitCode(); } else { ContainerStartContext csc = new ContainerStartContext.Builder() .setContainer(container) .setLocalizedResources(localResources) .setNmPrivateContainerScriptPath(nmPrivateContainerScriptPath) .setNmPrivateTokensPath(nmPrivateTokensPath) .setUser(user) .setAppId(appIdStr) .setContainerWorkDir(containerWorkDir) .setLocalDirs(localDirs) .setLogDirs(logDirs) .setFilecacheDirs(filecacheDirs) .setUserLocalDirs(userLocalDirs) .setContainerLocalDirs(containerLocalDirs) .setContainerLogDirs(containerLogDirs) .build(); ret = exec.prepareContainer(csc); if (ret == 0) { exec.activateContainer(containerID, pidFilePath); ret = exec.launchContainer(csc); } else { LOG.warn("Failed to prepare contaner." + ret); return ret; } } } catch (Throwable e) { LOG.warn("Failed to launch container.", e); dispatcher.getEventHandler().handle(new ContainerExitEvent( containerID, ContainerEventType.CONTAINER_EXITED_WITH_FAILURE, ret, e.getMessage())); return ret; } 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()) { LOG.debug("Container " + containerIdStr + " completed with exit code " + ret); } StringBuilder diagnosticInfo = new StringBuilder("Container exited with a non-zero exit code "); diagnosticInfo.append(ret); diagnosticInfo.append(". "); if (ret == ExitCode.FORCE_KILLED.getExitCode() || ret == ExitCode.TERMINATED.getExitCode()) { dispatcher.getEventHandler().handle( new ContainerExitEvent(containerID, ContainerEventType.CONTAINER_KILLED_ON_REQUEST, ret, diagnosticInfo.toString())); return ret; } if (ret != 0) { handleContainerExitWithFailure(containerID, ret, containerLogDir, diagnosticInfo); return ret; } LOG.info("Container " + containerIdStr + " succeeded "); dispatcher.getEventHandler().handle( new ContainerEvent(containerID, ContainerEventType.CONTAINER_EXITED_WITH_SUCCESS)); return 0; } }
ContainerLaunch implements Callable<Integer> { @Override @SuppressWarnings("unchecked") public Integer call() { final ContainerLaunchContext launchContext = container.getLaunchContext(); Map<Path,List<String>> localResources = null; ContainerId containerID = container.getContainerId(); String containerIdStr = ConverterUtils.toString(containerID); final List<String> command = launchContext.getCommands(); int ret = -1; if (container.getContainerState() == ContainerState.KILLING) { dispatcher.getEventHandler().handle( new ContainerExitEvent(containerID, ContainerEventType.CONTAINER_KILLED_ON_REQUEST, Shell.WINDOWS ? ExitCode.FORCE_KILLED.getExitCode() : ExitCode.TERMINATED.getExitCode(), "Container terminated before launch.")); return 0; } Path containerLogDir; try { localResources = container.getLocalizedResources(); if (localResources == null) { throw RPCUtil.getRemoteException( "Unable to get local resources when Container " + containerID + " is at " + container.getContainerState()); } localResources = olr.filterLocalResources(localResources); final String user = container.getUser(); List<String> newCmds = new ArrayList<String>(command.size()); String appIdStr = app.getAppId().toString(); String relativeContainerLogDir = ContainerLaunch .getRelativeContainerLogDir(appIdStr, containerIdStr); containerLogDir = dirsHandler.getLogPathForWrite(relativeContainerLogDir, false); for (String str : command) { newCmds.add(expandEnvironment(str, containerLogDir)); } launchContext.setCommands(newCmds); Map<String, String> environment = launchContext.getEnvironment(); for (Entry<String, String> entry : environment.entrySet()) { String value = entry.getValue(); value = expandEnvironment(value, containerLogDir); entry.setValue(value); } FileContext lfs = FileContext.getLocalFSFileContext(); Path nmPrivateContainerScriptPath = dirsHandler.getLocalPathForWrite( getContainerPrivateDir(appIdStr, containerIdStr) + Path.SEPARATOR + CONTAINER_SCRIPT); Path nmPrivateTokensPath = dirsHandler.getLocalPathForWrite( getContainerPrivateDir(appIdStr, containerIdStr) + Path.SEPARATOR + String.format(ContainerLocalizer.TOKEN_FILE_NAME_FMT, containerIdStr)); Path nmPrivateClasspathJarDir = dirsHandler.getLocalPathForWrite( getContainerPrivateDir(appIdStr, containerIdStr)); DataOutputStream containerScriptOutStream = null; DataOutputStream tokensOutStream = null; Path containerWorkDir = dirsHandler.getLocalPathForWrite(ContainerLocalizer.USERCACHE + Path.SEPARATOR + user + Path.SEPARATOR + ContainerLocalizer.APPCACHE + Path.SEPARATOR + appIdStr + Path.SEPARATOR + containerIdStr, LocalDirAllocator.SIZE_UNKNOWN, false); String pidFileSubpath = getPidFileSubpath(appIdStr, containerIdStr); pidFilePath = dirsHandler.getLocalPathForWrite(pidFileSubpath); List<String> localDirs = dirsHandler.getLocalDirs(); List<String> logDirs = dirsHandler.getLogDirs(); List<String> filecacheDirs = getNMFilecacheDirs(localDirs); List<String> userLocalDirs = getUserLocalDirs(localDirs); List<String> containerLogDirs = new ArrayList<String>(); for( String logDir : logDirs) { containerLogDirs.add(logDir + Path.SEPARATOR + relativeContainerLogDir); } if (!dirsHandler.areDisksHealthy()) { ret = ContainerExitStatus.DISKS_FAILED; throw new IOException("Most of the disks failed. " + dirsHandler.getDisksHealthReport(false)); } List<String> containerLocalDirs = new ArrayList<>(localDirs.size()); try { List<Path> appDirs = new ArrayList<Path>(localDirs.size()); for (String localDir : localDirs) { Path usersdir = new Path(localDir, ContainerLocalizer.USERCACHE); Path userdir = new Path(usersdir, user); Path appsdir = new Path(userdir, ContainerLocalizer.APPCACHE); appDirs.add(new Path(appsdir, appIdStr)); String containerLocalDir = localDir + Path.SEPARATOR + ContainerLocalizer.USERCACHE + Path.SEPARATOR + user + Path.SEPARATOR + ContainerLocalizer.APPCACHE + Path.SEPARATOR + appIdStr + Path.SEPARATOR; containerLocalDirs.add(containerLocalDir); } containerScriptOutStream = lfs.create(nmPrivateContainerScriptPath, EnumSet.of(CREATE, OVERWRITE)); environment.put( ApplicationConstants.CONTAINER_TOKEN_FILE_ENV_NAME, new Path(containerWorkDir, FINAL_CONTAINER_TOKENS_FILE).toUri().getPath()); sanitizeEnv(environment, containerWorkDir, appDirs, userLocalDirs, containerLogDirs, localResources, nmPrivateClasspathJarDir); exec.writeLaunchEnv(containerScriptOutStream, environment, localResources, launchContext.getCommands()); tokensOutStream = lfs.create(nmPrivateTokensPath, EnumSet.of(CREATE, OVERWRITE)); Credentials creds = container.getCredentials(); creds.writeTokenStorageToStream(tokensOutStream); } finally { IOUtils.cleanup(LOG, containerScriptOutStream, tokensOutStream); } dispatcher.getEventHandler().handle(new ContainerEvent( containerID, ContainerEventType.CONTAINER_LAUNCHED)); context.getNMStateStore().storeContainerLaunched(containerID); if (!shouldLaunchContainer.compareAndSet(false, true)) { LOG.info("Container " + containerIdStr + " not launched as " + "cleanup already called"); ret = ExitCode.TERMINATED.getExitCode(); } else { ContainerStartContext csc = new ContainerStartContext.Builder() .setContainer(container) .setLocalizedResources(localResources) .setNmPrivateContainerScriptPath(nmPrivateContainerScriptPath) .setNmPrivateTokensPath(nmPrivateTokensPath) .setUser(user) .setAppId(appIdStr) .setContainerWorkDir(containerWorkDir) .setLocalDirs(localDirs) .setLogDirs(logDirs) .setFilecacheDirs(filecacheDirs) .setUserLocalDirs(userLocalDirs) .setContainerLocalDirs(containerLocalDirs) .setContainerLogDirs(containerLogDirs) .build(); ret = exec.prepareContainer(csc); if (ret == 0) { exec.activateContainer(containerID, pidFilePath); ret = exec.launchContainer(csc); } else { LOG.warn("Failed to prepare contaner." + ret); return ret; } } } catch (Throwable e) { LOG.warn("Failed to launch container.", e); dispatcher.getEventHandler().handle(new ContainerExitEvent( containerID, ContainerEventType.CONTAINER_EXITED_WITH_FAILURE, ret, e.getMessage())); return ret; } 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()) { LOG.debug("Container " + containerIdStr + " completed with exit code " + ret); } StringBuilder diagnosticInfo = new StringBuilder("Container exited with a non-zero exit code "); diagnosticInfo.append(ret); diagnosticInfo.append(". "); if (ret == ExitCode.FORCE_KILLED.getExitCode() || ret == ExitCode.TERMINATED.getExitCode()) { dispatcher.getEventHandler().handle( new ContainerExitEvent(containerID, ContainerEventType.CONTAINER_KILLED_ON_REQUEST, ret, diagnosticInfo.toString())); return ret; } if (ret != 0) { handleContainerExitWithFailure(containerID, ret, containerLogDir, diagnosticInfo); return ret; } LOG.info("Container " + containerIdStr + " succeeded "); dispatcher.getEventHandler().handle( new ContainerEvent(containerID, ContainerEventType.CONTAINER_EXITED_WITH_SUCCESS)); return 0; } ContainerLaunch(Context context, Configuration configuration,
      Dispatcher dispatcher, ContainerExecutor exec, Application app,
      Container container, LocalDirsHandlerService dirsHandler,
      ContainerManagerImpl containerManager); }
ContainerLaunch implements Callable<Integer> { @Override @SuppressWarnings("unchecked") public Integer call() { final ContainerLaunchContext launchContext = container.getLaunchContext(); Map<Path,List<String>> localResources = null; ContainerId containerID = container.getContainerId(); String containerIdStr = ConverterUtils.toString(containerID); final List<String> command = launchContext.getCommands(); int ret = -1; if (container.getContainerState() == ContainerState.KILLING) { dispatcher.getEventHandler().handle( new ContainerExitEvent(containerID, ContainerEventType.CONTAINER_KILLED_ON_REQUEST, Shell.WINDOWS ? ExitCode.FORCE_KILLED.getExitCode() : ExitCode.TERMINATED.getExitCode(), "Container terminated before launch.")); return 0; } Path containerLogDir; try { localResources = container.getLocalizedResources(); if (localResources == null) { throw RPCUtil.getRemoteException( "Unable to get local resources when Container " + containerID + " is at " + container.getContainerState()); } localResources = olr.filterLocalResources(localResources); final String user = container.getUser(); List<String> newCmds = new ArrayList<String>(command.size()); String appIdStr = app.getAppId().toString(); String relativeContainerLogDir = ContainerLaunch .getRelativeContainerLogDir(appIdStr, containerIdStr); containerLogDir = dirsHandler.getLogPathForWrite(relativeContainerLogDir, false); for (String str : command) { newCmds.add(expandEnvironment(str, containerLogDir)); } launchContext.setCommands(newCmds); Map<String, String> environment = launchContext.getEnvironment(); for (Entry<String, String> entry : environment.entrySet()) { String value = entry.getValue(); value = expandEnvironment(value, containerLogDir); entry.setValue(value); } FileContext lfs = FileContext.getLocalFSFileContext(); Path nmPrivateContainerScriptPath = dirsHandler.getLocalPathForWrite( getContainerPrivateDir(appIdStr, containerIdStr) + Path.SEPARATOR + CONTAINER_SCRIPT); Path nmPrivateTokensPath = dirsHandler.getLocalPathForWrite( getContainerPrivateDir(appIdStr, containerIdStr) + Path.SEPARATOR + String.format(ContainerLocalizer.TOKEN_FILE_NAME_FMT, containerIdStr)); Path nmPrivateClasspathJarDir = dirsHandler.getLocalPathForWrite( getContainerPrivateDir(appIdStr, containerIdStr)); DataOutputStream containerScriptOutStream = null; DataOutputStream tokensOutStream = null; Path containerWorkDir = dirsHandler.getLocalPathForWrite(ContainerLocalizer.USERCACHE + Path.SEPARATOR + user + Path.SEPARATOR + ContainerLocalizer.APPCACHE + Path.SEPARATOR + appIdStr + Path.SEPARATOR + containerIdStr, LocalDirAllocator.SIZE_UNKNOWN, false); String pidFileSubpath = getPidFileSubpath(appIdStr, containerIdStr); pidFilePath = dirsHandler.getLocalPathForWrite(pidFileSubpath); List<String> localDirs = dirsHandler.getLocalDirs(); List<String> logDirs = dirsHandler.getLogDirs(); List<String> filecacheDirs = getNMFilecacheDirs(localDirs); List<String> userLocalDirs = getUserLocalDirs(localDirs); List<String> containerLogDirs = new ArrayList<String>(); for( String logDir : logDirs) { containerLogDirs.add(logDir + Path.SEPARATOR + relativeContainerLogDir); } if (!dirsHandler.areDisksHealthy()) { ret = ContainerExitStatus.DISKS_FAILED; throw new IOException("Most of the disks failed. " + dirsHandler.getDisksHealthReport(false)); } List<String> containerLocalDirs = new ArrayList<>(localDirs.size()); try { List<Path> appDirs = new ArrayList<Path>(localDirs.size()); for (String localDir : localDirs) { Path usersdir = new Path(localDir, ContainerLocalizer.USERCACHE); Path userdir = new Path(usersdir, user); Path appsdir = new Path(userdir, ContainerLocalizer.APPCACHE); appDirs.add(new Path(appsdir, appIdStr)); String containerLocalDir = localDir + Path.SEPARATOR + ContainerLocalizer.USERCACHE + Path.SEPARATOR + user + Path.SEPARATOR + ContainerLocalizer.APPCACHE + Path.SEPARATOR + appIdStr + Path.SEPARATOR; containerLocalDirs.add(containerLocalDir); } containerScriptOutStream = lfs.create(nmPrivateContainerScriptPath, EnumSet.of(CREATE, OVERWRITE)); environment.put( ApplicationConstants.CONTAINER_TOKEN_FILE_ENV_NAME, new Path(containerWorkDir, FINAL_CONTAINER_TOKENS_FILE).toUri().getPath()); sanitizeEnv(environment, containerWorkDir, appDirs, userLocalDirs, containerLogDirs, localResources, nmPrivateClasspathJarDir); exec.writeLaunchEnv(containerScriptOutStream, environment, localResources, launchContext.getCommands()); tokensOutStream = lfs.create(nmPrivateTokensPath, EnumSet.of(CREATE, OVERWRITE)); Credentials creds = container.getCredentials(); creds.writeTokenStorageToStream(tokensOutStream); } finally { IOUtils.cleanup(LOG, containerScriptOutStream, tokensOutStream); } dispatcher.getEventHandler().handle(new ContainerEvent( containerID, ContainerEventType.CONTAINER_LAUNCHED)); context.getNMStateStore().storeContainerLaunched(containerID); if (!shouldLaunchContainer.compareAndSet(false, true)) { LOG.info("Container " + containerIdStr + " not launched as " + "cleanup already called"); ret = ExitCode.TERMINATED.getExitCode(); } else { ContainerStartContext csc = new ContainerStartContext.Builder() .setContainer(container) .setLocalizedResources(localResources) .setNmPrivateContainerScriptPath(nmPrivateContainerScriptPath) .setNmPrivateTokensPath(nmPrivateTokensPath) .setUser(user) .setAppId(appIdStr) .setContainerWorkDir(containerWorkDir) .setLocalDirs(localDirs) .setLogDirs(logDirs) .setFilecacheDirs(filecacheDirs) .setUserLocalDirs(userLocalDirs) .setContainerLocalDirs(containerLocalDirs) .setContainerLogDirs(containerLogDirs) .build(); ret = exec.prepareContainer(csc); if (ret == 0) { exec.activateContainer(containerID, pidFilePath); ret = exec.launchContainer(csc); } else { LOG.warn("Failed to prepare contaner." + ret); return ret; } } } catch (Throwable e) { LOG.warn("Failed to launch container.", e); dispatcher.getEventHandler().handle(new ContainerExitEvent( containerID, ContainerEventType.CONTAINER_EXITED_WITH_FAILURE, ret, e.getMessage())); return ret; } 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()) { LOG.debug("Container " + containerIdStr + " completed with exit code " + ret); } StringBuilder diagnosticInfo = new StringBuilder("Container exited with a non-zero exit code "); diagnosticInfo.append(ret); diagnosticInfo.append(". "); if (ret == ExitCode.FORCE_KILLED.getExitCode() || ret == ExitCode.TERMINATED.getExitCode()) { dispatcher.getEventHandler().handle( new ContainerExitEvent(containerID, ContainerEventType.CONTAINER_KILLED_ON_REQUEST, ret, diagnosticInfo.toString())); return ret; } if (ret != 0) { handleContainerExitWithFailure(containerID, ret, containerLogDir, diagnosticInfo); return ret; } LOG.info("Container " + containerIdStr + " succeeded "); dispatcher.getEventHandler().handle( new ContainerEvent(containerID, ContainerEventType.CONTAINER_EXITED_WITH_SUCCESS)); return 0; } ContainerLaunch(Context context, Configuration configuration,
      Dispatcher dispatcher, ContainerExecutor exec, Application app,
      Container container, LocalDirsHandlerService dirsHandler,
      ContainerManagerImpl containerManager); @VisibleForTesting static String expandEnvironment(String var,
      Path containerLogDir); @Override @SuppressWarnings("unchecked") // dispatcher not typed Integer call(); @SuppressWarnings("unchecked") // dispatcher not typed void cleanupContainer(); static String getRelativeContainerLogDir(String appIdStr,
      String containerIdStr); void sanitizeEnv(Map<String, String> environment, Path pwd,
      List<Path> appDirs, List<String> userLocalDirs, List<String>
      containerLogDirs,
      Map<Path, List<String>> resources,
      Path nmPrivateClasspathJarDir); static String getExitCodeFile(String pidFile); }
ContainerLaunch implements Callable<Integer> { @Override @SuppressWarnings("unchecked") public Integer call() { final ContainerLaunchContext launchContext = container.getLaunchContext(); Map<Path,List<String>> localResources = null; ContainerId containerID = container.getContainerId(); String containerIdStr = ConverterUtils.toString(containerID); final List<String> command = launchContext.getCommands(); int ret = -1; if (container.getContainerState() == ContainerState.KILLING) { dispatcher.getEventHandler().handle( new ContainerExitEvent(containerID, ContainerEventType.CONTAINER_KILLED_ON_REQUEST, Shell.WINDOWS ? ExitCode.FORCE_KILLED.getExitCode() : ExitCode.TERMINATED.getExitCode(), "Container terminated before launch.")); return 0; } Path containerLogDir; try { localResources = container.getLocalizedResources(); if (localResources == null) { throw RPCUtil.getRemoteException( "Unable to get local resources when Container " + containerID + " is at " + container.getContainerState()); } localResources = olr.filterLocalResources(localResources); final String user = container.getUser(); List<String> newCmds = new ArrayList<String>(command.size()); String appIdStr = app.getAppId().toString(); String relativeContainerLogDir = ContainerLaunch .getRelativeContainerLogDir(appIdStr, containerIdStr); containerLogDir = dirsHandler.getLogPathForWrite(relativeContainerLogDir, false); for (String str : command) { newCmds.add(expandEnvironment(str, containerLogDir)); } launchContext.setCommands(newCmds); Map<String, String> environment = launchContext.getEnvironment(); for (Entry<String, String> entry : environment.entrySet()) { String value = entry.getValue(); value = expandEnvironment(value, containerLogDir); entry.setValue(value); } FileContext lfs = FileContext.getLocalFSFileContext(); Path nmPrivateContainerScriptPath = dirsHandler.getLocalPathForWrite( getContainerPrivateDir(appIdStr, containerIdStr) + Path.SEPARATOR + CONTAINER_SCRIPT); Path nmPrivateTokensPath = dirsHandler.getLocalPathForWrite( getContainerPrivateDir(appIdStr, containerIdStr) + Path.SEPARATOR + String.format(ContainerLocalizer.TOKEN_FILE_NAME_FMT, containerIdStr)); Path nmPrivateClasspathJarDir = dirsHandler.getLocalPathForWrite( getContainerPrivateDir(appIdStr, containerIdStr)); DataOutputStream containerScriptOutStream = null; DataOutputStream tokensOutStream = null; Path containerWorkDir = dirsHandler.getLocalPathForWrite(ContainerLocalizer.USERCACHE + Path.SEPARATOR + user + Path.SEPARATOR + ContainerLocalizer.APPCACHE + Path.SEPARATOR + appIdStr + Path.SEPARATOR + containerIdStr, LocalDirAllocator.SIZE_UNKNOWN, false); String pidFileSubpath = getPidFileSubpath(appIdStr, containerIdStr); pidFilePath = dirsHandler.getLocalPathForWrite(pidFileSubpath); List<String> localDirs = dirsHandler.getLocalDirs(); List<String> logDirs = dirsHandler.getLogDirs(); List<String> filecacheDirs = getNMFilecacheDirs(localDirs); List<String> userLocalDirs = getUserLocalDirs(localDirs); List<String> containerLogDirs = new ArrayList<String>(); for( String logDir : logDirs) { containerLogDirs.add(logDir + Path.SEPARATOR + relativeContainerLogDir); } if (!dirsHandler.areDisksHealthy()) { ret = ContainerExitStatus.DISKS_FAILED; throw new IOException("Most of the disks failed. " + dirsHandler.getDisksHealthReport(false)); } List<String> containerLocalDirs = new ArrayList<>(localDirs.size()); try { List<Path> appDirs = new ArrayList<Path>(localDirs.size()); for (String localDir : localDirs) { Path usersdir = new Path(localDir, ContainerLocalizer.USERCACHE); Path userdir = new Path(usersdir, user); Path appsdir = new Path(userdir, ContainerLocalizer.APPCACHE); appDirs.add(new Path(appsdir, appIdStr)); String containerLocalDir = localDir + Path.SEPARATOR + ContainerLocalizer.USERCACHE + Path.SEPARATOR + user + Path.SEPARATOR + ContainerLocalizer.APPCACHE + Path.SEPARATOR + appIdStr + Path.SEPARATOR; containerLocalDirs.add(containerLocalDir); } containerScriptOutStream = lfs.create(nmPrivateContainerScriptPath, EnumSet.of(CREATE, OVERWRITE)); environment.put( ApplicationConstants.CONTAINER_TOKEN_FILE_ENV_NAME, new Path(containerWorkDir, FINAL_CONTAINER_TOKENS_FILE).toUri().getPath()); sanitizeEnv(environment, containerWorkDir, appDirs, userLocalDirs, containerLogDirs, localResources, nmPrivateClasspathJarDir); exec.writeLaunchEnv(containerScriptOutStream, environment, localResources, launchContext.getCommands()); tokensOutStream = lfs.create(nmPrivateTokensPath, EnumSet.of(CREATE, OVERWRITE)); Credentials creds = container.getCredentials(); creds.writeTokenStorageToStream(tokensOutStream); } finally { IOUtils.cleanup(LOG, containerScriptOutStream, tokensOutStream); } dispatcher.getEventHandler().handle(new ContainerEvent( containerID, ContainerEventType.CONTAINER_LAUNCHED)); context.getNMStateStore().storeContainerLaunched(containerID); if (!shouldLaunchContainer.compareAndSet(false, true)) { LOG.info("Container " + containerIdStr + " not launched as " + "cleanup already called"); ret = ExitCode.TERMINATED.getExitCode(); } else { ContainerStartContext csc = new ContainerStartContext.Builder() .setContainer(container) .setLocalizedResources(localResources) .setNmPrivateContainerScriptPath(nmPrivateContainerScriptPath) .setNmPrivateTokensPath(nmPrivateTokensPath) .setUser(user) .setAppId(appIdStr) .setContainerWorkDir(containerWorkDir) .setLocalDirs(localDirs) .setLogDirs(logDirs) .setFilecacheDirs(filecacheDirs) .setUserLocalDirs(userLocalDirs) .setContainerLocalDirs(containerLocalDirs) .setContainerLogDirs(containerLogDirs) .build(); ret = exec.prepareContainer(csc); if (ret == 0) { exec.activateContainer(containerID, pidFilePath); ret = exec.launchContainer(csc); } else { LOG.warn("Failed to prepare contaner." + ret); return ret; } } } catch (Throwable e) { LOG.warn("Failed to launch container.", e); dispatcher.getEventHandler().handle(new ContainerExitEvent( containerID, ContainerEventType.CONTAINER_EXITED_WITH_FAILURE, ret, e.getMessage())); return ret; } 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()) { LOG.debug("Container " + containerIdStr + " completed with exit code " + ret); } StringBuilder diagnosticInfo = new StringBuilder("Container exited with a non-zero exit code "); diagnosticInfo.append(ret); diagnosticInfo.append(". "); if (ret == ExitCode.FORCE_KILLED.getExitCode() || ret == ExitCode.TERMINATED.getExitCode()) { dispatcher.getEventHandler().handle( new ContainerExitEvent(containerID, ContainerEventType.CONTAINER_KILLED_ON_REQUEST, ret, diagnosticInfo.toString())); return ret; } if (ret != 0) { handleContainerExitWithFailure(containerID, ret, containerLogDir, diagnosticInfo); return ret; } LOG.info("Container " + containerIdStr + " succeeded "); dispatcher.getEventHandler().handle( new ContainerEvent(containerID, ContainerEventType.CONTAINER_EXITED_WITH_SUCCESS)); return 0; } ContainerLaunch(Context context, Configuration configuration,
      Dispatcher dispatcher, ContainerExecutor exec, Application app,
      Container container, LocalDirsHandlerService dirsHandler,
      ContainerManagerImpl containerManager); @VisibleForTesting static String expandEnvironment(String var,
      Path containerLogDir); @Override @SuppressWarnings("unchecked") // dispatcher not typed Integer call(); @SuppressWarnings("unchecked") // dispatcher not typed void cleanupContainer(); static String getRelativeContainerLogDir(String appIdStr,
      String containerIdStr); void sanitizeEnv(Map<String, String> environment, Path pwd,
      List<Path> appDirs, List<String> userLocalDirs, List<String>
      containerLogDirs,
      Map<Path, List<String>> resources,
      Path nmPrivateClasspathJarDir); static String getExitCodeFile(String pidFile); static final String CONTAINER_SCRIPT; static final String FINAL_CONTAINER_TOKENS_FILE; }