feat: migrate per connection error exporter to otel

google-cloud-bigtable-bom/pom.xml

@@ -95,11 +95,6 @@
           <artifactId>proto-google-cloud-bigtable-v2</artifactId>
           <version>2.34.1-SNAPSHOT</version><!-- {x-version-update:proto-google-cloud-bigtable-v2:current} -->
         </dependency>
-          <dependency>
-              <groupId>com.google.cloud</groupId>
-              <artifactId>google-cloud-bigtable-stats</artifactId>
-              <version>2.34.1-SNAPSHOT</version><!-- {x-version-update:google-cloud-bigtable:current} -->
-          </dependency>
       </dependencies>
     </dependencyManagement>
 

google-cloud-bigtable-deps-bom/pom.xml

@@ -84,6 +84,11 @@
         <type>pom</type>
         <scope>import</scope>
       </dependency>
+      <dependency>
+        <groupId>com.google.cloud.opentelemetry</groupId>
+        <artifactId>detector-resources-support</artifactId>
+        <version>0.27.0</version>
+      </dependency>
       <!-- Other opencensus packages' versions are pulled through com.google.cloud:third-party-dependencies, but has to be manually specified for this one. -->
       <dependency>
         <groupId>io.opencensus</groupId>

google-cloud-bigtable/pom.xml

@@ -62,10 +62,6 @@
   </dependencyManagement>
 
   <dependencies>
-    <dependency>
-      <groupId>com.google.cloud</groupId>
-      <artifactId>google-cloud-bigtable-stats</artifactId>
-    </dependency>
     <!-- NOTE: Dependencies are organized into two groups, production and test.
          Within a group, dependencies are sorted by (groupId, artifactId) -->
     <!-- Production dependencies -->
@@ -342,6 +338,10 @@
       <groupId>io.opentelemetry</groupId>
       <artifactId>opentelemetry-sdk-common</artifactId>
     </dependency>
+    <dependency>
+      <groupId>com.google.cloud.opentelemetry</groupId>
+      <artifactId>detector-resources-support</artifactId>
+    </dependency>
     <dependency>
       <groupId>io.opentelemetry</groupId>
       <artifactId>opentelemetry-sdk-testing</artifactId>

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/metrics/BigtableCloudMonitoringExporter.java

@@ -28,6 +28,7 @@
 import com.google.cloud.monitoring.v3.MetricServiceSettings;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.MoreObjects;
+import com.google.common.collect.ImmutableList;
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.monitoring.v3.CreateTimeSeriesRequest;
 import com.google.monitoring.v3.ProjectName;
@@ -45,6 +46,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.logging.Level;
 import java.util.logging.Logger;
+import java.util.stream.Collectors;
 import javax.annotation.Nullable;
 import org.threeten.bp.Duration;
 
@@ -67,14 +69,16 @@ public final class BigtableCloudMonitoringExporter implements MetricExporter {
           System.getProperty("bigtable.test-monitoring-endpoint"),
           MetricServiceSettings.getDefaultEndpoint());
 
+  private static String GCE_OR_GKE_PROJECT_ID_LABEL = "project_id";
+
   private final MetricServiceClient client;
 
-  private final String projectId;
+  private final String bigtableProjectId;
   private final String taskId;
-  private final MonitoredResource monitoredResource;
-  private final AtomicBoolean isShutdown = new AtomicBoolean(false);
 
-  private static final String RESOURCE_TYPE = "bigtable_client_raw";
+  private final MonitoredResource gceOrGkeResource;
+
+  private final AtomicBoolean isShutdown = new AtomicBoolean(false);
 
   private CompletableResultCode lastExportCode;
 
@@ -94,23 +98,28 @@ public static BigtableCloudMonitoringExporter create(
     // TODO: createServiceTimeSeries needs special handling if the request failed. Leaving
     // it as not retried for now.
     settingsBuilder.createServiceTimeSeriesSettings().setSimpleTimeoutNoRetries(timeout);
+
+    MonitoredResource gceOrGkeResource = BigtableExporterUtils.detectResource();
+
+    logger.log(Level.INFO, "Detected resource: " + gceOrGkeResource);
+
     return new BigtableCloudMonitoringExporter(
         projectId,
         MetricServiceClient.create(settingsBuilder.build()),
-        MonitoredResource.newBuilder().setType(RESOURCE_TYPE).build(),
+        gceOrGkeResource,
         BigtableExporterUtils.getDefaultTaskValue());
   }
 
   @VisibleForTesting
   BigtableCloudMonitoringExporter(
       String projectId,
       MetricServiceClient client,
-      MonitoredResource monitoredResource,
+      @Nullable MonitoredResource gceOrGkeResource,
       String taskId) {
     this.client = client;
-    this.monitoredResource = monitoredResource;
     this.taskId = taskId;
-    this.projectId = projectId;
+    this.gceOrGkeResource = gceOrGkeResource;
+    this.bigtableProjectId = projectId;
   }
 
   @Override
@@ -121,48 +130,64 @@ public CompletableResultCode export(Collection<MetricData> collection) {
     }
     if (!collection.stream()
         .flatMap(metricData -> metricData.getData().getPoints().stream())
-        .allMatch(pd -> projectId.equals(BigtableExporterUtils.getProjectId(pd)))) {
+        .allMatch(pd -> bigtableProjectId.equals(BigtableExporterUtils.getProjectId(pd)))) {
       logger.log(Level.WARNING, "Metric data has different a projectId. Skip exporting.");
       return CompletableResultCode.ofFailure();
     }
 
