YARN-4905. Improved "yarn logs" command-line to optionally show log metadata also. Contributed by Xuan Gong.

(cherry picked from commit 9e37fe3b7a)

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/cli/LogsCLI.java

@@ -45,6 +45,7 @@ import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.util.Tool;
 import org.apache.hadoop.yarn.api.records.ApplicationId;
 import org.apache.hadoop.yarn.api.records.ApplicationReport;
+import org.apache.hadoop.yarn.api.records.ContainerId;
 import org.apache.hadoop.yarn.api.records.ContainerReport;
 import org.apache.hadoop.yarn.api.records.YarnApplicationState;
 import org.apache.hadoop.yarn.client.api.YarnClient;
@@ -78,59 +79,16 @@ public class LogsCLI extends Configured implements Tool {
   private static final String APP_OWNER_OPTION = "appOwner";
   private static final String AM_CONTAINER_OPTION = "am";
   private static final String CONTAINER_LOG_FILES = "logFiles";
+  private static final String SHOW_META_INFO = "show_meta_info";
+  private static final String LIST_NODES_OPTION = "list_nodes";
   public static final String HELP_CMD = "help";
 
   @Override
   public int run(String[] args) throws Exception {
 
-    Options opts = new Options();
+    Options opts = createCommandOpts();
-    opts.addOption(HELP_CMD, false, "Displays help for all commands.");
-    Option appIdOpt =
-        new Option(APPLICATION_ID_OPTION, true, "ApplicationId (required)");
-    appIdOpt.setRequired(true);
-    opts.addOption(appIdOpt);
-    opts.addOption(CONTAINER_ID_OPTION, true, "ContainerId. "
-        + "By default, it will only print syslog if the application is runing."
-        + " Work with -logFiles to get other logs.");
-    opts.addOption(NODE_ADDRESS_OPTION, true, "NodeAddress in the format "
-      + "nodename:port");
-    opts.addOption(APP_OWNER_OPTION, true,
-      "AppOwner (assumed to be current user if not specified)");
-    Option amOption = new Option(AM_CONTAINER_OPTION, true, 
-      "Prints the AM Container logs for this application. "
-      + "Specify comma-separated value to get logs for related AM Container. "
-      + "For example, If we specify -am 1,2, we will get the logs for "
-      + "the first AM Container as well as the second AM Container. "
-      + "To get logs for all AM Containers, use -am ALL. "
-      + "To get logs for the latest AM Container, use -am -1. "
-      + "By default, it will only print out syslog. Work with -logFiles "
-      + "to get other logs");
-    amOption.setValueSeparator(',');
-    amOption.setArgs(Option.UNLIMITED_VALUES);
-    amOption.setArgName("AM Containers");
-    opts.addOption(amOption);
-    Option logFileOpt = new Option(CONTAINER_LOG_FILES, true,
-      "Work with -am/-containerId and specify comma-separated value "
-        + "to get specified container log files. Use \"ALL\" to fetch all the "
-        + "log files for the container.");
-    logFileOpt.setValueSeparator(',');
-    logFileOpt.setArgs(Option.UNLIMITED_VALUES);
-    logFileOpt.setArgName("Log File Name");
-    opts.addOption(logFileOpt);
 
-    opts.getOption(APPLICATION_ID_OPTION).setArgName("Application ID");
+    Options printOpts = createPrintOpts(opts);
-    opts.getOption(CONTAINER_ID_OPTION).setArgName("Container ID");
-    opts.getOption(NODE_ADDRESS_OPTION).setArgName("Node Address");
-    opts.getOption(APP_OWNER_OPTION).setArgName("Application Owner");
-    opts.getOption(AM_CONTAINER_OPTION).setArgName("AM Containers");
-
-    Options printOpts = new Options();
-    printOpts.addOption(opts.getOption(HELP_CMD));
-    printOpts.addOption(opts.getOption(CONTAINER_ID_OPTION));
-    printOpts.addOption(opts.getOption(NODE_ADDRESS_OPTION));
-    printOpts.addOption(opts.getOption(APP_OWNER_OPTION));
-    printOpts.addOption(opts.getOption(AM_CONTAINER_OPTION));
-    printOpts.addOption(opts.getOption(CONTAINER_LOG_FILES));
 
     if (args.length < 1) {
       printHelpMessage(printOpts);
@@ -146,6 +104,8 @@ public class LogsCLI extends Configured implements Tool {
     String nodeAddress = null;
     String appOwner = null;
     boolean getAMContainerLogs = false;
+    boolean showMetaInfo = false;
+    boolean nodesList = false;
     String[] logFiles = null;
     List<String> amContainersList = new ArrayList<String>();
     try {
@@ -155,32 +115,15 @@ public class LogsCLI extends Configured implements Tool {
       nodeAddress = commandLine.getOptionValue(NODE_ADDRESS_OPTION);
       appOwner = commandLine.getOptionValue(APP_OWNER_OPTION);
       getAMContainerLogs = commandLine.hasOption(AM_CONTAINER_OPTION);
+      showMetaInfo = commandLine.hasOption(SHOW_META_INFO);
+      nodesList = commandLine.hasOption(LIST_NODES_OPTION);
       if (getAMContainerLogs) {
-        String[] amContainers = commandLine.getOptionValues(AM_CONTAINER_OPTION);
-        for (String am : amContainers) {
-          boolean errorInput = false;
-          if (!am.trim().equalsIgnoreCase("ALL")) {
         try {
-              int id = Integer.parseInt(am.trim());
+          amContainersList = parseAMContainer(commandLine, printOpts);
-              if (id != -1 && id <= 0) {
-                errorInput = true;
-              }
         } catch (NumberFormatException ex) {
-              errorInput = true;
+          System.err.println(ex.getMessage());
-            }
-            if (errorInput) {
-              System.err.println(
-                "Invalid input for option -am. Valid inputs are 'ALL', -1 "
-                + "and any other integer which is larger than 0.");
-              printHelpMessage(printOpts);
           return -1;
         }
-            amContainersList.add(am.trim());
-          } else {
-            amContainersList.add("ALL");
-            break;
-          }
-        }
       }
       if (commandLine.hasOption(CONTAINER_LOG_FILES)) {
         logFiles = commandLine.getOptionValues(CONTAINER_LOG_FILES);
@@ -212,138 +155,54 @@ public class LogsCLI extends Configured implements Tool {
       appOwner = UserGroupInformation.getCurrentUser().getShortUserName();
     }
 
-    boolean appStateKnown = true;
+    boolean appStateObtainedSuccessfully = true;
     YarnApplicationState appState = YarnApplicationState.NEW;
     try {
       appState = getApplicationState(appId);
       if (appState == YarnApplicationState.NEW
           || appState == YarnApplicationState.NEW_SAVING
           || appState == YarnApplicationState.SUBMITTED) {
-        System.out.println("Logs are not avaiable right now.");
+        System.err.println("Logs are not avaiable right now.");
         return -1;
       }
     } catch (IOException | YarnException e) {
-      appStateKnown = false;
+      appStateObtainedSuccessfully = false;
       System.err.println("Unable to get ApplicationState."
           + " Attempting to fetch logs directly from the filesystem.");
     }
 
+    if (showMetaInfo) {
+      return showMetaInfo(appState, appStateObtainedSuccessfully,
+          logCliHelper, appId, containerIdStr, nodeAddress, appOwner);
+    }
+
+    if (nodesList) {
+      return showNodeLists(appState, appStateObtainedSuccessfully,
+          logCliHelper, appId, appOwner);
+    }
     // To get am logs
     if (getAMContainerLogs) {
-      // if we do not specify the value for CONTAINER_LOG_FILES option,
+      return fetchAMContainerLogs(logFiles, appState, appId, appOwner,
-      // we will only output syslog
+          amContainersList, logCliHelper);
-      if (logFiles == null || logFiles.length == 0) {
-        logFiles = new String[] { "syslog" };
-      }
-      // If the application is running, we will call the RM WebService
-      // to get the AppAttempts which includes the nodeHttpAddress
-      // and containerId for all the AM Containers.
-      // After that, we will call NodeManager webService to get the
-      // related logs
-      if (appState == YarnApplicationState.ACCEPTED
-          || appState == YarnApplicationState.RUNNING) {
-        return printAMContainerLogs(getConf(), appIdStr, amContainersList,
-          logFiles, logCliHelper, appOwner, false);
-      } else {
-        // If the application is in the final state, we will call RM webservice
-        // to get all AppAttempts information first. If we get nothing,
-        // we will try to call AHS webservice to get related AppAttempts
-        // which includes nodeAddress for the AM Containers.
-        // After that, we will use nodeAddress and containerId
-        // to get logs from HDFS directly.
-        if (getConf().getBoolean(YarnConfiguration.APPLICATION_HISTORY_ENABLED,
-          YarnConfiguration.DEFAULT_APPLICATION_HISTORY_ENABLED)) {
-          return printAMContainerLogs(getConf(), appIdStr, amContainersList,
-            logFiles, logCliHelper, appOwner, true);
-        } else {
-          System.out
-            .println(
-                "Can not get AMContainers logs for the application:" + appId);
-          System.out.println("This application:" + appId + " is finished."
-              + " Please enable the application history service. Or Using "
-              + "yarn logs -applicationId <appId> -containerId <containerId> "
-              + "--nodeAddress <nodeHttpAddress> to get the container logs");
-          return -1;
-        }
-      }
     }
 
     int resultCode = 0;
     if (containerIdStr != null) {
-      // if we provide the node address and the application is in the final
+      ContainerId containerId = ContainerId.fromString(containerIdStr);
-      // state, we could directly get logs from HDFS.
+      if (!containerId.getApplicationAttemptId().getApplicationId()
-      if (nodeAddress != null && (!appStateKnown ||
+          .equals(appId)) {
-          isApplicationFinished(appState))) {
+        System.err.println("The Application:" + appId
-        // if user specified "ALL" as the logFiles param, pass null
+            + " does not have the container:" + containerId);
-        // to logCliHelper so that it fetches all the logs
-        List<String> logs;
-        if (logFiles == null) {
-          logs = null;
-        } else if (fetchAllLogFiles(logFiles)) {
-          logs = null;
-        } else {
-          logs = Arrays.asList(logFiles);
-        }
-        return logCliHelper.dumpAContainersLogsForALogType(appIdStr,
-            containerIdStr, nodeAddress, appOwner, logs);
-      }
-      String nodeHttpAddress = null;
-      String nodeId = null;
-      try {
-        // If the nodeAddress is not provided, we will try to get
-        // the ContainerReport. In the containerReport, we could get
-        // nodeAddress and nodeHttpAddress
-        ContainerReport report = getContainerReport(containerIdStr);
-        nodeHttpAddress =
-            report.getNodeHttpAddress().replaceFirst(
-              WebAppUtils.getHttpSchemePrefix(getConf()), "");
-        nodeId = report.getAssignedNode().toString();
-      } catch (IOException | YarnException ex) {
-        if (!appStateKnown || isApplicationFinished(appState)) {
-          String [] requestedLogFiles = logFiles;
-          if(fetchAllLogFiles(logFiles)) {
-            requestedLogFiles = null;
-          }
-          return printContainerLogsForFinishedApplicationWithoutNodeId(
-              appIdStr, containerIdStr, requestedLogFiles, logCliHelper,
-              appOwner);
-        } else if (!isApplicationFinished(appState)) {
-          System.err.println("Unable to get logs for this container:"
-              + containerIdStr + "for the application:" + appId);
-          System.out.println("The application: " + appId + " is still running, "
-              + "and we can not get Container report for the container: "
-              + containerIdStr +". Please try later or after the application "
-              + "finishes.");
         return -1;
       }
-      }
+      return fetchContainerLogs(appState, appStateObtainedSuccessfully,
-      // If the application is not in the final state,
+          logFiles, appOwner, nodeAddress, containerId, logCliHelper);
-      // we will provide the NodeHttpAddress and get the container logs
-      // by calling NodeManager webservice.
-      if (!isApplicationFinished(appState)) {
-        if (logFiles == null || logFiles.length == 0) {
-          logFiles = new String[] {"syslog"};
-        }
-        printContainerLogsFromRunningApplication(getConf(), appIdStr,
-            containerIdStr, nodeHttpAddress, nodeId, logFiles, logCliHelper,
-            appOwner);
-      } else {
-        String[] requestedLogFiles = logFiles;
-        if(fetchAllLogFiles(logFiles)) {
-          requestedLogFiles = null;
-        }
-        // If the application is in the final state, we will directly
-        // get the container logs from HDFS.
-        printContainerLogsForFinishedApplication(appIdStr, containerIdStr,
-            nodeId, requestedLogFiles, logCliHelper, appOwner);
-      }
-      return resultCode;
     } else {
       if (nodeAddress == null) {
         resultCode =
             logCliHelper.dumpAllContainersLogs(appId, appOwner, System.out);
       } else {
-        System.out.println("Should at least provide ContainerId!");
+        System.err.println("Should at least provide ContainerId!");
         printHelpMessage(printOpts);
         resultCode = -1;
       }
@@ -382,7 +241,8 @@ public class LogsCLI extends Configured implements Tool {
   private void printHelpMessage(Options options) {
     System.out.println("Retrieve logs for completed YARN applications.");
     HelpFormatter formatter = new HelpFormatter();
-    formatter.printHelp("yarn logs -applicationId <application ID> [OPTIONS]", new Options());
+    formatter.printHelp("yarn logs -applicationId <application ID> [OPTIONS]",
+        new Options());
     formatter.setSyntaxPrefix("");
     formatter.printHelp("general options are:", options);
   }
@@ -410,9 +270,9 @@ public class LogsCLI extends Configured implements Tool {
     return amContainersList;
   }
 
-  private List<JSONObject> getAMContainerInfoForAHSWebService(Configuration conf,
+  private List<JSONObject> getAMContainerInfoForAHSWebService(
-      String appId) throws ClientHandlerException, UniformInterfaceException,
+      Configuration conf, String appId) throws ClientHandlerException,
-      JSONException {
+      UniformInterfaceException, JSONException {
     Client webServiceClient = Client.create();
     String webAppAddress =
         WebAppUtils.getHttpSchemePrefix(conf)
@@ -420,8 +280,9 @@ public class LogsCLI extends Configured implements Tool {
     WebResource webResource = webServiceClient.resource(webAppAddress);
 
     ClientResponse response =
-        webResource.path("ws").path("v1").path("applicationhistory").path("apps")
+        webResource.path("ws").path("v1").path("applicationhistory")
-          .path(appId).path("appattempts").accept(MediaType.APPLICATION_JSON)
+          .path("apps").path(appId).path("appattempts")
+          .accept(MediaType.APPLICATION_JSON)
           .get(ClientResponse.class);
     JSONObject json = response.getEntity(JSONObject.class);
     JSONArray requests = json.getJSONArray("appAttempt");
@@ -467,24 +328,27 @@ public class LogsCLI extends Configured implements Tool {
             logFiles.add(elements.item(i).getTextContent());
           }
         } catch (Exception e) {
-          System.out.println("Unable to parse xml from webservice. Error:");
+          System.err.println("Unable to parse xml from webservice. Error:");
-          System.out.println(e.getMessage());
+          System.err.println(e.getMessage());
           throw new IOException(e);
         }
       }
 
     } catch (ClientHandlerException | UniformInterfaceException ex) {
-      System.out.println("Unable to fetch log files list");
+      System.err.println("Unable to fetch log files list");
       throw new IOException(ex);
     }
     return logFiles.toArray(new String[0]);
   }
 
   private void printContainerLogsFromRunningApplication(Configuration conf,
-      String appId, String containerIdStr, String nodeHttpAddress,
+      ContainerId containerId, String nodeHttpAddress,
       String nodeId, String[] logFiles, LogCLIHelpers logCliHelper,
       String appOwner) throws IOException {
-    String [] requestedLogFiles = logFiles;
+    String appId = containerId.getApplicationAttemptId()
+        .getApplicationId().toString();
+    String containerIdStr = containerId.toString();
+    String[] requestedLogFiles = logFiles;
     // fetch all the log files for the container
     if (fetchAllLogFiles(logFiles)) {
       requestedLogFiles =
@@ -511,7 +375,7 @@ public class LogsCLI extends Configured implements Tool {
         System.out.println(response.getEntity(String.class));
         System.out.println("End of LogType:" + logFile);
       } catch (ClientHandlerException | UniformInterfaceException ex) {
-        System.out.println("Can not find the log file:" + logFile
+        System.err.println("Can not find the log file:" + logFile
             + " for the container:" + containerIdStr + " in NodeManager:"
             + nodeId);
       }
@@ -528,7 +392,8 @@ public class LogsCLI extends Configured implements Tool {
     System.out.println(containerString);
     System.out.println(StringUtils.repeat("=", containerString.length()));
     logCliHelper.dumpAContainersLogsForALogType(appId, containerId,
-      nodeAddress, appOwner, logFiles != null ? Arrays.asList(logFiles) : null);
+        nodeAddress, appOwner, logFiles != null ? Arrays.asList(logFiles)
+        : null);
   }
 
   private int printContainerLogsForFinishedApplicationWithoutNodeId(
@@ -648,7 +513,7 @@ public class LogsCLI extends Configured implements Tool {
           }
         }
         if (nodeId != null && !nodeId.isEmpty()) {
-          String [] requestedLogFilesList = null;
+          String[] requestedLogFilesList = null;
           if(!fetchAllLogFiles(logFiles)) {
             requestedLogFilesList = logFiles;
           }
@@ -659,18 +524,262 @@ public class LogsCLI extends Configured implements Tool {
     } else {
       if (nodeHttpAddress != null && containerId != null
           && !nodeHttpAddress.isEmpty() && !containerId.isEmpty()) {
-        String [] requestedLogFiles = logFiles;
+        String[] requestedLogFiles = logFiles;
         // fetch all the log files for the AM
         if (fetchAllLogFiles(logFiles)) {
           requestedLogFiles =
               getContainerLogFiles(getConf(), containerId, nodeHttpAddress);
         }
-        printContainerLogsFromRunningApplication(conf, appId, containerId,
+        printContainerLogsFromRunningApplication(conf,
-          nodeHttpAddress, nodeId, requestedLogFiles, logCliHelper, appOwner);
+            ContainerId.fromString(containerId), nodeHttpAddress, nodeId,
+            requestedLogFiles, logCliHelper, appOwner);
       }
     }
   }
 
+  private int showMetaInfo(YarnApplicationState appState,
+      boolean appStateObtainedSuccessfully, LogCLIHelpers logCliHelper,
+      ApplicationId appId, String containerIdStr, String nodeAddress,
+      String appOwner) throws IOException {
+    if (!isApplicationFinished(appState) && appStateObtainedSuccessfully) {
+      System.err.println("The -show_meta_info command can be only used "
+          + "with finished applications");
+      return -1;
+    } else {
+      logCliHelper.printLogMetadata(appId, containerIdStr, nodeAddress,
+          appOwner, System.out, System.err);
+      return 0;
+    }
+  }
+
+  private int showNodeLists(YarnApplicationState appState,
+      boolean appStateObtainedSuccessfully, LogCLIHelpers logCliHelper,
+      ApplicationId appId, String appOwner) throws IOException {
+    if (!isApplicationFinished(appState) && appStateObtainedSuccessfully) {
+      System.err.println("The -list_nodes command can be only used with "
+          + "finished applications");
+      return -1;
+    } else {
+      logCliHelper.printNodesList(appId, appOwner, System.out, System.err);
+      return 0;
+    }
+  }
+
+  private Options createCommandOpts() {
+    Options opts = new Options();
+    opts.addOption(HELP_CMD, false, "Displays help for all commands.");
+    Option appIdOpt =
+        new Option(APPLICATION_ID_OPTION, true, "ApplicationId (required)");
+    appIdOpt.setRequired(true);
+    opts.addOption(appIdOpt);
+    opts.addOption(CONTAINER_ID_OPTION, true, "ContainerId. "
+        + "By default, it will only print syslog if the application is runing."
+        + " Work with -logFiles to get other logs.");
+    opts.addOption(NODE_ADDRESS_OPTION, true, "NodeAddress in the format "
+        + "nodename:port");
+    opts.addOption(APP_OWNER_OPTION, true,
+        "AppOwner (assumed to be current user if not specified)");
+    Option amOption = new Option(AM_CONTAINER_OPTION, true,
+        "Prints the AM Container logs for this application. "
+        + "Specify comma-separated value to get logs for related AM "
+        + "Container. For example, If we specify -am 1,2, we will get "
+        + "the logs for the first AM Container as well as the second "
+        + "AM Container. To get logs for all AM Containers, use -am ALL. "
+        + "To get logs for the latest AM Container, use -am -1. "
+        + "By default, it will only print out syslog. Work with -logFiles "
+        + "to get other logs");
+    amOption.setValueSeparator(',');
+    amOption.setArgs(Option.UNLIMITED_VALUES);
+    amOption.setArgName("AM Containers");
+    opts.addOption(amOption);
+    Option logFileOpt = new Option(CONTAINER_LOG_FILES, true,
+        "Work with -am/-containerId and specify comma-separated value "
+        + "to get specified container log files. Use \"ALL\" to fetch all the "
+        + "log files for the container.");
+    logFileOpt.setValueSeparator(',');
+    logFileOpt.setArgs(Option.UNLIMITED_VALUES);
+    logFileOpt.setArgName("Log File Name");
+    opts.addOption(logFileOpt);
+    opts.addOption(SHOW_META_INFO, false, "Show the log metadata, "
+        + "including log-file names, the size of the log files. "
+        + "You can combine this with --containerId to get log metadata for "
+        + "the specific container, or with --nodeAddress to get log metadata "
+        + "for all the containers on the specific NodeManager. "
+        + "Currently, this option can only be used for finished "
+        + "applications.");
+    opts.addOption(LIST_NODES_OPTION, false,
+        "Show the list of nodes that successfully aggregated logs. "
+        + "This option can only be used with finished applications.");
+    opts.getOption(APPLICATION_ID_OPTION).setArgName("Application ID");
+    opts.getOption(CONTAINER_ID_OPTION).setArgName("Container ID");
+    opts.getOption(NODE_ADDRESS_OPTION).setArgName("Node Address");
+    opts.getOption(APP_OWNER_OPTION).setArgName("Application Owner");
+    opts.getOption(AM_CONTAINER_OPTION).setArgName("AM Containers");
+    return opts;
+  }
+
+  private Options createPrintOpts(Options commandOpts) {
+    Options printOpts = new Options();
+    printOpts.addOption(commandOpts.getOption(HELP_CMD));
+    printOpts.addOption(commandOpts.getOption(CONTAINER_ID_OPTION));
+    printOpts.addOption(commandOpts.getOption(NODE_ADDRESS_OPTION));
+    printOpts.addOption(commandOpts.getOption(APP_OWNER_OPTION));
+    printOpts.addOption(commandOpts.getOption(AM_CONTAINER_OPTION));
+    printOpts.addOption(commandOpts.getOption(CONTAINER_LOG_FILES));
+    printOpts.addOption(commandOpts.getOption(SHOW_META_INFO));
+    printOpts.addOption(commandOpts.getOption(LIST_NODES_OPTION));
+    return printOpts;
+  }
+
+  private List<String> parseAMContainer(CommandLine commandLine,
+      Options printOpts) throws NumberFormatException {
+    List<String> amContainersList = new ArrayList<String>();
+    String[] amContainers = commandLine.getOptionValues(AM_CONTAINER_OPTION);
+    for (String am : amContainers) {
+      boolean errorInput = false;
+      if (!am.trim().equalsIgnoreCase("ALL")) {
+        try {
+          int id = Integer.parseInt(am.trim());
+          if (id != -1 && id <= 0) {
+            errorInput = true;
+          }
+        } catch (NumberFormatException ex) {
+          errorInput = true;
+        }
+        if (errorInput) {
+          String errMessage =
+              "Invalid input for option -am. Valid inputs are 'ALL', -1 "
+              + "and any other integer which is larger than 0.";
+          printHelpMessage(printOpts);
+          throw new NumberFormatException(errMessage);
+        }
+        amContainersList.add(am.trim());
+      } else {
+        amContainersList.add("ALL");
+        break;
+      }
+    }
+    return amContainersList;
+  }
+
+  private int fetchAMContainerLogs(String[] logFiles,
+      YarnApplicationState appState, ApplicationId appId,
+      String appOwner, List<String> amContainersList,
+      LogCLIHelpers logCliHelper) throws Exception {
+    // if we do not specify the value for CONTAINER_LOG_FILES option,
+    // we will only output syslog
+    if (logFiles == null || logFiles.length == 0) {
+      logFiles = new String[] {"syslog"};
+    }
+    // If the application is running, we will call the RM WebService
+    // to get the AppAttempts which includes the nodeHttpAddress
+    // and containerId for all the AM Containers.
+    // After that, we will call NodeManager webService to get the
+    // related logs
+    if (appState == YarnApplicationState.ACCEPTED
+        || appState == YarnApplicationState.RUNNING) {
+      return printAMContainerLogs(getConf(), appId.toString(), amContainersList,
+        logFiles, logCliHelper, appOwner, false);
+    } else {
+      // If the application is in the final state, we will call RM webservice
+      // to get all AppAttempts information first. If we get nothing,
+      // we will try to call AHS webservice to get related AppAttempts
+      // which includes nodeAddress for the AM Containers.
+      // After that, we will use nodeAddress and containerId
+      // to get logs from HDFS directly.
+      if (getConf().getBoolean(YarnConfiguration.APPLICATION_HISTORY_ENABLED,
+          YarnConfiguration.DEFAULT_APPLICATION_HISTORY_ENABLED)) {
+        return printAMContainerLogs(getConf(), appId.toString(),
+            amContainersList, logFiles, logCliHelper, appOwner, true);
+      } else {
+        System.err.println("Can not get AMContainers logs for "
+            + "the application:" + appId);
+        System.err.println("This application:" + appId + " is finished."
+            + " Please enable the application history service. Or Using "
+            + "yarn logs -applicationId <appId> -containerId <containerId> "
+            + "--nodeAddress <nodeHttpAddress> to get the container logs");
+        return -1;
+      }
+    }
+  }
+
+  private int fetchContainerLogs(YarnApplicationState appState,
+      boolean appStateObtainedSuccessfully, String[] logFiles,
+      String appOwner, String nodeAddress,
+      ContainerId containerId, LogCLIHelpers logCliHelper) throws IOException {
+    int resultCode = 0;
+    String appIdStr = containerId.getApplicationAttemptId()
+        .getApplicationId().toString();
+    String containerIdStr = containerId.toString();
+    // if we provide the node address and the application is in the final
+    // state, we could directly get logs from HDFS.
+    if (nodeAddress != null && (!appStateObtainedSuccessfully ||
+        isApplicationFinished(appState))) {
+      // if user specified "ALL" as the logFiles param, pass null
+      // to logCliHelper so that it fetches all the logs
+      List<String> logs;
+      if (logFiles == null) {
+        logs = null;
+      } else if (fetchAllLogFiles(logFiles)) {
+        logs = null;
+      } else {
+        logs = Arrays.asList(logFiles);
+      }
+      return logCliHelper.dumpAContainersLogsForALogType(appIdStr,
+          containerIdStr, nodeAddress, appOwner, logs);
+    }
+    String nodeHttpAddress = null;
+    String nodeId = null;
+    try {
+      // If the nodeAddress is not provided, we will try to get
+      // the ContainerReport. In the containerReport, we could get
+      // nodeAddress and nodeHttpAddress
+      ContainerReport report = getContainerReport(containerIdStr);
+      nodeHttpAddress =
+          report.getNodeHttpAddress().replaceFirst(
+            WebAppUtils.getHttpSchemePrefix(getConf()), "");
+      nodeId = report.getAssignedNode().toString();
+    } catch (IOException | YarnException ex) {
+      if (!appStateObtainedSuccessfully || isApplicationFinished(appState)) {
+        String[] requestedLogFiles = logFiles;
+        if(fetchAllLogFiles(logFiles)) {
+          requestedLogFiles = null;
+        }
+        return printContainerLogsForFinishedApplicationWithoutNodeId(
+            appIdStr, containerIdStr, requestedLogFiles, logCliHelper,
+            appOwner);
+      } else if (!isApplicationFinished(appState)) {
+        System.err.println("Unable to get logs for this container:"
+            + containerIdStr + "for the application:" + appIdStr);
+        System.err.println("The application: " + appIdStr
+            + " is still running, and we can not get Container report "
+            + "for the container: " + containerIdStr +". Please try later "
+            + "or after the application finishes.");
+        return -1;
+      }
+    }
+    // If the application is not in the final state,
+    // we will provide the NodeHttpAddress and get the container logs
+    // by calling NodeManager webservice.
+    if (!isApplicationFinished(appState)) {
+      if (logFiles == null || logFiles.length == 0) {
+        logFiles = new String[] {"syslog"};
+      }
+      printContainerLogsFromRunningApplication(getConf(), containerId,
+          nodeHttpAddress, nodeId, logFiles, logCliHelper, appOwner);
+    } else {
+      String[] requestedLogFiles = logFiles;
+      if(fetchAllLogFiles(logFiles)) {
+        requestedLogFiles = null;
+      }
+      // If the application is in the final state, we will directly
+      // get the container logs from HDFS.
+      printContainerLogsForFinishedApplication(appIdStr, containerIdStr,
+          nodeId, requestedLogFiles, logCliHelper, appOwner);
+    }
+    return resultCode;
+  }
+
   private static class AMLogsRequest {
     private String amContainerId;
     private String nodeId;

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/cli/TestLogsCLI.java

@@ -18,6 +18,7 @@
 
 package org.apache.hadoop.yarn.client.cli;
 
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.Matchers.any;
@@ -166,11 +167,22 @@ public class TestLogsCLI {
     pw.println("                                 runing. Work with -logFiles to get other");
     pw.println("                                 logs.");
     pw.println(" -help                           Displays help for all commands.");
+    pw.println(" -list_nodes                     Show the list of nodes that successfully");
+    pw.println("                                 aggregated logs. This option can only be");
+    pw.println("                                 used with finished applications.");
     pw.println(" -logFiles <Log File Name>       Work with -am/-containerId and specify");
     pw.println("                                 comma-separated value to get specified");
     pw.println("                                 container log files. Use \"ALL\" to fetch");
     pw.println("                                 all the log files for the container.");
     pw.println(" -nodeAddress <Node Address>     NodeAddress in the format nodename:port");
+    pw.println(" -show_meta_info                 Show the log metadata, including log-file");
+    pw.println("                                 names, the size of the log files. You can");
+    pw.println("                                 combine this with --containerId to get");
+    pw.println("                                 log metadata for the specific container,");
+    pw.println("                                 or with --nodeAddress to get log metadata");
+    pw.println("                                 for all the containers on the specific");
+    pw.println("                                 NodeManager. Currently, this option can");
+    pw.println("                                 only be used for finished applications.");
     pw.close();
     String appReportStr = baos.toString("UTF-8");
     Assert.assertEquals(appReportStr, sysOutStream.toString());
@@ -286,10 +298,10 @@ public class TestLogsCLI {
             "-nodeAddress", nodeId.toString(), "-containerId",
             containerId0.toString() });
     assertTrue(exitCode == -1);
-    assertTrue(sysOutStream.toString().contains(
+    assertTrue(sysErrStream.toString().contains(
       "Logs for container " + containerId0.toString()
           + " are not present in this log-file."));
-    sysOutStream.reset();
+    sysErrStream.reset();
 
     // uploaded two logs for container3. The first log is named as syslog.
     // The second one is named as stdout.
@@ -336,6 +348,140 @@ public class TestLogsCLI {
     fs.delete(new Path(rootLogDir), true);
   }
 
+  @Test (timeout = 15000)
+  public void testPrintContainerLogMetadata() throws Exception {
+    String remoteLogRootDir = "target/logs/";
+    Configuration configuration = new Configuration();
+    configuration.setBoolean(YarnConfiguration.LOG_AGGREGATION_ENABLED, true);
+    configuration
+      .set(YarnConfiguration.NM_REMOTE_APP_LOG_DIR, remoteLogRootDir);
+    configuration.setBoolean(YarnConfiguration.YARN_ACL_ENABLE, true);
+    configuration.set(YarnConfiguration.YARN_ADMIN_ACL, "admin");
+    FileSystem fs = FileSystem.get(configuration);
+    String rootLogDir = "target/LocalLogs";
+
+    ApplicationId appId = ApplicationId.newInstance(0, 1);
+    ApplicationAttemptId appAttemptId =
+        ApplicationAttemptId.newInstance(appId, 1);
+    List<ContainerId> containerIds = new ArrayList<ContainerId>();
+    ContainerId containerId1 = ContainerId.newContainerId(
+        appAttemptId, 1);
+    ContainerId containerId2 = ContainerId.newContainerId(
+        appAttemptId, 2);
+    containerIds.add(containerId1);
+    containerIds.add(containerId2);
+
+    List<NodeId> nodeIds = new ArrayList<NodeId>();
+    NodeId nodeId = NodeId.newInstance("localhost", 1234);
+    nodeIds.add(nodeId);
+    nodeIds.add(nodeId);
+
+    createContainerLogs(configuration, remoteLogRootDir, rootLogDir, fs,
+        appId, containerIds, nodeIds);
+
+    YarnClient mockYarnClient =
+        createMockYarnClient(YarnApplicationState.FINISHED);
+    LogsCLI cli = new LogsCLIForTest(mockYarnClient);
+    cli.setConf(configuration);
+
+    cli.run(new String[] { "-applicationId", appId.toString(),
+        "-show_meta_info" });
+    assertTrue(sysOutStream.toString().contains(
+        "Container: container_0_0001_01_000001 on localhost_"));
+    assertTrue(sysOutStream.toString().contains(
+        "Container: container_0_0001_01_000002 on localhost_"));
+    assertTrue(sysOutStream.toString().contains(
+        "LogType:syslog"));
+    assertTrue(sysOutStream.toString().contains(
+        "LogLength:43"));
+    sysOutStream.reset();
+
+    cli.run(new String[] { "-applicationId", appId.toString(),
+        "-show_meta_info", "-containerId", "container_0_0001_01_000001" });
+    assertTrue(sysOutStream.toString().contains(
+        "Container: container_0_0001_01_000001 on localhost_"));
+    assertFalse(sysOutStream.toString().contains(
+        "Container: container_0_0001_01_000002 on localhost_"));
+    assertTrue(sysOutStream.toString().contains(
+        "LogType:syslog"));
+    assertTrue(sysOutStream.toString().contains(
+        "LogLength:43"));
+    sysOutStream.reset();
+
+    cli.run(new String[] { "-applicationId", appId.toString(),
+        "-show_meta_info", "-nodeAddress", "localhost" });
+    assertTrue(sysOutStream.toString().contains(
+        "Container: container_0_0001_01_000001 on localhost_"));
+    assertTrue(sysOutStream.toString().contains(
+        "Container: container_0_0001_01_000002 on localhost_"));
+    assertTrue(sysOutStream.toString().contains(
+        "LogType:syslog"));
+    assertTrue(sysOutStream.toString().contains(
+        "LogLength:43"));
+    sysOutStream.reset();
+
+    cli.run(new String[] { "-applicationId", appId.toString(),
+        "-show_meta_info", "-nodeAddress", "localhost", "-containerId",
+        "container_1234" });
+    assertTrue(sysErrStream.toString().contains(
+        "The container container_1234 couldn't be found on the node "
+        + "specified: localhost"));
+    sysErrStream.reset();
+
+    fs.delete(new Path(remoteLogRootDir), true);
+    fs.delete(new Path(rootLogDir), true);
+  }
+
+  @Test (timeout = 15000)
+  public void testListNodeInfo() throws Exception {
+    String remoteLogRootDir = "target/logs/";
+    Configuration configuration = new Configuration();
+    configuration.setBoolean(YarnConfiguration.LOG_AGGREGATION_ENABLED, true);
+    configuration
+      .set(YarnConfiguration.NM_REMOTE_APP_LOG_DIR, remoteLogRootDir);
+    configuration.setBoolean(YarnConfiguration.YARN_ACL_ENABLE, true);
+    configuration.set(YarnConfiguration.YARN_ADMIN_ACL, "admin");
+
+    ApplicationId appId = ApplicationId.newInstance(0, 1);
+    ApplicationAttemptId appAttemptId =
+        ApplicationAttemptId.newInstance(appId, 1);
+    List<ContainerId> containerIds = new ArrayList<ContainerId>();
+    ContainerId containerId1 = ContainerId.newContainerId(
+        appAttemptId, 1);
+    ContainerId containerId2 = ContainerId.newContainerId(
+        appAttemptId, 2);
+    containerIds.add(containerId1);
+    containerIds.add(containerId2);
+
+    List<NodeId> nodeIds = new ArrayList<NodeId>();
+    NodeId nodeId1 = NodeId.newInstance("localhost1", 1234);
+    NodeId nodeId2 = NodeId.newInstance("localhost2", 2345);
+    nodeIds.add(nodeId1);
+    nodeIds.add(nodeId2);
+
+    String rootLogDir = "target/LocalLogs";
+    FileSystem fs = FileSystem.get(configuration);
+
+    createContainerLogs(configuration, remoteLogRootDir, rootLogDir, fs,
+        appId, containerIds, nodeIds);
+
+    YarnClient mockYarnClient =
+        createMockYarnClient(YarnApplicationState.FINISHED);
+    LogsCLI cli = new LogsCLIForTest(mockYarnClient);
+    cli.setConf(configuration);
+
+    cli.run(new String[] { "-applicationId", appId.toString(),
+        "-list_nodes" });
+    assertTrue(sysOutStream.toString().contains(
+        LogAggregationUtils.getNodeString(nodeId1)));
+    assertTrue(sysOutStream.toString().contains(
+        LogAggregationUtils.getNodeString(nodeId2)));
+    sysOutStream.reset();
+
+    fs.delete(new Path(remoteLogRootDir), true);
+    fs.delete(new Path(rootLogDir), true);
+  }
+
   @Test (timeout = 15000)
   public void testFetchApplictionLogsHar() throws Exception {
     String remoteLogRootDir = "target/logs/";
@@ -384,6 +530,46 @@ public class TestLogsCLI {
     fs.delete(new Path(remoteLogRootDir), true);
   }
 
+  private void createContainerLogs(Configuration configuration,
+      String remoteLogRootDir, String rootLogDir, FileSystem fs,
+      ApplicationId appId, List<ContainerId> containerIds,
+      List<NodeId> nodeIds) throws Exception {
+
+    UserGroupInformation ugi = UserGroupInformation.getCurrentUser();
+
+    // create local logs
+    Path rootLogDirPath = new Path(rootLogDir);
+    if (fs.exists(rootLogDirPath)) {
+      fs.delete(rootLogDirPath, true);
+    }
+    assertTrue(fs.mkdirs(rootLogDirPath));
+
+    Path appLogsDir = new Path(rootLogDirPath, appId.toString());
+    if (fs.exists(appLogsDir)) {
+      fs.delete(appLogsDir, true);
+    }
+    assertTrue(fs.mkdirs(appLogsDir));
+    List<String> rootLogDirs = Arrays.asList(rootLogDir);
+    List<String> logTypes = new ArrayList<String>();
+    logTypes.add("syslog");
+    // create container logs in localLogDir
+    for (ContainerId containerId : containerIds) {
+      createContainerLogInLocalDir(appLogsDir, containerId, fs, logTypes);
+    }
+    Path path =
+        new Path(remoteLogRootDir + ugi.getShortUserName()
+        + "/logs/application_0_0001");
+
+    if (fs.exists(path)) {
+      fs.delete(path, true);
+    }
+    assertTrue(fs.mkdirs(path));
+    for (int i=0; i<containerIds.size(); i++) {
+      uploadContainerLogIntoRemoteDir(ugi, configuration, rootLogDirs, nodeIds.get(i),
+          containerIds.get(i), path, fs);
+    }
+  }
+
   private static void createContainerLogInLocalDir(Path appLogsDir,
       ContainerId containerId, FileSystem fs, List<String> logTypes) throws Exception {
     Path containerLogsDir = new Path(appLogsDir, containerId.toString());

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

@@ -808,6 +808,26 @@ public class AggregatedLogFormat {
       }
     }
 
+    @Private
+    public static void readContainerMetaDataAndSkipData(
+        DataInputStream valueStream, PrintStream out) throws IOException {
+
+      String fileType = valueStream.readUTF();
+      String fileLengthStr = valueStream.readUTF();
+      long fileLength = Long.parseLong(fileLengthStr);
+      out.print("LogType:");
+      out.println(fileType);
+      out.print("LogLength:");
+      out.println(fileLengthStr);
+
+      long totalSkipped = 0;
+      long currSkipped = 0;
+      while (currSkipped != -1 && totalSkipped < fileLength) {
+        currSkipped = valueStream.skip(fileLength - totalSkipped);
+        totalSkipped += currSkipped;
+      }
+    }
+
     public void close() {
       IOUtils.cleanup(LOG, scanner, reader, fsDataIStream);
     }

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

@@ -68,24 +68,10 @@ public class LogCLIHelpers implements Configurable {
   public int dumpAContainersLogsForALogType(String appId, String containerId,
       String nodeId, String jobOwner, List<String> logType,
       boolean outputFailure) throws IOException {
-    Path remoteRootLogDir = new Path(getConf().get(
-        YarnConfiguration.NM_REMOTE_APP_LOG_DIR,
-        YarnConfiguration.DEFAULT_NM_REMOTE_APP_LOG_DIR));
-    String suffix = LogAggregationUtils.getRemoteNodeLogDirSuffix(getConf());
     ApplicationId applicationId = ConverterUtils.toApplicationId(appId);
-    Path remoteAppLogDir = LogAggregationUtils.getRemoteAppLogDir(
+    RemoteIterator<FileStatus> nodeFiles = getRemoteNodeFileDir(
-        remoteRootLogDir, applicationId, jobOwner,
+        applicationId, jobOwner);
-        suffix);
+    if (nodeFiles == null) {
-    RemoteIterator<FileStatus> nodeFiles;
-    try {
-      Path qualifiedLogDir =
-          FileContext.getFileContext(getConf()).makeQualified(
-            remoteAppLogDir);
-      nodeFiles =
-          FileContext.getFileContext(qualifiedLogDir.toUri(), getConf())
-            .listStatus(remoteAppLogDir);
-    } catch (FileNotFoundException fnf) {
-      logDirNotExist(remoteAppLogDir.toString());
       return -1;
     }
     boolean foundContainerLogs = false;
@@ -134,23 +120,10 @@ public class LogCLIHelpers implements Configurable {
   public int dumpAContainersLogsForALogTypeWithoutNodeId(String appId,
       String containerId, String jobOwner, List<String> logType)
     throws IOException {
-    Path remoteRootLogDir = new Path(getConf().get(
-        YarnConfiguration.NM_REMOTE_APP_LOG_DIR,
-        YarnConfiguration.DEFAULT_NM_REMOTE_APP_LOG_DIR));
     ApplicationId applicationId = ConverterUtils.toApplicationId(appId);
-    String user = jobOwner;
+    RemoteIterator<FileStatus> nodeFiles = getRemoteNodeFileDir(
-    String logDirSuffix = LogAggregationUtils.getRemoteNodeLogDirSuffix(
+        applicationId, jobOwner);
-        getConf());
+    if (nodeFiles == null) {
-    Path remoteAppLogDir = LogAggregationUtils.getRemoteAppLogDir(
-        remoteRootLogDir, applicationId, user, logDirSuffix);
-    RemoteIterator<FileStatus> nodeFiles;
-    try {
-      Path qualifiedLogDir =
-          FileContext.getFileContext(getConf()).makeQualified(remoteAppLogDir);
-      nodeFiles = FileContext.getFileContext(qualifiedLogDir.toUri(),
-          getConf()).listStatus(remoteAppLogDir);
-    } catch (FileNotFoundException fnf) {
-      logDirNotExist(remoteAppLogDir.toString());
       return -1;
     }
     boolean foundContainerLogs = false;
@@ -261,22 +234,9 @@ public class LogCLIHelpers implements Configurable {
   @Private
   public int dumpAllContainersLogs(ApplicationId appId, String appOwner,
       PrintStream out) throws IOException {
-    Path remoteRootLogDir = new Path(getConf().get(
+    RemoteIterator<FileStatus> nodeFiles = getRemoteNodeFileDir(
-        YarnConfiguration.NM_REMOTE_APP_LOG_DIR,
+        appId, appOwner);
-        YarnConfiguration.DEFAULT_NM_REMOTE_APP_LOG_DIR));
+    if (nodeFiles == null) {
-    String user = appOwner;
-    String logDirSuffix = LogAggregationUtils.getRemoteNodeLogDirSuffix(getConf());
-    // TODO Change this to get a list of files from the LAS.
-    Path remoteAppLogDir = LogAggregationUtils.getRemoteAppLogDir(
-        remoteRootLogDir, appId, user, logDirSuffix);
-    RemoteIterator<FileStatus> nodeFiles;
-    try {
-      Path qualifiedLogDir =
-          FileContext.getFileContext(getConf()).makeQualified(remoteAppLogDir);
-      nodeFiles = FileContext.getFileContext(qualifiedLogDir.toUri(),
-          getConf()).listStatus(remoteAppLogDir);
-    } catch (FileNotFoundException fnf) {
-      logDirNotExist(remoteAppLogDir.toString());
       return -1;
     }
     boolean foundAnyLogs = false;
@@ -324,12 +284,136 @@ public class LogCLIHelpers implements Configurable {
       }
     }
     if (! foundAnyLogs) {
-      emptyLogDir(remoteAppLogDir.toString());
+      emptyLogDir(getRemoteAppLogDir(appId, appOwner).toString());
       return -1;
     }
     return 0;
   }
 
+  @Private
+  public void printLogMetadata(ApplicationId appId,
+      String containerIdStr, String nodeId, String appOwner,
+      PrintStream out, PrintStream err)
+      throws IOException {
+    boolean getAllContainers = (containerIdStr == null);
+    String nodeIdStr = (nodeId == null) ? null
+        : LogAggregationUtils.getNodeString(nodeId);
+    RemoteIterator<FileStatus> nodeFiles = getRemoteNodeFileDir(
+        appId, appOwner);
+    if (nodeFiles == null) {
+      return;
+    }
+    boolean foundAnyLogs = false;
+    while (nodeFiles.hasNext()) {
+      FileStatus thisNodeFile = nodeFiles.next();
+      if (nodeIdStr != null) {
+        if (!thisNodeFile.getPath().getName().contains(nodeIdStr)) {
+          continue;
+        }
+      }
+      if (!thisNodeFile.getPath().getName()
+          .endsWith(LogAggregationUtils.TMP_FILE_SUFFIX)) {
+        AggregatedLogFormat.LogReader reader =
+            new AggregatedLogFormat.LogReader(getConf(),
+            thisNodeFile.getPath());
+        try {
+          DataInputStream valueStream;
+          LogKey key = new LogKey();
+          valueStream = reader.next(key);
+          while (valueStream != null) {
+            if (getAllContainers || (key.toString().equals(containerIdStr))) {
+              String containerString =
+                  "\n\nContainer: " + key + " on "
+                  + thisNodeFile.getPath().getName();
+              out.println(containerString);
+              out.println("Log Upload Time:"
+                  + thisNodeFile.getModificationTime());
+              out.println(StringUtils.repeat("=", containerString.length()));
+              while (true) {
+                try {
+                  LogReader.readContainerMetaDataAndSkipData(valueStream, out);
+                } catch (EOFException eof) {
+                  break;
+                }
+              }
+              foundAnyLogs = true;
+              if (!getAllContainers) {
+                break;
+              }
+            }
+            // Next container
+            key = new LogKey();
+            valueStream = reader.next(key);
+          }
+        } finally {
+          reader.close();
+        }
+      }
+    }
+    if (!foundAnyLogs) {
+      if (containerIdStr != null && nodeId != null) {
+        err.println("The container " + containerIdStr + " couldn't be found "
+            + "on the node specified: " + nodeId);
+      } else if (nodeId != null) {
+        err.println("Can not find log metadata for any containers on "
+            + nodeId);
+      } else if (containerIdStr != null) {
+        err.println("Can not find log metadata for container: "
+            + containerIdStr);
+      }
+    }
+  }
+
+  @Private
+  public void printNodesList(ApplicationId appId, String appOwner,
+      PrintStream out, PrintStream err) throws IOException {
+    RemoteIterator<FileStatus> nodeFiles = getRemoteNodeFileDir(
+        appId, appOwner);
+    if (nodeFiles == null) {
+      return;
+    }
+    boolean foundNode = false;
+    StringBuilder sb = new StringBuilder();
+    while (nodeFiles.hasNext()) {
+      FileStatus thisNodeFile = nodeFiles.next();
+      sb.append(thisNodeFile.getPath().getName() + "\n");
+      foundNode = true;
+    }
+    if (!foundNode) {
+      err.println("No nodes found that aggregated logs for "
+          + "the application: " + appId);
+    } else {
+      out.println(sb.toString());
+    }
+  }
+
+  private RemoteIterator<FileStatus> getRemoteNodeFileDir(ApplicationId appId,
+      String appOwner) throws IOException {
+    Path remoteAppLogDir = getRemoteAppLogDir(appId, appOwner);
+    RemoteIterator<FileStatus> nodeFiles = null;
+    try {
+      Path qualifiedLogDir =
+          FileContext.getFileContext(getConf()).makeQualified(remoteAppLogDir);
+      nodeFiles = FileContext.getFileContext(qualifiedLogDir.toUri(),
+          getConf()).listStatus(remoteAppLogDir);
+    } catch (FileNotFoundException fnf) {
+      logDirNotExist(remoteAppLogDir.toString());
+    }
+    return nodeFiles;
+  }
+
+  private Path getRemoteAppLogDir(ApplicationId appId, String appOwner) {
+    Path remoteRootLogDir = new Path(getConf().get(
+        YarnConfiguration.NM_REMOTE_APP_LOG_DIR,
+        YarnConfiguration.DEFAULT_NM_REMOTE_APP_LOG_DIR));
+    String user = appOwner;
+    String logDirSuffix = LogAggregationUtils
+        .getRemoteNodeLogDirSuffix(getConf());
+    // TODO Change this to get a list of files from the LAS.
+    return LogAggregationUtils.getRemoteAppLogDir(
+        remoteRootLogDir, appId, user, logDirSuffix);
+  }
+
   @Override
   public void setConf(Configuration conf) {
     this.conf = conf;
@@ -341,16 +425,16 @@ public class LogCLIHelpers implements Configurable {
   }
 
   private static void containerLogNotFound(String containerId) {
-    System.out.println("Logs for container " + containerId
+    System.err.println("Logs for container " + containerId
       + " are not present in this log-file.");
   }
 
   private static void logDirNotExist(String remoteAppLogDir) {
-    System.out.println(remoteAppLogDir + " does not exist.");
+    System.err.println(remoteAppLogDir + " does not exist.");
-    System.out.println("Log aggregation has not completed or is not enabled.");
+    System.err.println("Log aggregation has not completed or is not enabled.");
   }
 
   private static void emptyLogDir(String remoteAppLogDir) {
-    System.out.println(remoteAppLogDir + " does not have any log files.");
+    System.err.println(remoteAppLogDir + " does not have any log files.");
   }
 }