chore: use 2 different exporter instances for internal and external metrics (#2515)

* chore: use 2 different exporter instances for internal and external metrics

This is necessary to unblock upcoming changes that will track additional metrics and a new unified BigtableClient monitored resource.

Change-Id: I3a051f1faccaba1fc17e4151d1ed85ff8e5952f6

* skip export if the application resource is unsupported

Change-Id: I3f3765ed816922ecf30f28ce66c2271158d9f6bf

* tweaks

Change-Id: Ia3f7698fb9e17c015e6d99e8599d31dd93b5efe5

* ensure that exporters use the same task id

Change-Id: I9e4cda1ade196cc4f22191f6284e0ec85642b482

* chore: generate libraries at Thu Mar 13 02:31:04 UTC 2025

* tweaks

Change-Id: I145e75f3e7628decd844e61c36936cabbac7f3eb

* clean up tests

Change-Id: I16d2fb78e10e0a6b8ce8f89617269ed42e0a3c9f

* fix clirr

Change-Id: Ie971e3a06da180639d9d36d4941bbb1fd6bcebf8

* fix clirr

Change-Id: I23352ecfaf13a95fa2cfed165b0dba8ec4c47cfc

* address fedback

Change-Id: I07a36299cb0004dae4601a8fbb8d37ee93486b8f

* deflake unrelated test again

Change-Id: I9fa2b8fa840eae9bf434e72510f04a77b2166c74

* try to deflake test

Change-Id: I98ad5c58e33f35f6165db39963467b17cf7d8745

---------

Co-authored-by: cloud-java-bot <[email protected]>

Author: igorbernstein2

google-cloud-bigtable/clirr-ignored-differences.xml

@@ -314,4 +314,10 @@
         <className>com/google/cloud/bigtable/data/v2/models/TargetId</className>
         <method>*scopedForMaterializedView()</method>
     </difference>
+    <difference>
+        <!-- BigtableCloudMonitoringExporter is InternalApi -->
+        <differenceType>7009</differenceType>
+        <className>com/google/cloud/bigtable/data/v2/stub/metrics/BigtableCloudMonitoringExporter</className>
+        <method>*</method>
+    </difference>
 </differences>

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

@@ -40,9 +40,10 @@
 import com.google.cloud.monitoring.v3.MetricServiceClient;
 import com.google.cloud.monitoring.v3.MetricServiceSettings;
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
 import com.google.common.base.Supplier;
-import com.google.common.base.Suppliers;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Iterables;
 import com.google.common.util.concurrent.MoreExecutors;
@@ -56,6 +57,7 @@
 import io.opentelemetry.sdk.metrics.data.MetricData;
 import io.opentelemetry.sdk.metrics.export.MetricExporter;
 import java.io.IOException;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -94,43 +96,25 @@ public final class BigtableCloudMonitoringExporter implements MetricExporter {
   // https://cloud.google.com/monitoring/quotas#custom_metrics_quotas.
   private static final int EXPORT_BATCH_SIZE_LIMIT = 200;
 
-  private final MetricServiceClient client;
+  private final String exporterName;
 
-  private final String taskId;
+  private final MetricServiceClient client;
 
-  // Application resource is initialized on the first export, which runs on a background thread
-  // to avoid slowness when starting the client.
-  private final Supplier<MonitoredResource> applicationResource;
+  private final TimeSeriesConverter timeSeriesConverter;
 
   private final AtomicBoolean isShutdown = new AtomicBoolean(false);
 
   private CompletableResultCode lastExportCode;
 
-  private final AtomicBoolean bigtableExportFailureLogged = new AtomicBoolean(false);
-  private final AtomicBoolean applicationExportFailureLogged = new AtomicBoolean(false);
-
-  private static final ImmutableList<String> BIGTABLE_TABLE_METRICS =
-      ImmutableSet.of(
-              OPERATION_LATENCIES_NAME,
-              ATTEMPT_LATENCIES_NAME,
-              SERVER_LATENCIES_NAME,
-              FIRST_RESPONSE_LATENCIES_NAME,
-              CLIENT_BLOCKING_LATENCIES_NAME,
-              APPLICATION_BLOCKING_LATENCIES_NAME,
-              RETRY_COUNT_NAME,
-              CONNECTIVITY_ERROR_COUNT_NAME,
-              REMAINING_DEADLINE_NAME)
-          .stream()
-          .map(m -> METER_NAME + m)
-          .collect(ImmutableList.toImmutableList());
-
-  private static final ImmutableList<String> APPLICATION_METRICS =
-      ImmutableSet.of(PER_CONNECTION_ERROR_COUNT_NAME).stream()
-          .map(m -> METER_NAME + m)
-          .collect(ImmutableList.toImmutableList());
-
-  public static BigtableCloudMonitoringExporter create(
-      @Nullable Credentials credentials, @Nullable String endpoint) throws IOException {
+  private final AtomicBoolean exportFailureLogged = new AtomicBoolean(false);
+
+  static BigtableCloudMonitoringExporter create(
+      String exporterName,
+      @Nullable Credentials credentials,
+      @Nullable String endpoint,
+      TimeSeriesConverter converter)
+      throws IOException {
+
     MetricServiceSettings.Builder settingsBuilder = MetricServiceSettings.newBuilder();
     CredentialsProvider credentialsProvider =
         Optional.ofNullable(credentials)
@@ -146,79 +130,64 @@ public static BigtableCloudMonitoringExporter create(
       settingsBuilder.setEndpoint(endpoint);
     }
 
-    java.time.Duration timeout = java.time.Duration.ofMinutes(1);
+    Duration timeout = Duration.ofMinutes(1);
     // TODO: createServiceTimeSeries needs special handling if the request failed. Leaving
     // it as not retried for now.
     settingsBuilder.createServiceTimeSeriesSettings().setSimpleTimeoutNoRetriesDuration(timeout);
 
     return new BigtableCloudMonitoringExporter(
-        MetricServiceClient.create(settingsBuilder.build()),
-        Suppliers.memoize(BigtableExporterUtils::detectResourceSafe),
-        BigtableExporterUtils.getDefaultTaskValue());
+        exporterName, MetricServiceClient.create(settingsBuilder.build()), converter);
   }
 
   @VisibleForTesting
   BigtableCloudMonitoringExporter(
-      MetricServiceClient client, Supplier<MonitoredResource> applicationResource, String taskId) {
+      String exporterName, MetricServiceClient client, TimeSeriesConverter converter) {
+    this.exporterName = exporterName;
     this.client = client;
-    this.taskId = taskId;
-    this.applicationResource = applicationResource;
+    this.timeSeriesConverter = converter;
   }
 
   @Override
-  public CompletableResultCode export(Collection<MetricData> collection) {
-    if (isShutdown.get()) {
-      logger.log(Level.WARNING, "Exporter is shutting down");
-      return CompletableResultCode.ofFailure();
-    }
-
-    CompletableResultCode bigtableExportCode = exportBigtableResourceMetrics(collection);
-    CompletableResultCode applicationExportCode = exportApplicationResourceMetrics(collection);
-
-    lastExportCode =
-        CompletableResultCode.ofAll(ImmutableList.of(applicationExportCode, bigtableExportCode));
+  public CompletableResultCode export(Collection<MetricData> metricData) {
+    Preconditions.checkState(!isShutdown.get(), "Exporter is shutting down");
 
+    lastExportCode = doExport(metricData);
     return lastExportCode;
   }
 
   /** Export metrics associated with a BigtableTable resource. */
-  private CompletableResultCode exportBigtableResourceMetrics(Collection<MetricData> collection) {
-    // Filter bigtable table metrics
-    List<MetricData> bigtableMetricData =
-        collection.stream()
-            .filter(md -> BIGTABLE_TABLE_METRICS.contains(md.getName()))
-            .collect(Collectors.toList());
+  private CompletableResultCode doExport(Collection<MetricData> metricData) {
+    Map<ProjectName, List<TimeSeries>> bigtableTimeSeries;
 
-    // Skips exporting if there's none
-    if (bigtableMetricData.isEmpty()) {
-      return CompletableResultCode.ofSuccess();
-    }
-
-    // List of timeseries by project id
-    Map<String, List<TimeSeries>> bigtableTimeSeries;
     try {
-      bigtableTimeSeries =
-          BigtableExporterUtils.convertToBigtableTimeSeries(bigtableMetricData, taskId);
-    } catch (Throwable e) {
+      bigtableTimeSeries = timeSeriesConverter.convert(metricData);
+    } catch (Throwable t) {
       logger.log(
           Level.WARNING,
-          "Failed to convert bigtable table metric data to cloud monitoring timeseries.",
-          e);
+          String.format(
+              "Failed to convert %s metric data to cloud monitoring timeseries.", exporterName),
+          t);
       return CompletableResultCode.ofFailure();
     }
 
-    CompletableResultCode bigtableExportCode = new CompletableResultCode();
+    // Skips exporting if there's none
+    if (bigtableTimeSeries.isEmpty()) {
+      return CompletableResultCode.ofSuccess();
+    }
+
+    CompletableResultCode exportCode = new CompletableResultCode();
     bigtableTimeSeries.forEach(
-        (projectId, ts) -> {
-          ProjectName projectName = ProjectName.of(projectId);
+        (projectName, ts) -> {
           ApiFuture<List<Empty>> future = exportTimeSeries(projectName, ts);
           ApiFutures.addCallback(
               future,
               new ApiFutureCallback<List<Empty>>() {
                 @Override
                 public void onFailure(Throwable throwable) {
-                  if (bigtableExportFailureLogged.compareAndSet(false, true)) {
-                    String msg = "createServiceTimeSeries request failed for bigtable metrics.";
+                  if (exportFailureLogged.compareAndSet(false, true)) {
+                    String msg =
+                        String.format(
+                            "createServiceTimeSeries request failed for %s.", exporterName);
                     if (throwable instanceof PermissionDeniedException) {
                       msg +=
                           String.format(
@@ -227,100 +196,20 @@ public void onFailure(Throwable throwable) {
                     }
                     logger.log(Level.WARNING, msg, throwable);
                   }
-                  bigtableExportCode.fail();
+                  exportCode.fail();
                 }
 
                 @Override
                 public void onSuccess(List<Empty> emptyList) {
                   // When an export succeeded reset the export failure flag to false so if there's a
                   // transient failure it'll be logged.
-                  bigtableExportFailureLogged.set(false);
-                  bigtableExportCode.succeed();
+                  exportFailureLogged.set(false);
+                  exportCode.succeed();
                 }
               },
               MoreExecutors.directExecutor());
         });
 
-    return bigtableExportCode;
-  }
-
-  /** Export metrics associated with the resource the Application is running on. */
-  private CompletableResultCode exportApplicationResourceMetrics(
-      Collection<MetricData> collection) {
-    if (applicationResource.get() == null) {
-      return CompletableResultCode.ofSuccess();
-    }
-
-    // Filter application level metrics
-    List<MetricData> metricData =
-        collection.stream()
-            .filter(md -> APPLICATION_METRICS.contains(md.getName()))
-            .collect(Collectors.toList());
-
-    // Skip exporting if there's none
-    if (metricData.isEmpty()) {
-      return CompletableResultCode.ofSuccess();
-    }
-
-    List<TimeSeries> timeSeries;
-    try {
-      timeSeries =
-          BigtableExporterUtils.convertToApplicationResourceTimeSeries(
-              metricData, taskId, applicationResource.get());
-    } catch (Throwable e) {
-      logger.log(
-          Level.WARNING,
-          "Failed to convert application metric data to cloud monitoring timeseries.",
-          e);
-      return CompletableResultCode.ofFailure();
-    }
-
-    // Construct the request. The project id will be the project id of the detected monitored
-    // resource.
-    ApiFuture<List<Empty>> gceOrGkeFuture;
-    CompletableResultCode exportCode = new CompletableResultCode();
-    try {
-      ProjectName projectName =
-          ProjectName.of(
-              applicationResource.get().getLabelsOrThrow(APPLICATION_RESOURCE_PROJECT_ID));
-
-      gceOrGkeFuture = exportTimeSeries(projectName, timeSeries);
-
-      ApiFutures.addCallback(
-          gceOrGkeFuture,
-          new ApiFutureCallback<List<Empty>>() {
-            @Override
-            public void onFailure(Throwable throwable) {
-              if (applicationExportFailureLogged.compareAndSet(false, true)) {
-                String msg = "createServiceTimeSeries request failed for bigtable metrics.";
-                if (throwable instanceof PermissionDeniedException) {
-                  msg +=
-                      String.format(
-                          " Need monitoring metric writer permission on project=%s. Follow https://cloud.google.com/bigtable/docs/client-side-metrics-setup to set up permissions.",
-                          projectName.getProject());
-                }
-                logger.log(Level.WARNING, msg, throwable);
-              }
-              exportCode.fail();
-            }
-
-            @Override
-            public void onSuccess(List<Empty> emptyList) {
-              // When an export succeeded reset the export failure flag to false so if there's a
-              // transient failure it'll be logged.
-              applicationExportFailureLogged.set(false);
-              exportCode.succeed();
-            }
-          },
-          MoreExecutors.directExecutor());
-
-    } catch (Exception e) {
-      logger.log(
-          Level.WARNING,
-          "Failed to get projectName for application resource " + applicationResource);
-      return CompletableResultCode.ofFailure();
-    }
-
     return exportCode;
   }
 
@@ -383,4 +272,87 @@ public CompletableResultCode shutdown() {
   public AggregationTemporality getAggregationTemporality(InstrumentType instrumentType) {
     return AggregationTemporality.CUMULATIVE;
   }
+
+  interface TimeSeriesConverter {
+    Map<ProjectName, List<TimeSeries>> convert(Collection<MetricData> metricData);
+  }
+
+  static class PublicTimeSeriesConverter implements TimeSeriesConverter {
+    private static final ImmutableList<String> BIGTABLE_TABLE_METRICS =
+        ImmutableSet.of(
+                OPERATION_LATENCIES_NAME,
+                ATTEMPT_LATENCIES_NAME,
+                SERVER_LATENCIES_NAME,
+                FIRST_RESPONSE_LATENCIES_NAME,
+                CLIENT_BLOCKING_LATENCIES_NAME,
+                APPLICATION_BLOCKING_LATENCIES_NAME,
+                RETRY_COUNT_NAME,
+                CONNECTIVITY_ERROR_COUNT_NAME,
+                REMAINING_DEADLINE_NAME)
+            .stream()
+            .map(m -> METER_NAME + m)
+            .collect(ImmutableList.toImmutableList());
+
+    private final String taskId;
+
+    PublicTimeSeriesConverter() {
+      this(BigtableExporterUtils.DEFAULT_TABLE_VALUE.get());
+    }
+
+    PublicTimeSeriesConverter(String taskId) {
+      this.taskId = taskId;
+    }
+
+    @Override
+    public Map<ProjectName, List<TimeSeries>> convert(Collection<MetricData> metricData) {
+      List<MetricData> relevantData =
+          metricData.stream()
+              .filter(md -> BIGTABLE_TABLE_METRICS.contains(md.getName()))
+              .collect(Collectors.toList());
+      if (relevantData.isEmpty()) {
+        return ImmutableMap.of();
+      }
+      return BigtableExporterUtils.convertToBigtableTimeSeries(relevantData, taskId);
+    }
+  }
+
+  static class InternalTimeSeriesConverter implements TimeSeriesConverter {
+    private static final ImmutableList<String> APPLICATION_METRICS =
+        ImmutableSet.of(PER_CONNECTION_ERROR_COUNT_NAME).stream()
+            .map(m -> METER_NAME + m)
+            .collect(ImmutableList.toImmutableList());
+
+    private final String taskId;
+    private final Supplier<MonitoredResource> monitoredResource;
+
+    InternalTimeSeriesConverter(Supplier<MonitoredResource> monitoredResource) {
+      this(monitoredResource, BigtableExporterUtils.DEFAULT_TABLE_VALUE.get());
+    }
+
+    InternalTimeSeriesConverter(Supplier<MonitoredResource> monitoredResource, String taskId) {
+      this.monitoredResource = monitoredResource;
+      this.taskId = taskId;
+    }
+
+    @Override
+    public Map<ProjectName, List<TimeSeries>> convert(Collection<MetricData> metricData) {
+      MonitoredResource monitoredResource = this.monitoredResource.get();
+      if (monitoredResource == null) {
+        return ImmutableMap.of();
+      }
+
+      List<MetricData> relevantData =
+          metricData.stream()
+              .filter(md -> APPLICATION_METRICS.contains(md.getName()))
+              .collect(Collectors.toList());
+      if (relevantData.isEmpty()) {
+        return ImmutableMap.of();
+      }
+
+      return ImmutableMap.of(
+          ProjectName.of(monitoredResource.getLabelsOrThrow(APPLICATION_RESOURCE_PROJECT_ID)),
+          BigtableExporterUtils.convertToApplicationResourceTimeSeries(
+              relevantData, taskId, monitoredResource));
+    }
+  }
 }