fix(spanner): derive built-in metrics project from database client (#13262)

Fixes #13240

### Problem

The built-in-metrics Cloud Monitoring exporter can initialize before any
DatabaseClient exists, so the OpenTelemetry Resource may carry the
host/GKE project instead of the Spanner project — metrics get attributed
to the wrong project. A single resource-level project also misattributes
metrics for multi-project clients (RPCs to databases in other projects).

 ### Solution

Route each metric by its point-level project_id instead of the resource
project:

1. HeaderInterceptor adds project_id to its existing cached built-in
attribute map (cache-miss only, only when parsed from the resource
prefix and not the undefined-project sentinel). No new parsing or
per-RPC work.
2. The exporter ignores the resource-level project_id, derives each
TimeSeries' project from the point, groups by
project, and calls createServiceTimeSeries once per project (still
batched at the 200 quota limit).
3. Fallback for points with no usable project is meter-scoped:
grpc-java/grpc-gcp core metrics use a fallback project (first
DatabaseClient/BatchClient project, set in initializeBuiltInMetrics);
Spanner/GAX DB-scoped metrics are skipped rather than exported under an
unrelated project. No fallback yet → skip.

#### Notes / trade-offs

- gRPC/core metrics for multi-project clients are attributed to the
first client's project (they carry no point project).
- gRPC/core metrics before the first data/batch client, or in admin-only
processes, are dropped.
- Per-project export-failure logging (once per project, with
PermissionDeniedException setup guidance + project id; cleared on a
successful export).

Author: rahul2393

java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/BuiltInMetricsConstant.java

@@ -144,6 +144,7 @@ public class BuiltInMetricsConstant {
       ImmutableList.of(EEF_FALLBACK_COUNT_NAME, EEF_CALL_STATUS_NAME);
 
   public static final String SPANNER_RESOURCE_TYPE = "spanner_instance_client";
+  public static final String UNDEFINED_PROJECT_ID = "undefined-project";
 
   public static final AttributeKey<String> PROJECT_ID_KEY = AttributeKey.stringKey("project_id");
   public static final AttributeKey<String> INSTANCE_ID_KEY = AttributeKey.stringKey("instance_id");

java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/BuiltInMetricsProvider.java

@@ -32,6 +32,7 @@
 import com.google.cloud.opentelemetry.detection.AttributeKeys;
 import com.google.cloud.opentelemetry.detection.DetectedPlatform;
 import com.google.cloud.opentelemetry.detection.GCPPlatformDetector;
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Strings;
 import com.google.common.hash.HashFunction;
 import com.google.common.hash.Hashing;
@@ -75,10 +76,13 @@ final class BuiltInMetricsProvider {
   private static final String default_location = "global";
 
   private OpenTelemetry openTelemetry;
+  private String projectId;
+  private boolean mismatchedProjectIdLogged;
+  private Thread shutdownHook;
 
   private BuiltInMetricsProvider() {}
 
-  OpenTelemetry getOrCreateOpenTelemetry(
+  synchronized OpenTelemetry getOrCreateOpenTelemetry(
       String projectId,
       @Nullable Credentials credentials,
       @Nullable String monitoringHost,
@@ -88,12 +92,13 @@ OpenTelemetry getOrCreateOpenTelemetry(
         SdkMeterProviderBuilder sdkMeterProviderBuilder = SdkMeterProvider.builder();
         BuiltInMetricsView.registerBuiltinMetrics(
             SpannerCloudMonitoringExporter.create(
-                projectId, credentials, monitoringHost, universeDomain),
+                this::getProjectId, credentials, monitoringHost, universeDomain),
             sdkMeterProviderBuilder);
         sdkMeterProviderBuilder.setResource(Resource.create(createResourceAttributes(projectId)));
         SdkMeterProvider sdkMeterProvider = sdkMeterProviderBuilder.build();
         this.openTelemetry = OpenTelemetrySdk.builder().setMeterProvider(sdkMeterProvider).build();
-        Runtime.getRuntime().addShutdownHook(new Thread(sdkMeterProvider::close));
+        this.shutdownHook = new Thread(sdkMeterProvider::close);
+        Runtime.getRuntime().addShutdownHook(this.shutdownHook);
       }
       return this.openTelemetry;
     } catch (IOException ex) {
@@ -106,6 +111,47 @@ OpenTelemetry getOrCreateOpenTelemetry(
     }
   }
 
+  synchronized void setProjectIdIfAbsent(String projectId) {
+    if (this.projectId == null) {
+      this.projectId = projectId;
+    } else if (!this.projectId.equals(projectId) && !mismatchedProjectIdLogged) {
+      mismatchedProjectIdLogged = true;
+      logger.log(
+          Level.WARNING,
+          "Built-in metrics fallback project is already initialized to project {0}. Non-Spanner"
+              + " metrics without project information will be exported using that project instead"
+              + " of project {1}.",
+          new Object[] {this.projectId, projectId});
+    }
+  }
+
+  @Nullable
+  synchronized OpenTelemetry getOpenTelemetry() {
+    return this.openTelemetry;
+  }
+
+  synchronized String getProjectId() {
+    return this.projectId;
+  }
+
+  @VisibleForTesting
+  synchronized void reset() {
+    if (this.openTelemetry instanceof OpenTelemetrySdk) {
+      ((OpenTelemetrySdk) this.openTelemetry).getSdkMeterProvider().close();
+    }
+    if (this.shutdownHook != null) {
+      try {
+        Runtime.getRuntime().removeShutdownHook(this.shutdownHook);
+      } catch (IllegalStateException ignored) {
+        // The JVM is already shutting down.
+      }
+    }
+    this.openTelemetry = null;
+    this.projectId = null;
+    this.mismatchedProjectIdLogged = false;
+    this.shutdownHook = null;
+  }
+
   // TODO: Remove when
   // https://github.com/GoogleCloudPlatform/opentelemetry-operations-java/issues/421
   //       has been fixed.

java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/SpannerCloudMonitoringExporter.java

@@ -42,13 +42,15 @@
 import io.opentelemetry.sdk.metrics.data.AggregationTemporality;
 import io.opentelemetry.sdk.metrics.data.MetricData;
 import io.opentelemetry.sdk.metrics.export.MetricExporter;
-import io.opentelemetry.sdk.resources.Resource;
 import java.io.IOException;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Supplier;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 import java.util.stream.Collectors;
@@ -69,13 +71,12 @@ class SpannerCloudMonitoringExporter implements MetricExporter {
   // This the quota limit from Cloud Monitoring. More details in
   // https://cloud.google.com/monitoring/quotas#custom_metrics_quotas.
   private static final int EXPORT_BATCH_SIZE_LIMIT = 200;
-  private final AtomicBoolean spannerExportFailureLogged = new AtomicBoolean(false);
-  private final AtomicBoolean lastExportSkippedData = new AtomicBoolean(false);
+  private final Set<String> spannerExportFailureLoggedProjects = ConcurrentHashMap.newKeySet();
   private final MetricServiceClient client;
-  private final String spannerProjectId;
+  private final Supplier<String> fallbackProjectIdSupplier;
 
   static SpannerCloudMonitoringExporter create(
-      String projectId,
+      Supplier<String> fallbackProjectIdSupplier,
       @Nullable Credentials credentials,
       @Nullable String monitoringHost,
       String universeDomain)
@@ -114,13 +115,19 @@ static SpannerCloudMonitoringExporter create(
     settingsBuilder.createServiceTimeSeriesSettings().setSimpleTimeoutNoRetriesDuration(timeout);
 
     return new SpannerCloudMonitoringExporter(
-        projectId, MetricServiceClient.create(settingsBuilder.build()));
+        fallbackProjectIdSupplier, MetricServiceClient.create(settingsBuilder.build()));
   }
 
   @VisibleForTesting
-  SpannerCloudMonitoringExporter(String projectId, MetricServiceClient client) {
+  SpannerCloudMonitoringExporter(MetricServiceClient client) {
+    this(() -> null, client);
+  }
+
+  @VisibleForTesting
+  SpannerCloudMonitoringExporter(
+      Supplier<String> fallbackProjectIdSupplier, MetricServiceClient client) {
     this.client = client;
-    this.spannerProjectId = projectId;
+    this.fallbackProjectIdSupplier = fallbackProjectIdSupplier;
   }
 
   @Override
@@ -140,37 +147,16 @@ MetricServiceClient getMetricServiceClient() {
 
   /** Export client built in metrics */
   private CompletableResultCode exportSpannerClientMetrics(Collection<MetricData> collection) {
-    // Filter spanner metrics. Only include metrics that contain a valid project.
-    List<MetricData> spannerMetricData = collection.stream().collect(Collectors.toList());
-
-    // Log warnings for metrics that will be skipped.
-    boolean mustFilter = false;
-    if (spannerMetricData.stream()
-        .map(metricData -> metricData.getResource())
-        .anyMatch(this::shouldSkipPointDataDueToProjectId)) {
-      logger.log(
-          Level.WARNING, "Some metric data contain a different projectId. These will be skipped.");
-      mustFilter = true;
-    }
-
-    if (mustFilter) {
-      spannerMetricData =
-          spannerMetricData.stream()
-              .filter(this::shouldSkipMetricData)
-              .collect(Collectors.toList());
-    }
-    lastExportSkippedData.set(mustFilter);
-
     // Skips exporting if there's none
-    if (spannerMetricData.isEmpty()) {
+    if (collection.isEmpty()) {
       return CompletableResultCode.ofSuccess();
     }
 
     List<TimeSeries> spannerTimeSeries;
     try {
       spannerTimeSeries =
           SpannerCloudMonitoringExporterUtils.convertToSpannerTimeSeries(
-              spannerMetricData, this.spannerProjectId);
+              collection, fallbackProjectIdSupplier.get());
     } catch (Throwable e) {
       logger.log(
           Level.WARNING,
@@ -179,37 +165,60 @@ private CompletableResultCode exportSpannerClientMetrics(Collection<MetricData>
       return CompletableResultCode.ofFailure();
     }
 
-    ProjectName projectName = ProjectName.of(spannerProjectId);
+    if (spannerTimeSeries.isEmpty()) {
+      return CompletableResultCode.ofSuccess();
+    }
+
+    Map<String, List<TimeSeries>> timeSeriesByProject =
+        spannerTimeSeries.stream()
+            .collect(
+                Collectors.groupingBy(
+                    timeSeries ->
+                        timeSeries
+                            .getResource()
+                            .getLabelsMap()
+                            .get(BuiltInMetricsConstant.PROJECT_ID_KEY.getKey())));
+
+    List<ApiFuture<List<Empty>>> futures = new ArrayList<>();
+    for (Map.Entry<String, List<TimeSeries>> entry : timeSeriesByProject.entrySet()) {
+      ProjectName projectName = ProjectName.of(entry.getKey());
+      ApiFuture<List<Empty>> future = exportTimeSeriesInBatch(projectName, entry.getValue());
+      ApiFutures.addCallback(
+          future,
+          new ApiFutureCallback<List<Empty>>() {
+            @Override
+            public void onFailure(Throwable throwable) {
+              logExportFailure(throwable, projectName);
+            }
+
+            @Override
+            public void onSuccess(List<Empty> ignored) {
+              spannerExportFailureLoggedProjects.remove(projectName.getProject());
+            }
+          },
+          MoreExecutors.directExecutor());
+      futures.add(future);
+    }
 
-    ApiFuture<List<Empty>> futureList = exportTimeSeriesInBatch(projectName, spannerTimeSeries);
+    ApiFuture<List<List<Empty>>> groupedFuture = ApiFutures.allAsList(futures);
+    ApiFuture<List<Empty>> futureList =
+        ApiFutures.transform(
+            groupedFuture,
+            groupedResults ->
+                groupedResults.stream().flatMap(List::stream).collect(Collectors.toList()),
+            MoreExecutors.directExecutor());
 
     CompletableResultCode spannerExportCode = new CompletableResultCode();
     ApiFutures.addCallback(
         futureList,
         new ApiFutureCallback<List<Empty>>() {
           @Override
           public void onFailure(Throwable throwable) {
-            if (spannerExportFailureLogged.compareAndSet(false, true)) {
-              String msg = "createServiceTimeSeries request failed for spanner metrics.";
-              if (throwable instanceof PermissionDeniedException) {
-                // TODO: Add the link of public documentation when available in the log message.
-                msg +=
-                    String.format(
-                        " Need monitoring metric writer permission on project=%s. Follow"
-                            + " https://cloud.google.com/spanner/docs/view-manage-client-side-metrics#access-client-side-metrics"
-                            + " to set up permissions",
-                        projectName.getProject());
-              }
-              logger.log(Level.WARNING, msg, throwable);
-            }
             spannerExportCode.fail();
           }
 
           @Override
           public void onSuccess(List<Empty> empty) {
-            // When an export succeeded reset the export failure flag to false so if there's a
-            // transient failure it'll be logged.
-            spannerExportFailureLogged.set(false);
             spannerExportCode.succeed();
           }
         },
@@ -218,16 +227,22 @@ public void onSuccess(List<Empty> empty) {
     return spannerExportCode;
   }
 
-  private boolean shouldSkipMetricData(MetricData metricData) {
-    return shouldSkipPointDataDueToProjectId(metricData.getResource());
-  }
-
-  private boolean shouldSkipPointDataDueToProjectId(Resource resource) {
-    return !spannerProjectId.equals(SpannerCloudMonitoringExporterUtils.getProjectId(resource));
-  }
-
-  boolean lastExportSkippedData() {
-    return this.lastExportSkippedData.get();
+  private void logExportFailure(Throwable throwable, ProjectName projectName) {
+    if (spannerExportFailureLoggedProjects.add(projectName.getProject())) {
+      String msg = "createServiceTimeSeries request failed for spanner metrics.";
+      if (throwable instanceof PermissionDeniedException) {
+        msg +=
+            String.format(
+                " Need monitoring metric writer permission on project=%s. Follow"
+                    + " https://cloud.google.com/spanner/docs/view-manage-client-side-metrics"
+                    + "#access-client-side-metrics"
+                    + " to set up permissions",
+                projectName.getProject());
+      } else {
+        msg += String.format(" project=%s.", projectName.getProject());
+      }
+      logger.log(Level.WARNING, msg, throwable);
+    }
   }
 
   private ApiFuture<List<Empty>> exportTimeSeriesInBatch(