diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/AppLogAggregatorImpl.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/AppLogAggregatorImpl.java index ba7836a..20b25ea 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/AppLogAggregatorImpl.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/AppLogAggregatorImpl.java @@ -90,12 +90,6 @@ = YarnConfiguration.NM_PREFIX + "log-aggregation.num-log-files-per-app"; private static final int DEFAULT_NM_LOG_AGGREGATION_NUM_LOG_FILES_SIZE_PER_APP = 30; - - // This configuration is for debug and test purpose. By setting - // this configuration as true. We can break the lower bound of - // NM_LOG_AGGREGATION_ROLL_MONITORING_INTERVAL_SECONDS. - private static final String NM_LOG_AGGREGATION_DEBUG_ENABLED - = YarnConfiguration.NM_PREFIX + "log-aggregation.debug-enabled"; private final LocalDirsHandlerService dirsHandler; private final Dispatcher dispatcher; @@ -279,10 +273,17 @@ private void uploadLogsForContainers(boolean appFinished) { // b) some set of running containers: For all the Running containers, // we use exitCode of 0 to find those which satisfy the // ContainerLogAggregationPolicy. - Set pendingContainerInThisCycle = new HashSet(); - this.pendingContainers.drainTo(pendingContainerInThisCycle); - Set finishedContainers = - new HashSet(pendingContainerInThisCycle); + + // only make the copy of pendingContainers. + // We do not remove the finished container. + // For the application, it is very possible that we have set + // includePattern/excludePattern which will not be checked during + // the log aggregation in rollingMonitorInterval. + // If we remove the finished container here, we would miss the logs when + // the app finishes. + Set pendingContainerInThisCycle = new HashSet( + pendingContainers); + if (this.context.getApplications().get(this.appId) != null) { for (Container container : this.context.getApplications() .get(this.appId).getContainers().values()) { @@ -335,12 +336,6 @@ private void uploadLogsForContainers(boolean appFinished) { uploadedFilePathsInThisCycle .toArray(new Path[uploadedFilePathsInThisCycle.size()])); } - - // This container is finished, and all its logs have been uploaded, - // remove it from containerLogAggregators. - if (finishedContainers.contains(container)) { - containerLogAggregators.remove(container); - } } // Before upload logs, make sure the number of existing logs diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/TestLogAggregationService.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/TestLogAggregationService.java index 3961e1a..589c9ef 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/TestLogAggregationService.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/TestLogAggregationService.java @@ -2303,7 +2303,7 @@ public void testSkipUnnecessaryNNOperationsForService() throws Exception { AMOnlyLogAggregationPolicy.class.getName()); contextWithAMOnly.setRolledLogsIncludePattern("sys*"); contextWithAMOnly.setRolledLogsExcludePattern("std_final"); - verifySkipUnnecessaryNNOperations(contextWithAMOnly, 1, 4, 1); + verifySkipUnnecessaryNNOperations(contextWithAMOnly, 3, 4, 1); } private void verifySkipUnnecessaryNNOperations( @@ -2326,13 +2326,17 @@ private void verifySkipUnnecessaryNNOperations( AppLogAggregatorImpl aggregator = (AppLogAggregatorImpl) logAggregationService.getAppLogAggregators() .get(appId); + + // Try to aggregate the logs for the container at the first time aggregator.doLogAggregationOutOfBand(); Thread.sleep(2000); + // We did not remove the finished containers and the application is still + // running, try to aggregate the logs for the container at second time. aggregator.doLogAggregationOutOfBand(); Thread.sleep(2000); - // App finishes. + // App finishes. Try to aggregate this container for the third time. logAggregationService.handle(new LogHandlerAppFinishedEvent(appId)); logAggregationService.stop();