ResourceLocalizationService extends CompositeService implements EventHandler<LocalizationEvent>, LocalizationProtocol { FileContext getLocalFileContext(Configuration conf) { try { return FileContext.getLocalFSFileContext(conf); } catch (IOException e) { throw new YarnRuntimeException("Failed to access local fs"); } } ResourceLocalizationService(Dispatcher dispatcher,
      ContainerExecutor exec, DeletionService delService,
      LocalDirsHandlerService dirsHandler, Context context,
      NodeManagerMetrics metrics); @Override void serviceInit(Configuration conf); void recoverLocalizedResources(RecoveredLocalizationState state); @Override LocalizerHeartbeatResponse heartbeat(LocalizerStatus status); @Override void serviceStart(); @Override void serviceStop(); @Override void handle(LocalizationEvent event); @VisibleForTesting @Private PublicLocalizer getPublicLocalizer(); @VisibleForTesting @Private LocalizerRunner getLocalizerRunner(String locId); @VisibleForTesting @Private Map<String, LocalizerRunner> getPrivateLocalizers(); static final String NM_PRIVATE_DIR; static final FsPermission NM_PRIVATE_PERM; }
@Test public void testLocalizationInit() throws Exception { conf.set(CommonConfigurationKeys.FS_PERMISSIONS_UMASK_KEY, "077"); AsyncDispatcher dispatcher = new AsyncDispatcher(); dispatcher.init(new Configuration()); ContainerExecutor exec = mock(ContainerExecutor.class); DeletionService delService = spy(new DeletionService(exec)); delService.init(conf); delService.start(); List<Path> localDirs = new ArrayList<Path>(); String[] sDirs = new String[4]; for (int i = 0; i < 4; ++i) { localDirs.add(lfs.makeQualified(new Path(basedir, i + ""))); sDirs[i] = localDirs.get(i).toString(); } conf.setStrings(YarnConfiguration.NM_LOCAL_DIRS, sDirs); LocalDirsHandlerService diskhandler = new LocalDirsHandlerService(); diskhandler.init(conf); ResourceLocalizationService locService = spy(new ResourceLocalizationService(dispatcher, exec, delService, diskhandler, nmContext, metrics)); doReturn(lfs) .when(locService).getLocalFileContext(isA(Configuration.class)); try { dispatcher.start(); locService.init(conf); final FsPermission defaultPerm = new FsPermission((short)0755); for (Path p : localDirs) { p = new Path((new URI(p.toString())).getPath()); Path usercache = new Path(p, ContainerLocalizer.USERCACHE); verify(spylfs) .mkdir(eq(usercache), eq(defaultPerm), eq(true)); Path publicCache = new Path(p, ContainerLocalizer.FILECACHE); verify(spylfs) .mkdir(eq(publicCache), eq(defaultPerm), eq(true)); Path nmPriv = new Path(p, ResourceLocalizationService.NM_PRIVATE_DIR); verify(spylfs).mkdir(eq(nmPriv), eq(ResourceLocalizationService.NM_PRIVATE_PERM), eq(true)); } } finally { dispatcher.stop(); delService.stop(); } }
@Test public void testDirectoryCleanupOnNewlyCreatedStateStore() throws IOException, URISyntaxException { conf.set(CommonConfigurationKeys.FS_PERMISSIONS_UMASK_KEY, "077"); AsyncDispatcher dispatcher = new AsyncDispatcher(); dispatcher.init(new Configuration()); ContainerExecutor exec = mock(ContainerExecutor.class); DeletionService delService = spy(new DeletionService(exec)); delService.init(conf); delService.start(); List<Path> localDirs = new ArrayList<Path>(); String[] sDirs = new String[4]; for (int i = 0; i < 4; ++i) { localDirs.add(lfs.makeQualified(new Path(basedir, i + ""))); sDirs[i] = localDirs.get(i).toString(); } conf.setStrings(YarnConfiguration.NM_LOCAL_DIRS, sDirs); LocalDirsHandlerService diskhandler = new LocalDirsHandlerService(); diskhandler.init(conf); NMStateStoreService nmStateStoreService = mock(NMStateStoreService.class); when(nmStateStoreService.canRecover()).thenReturn(true); when(nmStateStoreService.isNewlyCreated()).thenReturn(true); ResourceLocalizationService locService = spy(new ResourceLocalizationService(dispatcher, exec, delService, diskhandler, nmContext, metrics)); doReturn(lfs) .when(locService).getLocalFileContext(isA(Configuration.class)); try { dispatcher.start(); locService.init(conf); final FsPermission defaultPerm = new FsPermission((short)0755); for (Path p : localDirs) { p = new Path((new URI(p.toString())).getPath()); Path usercache = new Path(p, ContainerLocalizer.USERCACHE); verify(spylfs) .rename(eq(usercache), any(Path.class), any(Options.Rename.class)); verify(spylfs) .mkdir(eq(usercache), eq(defaultPerm), eq(true)); Path publicCache = new Path(p, ContainerLocalizer.FILECACHE); verify(spylfs) .rename(eq(usercache), any(Path.class), any(Options.Rename.class)); verify(spylfs) .mkdir(eq(publicCache), eq(defaultPerm), eq(true)); Path nmPriv = new Path(p, ResourceLocalizationService.NM_PRIVATE_DIR); verify(spylfs) .rename(eq(usercache), any(Path.class), any(Options.Rename.class)); verify(spylfs).mkdir(eq(nmPriv), eq(ResourceLocalizationService.NM_PRIVATE_PERM), eq(true)); } } finally { dispatcher.stop(); delService.stop(); } }