chore: Collect Operation and Attempt metrics for gRPC requests (#12060)

Author: lqiu96

java-datastore/google-cloud-datastore/src/main/java/com/google/cloud/datastore/spi/v1/GrpcDatastoreRpc.java

@@ -31,9 +31,12 @@
 import com.google.api.gax.rpc.HeaderProvider;
 import com.google.api.gax.rpc.NoHeaderProvider;
 import com.google.api.gax.rpc.TransportChannel;
+import com.google.api.gax.tracing.MetricsTracerFactory;
+import com.google.api.gax.tracing.OpenTelemetryMetricsRecorder;
 import com.google.cloud.ServiceOptions;
 import com.google.cloud.datastore.DatastoreException;
 import com.google.cloud.datastore.DatastoreOptions;
+import com.google.cloud.datastore.telemetry.TelemetryConstants;
 import com.google.cloud.datastore.v1.DatastoreSettings;
 import com.google.cloud.datastore.v1.stub.DatastoreStubSettings;
 import com.google.cloud.datastore.v1.stub.GrpcDatastoreStub;
@@ -58,8 +61,12 @@
 import io.grpc.CallOptions;
 import io.grpc.ManagedChannel;
 import io.grpc.ManagedChannelBuilder;
+import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.api.OpenTelemetry;
 import java.io.IOException;
 import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
 
 @InternalApi
 public class GrpcDatastoreRpc implements DatastoreRpc {
@@ -76,7 +83,7 @@ public GrpcDatastoreRpc(DatastoreOptions datastoreOptions) throws IOException {
               : getClientContext(datastoreOptions);
 
       /* For grpc transport options, configure default gRPC Connection pool with minChannelCount = 1 */
-      DatastoreStubSettings datastoreStubSettings =
+      DatastoreStubSettings.Builder builder =
           DatastoreStubSettings.newBuilder(clientContext)
               .applyToAllUnaryMethods(retrySettingSetter(datastoreOptions))
               .setTransportChannelProvider(
@@ -86,14 +93,46 @@ public GrpcDatastoreRpc(DatastoreOptions datastoreOptions) throws IOException {
                               .setInitialChannelCount(DatastoreOptions.INIT_CHANNEL_COUNT)
                               .setMinChannelCount(DatastoreOptions.MIN_CHANNEL_COUNT)
                               .build())
-                      .build())
-              .build();
-      datastoreStub = GrpcDatastoreStub.create(datastoreStubSettings);
+                      .build());
+
+      // Hook into Gax's Metrics collection framework
+      MetricsTracerFactory metricsTracerFactory = buildMetricsTracerFactory(datastoreOptions);
+      if (metricsTracerFactory != null) {
+        builder.setTracerFactory(metricsTracerFactory);
+      }
+
+      datastoreStub = GrpcDatastoreStub.create(builder.build());
     } catch (IOException e) {
       throw new IOException(e);
     }
   }
 
