feat(spanner): support user-provided OpenTelemetry for built-in client metrics and endpoint attribute for location-aware routing

Adds a MetricsProvider family on SpannerOptions that controls how the
built-in client metrics are collected and exported: DefaultMetricsProvider
(collect and export to Cloud Monitoring, today's behavior and still the
default), NoopMetricsProvider (built-in metrics off), and
CustomOpenTelemetryMetricsProvider (record the built-in metrics on a
caller-provided OpenTelemetry instance instead of exporting them to Cloud
Monitoring, leaving the metrics pipeline - readers, exporters, resource -
fully owned by the caller). A custom provider records under the distinct
spanner/internal/client instrument namespace so a customer pipeline
re-exported to Cloud Monitoring cannot conflict with the reserved
spanner.googleapis.com descriptors. The new static
SpannerOptions.registerBuiltInMetricViews(SdkMeterProviderBuilder) registers
the renaming/bucketing/attribute-filtering views for the custom namespace.

Default behavior is unchanged: the default provider keeps today's Cloud
Monitoring export, the process-wide exporter singleton is never touched by
custom-sink clients, and OMNI still has built-in metrics off by default
(the default provider is coerced to noop) while an explicitly configured
CustomOpenTelemetryMetricsProvider is honored, which is the opt-in for
Spanner Omni where Cloud Monitoring export is unavailable. The
SPANNER_DISABLE_BUILTIN_METRICS environment variable disables the custom
sink as well.

Also records an endpoint attribute on attempt metrics for location-aware
routing: KeyAwareClientCall injects the resolved endpoint into the operation
tracer through the existing CallOptions tracer bridge. The attribute is only
allowed through the attribute filter of the custom-export views; the Cloud
Monitoring views drop it, so the exported Cloud Monitoring schema is
unchanged. Clients that do not enable the location API never produce the
attribute.

Author: rahul2393

java-spanner/.readme-partials.yaml

