Add a StackDriver exporter for stats using a Pull model (#713)

* Add a StackDriver exporter for stats

* Update to match our recent API changes

* Update to use current version of Mockito

* Run formatter

* Improve comments

* Add a TODO to remove exporter.registerView

* A few renaming

10 files changed, 1334 insertions, 0 deletions

diff --git a/all/build.gradle b/all/build.gradle
index 420ef9b6..76146322 100644
--- a/all/build.gradle
+++ b/all/build.gradle
@@ -27,6 +27,7 @@ def subprojects = [
         project(':opencensus-exporter-trace-logging'),
         project(':opencensus-exporter-trace-stackdriver'),
         project(':opencensus-exporter-trace-zipkin'),
+        project(':opencensus-exporter-stats-stackdriver'),
 ]
 
 // A subset of subprojects for which we want to publish javadoc.
@@ -40,6 +41,7 @@ def subprojects_javadoc = [
         project(':opencensus-exporter-trace-logging'),
         project(':opencensus-exporter-trace-stackdriver'),
         project(':opencensus-exporter-trace-zipkin'),
+        project(':opencensus-exporter-stats-stackdriver'),
 ]
 
 for (subproject in rootProject.subprojects) {
diff --git a/build.gradle b/build.gradle
index 7e9c4cbc..b1f64f9c 100644
--- a/build.gradle
+++ b/build.gradle
@@ -123,6 +123,7 @@ subprojects {
                 google_cloud_trace: "com.google.cloud:google-cloud-trace:${googleCloudVersion}",
                 zipkin_reporter: "io.zipkin.reporter2:zipkin-reporter:${zipkinReporterVersion}",
                 zipkin_urlconnection: "io.zipkin.reporter2:zipkin-sender-urlconnection:${zipkinReporterVersion}",
+                google_cloud_monitoring: "com.google.cloud:google-cloud-monitoring:${googleCloudVersion}",
                 grpc_context: "io.grpc:grpc-context:${grpcVersion}",
                 grpc_core: "io.grpc:grpc-core:${grpcVersion}",
                 guava: "com.google.guava:guava:${guavaVersion}",
diff --git a/exporters/stats/stackdriver/README.md b/exporters/stats/stackdriver/README.md
new file mode 100644
index 00000000..039f2658
--- /dev/null
+++ b/exporters/stats/stackdriver/README.md
@@ -0,0 +1,74 @@
+# OpenCensus Stackdriver Stats Exporter
+
+The *OpenCensus Stackdriver Stats Exporter* is a stats exporter that exports data to 
+Stackdriver Monitoring. [Stackdriver Monitoring][stackdriver-monitoring] provides visibility into 
+the performance, uptime, and overall health of cloud-powered applications. Stackdriver ingests that 
+data and generates insights via dashboards, charts, and alerts.
+
+## Quickstart
+
+### Prerequisites
+
+To use this exporter, you must have an application that you'd like to monitor. The app can be on 
+Google Cloud Platform, on-premise, or another cloud platform.
+
+In order to be able to push your stats to [Stackdriver Monitoring][stackdriver-monitoring], you must:
+
+1. [Create a Cloud project](https://support.google.com/cloud/answer/6251787?hl=en).
+2. [Enable billing](https://support.google.com/cloud/answer/6288653#new-billing).
+3. [Enable the Stackdriver Monitoring API](https://app.google.stackdriver.com/).
+
+These steps enable the API but don't require that your app is hosted on Google Cloud Platform.
+
+### Hello "Stackdriver Stats"
+
+#### Add the dependencies to your project
+
+For Maven add to your `pom.xml`: TODO
+
+For Gradle add to your dependencies: TODO
+
+#### Register the exporter
+
+This uses the default configuration for authentication and a given project ID.
+
+```java
+public class MyMainClass {
+  public static void main(String[] args) {
+    // Exporter will export to Stackdriver every 10 seconds.
+    StackdriverStatsExporter.createWithProjectId("MyStackdriverProjectId", Duration.create(10, 0));
+    StackdriverStatsExporter.registerView(myView);
+  }
+}
+```
+
+#### Authentication
+
+This exporter uses [google-cloud-java](https://github.com/GoogleCloudPlatform/google-cloud-java),
+for details about how to configure the authentication see [here](https://github.com/GoogleCloudPlatform/google-cloud-java#authentication).
+
+If you prefer to manually set the credentials use:
+```
+StackdriverStatsExporter.createWithCredentialsAndProjectId(
+    new GoogleCredentials(new AccessToken(accessToken, expirationTime)), 
+    "MyStackdriverProjectId",
+    Duration.create(10, 0));
+```
+
+#### Specifying a Project ID
+
+This exporter uses [google-cloud-java](https://github.com/GoogleCloudPlatform/google-cloud-java),
+for details about how to configure the project ID see [here](https://github.com/GoogleCloudPlatform/google-cloud-java#specifying-a-project-id).
+
+If you prefer to manually set the project ID use:
+```
+StackdriverStatsExporter.createWithProjectId("MyStackdriverProjectId", Duration.create(10, 0));
+```
+
+#### Java Versions
+
+Java 7 or above is required for using this exporter.
+
+## FAQ
+
+[stackdriver-monitoring]: https://cloud.google.com/monitoring/
\ No newline at end of file
diff --git a/exporters/stats/stackdriver/build.gradle b/exporters/stats/stackdriver/build.gradle
new file mode 100644
index 00000000..a831f086
--- /dev/null
+++ b/exporters/stats/stackdriver/build.gradle
@@ -0,0 +1,16 @@
+description = 'OpenCensus Stats Stackdriver Exporter'
+
+[compileJava, compileTestJava].each() {
+    it.sourceCompatibility = 1.7
+    it.targetCompatibility = 1.7
+}
+
+dependencies {
+    compile project(':opencensus-api'),
+            libraries.google_auth,
+            libraries.google_cloud_monitoring
+
+    testCompile project(':opencensus-api')
+
+    signature "org.codehaus.mojo.signature:java17:+@signature"
+}
\ No newline at end of file
diff --git a/exporters/stats/stackdriver/src/main/java/io/opencensus/exporter/stats/stackdriver/StackdriverExportUtils.java b/exporters/stats/stackdriver/src/main/java/io/opencensus/exporter/stats/stackdriver/StackdriverExportUtils.java
new file mode 100644
index 00000000..ea351034
--- /dev/null
+++ b/exporters/stats/stackdriver/src/main/java/io/opencensus/exporter/stats/stackdriver/StackdriverExportUtils.java
@@ -0,0 +1,301 @@
+/*
+ * 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.stackdriver;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+import com.google.api.Distribution;
+import com.google.api.Distribution.BucketOptions;
+import com.google.api.Distribution.BucketOptions.Explicit;
+import com.google.api.Distribution.Range;
+import com.google.api.LabelDescriptor;
+import com.google.api.LabelDescriptor.ValueType;
+import com.google.api.Metric;
+import com.google.api.MetricDescriptor;
+import com.google.api.MetricDescriptor.MetricKind;
+import com.google.api.MonitoredResource;
+import com.google.api.client.util.Maps;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.Lists;
+import com.google.monitoring.v3.Point;
+import com.google.monitoring.v3.TimeInterval;
+import com.google.monitoring.v3.TimeSeries;
+import com.google.monitoring.v3.TypedValue;
+import com.google.protobuf.Timestamp;
+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.BucketBoundaries;
+import io.opencensus.stats.Measure;
+import io.opencensus.stats.View;
+import io.opencensus.stats.View.AggregationWindow;
+import io.opencensus.stats.View.AggregationWindow.Cumulative;
+import io.opencensus.stats.ViewData;
+import io.opencensus.stats.ViewData.AggregationWindowData;
+import io.opencensus.stats.ViewData.AggregationWindowData.CumulativeData;
+import io.opencensus.stats.ViewData.AggregationWindowData.IntervalData;
+import io.opencensus.tags.TagKey;
+import io.opencensus.tags.TagValue;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import javax.annotation.Nullable;
+
+/** Util methods to convert OpenCensus Stats data models to StackDriver monitoring data models. */
+final class StackdriverExportUtils {
+
+  // TODO(songya): do we want these constants to be customizable?
+  @VisibleForTesting static final String LABEL_DESCRIPTION = "OpenCensus TagKey";
+
+  // Construct a MetricDescriptor using a View.
+  @Nullable
+  static MetricDescriptor createMetricDescriptor(View view, String projectId) {
+    if (!(view.getWindow() instanceof Cumulative)) {
+      // TODO(songya): Only Cumulative view will be exported to Stackdriver in this version.
+      return null;
+    }
+
+    MetricDescriptor.Builder builder = MetricDescriptor.newBuilder();
+    String viewName = view.getName().asString();
+    // Name format refers to
+    // cloud.google.com/monitoring/api/ref_v3/rest/v3/projects.metricDescriptors/create
+    builder.setName(String.format("projects/%s", projectId));
+    builder.setType(String.format("custom.googleapis.com/opencensus/%s", viewName));
+    builder.setDescription(view.getDescription());
+    builder.setUnit(view.getMeasure().getUnit());
+    builder.setDisplayName("OpenCensus/" + viewName);
+    for (TagKey tagKey : view.getColumns()) {
+      builder.addLabels(createLabelDescriptor(tagKey));
+    }
+    builder.setMetricKind(createMetricKind(view.getWindow()));
+    builder.setValueType(createValueType(view.getAggregation(), view.getMeasure()));
+    return builder.build();
+  }
+
+  // Construct a LabelDescriptor from a TagKey
+  @VisibleForTesting
+  static LabelDescriptor createLabelDescriptor(TagKey tagKey) {
+    LabelDescriptor.Builder builder = LabelDescriptor.newBuilder();
+    builder.setKey(tagKey.getName());
+    builder.setDescription(LABEL_DESCRIPTION);
+    // Now we only support String tags
+    builder.setValueType(ValueType.STRING);
+    return builder.build();
+  }
+
+  // Construct a MetricKind from an AggregationWindow
+  @VisibleForTesting
+  static MetricKind createMetricKind(AggregationWindow window) {
+    return window.match(
+        Functions.returnConstant(MetricKind.CUMULATIVE), // Cumulative
+        // TODO(songya): We don't support exporting Interval stats to StackDriver in this version.
+        Functions.returnConstant(MetricKind.UNRECOGNIZED), // Interval
+        Functions.returnConstant(MetricKind.UNRECOGNIZED));
+  }
+
+  // Construct a MetricDescriptor.ValueType from an Aggregation and a Measure
+  @VisibleForTesting
+  static MetricDescriptor.ValueType createValueType(
+      Aggregation aggregation, final Measure measure) {
+    return aggregation.match(
+        Functions.returnConstant(
+            measure.match(
+                Functions.returnConstant(MetricDescriptor.ValueType.DOUBLE), // Sum Double
+                Functions.returnConstant(MetricDescriptor.ValueType.INT64), // Sum Long
+                Functions.returnConstant(MetricDescriptor.ValueType.UNRECOGNIZED))),
+        Functions.returnConstant(MetricDescriptor.ValueType.INT64), // Count
+        Functions.returnConstant(MetricDescriptor.ValueType.DOUBLE), // Mean
+        Functions.returnConstant(MetricDescriptor.ValueType.DISTRIBUTION), // Distribution
+        Functions.returnConstant(MetricDescriptor.ValueType.UNRECOGNIZED));
+  }
+
+  // Convert ViewData to a list of TimeSeries, so that ViewData can be uploaded to Stackdriver.
+  static List<TimeSeries> createTimeSeriesList(ViewData viewData, String projectId) {
+    List<TimeSeries> timeSeriesList = Lists.newArrayList();
+    View view = viewData.getView();
+    if (!(view.getWindow() instanceof Cumulative)) {
+      // TODO(songya): Only Cumulative view will be exported to Stackdriver in this version.
+      return timeSeriesList;
+    }
+
+    // Shared fields for all TimeSeries generated from the same ViewData
+    TimeSeries.Builder shared = TimeSeries.newBuilder();
+    shared.setMetricKind(createMetricKind(view.getWindow()));
+    shared.setResource(
+        MonitoredResource.newBuilder().setType("global").putLabels("project_id", projectId));
+    shared.setValueType(createValueType(view.getAggregation(), view.getMeasure()));
+
+    // Each entry in AggregationMap will be converted into an independent TimeSeries object
+    for (Entry<List<TagValue>, AggregationData> entry : viewData.getAggregationMap().entrySet()) {
+      TimeSeries.Builder builder = shared.clone();
+      builder.setMetric(createMetric(view, entry.getKey()));
+      builder.addPoints(
+          createPoint(entry.getValue(), viewData.getWindowData(), view.getAggregation()));
+      timeSeriesList.add(builder.build());
+    }
+
+    return timeSeriesList;
+  }
+
+  // Create a Metric using the TagKeys and TagValues.
+  @VisibleForTesting
+  static Metric createMetric(View view, List<? extends TagValue> tagValues) {
+    Metric.Builder builder = Metric.newBuilder();
+    // TODO(songya): use pre-defined metrics for canonical views
+    builder.setType(
+        String.format("custom.googleapis.com/opencensus/%s", view.getName().asString()));
+    Map<String, String> stringTagMap = Maps.newHashMap();
+    List<TagKey> columns = view.getColumns();
+    checkArgument(
+        tagValues.size() == columns.size(), "TagKeys and TagValues don't have same size.");
+    for (int i = 0; i < tagValues.size(); i++) {
+      TagKey key = columns.get(i);
+      TagValue value = tagValues.get(i);
+      stringTagMap.put(key.getName(), value.asString());
+    }
+    builder.putAllLabels(stringTagMap);
+    return builder.build();
+  }
+
+  // Create Point from AggregationData, AggregationWindowData and Aggregation.
+  @VisibleForTesting
+  static Point createPoint(
+      AggregationData aggregationData, AggregationWindowData windowData, Aggregation aggregation) {
+    Point.Builder builder = Point.newBuilder();
+    builder.setInterval(createTimeInterval(windowData));
+    builder.setValue(createTypedValue(aggregation, aggregationData));
+    return builder.build();
+  }
+
+  // Convert AggregationWindowData to TimeInterval, currently only support CumulativeData.
+  @VisibleForTesting
+  static TimeInterval createTimeInterval(AggregationWindowData windowData) {
+    final TimeInterval.Builder builder = TimeInterval.newBuilder();
+    windowData.match(
+        new Function<CumulativeData, Void>() {
+          @Override
+          public Void apply(CumulativeData arg) {
+            builder.setStartTime(convertTimestamp(arg.getStart()));
+            builder.setEndTime(convertTimestamp(arg.getEnd()));
+            return null;
+          }
+        },
+        new Function<IntervalData, Void>() {
+          @Override
+          public Void apply(IntervalData arg) {
+            // TODO(songya): we don't export IntervalData in this version.
+            throw new IllegalArgumentException("IntervalData not supported");
+          }
+        },
+        Functions.<Void>throwIllegalArgumentException());
+    return builder.build();
+  }
+
+  // Create a TypedValue using AggregationData and Aggregation
+  // Note TypedValue is "A single strongly-typed value", i.e only one field should be set.
+  @VisibleForTesting
+  static TypedValue createTypedValue(
+      final Aggregation aggregation, AggregationData aggregationData) {
+    final TypedValue.Builder builder = TypedValue.newBuilder();
+    aggregationData.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.setInt64Value(arg.getSum());
+            return null;
+          }
+        },
+        new Function<CountData, Void>() {
+          @Override
+          public Void apply(CountData arg) {
+            builder.setInt64Value(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) {
+            checkArgument(
+                aggregation instanceof Aggregation.Distribution,
+                "Aggregation and AggregationData mismatch.");
+            builder.setDistributionValue(
+                createDistribution(
+                    arg, ((Aggregation.Distribution) aggregation).getBucketBoundaries()));
+            return null;
+          }
+        },
+        Functions.<Void>throwIllegalArgumentException());
+    return builder.build();
+  }
+
+  // Create a StackDriver Distribution from DistributionData and BucketBoundaries
+  @VisibleForTesting
+  static Distribution createDistribution(
+      DistributionData distributionData, BucketBoundaries bucketBoundaries) {
+    return Distribution.newBuilder()
+        .setBucketOptions(createBucketOptions(bucketBoundaries))
+        .addAllBucketCounts(distributionData.getBucketCounts())
+        .setCount(distributionData.getCount())
+        .setMean(distributionData.getMean())
+        .setRange(
+            Range.newBuilder()
+                .setMax(distributionData.getMax())
+                .setMin(distributionData.getMin())
+                .build())
+        .setSumOfSquaredDeviation(distributionData.getSumOfSquaredDeviations())
+        .build();
+  }
+
+  // Create BucketOptions from BucketBoundaries
+  @VisibleForTesting
+  static BucketOptions createBucketOptions(BucketBoundaries bucketBoundaries) {
+    return BucketOptions.newBuilder()
+        .setExplicitBuckets(Explicit.newBuilder().addAllBounds(bucketBoundaries.getBoundaries()))
+        .build();
+  }
+
+  // Convert a Census Timestamp to a StackDriver Timestamp
+  @VisibleForTesting
+  static Timestamp convertTimestamp(io.opencensus.common.Timestamp censusTimestamp) {
+    return Timestamp.newBuilder()
+        .setSeconds(censusTimestamp.getSeconds())
+        .setNanos(censusTimestamp.getNanos())
+        .build();
+  }
+
+  private StackdriverExportUtils() {}
+}
diff --git a/exporters/stats/stackdriver/src/main/java/io/opencensus/exporter/stats/stackdriver/StackdriverExporterWorkerThread.java b/exporters/stats/stackdriver/src/main/java/io/opencensus/exporter/stats/stackdriver/StackdriverExporterWorkerThread.java
new file mode 100644
index 00000000..0720a5c9
--- /dev/null
+++ b/exporters/stats/stackdriver/src/main/java/io/opencensus/exporter/stats/stackdriver/StackdriverExporterWorkerThread.java
@@ -0,0 +1,118 @@
+/*
+ * 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.stackdriver;
+
+import com.google.cloud.monitoring.v3.MetricServiceClient;
+import com.google.common.collect.Lists;
+import com.google.monitoring.v3.CreateTimeSeriesRequest;
+import io.opencensus.common.Duration;
+import io.opencensus.stats.View;
+import io.opencensus.stats.ViewData;
+import io.opencensus.stats.ViewManager;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import javax.annotation.concurrent.GuardedBy;
+
+/**
+ * Worker {@code Thread} that polls ViewData from Stats library and batch export to StackDriver.
+ *
+ * <p>{@code StackdriverExporterWorkerThread} is a daemon {@code Thread}.
+ */
+final class StackdriverExporterWorkerThread extends Thread {
+
+  private static final Object monitor = new Object();
+
+  private final long scheduleDelayMillis;
+  private final String projectId;
+  private final MetricServiceClient metricServiceClient;
+
+  @GuardedBy("monitor")
+  private final ViewManager viewManager;
+
+  @GuardedBy("monitor")
+  private final Map<View.Name, View> registeredViews = new HashMap<View.Name, View>();
+
+  StackdriverExporterWorkerThread(
+      String projectId,
+      MetricServiceClient metricServiceClient,
+      Duration exportInterval,
+      ViewManager viewManager) {
+    this.scheduleDelayMillis = toMillis(exportInterval);
+    this.projectId = projectId;
+    this.metricServiceClient = metricServiceClient;
+    this.viewManager = viewManager;
+    setDaemon(true);
+    setName("ExportWorkerThread");
+  }
+
+  void registerView(View view) {
+    synchronized (monitor) {
+      View existing = registeredViews.get(view.getName());
+      if (existing != null) {
+        if (existing.equals(view)) {
+          // Ignore views that are already registered.
+          return;
+        } else {
+          throw new IllegalArgumentException(
+              "A different view with the same name is already registered: " + existing);
+        }
+      }
+      registeredViews.put(view.getName(), view);
+      viewManager.registerView(view);
+    }
+  }
+
+  // Polls ViewData from Stats library for all registered views, and upload them as TimeSeries to
+  // StackDriver.
+  private void export() {
+    List<ViewData> viewDataList = Lists.newArrayList();
+    synchronized (monitor) {
+      for (View.Name view : registeredViews.keySet()) {
+        viewDataList.add(viewManager.getView(view));
+      }
+    }
+    CreateTimeSeriesRequest.Builder builder = CreateTimeSeriesRequest.newBuilder();
+    for (ViewData viewData : viewDataList) {
+      builder.addAllTimeSeries(StackdriverExportUtils.createTimeSeriesList(viewData, projectId));
+    }
+    if (!builder.getTimeSeriesList().isEmpty()) {
+      metricServiceClient.createTimeSeries(builder.build());
+    }
+  }
+
+  @Override
+  public void run() {
+    while (true) {
+      try {
+        export();
+        Thread.sleep(scheduleDelayMillis);
+      } catch (InterruptedException ie) {
+        // Preserve the interruption status as per guidance and stop doing any work.
+        Thread.currentThread().interrupt();
+        return;
+      }
+    }
+  }
+
+  private static final long MILLIS_PER_SECOND = 1000L;
+  private static final long NANOS_PER_MILLI = 1000 * 1000;
+
+  private static long toMillis(Duration duration) {
+    return duration.getSeconds() * MILLIS_PER_SECOND + duration.getNanos() / NANOS_PER_MILLI;
+  }
+}
diff --git a/exporters/stats/stackdriver/src/main/java/io/opencensus/exporter/stats/stackdriver/StackdriverStatsExporter.java b/exporters/stats/stackdriver/src/main/java/io/opencensus/exporter/stats/stackdriver/StackdriverStatsExporter.java
new file mode 100644
index 00000000..9e66bf90
--- /dev/null
+++ b/exporters/stats/stackdriver/src/main/java/io/opencensus/exporter/stats/stackdriver/StackdriverStatsExporter.java
@@ -0,0 +1,211 @@
+/*
+ * 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.stackdriver;
+
+import static com.google.api.client.util.Preconditions.checkArgument;
+import static com.google.api.client.util.Preconditions.checkNotNull;
+import static com.google.common.base.Preconditions.checkState;
+
+import com.google.api.MetricDescriptor;
+import com.google.api.gax.core.FixedCredentialsProvider;
+import com.google.auth.Credentials;
+import com.google.auth.oauth2.GoogleCredentials;
+import com.google.cloud.ServiceOptions;
+import com.google.cloud.monitoring.v3.MetricServiceClient;
+import com.google.cloud.monitoring.v3.MetricServiceSettings;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.monitoring.v3.CreateMetricDescriptorRequest;
+import io.opencensus.common.Duration;
+import io.opencensus.stats.Stats;
+import io.opencensus.stats.View;
+import io.opencensus.stats.ViewManager;
+import java.io.IOException;
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.GuardedBy;
+
+/**
+ * Exporter to Stackdriver Monitoring Client API v3.
+ *
+ * <p>Example of usage on Google Cloud VMs:
+ *
+ * <pre><code>
+ *   public static void main(String[] args) {
+ *     StackdriverStatsExporter.createWithProjectId(
+ *         "MyStackdriverProjectId", Duration.fromMillis(100000));
+ *     StackdriverStatsExporter.registerView(myView);
+ *     ... // Do work.
+ *   }
+ * </code></pre>
+ */
+public final class StackdriverStatsExporter {
+
+  private static final Object monitor = new Object();
+
+  private final String projectId;
+  private final MetricServiceClient metricServiceClient;
+  private final StackdriverExporterWorkerThread workerThread;
+
+  @GuardedBy("monitor")
+  private static StackdriverStatsExporter exporter = null;
+
+  private static final Duration ZERO = Duration.create(0, 0);
+
+  @VisibleForTesting
+  StackdriverStatsExporter(
+      String projectId,
+      MetricServiceClient metricServiceClient,
+      Duration exportInterval,
+      ViewManager viewManager) {
+    checkArgument(exportInterval.compareTo(ZERO) > 0, "Duration must be positive");
+    this.projectId = projectId;
+    this.metricServiceClient = metricServiceClient;
+    this.workerThread =
+        new StackdriverExporterWorkerThread(
+            projectId, metricServiceClient, exportInterval, viewManager);
+  }
+
+  /**
+   * Creates a StackdriverStatsExporter for an explicit project ID and using explicit credentials.
+   *
+   * <p>Only one Stackdriver exporter can be created.
+   *
+   * @param credentials a credentials used to authenticate API calls.
+   * @param projectId the cloud project id.
+   * @param exportInterval the interval between pushing stats to StackDriver.
+   * @throws IllegalStateException if a Stackdriver exporter already exists.
+   */
+  public static void createWithCredentialsAndProjectId(
+      Credentials credentials, String projectId, Duration exportInterval) throws IOException {
+    checkNotNull(credentials, "credentials");
+    checkNotNull(projectId, "projectId");
+    checkNotNull(exportInterval, "exportInterval");
+    createInternal(credentials, projectId, exportInterval);
+  }
+
+  /**
+   * Creates a Stackdriver Stats exporter for an explicit project ID.
+   *
+   * <p>Only one Stackdriver exporter can be created.
+   *
+   * <p>This uses the default application credentials. See {@link
+   * GoogleCredentials#getApplicationDefault}.
+   *
+   * <p>This is equivalent with:
+   *
+   * <pre>{@code
+   * StackdriverStatsExporter.createWithCredentialsAndProjectId(
+   *     GoogleCredentials.getApplicationDefault(), projectId);
+   * }</pre>
+   *
+   * @param projectId the cloud project id.
+   * @param exportInterval the interval between pushing stats to StackDriver.
+   * @throws IllegalStateException if a Stackdriver exporter is already created.
+   */
+  public static void createWithProjectId(String projectId, Duration exportInterval)
+      throws IOException {
+    checkNotNull(projectId, "projectId");
+    checkNotNull(exportInterval, "exportInterval");
+    createInternal(null, projectId, exportInterval);
+  }
+
+  /**
+   * Creates a Stackdriver Stats exporter.
+   *
+   * <p>Only one Stackdriver exporter can be created.
+   *
+   * <p>This uses the default application credentials. See {@link
+   * GoogleCredentials#getApplicationDefault}.
+   *
+   * <p>This uses the default project ID configured see {@link ServiceOptions#getDefaultProjectId}.
+   *
+   * <p>This is equivalent with:
+   *
+   * <pre>{@code
+   * StackdriverStatsExporter.createWithProjectId(ServiceOptions.getDefaultProjectId());
+   * }</pre>
+   *
+   * @param exportInterval the interval between pushing stats to StackDriver.
+   * @throws IllegalStateException if a Stackdriver exporter is already created.
+   */
+  public static void create(Duration exportInterval) throws IOException {
+    checkNotNull(exportInterval, "exportInterval");
+    createInternal(null, ServiceOptions.getDefaultProjectId(), exportInterval);
+  }
+
+  // Use createInternal() (instead of constructor) to enforce singleton.
+  private static void createInternal(
+      @Nullable Credentials credentials, String projectId, Duration exportInterval)
+      throws IOException {
+    synchronized (monitor) {
+      checkState(exporter == null, "Stackdriver stats exporter is already created.");
+      MetricServiceClient metricServiceClient;
+      // Initialize MetricServiceClient inside lock to avoid creating multiple clients.
+      if (credentials == null) {
+        metricServiceClient = MetricServiceClient.create();
+      } else {
+        metricServiceClient =
+            MetricServiceClient.create(
+                MetricServiceSettings.newBuilder()
+                    .setCredentialsProvider(FixedCredentialsProvider.create(credentials))
+                    .build());
+      }
+      exporter =
+          new StackdriverStatsExporter(
+              projectId, metricServiceClient, exportInterval, Stats.getViewManager());
+      exporter.workerThread.start();
+    }
+  }
+
+  // Method for setting exporter to a fake exporter or null (reset) for unit tests.
+  @VisibleForTesting
+  static void unsafeSetExporter(StackdriverStatsExporter exporter) {
+    synchronized (monitor) {
+      StackdriverStatsExporter.exporter = exporter;
+      if (exporter != null) {
+        exporter.workerThread.start();
+      }
+    }
+  }
+
+  /**
+   * Register a {@link View} against this exporter, and upload it as a {@link MetricDescriptor} to
+   * StackDriver.
+   *
+   * @param view the {@code View} to be registered.
+   * @throws IllegalStateException if a Stackdriver stats exporter has not been created yet.
+   */
+  // TODO(songya): remove this API and have exporter polls stats using getAllExportedView(). Views
+  // should not be registered against exporter, since in the future we'll probably switch to a push
+  // model.
+  public static void registerView(View view) {
+    synchronized (monitor) {
+      checkState(exporter != null, "Stackdriver stats exporter has not been created.");
+      // TODO(songya): don't need to create MetricDescriptor for RpcViewConstants once we defined
+      // canonical metrics. Registration is required only for custom view definitions. Canonical
+      // views should be pre-registered.
+      MetricDescriptor metricDescriptor =
+          StackdriverExportUtils.createMetricDescriptor(view, exporter.projectId);
+      if (metricDescriptor != null) {
+        exporter.metricServiceClient.createMetricDescriptor(
+            CreateMetricDescriptorRequest.newBuilder()
+                .setMetricDescriptor(metricDescriptor)
+                .build());
+        exporter.workerThread.registerView(view);
+      }
+    }
+  }
+}
diff --git a/exporters/stats/stackdriver/src/test/java/io/opencensus/exporter/stats/stackdriver/StackdriverExportUtilsTest.java b/exporters/stats/stackdriver/src/test/java/io/opencensus/exporter/stats/stackdriver/StackdriverExportUtilsTest.java
new file mode 100644
index 00000000..d6bfc69e
--- /dev/null
+++ b/exporters/stats/stackdriver/src/test/java/io/opencensus/exporter/stats/stackdriver/StackdriverExportUtilsTest.java
@@ -0,0 +1,376 @@
+/*
+ * 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.stackdriver;
+
+import static com.google.common.truth.Truth.assertThat;
+
+import com.google.api.Distribution.BucketOptions;
+import com.google.api.Distribution.BucketOptions.Explicit;
+import com.google.api.LabelDescriptor;
+import com.google.api.LabelDescriptor.ValueType;
+import com.google.api.Metric;
+import com.google.api.MetricDescriptor;
+import com.google.api.MetricDescriptor.MetricKind;
+import com.google.api.MonitoredResource;
+import com.google.common.collect.ImmutableMap;
+import com.google.monitoring.v3.Point;
+import com.google.monitoring.v3.TimeInterval;
+import com.google.monitoring.v3.TimeSeries;
+import com.google.monitoring.v3.TypedValue;
+import io.opencensus.common.Duration;
+import io.opencensus.common.Timestamp;
+import io.opencensus.stats.Aggregation.Count;
+import io.opencensus.stats.Aggregation.Distribution;
+import io.opencensus.stats.Aggregation.Mean;
+import io.opencensus.stats.Aggregation.Sum;
+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.BucketBoundaries;
+import io.opencensus.stats.Measure.MeasureDouble;
+import io.opencensus.stats.Measure.MeasureLong;
+import io.opencensus.stats.View;
+import io.opencensus.stats.View.AggregationWindow.Cumulative;
+import io.opencensus.stats.View.AggregationWindow.Interval;
+import io.opencensus.stats.View.Name;
+import io.opencensus.stats.ViewData;
+import io.opencensus.stats.ViewData.AggregationWindowData.CumulativeData;
+import io.opencensus.stats.ViewData.AggregationWindowData.IntervalData;
+import io.opencensus.tags.TagKey;
+import io.opencensus.tags.TagValue;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Unit tests for {@link StackdriverExportUtils}. */
+@RunWith(JUnit4.class)
+public class StackdriverExportUtilsTest {
+
+  @Rule public final ExpectedException thrown = ExpectedException.none();
+
+  private static final TagKey KEY = TagKey.create("KEY");
+  private static final TagValue VALUE_1 = TagValue.create("VALUE1");
+  private static final TagValue VALUE_2 = TagValue.create("VALUE2");
+  private static final String MEASURE_UNIT = "us";
+  private static final String MEASURE_DESCRIPTION = "measure description";
+  private static final MeasureDouble MEASURE_DOUBLE =
+      MeasureDouble.create("measure1", MEASURE_DESCRIPTION, MEASURE_UNIT);
+  private static final MeasureLong MEASURE_LONG =
+      MeasureLong.create("measure2", MEASURE_DESCRIPTION, MEASURE_UNIT);
+  private static final String VIEW_NAME = "view";
+  private static final String VIEW_DESCRIPTION = "view description";
+  private static final Duration TEN_SECONDS = Duration.create(10, 0);
+  private static final Cumulative CUMULATIVE = Cumulative.create();
+  private static final Interval INTERVAL = Interval.create(TEN_SECONDS);
+  private static final BucketBoundaries BUCKET_BOUNDARIES =
+      BucketBoundaries.create(Arrays.asList(0.0, 1.0, 3.0, 5.0));
+  private static final Sum SUM = Sum.create();
+  private static final Count COUNT = Count.create();
+  private static final Mean MEAN = Mean.create();
+  private static final Distribution DISTRIBUTION = Distribution.create(BUCKET_BOUNDARIES);
+  private static final String PROJECT_ID = "id";
+
+  @Test
+  public void testConstant() {
+    assertThat(StackdriverExportUtils.LABEL_DESCRIPTION).isEqualTo("OpenCensus TagKey");
+  }
+
+  @Test
+  public void createLabelDescriptor() {
+    assertThat(StackdriverExportUtils.createLabelDescriptor(TagKey.create("string")))
+        .isEqualTo(
+            LabelDescriptor.newBuilder()
+                .setKey("string")
+                .setDescription(StackdriverExportUtils.LABEL_DESCRIPTION)
+                .setValueType(ValueType.STRING)
+                .build());
+    // TODO(songya): test TagKeyLong and TagKeyBoolean once they are exposed
+  }
+
+  @Test
+  public void createMetricKind() {
+    assertThat(StackdriverExportUtils.createMetricKind(CUMULATIVE))
+        .isEqualTo(MetricKind.CUMULATIVE);
+    assertThat(StackdriverExportUtils.createMetricKind(INTERVAL))
+        .isEqualTo(MetricKind.UNRECOGNIZED);
+  }
+
+  @Test
+  public void createValueType() {
+    assertThat(StackdriverExportUtils.createValueType(SUM, MEASURE_DOUBLE))
+        .isEqualTo(MetricDescriptor.ValueType.DOUBLE);
+    assertThat(StackdriverExportUtils.createValueType(SUM, MEASURE_LONG))
+        .isEqualTo(MetricDescriptor.ValueType.INT64);
+    assertThat(StackdriverExportUtils.createValueType(COUNT, MEASURE_DOUBLE))
+        .isEqualTo(MetricDescriptor.ValueType.INT64);
+    assertThat(StackdriverExportUtils.createValueType(COUNT, MEASURE_LONG))
+        .isEqualTo(MetricDescriptor.ValueType.INT64);
+    assertThat(StackdriverExportUtils.createValueType(MEAN, MEASURE_DOUBLE))
+        .isEqualTo(MetricDescriptor.ValueType.DOUBLE);
+    assertThat(StackdriverExportUtils.createValueType(MEAN, MEASURE_LONG))
+        .isEqualTo(MetricDescriptor.ValueType.DOUBLE);
+    assertThat(StackdriverExportUtils.createValueType(DISTRIBUTION, MEASURE_DOUBLE))
+        .isEqualTo(MetricDescriptor.ValueType.DISTRIBUTION);
+    assertThat(StackdriverExportUtils.createValueType(DISTRIBUTION, MEASURE_LONG))
+        .isEqualTo(MetricDescriptor.ValueType.DISTRIBUTION);
+  }
+
+  @Test
+  public void createMetric() {
+    View view =
+        View.create(
+            Name.create(VIEW_NAME),
+            VIEW_DESCRIPTION,
+            MEASURE_DOUBLE,
+            DISTRIBUTION,
+            Arrays.asList(KEY),
+            CUMULATIVE);
+    assertThat(StackdriverExportUtils.createMetric(view, Arrays.asList(VALUE_1)))
+        .isEqualTo(
+            Metric.newBuilder()
+                .setType("custom.googleapis.com/opencensus/" + VIEW_NAME)
+                .putLabels("KEY", "VALUE1")
+                .build());
+  }
+
+  @Test
+  public void convertTimestamp() {
+    Timestamp censusTimestamp1 = Timestamp.create(100, 3000);
+    assertThat(StackdriverExportUtils.convertTimestamp(censusTimestamp1))
+        .isEqualTo(
+            com.google.protobuf.Timestamp.newBuilder().setSeconds(100).setNanos(3000).build());
+
+    // Proto timestamp doesn't allow negative values, instead it will replace the negative values
+    // by returning a default instance.
+    Timestamp censusTimestamp2 = Timestamp.create(-100, -3000);
+    assertThat(StackdriverExportUtils.convertTimestamp(censusTimestamp2))
+        .isEqualTo(com.google.protobuf.Timestamp.newBuilder().build());
+  }
+
+  @Test
+  public void createTimeInterval_cumulative() {
+    Timestamp censusTimestamp1 = Timestamp.create(100, 3000);
+    Timestamp censusTimestamp2 = Timestamp.create(200, 0);
+    assertThat(
+            StackdriverExportUtils.createTimeInterval(
+                CumulativeData.create(censusTimestamp1, censusTimestamp2)))
+        .isEqualTo(
+            TimeInterval.newBuilder()
+                .setStartTime(StackdriverExportUtils.convertTimestamp(censusTimestamp1))
+                .setEndTime(StackdriverExportUtils.convertTimestamp(censusTimestamp2))
+                .build());
+  }
+
+  @Test
+  public void createTimeInterval_interval() {
+    IntervalData intervalData = IntervalData.create(Timestamp.create(200, 0));
+    // Only Cumulative view will supported in this version.
+    thrown.expect(IllegalArgumentException.class);
+    StackdriverExportUtils.createTimeInterval(intervalData);
+  }
+
+  @Test
+  public void createBucketOptions() {
+    assertThat(StackdriverExportUtils.createBucketOptions(BUCKET_BOUNDARIES))
+        .isEqualTo(
+            BucketOptions.newBuilder()
+                .setExplicitBuckets(
+                    Explicit.newBuilder().addAllBounds(Arrays.asList(0.0, 1.0, 3.0, 5.0)))
+                .build());
+  }
+
+  @Test
+  public void createDistribution() {
+    DistributionData distributionData =
+        DistributionData.create(2, 3, 0, 5, 14, Arrays.asList(0L, 1L, 1L, 0L, 1L));
+    assertThat(StackdriverExportUtils.createDistribution(distributionData, BUCKET_BOUNDARIES))
+        .isEqualTo(
+            com.google.api.Distribution.newBuilder()
+                .setMean(2)
+                .setCount(3)
+                .setRange(
+                    com.google.api.Distribution.Range.newBuilder().setMin(0).setMax(5).build())
+                .setBucketOptions(StackdriverExportUtils.createBucketOptions(BUCKET_BOUNDARIES))
+                .addAllBucketCounts(Arrays.asList(0L, 1L, 1L, 0L, 1L))
+                .setSumOfSquaredDeviation(14)
+                .build());
+  }
+
+  @Test
+  public void createTypedValue() {
+    assertThat(StackdriverExportUtils.createTypedValue(SUM, SumDataDouble.create(1.1)))
+        .isEqualTo(TypedValue.newBuilder().setDoubleValue(1.1).build());
+    assertThat(StackdriverExportUtils.createTypedValue(SUM, SumDataLong.create(10000)))
+        .isEqualTo(TypedValue.newBuilder().setInt64Value(10000).build());
+    assertThat(StackdriverExportUtils.createTypedValue(COUNT, CountData.create(55)))
+        .isEqualTo(TypedValue.newBuilder().setInt64Value(55).build());
+    assertThat(StackdriverExportUtils.createTypedValue(MEAN, MeanData.create(7.7, 8)))
+        .isEqualTo(TypedValue.newBuilder().setDoubleValue(7.7).build());
+    DistributionData distributionData =
+        DistributionData.create(2, 3, 0, 5, 14, Arrays.asList(0L, 1L, 1L, 0L, 1L));
+    assertThat(StackdriverExportUtils.createTypedValue(DISTRIBUTION, distributionData))
+        .isEqualTo(
+            TypedValue.newBuilder()
+                .setDistributionValue(
+                    StackdriverExportUtils.createDistribution(distributionData, BUCKET_BOUNDARIES))
+                .build());
+  }
+
+  @Test
+  public void createPoint_cumulative() {
+    Timestamp censusTimestamp1 = Timestamp.create(100, 3000);
+    Timestamp censusTimestamp2 = Timestamp.create(200, 0);
+    CumulativeData cumulativeData = CumulativeData.create(censusTimestamp1, censusTimestamp2);
+    SumDataDouble sumDataDouble = SumDataDouble.create(33.3);
+
+    assertThat(StackdriverExportUtils.createPoint(sumDataDouble, cumulativeData, SUM))
+        .isEqualTo(
+            Point.newBuilder()
+                .setInterval(StackdriverExportUtils.createTimeInterval(cumulativeData))
+                .setValue(StackdriverExportUtils.createTypedValue(SUM, sumDataDouble))
+                .build());
+  }
+
+  @Test
+  public void createPoint_interval() {
+    IntervalData intervalData = IntervalData.create(Timestamp.create(200, 0));
+    SumDataDouble sumDataDouble = SumDataDouble.create(33.3);
+    // Only Cumulative view will supported in this version.
+    thrown.expect(IllegalArgumentException.class);
+    StackdriverExportUtils.createPoint(sumDataDouble, intervalData, SUM);
+  }
+
+  @Test
+  public void createMetricDescriptor_cumulative() {
+    View view =
+        View.create(
+            Name.create(VIEW_NAME),
+            VIEW_DESCRIPTION,
+            MEASURE_DOUBLE,
+            DISTRIBUTION,
+            Arrays.asList(KEY),
+            CUMULATIVE);
+    MetricDescriptor metricDescriptor =
+        StackdriverExportUtils.createMetricDescriptor(view, PROJECT_ID);
+    assertThat(metricDescriptor.getName()).isEqualTo("projects/" + PROJECT_ID);
+    assertThat(metricDescriptor.getDescription()).isEqualTo(VIEW_DESCRIPTION);
+    assertThat(metricDescriptor.getDisplayName()).isEqualTo("OpenCensus/" + VIEW_NAME);
+    assertThat(metricDescriptor.getType())
+        .isEqualTo("custom.googleapis.com/opencensus/" + VIEW_NAME);
+    assertThat(metricDescriptor.getUnit()).isEqualTo(MEASURE_UNIT);
+    assertThat(metricDescriptor.getMetricKind()).isEqualTo(MetricKind.CUMULATIVE);
+    assertThat(metricDescriptor.getValueType()).isEqualTo(MetricDescriptor.ValueType.DISTRIBUTION);
+    assertThat(metricDescriptor.getLabelsList())
+        .containsExactly(
+            LabelDescriptor.newBuilder()
+                .setKey(KEY.getName())
+                .setDescription(StackdriverExportUtils.LABEL_DESCRIPTION)
+                .setValueType(ValueType.STRING)
+                .build());
+  }
+
+  @Test
+  public void createMetricDescriptor_interval() {
+    View view =
+        View.create(
+            Name.create(VIEW_NAME),
+            VIEW_DESCRIPTION,
+            MEASURE_DOUBLE,
+            DISTRIBUTION,
+            Arrays.asList(KEY),
+            INTERVAL);
+    assertThat(StackdriverExportUtils.createMetricDescriptor(view, PROJECT_ID)).isNull();
+  }
+
+  @Test
+  public void createTimeSeriesList_cumulative() {
+    View view =
+        View.create(
+            Name.create(VIEW_NAME),
+            VIEW_DESCRIPTION,
+            MEASURE_DOUBLE,
+            DISTRIBUTION,
+            Arrays.asList(KEY),
+            CUMULATIVE);
+    DistributionData distributionData1 =
+        DistributionData.create(2, 3, 0, 5, 14, Arrays.asList(0L, 1L, 1L, 0L, 1L));
+    DistributionData distributionData2 =
+        DistributionData.create(-1, 1, -1, -1, 0, Arrays.asList(1L, 0L, 0L, 0L, 0L));
+    Map<List<TagValue>, DistributionData> aggregationMap =
+        ImmutableMap.of(
+            Arrays.asList(VALUE_1), distributionData1, Arrays.asList(VALUE_2), distributionData2);
+    CumulativeData cumulativeData =
+        CumulativeData.create(Timestamp.fromMillis(1000), Timestamp.fromMillis(2000));
+    ViewData viewData = ViewData.create(view, aggregationMap, cumulativeData);
+    List<TimeSeries> timeSeriesList =
+        StackdriverExportUtils.createTimeSeriesList(viewData, PROJECT_ID);
+    assertThat(timeSeriesList).hasSize(2);
+    TimeSeries expected1 =
+        TimeSeries.newBuilder()
+            .setMetricKind(MetricKind.CUMULATIVE)
+            .setValueType(MetricDescriptor.ValueType.DISTRIBUTION)
+            .setMetric(StackdriverExportUtils.createMetric(view, Arrays.asList(VALUE_1)))
+            .setResource(
+                MonitoredResource.newBuilder()
+                    .setType("global")
+                    .putLabels("project_id", PROJECT_ID))
+            .addPoints(
+                StackdriverExportUtils.createPoint(distributionData1, cumulativeData, DISTRIBUTION))
+            .build();
+    TimeSeries expected2 =
+        TimeSeries.newBuilder()
+            .setMetricKind(MetricKind.CUMULATIVE)
+            .setValueType(MetricDescriptor.ValueType.DISTRIBUTION)
+            .setMetric(StackdriverExportUtils.createMetric(view, Arrays.asList(VALUE_2)))
+            .setResource(
+                MonitoredResource.newBuilder()
+                    .setType("global")
+                    .putLabels("project_id", PROJECT_ID))
+            .addPoints(
+                StackdriverExportUtils.createPoint(distributionData2, cumulativeData, DISTRIBUTION))
+            .build();
+    assertThat(timeSeriesList).containsExactly(expected1, expected2);
+  }
+
+  @Test
+  public void createTimeSeriesList_interval() {
+    View view =
+        View.create(
+            Name.create(VIEW_NAME),
+            VIEW_DESCRIPTION,
+            MEASURE_DOUBLE,
+            DISTRIBUTION,
+            Arrays.asList(KEY),
+            INTERVAL);
+    Map<List<TagValue>, DistributionData> aggregationMap =
+        ImmutableMap.of(
+            Arrays.asList(VALUE_1),
+            DistributionData.create(2, 3, 0, 5, 14, Arrays.asList(0L, 1L, 1L, 0L, 1L)),
+            Arrays.asList(VALUE_2),
+            DistributionData.create(-1, 1, -1, -1, 0, Arrays.asList(1L, 0L, 0L, 0L, 0L)));
+    ViewData viewData =
+        ViewData.create(view, aggregationMap, IntervalData.create(Timestamp.fromMillis(2000)));
+    assertThat(StackdriverExportUtils.createTimeSeriesList(viewData, PROJECT_ID)).isEmpty();
+  }
+}
diff --git a/exporters/stats/stackdriver/src/test/java/io/opencensus/exporter/stats/stackdriver/StackdriverStatsExporterTest.java b/exporters/stats/stackdriver/src/test/java/io/opencensus/exporter/stats/stackdriver/StackdriverStatsExporterTest.java
new file mode 100644
index 00000000..25deace6
--- /dev/null
+++ b/exporters/stats/stackdriver/src/test/java/io/opencensus/exporter/stats/stackdriver/StackdriverStatsExporterTest.java
@@ -0,0 +1,233 @@
+/*
+ * 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.stackdriver;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+import com.google.api.MetricDescriptor;
+import com.google.api.gax.rpc.UnaryCallable;
+import com.google.auth.oauth2.GoogleCredentials;
+import com.google.cloud.monitoring.v3.MetricServiceClient;
+import com.google.cloud.monitoring.v3.stub.MetricServiceStub;
+import com.google.common.collect.ImmutableMap;
+import com.google.monitoring.v3.CreateMetricDescriptorRequest;
+import com.google.monitoring.v3.CreateTimeSeriesRequest;
+import com.google.monitoring.v3.TimeSeries;
+import com.google.protobuf.Empty;
+import io.opencensus.common.Duration;
+import io.opencensus.common.Timestamp;
+import io.opencensus.stats.Aggregation.Sum;
+import io.opencensus.stats.AggregationData.SumDataLong;
+import io.opencensus.stats.Measure.MeasureLong;
+import io.opencensus.stats.View;
+import io.opencensus.stats.View.AggregationWindow.Cumulative;
+import io.opencensus.stats.View.Name;
+import io.opencensus.stats.ViewData;
+import io.opencensus.stats.ViewData.AggregationWindowData.CumulativeData;
+import io.opencensus.stats.ViewManager;
+import io.opencensus.tags.TagKey;
+import io.opencensus.tags.TagValue;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+
+/** Unit tests for {@link StackdriverStatsExporter}. */
+@RunWith(JUnit4.class)
+public class StackdriverStatsExporterTest {
+
+  private static final String PROJECT_ID = "projectId";
+  private static final Duration ONE_SECOND = Duration.create(1, 0);
+  private static final TagKey KEY = TagKey.create("KEY");
+  private static final TagValue VALUE = TagValue.create("VALUE");
+  private static final String MEASURE_NAME = "my measurement";
+  private static final String MEASURE_UNIT = "us";
+  private static final String MEASURE_DESCRIPTION = "measure description";
+  private static final MeasureLong MEASURE =
+      MeasureLong.create(MEASURE_NAME, MEASURE_DESCRIPTION, MEASURE_UNIT);
+  private static final Name VIEW_NAME = Name.create("my view");
+  private static final String VIEW_DESCRIPTION = "view description";
+  private static final Cumulative CUMULATIVE = Cumulative.create();
+  private static final Sum SUM = Sum.create();
+
+  @Mock private ViewManager mockViewManager;
+
+  @Mock private MetricServiceStub mockStub;
+
+  @Mock
+  private UnaryCallable<CreateMetricDescriptorRequest, MetricDescriptor>
+      mockCreateMetricDescriptorCallable;
+
+  @Mock private UnaryCallable<CreateTimeSeriesRequest, Empty> mockCreateTimeSeriesCallable;
+
+  @Rule public final ExpectedException thrown = ExpectedException.none();
+
+  @Before
+  public void setUp() {
+    MockitoAnnotations.initMocks(this);
+
+    doReturn(mockCreateMetricDescriptorCallable).when(mockStub).createMetricDescriptorCallable();
+    doReturn(mockCreateTimeSeriesCallable).when(mockStub).createTimeSeriesCallable();
+    doReturn(null)
+        .when(mockCreateMetricDescriptorCallable)
+        .call(any(CreateMetricDescriptorRequest.class));
+    doReturn(null).when(mockCreateTimeSeriesCallable).call(any(CreateTimeSeriesRequest.class));
+  }
+
+  @Test
+  public void createWithNullCredentials() throws IOException {
+    thrown.expect(NullPointerException.class);
+    thrown.expectMessage("credentials");
+    StackdriverStatsExporter.createWithCredentialsAndProjectId(null, PROJECT_ID, ONE_SECOND);
+  }
+
+  @Test
+  public void createWithNullProjectId() throws IOException {
+    thrown.expect(NullPointerException.class);
+    thrown.expectMessage("projectId");
+    StackdriverStatsExporter.createWithCredentialsAndProjectId(
+        GoogleCredentials.newBuilder().build(), null, ONE_SECOND);
+  }
+
+  @Test
+  public void createWithNullDuration() throws IOException {
+    thrown.expect(NullPointerException.class);
+    thrown.expectMessage("exportInterval");
+    StackdriverStatsExporter.createWithCredentialsAndProjectId(
+        GoogleCredentials.newBuilder().build(), PROJECT_ID, null);
+  }
+
+  @Test
+  public void createWithNegativeDuration() throws IOException {
+    thrown.expect(IllegalArgumentException.class);
+    thrown.expectMessage("Duration must be positive");
+    StackdriverStatsExporter.createWithCredentialsAndProjectId(
+        GoogleCredentials.newBuilder().build(), PROJECT_ID, Duration.create(-1, 0));
+  }
+
+  @Test
+  public void createHandlerTwice() throws IOException {
+    StackdriverStatsExporter.createWithCredentialsAndProjectId(
+        GoogleCredentials.newBuilder().build(), PROJECT_ID, ONE_SECOND);
+    try {
+      thrown.expect(IllegalStateException.class);
+      thrown.expectMessage("Stackdriver stats exporter is already created.");
+      StackdriverStatsExporter.createWithCredentialsAndProjectId(
+          GoogleCredentials.newBuilder().build(), PROJECT_ID, ONE_SECOND);
+    } finally {
+      StackdriverStatsExporter.unsafeSetExporter(null);
+    }
+  }
+
+  @Test
+  public void registerViewAndExport() throws IOException {
+    View view =
+        View.create(VIEW_NAME, VIEW_DESCRIPTION, MEASURE, SUM, Arrays.asList(KEY), CUMULATIVE);
+    ViewData viewData =
+        ViewData.create(
+            view,
+            ImmutableMap.of(Arrays.asList(VALUE), SumDataLong.create(1)),
+            CumulativeData.create(Timestamp.fromMillis(100), Timestamp.fromMillis(200)));
+    doNothing().when(mockViewManager).registerView(view);
+    doReturn(viewData).when(mockViewManager).getView(VIEW_NAME);
+
+    try {
+      StackdriverStatsExporter.unsafeSetExporter(
+          new StackdriverStatsExporter(
+              PROJECT_ID, new FakeMetricServiceClient(mockStub), ONE_SECOND, mockViewManager));
+      StackdriverStatsExporter.registerView(view);
+
+      verify(mockStub, times(1)).createMetricDescriptorCallable();
+      // The timeout for verifying createTimeSeries needs to match the export interval of exporter.
+      verify(mockStub, timeout(1000).times(1)).createTimeSeriesCallable();
+
+      MetricDescriptor descriptor = StackdriverExportUtils.createMetricDescriptor(view, PROJECT_ID);
+      List<TimeSeries> timeSeries =
+          StackdriverExportUtils.createTimeSeriesList(viewData, PROJECT_ID);
+      verify(mockCreateMetricDescriptorCallable, times(1))
+          .call(
+              eq(
+                  CreateMetricDescriptorRequest.newBuilder()
+                      .setMetricDescriptor(descriptor)
+                      .build()));
+      verify(mockCreateTimeSeriesCallable, times(1))
+          .call(eq(CreateTimeSeriesRequest.newBuilder().addAllTimeSeries(timeSeries).build()));
+    } finally {
+      StackdriverStatsExporter.unsafeSetExporter(null);
+    }
+  }
+
+  @Test
+  public void preventRegisterViewBeforeCreateExporter() throws IOException {
+    View view1 =
+        View.create(VIEW_NAME, VIEW_DESCRIPTION, MEASURE, SUM, Arrays.asList(KEY), CUMULATIVE);
+    thrown.expect(IllegalStateException.class);
+    thrown.expectMessage("Stackdriver stats exporter has not been created.");
+    StackdriverStatsExporter.registerView(view1);
+  }
+
+  @Test
+  public void preventRegisteringDifferentViewWithSameName() throws IOException {
+    doNothing().when(mockViewManager).registerView(any(View.class));
+    StackdriverStatsExporter.unsafeSetExporter(
+        new StackdriverStatsExporter(
+            PROJECT_ID, new FakeMetricServiceClient(mockStub), ONE_SECOND, mockViewManager));
+    View view1 =
+        View.create(VIEW_NAME, VIEW_DESCRIPTION, MEASURE, SUM, Arrays.asList(KEY), CUMULATIVE);
+    StackdriverStatsExporter.registerView(view1);
+    View view2 =
+        View.create(
+            VIEW_NAME,
+            "This is a different description.",
+            MEASURE,
+            SUM,
+            Arrays.asList(KEY),
+            CUMULATIVE);
+    try {
+      thrown.expect(IllegalArgumentException.class);
+      thrown.expectMessage("A different view with the same name is already registered: ");
+      StackdriverStatsExporter.registerView(view2);
+    } finally {
+      StackdriverStatsExporter.unsafeSetExporter(null);
+    }
+  }
+
+  /*
+   * MetricServiceClient.createMetricDescriptor() and MetricServiceClient.createTimeSeries() are
+   * final methods and cannot be mocked. We have to use a mock MetricServiceStub in order to verify
+   * the output.
+   */
+  private static final class FakeMetricServiceClient extends MetricServiceClient {
+
+    protected FakeMetricServiceClient(MetricServiceStub stub) {
+      super(stub);
+    }
+  }
+}
diff --git a/settings.gradle b/settings.gradle
index cc494a4a..909c171f 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -8,6 +8,7 @@ include ":opencensus-testing"
 include ":opencensus-exporter-trace-logging"
 include ":opencensus-exporter-trace-stackdriver"
 include ":opencensus-exporter-trace-zipkin"
+include ":opencensus-exporter-stats-stackdriver"
 include ":core_impl"
 include ":core_impl_java"
 include ":core_impl_android"
@@ -28,6 +29,7 @@ project(':opencensus-exporter-trace-logging').projectDir =
 project(':opencensus-exporter-trace-stackdriver').projectDir =
         "$rootDir/exporters/trace/stackdriver" as File
 project(':opencensus-exporter-trace-zipkin').projectDir = "$rootDir/exporters/trace/zipkin" as File
+project(':opencensus-exporter-stats-stackdriver').projectDir = "$rootDir/exporters/stats/stackdriver" as File
 
 // Java8 projects only
 if (JavaVersion.current().isJava8Compatible()) {