Add distribution support to semantic-metrics FfwdhttpReporter

core/src/main/java/com/spotify/metrics/core/SemanticMetricDistribution.java

@@ -47,7 +47,7 @@
  * From P99.9 to P99.999 the error rate is slightly higher than 2%.
  *
  */
-public final class SemanticMetricDistribution implements Distribution {
+public class SemanticMetricDistribution implements Distribution {
 
     private static final int COMPRESSION_DEFAULT_LEVEL = 100;
     private final AtomicReference<TDigest> distRef;

examples/pom.xml

@@ -43,5 +43,9 @@
       <artifactId>mockito-core</artifactId>
       <scope>test</scope>
     </dependency>
+      <dependency>
+          <groupId>com.spotify.metrics</groupId>
+          <artifactId>semantic-metrics-ffwd-http-reporter</artifactId>
+      </dependency>
   </dependencies>
 </project>

examples/src/main/java/com/spotify/metrics/example/FfwdHttpReportExample.java

@@ -0,0 +1,58 @@
+/*
+ * Copyright (c) 2016 Spotify AB.
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.spotify.metrics.example;
+
+import com.spotify.ffwd.http.HttpClient;
+import com.spotify.metrics.core.MetricId;
+import com.spotify.metrics.core.SemanticMetricRegistry;
+import com.spotify.metrics.ffwdhttp.FastForwardHttpReporter;
+
+
+public class FfwdHttpReportExample {
+    private static final MetricId APP_PREFIX = MetricId.build("distribution-metric-example");
+    private static final SemanticMetricRegistry registry = new SemanticMetricRegistry();
+
+    public static void main(final String[] args) throws Exception {
+        final HttpClient.Builder builder = new HttpClient.Builder();
+        final HttpClient httpClient = builder.build();
+
+        //client should send a batch of 5 data points
+        for (int i = 0; i < 1000; i++) {
+            registry.distribution(MetricId.build("distributionExample")
+                    .tagged("what", "service-latency").tagged("host",
+                            "host" + i / 200)).record(Math.random());
+        }
+
+        final FastForwardHttpReporter reporter = FastForwardHttpReporter
+                .forRegistry(registry, httpClient)
+                .prefix(APP_PREFIX)
+                .build();
+
+
+        reporter.start();
+
+        System.out.println("Sending dynamic metrics...");
+        System.in.read();
+
+    }
+}
+

ffwd-http-reporter/pom.xml

@@ -25,7 +25,7 @@
     <dependency>
       <groupId>com.spotify.ffwd</groupId>
       <artifactId>ffwd-http-client</artifactId>
-      <version>0.4.0</version>
+      <version>0.4.4</version>
     </dependency>
 
     <dependency>

ffwd-http-reporter/src/main/java/com/spotify/metrics/ffwdhttp/FastForwardHttpReporter.java

@@ -30,9 +30,12 @@
 import com.codahale.metrics.Snapshot;
 import com.codahale.metrics.Timer;
 import com.google.common.collect.Sets;
-import com.spotify.ffwd.http.Batch;
+import com.google.protobuf.ByteString;
 import com.spotify.ffwd.http.HttpClient;
+import com.spotify.ffwd.http.model.v2.Batch;
+import com.spotify.ffwd.http.model.v2.Value;
 import com.spotify.metrics.core.DerivingMeter;
+import com.spotify.metrics.core.Distribution;
 import com.spotify.metrics.core.MetricId;
 import com.spotify.metrics.core.SemanticMetricFilter;
 import com.spotify.metrics.core.SemanticMetricRegistry;
@@ -122,7 +125,7 @@ public static final class Builder {
         private TagExtractor tagExtractor;
 
         private Set<Percentile> histogramPercentiles =
-            Sets.newHashSet(new Percentile(0.75), new Percentile(0.99));
+                Sets.newHashSet(new Percentile(0.75), new Percentile(0.99));
         private ScheduledExecutorService executorService;
 
         public Builder(SemanticMetricRegistry registry, HttpClient client) {
@@ -212,7 +215,7 @@ public FastForwardHttpReporter build() throws IOException {
             }
 
             final TagExtractor tagExtractor =
-                this.tagExtractor != null ? this.tagExtractor : new NoopTagExtractor();
+                    this.tagExtractor != null ? this.tagExtractor : new NoopTagExtractor();
 
             final boolean executorOwner;
             final ScheduledExecutorService executorService;
@@ -224,7 +227,7 @@ public FastForwardHttpReporter build() throws IOException {
                 executorOwner = true;
             }
             return new FastForwardHttpReporter(registry, prefix, unit, time, client,
-                histogramPercentiles, clock, tagExtractor, executorService, executorOwner);
+                    histogramPercentiles, clock, tagExtractor, executorService, executorOwner);
         }
 
         private ScheduledExecutorService createExecutor() {
@@ -238,21 +241,21 @@ private void report() {
         final long timestamp = clock.currentTimeMillis();
 
         for (@SuppressWarnings("rawtypes") Map.Entry<MetricId, Gauge> entry : registry
-            .getGauges(FILTER_ALL)
-            .entrySet()) {
+                .getGauges(FILTER_ALL)
+                .entrySet()) {
             final BatchBuilder builder = createBuilder(points, timestamp, entry.getKey(), "gauge");
             reportGauge(builder, entry.getValue());
         }
 
         for (Map.Entry<MetricId, Counter> entry : registry.getCounters(FILTER_ALL).entrySet()) {
             final BatchBuilder builder =
-                createBuilder(points, timestamp, entry.getKey(), "counter");
+                    createBuilder(points, timestamp, entry.getKey(), "counter");
             reportCounter(builder, entry.getValue());
         }
 
         for (Map.Entry<MetricId, Histogram> entry : registry.getHistograms(FILTER_ALL).entrySet()) {
             final BatchBuilder builder =
-                createBuilder(points, timestamp, entry.getKey(), "histogram");
+                    createBuilder(points, timestamp, entry.getKey(), "histogram");
             reportHistogram(builder, entry.getValue().getSnapshot());
         }
 
@@ -267,30 +270,38 @@ private void report() {
         }
 
         for (Map.Entry<MetricId, DerivingMeter> entry : registry
-            .getDerivingMeters(FILTER_ALL)
-            .entrySet()) {
+                .getDerivingMeters(FILTER_ALL)
+                .entrySet()) {
             final BatchBuilder builder =
-                createBuilder(points, timestamp, entry.getKey(), "deriving-meter");
+                    createBuilder(points, timestamp, entry.getKey(), "deriving-meter");
             reportDerivingMeter(builder, entry.getValue());
         }
 
+        for (Map.Entry<MetricId, Distribution> entry : registry
+                .getDistributions(FILTER_ALL)
+                .entrySet()) {
+            final BatchBuilder builder =
+                    createBuilder(points, timestamp, entry.getKey(), "distribution");
+            reportDistribution(builder, entry.getValue());
+        }
+
         final Map<String, String> commonTags = tagExtractor.addTags(prefix.getTags());
-        final Batch batch = new Batch(commonTags, points);
+        final Batch batch = new Batch(commonTags, createResource(), points);
 
         client.sendBatch(batch).toCompletable().await();
     }
 
     private BatchBuilder createBuilder(
-        final List<Batch.Point> points, final long timestamp, final MetricId id,
-        final String metricType
+            final List<Batch.Point> points, final long timestamp, final MetricId id,
+            final String metricType
     ) {
         final String key = joinKeys(prefix, id);
         final String unit = getUnit(id.getTags());
         return new BatchBuilder(points, timestamp, key, id.getTags(), unit, metricType);
     }
 
     private void reportGauge(
-        final BatchBuilder builder, @SuppressWarnings("rawtypes") Gauge value
+            final BatchBuilder builder, @SuppressWarnings("rawtypes") Gauge value
     ) {
         if (value == null) {
             return;
@@ -308,34 +319,34 @@ private double convert(Object value) {
     }
 
     private void reportCounter(
-        final BatchBuilder builder, final Counting value
+            final BatchBuilder builder, final Counting value
     ) {
         builder.buildPoint("count", value.getCount());
     }
 
     private void reportMeter(
-        final BatchBuilder builder, Meter value
+            final BatchBuilder builder, Meter value
     ) {
         reportMetered(builder, value);
         reportCounter(builder, value);
     }
 
     private void reportTimer(
-        final BatchBuilder builderIn, Timer value
+            final BatchBuilder builderIn, Timer value
     ) {
         final BatchBuilder builder = builderIn.withUnit("ns");
         reportMetered(builder, value);
         reportHistogram(builder, value.getSnapshot());
     }
 
     private void reportDerivingMeter(
-        final BatchBuilder builder, DerivingMeter value
+            final BatchBuilder builder, DerivingMeter value
     ) {
         reportMetered(builder, value);
     }
 
     private void reportHistogram(
-        final BatchBuilder builder, final Snapshot s
+            final BatchBuilder builder, final Snapshot s
     ) {
         builder.buildPoint("min", s.getMin());
         builder.buildPoint("max", s.getMax());
@@ -347,22 +358,30 @@ private void reportHistogram(
     }
 
     private void reportHistogramQuantiles(
-        final BatchBuilder builder, final Snapshot s
+            final BatchBuilder builder, final Snapshot s
     ) {
         for (Percentile q : histogramPercentiles) {
             builder.buildPoint(q.getPercentileString(), s.getValue(q.getQuantile()));
         }
     }
 
     private void reportMetered(
-        final BatchBuilder builder, final Metered value
+            final BatchBuilder builder, final Metered value
     ) {
         final BatchBuilder b = builder.withUnit(builder.getUnit() + "/s");
 
         b.buildPoint("1m", value.getOneMinuteRate());
         b.buildPoint("5m", value.getFiveMinuteRate());
     }
 
+    private void reportDistribution(
+            final BatchBuilder builder, Distribution distribution) {
+        if (distribution.getCount() == 0) {
+            return;
+        }
+        builder.buildPoint("distribution", distribution.getValueAndFlush());
+    }
+
     private String getUnit(final Map<String, String> tags) {
         final String unit = tags.get("unit");
 
@@ -391,6 +410,10 @@ private String joinKeys(MetricId... parts) {
         return key.toString();
     }
 
+    private static Map<String, String> createResource() {
+        return new HashMap<>();
+    }
+
     public void start() {
         if (running.getAndSet(true)) {
             return;
@@ -431,8 +454,8 @@ private static class BatchBuilder {
         private final String metricType;
 
         public BatchBuilder(
-            final List<Batch.Point> points, final long timestamp, final String key,
-            final Map<String, String> tags, final String unit, final String metricType
+                final List<Batch.Point> points, final long timestamp, final String key,
+                final Map<String, String> tags, final String unit, final String metricType
         ) {
             this.points = points;
             this.timestamp = timestamp;
@@ -455,11 +478,17 @@ public BatchBuilder withMetricType(final String metricType) {
         }
 
         public void buildPoint(final String stat, final double value) {
-            points.add(new Batch.Point(key, statsMap(stat), value, timestamp));
+            points.add(new Batch.Point(key, statsMap(stat), createResource(),
+                    Value.DoubleValue.create(value), timestamp));
+        }
+
+        public void buildPoint(final String stat, final ByteString value) {
+            points.add(new Batch.Point(key, statsMap(stat), createResource(),
+                    Value.DistributionValue.create(value), timestamp));
         }
 
         private Map<String, String> statsMap(
-            final String stat
+                final String stat
         ) {
             final boolean sameUnit = this.unit.equals(tags.get("unit"));
             final boolean sameMetricType = this.metricType.equals(tags.get("unit"));