HadoopArchiveLogs implements Tool { @VisibleForTesting void checkFilesAndSeedApps(FileSystem fs, Path remoteRootLogDir, String suffix, Path workingDir) throws IOException { for (RemoteIterator<FileStatus> userIt = fs.listStatusIterator(remoteRootLogDir); userIt.hasNext();) { Path userLogPath = userIt.next().getPath(); try { for (RemoteIterator<FileStatus> appIt = fs.listStatusIterator(new Path(userLogPath, suffix)); appIt.hasNext();) { Path appLogPath = appIt.next().getPath(); try { FileStatus[] files = fs.listStatus(appLogPath); if (files.length >= minNumLogFiles) { boolean eligible = true; long totalFileSize = 0L; for (FileStatus file : files) { if (file.getPath().getName().equals(appLogPath.getName() + ".har")) { eligible = false; if (verbose) { LOG.info("Skipping " + appLogPath.getName() + " due to existing .har file"); } break; } totalFileSize += file.getLen(); if (totalFileSize > maxTotalLogsSize) { eligible = false; if (verbose) { LOG.info("Skipping " + appLogPath.getName() + " due to " + "total file size being too large (" + totalFileSize + " > " + maxTotalLogsSize + ")"); } break; } } if (eligible) { if (verbose) { LOG.info("Adding " + appLogPath.getName() + " for user " + userLogPath.getName()); } AppInfo context = new AppInfo(); context.setAppId(appLogPath.getName()); context.setUser(userLogPath.getName()); context.setSuffix(suffix); context.setRemoteRootLogDir(remoteRootLogDir); context.setWorkingDir(workingDir); eligibleApplications.add(context); } } else { if (verbose) { LOG.info("Skipping " + appLogPath.getName() + " due to not " + "having enough log files (" + files.length + " < " + minNumLogFiles + ")"); } } } catch (IOException ioe) { if (verbose) { LOG.info("Skipping logs under " + appLogPath + " due to " + ioe.getMessage()); } } } } catch (IOException ioe) { if (verbose) { LOG.info("Skipping all logs under " + userLogPath + " due to " + ioe.getMessage()); } } } } HadoopArchiveLogs(Configuration conf); static void main(String[] args); @Override int run(String[] args); @Override void setConf(Configuration conf); @Override Configuration getConf();  }
@Test(timeout = 10000) public void testCheckFilesAndSeedApps() throws Exception { Configuration conf = new Configuration(); HadoopArchiveLogs hal = new HadoopArchiveLogs(conf); FileSystem fs = FileSystem.getLocal(conf); Path rootLogDir = new Path("target", "logs"); String suffix = "logs"; Path logDir = new Path(rootLogDir, new Path(USER, suffix)); fs.delete(logDir, true); Assert.assertFalse(fs.exists(logDir)); fs.mkdirs(logDir); ApplicationId appId1 = ApplicationId.newInstance(CLUSTER_TIMESTAMP, 1); Path app1Path = new Path(logDir, appId1.toString()); fs.mkdirs(app1Path); ApplicationId appId2 = ApplicationId.newInstance(CLUSTER_TIMESTAMP, 2); Path app2Path = new Path(logDir, appId2.toString()); fs.mkdirs(app2Path); createFile(fs, new Path(app2Path, "file1"), 1); hal.minNumLogFiles = 2; ApplicationId appId3 = ApplicationId.newInstance(CLUSTER_TIMESTAMP, 3); Path app3Path = new Path(logDir, appId3.toString()); fs.mkdirs(app3Path); createFile(fs, new Path(app3Path, "file1"), 2); createFile(fs, new Path(app3Path, "file2"), 5); hal.maxTotalLogsSize = FILE_SIZE_INCREMENT * 6; ApplicationId appId4 = ApplicationId.newInstance(CLUSTER_TIMESTAMP, 4); Path app4Path = new Path(logDir, appId4.toString()); fs.mkdirs(app4Path); createFile(fs, new Path(app4Path, appId4 + ".har"), 1); ApplicationId appId5 = ApplicationId.newInstance(CLUSTER_TIMESTAMP, 5); Path app5Path = new Path(logDir, appId5.toString()); fs.mkdirs(app5Path); createFile(fs, new Path(app5Path, "file1"), 2); createFile(fs, new Path(app5Path, "file2"), 3); Assert.assertEquals(0, hal.eligibleApplications.size()); hal.checkFilesAndSeedApps(fs, rootLogDir, suffix, new Path(rootLogDir, "archive-logs-work")); Assert.assertEquals(1, hal.eligibleApplications.size()); Assert.assertEquals(appId5.toString(), hal.eligibleApplications.iterator().next().getAppId()); }