From 810470508bacfddcbb54241e27046168edca2f48 Mon Sep 17 00:00:00 2001 From: Junping Du Date: Thu, 16 Jun 2016 07:18:36 -0700 Subject: [PATCH] YARN-5237. Fix missing log files issue in rolling log aggregation. Contributed by Xuan Gong. --- .../logaggregation/AggregatedLogFormat.java | 47 +++++--- .../logaggregation/AppLogAggregatorImpl.java | 9 +- .../TestLogAggregationService.java | 100 +++++++++++++++++- 3 files changed, 134 insertions(+), 22 deletions(-) diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/logaggregation/AggregatedLogFormat.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/logaggregation/AggregatedLogFormat.java index f5dbc9272d8..50517310736 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/logaggregation/AggregatedLogFormat.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/logaggregation/AggregatedLogFormat.java @@ -168,17 +168,20 @@ public class AggregatedLogFormat { private final Set alreadyUploadedLogFiles; private Set allExistingFileMeta = new HashSet(); private final boolean appFinished; + private final boolean containerFinished; // TODO Maybe add a version string here. Instead of changing the version of // the entire k-v format public LogValue(List rootLogDirs, ContainerId containerId, String user) { - this(rootLogDirs, containerId, user, null, new HashSet(), true); + this(rootLogDirs, containerId, user, null, new HashSet(), true, + true); } public LogValue(List rootLogDirs, ContainerId containerId, String user, LogAggregationContext logAggregationContext, - Set alreadyUploadedLogFiles, boolean appFinished) { + Set alreadyUploadedLogFiles, boolean appFinished, + boolean containerFinished) { this.rootLogDirs = new ArrayList(rootLogDirs); this.containerId = containerId; this.user = user; @@ -188,6 +191,7 @@ public class AggregatedLogFormat { this.logAggregationContext = logAggregationContext; this.alreadyUploadedLogFiles = alreadyUploadedLogFiles; this.appFinished = appFinished; + this.containerFinished = containerFinished; } private Set getPendingLogFilesToUploadForThisContainer() { @@ -294,28 +298,39 @@ public class AggregatedLogFormat { this.allExistingFileMeta.add(getLogFileMetaData(logFile)); } + Set fileCandidates = new HashSet(candidates); if (this.logAggregationContext != null && candidates.size() > 0) { - filterFiles( - this.appFinished ? this.logAggregationContext.getIncludePattern() + fileCandidates = getFileCandidates(fileCandidates, this.appFinished); + if (!this.appFinished && this.containerFinished) { + Set addition = new HashSet(candidates); + addition = getFileCandidates(addition, true); + fileCandidates.addAll(addition); + } + } + return fileCandidates; + } + + private Set getFileCandidates(Set candidates, + boolean useRegularPattern) { + filterFiles( + useRegularPattern ? this.logAggregationContext.getIncludePattern() : this.logAggregationContext.getRolledLogsIncludePattern(), candidates, false); - filterFiles( - this.appFinished ? this.logAggregationContext.getExcludePattern() + filterFiles( + useRegularPattern ? this.logAggregationContext.getExcludePattern() : this.logAggregationContext.getRolledLogsExcludePattern(), candidates, true); - Iterable mask = - Iterables.filter(candidates, new Predicate() { - @Override - public boolean apply(File next) { - return !alreadyUploadedLogFiles + Iterable mask = + Iterables.filter(candidates, new Predicate() { + @Override + public boolean apply(File next) { + return !alreadyUploadedLogFiles .contains(getLogFileMetaData(next)); - } - }); - candidates = Sets.newHashSet(mask); - } - return candidates; + } + }); + return Sets.newHashSet(mask); } private void filterFiles(String pattern, Set candidates, diff --git a/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 b/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 a5b1e2c5ec8..fa07c593c98 100644 --- a/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 +++ b/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 @@ -348,7 +348,8 @@ public class AppLogAggregatorImpl implements AppLogAggregator { containerLogAggregators.put(container, aggregator); } Set uploadedFilePathsInThisCycle = - aggregator.doContainerLogAggregation(writer, appFinished); + aggregator.doContainerLogAggregation(writer, appFinished, + finishedContainers.contains(container)); if (uploadedFilePathsInThisCycle.size() > 0) { uploadedLogsInThisCycle = true; this.delService.delete(this.userUgi.getShortUserName(), null, @@ -650,15 +651,15 @@ public class AppLogAggregatorImpl implements AppLogAggregator { } public Set doContainerLogAggregation(LogWriter writer, - boolean appFinished) { + boolean appFinished, boolean containerFinished) { LOG.info("Uploading logs for container " + containerId + ". Current good log dirs are " + StringUtils.join(",", dirsHandler.getLogDirsForRead())); final LogKey logKey = new LogKey(containerId); final LogValue logValue = new LogValue(dirsHandler.getLogDirsForRead(), containerId, - userUgi.getShortUserName(), logAggregationContext, - this.uploadedFileMeta, appFinished); + userUgi.getShortUserName(), logAggregationContext, + this.uploadedFileMeta, appFinished, containerFinished); try { writer.append(logKey, logValue); } catch (Exception e) { diff --git a/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 b/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 d56d030fd36..c0a3f3a853a 100644 --- a/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 +++ b/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 @@ -1435,6 +1435,102 @@ public class TestLogAggregationService extends BaseContainerManagerTest { "getApplicationID"); } + @SuppressWarnings("resource") + @Test (timeout = 50000) + public void testLogAggregationServiceWithPatternsAndIntervals() + throws Exception { + LogAggregationContext logAggregationContext = + Records.newRecord(LogAggregationContext.class); + // set IncludePattern and RolledLogsIncludePattern. + // When the app is running, we only aggregate the log with + // the name stdout. After the app finishes, we only aggregate + // the log with the name std_final. + logAggregationContext.setRolledLogsIncludePattern("stdout"); + logAggregationContext.setIncludePattern("std_final"); + this.conf.set( + YarnConfiguration.NM_LOG_DIRS, localLogDir.getAbsolutePath()); + //configure YarnConfiguration.NM_REMOTE_APP_LOG_DIR to + //have fully qualified path + this.conf.set(YarnConfiguration.NM_REMOTE_APP_LOG_DIR, + this.remoteRootLogDir.toURI().toString()); + this.conf.setLong( + YarnConfiguration.NM_LOG_AGGREGATION_ROLL_MONITORING_INTERVAL_SECONDS, + 3600); + + this.conf.setLong(YarnConfiguration.DEBUG_NM_DELETE_DELAY_SEC, 3600); + + ApplicationId application = + BuilderUtils.newApplicationId(System.currentTimeMillis(), 1); + ApplicationAttemptId appAttemptId = + BuilderUtils.newApplicationAttemptId(application, 1); + ContainerId container = createContainer(appAttemptId, 1, + ContainerType.APPLICATION_MASTER); + + ConcurrentMap maps = + this.context.getApplications(); + Application app = mock(Application.class); + maps.put(application, app); + when(app.getContainers()).thenReturn(this.context.getContainers()); + + LogAggregationService logAggregationService = + new LogAggregationService(dispatcher, context, this.delSrvc, + super.dirsHandler); + + logAggregationService.init(this.conf); + logAggregationService.start(); + + // AppLogDir should be created + File appLogDir = + new File(localLogDir, ConverterUtils.toString(application)); + appLogDir.mkdir(); + logAggregationService.handle(new LogHandlerAppStartedEvent(application, + this.user, null, this.acls, logAggregationContext)); + + // Simulate log-file creation + // create std_final in log directory which will not be aggregated + // until the app finishes. + String[] logFilesWithFinalLog = + new String[] {"stdout", "std_final"}; + writeContainerLogs(appLogDir, container, logFilesWithFinalLog); + + // Do log aggregation + AppLogAggregatorImpl aggregator = + (AppLogAggregatorImpl) logAggregationService.getAppLogAggregators() + .get(application); + + aggregator.doLogAggregationOutOfBand(); + + Assert.assertTrue(waitAndCheckLogNum(logAggregationService, application, + 50, 1, false, null)); + + String[] logFiles = new String[] { "stdout" }; + verifyContainerLogs(logAggregationService, application, + new ContainerId[] {container}, logFiles, 1, true); + + logAggregationService.handle( + new LogHandlerContainerFinishedEvent(container, 0)); + + dispatcher.await(); + + // Do the log aggregation after ContainerFinishedEvent but before + // AppFinishedEvent. The std_final is expected to be aggregated this time + // even if the app is running but the container finishes. + aggregator.doLogAggregationOutOfBand(); + + Assert.assertTrue(waitAndCheckLogNum(logAggregationService, application, + 50, 2, false, null)); + + // This container finishes. + // The log "std_final" should be aggregated this time. + String[] logFinalLog = new String[] {"std_final"}; + verifyContainerLogs(logAggregationService, application, + new ContainerId[] {container}, logFinalLog, 1, true); + + logAggregationService.handle(new LogHandlerAppFinishedEvent(application)); + + logAggregationService.stop(); + } + @Test (timeout = 50000) @SuppressWarnings("unchecked") public void testNoneContainerPolicy() throws Exception { @@ -1443,14 +1539,14 @@ public class TestLogAggregationService extends BaseContainerManagerTest { LogAggregationService logAggregationService = createLogAggregationService( appId, NoneContainerLogAggregationPolicy.class, null); - String[] logFiles = new String[] { "stdout" }; + String[] logFiles = new String[] {"stdout"}; ContainerId container1 = finishContainer(appId, logAggregationService, ContainerType.APPLICATION_MASTER, 1, 0, logFiles); finishApplication(appId, logAggregationService); verifyContainerLogs(logAggregationService, appId, - new ContainerId[] { container1 }, logFiles, 0, false); + new ContainerId[] {container1}, logFiles, 0, false); verifyLogAggFinishEvent(appId); }