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.getApplicationAcls(), appStartEvent.getLogAggregationContext(), appStartEvent.getRecoveredAppLogInitedTime()); break; case CONTAINER_FINISHED: LogHandlerContainerFinishedEvent containerFinishEvent = (LogHandlerContainerFinishedEvent) event; stopContainer(containerFinishEvent.getContainerId(), containerFinishEvent.getContainerType(), containerFinishEvent.getExitCode()); break; case APPLICATION_FINISHED: LogHandlerAppFinishedEvent appFinishedEvent = (LogHandlerAppFinishedEvent) event; stopApp(appFinishedEvent.getApplicationId()); break; case LOG_AGG_TOKEN_UPDATE: checkAndEnableAppAggregators(); break; default: ; } } LogAggregationService(Dispatcher dispatcher, Context context,
      DeletionService deletionService, LocalDirsHandlerService dirsHandler); @Override void handle(LogHandlerEvent event); @Override Set<ApplicationId> getInvalidTokenApps(); @VisibleForTesting ConcurrentMap<ApplicationId, AppLogAggregator> getAppLogAggregators(); @VisibleForTesting NodeId getNodeId(); @VisibleForTesting long getRollingMonitorInterval(); @VisibleForTesting LogAggregationFileController getLogAggregationFileController(
      Configuration conf);  }
