Support custom MonitoredResource for Stackdriver Stats Exporter (#870)

exporters/stats/stackdriver/README.md

@@ -67,6 +67,43 @@ public class MyMainClass {
 }
 ```
 
+#### Set Monitored Resource for exporter
+
+By default, the Stackdriver Stats Exporter uses [a global Stackdriver monitored resource with no 
+labels](https://cloud.google.com/monitoring/api/resources#tag_global), and this works fine when you 
+have only one exporter running. However, if you want to have multiple processes exporting stats for 
+the same metric concurrently, using the default monitored resource for all exporters will not work. 
+In this case you need to associate a unique monitored resource with each exporter:
+
+```java
+public class MyMainClass {
+  public static void main(String[] args) {
+    // A sample AWS EC2 monitored resource.
+    // This will only work if each EC2 has one process that records stats. If there are multiple 
+    // processes, you'll need an extra label such as pid.
+    MonitoredResource myResource = MonitoredResource.newBuilder()
+                                               .setType("aws_ec2_instance")
+                                               .putLabels("instance_id", "instance")
+                                               .putLabels("aws_account", "account")
+                                               .putLabels("region", "aws:us-west-2")
+                                               .build();
+
+    // Set a custom MonitoredResource. Please make sure each Stackdriver Stats Exporter has a 
+    // unique MonitoredResource.      
+    StackdriverStatsExporter.createAndRegisterWithProjectIdAndMonitoredResource(
+        "MyStackdriverProjectId", 
+        Duration.create(10, 0), 
+        myResource);
+  }
+}
+```
+
+For a complete list of valid Stackdriver monitored resources, please refer to [Stackdriver 
+Documentation](https://cloud.google.com/monitoring/custom-metrics/creating-metrics#which-resource).
+Please also note that although there are a lot of monitored resources available on [Stackdriver](https://cloud.google.com/monitoring/api/resources), 
+only [a small subset of them](https://cloud.google.com/monitoring/custom-metrics/creating-metrics#which-resource) 
+are compatible with the Opencensus Stackdriver Stats Exporter.
+
 #### Authentication
 
 This exporter uses [google-cloud-java](https://github.com/GoogleCloudPlatform/google-cloud-java),

exporters/stats/stackdriver/src/main/java/io/opencensus/exporter/stats/stackdriver/StackdriverExportUtils.java

@@ -27,9 +27,9 @@
 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.common.collect.Maps;
 import com.google.monitoring.v3.Point;
 import com.google.monitoring.v3.TimeInterval;
 import com.google.monitoring.v3.TimeSeries;
@@ -137,7 +137,8 @@ static MetricDescriptor.ValueType createValueType(
   }
 
   // Convert ViewData to a list of TimeSeries, so that ViewData can be uploaded to Stackdriver.
-  static List<TimeSeries> createTimeSeriesList(ViewData viewData, String projectId) {
+  static List<TimeSeries> createTimeSeriesList(
+      ViewData viewData, MonitoredResource monitoredResource) {
     List<TimeSeries> timeSeriesList = Lists.newArrayList();
     if (viewData == null) {
       return timeSeriesList;
@@ -151,8 +152,7 @@ static List<TimeSeries> createTimeSeriesList(ViewData viewData, String projectId
     // Shared fields for all TimeSeries generated from the same ViewData
     TimeSeries.Builder shared = TimeSeries.newBuilder();
     shared.setMetricKind(createMetricKind(view.getWindow()));
-    // TODO(songya): add support for custom resource labels.
-    shared.setResource(MonitoredResource.newBuilder().setType("global"));
+    shared.setResource(monitoredResource);
     shared.setValueType(createValueType(view.getAggregation(), view.getMeasure()));
 
     // Each entry in AggregationMap will be converted into an independent TimeSeries object

exporters/stats/stackdriver/src/main/java/io/opencensus/exporter/stats/stackdriver/StackdriverExporterWorkerThread.java

@@ -17,6 +17,7 @@
 package io.opencensus.exporter.stats.stackdriver;
 
 import com.google.api.MetricDescriptor;
+import com.google.api.MonitoredResource;
 import com.google.cloud.monitoring.v3.MetricServiceClient;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.Lists;
@@ -56,18 +57,21 @@ final class StackdriverExporterWorkerThread extends Thread {
   private final ProjectName projectName;
   private final MetricServiceClient metricServiceClient;
   private final ViewManager viewManager;
+  private final MonitoredResource monitoredResource;
   private final Map<View.Name, View> registeredViews = new HashMap<View.Name, View>();
 
   StackdriverExporterWorkerThread(
       String projectId,
       MetricServiceClient metricServiceClient,
       Duration exportInterval,
-      ViewManager viewManager) {
+      ViewManager viewManager,
+      MonitoredResource monitoredResource) {
     this.scheduleDelayMillis = toMillis(exportInterval);
     this.projectId = projectId;
     projectName = ProjectName.newBuilder().setProject(projectId).build();
     this.metricServiceClient = metricServiceClient;
     this.viewManager = viewManager;
+    this.monitoredResource = monitoredResource;
     setDaemon(true);
     setName("ExportWorkerThread");
   }
@@ -129,7 +133,8 @@ void export() {
     }
     List<TimeSeries> timeSeriesList = Lists.newArrayList();
     for (ViewData viewData : viewDataList) {
-      timeSeriesList.addAll(StackdriverExportUtils.createTimeSeriesList(viewData, projectId));
+      timeSeriesList.addAll(
+          StackdriverExportUtils.createTimeSeriesList(viewData, monitoredResource));
     }
 
     for (List<TimeSeries> batchedTimeSeries :

exporters/stats/stackdriver/src/main/java/io/opencensus/exporter/stats/stackdriver/StackdriverStatsExporter.java

@@ -16,10 +16,11 @@
 
 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.checkArgument;
+import static com.google.common.base.Preconditions.checkNotNull;
 import static com.google.common.base.Preconditions.checkState;
 
+import com.google.api.MonitoredResource;
 import com.google.api.gax.core.FixedCredentialsProvider;
 import com.google.auth.Credentials;
 import com.google.auth.oauth2.GoogleCredentials;
@@ -57,21 +58,25 @@ public final class StackdriverStatsExporter {
   private static StackdriverStatsExporter exporter = null;
 
   private static final Duration ZERO = Duration.create(0, 0);
+  private static final MonitoredResource DEFAULT_RESOURCE =
+      MonitoredResource.newBuilder().setType("global").build();
 
   @VisibleForTesting
   StackdriverStatsExporter(
       String projectId,
       MetricServiceClient metricServiceClient,
       Duration exportInterval,
-      ViewManager viewManager) {
+      ViewManager viewManager,
+      MonitoredResource monitoredResource) {
     checkArgument(exportInterval.compareTo(ZERO) > 0, "Duration must be positive");
     this.workerThread =
         new StackdriverExporterWorkerThread(
-            projectId, metricServiceClient, exportInterval, viewManager);
+            projectId, metricServiceClient, exportInterval, viewManager, monitoredResource);
   }
 
   /**
-   * Creates a StackdriverStatsExporter for an explicit project ID and using explicit credentials.
+   * Creates a StackdriverStatsExporter for an explicit project ID and using explicit credentials,
+   * with default Monitored Resource.
    *
    * <p>Only one Stackdriver exporter can be created.
    *
@@ -85,11 +90,12 @@ public static void createAndRegisterWithCredentialsAndProjectId(
     checkNotNull(credentials, "credentials");
     checkNotNull(projectId, "projectId");
     checkNotNull(exportInterval, "exportInterval");
-    createInternal(credentials, projectId, exportInterval);
+    createInternal(credentials, projectId, exportInterval, null);
   }
 
   /**
-   * Creates a Stackdriver Stats exporter for an explicit project ID.
+   * Creates a Stackdriver Stats exporter for an explicit project ID, with default Monitored
+   * Resource.
    *
    * <p>Only one Stackdriver exporter can be created.
    *
@@ -111,11 +117,11 @@ public static void createAndRegisterWithProjectId(String projectId, Duration exp
       throws IOException {
     checkNotNull(projectId, "projectId");
     checkNotNull(exportInterval, "exportInterval");
-    createInternal(null, projectId, exportInterval);
+    createInternal(null, projectId, exportInterval, null);
   }
 
   /**
-   * Creates a Stackdriver Stats exporter.
+   * Creates a Stackdriver Stats exporter with default Monitored Resource.
    *
    * <p>Only one Stackdriver exporter can be created.
    *
@@ -135,12 +141,65 @@ public static void createAndRegisterWithProjectId(String projectId, Duration exp
    */
   public static void createAndRegister(Duration exportInterval) throws IOException {
     checkNotNull(exportInterval, "exportInterval");
-    createInternal(null, ServiceOptions.getDefaultProjectId(), exportInterval);
+    createInternal(null, ServiceOptions.getDefaultProjectId(), exportInterval, null);
+  }
+
+  /**
+   * Creates a Stackdriver Stats exporter with an explicit project ID and a custom Monitored
+   * Resource.
+   *
+   * <p>Only one Stackdriver exporter can be created.
+   *
+   * <p>Please refer to cloud.google.com/monitoring/custom-metrics/creating-metrics#which-resource
+   * for a list of valid {@code MonitoredResource}s.
+   *
+   * <p>This uses the default application credentials. See {@link
+   * GoogleCredentials#getApplicationDefault}.
+   *
+   * @param projectId the cloud project id.
+   * @param exportInterval the interval between pushing stats to StackDriver.
+   * @param monitoredResource the Monitored Resource used by exporter.
+   * @throws IllegalStateException if a Stackdriver exporter is already created.
+   */
+  public static void createAndRegisterWithProjectIdAndMonitoredResource(
+      String projectId, Duration exportInterval, MonitoredResource monitoredResource)
+      throws IOException {
+    checkNotNull(projectId, "projectId");
+    checkNotNull(exportInterval, "exportInterval");
+    checkNotNull(monitoredResource, "monitoredResource");
+    createInternal(null, projectId, exportInterval, monitoredResource);
+  }
+
+  /**
+   * Creates a Stackdriver Stats exporter with a custom Monitored Resource.
+   *
+   * <p>Only one Stackdriver exporter can be created.
+   *
+   * <p>Please refer to cloud.google.com/monitoring/custom-metrics/creating-metrics#which-resource
+   * for a list of valid {@code MonitoredResource}s.
+   *
+   * <p>This uses the default application credentials. See {@link
+   * GoogleCredentials#getApplicationDefault}.
+   *
+   * <p>This uses the default project ID configured see {@link ServiceOptions#getDefaultProjectId}.
+   *
+   * @param exportInterval the interval between pushing stats to StackDriver.
+   * @param monitoredResource the Monitored Resource used by exporter.
+   * @throws IllegalStateException if a Stackdriver exporter is already created.
+   */
+  public static void createAndRegisterWithMonitoredResource(
+      Duration exportInterval, MonitoredResource monitoredResource) throws IOException {
+    checkNotNull(exportInterval, "exportInterval");
+    checkNotNull(monitoredResource, "monitoredResource");
+    createInternal(null, ServiceOptions.getDefaultProjectId(), exportInterval, monitoredResource);
   }
 
   // Use createInternal() (instead of constructor) to enforce singleton.
   private static void createInternal(
-      @Nullable Credentials credentials, String projectId, Duration exportInterval)
+      @Nullable Credentials credentials,
+      String projectId,
+      Duration exportInterval,
+      @Nullable MonitoredResource monitoredResource)
       throws IOException {
     synchronized (monitor) {
       checkState(exporter == null, "Stackdriver stats exporter is already created.");
@@ -155,9 +214,16 @@ private static void createInternal(
                     .setCredentialsProvider(FixedCredentialsProvider.create(credentials))
                     .build());
       }
+      if (monitoredResource == null) {
+        monitoredResource = DEFAULT_RESOURCE;
+      }
       exporter =
           new StackdriverStatsExporter(
-              projectId, metricServiceClient, exportInterval, Stats.getViewManager());
+              projectId,
+              metricServiceClient,
+              exportInterval,
+              Stats.getViewManager(),
+              monitoredResource);
       exporter.workerThread.start();
     }
   }

exporters/stats/stackdriver/src/test/java/io/opencensus/exporter/stats/stackdriver/StackdriverExportUtilsTest.java

@@ -92,6 +92,8 @@ public class StackdriverExportUtilsTest {
   private static final Mean MEAN = Mean.create();
   private static final Distribution DISTRIBUTION = Distribution.create(BUCKET_BOUNDARIES);
   private static final String PROJECT_ID = "id";
+  private static final MonitoredResource DEFAULT_RESOURCE =
+      MonitoredResource.newBuilder().setType("global").build();
 
   @Test
   public void testConstant() {
@@ -366,7 +368,7 @@ public void createTimeSeriesList_cumulative() {
         CumulativeData.create(Timestamp.fromMillis(1000), Timestamp.fromMillis(2000));
     ViewData viewData = ViewData.create(view, aggregationMap, cumulativeData);
     List<TimeSeries> timeSeriesList =
-        StackdriverExportUtils.createTimeSeriesList(viewData, PROJECT_ID);
+        StackdriverExportUtils.createTimeSeriesList(viewData, DEFAULT_RESOURCE);
     assertThat(timeSeriesList).hasSize(2);
     TimeSeries expected1 =
         TimeSeries.newBuilder()
@@ -407,6 +409,37 @@ public void createTimeSeriesList_interval() {
             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();
+    assertThat(StackdriverExportUtils.createTimeSeriesList(viewData, DEFAULT_RESOURCE)).isEmpty();
+  }
+
+  @Test
+  public void createTimeSeriesList_withCustomMonitoredResource() {
+    MonitoredResource resource =
+        MonitoredResource.newBuilder().setType("global").putLabels("key", "value").build();
+    View view =
+        View.create(
+            Name.create(VIEW_NAME),
+            VIEW_DESCRIPTION,
+            MEASURE_DOUBLE,
+            SUM,
+            Arrays.asList(KEY),
+            CUMULATIVE);
+    SumDataDouble sumData = SumDataDouble.create(55.5);
+    Map<List<TagValue>, SumDataDouble> aggregationMap =
+        ImmutableMap.of(Arrays.asList(VALUE_1), sumData);
+    CumulativeData cumulativeData =
+        CumulativeData.create(Timestamp.fromMillis(1000), Timestamp.fromMillis(2000));
+    ViewData viewData = ViewData.create(view, aggregationMap, cumulativeData);
+    List<TimeSeries> timeSeriesList =
+        StackdriverExportUtils.createTimeSeriesList(viewData, resource);
+    assertThat(timeSeriesList)
+        .containsExactly(
+            TimeSeries.newBuilder()
+                .setMetricKind(MetricKind.CUMULATIVE)
+                .setValueType(MetricDescriptor.ValueType.DOUBLE)
+                .setMetric(StackdriverExportUtils.createMetric(view, Arrays.asList(VALUE_1)))
+                .setResource(resource)
+                .addPoints(StackdriverExportUtils.createPoint(sumData, cumulativeData, SUM))
+                .build());
   }
 }