address comments and add a test

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/BigtableDataClientFactory.java

@@ -88,7 +88,6 @@ public static BigtableDataClientFactory create(BigtableDataSettings defaultSetti
       // the OTEL instance and log the exception instead.
       openTelemetry =
           EnhancedBigtableStub.getOpenTelemetry(
-              defaultSettings.getProjectId(),
               defaultSettings.getMetricsProvider(),
               sharedClientContext.getCredentials(),
               defaultSettings.getStubSettings().getMetricsEndpoint());

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

@@ -221,7 +221,6 @@ public static EnhancedBigtableStub create(EnhancedBigtableStubSettings settings)
       // the OTEL instance and log the exception instead.
       openTelemetry =
           getOpenTelemetry(
-              settings.getProjectId(),
               settings.getMetricsProvider(),
               clientContext.getCredentials(),
               settings.getMetricsEndpoint());
@@ -270,10 +269,7 @@ public static ClientContext createClientContext(EnhancedBigtableStubSettings set
       // the OTEL instance and log the exception instead.
       openTelemetry =
           getOpenTelemetry(
-              settings.getProjectId(),
-              settings.getMetricsProvider(),
-              credentials,
-              settings.getMetricsEndpoint());
+              settings.getMetricsProvider(), credentials, settings.getMetricsEndpoint());
     } catch (Throwable t) {
       logger.log(Level.WARNING, "Failed to get OTEL, will skip exporting client side metrics", t);
     }
@@ -381,9 +377,23 @@ public static ApiTracerFactory createBigtableTracerFactory(
     return new CompositeTracerFactory(tracerFactories.build());
   }
 
+  /**
+   * getOpenTelemetry is called from the following places:
+   *
+   * <ul>
+   *   <li>EnhancedBigtableStub.createClientContext: gets the openTelemetry to create meter for
+   *       per_connection_error_count
+   *   <li>EnhancedBigtableStub.create(settings): gets the openTelemetry to pass into
+   *       BigtableTracerFactory to create meters for built in metrics
+   *   <li>BigtableDataClientFactory.create(defaultSettings): gets the openTelemetry to pass around
+   *       when creating BigtableTracerFactory
+   * </ul>
+   *
+   * <p>DefaultMetricsProvider will always return the same openTelemetry instance for the same
+   * credentials and metrics endpoint.
+   */
   @Nullable
   public static OpenTelemetry getOpenTelemetry(
-      String projectId,
       MetricsProvider metricsProvider,
       @Nullable Credentials defaultCredentials,
       @Nullable String metricsEndpoint)
@@ -398,7 +408,7 @@ public static OpenTelemetry getOpenTelemetry(
               ? BigtableDataSettings.getMetricsCredentials()
               : defaultCredentials;
       DefaultMetricsProvider defaultMetricsProvider = (DefaultMetricsProvider) metricsProvider;
-      return defaultMetricsProvider.getOpenTelemetry(projectId, metricsEndpoint, credentials);
+      return defaultMetricsProvider.getOpenTelemetry(metricsEndpoint, credentials);
     } else if (metricsProvider instanceof NoopMetricsProvider) {
       return null;
     }

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

@@ -203,6 +203,7 @@ private CompletableResultCode exportBigtableResourceMetrics(Collection<MetricDat
       return CompletableResultCode.ofSuccess();
     }
 
