TailDirSource extends EventSource.Base { @Override synchronized public ReportEvent getMetrics() { ReportEvent rpt = super.getMetrics(); rpt.setLongMetric(A_FILESADDED, filesAdded.get()); rpt.setLongMetric(A_FILESDELETED, filesDeleted.get()); rpt.setLongMetric(A_SUBDIRSADDED, subdirsAdded.get()); rpt.setLongMetric(A_SUBDIRSDELETED, subdirsDeleted.get()); rpt.setLongMetric(A_FILESPRESENT, tail.cursors.size()); return rpt; } TailDirSource(File f, String regex);  TailDirSource(File f, String regex, boolean startFromEnd);  TailDirSource(File f, String regex, boolean startFromEnd,
      int recurseDepth);  TailDirSource(File f, String regex, boolean startFromEnd,
      int recurseDepth, String delimRegex, DelimMode delimMode); @Override synchronized void open(); @Override synchronized void close(); @Override synchronized ReportEvent getMetrics(); @Override Event next(); static SourceBuilder builder(); static final Logger LOG; static final String USAGE; final static String A_FILESADDED; final static String A_FILESDELETED; final static String A_FILESPRESENT; final static String A_SUBDIRSADDED; final static String A_SUBDIRSDELETED; }
@Test public void testExistingDir() throws IOException, InterruptedException { File tmpdir = FileUtil.mktempdir(); File subDir = new File(tmpdir, "subdir"); subDir.mkdirs(); TailDirSource src = new TailDirSource(tmpdir, ".*"); AccumulatorSink cnt = new AccumulatorSink("tailcount"); DirectDriver drv = new DirectDriver(src, cnt); genFiles(tmpdir, "foo", 10, 100); drv.start(); Clock.sleep(1000); assertEquals(1000, cnt.getCount()); drv.stop(); FileUtil.rmr(tmpdir); assertEquals(Long.valueOf(10), src.getMetrics().getLongMetric(TailDirSource.A_FILESADDED)); }
@Test public void testTailDirSourceStartFromEnd() throws IOException, FlumeSpecException, InterruptedException { File tmpdir = FileUtil.mktempdir(); genFiles(tmpdir, "foo", 10, 100); TailDirSource src = new TailDirSource(tmpdir, ".*", true); AccumulatorSink cnt = new AccumulatorSink("tailcount"); DirectDriver drv = new DirectDriver(src, cnt); drv.start(); Clock.sleep(1000); assertEquals(0, cnt.getCount()); addLinesToExistingFiles(tmpdir, 10); Clock.sleep(1000); assertEquals(10 * 10, cnt.getCount()); genFiles(tmpdir, "bar", 10, 100); Clock.sleep(1000); assertEquals(10 * 10 + 1000, cnt.getCount()); drv.stop(); FileUtil.rmr(tmpdir); assertEquals(Long.valueOf(20), src.getMetrics().getLongMetric(TailDirSource.A_FILESADDED)); }
@Test public void testTailDirsRecursively() throws IOException, FlumeSpecException, InterruptedException { File tmpdir = FileUtil.mktempdir(); genFiles(tmpdir, "2tail-old", 10, 100); File subDirL1 = new File(tmpdir, "subdir-level1"); subDirL1.mkdirs(); genFiles(subDirL1, "2tail-old1", 10, 10); File subDirL2 = new File(subDirL1, "subdir-level2"); subDirL2.mkdirs(); genFiles(subDirL2, "2tail-old2", 10, 20); File subDirL3 = new File(subDirL2, "subdir-level3"); subDirL3.mkdirs(); genFiles(subDirL3, "2tail-old3", 10, 30); TailDirSource src = new TailDirSource(tmpdir, ".(2tail)?.*", true, 2); AccumulatorSink cnt = new AccumulatorSink("tailcount"); DirectDriver drv = new DirectDriver(src, cnt); drv.start(); Clock.sleep(1000); assertEquals(0, cnt.getCount()); addLinesToExistingFiles(tmpdir, 10); addLinesToExistingFiles(subDirL1, 20); addLinesToExistingFiles(subDirL2, 30); addLinesToExistingFiles(subDirL3, 40); Clock.sleep(1000); int expEventsCount = 10 * 10 + 10 * 20 + 10 * 30; assertEquals(expEventsCount, cnt.getCount()); genFiles(tmpdir, "2tail-new", 10, 40); genFiles(subDirL1, "2tail-new1", 10, 30); genFiles(subDirL2, "2tail-new2", 10, 20); genFiles(subDirL3, "2tail-new3", 10, 10); Clock.sleep(1000); expEventsCount += 10 * 40 + 10 * 30 + 10 * 20; assertEquals(expEventsCount, cnt.getCount()); File newSubDirL2 = new File(subDirL1, "subdir-level2-new"); newSubDirL2.mkdirs(); genFiles(newSubDirL2, "2tail-new2-new", 10, 100); Clock.sleep(1000); expEventsCount += 10 * 100; assertEquals(expEventsCount, cnt.getCount()); FileUtil.rmr(subDirL3); FileUtil.rmr(subDirL2); Clock.sleep(1000); assertEquals(expEventsCount, cnt.getCount()); subDirL2.mkdirs(); genFiles(subDirL2, "2tail-new2", 10, 100); Clock.sleep(1000); expEventsCount += 10 * 100; assertEquals(expEventsCount, cnt.getCount()); drv.stop(); FileUtil.rmr(tmpdir); ReportEvent report = src.getMetrics(); assertEquals(Long.valueOf(80), report.getLongMetric(TailDirSource.A_FILESADDED)); assertEquals(Long.valueOf(20), report.getLongMetric(TailDirSource.A_FILESDELETED)); assertEquals(Long.valueOf(4), report.getLongMetric(TailDirSource.A_SUBDIRSADDED)); assertEquals(Long.valueOf(1), report.getLongMetric(TailDirSource.A_SUBDIRSDELETED)); }
@Test public void testCursorExhaustion() throws IOException, InterruptedException { Assume.assumeTrue(!OSUtils.isWindowsOS()); File tmpdir = FileUtil.mktempdir(); TailDirSource src = new TailDirSource(tmpdir, ".*"); AccumulatorSink cnt = new AccumulatorSink("tailcount"); DirectDriver drv = new DirectDriver(src, cnt); drv.start(); genFiles(tmpdir, "foo", 200, 10); Clock.sleep(1000); assertEquals(2000, cnt.getCount()); ReportEvent rpt1 = src.getMetrics(); assertEquals(Long.valueOf(200), rpt1.getLongMetric(TailDirSource.A_FILESPRESENT)); FileUtil.rmr(tmpdir); tmpdir.mkdirs(); Clock.sleep(1000); assertEquals(2000, cnt.getCount()); ReportEvent rpt = src.getMetrics(); assertEquals(rpt.getLongMetric(TailDirSource.A_FILESADDED), rpt.getLongMetric(TailDirSource.A_FILESDELETED)); assertEquals(Long.valueOf(0), rpt.getLongMetric(TailDirSource.A_FILESPRESENT)); drv.stop(); FileUtil.rmr(tmpdir); }