@@ -69,6 +69,45 @@ custom_content: |
   
   > Note: Client-side metrics needs `monitoring.timeSeries.create` IAM permission to export metrics data. Ask your administrator to grant your service account the [Monitoring Metric Writer](https://cloud.google.com/iam/docs/roles-permissions/monitoring#monitoring.metricWriter) (roles/monitoring.metricWriter) IAM role on the project.
 
+  #### Customizing how built-in metrics are collected
+
+  How built-in metrics are collected and exported is controlled by a `MetricsProvider`,
+  set with `SpannerOptions.Builder.setMetricsProvider(MetricsProvider)`. The available
+  providers are:
+
+  * `DefaultMetricsProvider` (the default): built-in metrics are collected and exported to
+    Google Cloud Monitoring.
+  * `NoopMetricsProvider`: built-in metrics are disabled (equivalent to
+    `setBuiltInMetricsEnabled(false)`).
+  * `CustomOpenTelemetryMetricsProvider`: built-in metrics are recorded on an
+    `OpenTelemetry` instance that you provide instead of being exported to Cloud Monitoring.
+    You own the metrics pipeline (readers, exporters and resource). This is the way to
+    collect built-in metrics for Spanner Omni (`InstanceType.OMNI`), for which Cloud
+    Monitoring export is not available.
+
+  An explicitly set provider takes precedence over `setBuiltInMetricsEnabled(boolean)`, and
+  the `SPANNER_DISABLE_BUILTIN_METRICS` environment variable disables built-in metrics for
+  all providers.
+
+  When using `CustomOpenTelemetryMetricsProvider`, it is recommended to register the Spanner
+  built-in metric views on a dedicated `SdkMeterProviderBuilder` with
+  `SpannerOptions.registerBuiltInMetricViews(SdkMeterProviderBuilder)` before creating the
+  `OpenTelemetry` instance. The views rename the raw instruments, apply the Spanner latency
+  histogram buckets and restrict the recorded attributes to the supported built-in metric
+  labels.
+
+  ```java
+  SdkMeterProviderBuilder meterProviderBuilder =
+      SdkMeterProvider.builder().registerMetricReader(PeriodicMetricReader.create(myExporter));
+  SpannerOptions.registerBuiltInMetricViews(meterProviderBuilder);
+  OpenTelemetry openTelemetry =
+      OpenTelemetrySdk.builder().setMeterProvider(meterProviderBuilder.build()).build();
+  SpannerOptions options =
+      SpannerOptions.newBuilder()
+          .setMetricsProvider(CustomOpenTelemetryMetricsProvider.create(openTelemetry))
+          .build();
+  ```
+
   ## Traces
   Cloud Spanner client supports OpenTelemetry Traces, which gives insight into the client internals and aids in debugging/troubleshooting production issues. 
 

java-spanner/README.md

@@ -167,6 +167,45 @@ You can also disable these metrics by setting `SPANNER_DISABLE_BUILTIN_METRICS`
 
 > Note: Client-side metrics needs `monitoring.timeSeries.create` IAM permission to export metrics data. Ask your administrator to grant your service account the [Monitoring Metric Writer](https://cloud.google.com/iam/docs/roles-permissions/monitoring#monitoring.metricWriter) (roles/monitoring.metricWriter) IAM role on the project.
 
+#### Customizing how built-in metrics are collected
+
+How built-in metrics are collected and exported is controlled by a `MetricsProvider`,
+set with `SpannerOptions.Builder.setMetricsProvider(MetricsProvider)`. The available
+providers are:
+
+* `DefaultMetricsProvider` (the default): built-in metrics are collected and exported to
+  Google Cloud Monitoring.
+* `NoopMetricsProvider`: built-in metrics are disabled (equivalent to
+  `setBuiltInMetricsEnabled(false)`).
+* `CustomOpenTelemetryMetricsProvider`: built-in metrics are recorded on an
+  `OpenTelemetry` instance that you provide instead of being exported to Cloud Monitoring.
+  You own the metrics pipeline (readers, exporters and resource). This is the way to
+  collect built-in metrics for Spanner Omni (`InstanceType.OMNI`), for which Cloud
+  Monitoring export is not available.
+
+An explicitly set provider takes precedence over `setBuiltInMetricsEnabled(boolean)`, and
+the `SPANNER_DISABLE_BUILTIN_METRICS` environment variable disables built-in metrics for
+all providers.
+
+When using `CustomOpenTelemetryMetricsProvider`, it is recommended to register the Spanner
+built-in metric views on a dedicated `SdkMeterProviderBuilder` with
+`SpannerOptions.registerBuiltInMetricViews(SdkMeterProviderBuilder)` before creating the
+`OpenTelemetry` instance. The views rename the raw instruments, apply the Spanner latency
+histogram buckets and restrict the recorded attributes to the supported built-in metric
+labels.
+
+```java
+SdkMeterProviderBuilder meterProviderBuilder =
+    SdkMeterProvider.builder().registerMetricReader(PeriodicMetricReader.create(myExporter));
+SpannerOptions.registerBuiltInMetricViews(meterProviderBuilder);
+OpenTelemetry openTelemetry =
+    OpenTelemetrySdk.builder().setMeterProvider(meterProviderBuilder.build()).build();
+SpannerOptions options =
+    SpannerOptions.newBuilder()
+        .setMetricsProvider(CustomOpenTelemetryMetricsProvider.create(openTelemetry))
+        .build();
+```
+
 ## Traces
 Cloud Spanner client supports OpenTelemetry Traces, which gives insight into the client internals and aids in debugging/troubleshooting production issues. 
 

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

@@ -40,6 +40,16 @@
 public class BuiltInMetricsConstant {
 
   public static final String METER_NAME = "spanner.googleapis.com/internal/client";
+
+  /**
+   * Instrument namespace used when built-in metrics are recorded on a caller-provided {@link
+   * io.opentelemetry.api.OpenTelemetry} via {@link CustomOpenTelemetryMetricsProvider}. This is
+   * deliberately distinct from the reserved {@code spanner.googleapis.com} namespace, so that
+   * re-exporting a caller-owned pipeline to Cloud Monitoring cannot conflict with the curated
+   * Spanner metric descriptors.
+   */
+  public static final String CUSTOM_EXPORT_METER_NAME = "spanner/internal/client";
+
   public static final String GAX_METER_NAME = OpenTelemetryMetricsRecorder.GAX_METER_NAME;
   public static final String GRPC_GCP_METER_NAME = "grpc-gcp";
   static final String SPANNER_METER_NAME = "spanner-java";
@@ -168,6 +178,14 @@ public class BuiltInMetricsConstant {
       AttributeKey.stringKey("directpath_used");
   public static final AttributeKey<String> REQUEST_ID_KEY =
       AttributeKey.stringKey(REQUEST_ID_HEADER_NAME);
+
+  /**
+   * The endpoint an attempt was routed to by the location-aware fastpath. Only produced when the
+   * experimental location API is enabled (Spanner Omni), and only allowed through the attribute
+   * filter of the custom-export views; the Cloud Monitoring views never include it.
+   */
+  public static final AttributeKey<String> ENDPOINT_KEY = AttributeKey.stringKey("endpoint");
+
   public static Set<String> ALLOWED_EXEMPLARS_ATTRIBUTES =
       new HashSet<>(Arrays.asList(REQUEST_ID_HEADER_NAME));
 
@@ -208,102 +226,138 @@ public class BuiltInMetricsConstant {
 
   static Map<InstrumentSelector, View> getAllViews() {
     ImmutableMap.Builder<InstrumentSelector, View> views = ImmutableMap.builder();
+    addViews(views, METER_NAME, false);
+    defineGrpcGcpView(views);
+    return views.build();
+  }
+
+  /**
+   * Views for built-in metrics recorded on a caller-provided OpenTelemetry via {@link
+   * CustomOpenTelemetryMetricsProvider}. The instruments live in the {@link
+   * #CUSTOM_EXPORT_METER_NAME} namespace, and the attribute filter additionally allows the {@link
+   * #ENDPOINT_KEY} label. The grpc-gcp DirectPath-fallback views are deliberately excluded: those
+   * metrics are only recorded on the Cloud Monitoring path, so registering their views on a custom
+   * sink would be dead configuration.
+   */
+  static Map<InstrumentSelector, View> getCustomExportViews() {
+    ImmutableMap.Builder<InstrumentSelector, View> views = ImmutableMap.builder();
+    addViews(views, CUSTOM_EXPORT_METER_NAME, true);
+    return views.build();
+  }
+
+  private static void addViews(
+      ImmutableMap.Builder<InstrumentSelector, View> views,
+      String metricPrefix,
+      boolean includeEndpointAttribute) {
     defineView(
         views,
+        metricPrefix,
         BuiltInMetricsConstant.GAX_METER_NAME,
         BuiltInMetricsConstant.OPERATION_LATENCY_NAME,
         BuiltInMetricsConstant.OPERATION_LATENCIES_NAME,
         BuiltInMetricsConstant.AGGREGATION_WITH_MILLIS_HISTOGRAM,
         InstrumentType.HISTOGRAM,
-        "ms");
+        "ms",
+        includeEndpointAttribute);
     defineView(
         views,
+        metricPrefix,
         BuiltInMetricsConstant.GAX_METER_NAME,
         BuiltInMetricsConstant.ATTEMPT_LATENCY_NAME,
         BuiltInMetricsConstant.ATTEMPT_LATENCIES_NAME,
         BuiltInMetricsConstant.AGGREGATION_WITH_MILLIS_HISTOGRAM,
         InstrumentType.HISTOGRAM,
-        "ms");
+        "ms",
+        includeEndpointAttribute);
     defineView(
         views,
+        metricPrefix,
         BuiltInMetricsConstant.GAX_METER_NAME,
         BuiltInMetricsConstant.OPERATION_COUNT_NAME,
         BuiltInMetricsConstant.OPERATION_COUNT_NAME,
         Aggregation.sum(),
         InstrumentType.COUNTER,
-        "1");
+        "1",
+        includeEndpointAttribute);
     defineView(
         views,
+        metricPrefix,
         BuiltInMetricsConstant.GAX_METER_NAME,
         BuiltInMetricsConstant.ATTEMPT_COUNT_NAME,
         BuiltInMetricsConstant.ATTEMPT_COUNT_NAME,
         Aggregation.sum(),
         InstrumentType.COUNTER,
-        "1");
-    defineSpannerView(views);
-    defineGRPCView(views);
-    defineGrpcGcpView(views);
-    return views.build();
+        "1",
+        includeEndpointAttribute);
+    defineSpannerView(views, includeEndpointAttribute);
+    defineGRPCView(views, metricPrefix, includeEndpointAttribute);
+  }
+
+  private static Set<String> baseAttributesFilter(boolean includeEndpointAttribute) {
+    Set<String> attributesFilter =
+        BuiltInMetricsConstant.COMMON_ATTRIBUTES.stream()
+            .map(AttributeKey::getKey)
+            .collect(Collectors.toSet());
+    if (includeEndpointAttribute) {
+      attributesFilter.add(ENDPOINT_KEY.getKey());
+    }
+    return attributesFilter;
   }
 
   private static void defineView(
       ImmutableMap.Builder<InstrumentSelector, View> viewMap,
+      String metricPrefix,
       String meterName,
       String metricName,
       String metricViewName,
       Aggregation aggregation,
       InstrumentType type,
-      String unit) {
+      String unit,
+      boolean includeEndpointAttribute) {
     InstrumentSelector selector =
         InstrumentSelector.builder()
-            .setName(BuiltInMetricsConstant.METER_NAME + '/' + metricName)
+            .setName(metricPrefix + '/' + metricName)
             .setMeterName(meterName)
             .setType(type)
             .setUnit(unit)
             .build();
-    Set<String> attributesFilter =
-        BuiltInMetricsConstant.COMMON_ATTRIBUTES.stream()
-            .map(AttributeKey::getKey)
-            .collect(Collectors.toSet());
     View view =
         View.builder()
-            .setName(BuiltInMetricsConstant.METER_NAME + '/' + metricViewName)
+            .setName(metricPrefix + '/' + metricViewName)
             .setAggregation(aggregation)
-            .setAttributeFilter(attributesFilter)
+            .setAttributeFilter(baseAttributesFilter(includeEndpointAttribute))
             .build();
     viewMap.put(selector, view);
   }
 
-  private static void defineSpannerView(ImmutableMap.Builder<InstrumentSelector, View> viewMap) {
+  private static void defineSpannerView(
+      ImmutableMap.Builder<InstrumentSelector, View> viewMap, boolean includeEndpointAttribute) {
     InstrumentSelector selector =
         InstrumentSelector.builder()
             .setMeterName(BuiltInMetricsConstant.SPANNER_METER_NAME)
             .build();
-    Set<String> attributesFilter =
-        BuiltInMetricsConstant.COMMON_ATTRIBUTES.stream()
-            .map(AttributeKey::getKey)
-            .collect(Collectors.toSet());
-    View view = View.builder().setAttributeFilter(attributesFilter).build();
+    View view =
+        View.builder().setAttributeFilter(baseAttributesFilter(includeEndpointAttribute)).build();
     viewMap.put(selector, view);
   }
 
-  private static void defineGRPCView(ImmutableMap.Builder<InstrumentSelector, View> viewMap) {
+  private static void defineGRPCView(
+      ImmutableMap.Builder<InstrumentSelector, View> viewMap,
+      String metricPrefix,
+      boolean includeEndpointAttribute) {
     for (String metric : BuiltInMetricsConstant.GRPC_METRICS_TO_ENABLE) {
       InstrumentSelector selector =
           InstrumentSelector.builder()
               .setName(metric)
               .setMeterName(BuiltInMetricsConstant.GRPC_METER_NAME)
               .build();
-      Set<String> attributesFilter =
-          BuiltInMetricsConstant.COMMON_ATTRIBUTES.stream()
-              .map(AttributeKey::getKey)
-              .collect(Collectors.toSet());
+      Set<String> attributesFilter = baseAttributesFilter(includeEndpointAttribute);
       attributesFilter.addAll(
           GRPC_METRIC_ADDITIONAL_ATTRIBUTES.getOrDefault(metric, ImmutableSet.of()));
 
       View view =
           View.builder()
-              .setName(BuiltInMetricsConstant.METER_NAME + '/' + metric.replace(".", "/"))
+              .setName(metricPrefix + '/' + metric.replace(".", "/"))
               .setAttributeFilter(attributesFilter)
               .build();
       viewMap.put(selector, view);
@@ -318,11 +372,7 @@ private static void defineGrpcGcpView(ImmutableMap.Builder<InstrumentSelector, V
               .setMeterName(BuiltInMetricsConstant.GRPC_GCP_METER_NAME)
               .build();
 
-      Set<String> attributesFilter =
-          BuiltInMetricsConstant.COMMON_ATTRIBUTES.stream()
-              .map(AttributeKey::getKey)
-              .collect(Collectors.toSet());
-
+      Set<String> attributesFilter = baseAttributesFilter(false);
       attributesFilter.addAll(
           GRPC_GCP_METRIC_ADDITIONAL_ATTRIBUTES.getOrDefault(metric, ImmutableSet.of()));
 

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

@@ -188,11 +188,22 @@ void enableGrpcMetrics(
       @Nullable Credentials credentials,
       @Nullable String monitoringHost,
       String universeDomain) {
+    enableGrpcMetrics(
+        channelProviderBuilder,
+        this.getOrCreateOpenTelemetry(projectId, credentials, monitoringHost, universeDomain));
+  }
+
+  /**
+   * Wires gRPC-layer metrics into the channel builder using the given OpenTelemetry instance
+   * instead of the process-wide Cloud Monitoring one. Used for {@link
+   * CustomOpenTelemetryMetricsProvider}; does not touch any state of this singleton.
+   */
+  void enableGrpcMetrics(
+      InstantiatingGrpcChannelProvider.Builder channelProviderBuilder,
+      OpenTelemetry openTelemetry) {
     GrpcOpenTelemetry grpcOpenTelemetry =
         GrpcOpenTelemetry.newBuilder()
-            .sdk(
-                this.getOrCreateOpenTelemetry(
-                    projectId, credentials, monitoringHost, universeDomain))
+            .sdk(openTelemetry)
             .enableMetrics(BuiltInMetricsConstant.GRPC_METRICS_TO_ENABLE)
             // Disable gRPCs default metrics as they are not needed for Spanner.
             .disableMetrics(BuiltInMetricsConstant.GRPC_METRICS_ENABLED_BY_DEFAULT)