@Test public void testNoLogsUploadedOnAppFinish() throws Exception { this.delSrvc = new DeletionService(createContainerExecutor()); delSrvc = spy(delSrvc); this.delSrvc.init(conf); this.conf.set(YarnConfiguration.NM_LOG_DIRS, localLogDir.getAbsolutePath()); this.conf.set(YarnConfiguration.NM_REMOTE_APP_LOG_DIR, this.remoteRootLogDir.getAbsolutePath()); LogAggregationService logAggregationService = new LogAggregationService( dispatcher, this.context, this.delSrvc, super.dirsHandler); logAggregationService.init(this.conf); logAggregationService.start(); ApplicationId app = BuilderUtils.newApplicationId(1234, 1); File appLogDir = new File(localLogDir, app.toString()); appLogDir.mkdir(); LogAggregationContext context = LogAggregationContext.newInstance("HOST*", "sys*"); logAggregationService.handle(new LogHandlerAppStartedEvent(app, this.user, null, this.acls, context)); ApplicationAttemptId appAttemptId = BuilderUtils.newApplicationAttemptId(app, 1); ContainerId cont = ContainerId.newContainerId(appAttemptId, 1); writeContainerLogs(appLogDir, cont, new String[] { "stdout", "stderr", "syslog" }, EMPTY_FILES); logAggregationService.handle(new LogHandlerContainerFinishedEvent(cont, ContainerType.APPLICATION_MASTER, 0)); logAggregationService.handle(new LogHandlerAppFinishedEvent(app)); logAggregationService.stop(); delSrvc.stop(); FileDeletionTask deletionTask = new FileDeletionTask(delSrvc, user, null, null); verify(delSrvc, times(0)).delete(deletionTask); }
@Test public void testVerifyAndCreateRemoteDirsFailure() throws Exception { this.conf.set(YarnConfiguration.NM_LOG_DIRS, localLogDir.getAbsolutePath()); this.conf.set(YarnConfiguration.NM_REMOTE_APP_LOG_DIR, this.remoteRootLogDir.getAbsolutePath()); LogAggregationFileControllerFactory factory = new LogAggregationFileControllerFactory(conf); LogAggregationFileController logAggregationFileFormat = factory .getFileControllerForWrite(); LogAggregationFileController spyLogAggregationFileFormat = spy(logAggregationFileFormat); YarnRuntimeException e = new YarnRuntimeException("KABOOM!"); doThrow(e).doNothing().when(spyLogAggregationFileFormat) .verifyAndCreateRemoteLogDir(); LogAggregationService logAggregationService = spy( new LogAggregationService(dispatcher, this.context, this.delSrvc, super.dirsHandler) { @Override public LogAggregationFileController getLogAggregationFileController( Configuration conf) { return spyLogAggregationFileFormat; } }); logAggregationService.init(this.conf); logAggregationService.start(); ApplicationId appId = BuilderUtils.newApplicationId(System.currentTimeMillis(), (int) (Math.random() * 1000)); LogAggregationContext contextWithAMAndFailed = Records.newRecord(LogAggregationContext.class); contextWithAMAndFailed.setLogAggregationPolicyClassName( AMOrFailedContainerLogAggregationPolicy.class.getName()); logAggregationService.handle(new LogHandlerAppStartedEvent(appId, this.user, null, this.acls, contextWithAMAndFailed)); dispatcher.await(); ApplicationEvent[] expectedEvents = new ApplicationEvent[] { new ApplicationEvent(appId, ApplicationEventType.APPLICATION_LOG_HANDLING_FAILED) }; checkEvents(appEventHandler, expectedEvents, false, "getType", "getApplicationID", "getDiagnostic"); Mockito.reset(logAggregationService); ApplicationId appId2 = BuilderUtils.newApplicationId(System.currentTimeMillis(), (int) (Math.random() * 1000)); File appLogDir = new File(localLogDir, appId2.toString()); appLogDir.mkdir(); logAggregationService.handle(new LogHandlerAppStartedEvent(appId2, this.user, null, this.acls, contextWithAMAndFailed)); dispatcher.await(); expectedEvents = new ApplicationEvent[] { new ApplicationEvent(appId, ApplicationEventType.APPLICATION_LOG_HANDLING_FAILED), new ApplicationEvent(appId2, ApplicationEventType.APPLICATION_LOG_HANDLING_INITED) }; checkEvents(appEventHandler, expectedEvents, false, "getType", "getApplicationID", "getDiagnostic"); logAggregationService.stop(); }
@Test public void testVerifyAndCreateRemoteDirNonExistence() throws Exception { this.conf.set(YarnConfiguration.NM_LOG_DIRS, localLogDir.getAbsolutePath()); File aNewFile = new File(String.valueOf("tmp"+System.currentTimeMillis())); this.conf.set(YarnConfiguration.NM_REMOTE_APP_LOG_DIR, aNewFile.getAbsolutePath()); LogAggregationService logAggregationService = spy( new LogAggregationService(dispatcher, this.context, this.delSrvc, super.dirsHandler)); logAggregationService.init(this.conf); logAggregationService.start(); boolean existsBefore = aNewFile.exists(); assertTrue("The new file already exists!", !existsBefore); ApplicationId appId = ApplicationId.newInstance( System.currentTimeMillis(), 1); LogAggregationContext contextWithAMAndFailed = Records.newRecord(LogAggregationContext.class); contextWithAMAndFailed.setLogAggregationPolicyClassName( AMOrFailedContainerLogAggregationPolicy.class.getName()); logAggregationService.handle(new LogHandlerAppStartedEvent(appId, this.user, null, this.acls, contextWithAMAndFailed)); dispatcher.await(); boolean existsAfter = aNewFile.exists(); assertTrue("The new aggregate file is not successfully created", existsAfter); aNewFile.delete(); logAggregationService.stop(); }
@Test public void testRemoteRootLogDirIsCreatedWithCorrectGroupOwner() throws IOException { this.conf.set(YarnConfiguration.NM_LOG_DIRS, localLogDir.getAbsolutePath()); Path aNewFile = new Path(String.valueOf("tmp"+System.currentTimeMillis())); this.conf.set(YarnConfiguration.NM_REMOTE_APP_LOG_DIR, aNewFile.getName()); LogAggregationService logAggregationService = new LogAggregationService( dispatcher, this.context, this.delSrvc, super.dirsHandler); logAggregationService.init(this.conf); logAggregationService.start(); ApplicationId appId = ApplicationId.newInstance( System.currentTimeMillis(), 1); LogAggregationContext contextWithAMAndFailed = Records.newRecord(LogAggregationContext.class); contextWithAMAndFailed.setLogAggregationPolicyClassName( AMOrFailedContainerLogAggregationPolicy.class.getName()); logAggregationService.handle(new LogHandlerAppStartedEvent(appId, this.user, null, this.acls, contextWithAMAndFailed)); dispatcher.await(); String targetGroup = UserGroupInformation.getLoginUser().getPrimaryGroupName(); FileSystem fs = FileSystem.get(this.conf); FileStatus fileStatus = fs.getFileStatus(aNewFile); Assert.assertEquals("The new aggregate file is not successfully created", fileStatus.getGroup(), targetGroup); fs.delete(aNewFile, true); logAggregationService.stop(); }
@Test public void testAppLogDirCreation() throws Exception { final String inputSuffix = "logs-tfile"; 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, "logs"); InlineDispatcher dispatcher = new InlineDispatcher(); dispatcher.init(this.conf); dispatcher.start(); FileSystem fs = FileSystem.get(this.conf); final FileSystem spyFs = spy(FileSystem.get(this.conf)); final LogAggregationTFileController spyFileFormat = new LogAggregationTFileController() { @Override public FileSystem getFileSystem(Configuration conf) throws IOException { return spyFs; } }; spyFileFormat.initialize(conf, "TFile"); LogAggregationService aggSvc = new LogAggregationService(dispatcher, this.context, this.delSrvc, super.dirsHandler) { @Override public LogAggregationFileController getLogAggregationFileController( Configuration conf) { return spyFileFormat; } }; aggSvc.init(this.conf); aggSvc.start(); ApplicationId appId = BuilderUtils.newApplicationId(1, 1); Path userDir = fs.makeQualified(new Path( remoteRootLogDir.getAbsolutePath(), this.user)); Path bucketDir = fs.makeQualified(LogAggregationUtils.getRemoteBucketDir( new Path(remoteRootLogDir.getAbsolutePath()), this.user, inputSuffix, appId)); Path suffixDir = bucketDir.getParent(); Path appDir = fs.makeQualified(LogAggregationUtils.getRemoteAppLogDir( new Path(remoteRootLogDir.getAbsolutePath()), appId, this.user, inputSuffix)); LogAggregationContext contextWithAllContainers = Records.newRecord(LogAggregationContext.class); contextWithAllContainers.setLogAggregationPolicyClassName( AllContainerLogAggregationPolicy.class.getName()); aggSvc.handle(new LogHandlerAppStartedEvent(appId, this.user, null, this.acls, contextWithAllContainers)); verify(spyFs).mkdirs(eq(userDir), isA(FsPermission.class)); verify(spyFs).mkdirs(eq(suffixDir), isA(FsPermission.class)); verify(spyFs).mkdirs(eq(bucketDir), isA(FsPermission.class)); verify(spyFs).mkdirs(eq(appDir), isA(FsPermission.class)); ApplicationId appId2 = BuilderUtils.newApplicationId(1, 2); Path appDir2 = fs.makeQualified(LogAggregationUtils.getRemoteAppLogDir( new Path(remoteRootLogDir.getAbsolutePath()), appId2, this.user, inputSuffix)); aggSvc.handle(new LogHandlerAppStartedEvent(appId2, this.user, null, this.acls, contextWithAllContainers)); verify(spyFs).mkdirs(eq(appDir2), isA(FsPermission.class)); ApplicationId appId3 = BuilderUtils.newApplicationId(2, 2); Path appDir3 = fs.makeQualified(LogAggregationUtils.getRemoteAppLogDir( new Path(remoteRootLogDir.getAbsolutePath()), appId3, this.user, inputSuffix)); new File(appDir3.toUri().getPath()).mkdir(); aggSvc.handle(new LogHandlerAppStartedEvent(appId3, this.user, null, this.acls, contextWithAllContainers)); verify(spyFs, never()).mkdirs(eq(appDir3), isA(FsPermission.class)); ApplicationId appId4 = BuilderUtils.newApplicationId(2, 10003); Path appDir4 = fs.makeQualified(LogAggregationUtils.getRemoteAppLogDir( new Path(remoteRootLogDir.getAbsolutePath()), appId4, this.user, inputSuffix)); Path bucketDir4 = appDir4.getParent(); new File(bucketDir4.toUri().getPath()).mkdir(); aggSvc.handle(new LogHandlerAppStartedEvent(appId4, this.user, null, this.acls, contextWithAllContainers)); verify(spyFs, never()).mkdirs(eq(bucketDir4), isA(FsPermission.class)); verify(spyFs).mkdirs(eq(appDir4), isA(FsPermission.class)); aggSvc.stop(); aggSvc.close(); dispatcher.stop(); }
@Test(timeout = 50000) public void testLogAggregationAbsentContainer() throws Exception { ApplicationId appId = createApplication(); LogAggregationService logAggregationService = createLogAggregationService(appId, FailedOrKilledContainerLogAggregationPolicy.class, null); ApplicationAttemptId appAttemptId1 = BuilderUtils.newApplicationAttemptId(appId, 1); ContainerId containerId = BuilderUtils.newContainerId(appAttemptId1, 2l); logAggregationService.handle(new LogHandlerContainerFinishedEvent( containerId, ContainerType.APPLICATION_MASTER, 100)); }