chore(Datastore): Refactor Datastore Metric classes to inherit from Gax

java-datastore/google-cloud-datastore/src/main/java/com/google/cloud/datastore/DatastoreImpl.java

@@ -48,12 +48,11 @@
 import com.google.cloud.TransportOptions;
 import com.google.cloud.datastore.execution.AggregationQueryExecutor;
 import com.google.cloud.datastore.spi.v1.DatastoreRpc;
-import com.google.cloud.datastore.telemetry.MetricsRecorder;
+import com.google.cloud.datastore.telemetry.DatastoreMetricsRecorder;
 import com.google.cloud.datastore.telemetry.TelemetryConstants;
 import com.google.cloud.datastore.telemetry.TelemetryUtils;
 import com.google.cloud.datastore.telemetry.TraceUtil;
 import com.google.cloud.datastore.telemetry.TraceUtil.Scope;
-import com.google.cloud.http.HttpTransportOptions;
 import com.google.common.base.MoreObjects;
 import com.google.common.base.Preconditions;
 import com.google.common.base.Stopwatch;
@@ -101,16 +100,13 @@ final class DatastoreImpl extends BaseService<DatastoreOptions> implements Datas
 
   private final com.google.cloud.datastore.telemetry.TraceUtil otelTraceUtil =
       getOptions().getTraceUtil();
