SignalFx stats exporter (#903)

* SignalFx stats exporter

This commit implements a stats exporter that sends stats as metrics to
SignalFx (https://signalfx.com). It is modeled on the Stackdriver stats
exporter in that it uses a background worker thread that periodically
inspects all the views and constructs datapoints to send to SignalFx for
their corresponding metric timeseries (identified by the view name as
metric name, and tag key/value pairs as dimensions).

* Review feedback (squash me)

* More review feedback (squash me)

* Use a configuration builder pattern (squash me)

* Fixes for checker framework compliance (squash me)

* Review feedback (squash me)

* Remove support for AggregationWindow.Interval (squash me)

Author: mpetazzoni

README.md

@@ -111,6 +111,7 @@ runtime 'io.opencensus:opencensus-impl:0.10.1'
 
 #### Stats exporters
 * [Stackdriver][StatsExporterStackdriver]
+* [SignalFx][StatsExporterSignalFx]
 
 ### How to setup debugging Z-Pages?
 
@@ -133,3 +134,4 @@ see this [link](https://github.com/census-instrumentation/opencensus-java/tree/m
 [TraceExporterStackdriver]: https://github.com/census-instrumentation/opencensus-java/tree/master/exporters/trace/stackdriver#quickstart
 [TraceExporterZipkin]: https://github.com/census-instrumentation/opencensus-java/tree/master/exporters/trace/zipkin#quickstart
 [StatsExporterStackdriver]: https://github.com/census-instrumentation/opencensus-java/tree/master/exporters/stats/stackdriver#quickstart
+[StatsExporterSignalFx]: https://github.com/census-instrumentation/opencensus-java/tree/master/exporters/stats/signalfx#quickstart

all/build.gradle

@@ -24,6 +24,7 @@ def subprojects = [
         project(':opencensus-exporter-trace-logging'),
         project(':opencensus-exporter-trace-stackdriver'),
         project(':opencensus-exporter-trace-zipkin'),
+        project(':opencensus-exporter-stats-signalfx'),
         project(':opencensus-exporter-stats-stackdriver'),
 ]
 
@@ -38,6 +39,7 @@ def subprojects_javadoc = [
         project(':opencensus-exporter-trace-logging'),
         project(':opencensus-exporter-trace-stackdriver'),
         project(':opencensus-exporter-trace-zipkin'),
+        project(':opencensus-exporter-stats-signalfx'),
         project(':opencensus-exporter-stats-stackdriver'),
 ]
 

build.gradle

@@ -138,6 +138,7 @@ subprojects {
         guavaVersion = '19.0'
         googleAuthVersion = '0.9.0'
         googleCloudVersion = '0.30.0-beta'
+        signalfxVersion = '0.0.39'
         zipkinReporterVersion = '2.0.0'
 
         libraries = [
@@ -157,6 +158,7 @@ subprojects {
                 grpc_core: "io.grpc:grpc-core:${grpcVersion}",
                 guava: "com.google.guava:guava:${guavaVersion}",
                 jsr305: "com.google.code.findbugs:jsr305:${findBugsVersion}",
+                signalfx_java: "com.signalfx.public:signalfx-java:${signalfxVersion}",
 
                 // Test dependencies.
                 guava_testlib: "com.google.guava:guava-testlib:${guavaVersion}",
@@ -319,6 +321,7 @@ subprojects {
                  'opencensus-contrib-grpc-metrics',
                  'opencensus-contrib-grpc-util',
                  'opencensus-contrib-zpages',
+                 'opencensus-exporter-stats-signalfx',
                  'opencensus-exporter-stats-stackdriver',
                  'opencensus-exporter-trace-logging',
                  'opencensus-exporter-trace-stackdriver',

exporters/stats/signalfx/README.md

@@ -0,0 +1,75 @@
+# OpenCensus SignalFx Stats Exporter
+
+The _OpenCensus SignalFx Stats Exporter_ is a stats exporter that
+exports data to [SignalFx](https://signalfx.com), a real-time monitoring
+solution for cloud and distributed applications. SignalFx ingests that
+data and offers various visualizations on charts, dashboards and service
+maps, as well as real-time anomaly detection.
+
+## Quickstart
+
+### Prerequisites
+
+To use this exporter, you must have a [SignalFx](https://signalfx.com)
+account and corresponding [data ingest
+token](https://docs.signalfx.com/en/latest/admin-guide/tokens.html).
+
+#### Java versions
+
+This exporter requires Java 7 or above.
+
+### Add the dependencies to your project
+
+For Maven add to your `pom.xml`:
+
+```xml
+<dependencies>
+  <dependency>
+    <groupId>io.opencensus</groupId>
+    <artifactId>opencensus-api</artifactId>
+    <version>0.10.1</version>
+  </dependency>
+  <dependency>
+    <groupId>io.opencensus</groupId>
+    <artifactId>opencensus-exporter-stats-signalfx</artifactId>
+    <version>0.10.1</version>
+  </dependency>
+  <dependency>
+    <groupId>io.opencensus</groupId>
+    <artifactId>opencensus-impl</artifactId>
+    <version>0.10.1</version>
+    <scope>runtime</scope>
+  </dependency>
+</dependencies>
+```
+
+For Gradle add to your dependencies:
+
+```
+compile 'io.opencensus:opencensus-api:0.10.1'
+compile 'io.opencensus:opencensus-exporter-stats-signalfx:0.10.1'
+runtime 'io.opencensus:opencensus-impl:0.10.1'
+```
+
+### Register the exporter
+
+```java
+public class MyMainClass {
+  public static void main(String[] args) {
+    // SignalFx token is read from Java system properties.
+    // Stats will be reported every second by default.
+    SignalFxStatsExporter.create();
+  }
+}
+```
+
+If you want to pass in the token yourself, or set a different reporting
+interval, use:
+
+```java
+// Use token "your_signalfx_token" and report every 5 seconds.
+SignalFxStatsExporter.create(
+    SignalFxStatsExporter.DEFAULT_SIGNALFX_ENDPOINT,
+    "your_signalfx_token",
+    Duration.create(5, 0));
+```

exporters/stats/signalfx/build.gradle

@@ -0,0 +1,17 @@
+description = 'OpenCensus SignalFx Stats Exporter'
+
+[compileJava, compileTestJava].each() {
+    it.sourceCompatibility = 1.7
+    it.targetCompatibility = 1.7
+}
+
+dependencies {
+    compileOnly libraries.auto_value
+
+    compile project(':opencensus-api'),
+            libraries.signalfx_java
+
+    testCompile project(':opencensus-api')
+
+    signature "org.codehaus.mojo.signature:java17:+@signature"
+}

exporters/stats/signalfx/src/main/java/io/opencensus/exporter/stats/signalfx/SignalFxMetricsSenderFactory.java

@@ -0,0 +1,59 @@
+/*
+ * Copyright 2017, OpenCensus Authors
+ *
+ * Licensed 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 io.opencensus.exporter.stats.signalfx;
+
+import com.signalfx.endpoint.SignalFxEndpoint;
+import com.signalfx.metrics.auth.StaticAuthToken;
+import com.signalfx.metrics.connection.HttpDataPointProtobufReceiverFactory;
+import com.signalfx.metrics.connection.HttpEventProtobufReceiverFactory;
+import com.signalfx.metrics.errorhandler.OnSendErrorHandler;
+import com.signalfx.metrics.flush.AggregateMetricSender;
+import java.net.URI;
+import java.util.Collections;
+
+/** Interface for creators of {@link AggregateMetricSender}. */
+interface SignalFxMetricsSenderFactory {
+
+  /**
+   * Creates a new SignalFx metrics sender instance.
+   *
+   * @param endpoint The SignalFx ingest endpoint URL.
+   * @param token The SignalFx ingest token.
+   * @param errorHandler An {@link OnSendErrorHandler} through which errors when sending data to
+   *     SignalFx will be communicated.
+   * @return The created {@link AggregateMetricSender} instance.
+   */
+  AggregateMetricSender create(URI endpoint, String token, OnSendErrorHandler errorHandler);
+
+  /** The default, concrete implementation of this interface. */
+  SignalFxMetricsSenderFactory DEFAULT =
+      new SignalFxMetricsSenderFactory() {
+        @Override
+        @SuppressWarnings("nullness")
+        public AggregateMetricSender create(
+            URI endpoint, String token, OnSendErrorHandler errorHandler) {
+          SignalFxEndpoint sfx =
+              new SignalFxEndpoint(endpoint.getScheme(), endpoint.getHost(), endpoint.getPort());
+          return new AggregateMetricSender(
+              null,
+              new HttpDataPointProtobufReceiverFactory(sfx).setVersion(2),
+              new HttpEventProtobufReceiverFactory(sfx),
+              new StaticAuthToken(token),
+              Collections.singleton(errorHandler));
+        }
+      };
+}

exporters/stats/signalfx/src/main/java/io/opencensus/exporter/stats/signalfx/SignalFxSessionAdaptor.java

@@ -0,0 +1,166 @@
+/*
+ * Copyright 2017, OpenCensus Authors
+ *
+ * Licensed 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 io.opencensus.exporter.stats.signalfx;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
+import com.signalfx.metrics.protobuf.SignalFxProtocolBuffers.DataPoint;
+import com.signalfx.metrics.protobuf.SignalFxProtocolBuffers.Datum;
+import com.signalfx.metrics.protobuf.SignalFxProtocolBuffers.Dimension;
+import com.signalfx.metrics.protobuf.SignalFxProtocolBuffers.MetricType;
+import io.opencensus.common.Function;
+import io.opencensus.common.Functions;
+import io.opencensus.stats.Aggregation;
+import io.opencensus.stats.AggregationData;
+import io.opencensus.stats.AggregationData.CountData;
+import io.opencensus.stats.AggregationData.DistributionData;
+import io.opencensus.stats.AggregationData.MeanData;
+import io.opencensus.stats.AggregationData.SumDataDouble;
+import io.opencensus.stats.AggregationData.SumDataLong;
+import io.opencensus.stats.View;
+import io.opencensus.stats.View.AggregationWindow;
+import io.opencensus.stats.ViewData;
+import io.opencensus.tags.TagKey;
+import io.opencensus.tags.TagValue;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.ListIterator;
+import java.util.Map;
+
+/*>>>
+import org.checkerframework.checker.nullness.qual.Nullable;
+*/
+
+/** Adapter for a {@code ViewData}'s contents into SignalFx datapoints. */
+final class SignalFxSessionAdaptor {
+
+  private SignalFxSessionAdaptor() {}
+
+  /**
+   * Converts the given view data into datapoints that can be sent to SignalFx.
+   *
+   * <p>The view name is used as the metric name, and the aggregation type and aggregation window
+   * type determine the metric type.
+   *
+   * @param data The {@link ViewData} containing the aggregation data of each combination of tag
+   *     values.
+   * @return A list of datapoints for the corresponding metric timeseries of this view's metric.
+   */
+  static List<DataPoint> adapt(ViewData data) {
+    View view = data.getView();
+    List<TagKey> keys = view.getColumns();
+
+    MetricType metricType = getMetricTypeForAggregation(view.getAggregation(), view.getWindow());
+    if (metricType == null) {
+      return Collections.emptyList();
+    }
+
+    List<DataPoint> datapoints = new ArrayList<>(data.getAggregationMap().size());
+    for (Map.Entry<List<TagValue>, AggregationData> entry : data.getAggregationMap().entrySet()) {
+      datapoints.add(
+          DataPoint.newBuilder()
+              .setMetric(view.getName().asString())
+              .setMetricType(metricType)
+              .addAllDimensions(createDimensions(keys, entry.getKey()))
+              .setValue(createDatum(entry.getValue()))
+              .build());
+    }
+    return datapoints;
+  }
+
+  @VisibleForTesting
+  @javax.annotation.Nullable
+  static MetricType getMetricTypeForAggregation(Aggregation aggregation, AggregationWindow window) {
+    if (aggregation instanceof Aggregation.Mean) {
+      return MetricType.GAUGE;
+    } else if (aggregation instanceof Aggregation.Count || aggregation instanceof Aggregation.Sum) {
+      if (window instanceof AggregationWindow.Cumulative) {
+        return MetricType.CUMULATIVE_COUNTER;
+      }
+      // TODO(mpetazzoni): support incremental counters when AggregationWindow.Interval is ready
+    }
+
+    // TODO(mpetazzoni): add support for histograms (Aggregation.Distribution).
+    return null;
+  }
+
+  @VisibleForTesting
+  static Iterable<Dimension> createDimensions(List<TagKey> keys, List<TagValue> values) {
+    Preconditions.checkArgument(
+        keys.size() == values.size(), "TagKeys and TagValues don't have the same size.");
+    List<Dimension> dimensions = new ArrayList<>(keys.size());
+    for (ListIterator<TagKey> it = keys.listIterator(); it.hasNext(); ) {
+      TagKey key = it.next();
+      TagValue value = values.get(it.previousIndex());
+      if (value == null || Strings.isNullOrEmpty(value.asString())) {
+        continue;
+      }
+      dimensions.add(createDimension(key, value));
+    }
+    return dimensions;
+  }
+
+  @VisibleForTesting
+  static Dimension createDimension(TagKey key, TagValue value) {
+    return Dimension.newBuilder().setKey(key.getName()).setValue(value.asString()).build();
+  }
+
+  @VisibleForTesting
+  static Datum createDatum(AggregationData data) {
+    final Datum.Builder builder = Datum.newBuilder();
+    data.match(
+        new Function<SumDataDouble, Void>() {
+          @Override
+          public Void apply(SumDataDouble arg) {
+            builder.setDoubleValue(arg.getSum());
+            return null;
+          }
+        },
+        new Function<SumDataLong, Void>() {
+          @Override
+          public Void apply(SumDataLong arg) {
+            builder.setIntValue(arg.getSum());
+            return null;
+          }
+        },
+        new Function<CountData, Void>() {
+          @Override
+          public Void apply(CountData arg) {
+            builder.setIntValue(arg.getCount());
+            return null;
+          }
+        },
+        new Function<MeanData, Void>() {
+          @Override
+          public Void apply(MeanData arg) {
+            builder.setDoubleValue(arg.getMean());
+            return null;
+          }
+        },
+        new Function<DistributionData, Void>() {
+          @Override
+          public Void apply(DistributionData arg) {
+            // TODO(mpetazzoni): add histogram support.
+            throw new IllegalArgumentException("Distribution aggregations are not supported");
+          }
+        },
+        Functions.</*@Nullable*/ Void>throwIllegalArgumentException());
+    return builder.build();
+  }
+}