Fix duplicate metrics in timeseries (#231)

* Fixes export error due to multiple metric points in TimeSeries

* Cleanup debug statements

* Fix formatting

* Refactor: removes duplicate code

* Updates test with expected instrumentation labels

* Refactor: extract label values to final constants

Author: psx95

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

@@ -22,17 +22,22 @@
 import static com.google.cloud.opentelemetry.metric.ResourceTranslator.mapResource;
 
 import com.google.api.MetricDescriptor;
+import com.google.monitoring.v3.Point;
 import com.google.monitoring.v3.TimeSeries;
 import com.google.monitoring.v3.TypedValue;
+import io.opentelemetry.api.common.AttributeKey;
 import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.sdk.common.InstrumentationScopeInfo;
 import io.opentelemetry.sdk.metrics.data.DoublePointData;
 import io.opentelemetry.sdk.metrics.data.HistogramPointData;
 import io.opentelemetry.sdk.metrics.data.LongPointData;
 import io.opentelemetry.sdk.metrics.data.MetricData;
+import io.opentelemetry.sdk.metrics.data.PointData;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.stream.Collectors;
 
 /**
@@ -41,6 +46,9 @@
  */
 public final class AggregateByLabelMetricTimeSeriesBuilder implements MetricTimeSeriesBuilder {
 
+  public static final String LABEL_INSTRUMENTATION_LIBRARY_NAME = "instrumentation_source";
+  public static final String LABEL_INSTRUMENTATION_LIBRARY_VERSION = "instrumentation_version";
+
   private final Map<String, MetricDescriptor> descriptors = new HashMap<>();
   private final Map<MetricWithLabels, TimeSeries.Builder> pendingTimeSeries = new HashMap<>();
   private final String projectId;
@@ -52,58 +60,53 @@ public AggregateByLabelMetricTimeSeriesBuilder(String projectId, String prefix)
   }
 
   @Override
-  public void recordPoint(MetricData metric, LongPointData point) {
-    MetricDescriptor descriptor = mapMetricDescriptor(this.prefix, metric, point);
-    if (descriptor == null) {
-      // Unsupported type.
-      return;
-    }
-    descriptors.putIfAbsent(descriptor.getType(), descriptor);
-    MetricWithLabels key = new MetricWithLabels(descriptor.getType(), point.getAttributes());
-    // TODO: Check lastExportTime and ensure we don't send too often...
-    pendingTimeSeries
-        .computeIfAbsent(key, k -> makeTimeSeriesHeader(metric, point.getAttributes(), descriptor))
-        .addPoints(
-            com.google.monitoring.v3.Point.newBuilder()
-                .setValue(TypedValue.newBuilder().setInt64Value(point.getValue()))
-                .setInterval(mapInterval(point, metric))
-                .build());
+  public void recordPoint(MetricData metricData, LongPointData pointData) {
+    recordPoint(
+        metricData,
+        pointData,
+        Point.newBuilder()
+            .setValue(TypedValue.newBuilder().setInt64Value(pointData.getValue()))
+            .setInterval(mapInterval(pointData, metricData))
+            .build());
   }
 
   @Override
-  public void recordPoint(MetricData metric, DoublePointData point) {
-    MetricDescriptor descriptor = mapMetricDescriptor(this.prefix, metric, point);
-    if (descriptor == null) {
-      // Unsupported type.
-      return;
-    }
-    descriptors.putIfAbsent(descriptor.getType(), descriptor);
-    MetricWithLabels key = new MetricWithLabels(descriptor.getType(), point.getAttributes());
-    // TODO: Check lastExportTime and ensure we don't send too often...
-    pendingTimeSeries
-        .computeIfAbsent(key, k -> makeTimeSeriesHeader(metric, point.getAttributes(), descriptor))
-        .addPoints(
-            com.google.monitoring.v3.Point.newBuilder()
-                .setValue(TypedValue.newBuilder().setDoubleValue(point.getValue()))
-                .setInterval(mapInterval(point, metric)));
+  public void recordPoint(MetricData metricData, DoublePointData pointData) {
+    recordPoint(
+        metricData,
+        pointData,
+        Point.newBuilder()
+            .setValue(TypedValue.newBuilder().setDoubleValue(pointData.getValue()))
+            .setInterval(mapInterval(pointData, metricData))
+            .build());
   }
 
   @Override
-  public void recordPoint(MetricData metric, HistogramPointData point) {
+  public void recordPoint(MetricData metricData, HistogramPointData pointData) {
+    recordPoint(
+        metricData,
+        pointData,
+        Point.newBuilder()
+            .setValue(
+                TypedValue.newBuilder().setDistributionValue(mapDistribution(pointData, projectId)))
+            .setInterval(mapInterval(pointData, metricData))
+            .build());
+  }
+
+  private void recordPoint(MetricData metric, PointData point, Point builtPoint) {
     MetricDescriptor descriptor = mapMetricDescriptor(this.prefix, metric, point);
     if (descriptor == null) {
       // Unsupported type.
       return;
     }
     descriptors.putIfAbsent(descriptor.getType(), descriptor);
-    MetricWithLabels key = new MetricWithLabels(descriptor.getType(), point.getAttributes());
+    Attributes metricAttributes =
+        attachInstrumentationLibraryLabels(
+            point.getAttributes(), metric.getInstrumentationScopeInfo());
+    MetricWithLabels key = new MetricWithLabels(descriptor.getType(), metricAttributes);
     pendingTimeSeries
-        .computeIfAbsent(key, k -> makeTimeSeriesHeader(metric, point.getAttributes(), descriptor))
-        .addPoints(
-            com.google.monitoring.v3.Point.newBuilder()
-                .setValue(
-                    TypedValue.newBuilder().setDistributionValue(mapDistribution(point, projectId)))
-                .setInterval(mapInterval(point, metric)));
+        .computeIfAbsent(key, k -> makeTimeSeriesHeader(metric, metricAttributes, descriptor))
+        .addPoints(builtPoint);
   }
 
   private TimeSeries.Builder makeTimeSeriesHeader(
@@ -114,6 +117,18 @@ private TimeSeries.Builder makeTimeSeriesHeader(
         .setResource(mapResource(metric.getResource()));
   }
 
+  private Attributes attachInstrumentationLibraryLabels(
+      Attributes attributes, InstrumentationScopeInfo instrumentationScopeInfo) {
+    return attributes.toBuilder()
+        .put(
+            AttributeKey.stringKey(LABEL_INSTRUMENTATION_LIBRARY_NAME),
+            instrumentationScopeInfo.getName())
+        .put(
+            AttributeKey.stringKey(LABEL_INSTRUMENTATION_LIBRARY_VERSION),
+            Objects.requireNonNullElse(instrumentationScopeInfo.getVersion(), ""))
+        .build();
+  }
+
   @Override
   public Collection<MetricDescriptor> getDescriptors() {
     return descriptors.values();

exporters/metrics/src/test/java/com/google/cloud/opentelemetry/metric/GoogleCloudMetricExporterTest.java

@@ -15,6 +15,8 @@
  */
 package com.google.cloud.opentelemetry.metric;
 
+import static com.google.cloud.opentelemetry.metric.AggregateByLabelMetricTimeSeriesBuilder.LABEL_INSTRUMENTATION_LIBRARY_NAME;
+import static com.google.cloud.opentelemetry.metric.AggregateByLabelMetricTimeSeriesBuilder.LABEL_INSTRUMENTATION_LIBRARY_VERSION;
 import static com.google.cloud.opentelemetry.metric.FakeData.aCloudZone;
 import static com.google.cloud.opentelemetry.metric.FakeData.aDoubleSummaryPoint;
 import static com.google.cloud.opentelemetry.metric.FakeData.aFakeCredential;
@@ -190,6 +192,8 @@ public void testExportSucceeds() {
                     .setType(expectedDescriptor.getType())
                     .putLabels("label1", "value1")
                     .putLabels("label2", "false")
+                    .putLabels(LABEL_INSTRUMENTATION_LIBRARY_NAME, "instrumentName")
+                    .putLabels(LABEL_INSTRUMENTATION_LIBRARY_VERSION, "0")
                     .build())
             .addPoints(expectedPoint)
             .setMetricKind(expectedDescriptor.getMetricKind())
@@ -294,6 +298,8 @@ public void testExportWithHistogram_Succeeds() {
                 Metric.newBuilder()
                     .setType(expectedDescriptor.getType())
                     .putLabels("test", "one")
+                    .putLabels("instrumentation_source", "instrumentName")
+                    .putLabels("instrumentation_version", "0")
                     .build())
             .addPoints(expectedPoint)
             .setMetricKind(expectedDescriptor.getMetricKind())