Resolve graphite growth problem, introduce hopper 0.1.2 (#181)

We noticed that there was something fishy going on with graphite.
In particular, the number of lines push to the programmable_filter
grew _way_ too quickly for the inputs. It turns out, that fancy
line buffer we pass in to avoid allocing small strings repeatedly
was not being cleared and so everything that came in would be
appended there.

This commit also includes the work done as a part of
postmates/hopper:#5 by updating to 0.1.2. All messages will be
delivered in order--as they should have been--and hopper is now
quite a bit faster to boot as it can keep cramming in low-memory
when there's space.

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

Cargo.toml

@@ -16,7 +16,7 @@ clap = "2.10.0"
 fern = "0.3.5"
 flate2 = "0.2"
 glob = "0.2.11"
-hopper = "0.1.0"
+hopper = "0.1.2"
 hyper = "0.9"
 lazy_static = "0.2.1"
 libc = "0.2"

src/bin/cernan.rs

@@ -146,13 +146,13 @@ fn main() {
         let (flt_send, flt_recv) = hopper::channel(&config.config_path, &args.data_directory)
             .unwrap();
         sends.insert(config.config_path.clone(), flt_send);
-        let mut upstream_sends = Vec::new();
-        populate_forwards(&mut upstream_sends,
+        let mut downstream_sends = Vec::new();
+        populate_forwards(&mut downstream_sends,
                           &config.forwards,
                           &config.config_path,
                           &sends);
         joins.push(thread::spawn(move || {
-            cernan::filter::ProgrammableFilter::new(c).run(flt_recv, upstream_sends);
+            cernan::filter::ProgrammableFilter::new(c).run(flt_recv, downstream_sends);
         }));
     }
 

src/filter/mod.rs

@@ -25,18 +25,22 @@ fn event_in_fe(fe: FilterError) -> metric::Event {
 }
 
 pub trait Filter {
-    fn process(&mut self, event: metric::Event) -> Result<Vec<metric::Event>, FilterError>;
+    fn process(&mut self,
+               event: metric::Event,
+               res: &mut Vec<metric::Event>)
+               -> Result<(), FilterError>;
     fn run(&mut self, mut recv: hopper::Receiver<metric::Event>, mut chans: util::Channel) {
         let mut attempts = 0;
+        let mut events = Vec::with_capacity(64);
         loop {
             time::delay(attempts);
             match recv.next() {
                 None => attempts += 1,
                 Some(event) => {
                     attempts = 0;
-                    match self.process(event) {
-                        Ok(events) => {
-                            for ev in events {
+                    match self.process(event, &mut events) {
+                        Ok(()) => {
+                            for ev in events.drain(..) {
                                 util::send("filter", &mut chans, ev)
                             }
                         }

src/filter/programmable_filter.rs

@@ -370,7 +370,7 @@ impl ProgrammableFilter {
 }
 
 impl filter::Filter for ProgrammableFilter {
-    fn process(&mut self, event: metric::Event) -> Result<Vec<metric::Event>, filter::FilterError> {
+    fn process(&mut self, event: metric::Event, res: &mut Vec<metric::Event>) -> Result<(), filter::FilterError> {
         match event {
             metric::Event::Telemetry(mut m) => {
                 self.state.get_global("process_metric");
@@ -396,11 +396,13 @@ impl filter::Filter for ProgrammableFilter {
 
                 self.state.call(1, 0);
 
-                Ok(pyld.logs
-                    .into_iter()
-                    .map(|m| metric::Event::new_log(*m))
-                    .chain(pyld.metrics.into_iter().map(|m| metric::Event::new_telemetry(*m)))
-                    .collect())
+                for lg in pyld.logs {
+                    res.push(metric::Event::new_log(*lg));
+                }
+                for mt in pyld.metrics {
+                    res.push(metric::Event::new_telemetry(*mt));
+                }
+                Ok(())
             }
             metric::Event::TimerFlush => {
                 self.state.get_global("tick");
@@ -423,11 +425,13 @@ impl filter::Filter for ProgrammableFilter {
 
                 self.state.call(1, 0);
 
-                Ok(pyld.logs
-                    .into_iter()
-                    .map(|m| metric::Event::new_log(*m))
-                    .chain(pyld.metrics.into_iter().map(|m| metric::Event::new_telemetry(*m)))
-                    .collect())
+                for lg in pyld.logs {
+                    res.push(metric::Event::new_log(*lg));
+                }
+                for mt in pyld.metrics {
+                    res.push(metric::Event::new_telemetry(*mt));
+                }
+                Ok(())
             }
             metric::Event::Log(mut l) => {
                 self.state.get_global("process_log");
@@ -453,11 +457,13 @@ impl filter::Filter for ProgrammableFilter {
 
                 self.state.call(1, 0);
 
-                Ok(pyld.logs
-                    .into_iter()
-                    .map(|m| metric::Event::new_log(*m))
-                    .chain(pyld.metrics.into_iter().map(|m| metric::Event::new_telemetry(*m)))
-                    .collect())
+                for lg in pyld.logs {
+                    res.push(metric::Event::new_log(*lg));
+                }
+                for mt in pyld.metrics {
+                    res.push(metric::Event::new_telemetry(*mt));
+                }
+                Ok(())
             }
         }
     }

src/source/graphite.rs

@@ -90,6 +90,7 @@ fn handle_stream(mut chans: util::Channel, tags: Arc<metric::TagMap>, stream: Tc
                              &mut chans,
                              metric::Event::Telemetry(Arc::new(Some(m))));
                     }
+                    line.clear();
                 } else {
                     let metric = metric::Telemetry::new("cernan.graphite.bad_packet", 1.0)
                         .aggr_sum()
@@ -98,6 +99,7 @@ fn handle_stream(mut chans: util::Channel, tags: Arc<metric::TagMap>, stream: Tc
                          &mut chans,
                          metric::Event::Telemetry(Arc::new(Some(metric))));
                     error!("bad packet: {:?}", line);
+                    line.clear();
                 }
             } else {
                 break;

tests/programmable_filter.rs

@@ -26,9 +26,9 @@ mod integration {
                 .overlay_tag("bizz", "bazz");
             let event = metric::Event::new_telemetry(metric);
 
-            let res = cs.process(event.clone());
+            let mut events = Vec::new();
+            let res = cs.process(event.clone(), &mut events);
             assert!(res.is_ok());
-            let events = res.ok().unwrap();
             assert!(!events.is_empty());
             assert_eq!(events.len(), 1);
             assert_eq!(events[0], event);
@@ -58,9 +58,9 @@ mod integration {
             let orig_event = metric::Event::new_log(orig_log);
             let expected_event = metric::Event::new_log(expected_log);
 
-            let res = cs.process(orig_event.clone());
+            let mut events = Vec::new();
+            let res = cs.process(orig_event.clone(), &mut events);
             assert!(res.is_ok());
-            let events = res.ok().unwrap();
             assert!(!events.is_empty());
             assert_eq!(events.len(), 1);
             assert_eq!(events[0], expected_event);
@@ -87,9 +87,9 @@ mod integration {
             let orig_event = metric::Event::new_telemetry(orig_metric);
             let expected_event = metric::Event::new_telemetry(expected_metric);
 
-            let res = cs.process(orig_event.clone());
+            let mut events = Vec::new();
+            let res = cs.process(orig_event.clone(), &mut events);
             assert!(res.is_ok());
-            let events = res.ok().unwrap();
             assert!(!events.is_empty());
             assert_eq!(events.len(), 1);
             assert_eq!(events[0], expected_event);
@@ -114,9 +114,9 @@ mod integration {
 
             let orig_event = metric::Event::new_telemetry(orig_metric);
 
-            let res = cs.process(orig_event.clone());
+            let mut events = Vec::new();
+            let res = cs.process(orig_event.clone(), &mut events);
             assert!(res.is_ok());
-            let events = res.ok().unwrap();
             assert!(!events.is_empty());
             assert_eq!(events.len(), 1);
             assert_eq!(events[0], orig_event);
@@ -141,8 +141,10 @@ mod integration {
 
             let orig_event = metric::Event::new_telemetry(orig_metric);
 
-            let res = cs.process(orig_event);
+            let mut events = Vec::new();
+            let res = cs.process(orig_event, &mut events);
             assert!(res.is_err());
+            assert!(events.is_empty());
         }
 
         #[test]
@@ -169,9 +171,9 @@ mod integration {
             let orig_event = metric::Event::new_log(orig_log);
             let expected_event = metric::Event::new_log(expected_log);
 
-            let res = cs.process(orig_event);
+            let mut events = Vec::new();
+            let res = cs.process(orig_event, &mut events);
             assert!(res.is_ok());
-            let events = res.ok().unwrap();
             assert!(!events.is_empty());
             assert_eq!(events.len(), 1);
             assert_eq!(events[0], expected_event);
@@ -197,9 +199,9 @@ mod integration {
             let orig_event = metric::Event::new_telemetry(orig_metric);
             let expected_event = metric::Event::new_telemetry(expected_metric);
 
-            let res = cs.process(orig_event);
+            let mut events = Vec::new();
+            let res = cs.process(orig_event, &mut events);
             assert!(res.is_ok());
-            let events = res.ok().unwrap();
             assert!(!events.is_empty());
             assert_eq!(events.len(), 1);
             assert_eq!(events[0], expected_event);
@@ -229,12 +231,13 @@ mod integration {
 
             let flush = metric::Event::TimerFlush;
 
+            let mut events = Vec::new();
             for ev in &[metric0, metric1, metric2, log0, log1] {
-                let _ = cs.process(ev.clone());
+                let _ = cs.process(ev.clone(), &mut events);
             }
-            let res = cs.process(flush.clone());
+            events.clear();
+            let res = cs.process(flush.clone(), &mut events);
             assert!(res.is_ok());
-            let events = res.ok().unwrap();
 
             assert!(!events.is_empty());
             assert_eq!(events.len(), 2);
@@ -245,12 +248,13 @@ mod integration {
                        metric::Event::new_log(metric::LogLine::new("filters.keep_count",
                                                                    "count_per_tick: 5")));
 
+            events.clear();
             for ev in &[log2, log3] {
-                let _ = cs.process(ev.clone());
+                let _ = cs.process(ev.clone(), &mut events);
             }
-            let res = cs.process(flush);
+            events.clear();
+            let res = cs.process(flush, &mut events);
             assert!(res.is_ok());
-            let events = res.ok().unwrap();
 
             assert!(!events.is_empty());
             assert_eq!(events.len(), 2);
@@ -282,9 +286,9 @@ mod integration {
             let metric = metric::Telemetry::new(orig, 12.0);
             let event = metric::Event::new_telemetry(metric);
 
-            let res = cs.process(event);
+            let mut events = Vec::new();
+            let res = cs.process(event, &mut events);
             assert!(res.is_ok());
-            let events = res.ok().unwrap();
             assert!(!events.is_empty());
             assert_eq!(events.len(), 1);
 
@@ -320,9 +324,9 @@ mod integration {
             let metric = metric::Telemetry::new(orig, 12.0);
             let event = metric::Event::new_telemetry(metric);
 
-            let res = cs.process(event);
+            let mut events = Vec::new();
+            let res = cs.process(event, &mut events);
             assert!(res.is_ok());
-            let events = res.ok().unwrap();
             assert!(!events.is_empty());
             assert_eq!(events.len(), 1);