@Override
  public int run(String[] args) throws Exception {
    try {
      webServiceClient = WebServiceClient.getWebServiceClient().createClient();
      return runCommand(args);
    } finally {
      if (yarnClient != null) {
        yarnClient.close();
      }
      if (webServiceClient != null) {
        webServiceClient.destroy();
      }
    }
  }
@Test (timeout = 15000)
  public void testFetchFinishedApplictionLogs() throws Exception {
    String remoteLogRootDir = "target/logs/";
    conf.setBoolean(YarnConfiguration.LOG_AGGREGATION_ENABLED, true);
    conf
      .set(YarnConfiguration.NM_REMOTE_APP_LOG_DIR, remoteLogRootDir);
    conf.setBoolean(YarnConfiguration.YARN_ACL_ENABLE, true);
    conf.set(YarnConfiguration.YARN_ADMIN_ACL, "admin");
    FileSystem fs = FileSystem.get(conf);

    UserGroupInformation ugi = UserGroupInformation.getCurrentUser();
    ApplicationId appId = ApplicationId.newInstance(0, 1);
    ApplicationAttemptId appAttemptId1 =
        ApplicationAttemptId.newInstance(appId, 1);
    ApplicationAttemptId appAttemptId2 =
            ApplicationAttemptId.newInstance(appId, 2);
    ContainerId containerId0 = ContainerId.newContainerId(appAttemptId1, 0);
    ContainerId containerId1 = ContainerId.newContainerId(appAttemptId1, 1);
    ContainerId containerId2 = ContainerId.newContainerId(appAttemptId1, 2);
    ContainerId containerId3 = ContainerId.newContainerId(appAttemptId2, 3);
    ContainerId containerId4 = ContainerId.newContainerId(appAttemptId2, 4);
    final NodeId nodeId = NodeId.newInstance("localhost", 1234);
    final NodeId badNodeId = NodeId.newInstance("badhost", 5678);

    // create local logs
    String rootLogDir = "target/LocalLogs";
    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
    createContainerLogInLocalDir(appLogsDir, containerId1, fs, logTypes,
        ImmutableList.of("empty"));
    createContainerLogInLocalDir(appLogsDir, containerId2, fs, logTypes,
        Collections.emptyList());
    // create two logs for container3 in localLogDir
    logTypes.add("stdout");
    logTypes.add("stdout1234");
    createContainerLogInLocalDir(appLogsDir, containerId3, fs, logTypes,
        Collections.emptyList());

    Path path =
        new Path(remoteLogRootDir + ugi.getShortUserName()
        + "/bucket-logs-tfile/0001/application_0_0001");
    if (fs.exists(path)) {
      fs.delete(path, true);
    }
    assertTrue(fs.mkdirs(path));

    // upload container logs into remote directory
    // the first two logs is empty. When we try to read first two logs,
    // we will meet EOF exception, but it will not impact other logs.
    // Other logs should be read successfully.
    uploadEmptyContainerLogIntoRemoteDir(ugi, conf, rootLogDirs, nodeId,
      containerId0, path, fs);
    uploadEmptyContainerLogIntoRemoteDir(ugi, conf, rootLogDirs, nodeId,
      containerId1, path, fs);
    uploadContainerLogIntoRemoteDir(ugi, conf, rootLogDirs, nodeId,
      containerId1, path, fs);
    uploadContainerLogIntoRemoteDir(ugi, conf, rootLogDirs, nodeId,
      containerId2, path, fs);
    uploadContainerLogIntoRemoteDir(ugi, conf, rootLogDirs, nodeId,
      containerId3, path, fs);
    uploadTruncatedTFileIntoRemoteDir(ugi, conf, badNodeId,
        containerId4, fs);

    YarnClient mockYarnClient =
        createMockYarnClient(
            YarnApplicationState.FINISHED, ugi.getShortUserName());
    LogsCLI cli = new LogsCLIForTest(mockYarnClient) {
      @Override
      public ContainerReport getContainerReport(String containerIdStr)
          throws YarnException, IOException {
        ContainerReport mockReport = mock(ContainerReport.class);
        doReturn(nodeId).when(mockReport).getAssignedNode();
        doReturn("http://localhost:2345").when(mockReport).getNodeHttpAddress();
        return mockReport;
      }
    };
    cli.setConf(conf);

    int exitCode = cli.run(new String[] { "-applicationId", appId.toString() });
    LOG.info(sysOutStream.toString());
    assertTrue(exitCode == 0);
    assertTrue(sysOutStream.toString().contains(
        logMessage(containerId1, "syslog")));
    assertTrue(sysOutStream.toString().contains(
        logMessage(containerId2, "syslog")));
    assertTrue(sysOutStream.toString().contains(
        logMessage(containerId3, "syslog")));
    assertTrue(sysOutStream.toString().contains(
        logMessage(containerId3, "stdout")));
    assertTrue(sysOutStream.toString().contains(
        logMessage(containerId3, "stdout1234")));
    assertTrue(sysOutStream.toString().contains(
        createEmptyLog("empty")));
    sysOutStream.reset();

    // Check fetching data for application attempt with applicationId defined
    exitCode = cli.run(new String[] {"-applicationId", appId.toString(),
        "-applicationAttemptId", appAttemptId1.toString()});
    LOG.info(sysOutStream.toString());
    assertTrue(exitCode == 0);
    assertTrue(sysOutStream.toString().contains(
            logMessage(containerId1, "syslog")));
    assertTrue(sysOutStream.toString().contains(
            logMessage(containerId2, "syslog")));
    assertTrue(sysOutStream.toString().contains(
            logMessage(containerId3, "syslog")));
    assertFalse(sysOutStream.toString().contains(
            logMessage(containerId3, "stdout")));
    assertFalse(sysOutStream.toString().contains(
            logMessage(containerId3, "stdout1234")));
    assertTrue(sysOutStream.toString().contains(
            createEmptyLog("empty")));
    sysOutStream.reset();

    // Check fetching data for application attempt without application defined
    exitCode = cli.run(new String[] {
        "-applicationAttemptId", appAttemptId1.toString()});
    LOG.info(sysOutStream.toString());
    assertTrue(exitCode == 0);
    assertTrue(sysOutStream.toString().contains(
            logMessage(containerId1, "syslog")));
    assertTrue(sysOutStream.toString().contains(
            logMessage(containerId2, "syslog")));
    assertTrue(sysOutStream.toString().contains(
            logMessage(containerId3, "syslog")));
    assertFalse(sysOutStream.toString().contains(
            logMessage(containerId3, "stdout")));
    assertFalse(sysOutStream.toString().contains(
            logMessage(containerId3, "stdout1234")));
    assertTrue(sysOutStream.toString().contains(
            createEmptyLog("empty")));
    sysOutStream.reset();

    exitCode = cli.run(new String[] {"-applicationId", appId.toString(),
        "-log_files_pattern", ".*"});
    assertTrue(exitCode == 0);
    assertTrue(sysOutStream.toString().contains(
        logMessage(containerId1, "syslog")));
    assertTrue(sysOutStream.toString().contains(
        logMessage(containerId2, "syslog")));
    assertTrue(sysOutStream.toString().contains(
        logMessage(containerId3, "syslog")));
    assertTrue(sysOutStream.toString().contains(
        logMessage(containerId3, "stdout")));
    assertTrue(sysOutStream.toString().contains(
        logMessage(containerId3, "stdout1234")));
    assertTrue(sysOutStream.toString().contains(
        createEmptyLog("empty")));
    sysOutStream.reset();

    exitCode = cli.run(new String[] {"-applicationId", appId.toString(),
        "-log_files", "*"});
    assertTrue(exitCode == 0);
    assertTrue(sysOutStream.toString().contains(
        logMessage(containerId1, "syslog")));
    assertTrue(sysOutStream.toString().contains(
        logMessage(containerId2, "syslog")));
    assertTrue(sysOutStream.toString().contains(
        logMessage(containerId3, "syslog")));
    assertTrue(sysOutStream.toString().contains(
        logMessage(containerId3, "stdout")));
    assertTrue(sysOutStream.toString().contains(
        logMessage(containerId3, "stdout1234")));
    assertTrue(sysOutStream.toString().contains(
        createEmptyLog("empty")));
    int fullSize = sysOutStream.toByteArray().length;
    sysOutStream.reset();

    exitCode = cli.run(new String[] {"-applicationId", appId.toString(),
        "-log_files", "stdout"});
    assertTrue(exitCode == 0);
    assertFalse(sysOutStream.toString().contains(
        logMessage(containerId1, "syslog")));
    assertFalse(sysOutStream.toString().contains(
        logMessage(containerId2, "syslog")));
    assertFalse(sysOutStream.toString().contains(
        logMessage(containerId3, "syslog")));
    assertTrue(sysOutStream.toString().contains(
        logMessage(containerId3, "stdout")));
    assertFalse(sysOutStream.toString().contains(
        logMessage(containerId3, "stdout1234")));
    assertFalse(sysOutStream.toString().contains(
        createEmptyLog("empty")));
    sysOutStream.reset();

    // Check backward compatibility for -logFiles
    exitCode = cli.run(new String[] {"-applicationId", appId.toString(),
        "-logFiles", "stdout"});
    assertTrue("Failed with -logFiles", exitCode == 0);
    assertFalse("Failed with -logFiles", sysOutStream.toString().contains(
        logMessage(containerId1, "syslog")));
    assertFalse("Failed with -logFiles", sysOutStream.toString().contains(
        logMessage(containerId2, "syslog")));
    assertFalse("Failed with -logFiles", sysOutStream.toString().contains(
        logMessage(containerId3, "syslog")));
    assertTrue("Failed with -logFiles", sysOutStream.toString().contains(
        logMessage(containerId3, "stdout")));
    assertFalse("Failed with -logFiles", sysOutStream.toString().contains(
        logMessage(containerId3, "stdout1234")));
    assertFalse("Failed with -logFiles", sysOutStream.toString().contains(
        createEmptyLog("empty")));
    sysOutStream.reset();

    // Check -log_files supercedes -logFiles
    exitCode = cli.run(new String[] {"-applicationId", appId.toString(),
        "-log_files", "stdout", "-logFiles", "syslog"});
    assertTrue("Failed with -logFiles and -log_files", exitCode == 0);
    assertFalse("Failed with -logFiles and -log_files",
        sysOutStream.toString().contains(
        logMessage(containerId1, "syslog")));
    assertFalse("Failed with -logFiles and -log_files",
        sysOutStream.toString().contains(
        logMessage(containerId2, "syslog")));
    assertFalse("Failed with -logFiles and -log_files",
        sysOutStream.toString().contains(
        logMessage(containerId3, "syslog")));
    assertTrue("Failed with -logFiles and -log_files",
        sysOutStream.toString().contains(
        logMessage(containerId3, "stdout")));
    assertFalse("Failed with -logFiles and -log_files",
        sysOutStream.toString().contains(
        logMessage(containerId3, "stdout1234")));
    assertFalse("Failed with -logFiles and -log_files",
        sysOutStream.toString().contains(
        createEmptyLog("empty")));
    sysOutStream.reset();

    exitCode = cli.run(new String[] {"-applicationId", appId.toString(),
        "-log_files_pattern", "std*"});
    assertTrue(exitCode == 0);
    assertFalse(sysOutStream.toString().contains(
        logMessage(containerId1, "syslog")));
    assertFalse(sysOutStream.toString().contains(
        logMessage(containerId2, "syslog")));
    assertFalse(sysOutStream.toString().contains(
        logMessage(containerId3, "syslog")));
    assertTrue(sysOutStream.toString().contains(
        logMessage(containerId3, "stdout")));
    assertTrue(sysOutStream.toString().contains(
        logMessage(containerId3, "stdout1234")));
    assertFalse(sysOutStream.toString().contains(
        createEmptyLog("empty")));
    sysOutStream.reset();

    exitCode = cli.run(new String[] {"-applicationId", appId.toString(),
        "-log_files", "123"});
    assertTrue(exitCode == -1);
    assertTrue(sysErrStream.toString().contains(
        "Can not find any log file matching the pattern: [123] "
        + "for the application: " + appId.toString()));
    sysErrStream.reset();

    // specify the bytes which is larger than the actual file size,
    // we would get the full logs
    exitCode = cli.run(new String[] {"-applicationId", appId.toString(),
        "-log_files", "*", "-size", "10000" });
    assertTrue(exitCode == 0);
    assertTrue(sysOutStream.toByteArray().length == fullSize);
    sysOutStream.reset();

    // uploaded two logs for container1. The first log is empty.
    // The second one is not empty.
    // We can still successfully read logs for container1.
    exitCode =
        cli.run(new String[] { "-applicationId", appId.toString(),
            "-nodeAddress", nodeId.toString(), "-containerId",
            containerId1.toString() });
    assertTrue(exitCode == 0);
    assertTrue(sysOutStream.toString().contains(
        logMessage(containerId1, "syslog")));
    assertTrue(sysOutStream.toString().contains("LogLastModifiedTime"));
    assertTrue(!sysOutStream.toString().contains(
      "Logs for container " + containerId1.toString()
          + " are not present in this log-file."));
    sysOutStream.reset();

    exitCode = cli.run(new String[] {"-applicationId", appId.toString(),
        "-containerId", containerId3.toString(), "-log_files", "123" });
    assertTrue(exitCode == -1);
    assertTrue(sysErrStream.toString().contains(
        "Can not find any log file matching the pattern: [123] "
        + "for the container: " + containerId3
        + " within the application: " + appId.toString()));
    sysErrStream.reset();

    exitCode = cli.run(new String[] {"-applicationId", appId.toString(),
        "-containerId", containerId3.toString(), "-log_files", "stdout" });
    assertTrue(exitCode == 0);
    int fullContextSize = sysOutStream.toByteArray().length;
    String fullContext = sysOutStream.toString();
    sysOutStream.reset();

    String logMessage = logMessage(containerId3, "stdout");
    int fileContentSize = logMessage.getBytes().length;
    StringBuilder sb = new StringBuilder();
    String endOfFile = "End of LogType:stdout";
    sb.append("\n" + endOfFile + "\n");
    sb.append(StringUtils.repeat("*", endOfFile.length() + 50)
        + "\n\n");
    int tailContentSize = sb.toString().length();
    // specify how many bytes we should get from logs
    // specify a position number, it would get the first n bytes from
    // container log
    exitCode = cli.run(new String[] {"-applicationId", appId.toString(),
        "-containerId", containerId3.toString(), "-log_files", "stdout",
        "-size", "5"});
    assertTrue(exitCode == 0);
    Assert.assertEquals(new String(logMessage.getBytes(), 0, 5),
        new String(sysOutStream.toByteArray(),
        (fullContextSize - fileContentSize - tailContentSize), 5));
    sysOutStream.reset();

    // specify how many bytes we should get from an empty log
    exitCode = cli.run(new String[] {"-applicationId", appId.toString(),
        "-containerId", containerId1.toString(), "-log_files", "empty",
        "-size", "5"});
    assertTrue(exitCode == 0);
    assertTrue(sysOutStream.toString().contains(
        createEmptyLog("empty")));
    sysOutStream.reset();

    // specify a negative number, it would get the last n bytes from
    // container log
    exitCode = cli.run(new String[] {"-applicationId", appId.toString(),
        "-containerId", containerId3.toString(), "-log_files", "stdout",
        "-size", "-5"});
    assertTrue(exitCode == 0);
    Assert.assertEquals(new String(logMessage.getBytes(),
        logMessage.getBytes().length - 5, 5),
        new String(sysOutStream.toByteArray(),
        (fullContextSize - fileContentSize - tailContentSize), 5));
    sysOutStream.reset();

    long negative = (fullContextSize + 1000) * (-1);
    exitCode = cli.run(new String[] {"-applicationId", appId.toString(),
        "-containerId", containerId3.toString(), "-log_files", "stdout",
        "-size", Long.toString(negative)});
    assertTrue(exitCode == 0);
    Assert.assertEquals(fullContext, sysOutStream.toString());
    sysOutStream.reset();

    // Uploaded the empty log for container0.
    // We should see the message showing the log for container0
    // are not present.
    exitCode =
        cli.run(new String[] { "-applicationId", appId.toString(),
            "-nodeAddress", nodeId.toString(), "-containerId",
            containerId0.toString() });
    assertTrue(exitCode == -1);
    assertTrue(sysErrStream.toString().contains(
        "Can not find any log file matching the pattern"));
    sysErrStream.reset();

    // uploaded two logs for container3. The first log is named as syslog.
    // The second one is named as stdout.
    exitCode =
        cli.run(new String[] { "-applicationId", appId.toString(),
            "-nodeAddress", nodeId.toString(), "-containerId",
            containerId3.toString() });
    assertTrue(exitCode == 0);
    assertTrue(sysOutStream.toString().contains(
        logMessage(containerId3, "syslog")));
    assertTrue(sysOutStream.toString().contains(
        logMessage(containerId3, "stdout")));
    sysOutStream.reset();

    // set -log_files option as stdout
    // should only print log with the name as stdout
    exitCode =
        cli.run(new String[] { "-applicationId", appId.toString(),
            "-nodeAddress", nodeId.toString(), "-containerId",
            containerId3.toString() , "-log_files", "stdout"});
    assertTrue(exitCode == 0);
    assertTrue(sysOutStream.toString().contains(
        logMessage(containerId3, "stdout")));
    assertTrue(!sysOutStream.toString().contains(
        logMessage(containerId3, "syslog")));
    sysOutStream.reset();

    YarnClient mockYarnClientWithException =
        createMockYarnClientWithException();
    cli = new LogsCLIForTest(mockYarnClientWithException);
    cli.setConf(conf);

    exitCode =
        cli.run(new String[] { "-applicationId", appId.toString(),
            "-containerId", containerId3.toString() });
    assertTrue(exitCode == 0);
    assertTrue(sysOutStream.toString().contains(
        logMessage(containerId3, "syslog")));
    assertTrue(sysOutStream.toString().contains(
        logMessage(containerId3, "stdout")));
    assertTrue(sysOutStream.toString().contains(
        containerId3 + " on " + LogAggregationUtils.getNodeString(nodeId)));
    sysOutStream.reset();

    // The same should also work without the applicationId
    exitCode =
        cli.run(new String[] { "-containerId", containerId3.toString() });
    assertTrue(exitCode == 0);
    assertTrue(sysOutStream.toString().contains(
        logMessage(containerId3, "syslog")));
    assertTrue(sysOutStream.toString().contains(
        logMessage(containerId3, "stdout")));
    assertTrue(sysOutStream.toString().contains(
        containerId3 + " on " + LogAggregationUtils.getNodeString(nodeId)));
    sysOutStream.reset();

    exitCode = cli.run(new String[] { "-containerId", "invalid_container" });
    assertTrue(exitCode == -1);
    assertTrue(sysErrStream.toString().contains(
        "Invalid ContainerId specified"));
    sysErrStream.reset();

    // Uploaded the empty log for container4. We should see a message
    // showing the log for container4 is not present.
    exitCode =
        cli.run(new String[] {"-applicationId", appId.toString(),
            "-nodeAddress", badNodeId.toString(), "-containerId",
            containerId4.toString()});
    assertTrue(exitCode == -1);
    assertTrue(sysErrStream.toString().contains(
        "Can not find any log file matching the pattern"));
    sysErrStream.reset();

    fs.delete(new Path(remoteLogRootDir), true);
    fs.delete(new Path(rootLogDir), true);
  }