@Test public void testAppLogDirCreation() throws Exception { final String logSuffix = "logs"; this.conf.set(YarnConfiguration.NM_LOG_DIRS, localLogDir.getAbsolutePath()); this.conf.set(YarnConfiguration.NM_REMOTE_APP_LOG_DIR, this.remoteRootLogDir.getAbsolutePath()); this.conf.set(YarnConfiguration.NM_REMOTE_APP_LOG_DIR_SUFFIX, logSuffix); InlineDispatcher dispatcher = new InlineDispatcher(); dispatcher.init(this.conf); dispatcher.start(); FileSystem fs = FileSystem.get(this.conf); final FileSystem spyFs = spy(FileSystem.get(this.conf)); LogAggregationService aggSvc = new LogAggregationService(dispatcher, this.context, this.delSrvc, super.dirsHandler) { @Override protected FileSystem getFileSystem(Configuration conf) { return spyFs; } }; aggSvc.init(this.conf); aggSvc.start(); ApplicationId appId = BuilderUtils.newApplicationId(1, 1); Path userDir = fs.makeQualified(new Path( remoteRootLogDir.getAbsolutePath(), this.user)); Path suffixDir = new Path(userDir, logSuffix); Path appDir = new Path(suffixDir, appId.toString()); aggSvc.handle(new LogHandlerAppStartedEvent(appId, this.user, null, ContainerLogsRetentionPolicy.ALL_CONTAINERS, this.acls)); verify(spyFs).mkdirs(eq(userDir), isA(FsPermission.class)); verify(spyFs).mkdirs(eq(suffixDir), isA(FsPermission.class)); verify(spyFs).mkdirs(eq(appDir), isA(FsPermission.class)); ApplicationId appId2 = BuilderUtils.newApplicationId(1, 2); Path appDir2 = new Path(suffixDir, appId2.toString()); aggSvc.handle(new LogHandlerAppStartedEvent(appId2, this.user, null, ContainerLogsRetentionPolicy.ALL_CONTAINERS, this.acls)); verify(spyFs).mkdirs(eq(appDir2), isA(FsPermission.class)); ApplicationId appId3 = BuilderUtils.newApplicationId(1, 3); Path appDir3 = new Path(suffixDir, appId3.toString()); new File(appDir3.toUri().getPath()).mkdir(); aggSvc.handle(new LogHandlerAppStartedEvent(appId3, this.user, null, ContainerLogsRetentionPolicy.ALL_CONTAINERS, this.acls)); verify(spyFs, never()).mkdirs(eq(appDir3), isA(FsPermission.class)); aggSvc.stop(); aggSvc.close(); }
@Override public void handle(LogHandlerEvent event) { switch (event.getType()) { case APPLICATION_STARTED: LogHandlerAppStartedEvent appStartEvent = (LogHandlerAppStartedEvent) event; initApp(appStartEvent.getApplicationId(), appStartEvent.getUser(), appStartEvent.getCredentials(), appStartEvent.getLogRetentionPolicy(), appStartEvent.getApplicationAcls(), appStartEvent.getLogAggregationContext()); break; case CONTAINER_FINISHED: LogHandlerContainerFinishedEvent containerFinishEvent = (LogHandlerContainerFinishedEvent) event; stopContainer(containerFinishEvent.getContainerId(), containerFinishEvent.getExitCode()); break; case APPLICATION_FINISHED: LogHandlerAppFinishedEvent appFinishedEvent = (LogHandlerAppFinishedEvent) event; stopApp(appFinishedEvent.getApplicationId()); break; default: ; } }
LogAggregationService extends AbstractService implements
    LogHandler { @Override public void handle(LogHandlerEvent event) { switch (event.getType()) { case APPLICATION_STARTED: LogHandlerAppStartedEvent appStartEvent = (LogHandlerAppStartedEvent) event; initApp(appStartEvent.getApplicationId(), appStartEvent.getUser(), appStartEvent.getCredentials(), appStartEvent.getLogRetentionPolicy(), appStartEvent.getApplicationAcls(), appStartEvent.getLogAggregationContext()); break; case CONTAINER_FINISHED: LogHandlerContainerFinishedEvent containerFinishEvent = (LogHandlerContainerFinishedEvent) event; stopContainer(containerFinishEvent.getContainerId(), containerFinishEvent.getExitCode()); break; case APPLICATION_FINISHED: LogHandlerAppFinishedEvent appFinishedEvent = (LogHandlerAppFinishedEvent) event; stopApp(appFinishedEvent.getApplicationId()); break; default: ; } } }
