EntityGroupFSTimelineStore extends CompositeService implements TimelineStore { @InterfaceAudience.Private @VisibleForTesting void cleanLogs(Path dirpath, long retainMillis) throws IOException { long now = Time.now(); RemoteIterator<FileStatus> iter = list(dirpath); while (iter.hasNext()) { FileStatus stat = iter.next(); if (isValidClusterTimeStampDir(stat)) { Path clusterTimeStampPath = stat.getPath(); MutableBoolean appLogDirPresent = new MutableBoolean(false); cleanAppLogDir(clusterTimeStampPath, retainMillis, appLogDirPresent); if (appLogDirPresent.isFalse() && (now - stat.getModificationTime() > retainMillis)) { deleteDir(clusterTimeStampPath); } } } } 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(new Path(testDoneDirPath, Long.toString(mainTestAppId.getClusterTimestamp())), "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, 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, 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()); }
@Test public void testCleanBuckets() throws Exception { Path clusterTimeStampDir1 = new Path(testDoneDirPath, Long.toString(sampleAppIds.get(0).getClusterTimestamp())); Path appDir1 = new Path(new Path(new Path( clusterTimeStampDir1, "0000"), "000"), sampleAppIds.get(0).toString()); Path appDir2 = new Path(new Path(new Path( clusterTimeStampDir1, "0000"), "001"), sampleAppIds.get(1).toString()); Path appDir3 = new Path(new Path(new Path( clusterTimeStampDir1, "0000"), "002"), sampleAppIds.get(2).toString()); Path appDir4 = new Path(new Path(new Path( clusterTimeStampDir1, "0001"), "000"), sampleAppIds.get(3).toString()); Path clusterTimeStampDir2 = new Path(testDoneDirPath, "1235"); Path clusterTimeStampDir3 = new Path(testDoneDirPath, "irrevelant"); Path appDir5 = new Path(new Path(new Path( clusterTimeStampDir3, "0000"), "000"), sampleAppIds.get(4).toString()); fs.mkdirs(appDir1); fs.mkdirs(appDir2); fs.mkdirs(appDir3); fs.mkdirs(appDir4); fs.mkdirs(clusterTimeStampDir2); fs.mkdirs(appDir5); Thread.sleep(2000); store.cleanLogs(testDoneDirPath, 1000); assertTrue(fs.exists(clusterTimeStampDir1)); assertFalse(fs.exists(appDir1)); assertFalse(fs.exists(appDir2)); assertFalse(fs.exists(appDir3)); assertFalse(fs.exists(appDir4)); assertFalse(fs.exists(clusterTimeStampDir2)); assertTrue(fs.exists(appDir5)); store.cleanLogs(testDoneDirPath, 1000); assertFalse(fs.exists(clusterTimeStampDir1)); }