YARN-1145. Fixed a potential file-handle leak in the web interface for displaying aggregated logs. Contributed by Rohith Sharma.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/trunk@1551326 13f79535-47bb-0310-9956-ffa450edef68

hadoop-yarn-project/CHANGES.txt

@@ -249,6 +249,9 @@ Release 2.4.0 - UNRELEASED
     YARN-1505. Fixed Webapplication proxy server to not hardcode its bind
     address. (Xuan Gong via vinodkv)
 
+    YARN-1145. Fixed a potential file-handle leak in the web interface for
+    displaying aggregated logs. (Rohith Sharma via vinodkv)
+
 Release 2.3.0 - UNRELEASED
 
   INCOMPATIBLE CHANGES

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/logaggregation/AggregatedLogFormat.java

@@ -53,6 +53,7 @@
 import org.apache.hadoop.fs.Options;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.io.IOUtils;
 import org.apache.hadoop.io.SecureIOUtils;
 import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.io.file.tfile.TFile;
@@ -294,7 +295,7 @@ public void append(LogKey logKey, LogValue logValue) throws IOException {
       out.close();
     }
 
-    public void closeWriter() {
+    public void close() {
       try {
         this.writer.close();
       } catch (IOException e) {
@@ -569,9 +570,8 @@ public static void readAContainerLogsForALogType(
       out.println("");
     }
 
-    public void close() throws IOException {
+    public void close() {
-      this.scanner.close();
+      IOUtils.cleanup(LOG, scanner, reader, fsDataIStream);
-      this.fsDataIStream.close();
     }
   }
 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/webapp/log/AggregatedLogsBlock.java

@@ -59,109 +59,113 @@ public class AggregatedLogsBlock extends HtmlBlock {
 
   @Override
   protected void render(Block html) {
-    ContainerId containerId = verifyAndGetContainerId(html);
-    NodeId nodeId = verifyAndGetNodeId(html);
-    String appOwner = verifyAndGetAppOwner(html);
-    LogLimits logLimits = verifyAndGetLogLimits(html);
-    if (containerId == null || nodeId == null || appOwner == null
-        || appOwner.isEmpty() || logLimits == null) {
-      return;
-    }
-    
-    ApplicationId applicationId =
-        containerId.getApplicationAttemptId().getApplicationId();
-    String logEntity = $(ENTITY_STRING);
-    if (logEntity == null || logEntity.isEmpty()) {
-      logEntity = containerId.toString();
-    }
-
-    if (!conf.getBoolean(YarnConfiguration.LOG_AGGREGATION_ENABLED,
-        YarnConfiguration.DEFAULT_LOG_AGGREGATION_ENABLED)) {
-      html.h1()
-          ._("Aggregation is not enabled. Try the nodemanager at " + nodeId)
-          ._();
-      return;
-    }
-    
-    Path remoteRootLogDir =
-        new Path(conf.get(YarnConfiguration.NM_REMOTE_APP_LOG_DIR,
-            YarnConfiguration.DEFAULT_NM_REMOTE_APP_LOG_DIR));
     AggregatedLogFormat.LogReader reader = null;
     try {
-      reader =
+      ContainerId containerId = verifyAndGetContainerId(html);
-          new AggregatedLogFormat.LogReader(conf,
+      NodeId nodeId = verifyAndGetNodeId(html);
-              LogAggregationUtils.getRemoteNodeLogFileForApp(
+      String appOwner = verifyAndGetAppOwner(html);
-                  remoteRootLogDir, applicationId, appOwner, nodeId,
+      LogLimits logLimits = verifyAndGetLogLimits(html);
-                  LogAggregationUtils.getRemoteNodeLogDirSuffix(conf)));
+      if (containerId == null || nodeId == null || appOwner == null
-    } catch (FileNotFoundException e) {
+          || appOwner.isEmpty() || logLimits == null) {
-      // ACLs not available till the log file is opened.
-      html.h1()
-          ._("Logs not available for "
-              + logEntity
-              + ". Aggregation may not be complete, "
-              + "Check back later or try the nodemanager at "
-              + nodeId)._();
-      return;
-    } catch (IOException e) {
-      html.h1()._("Error getting logs for " + logEntity)._();
-      LOG.error("Error getting logs for " + logEntity, e);
-      return;
-    }
-
-    String owner = null;
-    Map<ApplicationAccessType, String> appAcls = null;
-    try {
-      owner = reader.getApplicationOwner();
-      appAcls = reader.getApplicationAcls();
-    } catch (IOException e) {
-      html.h1()._("Error getting logs for " + logEntity)._();
-      LOG.error("Error getting logs for " + logEntity, e);
-      return;
-    }
-    ApplicationACLsManager aclsManager = new ApplicationACLsManager(conf);
-    aclsManager.addApplication(applicationId, appAcls);
-
-    String remoteUser = request().getRemoteUser();
-    UserGroupInformation callerUGI = null;
-    if (remoteUser != null) {
-      callerUGI = UserGroupInformation.createRemoteUser(remoteUser);
-    }
-    if (callerUGI != null
-        && !aclsManager.checkAccess(callerUGI, ApplicationAccessType.VIEW_APP,
-            owner, applicationId)) {
-      html.h1()
-          ._("User [" + remoteUser
-              + "] is not authorized to view the logs for " + logEntity)._();
-      return;
-    }
-
-    String desiredLogType = $(CONTAINER_LOG_TYPE);
-    try {
-      AggregatedLogFormat.ContainerLogsReader logReader =
-          reader.getContainerLogsReader(containerId);
-      if (logReader == null) {
-        html.h1()._(
-            "Logs not available for " + logEntity
-                + ". Could be caused by the rentention policy")._();
         return;
       }
 
-      boolean foundLog = readContainerLogs(html, logReader, logLimits,
+      ApplicationId applicationId = containerId.getApplicationAttemptId()
-          desiredLogType);
+          .getApplicationId();
+      String logEntity = $(ENTITY_STRING);
+      if (logEntity == null || logEntity.isEmpty()) {
+        logEntity = containerId.toString();
+      }
 
-      if (!foundLog) {
+      if (!conf.getBoolean(YarnConfiguration.LOG_AGGREGATION_ENABLED,
-        if (desiredLogType.isEmpty()) {
+          YarnConfiguration.DEFAULT_LOG_AGGREGATION_ENABLED)) {
-          html.h1("No logs available for container " + containerId.toString());
+        html.h1()
-        } else {
+            ._("Aggregation is not enabled. Try the nodemanager at " + nodeId)
-          html.h1("Unable to locate '" + desiredLogType
+            ._();
-              + "' log for container " + containerId.toString());
+        return;
+      }
+
+      Path remoteRootLogDir = new Path(conf.get(
+          YarnConfiguration.NM_REMOTE_APP_LOG_DIR,
+          YarnConfiguration.DEFAULT_NM_REMOTE_APP_LOG_DIR));
+
+      try {
+        reader = new AggregatedLogFormat.LogReader(conf,
+            LogAggregationUtils.getRemoteNodeLogFileForApp(remoteRootLogDir,
+                applicationId, appOwner, nodeId,
+                LogAggregationUtils.getRemoteNodeLogDirSuffix(conf)));
+      } catch (FileNotFoundException e) {
+        // ACLs not available till the log file is opened.
+        html.h1()
+            ._("Logs not available for " + logEntity
+                + ". Aggregation may not be complete, "
+                + "Check back later or try the nodemanager at " + nodeId)._();
+        return;
+      } catch (IOException e) {
+        html.h1()._("Error getting logs for " + logEntity)._();
+        LOG.error("Error getting logs for " + logEntity, e);
+        return;
+      }
+
+      String owner = null;
+      Map<ApplicationAccessType, String> appAcls = null;
+      try {
+        owner = reader.getApplicationOwner();
+        appAcls = reader.getApplicationAcls();
+      } catch (IOException e) {
+        html.h1()._("Error getting logs for " + logEntity)._();
+        LOG.error("Error getting logs for " + logEntity, e);
+        return;
+      }
+      ApplicationACLsManager aclsManager = new ApplicationACLsManager(conf);
+      aclsManager.addApplication(applicationId, appAcls);
+
+      String remoteUser = request().getRemoteUser();
+      UserGroupInformation callerUGI = null;
+      if (remoteUser != null) {
+        callerUGI = UserGroupInformation.createRemoteUser(remoteUser);
+      }
+      if (callerUGI != null
+          && !aclsManager.checkAccess(callerUGI,
+              ApplicationAccessType.VIEW_APP, owner, applicationId)) {
+        html.h1()
+            ._("User [" + remoteUser
+                + "] is not authorized to view the logs for " + logEntity)._();
+        return;
+      }
+
+      String desiredLogType = $(CONTAINER_LOG_TYPE);
+      try {
+        AggregatedLogFormat.ContainerLogsReader logReader = reader
+            .getContainerLogsReader(containerId);
+        if (logReader == null) {
+          html.h1()
+              ._("Logs not available for " + logEntity
+                  + ". Could be caused by the rentention policy")._();
+          return;
         }
+
+        boolean foundLog = readContainerLogs(html, logReader, logLimits,
+            desiredLogType);
+
+        if (!foundLog) {
+          if (desiredLogType.isEmpty()) {
+            html.h1("No logs available for container " + containerId.toString());
+          } else {
+            html.h1("Unable to locate '" + desiredLogType
+                + "' log for container " + containerId.toString());
+          }
+          return;
+        }
+      } catch (IOException e) {
+        html.h1()._("Error getting logs for " + logEntity)._();
+        LOG.error("Error getting logs for " + logEntity, e);
         return;
       }
-    } catch (IOException e) {
+    } finally {
-      html.h1()._("Error getting logs for " + logEntity)._();
+      if (reader != null) {
-      LOG.error("Error getting logs for " + logEntity, e);
+        reader.close();
-      return;
+      }
     }
   }
 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/logaggregation/TestAggregatedLogFormat.java

@@ -114,7 +114,7 @@ public void testReadAcontainerLogs1() throws Exception {
             testContainerId, ugi.getShortUserName());
 
     logWriter.append(logKey, logValue);
-    logWriter.closeWriter();
+    logWriter.close();
 
     // make sure permission are correct on the file
     FileStatus fsStatus =  fs.getFileStatus(remoteAppLogFile);
@@ -194,7 +194,7 @@ public void testContainerLogsFileAccess() throws IOException {
         ugi.getShortUserName());
     logWriter.append(logKey, logValue);
 
-    logWriter.closeWriter();
+    logWriter.close();
     
     BufferedReader in =
         new BufferedReader(new FileReader(new File(remoteAppLogFile

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/logaggregation/TestAggregatedLogsBlock.java

@@ -229,7 +229,7 @@ private void writeLog(Configuration configuration, String user)
 
     writer.append(new AggregatedLogFormat.LogKey("container_0_0001_01_000001"),
         new AggregatedLogFormat.LogValue(rootLogDirs, containerId,UserGroupInformation.getCurrentUser().getShortUserName()));
-    writer.closeWriter();
+    writer.close();
   }
 
   private void writeLogs(String dirName) throws Exception {

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

@@ -178,7 +178,7 @@ private void doAppLogAggregation() {
         localAppLogDirs);
 
     if (this.writer != null) {
-      this.writer.closeWriter();
+      this.writer.close();
       LOG.info("Finished aggregate log-file for app " + this.applicationId);
     }