Drop deleted files from the file source (#310)

* Drop deleted files from the file source

Prior to this commit the file source has assumed that descriptors
to deleted files can be cleared out when a new file is dropped in
the path. This is a problem for clients where log files are made
only at boot and then deleted. We now clear files in the same way
we used too but also when NotFound is returned from io operations.

This resolves #244

Signed-off-by: Brian L. Troutwine <blt@postmates.com>

* Break up the file source

This commit breaks up the file source in a fashion which drops
FPs that are bound to paths which no longer exist, rotate faster
and with simpler code. Unfortunately we've changed how the file_watcher
read_line works and this has made the QC test for file_watcher sit
and spin. The next commit will have to correct that.

Signed-off-by: Brian L. Troutwine <blt@postmates.com>

* Adapt file source qc test to new API

With the changes made to the file_watcher we broke the quickcheck
tests. Well! They're now fixed. Mostly we had to cope with the extra
O's that come out of the file_watcher.

Signed-off-by: Brian L. Troutwine <blt@postmates.com>

src/bin/cernan.rs

@@ -465,7 +465,6 @@ fn main() {
 
     drop(args);
     drop(config_topology);
-    drop(level);
     drop(receivers);
 
     for jh in joins {

src/config.rs

@@ -690,13 +690,13 @@ pub fn parse_config_file(buffer: &str, verbosity: u64) -> Args {
                         //
                         // Someday a static analysis system will flag this as
                         // unsafe. Welcome.
-                        fl.max_read_lines = tbl.get("max_read_lines")
+                        fl.max_lines_read = tbl.get("max_read_lines")
                             .map(|mrl| {
                                 mrl.as_integer()
                                     .expect("could not parse sinks.wavefront.port") as
                                     usize
                             })
-                            .unwrap_or(fl.max_read_lines);
+                            .unwrap_or(fl.max_lines_read);
 
                         files.push(fl)
                     }
@@ -1406,7 +1406,7 @@ scripts-directory = "/foo/bar"
 
         assert_eq!(files[0].path, Some(PathBuf::from("/foo/bar.txt")));
         assert_eq!(files[0].forwards, vec!["sink.blech"]);
-        assert_eq!(files[0].max_read_lines, 10_000);
+        assert_eq!(files[0].max_lines_read, 10_000);
     }
 
     #[test]
@@ -1433,7 +1433,7 @@ scripts-directory = "/foo/bar"
         assert_eq!(files[0].forwards, vec!["sink.bar.blech"]);
 
         assert_eq!(files[1].path, Some(PathBuf::from("/foo/bar.txt")));
-        assert_eq!(files[1].max_read_lines, 10);
+        assert_eq!(files[1].max_lines_read, 10);
         assert_eq!(files[1].forwards, vec!["sink.blech"]);
     }
 }

src/sink/influxdb.rs

@@ -296,8 +296,6 @@ impl Sink for InfluxDB {
 
 #[cfg(test)]
 mod test {
-    extern crate quickcheck;
-
     use super::*;
     use chrono::{TimeZone, Utc};
     use metric::{TagMap, Telemetry};