+    // List of timeseries by project id
     Map<String, List<TimeSeries>> bigtableTimeSeries;
     try {
       bigtableTimeSeries =

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

@@ -113,6 +113,7 @@ static String getProjectId(PointData pointData) {
 
   static Map<String, List<TimeSeries>> convertToBigtableTimeSeries(
       List<MetricData> collection, String taskId) {
+    // List of timeseries by project id
     Map<String, List<TimeSeries>> allTimeSeries = new HashMap<>();
 
     for (MetricData metricData : collection) {

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

@@ -38,33 +38,64 @@ private BuiltinMetricsView() {}
   /**
    * Register built-in metrics on the {@link SdkMeterProviderBuilder} with application default
    * credentials and default endpoint.
+   *
+   * @deprecated projectId is no longer used. Call {@link
+   *     #registerBuiltinMetrics(SdkMeterProviderBuilder)} instead.
    */
+  @Deprecated
   public static void registerBuiltinMetrics(String projectId, SdkMeterProviderBuilder builder)
       throws IOException {
     BuiltinMetricsView.registerBuiltinMetrics(
-        projectId, GoogleCredentials.getApplicationDefault(), builder);
+        GoogleCredentials.getApplicationDefault(), builder, null);
+  }
+
+  /**
+   * Register built-in metrics on the {@link SdkMeterProviderBuilder} with application default
+   * credentials and default endpoint.
+   */
+  public static void registerBuiltinMetrics(SdkMeterProviderBuilder builder) throws IOException {
+    BuiltinMetricsView.registerBuiltinMetrics(
+        GoogleCredentials.getApplicationDefault(), builder, null);
   }
 
   /**
    * Register built-in metrics on the {@link SdkMeterProviderBuilder} with custom credentials and
    * default endpoint.
+   *
+   * @deprecated projectId is no longer used. Call {@link #registerBuiltinMetrics(Credentials,
+   *     SdkMeterProviderBuilder, String)} instead.
    */
+  @Deprecated
   public static void registerBuiltinMetrics(
       String projectId, @Nullable Credentials credentials, SdkMeterProviderBuilder builder)
       throws IOException {
-    BuiltinMetricsView.registerBuiltinMetrics(projectId, credentials, builder, null);
+    BuiltinMetricsView.registerBuiltinMetrics(credentials, builder, null);
   }
 
   /**
    * Register built-in metrics on the {@link SdkMeterProviderBuilder} with custom credentials and
    * endpoint.
+   *
+   * @deprecated projectId is no longer used. Call {@link #registerBuiltinMetrics(Credentials,
+   *     SdkMeterProviderBuilder, String)} instead.
    */
+  @Deprecated
   public static void registerBuiltinMetrics(
       String projectId,
       @Nullable Credentials credentials,
       SdkMeterProviderBuilder builder,
       @Nullable String endpoint)
       throws IOException {
+    registerBuiltinMetrics(credentials, builder, endpoint);
+  }
+
+  /**
+   * Register built-in metrics on the {@link SdkMeterProviderBuilder} with custom credentials and
+   * endpoint.
+   */
+  public static void registerBuiltinMetrics(
+      @Nullable Credentials credentials, SdkMeterProviderBuilder builder, @Nullable String endpoint)
+      throws IOException {
     MetricExporter metricExporter = BigtableCloudMonitoringExporter.create(credentials, endpoint);
     for (Map.Entry<InstrumentSelector, View> entry :
         BuiltinMetricsConstants.getAllViews().entrySet()) {

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

@@ -27,7 +27,7 @@
  * SdkMeterProviderBuilder sdkMeterProvider = SdkMeterProvider.builder();
  *
  * // register Builtin metrics on your meter provider with default credentials
- * BuiltinMetricsView.registerBuiltinMetrics("project-id", sdkMeterProvider);
+ * BuiltinMetricsView.registerBuiltinMetrics(sdkMeterProvider);
  *
  * // register other metrics reader and views
  * sdkMeterProvider.registerMetricReader(..);

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

@@ -17,12 +17,14 @@
 
 import com.google.api.core.InternalApi;
 import com.google.auth.Credentials;
+import com.google.auto.value.AutoValue;
 import com.google.common.base.MoreObjects;
 import io.opentelemetry.api.OpenTelemetry;
 import io.opentelemetry.sdk.OpenTelemetrySdk;
 import io.opentelemetry.sdk.metrics.SdkMeterProvider;
 import io.opentelemetry.sdk.metrics.SdkMeterProviderBuilder;
 import java.io.IOException;
+import java.util.concurrent.ConcurrentHashMap;
 import javax.annotation.Nullable;
 
 /**
@@ -36,30 +38,44 @@ public final class DefaultMetricsProvider implements MetricsProvider {
 
   public static DefaultMetricsProvider INSTANCE = new DefaultMetricsProvider();
 
-  private OpenTelemetry openTelemetry;
-  private String projectId;
+  private final ConcurrentHashMap<Key, OpenTelemetry> otels = new ConcurrentHashMap<>();
 
   private DefaultMetricsProvider() {}
 
   @InternalApi
   public OpenTelemetry getOpenTelemetry(
-      String projectId, String metricsEndpoint, @Nullable Credentials credentials)
-      throws IOException {
-    this.projectId = projectId;
-    if (openTelemetry == null) {
+      @Nullable String metricsEndpoint, @Nullable Credentials credentials) throws IOException {
+    Key key = Key.create(metricsEndpoint, credentials);
+    OpenTelemetry otel = otels.get(key);
+    if (otel == null) {
       SdkMeterProviderBuilder meterProvider = SdkMeterProvider.builder();
-      BuiltinMetricsView.registerBuiltinMetrics(
-          projectId, credentials, meterProvider, metricsEndpoint);
-      openTelemetry = OpenTelemetrySdk.builder().setMeterProvider(meterProvider.build()).build();
+      BuiltinMetricsView.registerBuiltinMetrics(credentials, meterProvider, metricsEndpoint);
+      otel = OpenTelemetrySdk.builder().setMeterProvider(meterProvider.build()).build();
+      otels.put(key, otel);
     }
-    return openTelemetry;
+    return otel;
   }
 
   @Override
   public String toString() {
-    return MoreObjects.toStringHelper(this)
-        .add("projectId", projectId)
-        .add("openTelemetry", openTelemetry)
-        .toString();
+    // don't log credentials
+    MoreObjects.ToStringHelper toStringHelper = MoreObjects.toStringHelper(this);
+    otels.forEach(
+        (k, v) ->
+            toStringHelper.add(k.metricsEndpoint() == null ? "null" : k.metricsEndpoint(), v));
+    return toStringHelper.toString();
+  }
+
+  @AutoValue
+  abstract static class Key {
+    @Nullable
+    abstract String metricsEndpoint();
+
+    @Nullable
+    abstract Credentials credentials();
+
+    static Key create(@Nullable String metricsEndpoint, @Nullable Credentials credentials) {
+      return new AutoValue_DefaultMetricsProvider_Key(metricsEndpoint, credentials);
+    }
   }
 }

google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/stub/metrics/BigtableCloudMonitoringExporterTest.java

@@ -24,7 +24,10 @@
 import static com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsConstants.TABLE_ID_KEY;
 import static com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsConstants.ZONE_ID_KEY;
 import static com.google.common.truth.Truth.assertThat;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import com.google.api.Distribution;
@@ -35,6 +38,7 @@
 import com.google.cloud.monitoring.v3.MetricServiceClient;
 import com.google.cloud.monitoring.v3.stub.MetricServiceStub;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
 import com.google.monitoring.v3.CreateTimeSeriesRequest;
 import com.google.monitoring.v3.TimeSeries;
 import com.google.protobuf.Empty;
@@ -53,6 +57,8 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.List;
+import java.util.Map;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Rule;
@@ -376,6 +382,114 @@ public void testTimeSeriesForMetricWithGceOrGkeResource() {
             taskId);
   }
 
+  @Test
+  public void testExportingToMultipleProjects() {
+    ArgumentCaptor<CreateTimeSeriesRequest> argumentCaptor =
+        ArgumentCaptor.forClass(CreateTimeSeriesRequest.class);
+
+    UnaryCallable<CreateTimeSeriesRequest, Empty> mockCallable = mock(UnaryCallable.class);
+    when(mockMetricServiceStub.createServiceTimeSeriesCallable()).thenReturn(mockCallable);
+    ApiFuture<Empty> future = ApiFutures.immediateFuture(Empty.getDefaultInstance());
+    when(mockCallable.futureCall(any())).thenReturn(future);
+
+    long startEpoch = 10;
+    long endEpoch = 15;
+    HistogramPointData histogramPointData1 =
+        ImmutableHistogramPointData.create(
+            startEpoch,
+            endEpoch,
+            attributes,
+            3d,
+            true,
+            1d, // min
+            true,
+            2d, // max
+            Arrays.asList(1.0),
+            Arrays.asList(1L, 2L));
+
+    MetricData histogramData1 =
+        ImmutableMetricData.createDoubleHistogram(
+            resource,
+            scope,
+            "bigtable.googleapis.com/internal/client/operation_latencies",
+            "description",
+            "ms",
+            ImmutableHistogramData.create(
+                AggregationTemporality.CUMULATIVE, ImmutableList.of(histogramPointData1)));
+
+    HistogramPointData histogramPointData2 =
+        ImmutableHistogramPointData.create(
+            startEpoch,
+            endEpoch,
+            attributes.toBuilder().put(BIGTABLE_PROJECT_ID_KEY, "another-project").build(),
+            50d,
+            true,
+            5d, // min
+            true,
+            30d, // max
+            Arrays.asList(1.0),
+            Arrays.asList(5L, 10L));
+
+    MetricData histogramData2 =
+        ImmutableMetricData.createDoubleHistogram(
+            resource,
+            scope,
+            "bigtable.googleapis.com/internal/client/operation_latencies",
+            "description",
+            "ms",
+            ImmutableHistogramData.create(
+                AggregationTemporality.CUMULATIVE, ImmutableList.of(histogramPointData2)));
+
+    exporter.export(Arrays.asList(histogramData1, histogramData2));
+
+    verify(mockCallable, times(2)).futureCall(argumentCaptor.capture());
+
+    List<CreateTimeSeriesRequest> allValues = argumentCaptor.getAllValues();
+
+    assertThat(allValues).hasSize(2);
+
+    List<Map<String, String>> labelsMap = new ArrayList<>();
+    List<Long> counts = new ArrayList<>();
+    allValues.forEach(
+        value -> {
+          labelsMap.add(value.getTimeSeriesList().get(0).getResource().getLabelsMap());
+          counts.add(
+              value
+                  .getTimeSeriesList()
+                  .get(0)
+                  .getPoints(0)
+                  .getValue()
+                  .getDistributionValue()
+                  .getCount());
+        });
+
+    assertThat(labelsMap)
+        .containsExactly(
+            ImmutableMap.of(
+                BIGTABLE_PROJECT_ID_KEY.getKey(),
+                projectId,
+                INSTANCE_ID_KEY.getKey(),
+                instanceId,
+                TABLE_ID_KEY.getKey(),
+                tableId,
+                CLUSTER_ID_KEY.getKey(),
+                cluster,
+                ZONE_ID_KEY.getKey(),
+                zone),
+            ImmutableMap.of(
+                BIGTABLE_PROJECT_ID_KEY.getKey(),
+                "another-project",
+                INSTANCE_ID_KEY.getKey(),
+                instanceId,
+                TABLE_ID_KEY.getKey(),
+                tableId,
+                CLUSTER_ID_KEY.getKey(),
+                cluster,
+                ZONE_ID_KEY.getKey(),
+                zone));
+    assertThat(counts).containsExactly(3l, 15l);
+  }
+
   private static class FakeMetricServiceClient extends MetricServiceClient {
 
     protected FakeMetricServiceClient(MetricServiceStub stub) {