+  /**
+   * Build the MetricsTracerFactory to hook into Gax's Otel Framework. Only hooks into Gax on two
+   * conditions: 1. OpenTelemetry instance is passed in by the user 2. Metrics are enabled
+   *
+   * <p>Sets default attributes to be recorded as part of the metrics.
+   */
+  static MetricsTracerFactory buildMetricsTracerFactory(DatastoreOptions datastoreOptions) {
+    if (!datastoreOptions.getOpenTelemetryOptions().isMetricsEnabled()) {
+      return null;
+    }
+    OpenTelemetry openTelemetry = datastoreOptions.getOpenTelemetryOptions().getOpenTelemetry();
+    if (openTelemetry == null) {
+      openTelemetry = GlobalOpenTelemetry.get();
+    }
+    OpenTelemetryMetricsRecorder gaxMetricsRecorder =
+        new OpenTelemetryMetricsRecorder(openTelemetry, TelemetryConstants.SERVICE_NAME);
+    Map<String, String> attributes = new HashMap<>();
+    attributes.put(TelemetryConstants.ATTRIBUTES_KEY_PROJECT_ID, datastoreOptions.getProjectId());
+    if (!Strings.isNullOrEmpty(datastoreOptions.getDatabaseId())) {
+      attributes.put(
+          TelemetryConstants.ATTRIBUTES_KEY_DATABASE_ID, datastoreOptions.getDatabaseId());
+    }
+    attributes.put(TelemetryConstants.ATTRIBUTES_KEY_TRANSPORT, "grpc");
+    return new MetricsTracerFactory(gaxMetricsRecorder, attributes);
+  }
+
   @Override
   public void close() throws Exception {
     if (!closed) {

java-datastore/google-cloud-datastore/src/main/java/com/google/cloud/datastore/telemetry/TelemetryConstants.java

@@ -29,7 +29,10 @@
  */
 @InternalApi
 public class TelemetryConstants {
-  static final String SERVICE_NAME = "datastore.googleapis.com";
+
+  // TODO(lawrenceqiu): For now, use `custom.googleapis.com` until metrics can be written to
+  // datastore domain
+  public static final String SERVICE_NAME = "custom.googleapis.com";
   static final String METER_NAME = "com.google.cloud.datastore";
 
   public static final String ATTRIBUTES_KEY_DOCUMENT_COUNT = "doc_count";
@@ -65,26 +68,11 @@ public class TelemetryConstants {
   public static final String METRIC_NAME_TRANSACTION_ATTEMPT_COUNT =
       SERVICE_NAME + "/client/transaction_attempt_count";
 
-  /* TODO(lawrenceqiu): For now, these are a duplicate of method names in TraceUtil. Those will use these eventually */
-  // Format is not SnakeCase to keep backward compatibility with the existing values TraceUtil spans
-  public static final String METHOD_ALLOCATE_IDS = "AllocateIds";
-  public static final String METHOD_BEGIN_TRANSACTION = "Transaction.Begin";
-  public static final String METHOD_COMMIT = "Commit";
-  public static final String METHOD_LOOKUP = "Lookup";
-  public static final String METHOD_RESERVE_IDS = "ReserveIds";
-  public static final String METHOD_RUN_QUERY = "RunQuery";
-  public static final String METHOD_TRANSACTION_COMMIT = "Transaction.Commit";
-  public static final String METHOD_TRANSACTION_LOOKUP = "Transaction.Lookup";
-  public static final String METHOD_TRANSACTION_RUN = "Transaction.Run";
-  public static final String METHOD_TRANSACTION_RUN_QUERY = "Transaction.RunQuery";
-  public static final String METHOD_TRANSACTION_ROLLBACK = "Transaction.Rollback";
-  public static final String METHOD_TRANSACTION_RUN_AGGREGATION_QUERY =
-      "Transaction.RunAggregationQuery";
-  public static final String METHOD_ADD = "add";
-  public static final String METHOD_PUT = "put";
-  public static final String METHOD_UPDATE = "update";
-  public static final String METHOD_DELETE = "delete";
-  public static final String METHOD_SUBMIT = "submit";
+  // Format is not SnakeCase to match the method name convention in Gax.
+  // The format is {ServiceName}.{MethodName}. For these methods, include `Transaction`
+  // to denote that the metrics are related specifically to transactions.
+  public static final String METHOD_TRANSACTION_COMMIT = "Datastore.Transaction.Commit";
+  public static final String METHOD_TRANSACTION_RUN = "Datastore.Transaction.Run";
 
   private TelemetryConstants() {}
 

java-datastore/google-cloud-datastore/src/test/java/com/google/cloud/datastore/DatastoreOpenTelemetryOptionsTestHelper.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2026 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.datastore;
+
+import io.opentelemetry.api.OpenTelemetry;
+
+// TODO(lawrenceqiu): This is a temporary class used to enabled metrics while `setMetricsEnabled`
+// is package private. This is to be removed later.
+public class DatastoreOpenTelemetryOptionsTestHelper {
+  public static DatastoreOpenTelemetryOptions withMetricsEnabled(OpenTelemetry openTelemetry) {
+    return DatastoreOpenTelemetryOptions.newBuilder()
+        .setMetricsEnabled(true)
+        .setOpenTelemetry(openTelemetry)
+        .build();
+  }
+
+  public static DatastoreOpenTelemetryOptions withMetricsEnabled() {
+    return DatastoreOpenTelemetryOptions.newBuilder().setMetricsEnabled(true).build();
+  }
+}