EntityGroupFSTimelineStore extends CompositeService implements TimelineStore { @InterfaceAudience.Private @VisibleForTesting void cleanLogs(Path dirpath, FileSystem fs, long retainMillis) throws IOException { long now = Time.now(); RemoteIterator<FileStatus> iter = list(dirpath); while (iter.hasNext()) { FileStatus stat = iter.next(); if (stat.isDirectory()) { ApplicationId appId = parseApplicationId(dirpath.getName()); if (appId != null) { if (shouldCleanAppLogDir(dirpath, now, fs, retainMillis)) { try { LOG.info("Deleting {}", dirpath); if (!fs.delete(dirpath, true)) { LOG.error("Unable to remove " + dirpath); } metrics.incrLogsDirsCleaned(); } catch (IOException e) { LOG.error("Unable to remove " + dirpath, e); } } } else { cleanLogs(stat.getPath(), fs, retainMillis); } } } } EntityGroupFSTimelineStore(); @Override TimelineEntities getEntities(String entityType, Long limit,
      Long windowStart, Long windowEnd, String fromId, Long fromTs,
      NameValuePair primaryFilter, Collection<NameValuePair> secondaryFilters,
      EnumSet<Field> fieldsToRetrieve, CheckAcl checkAcl); @Override TimelineEntity getEntity(String entityId, String entityType,
      EnumSet<Field> fieldsToRetrieve); @Override TimelineEvents getEntityTimelines(String entityType,
      SortedSet<String> entityIds, Long limit, Long windowStart,
      Long windowEnd, Set<String> eventTypes); @Override TimelineDomain getDomain(String domainId); @Override TimelineDomains getDomains(String owner); @Override TimelinePutResponse put(TimelineEntities data); @Override void put(TimelineDomain domain);  }
@Test public void testCleanLogs() throws Exception { String appDirName = mainTestAppId.toString(); String attemptDirName = ApplicationAttemptId.appAttemptIdStrPrefix + appDirName + "_1"; Path irrelevantFilePath = new Path( testDoneDirPath, "irrelevant.log"); FSDataOutputStream stream = fs.create(irrelevantFilePath); stream.close(); Path irrelevantDirPath = new Path(testDoneDirPath, "irrelevant"); fs.mkdirs(irrelevantDirPath); Path doneAppHomeDir = new Path(new Path(testDoneDirPath, "0000"), "001"); Path appDirClean = new Path(doneAppHomeDir, appDirName); Path attemptDirClean = new Path(appDirClean, attemptDirName); fs.mkdirs(attemptDirClean); Path filePath = new Path(attemptDirClean, "test.log"); stream = fs.create(filePath); stream.close(); Path appDirHoldByFile = new Path(doneAppHomeDir, appDirName + "1"); Path attemptDirHoldByFile = new Path(appDirHoldByFile, attemptDirName); fs.mkdirs(attemptDirHoldByFile); Path filePathHold = new Path(attemptDirHoldByFile, "test1.log"); stream = fs.create(filePathHold); stream.close(); Path appDirHoldByDir = new Path(doneAppHomeDir, appDirName + "2"); Path attemptDirHoldByDir = new Path(appDirHoldByDir, attemptDirName); fs.mkdirs(attemptDirHoldByDir); Path dirPathHold = new Path(attemptDirHoldByDir, "hold"); fs.mkdirs(dirPathHold); Path appDirEmpty = new Path(doneAppHomeDir, appDirName + "3"); Path attemptDirEmpty = new Path(appDirEmpty, attemptDirName); fs.mkdirs(attemptDirEmpty); Path dirPathEmpty = new Path(attemptDirEmpty, "empty"); fs.mkdirs(dirPathEmpty); MutableCounterLong dirsCleaned = store.metrics.getLogsDirsCleaned(); long before = dirsCleaned.value(); store.cleanLogs(testDoneDirPath, fs, 10000); assertTrue(fs.exists(irrelevantDirPath)); assertTrue(fs.exists(irrelevantFilePath)); assertTrue(fs.exists(filePath)); assertTrue(fs.exists(filePathHold)); assertTrue(fs.exists(dirPathHold)); assertTrue(fs.exists(dirPathEmpty)); Thread.sleep(2000); stream = fs.append(filePathHold); stream.writeBytes("append"); stream.close(); fs.mkdirs(new Path(dirPathHold, "holdByMe")); store.cleanLogs(testDoneDirPath, fs, 1000); assertTrue(fs.exists(irrelevantDirPath)); assertTrue(fs.exists(irrelevantFilePath)); assertTrue(fs.exists(filePathHold)); assertTrue(fs.exists(dirPathHold)); assertTrue(fs.exists(doneAppHomeDir)); assertFalse(fs.exists(appDirClean)); assertFalse(fs.exists(appDirEmpty)); assertEquals(before + 2L, dirsCleaned.value()); }