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;
@@ -274,8 +277,15 @@ private void report() {
             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();
     }
@@ -289,8 +299,12 @@ private BatchBuilder createBuilder(
         return new BatchBuilder(points, timestamp, key, id.getTags(), unit, metricType);
     }
 
+    private static Map<String, String> createResource() {
+        return new HashMap<>();
+    }
+
     private void reportGauge(
-        final BatchBuilder builder, @SuppressWarnings("rawtypes") Gauge value
+            final BatchBuilder builder, @SuppressWarnings("rawtypes") Gauge value
     ) {
         if (value == null) {
             return;
@@ -299,6 +313,13 @@ private void reportGauge(
         builder.buildPoint(null, convert(value.getValue()));
     }
 
+    private void reportDistribution(BatchBuilder builder, Distribution distribution) {
+        if (distribution.getCount() == 0) {
+            return;
+        }
+        builder.buildPoint("distribution", distribution.getValueAndFlush());
+    }
+
     private double convert(Object value) {
         if (value instanceof Number) {
             return Number.class.cast(value).doubleValue();
@@ -455,7 +476,13 @@ 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(

ffwd-http-reporter/src/test/java/com/spotify/metrics/ffwd/FastForwardHttpReporterTest.java

@@ -2,6 +2,8 @@
 
 import static com.google.common.collect.ImmutableMap.of;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
 import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.atLeastOnce;
 import static org.mockito.Mockito.doReturn;
@@ -12,13 +14,20 @@
 import com.google.common.base.Suppliers;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
-import com.spotify.ffwd.http.Batch;
+import com.google.protobuf.ByteString;
+import com.spotify.ffwd.http.model.v2.Batch;
 import com.spotify.ffwd.http.HttpClient;
+import com.spotify.ffwd.http.model.v2.Value;
+import com.spotify.metrics.core.Distribution;
 import com.spotify.metrics.core.MetricId;
+import com.spotify.metrics.core.SemanticMetricBuilder;
+import com.spotify.metrics.core.SemanticMetricDistribution;
+import com.spotify.metrics.core.SemanticMetricFilter;
 import com.spotify.metrics.core.SemanticMetricRegistry;
 import com.spotify.metrics.ffwdhttp.Clock;
 import com.spotify.metrics.ffwdhttp.FastForwardHttpReporter;
 import com.spotify.metrics.tags.EnvironmentTagExtractor;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
@@ -30,20 +39,32 @@
 import org.junit.runner.RunWith;
 import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
+import org.mockito.Mockito;
 import org.mockito.runners.MockitoJUnitRunner;
 import rx.Observable;
 
 @RunWith(MockitoJUnitRunner.class)
 public class FastForwardHttpReporterTest {
+    private static final Value.DoubleValue VALUE_0 = Value.DoubleValue.create(0);
+    private static final ByteString VALUE_1 = ByteString.copyFromUtf8("AAAAAAAAAAA");
+    private static final String APP_PREFIX = "prefix";
+    private static final Map<String,String> RESOURCE = new HashMap<>();
     private static final int REPORTING_PERIOD = 50;
     private static final long TIME = 42L;
     private FastForwardHttpReporter reporter;
 
+
     private SemanticMetricRegistry registry;
     private Clock.Fixed fixedClock;
     @Mock
     private HttpClient httpClient;
 
+    @Mock
+    private SemanticMetricDistribution distribution;
+
+    @Mock
+    private SemanticMetricBuilder<SemanticMetricDistribution> semanticMetricBuilder;
+
     private Map<String, String> commonTags;
     DeterministicScheduler executorService;
 
@@ -52,23 +73,29 @@ public void setUp() throws Exception {
         registry = new SemanticMetricRegistry();
         fixedClock = new Clock.Fixed(0L);
 
+
         commonTags = of("foo", "bar");
         executorService = new DeterministicScheduler();
 
         reporter = FastForwardHttpReporter
             .forRegistry(registry, httpClient)
             .schedule(REPORTING_PERIOD, TimeUnit.MILLISECONDS)
-            .prefix(MetricId.build("prefix").tagged(commonTags))
+            .prefix(MetricId.build(APP_PREFIX).tagged(commonTags))
             .clock(fixedClock)
             .executorService(executorService)
             .build();
     }
 
+    private void setupDistribution(){
+        Mockito.when(semanticMetricBuilder.newMetric()).thenReturn(distribution);
+        Mockito.when(distribution.getValueAndFlush()).thenReturn(VALUE_1);
+        registry.getOrAdd(MetricId.build("distribution"), semanticMetricBuilder);
+    }
+
     @Test
     public void someReporting() {
         doReturn(Observable.<Void>just(null)).when(httpClient).sendBatch(any(Batch.class));
         fixedClock.setCurrentTime(TIME);
-
         registry.counter(MetricId.build("counter"));
         registry.derivingMeter(MetricId.build("deriving-meter"));
         registry.histogram(MetricId.build("histogram"));
@@ -83,63 +110,68 @@ public Double getValue() {
                 }
             });
 
+        setupDistribution();
+
         final Set<Batch.Point> expected = new HashSet<>();
+        expected.add(new Batch.Point("prefix.distribution",
+                of( "unit", "n", "stat", "distribution","metric_type",
+                        "distribution"), RESOURCE, Value.DistributionValue.create(VALUE_1), TIME));
         expected.add(new Batch.Point("prefix.counter",
-            of("unit", "n", "stat", "count", "metric_type", "counter"), 0, TIME));
+            of("unit", "n", "stat", "count", "metric_type", "counter"), RESOURCE, VALUE_0, TIME));
         expected.add(new Batch.Point("prefix.deriving-meter",
-            of("unit", "n/s", "stat", "5m", "metric_type", "deriving-meter"), 0, TIME));
+            of("unit", "n/s", "stat", "5m", "metric_type", "deriving-meter"), RESOURCE, VALUE_0, TIME));
         expected.add(new Batch.Point("prefix.deriving-meter",
-            of("unit", "n/s", "stat", "1m", "metric_type", "deriving-meter"), 0, TIME));
+            of("unit", "n/s", "stat", "1m", "metric_type", "deriving-meter"), RESOURCE, VALUE_0, TIME));
         expected.add(new Batch.Point("prefix.histogram",
-            of("unit", "n", "stat", "max", "metric_type", "histogram"), 0, TIME));
+            of("unit", "n", "stat", "max", "metric_type", "histogram"), RESOURCE, VALUE_0, TIME));
         expected.add(new Batch.Point("prefix.histogram",
-            of("unit", "n", "stat", "min", "metric_type", "histogram"), 0, TIME));
+            of("unit", "n", "stat", "min", "metric_type", "histogram"), RESOURCE, VALUE_0, TIME));
         expected.add(new Batch.Point("prefix.histogram",
-            of("unit", "n", "stat", "mean", "metric_type", "histogram"), 0, TIME));
+            of("unit", "n", "stat", "mean", "metric_type", "histogram"), RESOURCE, VALUE_0, TIME));
         expected.add(new Batch.Point("prefix.histogram",
-            of("unit", "n", "stat", "p75", "metric_type", "histogram"), 0, TIME));
+            of("unit", "n", "stat", "p75", "metric_type", "histogram"), RESOURCE, VALUE_0, TIME));
         expected.add(new Batch.Point("prefix.histogram",
-            of("unit", "n", "stat", "median", "metric_type", "histogram"), 0, TIME));
+            of("unit", "n", "stat", "median", "metric_type", "histogram"), RESOURCE, VALUE_0, TIME));
         expected.add(new Batch.Point("prefix.histogram",
-            of("unit", "n", "stat", "stddev", "metric_type", "histogram"), 0, TIME));
+            of("unit", "n", "stat", "stddev", "metric_type", "histogram"), RESOURCE, VALUE_0, TIME));
         expected.add(new Batch.Point("prefix.histogram",
-            of("unit", "n", "stat", "p99", "metric_type", "histogram"), 0, TIME));
+            of("unit", "n", "stat", "p99", "metric_type", "histogram"), RESOURCE, VALUE_0, TIME));
         expected.add(new Batch.Point("prefix.meter",
-            of("unit", "spec", "stat", "count", "metric_type", "meter"), 0, TIME));
+            of("unit", "spec", "stat", "count", "metric_type", "meter"), RESOURCE, VALUE_0, TIME));
         expected.add(new Batch.Point("prefix.meter",
-            of("unit", "spec/s", "stat", "1m", "metric_type", "meter"), 0, TIME));
+            of("unit", "spec/s", "stat", "1m", "metric_type", "meter"), RESOURCE, VALUE_0, TIME));
         expected.add(new Batch.Point("prefix.meter",
-            of("unit", "spec/s", "stat", "5m", "metric_type", "meter"), 0, TIME));
+            of("unit", "spec/s", "stat", "5m", "metric_type", "meter"), RESOURCE, VALUE_0, TIME));
         expected.add(new Batch.Point("prefix.meter2",
-            of("unit", "n", "stat", "count", "metric_type", "meter"), 0, TIME));
+            of("unit", "n", "stat", "count", "metric_type", "meter"), RESOURCE, VALUE_0, TIME));
         expected.add(new Batch.Point("prefix.meter2",
-            of("unit", "n/s", "stat", "1m", "metric_type", "meter"), 0, TIME));
+            of("unit", "n/s", "stat", "1m", "metric_type", "meter"), RESOURCE, VALUE_0, TIME));
         expected.add(new Batch.Point("prefix.meter2",
-            of("unit", "n/s", "stat", "5m", "metric_type", "meter"), 0, TIME));
+            of("unit", "n/s", "stat", "5m", "metric_type", "meter"), RESOURCE, VALUE_0, TIME));
         expected.add(
             new Batch.Point("prefix.timer", of("unit", "ns", "stat", "max", "metric_type", "timer"),
-                0, TIME));
+                    RESOURCE, VALUE_0, TIME));
         expected.add(
             new Batch.Point("prefix.timer", of("unit", "ns", "stat", "min", "metric_type", "timer"),
-                0, TIME));
+                    RESOURCE, VALUE_0, TIME));
         expected.add(new Batch.Point("prefix.timer",
-            of("unit", "ns", "stat", "mean", "metric_type", "timer"), 0, TIME));
+            of("unit", "ns", "stat", "mean", "metric_type", "timer"), RESOURCE, VALUE_0, TIME));
         expected.add(
             new Batch.Point("prefix.timer", of("unit", "ns", "stat", "p75", "metric_type", "timer"),
-                0, TIME));
+                    RESOURCE, VALUE_0, TIME));
         expected.add(new Batch.Point("prefix.timer",
-            of("unit", "ns", "stat", "median", "metric_type", "timer"), 0, TIME));
+            of("unit", "ns", "stat", "median", "metric_type", "timer"), RESOURCE, VALUE_0, TIME));
         expected.add(new Batch.Point("prefix.timer",
-            of("unit", "ns", "stat", "stddev", "metric_type", "timer"), 0, TIME));
+            of("unit", "ns", "stat", "stddev", "metric_type", "timer"), RESOURCE, VALUE_0, TIME));
         expected.add(
             new Batch.Point("prefix.timer", of("unit", "ns", "stat", "p99", "metric_type", "timer"),
-                0, TIME));
+                    RESOURCE, VALUE_0, TIME));
         expected.add(new Batch.Point("prefix.timer",
-            of("unit", "ns/s", "stat", "1m", "metric_type", "timer"), 0, TIME));
+            of("unit", "ns/s", "stat", "1m", "metric_type", "timer"), RESOURCE, VALUE_0, TIME));
         expected.add(new Batch.Point("prefix.timer",
-            of("unit", "ns/s", "stat", "5m", "metric_type", "timer"), 0, TIME));
+            of("unit", "ns/s", "stat", "5m", "metric_type", "timer"), RESOURCE, VALUE_0, TIME));
         expected.add(new Batch.Point("prefix.gauge",
-            of("what", "some-gauge", "unit", "n", "metric_type", "gauge"), 0, TIME));
+            of("what", "some-gauge", "unit", "n", "metric_type", "gauge"), RESOURCE, VALUE_0, TIME));
 
         reporter.start();
 
@@ -148,7 +180,6 @@ public Double getValue() {
         executorService.tick(REPORTING_PERIOD * 2 + 20, TimeUnit.MILLISECONDS);
         verify(httpClient, atLeastOnce()).sendBatch(
             batch.capture());
-
         for (final Batch b : batch.getAllValues()) {
             assertEquals(commonTags, b.getCommonTags());
             final Set<Batch.Point> points = new HashSet<>(b.getPoints());