@Test(timeout = 30000) public void testStartLocalizer() throws IOException, InterruptedException { InetSocketAddress localizationServerAddress; final Path firstDir = new Path(BASE_TMP_PATH, "localDir1"); List<String> localDirs = new ArrayList<String>(); final Path secondDir = new Path(BASE_TMP_PATH, "localDir2"); List<String> logDirs = new ArrayList<String>(); final Path logDir = new Path(BASE_TMP_PATH, "logDir"); final Path tokenDir = new Path(BASE_TMP_PATH, "tokenDir"); FsPermission perms = new FsPermission((short)0770); Configuration conf = new Configuration(); localizationServerAddress = conf.getSocketAddr( YarnConfiguration.NM_BIND_HOST, YarnConfiguration.NM_LOCALIZER_ADDRESS, YarnConfiguration.DEFAULT_NM_LOCALIZER_ADDRESS, YarnConfiguration.DEFAULT_NM_LOCALIZER_PORT); final FileContext mockLfs = spy(FileContext.getLocalFSFileContext(conf)); final FileContext.Util mockUtil = spy(mockLfs.util()); doAnswer(new Answer() { @Override public Object answer(InvocationOnMock invocationOnMock) throws Throwable { return mockUtil; } }).when(mockLfs).util(); doAnswer(new Answer() { @Override public Object answer(InvocationOnMock invocationOnMock) throws Throwable { Path dest = (Path) invocationOnMock.getArguments()[1]; if (dest.toString().contains(firstDir.toString())) { throw new IOException("No space on this drive " + dest.toString()); } else { DataOutputStream tokenOut = null; try { Credentials credentials = new Credentials(); tokenOut = mockLfs.create(dest, EnumSet.of(CREATE, OVERWRITE)); credentials.writeTokenStorageToStream(tokenOut); } finally { if (tokenOut != null) { tokenOut.close(); } } } return null; } }).when(mockUtil).copy(any(Path.class), any(Path.class)); doAnswer(new Answer() { @Override public Object answer(InvocationOnMock invocationOnMock) throws Throwable { Path p = (Path) invocationOnMock.getArguments()[0]; if (p.toString().contains(firstDir.toString())) { return new FsStatus(2000, 2000, 0); } else { return new FsStatus(1000, 0, 1000); } } }).when(mockLfs).getFsStatus(any(Path.class)); DefaultContainerExecutor mockExec = spy(new DefaultContainerExecutor( mockLfs)); mockExec.setConf(conf); localDirs.add(mockLfs.makeQualified(firstDir).toString()); localDirs.add(mockLfs.makeQualified(secondDir).toString()); logDirs.add(mockLfs.makeQualified(logDir).toString()); conf.setStrings(YarnConfiguration.NM_LOCAL_DIRS, localDirs.toArray(new String[localDirs.size()])); conf.set(YarnConfiguration.NM_LOG_DIRS, logDir.toString()); mockLfs.mkdir(tokenDir, perms, true); Path nmPrivateCTokensPath = new Path(tokenDir, "test.tokens"); String appSubmitter = "nobody"; String appId = "APP_ID"; String locId = "LOC_ID"; LocalDirsHandlerService dirsHandler = mock(LocalDirsHandlerService.class); when(dirsHandler.getLocalDirs()).thenReturn(localDirs); when(dirsHandler.getLogDirs()).thenReturn(logDirs); try { mockExec.startLocalizer(new LocalizerStartContext.Builder() .setNmPrivateContainerTokens(nmPrivateCTokensPath) .setNmAddr(localizationServerAddress) .setUser(appSubmitter) .setAppId(appId) .setLocId(locId) .setDirsHandler(dirsHandler) .build()); } catch (IOException e) { Assert.fail("StartLocalizer failed to copy token file " + e); } finally { mockExec.deleteAsUser(new DeletionAsUserContext.Builder() .setUser(appSubmitter) .setSubDir(firstDir) .build()); mockExec.deleteAsUser(new DeletionAsUserContext.Builder() .setUser(appSubmitter) .setSubDir(secondDir) .build()); mockExec.deleteAsUser(new DeletionAsUserContext.Builder() .setUser(appSubmitter) .setSubDir(logDir) .build()); deleteTmpFiles(); } }
@Override public void startLocalizer(LocalizerStartContext ctx) throws IOException, InterruptedException { Path nmPrivateContainerTokensPath = ctx.getNmPrivateContainerTokens(); InetSocketAddress nmAddr = ctx.getNmAddr(); String user = ctx.getUser(); String appId = ctx.getAppId(); String locId = ctx.getLocId(); LocalDirsHandlerService dirsHandler = ctx.getDirsHandler(); List<String> localDirs = dirsHandler.getLocalDirs(); List<String> logDirs = dirsHandler.getLogDirs(); createUserLocalDirs(localDirs, user); createUserCacheDirs(localDirs, user); createAppDirs(localDirs, user, appId); createAppLogDirs(appId, logDirs, user); Path appStorageDir = getWorkingDir(localDirs, user, appId); String tokenFn = String.format(ContainerLocalizer.TOKEN_FILE_NAME_FMT, locId); Path tokenDst = new Path(appStorageDir, tokenFn); copyFile(nmPrivateContainerTokensPath, tokenDst, user); LOG.info("Copying from " + nmPrivateContainerTokensPath + " to " + tokenDst); FileContext localizerFc = FileContext.getFileContext( lfs.getDefaultFileSystem(), getConf()); localizerFc.setUMask(lfs.getUMask()); localizerFc.setWorkingDirectory(appStorageDir); LOG.info("Localizer CWD set to " + appStorageDir + " = " + localizerFc.getWorkingDirectory()); ContainerLocalizer localizer = new ContainerLocalizer(localizerFc, user, appId, locId, getPaths(localDirs), RecordFactoryProvider.getRecordFactory(getConf())); localizer.runLocalization(nmAddr); }
DefaultContainerExecutor extends ContainerExecutor { @Override public void startLocalizer(LocalizerStartContext ctx) throws IOException, InterruptedException { Path nmPrivateContainerTokensPath = ctx.getNmPrivateContainerTokens(); InetSocketAddress nmAddr = ctx.getNmAddr(); String user = ctx.getUser(); String appId = ctx.getAppId(); String locId = ctx.getLocId(); LocalDirsHandlerService dirsHandler = ctx.getDirsHandler(); List<String> localDirs = dirsHandler.getLocalDirs(); List<String> logDirs = dirsHandler.getLogDirs(); createUserLocalDirs(localDirs, user); createUserCacheDirs(localDirs, user); createAppDirs(localDirs, user, appId); createAppLogDirs(appId, logDirs, user); Path appStorageDir = getWorkingDir(localDirs, user, appId); String tokenFn = String.format(ContainerLocalizer.TOKEN_FILE_NAME_FMT, locId); Path tokenDst = new Path(appStorageDir, tokenFn); copyFile(nmPrivateContainerTokensPath, tokenDst, user); LOG.info("Copying from " + nmPrivateContainerTokensPath + " to " + tokenDst); FileContext localizerFc = FileContext.getFileContext( lfs.getDefaultFileSystem(), getConf()); localizerFc.setUMask(lfs.getUMask()); localizerFc.setWorkingDirectory(appStorageDir); LOG.info("Localizer CWD set to " + appStorageDir + " = " + localizerFc.getWorkingDirectory()); ContainerLocalizer localizer = new ContainerLocalizer(localizerFc, user, appId, locId, getPaths(localDirs), RecordFactoryProvider.getRecordFactory(getConf())); localizer.runLocalization(nmAddr); } }
DefaultContainerExecutor extends ContainerExecutor { @Override public void startLocalizer(LocalizerStartContext ctx) throws IOException, InterruptedException { Path nmPrivateContainerTokensPath = ctx.getNmPrivateContainerTokens(); InetSocketAddress nmAddr = ctx.getNmAddr(); String user = ctx.getUser(); String appId = ctx.getAppId(); String locId = ctx.getLocId(); LocalDirsHandlerService dirsHandler = ctx.getDirsHandler(); List<String> localDirs = dirsHandler.getLocalDirs(); List<String> logDirs = dirsHandler.getLogDirs(); createUserLocalDirs(localDirs, user); createUserCacheDirs(localDirs, user); createAppDirs(localDirs, user, appId); createAppLogDirs(appId, logDirs, user); Path appStorageDir = getWorkingDir(localDirs, user, appId); String tokenFn = String.format(ContainerLocalizer.TOKEN_FILE_NAME_FMT, locId); Path tokenDst = new Path(appStorageDir, tokenFn); copyFile(nmPrivateContainerTokensPath, tokenDst, user); LOG.info("Copying from " + nmPrivateContainerTokensPath + " to " + tokenDst); FileContext localizerFc = FileContext.getFileContext( lfs.getDefaultFileSystem(), getConf()); localizerFc.setUMask(lfs.getUMask()); localizerFc.setWorkingDirectory(appStorageDir); LOG.info("Localizer CWD set to " + appStorageDir + " = " + localizerFc.getWorkingDirectory()); ContainerLocalizer localizer = new ContainerLocalizer(localizerFc, user, appId, locId, getPaths(localDirs), RecordFactoryProvider.getRecordFactory(getConf())); localizer.runLocalization(nmAddr); } DefaultContainerExecutor();   DefaultContainerExecutor(FileContext lfs); }
DefaultContainerExecutor extends ContainerExecutor { @Override public void startLocalizer(LocalizerStartContext ctx) throws IOException, InterruptedException { Path nmPrivateContainerTokensPath = ctx.getNmPrivateContainerTokens(); InetSocketAddress nmAddr = ctx.getNmAddr(); String user = ctx.getUser(); String appId = ctx.getAppId(); String locId = ctx.getLocId(); LocalDirsHandlerService dirsHandler = ctx.getDirsHandler(); List<String> localDirs = dirsHandler.getLocalDirs(); List<String> logDirs = dirsHandler.getLogDirs(); createUserLocalDirs(localDirs, user); createUserCacheDirs(localDirs, user); createAppDirs(localDirs, user, appId); createAppLogDirs(appId, logDirs, user); Path appStorageDir = getWorkingDir(localDirs, user, appId); String tokenFn = String.format(ContainerLocalizer.TOKEN_FILE_NAME_FMT, locId); Path tokenDst = new Path(appStorageDir, tokenFn); copyFile(nmPrivateContainerTokensPath, tokenDst, user); LOG.info("Copying from " + nmPrivateContainerTokensPath + " to " + tokenDst); FileContext localizerFc = FileContext.getFileContext( lfs.getDefaultFileSystem(), getConf()); localizerFc.setUMask(lfs.getUMask()); localizerFc.setWorkingDirectory(appStorageDir); LOG.info("Localizer CWD set to " + appStorageDir + " = " + localizerFc.getWorkingDirectory()); ContainerLocalizer localizer = new ContainerLocalizer(localizerFc, user, appId, locId, getPaths(localDirs), RecordFactoryProvider.getRecordFactory(getConf())); localizer.runLocalization(nmAddr); } DefaultContainerExecutor();   DefaultContainerExecutor(FileContext lfs); @Override void init(); @Override void startLocalizer(LocalizerStartContext ctx); @Override int prepareContainer(ContainerStartContext ctx); @Override int launchContainer(ContainerStartContext ctx); @Override boolean signalContainer(ContainerSignalContext ctx); @Override boolean isContainerAlive(ContainerLivenessContext ctx); @VisibleForTesting static boolean containerIsAlive(String pid); @Override void deleteAsUser(DeletionAsUserContext ctx); }
DefaultContainerExecutor extends ContainerExecutor { @Override public void startLocalizer(LocalizerStartContext ctx) throws IOException, InterruptedException { Path nmPrivateContainerTokensPath = ctx.getNmPrivateContainerTokens(); InetSocketAddress nmAddr = ctx.getNmAddr(); String user = ctx.getUser(); String appId = ctx.getAppId(); String locId = ctx.getLocId(); LocalDirsHandlerService dirsHandler = ctx.getDirsHandler(); List<String> localDirs = dirsHandler.getLocalDirs(); List<String> logDirs = dirsHandler.getLogDirs(); createUserLocalDirs(localDirs, user); createUserCacheDirs(localDirs, user); createAppDirs(localDirs, user, appId); createAppLogDirs(appId, logDirs, user); Path appStorageDir = getWorkingDir(localDirs, user, appId); String tokenFn = String.format(ContainerLocalizer.TOKEN_FILE_NAME_FMT, locId); Path tokenDst = new Path(appStorageDir, tokenFn); copyFile(nmPrivateContainerTokensPath, tokenDst, user); LOG.info("Copying from " + nmPrivateContainerTokensPath + " to " + tokenDst); FileContext localizerFc = FileContext.getFileContext( lfs.getDefaultFileSystem(), getConf()); localizerFc.setUMask(lfs.getUMask()); localizerFc.setWorkingDirectory(appStorageDir); LOG.info("Localizer CWD set to " + appStorageDir + " = " + localizerFc.getWorkingDirectory()); ContainerLocalizer localizer = new ContainerLocalizer(localizerFc, user, appId, locId, getPaths(localDirs), RecordFactoryProvider.getRecordFactory(getConf())); localizer.runLocalization(nmAddr); } DefaultContainerExecutor();   DefaultContainerExecutor(FileContext lfs); @Override void init(); @Override void startLocalizer(LocalizerStartContext ctx); @Override int prepareContainer(ContainerStartContext ctx); @Override int launchContainer(ContainerStartContext ctx); @Override boolean signalContainer(ContainerSignalContext ctx); @Override boolean isContainerAlive(ContainerLivenessContext ctx); @VisibleForTesting static boolean containerIsAlive(String pid); @Override void deleteAsUser(DeletionAsUserContext ctx);  }