-  private final MetricsRecorder metricsRecorder = getOptions().getMetricsRecorder();
-  private final boolean isHttpTransport;
-
+  private final DatastoreMetricsRecorder metricsRecorder = getOptions().getMetricsRecorder();
   private final ReadOptionProtoPreparer readOptionProtoPreparer;
   private final AggregationQueryExecutor aggregationQueryExecutor;
 
   DatastoreImpl(DatastoreOptions options) {
     super(options);
     this.datastoreRpc = options.getDatastoreRpcV1();
-    this.isHttpTransport = options.getTransportOptions() instanceof HttpTransportOptions;
     retrySettings =
         MoreObjects.firstNonNull(options.getRetrySettings(), ServiceOptions.getNoRetrySettings());
 
@@ -185,15 +181,15 @@ public boolean isClosed() {
   static class ReadWriteTransactionCallable<T> implements Callable<T> {
     private final Datastore datastore;
     private final TransactionCallable<T> callable;
-    private final MetricsRecorder metricsRecorder;
+    private final DatastoreMetricsRecorder metricsRecorder;
     private volatile TransactionOptions options;
     private volatile Transaction transaction;
 
     ReadWriteTransactionCallable(
         Datastore datastore,
         TransactionCallable<T> callable,
         TransactionOptions options,
-        MetricsRecorder metricsRecorder) {
+        DatastoreMetricsRecorder metricsRecorder) {
       this.datastore = datastore;
       this.callable = callable;
       this.options = options;
@@ -222,7 +218,7 @@ public T call() throws DatastoreException {
         // or from `transaction.commit()`. If there is an exception thrown from either call site,
         // then the transaction is still active. Check if it is still active (e.g. not commited)
         // and roll back the transaction.
-        if (transaction.isActive()) {
+        if (transaction != null && transaction.isActive()) {
           transaction.rollback();
         }
         throw DatastoreException.propagateUserException(ex);
@@ -231,10 +227,11 @@ public T call() throws DatastoreException {
         // If the transaction is active, then commit the rollback. If it was already successfully
         // rolled back, the transaction is inactive (prevents rolling back an already rolled back
         // transaction).
-        if (transaction.isActive()) {
+        if (transaction != null && transaction.isActive()) {
           transaction.rollback();
         }
-        if (options != null
+        if (transaction != null
+            && options != null
             && options.getModeCase().equals(TransactionOptions.ModeCase.READ_WRITE)) {
           setPrevTransactionId(transaction.getTransactionId());
         }
@@ -805,15 +802,12 @@ private <T> T runWithObservability(
       ResultRetryAlgorithm<?> exceptionHandler) {
     com.google.cloud.datastore.telemetry.TraceUtil.Span span = otelTraceUtil.startSpan(spanName);
 
-    // Gax already records operation and attempt metrics. Since Datastore HttpJson does not
-    // integrate with Gax, manually instrument these metrics when using HttpJson for parity
-    Stopwatch operationStopwatch = isHttpTransport ? Stopwatch.createStarted() : null;
+    Stopwatch operationStopwatch = Stopwatch.createStarted();
     String operationStatus = StatusCode.Code.OK.toString();
 
     DatastoreOptions options = getOptions();
     Callable<T> attemptCallable =
-        TelemetryUtils.attemptMetricsCallable(
-            callable, metricsRecorder, options, isHttpTransport, methodName);
+        TelemetryUtils.attemptMetricsCallable(callable, metricsRecorder, options, methodName);
     try (TraceUtil.Scope ignored = span.makeCurrent()) {
       return RetryHelper.runWithRetries(
           attemptCallable, retrySettings, exceptionHandler, options.getClock());
@@ -823,12 +817,7 @@ private <T> T runWithObservability(
       throw DatastoreException.translateAndThrow(e);
     } finally {
       TelemetryUtils.recordOperationMetrics(
-          metricsRecorder,
-          options,
-          isHttpTransport,
-          operationStopwatch,
-          methodName,
-          operationStatus);
+          metricsRecorder, options, operationStopwatch, methodName, operationStatus);
       span.end();
     }
   }

java-datastore/google-cloud-datastore/src/main/java/com/google/cloud/datastore/DatastoreOptions.java

@@ -31,7 +31,7 @@
 import com.google.cloud.datastore.spi.v1.DatastoreRpc;
 import com.google.cloud.datastore.spi.v1.GrpcDatastoreRpc;
 import com.google.cloud.datastore.spi.v1.HttpDatastoreRpc;
-import com.google.cloud.datastore.telemetry.MetricsRecorder;
+import com.google.cloud.datastore.telemetry.DatastoreMetricsRecorder;
 import com.google.cloud.datastore.v1.DatastoreSettings;
 import com.google.cloud.grpc.GrpcTransportOptions;
 import com.google.cloud.http.HttpTransportOptions;
@@ -65,7 +65,7 @@ public class DatastoreOptions extends ServiceOptions<Datastore, DatastoreOptions
 
   private final transient @Nonnull DatastoreOpenTelemetryOptions openTelemetryOptions;
   private final transient @Nonnull com.google.cloud.datastore.telemetry.TraceUtil traceUtil;
-  private final transient @Nonnull MetricsRecorder metricsRecorder;
+  private final transient @Nonnull DatastoreMetricsRecorder metricsRecorder;
 
   public static class DefaultDatastoreFactory implements DatastoreFactory {
 
@@ -107,7 +107,7 @@ public DatastoreOpenTelemetryOptions getOpenTelemetryOptions() {
   }
 
   @Nonnull
-  MetricsRecorder getMetricsRecorder() {
+  DatastoreMetricsRecorder getMetricsRecorder() {
     return metricsRecorder;
   }
 
@@ -193,7 +193,7 @@ public Builder setDatabaseId(String databaseId) {
     }
 
     /**
-     * Sets the {@link DatastoreOpenTelemetryOptions} to be used for this Firestore instance.
+     * Sets the {@link DatastoreOpenTelemetryOptions} to be used for this Datastore instance.
      *
      * @param openTelemetryOptions The `DatastoreOpenTelemetryOptions` to use.
      */
@@ -223,7 +223,7 @@ private DatastoreOptions(Builder builder) {
             ? builder.openTelemetryOptions
             : DatastoreOpenTelemetryOptions.newBuilder().build();
     this.traceUtil = com.google.cloud.datastore.telemetry.TraceUtil.getInstance(this);
-    this.metricsRecorder = MetricsRecorder.getInstance(openTelemetryOptions);
+    this.metricsRecorder = DatastoreMetricsRecorder.getInstance(this);
 
     namespace = MoreObjects.firstNonNull(builder.namespace, defaultNamespace());
     databaseId = MoreObjects.firstNonNull(builder.databaseId, DEFAULT_DATABASE_ID);

java-datastore/google-cloud-datastore/src/main/java/com/google/cloud/datastore/RetryAndTraceDatastoreRpcDecorator.java

@@ -24,12 +24,11 @@
 import com.google.cloud.RetryHelper;
 import com.google.cloud.RetryHelper.RetryHelperException;
 import com.google.cloud.datastore.spi.v1.DatastoreRpc;
-import com.google.cloud.datastore.telemetry.MetricsRecorder;
-import com.google.cloud.datastore.telemetry.NoOpMetricsRecorder;
+import com.google.cloud.datastore.telemetry.DatastoreMetricsRecorder;
+import com.google.cloud.datastore.telemetry.NoOpDatastoreMetricsRecorder;
 import com.google.cloud.datastore.telemetry.TelemetryConstants;
 import com.google.cloud.datastore.telemetry.TelemetryUtils;
 import com.google.cloud.datastore.telemetry.TraceUtil;
-import com.google.cloud.http.HttpTransportOptions;
 import com.google.common.base.Preconditions;
 import com.google.common.base.Stopwatch;
 import com.google.datastore.v1.AllocateIdsRequest;
@@ -62,8 +61,7 @@ public class RetryAndTraceDatastoreRpcDecorator implements DatastoreRpc {
   private final com.google.cloud.datastore.telemetry.TraceUtil otelTraceUtil;
   private final RetrySettings retrySettings;
   private final DatastoreOptions datastoreOptions;
-  private final MetricsRecorder metricsRecorder;
-  private final boolean isHttpTransport;
+  private final DatastoreMetricsRecorder metricsRecorder;
 
   @ObsoleteApi("Prefer to create RetryAndTraceDatastoreRpcDecorator via the Builder")
   public RetryAndTraceDatastoreRpcDecorator(
@@ -75,8 +73,7 @@ public RetryAndTraceDatastoreRpcDecorator(
     this.retrySettings = retrySettings;
     this.datastoreOptions = datastoreOptions;
     this.otelTraceUtil = otelTraceUtil;
-    this.metricsRecorder = new NoOpMetricsRecorder();
-    this.isHttpTransport = datastoreOptions.getTransportOptions() instanceof HttpTransportOptions;
+    this.metricsRecorder = new NoOpDatastoreMetricsRecorder();
   }
 
   private RetryAndTraceDatastoreRpcDecorator(Builder builder) {
@@ -85,7 +82,6 @@ private RetryAndTraceDatastoreRpcDecorator(Builder builder) {
     this.retrySettings = builder.retrySettings;
     this.datastoreOptions = builder.datastoreOptions;
     this.metricsRecorder = builder.metricsRecorder;
-    this.isHttpTransport = builder.isHttpTransport;
   }
 
   public static Builder newBuilder() {
@@ -99,8 +95,7 @@ public static class Builder {
     private DatastoreOptions datastoreOptions;
 
     // Defaults configured for this class
-    private MetricsRecorder metricsRecorder = new NoOpMetricsRecorder();
-    private boolean isHttpTransport = false;
+    private DatastoreMetricsRecorder metricsRecorder = new NoOpDatastoreMetricsRecorder();
 
     private Builder() {}
 
@@ -124,7 +119,7 @@ public Builder setDatastoreOptions(DatastoreOptions datastoreOptions) {
       return this;
     }
 
-    public Builder setMetricsRecorder(MetricsRecorder metricsRecorder) {
+    public Builder setMetricsRecorder(DatastoreMetricsRecorder metricsRecorder) {
       Preconditions.checkNotNull(metricsRecorder, "metricsRecorder can not be null");
       this.metricsRecorder = metricsRecorder;
       return this;
@@ -135,7 +130,6 @@ public RetryAndTraceDatastoreRpcDecorator build() {
       Preconditions.checkNotNull(otelTraceUtil, "otelTraceUtil is required");
       Preconditions.checkNotNull(retrySettings, "retrySettings is required");
       Preconditions.checkNotNull(datastoreOptions, "datastoreOptions is required");
-      this.isHttpTransport = datastoreOptions.getTransportOptions() instanceof HttpTransportOptions;
       return new RetryAndTraceDatastoreRpcDecorator(this);
     }
   }
@@ -207,12 +201,12 @@ public <O> O invokeRpc(Callable<O> block, String startSpan) {
 
   <O> O invokeRpc(Callable<O> block, String startSpan, String methodName) {
     TraceUtil.Span span = otelTraceUtil.startSpan(startSpan);
-    Stopwatch stopwatch = isHttpTransport ? Stopwatch.createStarted() : null;
+    Stopwatch stopwatch = Stopwatch.createStarted();
     String operationStatus = StatusCode.Code.UNKNOWN.toString();
     try (TraceUtil.Scope ignored = span.makeCurrent()) {
       Callable<O> callable =
           TelemetryUtils.attemptMetricsCallable(
-              block, metricsRecorder, datastoreOptions, isHttpTransport, methodName);
+              block, metricsRecorder, datastoreOptions, methodName);
       O result =
           RetryHelper.runWithRetries(
               callable, this.retrySettings, EXCEPTION_HANDLER, this.datastoreOptions.getClock());
@@ -224,12 +218,7 @@ <O> O invokeRpc(Callable<O> block, String startSpan, String methodName) {
       throw DatastoreException.translateAndThrow(e);
     } finally {
       TelemetryUtils.recordOperationMetrics(
-          metricsRecorder,
-          datastoreOptions,
-          isHttpTransport,
-          stopwatch,
-          methodName,
-          operationStatus);
+          metricsRecorder, datastoreOptions, stopwatch, methodName, operationStatus);
       span.end();
     }
   }

java-datastore/google-cloud-datastore/src/main/java/com/google/cloud/datastore/samples/MetricsSample.java

@@ -0,0 +1,72 @@
+/*
+ * 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.samples;
+
+import com.google.cloud.datastore.Datastore;
+import com.google.cloud.datastore.DatastoreOptions;
+import com.google.cloud.datastore.Entity;
+import com.google.cloud.datastore.Key;
+
+/** A simple sample to verify that Datastore metrics are being recorded and exported correctly. */
+public class MetricsSample {
+
+  public static void main(String[] args) throws Exception {
+    String projectId = System.getenv("GOOGLE_CLOUD_PROJECT");
+    if (projectId == null || projectId.isEmpty()) {
+      System.err.println("Error: GOOGLE_CLOUD_PROJECT environment variable is not set.");
+      System.exit(1);
+    }
+
+    System.out.println("Starting MetricsSample for project: " + projectId);
+
+    // Initialize Datastore client with metrics enabled by default.
+    Datastore datastore =
+        DatastoreOptions.newBuilder().setProjectId(projectId).build().getService();
+
+    System.out.println("Executing operations to generate metrics...");
+
+    // 1. Basic Key Lookup
+    Key key = datastore.newKeyFactory().setKind("MetricsTask").newKey("sample-task");
+    datastore.get(key);
+    System.out.println("Performed lookup.");
+
+    // 2. Transactional Read/Write
+    datastore.runInTransaction(
+        tx -> {
+          Entity entity =
+              Entity.newBuilder(key)
+                  .set("description", "Recorded via MetricsSample")
+                  .set("timestamp", System.currentTimeMillis())
+                  .build();
+          tx.put(entity);
+          return null;
+        });
+    System.out.println("Performed transaction.");
+
+    // 3. Batch operation
+    datastore.delete(key);
+    System.out.println("Performed delete.");
+
+    System.out.println("Waiting 65 seconds for metrics to be flushed to Cloud Monitoring...");
+    // PeriodicMetricReader default interval is 60s.
+    Thread.sleep(65000);
+
+    System.out.println("Sample finished. Check the Cloud Monitoring dashboard for:");
+    System.out.println(" - firestore.googleapis.com/internal/client/operation_latency");
+    System.out.println(" - firestore.googleapis.com/internal/client/transaction_latency");
+  }
+}

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

@@ -31,12 +31,9 @@
 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;
@@ -61,12 +58,8 @@
 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 {
@@ -95,44 +88,12 @@ public GrpcDatastoreRpc(DatastoreOptions datastoreOptions) throws IOException {
                               .build())
                       .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) {