Add HISTOGRAM support to the wavefront sink (#322)

This commit adds emission of the histogram summarization to the
wavefront sink. The name of the metric is smushed together with
the bin boundary and the count / timestamp happen where you'd
expect.

The bin is separated by '_' rather than a dot to avoid confusion
with bins that have a decimal value.

Resolves #308. Related to #307.

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

src/sink/wavefront.rs

@@ -483,9 +483,32 @@ impl Wavefront {
     ) -> () {
         let mut tag_buf = String::with_capacity(1_024);
         match value.kind() {
-            AggregationMethod::Histogram => {
-                unimplemented!();
-            }
+            AggregationMethod::Histogram => if let Some(bins) = value.bins() {
+                use quantiles::histogram::Bound;
+                fmt_tags(&value.tags, &mut tag_buf);
+                for &(bound, count) in bins {
+                    self.stats.push_str(&value.name);
+                    self.stats.push_str("_");
+                    match bound {
+                        Bound::Finite(bnd) => {
+                            self.stats.push_str(get_from_cache(&mut value_cache, bnd));
+                        }
+                        Bound::PosInf => {
+                            self.stats.push_str("pos_inf");
+                        }
+                    };
+                    self.stats.push_str(" ");
+                    self.stats.push_str(get_from_cache(&mut count_cache, count));
+                    self.stats.push_str(" ");
+                    self.stats
+                        .push_str(get_from_cache(&mut time_cache, value.timestamp));
+                    self.stats.push_str(" ");
+                    self.stats.push_str(&tag_buf);
+                    self.stats.push_str("\n");
+                }
+
+                tag_buf.clear();
+            },
             AggregationMethod::Sum => if let Some(v) = value.sum() {
                 self.stats.push_str(&value.name);
                 self.stats.push_str(" ");

src/source/internal.rs

@@ -242,7 +242,7 @@ impl Source for Internal {
                     self.chans
                 );
                 atom_non_zero_telem!(
-                    "cernan.sinks.wavefront.value.open",
+                    "cernan.sinks.wavefront.valve.open",
                     sink::wavefront::WAVEFRONT_VALVE_OPEN,
                     self.tags,
                     self.chans