LogAggregationService extends AbstractService implements
    LogHandler { @Override public void handle(LogHandlerEvent event) { switch (event.getType()) { case APPLICATION_STARTED: LogHandlerAppStartedEvent appStartEvent = (LogHandlerAppStartedEvent) event; initApp(appStartEvent.getApplicationId(), appStartEvent.getUser(), appStartEvent.getCredentials(), appStartEvent.getLogRetentionPolicy(), appStartEvent.getApplicationAcls(), appStartEvent.getLogAggregationContext()); break; case CONTAINER_FINISHED: LogHandlerContainerFinishedEvent containerFinishEvent = (LogHandlerContainerFinishedEvent) event; stopContainer(containerFinishEvent.getContainerId(), containerFinishEvent.getExitCode()); break; case APPLICATION_FINISHED: LogHandlerAppFinishedEvent appFinishedEvent = (LogHandlerAppFinishedEvent) event; stopApp(appFinishedEvent.getApplicationId()); break; default: ; } } LogAggregationService(Dispatcher dispatcher, Context context,
      DeletionService deletionService, LocalDirsHandlerService dirsHandler); }
LogAggregationService extends AbstractService implements
    LogHandler { @Override public void handle(LogHandlerEvent event) { switch (event.getType()) { case APPLICATION_STARTED: LogHandlerAppStartedEvent appStartEvent = (LogHandlerAppStartedEvent) event; initApp(appStartEvent.getApplicationId(), appStartEvent.getUser(), appStartEvent.getCredentials(), appStartEvent.getLogRetentionPolicy(), appStartEvent.getApplicationAcls(), appStartEvent.getLogAggregationContext()); break; case CONTAINER_FINISHED: LogHandlerContainerFinishedEvent containerFinishEvent = (LogHandlerContainerFinishedEvent) event; stopContainer(containerFinishEvent.getContainerId(), containerFinishEvent.getExitCode()); break; case APPLICATION_FINISHED: LogHandlerAppFinishedEvent appFinishedEvent = (LogHandlerAppFinishedEvent) event; stopApp(appFinishedEvent.getApplicationId()); break; default: ; } } LogAggregationService(Dispatcher dispatcher, Context context,
      DeletionService deletionService, LocalDirsHandlerService dirsHandler); @Override void handle(LogHandlerEvent event); @VisibleForTesting ConcurrentMap<ApplicationId, AppLogAggregator> getAppLogAggregators(); @VisibleForTesting NodeId getNodeId(); }
LogAggregationService extends AbstractService implements
    LogHandler { @Override public void handle(LogHandlerEvent event) { switch (event.getType()) { case APPLICATION_STARTED: LogHandlerAppStartedEvent appStartEvent = (LogHandlerAppStartedEvent) event; initApp(appStartEvent.getApplicationId(), appStartEvent.getUser(), appStartEvent.getCredentials(), appStartEvent.getLogRetentionPolicy(), appStartEvent.getApplicationAcls(), appStartEvent.getLogAggregationContext()); break; case CONTAINER_FINISHED: LogHandlerContainerFinishedEvent containerFinishEvent = (LogHandlerContainerFinishedEvent) event; stopContainer(containerFinishEvent.getContainerId(), containerFinishEvent.getExitCode()); break; case APPLICATION_FINISHED: LogHandlerAppFinishedEvent appFinishedEvent = (LogHandlerAppFinishedEvent) event; stopApp(appFinishedEvent.getApplicationId()); break; default: ; } } LogAggregationService(Dispatcher dispatcher, Context context,
      DeletionService deletionService, LocalDirsHandlerService dirsHandler); @Override void handle(LogHandlerEvent event); @VisibleForTesting ConcurrentMap<ApplicationId, AppLogAggregator> getAppLogAggregators(); @VisibleForTesting NodeId getNodeId();  }