Increase existing telemetry, adjust file delays (#237)

This commit contains a smattering of adjustments. The main show
is the inclusion of Internal setup code in bin/cernan -- oops, it
was only present on my disk -- and fiddling with the way backoffs
occur in file.rs. We reduce the CPU load of each FileServer thread
by only gradually scaling down to faster polls. This means a group
of slow files will be polled more slowly that a group with one
fast file in it.

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

examples/configs/basic.toml

@@ -7,6 +7,9 @@ flush-interval = 5
 source = "cernan"
 
 [sources]
+  [sources.internal]
+  forwards = ["sinks.console", "sinks.null"]
+
   [sources.statsd.primary]
   enabled = true
   port = 8125

src/bin/cernan.rs

@@ -161,6 +161,18 @@ fn main() {
             cernan::source::NativeServer::new(native_server_send, config).run();
         }))
     }
+
+    let internal_config = args.internal;
+    let mut internal_send = Vec::new();
+    populate_forwards(&mut internal_send,
+                      Some(&mut flush_sends),
+                      &internal_config.forwards,
+                      &internal_config.config_path,
+                      &sends);
+    joins.push(thread::spawn(move || {
+        cernan::source::Internal::new(internal_send, internal_config).run();
+    }));
+
     for config in args.statsds.values() {
         let c = (*config).clone();
         let mut statsd_sends = Vec::new();

src/sink/wavefront.rs

@@ -1,6 +1,7 @@
 use buckets::Buckets;
 use metric::{AggregationMethod, LogLine, TagMap, Telemetry};
 use sink::{Sink, Valve};
+use source::report_telemetry;
 use std::cmp;
 use std::io::Write as IoWrite;
 use std::net::TcpStream;
@@ -143,6 +144,8 @@ impl Sink for Wavefront {
 
     fn flush(&mut self) {
         loop {
+            report_telemetry("cernan.sinks.wavefront.delivery_attempts",
+                             self.delivery_attempts as f64);
             if self.delivery_attempts > 0 {
                 debug!("delivery attempts: {}", self.delivery_attempts);
             }

src/source/file.rs

@@ -206,7 +206,7 @@ impl Source for FileServer {
                 }
             }
             let start = Instant::now();
-            let mut attempts = 0;
+            let mut attempts: u32 = 0;
             loop {
                 // file poll
                 if fp_map.is_empty() {
@@ -220,7 +220,7 @@ impl Source for FileServer {
                         let mut lines_read = 0;
                         match file.read_line(&mut buffer) {
                             Ok(sz) => {
-                                attempts = 0;
+                                attempts = attempts.saturating_sub(1);
                                 if sz > 0 {
                                     lines_read += 1;
                                     buffer.pop();

src/source/internal.rs

@@ -68,10 +68,10 @@ pub fn report_telemetry<S>(name: S, value: f64) -> ()
 /// floor.
 impl Source for Internal {
     fn run(&mut self) {
-        let mut attempts = 0;
+        let mut attempts: u32 = 0;
         loop {
             if let Some(mut telem) = Q.lock().unwrap().pop_front() {
-                attempts -= 1;
+                attempts = attempts.saturating_sub(1);
                 if !self.chans.is_empty() {
                     telem = telem.overlay_tags_from_map(&self.tags);
                     util::send("internal",

src/source/mod.rs

@@ -8,7 +8,7 @@ mod statsd;
 pub use self::file::{FileServer, FileServerConfig};
 pub use self::flush::FlushTimer;
 pub use self::graphite::{Graphite, GraphiteConfig};
-pub use self::internal::{Internal, InternalConfig};
+pub use self::internal::{Internal, InternalConfig, report_telemetry};
 pub use self::native::{NativeServer, NativeServerConfig};
 pub use self::statsd::{Statsd, StatsdConfig};
 

src/time.rs

@@ -23,7 +23,6 @@ pub fn update_time() {
         thread::sleep(dur);
         let now = UTC::now().timestamp() as usize;
         let order = Ordering::Relaxed;
-        trace!("updated cernan {:?} now, is: {}", order, now);
         NOW.store(now, order);
     }
 }