-    List<TimeSeries> allTimeSeries;
+    List<TimeSeries> bigtableTimeSeries;
     try {
-      allTimeSeries =
-          BigtableExporterUtils.convertCollectionToListOfTimeSeries(
-              collection, taskId, monitoredResource);
+      bigtableTimeSeries =
+          BigtableExporterUtils.convertToBigtableTimeSeries(
+              collection.stream()
+                  .filter(
+                      md ->
+                          !md.getName()
+                              .contains(BuiltinMetricsConstants.PER_CONNECTION_ERROR_COUNT_NAME))
+                  .collect(Collectors.toList()),
+              taskId);
     } catch (Throwable e) {
-      logger.log(Level.WARNING, "Failed to convert metric data to cloud monitoring timeseries.", e);
+      logger.log(
+          Level.WARNING,
+          "Failed to convert bigtable metric data to cloud monitoring timeseries.",
+          e);
       return CompletableResultCode.ofFailure();
     }
 
-    ProjectName projectName = ProjectName.of(projectId);
-    CreateTimeSeriesRequest request =
+    ProjectName projectName = ProjectName.of(bigtableProjectId);
+    CreateTimeSeriesRequest bigtableRequest =
         CreateTimeSeriesRequest.newBuilder()
             .setName(projectName.toString())
-            .addAllTimeSeries(allTimeSeries)
+            .addAllTimeSeries(bigtableTimeSeries)
             .build();
 
-    ApiFuture<Empty> future = this.client.createServiceTimeSeriesCallable().futureCall(request);
-
-    lastExportCode = new CompletableResultCode();
+    ApiFuture<Empty> future =
+        this.client.createServiceTimeSeriesCallable().futureCall(bigtableRequest);
 
+    CompletableResultCode bigtableExportCode = new CompletableResultCode();
     ApiFutures.addCallback(
         future,
         new ApiFutureCallback<Empty>() {
           @Override
           public void onFailure(Throwable throwable) {
-            logger.log(Level.WARNING, "createServiceTimeSeries request failed. ", throwable);
-            lastExportCode.fail();
+            logger.log(
+                Level.WARNING,
+                "createServiceTimeSeries request failed for bigtable metrics. ",
+                throwable);
+            bigtableExportCode.fail();
           }
 
           @Override
           public void onSuccess(Empty empty) {
-            lastExportCode.succeed();
+            bigtableExportCode.succeed();
           }
         },
         MoreExecutors.directExecutor());
 
+    CompletableResultCode gceOrGkeExportCode = exportGceOrGkeMetrics(collection);
+
+    lastExportCode =
+        CompletableResultCode.ofAll(ImmutableList.of(gceOrGkeExportCode, bigtableExportCode));
     return lastExportCode;
   }
 
@@ -208,4 +233,62 @@ public CompletableResultCode shutdown() {
   public AggregationTemporality getAggregationTemporality(InstrumentType instrumentType) {
     return AggregationTemporality.CUMULATIVE;
   }
+
+  private CompletableResultCode exportGceOrGkeMetrics(Collection<MetricData> collection) {
+    CompletableResultCode gceOrGkeExportCode = new CompletableResultCode();
+    if (gceOrGkeResource == null) {
+      return gceOrGkeExportCode.succeed();
+    }
+    List<TimeSeries> gceOrGceTimeSeries;
+    try {
+      gceOrGceTimeSeries =
+          BigtableExporterUtils.convertToGceOrGkeTimeSeries(
+              collection.stream()
+                  .filter(
+                      md ->
+                          md.getName()
+                              .contains(BuiltinMetricsConstants.PER_CONNECTION_ERROR_COUNT_NAME))
+                  .collect(Collectors.toList()),
+              taskId,
+              gceOrGkeResource);
+    } catch (Throwable e) {
+      logger.log(
+          Level.WARNING,
+          "Failed to convert per connection error count data to cloud monitoring timeseries.",
+          e);
+      return CompletableResultCode.ofFailure();
+    }
+
+    ProjectName gceOrGkeProjectName =
+        ProjectName.of(gceOrGkeResource.getLabelsOrThrow(GCE_OR_GKE_PROJECT_ID_LABEL));
+    CreateTimeSeriesRequest gceOrGkeRequest =
+        CreateTimeSeriesRequest.newBuilder()
+            .setName(gceOrGkeProjectName.toString())
+            .addAllTimeSeries(gceOrGceTimeSeries)
+            .build();
+
+    ApiFuture<Empty> gceOrGkeFuture =
+        this.client.createServiceTimeSeriesCallable().futureCall(gceOrGkeRequest);
+
+    ApiFutures.addCallback(
+        gceOrGkeFuture,
+        new ApiFutureCallback<Empty>() {
+          @Override
+          public void onFailure(Throwable throwable) {
+            logger.log(
+                Level.WARNING,
+                "createServiceTimeSeries request failed for per connection error metrics.",
+                throwable);
+            gceOrGkeExportCode.fail();
+          }
+
+          @Override
+          public void onSuccess(Empty empty) {
+            gceOrGkeExportCode.succeed();
+          }
+        },
+        MoreExecutors.directExecutor());
+
+    return gceOrGkeExportCode;
+  }
 }