Add metrics exporter

build.gradle

@@ -18,7 +18,9 @@ subprojects {
 
     ext {
         autoValueVersion = '1.7.4'
+        slf4jVersion = '1.7.30'
         googleCloudVersion = '1.0.2'
+        cloudMonitoringVersion = '2.0.1'
         openTelemetryVersion = '0.7.0'
         junitVersion = '4.13'
         opentelemetryOperationsVersion = "0.7.1"  // CURRENT_RELEASE_VERSION
@@ -27,14 +29,18 @@ subprojects {
                 auto_value_annotations       : "com.google.auto.value:auto-value-annotations:${autoValueVersion}",
                 auto_value                   : "com.google.auto.value:auto-value:${autoValueVersion}",
                 google_cloud_core            : "com.google.cloud:google-cloud-core:${googleCloudVersion}",
-                google_cloud_grpc            : "com.google.api.grpc:grpc-google-cloud-trace-v2:${googleCloudVersion}",
                 google_cloud_trace           : "com.google.cloud:google-cloud-trace:${googleCloudVersion}",
+                google_cloud_trace_grpc      : "com.google.api.grpc:grpc-google-cloud-trace-v2:${googleCloudVersion}",
+                google_cloud_monitoring      : "com.google.cloud:google-cloud-monitoring:${cloudMonitoringVersion}",
+                google_cloud_monitoring_grpc : "com.google.cloud:grpc-google-cloud-monitoring-v3:${cloudMonitoringVersion}",
+                slf4j                        : "org.slf4j:slf4j-api:${slf4jVersion}",
                 opentelemetry_api            : "io.opentelemetry:opentelemetry-api:${openTelemetryVersion}",
                 opentelemetry_sdk            : "io.opentelemetry:opentelemetry-sdk:${openTelemetryVersion}",
                 opentelemetry_operations_java: "com.google.cloud.opentelemetry:exporter-trace:${opentelemetryOperationsVersion}"
         ]
         testLibraries = [
-                junit: "junit:junit:${junitVersion}",
+                junit       : "junit:junit:${junitVersion}",
+                slf4j_simple: "org.slf4j:slf4j-simple:${slf4jVersion}"
         ]
     }
 

exporters/metrics/build.gradle

@@ -0,0 +1,13 @@
+description = 'Cloud Monitoring Exporter for OpenTelemetry'
+
+dependencies {
+    api(libraries.auto_value_annotations)
+    api(libraries.slf4j)
+    annotationProcessor(libraries.auto_value)
+    api(libraries.opentelemetry_api)
+    api(libraries.opentelemetry_sdk)
+    api(libraries.google_cloud_core)
+    api(libraries.google_cloud_monitoring)
+    testImplementation(testLibraries.junit)
+    testImplementation(testLibraries.slf4j_simple)
+}

exporters/metrics/src/main/java/com.google.cloud.opentelemetry.metric/MetricConfiguration.java

@@ -0,0 +1,104 @@
+package com.google.cloud.opentelemetry.metric;
+
+import static java.time.Duration.ZERO;
+
+import com.google.auth.Credentials;
+import com.google.auto.value.AutoValue;
+import com.google.cloud.ServiceOptions;
+import com.google.cloud.monitoring.v3.stub.MetricServiceStub;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
+import java.time.Duration;
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.Immutable;
+
+/**
+ * Configurations for {@link MetricExporter}.
+ */
+@AutoValue
+@Immutable
+public abstract class MetricConfiguration {
+
+    private static final String DEFAULT_PROJECT_ID =
+        Strings.nullToEmpty(ServiceOptions.getDefaultProjectId());
+    private static final Duration DEFAULT_DEADLINE = Duration.ofSeconds(10, 0);
+
+    MetricConfiguration() {
+    }
+
+    /**
+     * Returns the {@link Credentials}.
+     *
+     * @return the {@code Credentials}.
+     */
+    @Nullable
+    public abstract Credentials getCredentials();
+
+    /**
+     * Returns the cloud project id.
+     *
+     * @return the cloud project id.
+     */
+    public abstract String getProjectId();
+
+    /**
+     * Returns a MetricsServiceStub instance used to make RPC calls.
+     *
+     * @return the metrics service stub.
+     */
+    @Nullable
+    public abstract MetricServiceStub getMetricServiceStub();
+
+    /**
+     * Returns the deadline for exporting to Cloud Monitoring backend.
+     *
+     * <p>Default value is 10 seconds.
+     *
+     * @return the export deadline.
+     */
+    public abstract Duration getDeadline();
+
+    public static Builder builder() {
+        return new AutoValue_MetricConfiguration.Builder()
+            .setProjectId(DEFAULT_PROJECT_ID)
+            .setDeadline(DEFAULT_DEADLINE);
+    }
+
+    /**
+     * Builder for {@link MetricConfiguration}.
+     */
+    @AutoValue.Builder
+    public abstract static class Builder {
+
+        Builder() {
+        }
+
+        abstract String getProjectId();
+
+        abstract Duration getDeadline();
+
+        public abstract Builder setProjectId(String projectId);
+
+        public abstract Builder setCredentials(Credentials newCredentials);
+
+        public abstract Builder setMetricServiceStub(MetricServiceStub newMetricServiceStub);
+
+        public abstract Builder setDeadline(Duration deadline);
+
+        abstract MetricConfiguration autoBuild();
+
+        /**
+         * Builds a {@link MetricConfiguration}.
+         *
+         * @return a {@code MetricsConfiguration}.
+         */
+        public MetricConfiguration build() {
+            Preconditions.checkArgument(
+                !Strings.isNullOrEmpty(getProjectId()),
+                "Cannot find a project ID from either configurations or application default.");
+            Preconditions.checkArgument(getDeadline().compareTo(ZERO) > 0, "Deadline must be positive.");
+            return autoBuild();
+        }
+    }
+
+}

exporters/metrics/src/main/java/com.google.cloud.opentelemetry.metric/MetricExporter.java

@@ -0,0 +1,201 @@
+package com.google.cloud.opentelemetry.metric;
+
+import static com.google.api.client.util.Preconditions.checkNotNull;
+import static com.google.cloud.opentelemetry.metric.MetricTranslator.mapMetric;
+import static com.google.cloud.opentelemetry.metric.MetricTranslator.mapMetricDescriptor;
+import static com.google.cloud.opentelemetry.metric.MetricTranslator.mapPoint;
+
+import com.google.api.Metric;
+import com.google.api.MetricDescriptor;
+import com.google.api.MonitoredResource;
+import com.google.api.gax.core.FixedCredentialsProvider;
+import com.google.auth.Credentials;
+import com.google.auth.oauth2.GoogleCredentials;
+import com.google.cloud.monitoring.v3.MetricServiceClient;
+import com.google.cloud.monitoring.v3.MetricServiceSettings;
+import com.google.cloud.monitoring.v3.stub.MetricServiceStub;
+import com.google.common.collect.Lists;
+import com.google.monitoring.v3.CreateMetricDescriptorRequest;
+import com.google.monitoring.v3.Point;
+import com.google.monitoring.v3.ProjectName;
+import com.google.monitoring.v3.TimeSeries;
+import io.opentelemetry.common.Labels;
+import io.opentelemetry.sdk.metrics.data.MetricData;
+import java.io.IOException;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MetricExporter implements io.opentelemetry.sdk.metrics.export.MetricExporter {
+
+  private static final String PROJECT_NAME_PREFIX = "projects/";
+  private static final long WRITE_INTERVAL_SECOND = 12;
+  private static final int MAX_BATCH_SIZE = 200;
+  private static final long NANO_PER_SECOND = (long) 1e9;
+
+  private static final Logger logger = LoggerFactory.getLogger(MetricExporter.class);
+
+  private final MetricServiceClient metricServiceClient;
+  private final String projectId;
+  private final Instant exporterStartTime;
+  private final Map<MetricWithLabels, Long> lastUpdatedTime = new HashMap<>();
+
+  MetricExporter(
+      String projectId,
+      MetricServiceClient client
+  ) {
+    this.projectId = projectId;
+    this.metricServiceClient = client;
+    this.exporterStartTime = Instant.now();
+  }
+
+  public static MetricExporter createWithDefaultConfiguration() throws IOException {
+    MetricConfiguration configuration = MetricConfiguration.builder().build();
+    return MetricExporter.createWithConfiguration(configuration);
+  }
+
+  public static MetricExporter createWithConfiguration(MetricConfiguration configuration)
+      throws IOException {
+    String projectId = configuration.getProjectId();
+    MetricServiceStub stub = configuration.getMetricServiceStub();
+
+    if (stub == null) {
+      Credentials credentials =
+          configuration.getCredentials() == null
+              ? GoogleCredentials.getApplicationDefault()
+              : configuration.getCredentials();
+
+      return MetricExporter.createWithCredentials(
+          projectId, credentials, configuration.getDeadline());
+    }
+    return MetricExporter.createWithClient(
+        projectId, MetricServiceClient.create(stub));
+  }
+
+  private static MetricExporter createWithClient(
+      String projectId,
+      MetricServiceClient metricServiceClient) {
+    return new MetricExporter(projectId, metricServiceClient);
+  }
+
+  private static MetricExporter createWithCredentials(
+      String projectId,
+      Credentials credentials,
+      Duration deadline) throws IOException {
+    MetricServiceSettings.Builder builder =
+        MetricServiceSettings.newBuilder()
+            .setCredentialsProvider(
+                FixedCredentialsProvider.create(checkNotNull(credentials, "Credentials not provided.")));
+    builder.createMetricDescriptorSettings()
+        .setSimpleTimeoutNoRetries(org.threeten.bp.Duration.ofMillis(deadline.toMillis()));
+    return new MetricExporter(projectId, MetricServiceClient.create(builder.build()));
+  }
+
+  @Override
+  public ResultCode export(Collection<MetricData> metrics) {
+    List<TimeSeries> allTimesSeries = new ArrayList<>();
+
+    for (MetricData metricData : metrics) {
+      MetricDescriptor descriptor = mapMetricDescriptor(metricData);
+      if (descriptor == null) {
+        continue;
+      }
+      metricServiceClient.createMetricDescriptor(
+          CreateMetricDescriptorRequest.newBuilder().setName(PROJECT_NAME_PREFIX + projectId)
+              .setMetricDescriptor(descriptor)
+              .build());
+
+      MetricWithLabels updateKey = new MetricWithLabels(descriptor.getType(),
+          metricData.getDescriptor().getConstantLabels());
+
+      // We are expecting one point per MetricData
+      if (metricData.getPoints().size() != 1) {
+        logger.error("There should be exactly one point in each metricData, found {}",
+            metricData.getPoints().size());
+        continue;
+      }
+      MetricData.Point metricPoint = metricData.getPoints().iterator().next();
+
+      // Cloud Monitoring API allows, for any combination of labels and
+      // metric name, one update per WRITE_INTERVAL seconds
+      long pointCollectionTime = metricPoint.getEpochNanos();
+      if (lastUpdatedTime.containsKey(updateKey)
+          && pointCollectionTime <= lastUpdatedTime.get(updateKey) / NANO_PER_SECOND + WRITE_INTERVAL_SECOND) {
+        continue;
+      }
+
+      Metric metric = mapMetric(metricData, descriptor.getType());
+      Point point = mapPoint(metricData, metricPoint, updateKey, lastUpdatedTime);
+      if (point == null) {
+        continue;
+      }
+
+      allTimesSeries.add(TimeSeries.newBuilder()
+          .setMetric(metric)
+          .addPoints(point)
+          .setResource(MonitoredResource.newBuilder().build())
+          .setMetricKind(descriptor.getMetricKind())
+          .build());
+    }
+    createTimeSeriesBatch(metricServiceClient, ProjectName.of(projectId), allTimesSeries);
+    return ResultCode.SUCCESS;
+  }
+
+  private static void createTimeSeriesBatch(MetricServiceClient metricServiceClient, ProjectName projectName,
+      List<TimeSeries> allTimesSeries) {
+    List<List<TimeSeries>> batches = Lists.partition(allTimesSeries, MAX_BATCH_SIZE);
+    for (List<TimeSeries> timeSeries : batches) {
+      metricServiceClient.createTimeSeries(projectName, new ArrayList<>(timeSeries));
+    }
+  }
+
+  /**
+   * The exporter does not batch metrics, so this method will immediately return with success.
+   *
+   * @return always Success
+   */
+  @Override
+  public ResultCode flush() {
+    return ResultCode.SUCCESS;
+  }
+
+  @Override
+  public void shutdown() {
+    metricServiceClient.shutdown();
+  }
+
+  static class MetricWithLabels {
+
+    private final String metricType;
+    private final Labels labels;
+
+    MetricWithLabels(String metricType, Labels labels) {
+      this.metricType = metricType;
+      this.labels = labels;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) {
+        return true;
+      }
+      if (o == null || getClass() != o.getClass()) {
+        return false;
+      }
+      MetricWithLabels that = (MetricWithLabels) o;
+      return Objects.equals(metricType, that.metricType) && Objects.equals(labels, that.labels);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(metricType, labels);
+    }
+  }
+}