feat: add per-RPC transaction metrics for all transactional operations

Record transaction latency and attempt count at each individual RPC
when it involves a transaction:
- commit() when isTransactional
- lookup() when isTransactional
- runQuery() when isTransactional
- beginTransaction() (always transactional)
- rollback() (always transactional)
- AggregationQueryExecutor.execute() when transactional

Each RPC uses its own TelemetryConstants.METHOD_* constant. Removed
per-attempt recording from ReadWriteTransactionCallable since individual
RPCs now handle their own metrics. Made extractStatusCode public for
cross-package access from AggregationQueryExecutor.

Author: lqiu96

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

@@ -171,7 +171,7 @@ static DatastoreException propagateUserException(Exception ex) {
    * @param throwable the throwable to extract the status code from
    * @return the status code name, or "UNKNOWN" if not determinable
    */
-  static String extractStatusCode(Throwable throwable) {
+  public static String extractStatusCode(Throwable throwable) {
     Throwable current = throwable;
     while (current != null) {
       if (current instanceof DatastoreException) {

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

@@ -112,7 +112,8 @@ final class DatastoreImpl extends BaseService<DatastoreOptions> implements Datas
         new AggregationQueryExecutor(
             new RetryAndTraceDatastoreRpcDecorator(
                 datastoreRpc, otelTraceUtil, retrySettings, options),
-            options);
+            options,
+            metricsRecorder);
   }
 
   @Override

java-datastore/google-cloud-datastore/src/main/java/com/google/cloud/datastore/execution/AggregationQueryExecutor.java

@@ -16,18 +16,26 @@
 package com.google.cloud.datastore.execution;
 
 import com.google.api.core.InternalApi;
+import com.google.api.gax.rpc.StatusCode;
 import com.google.cloud.datastore.AggregationQuery;
 import com.google.cloud.datastore.AggregationResults;
+import com.google.cloud.datastore.DatastoreException;
 import com.google.cloud.datastore.DatastoreOptions;
 import com.google.cloud.datastore.ReadOption;
 import com.google.cloud.datastore.ReadOption.QueryConfig;
 import com.google.cloud.datastore.execution.request.AggregationQueryRequestProtoPreparer;
 import com.google.cloud.datastore.execution.response.AggregationQueryResponseTransformer;
 import com.google.cloud.datastore.models.ExplainOptions;
 import com.google.cloud.datastore.spi.v1.DatastoreRpc;
+import com.google.cloud.datastore.telemetry.MetricsRecorder;
+import com.google.cloud.datastore.telemetry.TelemetryConstants;
+import com.google.common.base.Stopwatch;
 import com.google.datastore.v1.RunAggregationQueryRequest;
 import com.google.datastore.v1.RunAggregationQueryResponse;
 import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
 
 /**
  * An implementation of {@link QueryExecutor} which executes {@link AggregationQuery} and returns
@@ -40,11 +48,16 @@ public class AggregationQueryExecutor
   private final DatastoreRpc datastoreRpc;
   private final AggregationQueryRequestProtoPreparer protoPreparer;
   private final AggregationQueryResponseTransformer responseTransformer;
+  private final MetricsRecorder metricsRecorder;
 
-  public AggregationQueryExecutor(DatastoreRpc datastoreRpc, DatastoreOptions datastoreOptions) {
+  public AggregationQueryExecutor(
+      DatastoreRpc datastoreRpc,
+      DatastoreOptions datastoreOptions,
+      MetricsRecorder metricsRecorder) {
     this.datastoreRpc = datastoreRpc;
     this.protoPreparer = new AggregationQueryRequestProtoPreparer(datastoreOptions);
     this.responseTransformer = new AggregationQueryResponseTransformer();
+    this.metricsRecorder = metricsRecorder;
   }
 
   @Override
@@ -53,9 +66,28 @@ public AggregationResults execute(
     RunAggregationQueryRequest runAggregationQueryRequest =
         getRunAggregationQueryRequest(
             query, explainOptions == null ? null : explainOptions.toPb(), readOptions);
-    RunAggregationQueryResponse runAggregationQueryResponse =
-        this.datastoreRpc.runAggregationQuery(runAggregationQueryRequest);
-    return this.responseTransformer.transform(runAggregationQueryResponse);
+    boolean isTransactional = runAggregationQueryRequest.getReadOptions().hasTransaction();
+    Stopwatch stopwatch = isTransactional ? Stopwatch.createStarted() : null;
+    String status = StatusCode.Code.OK.toString();
+    try {
+      RunAggregationQueryResponse runAggregationQueryResponse =
+          this.datastoreRpc.runAggregationQuery(runAggregationQueryRequest);
+      return this.responseTransformer.transform(runAggregationQueryResponse);
+    } catch (DatastoreException e) {
+      status = DatastoreException.extractStatusCode(e);
+      throw e;
+    } finally {
+      if (isTransactional) {
+        long latencyMs = stopwatch.elapsed(TimeUnit.MILLISECONDS);
+        Map<String, String> attributes = new HashMap<>();
+        attributes.put(TelemetryConstants.ATTRIBUTES_KEY_STATUS, status);
+        attributes.put(
+            TelemetryConstants.ATTRIBUTES_KEY_METHOD,
+            TelemetryConstants.METHOD_TRANSACTION_RUN_AGGREGATION_QUERY);
+        metricsRecorder.recordTransactionLatency(latencyMs, attributes);
+        metricsRecorder.recordTransactionAttemptCount(1, attributes);
+      }
+    }
   }
 
   private RunAggregationQueryRequest getRunAggregationQueryRequest(

java-datastore/google-cloud-datastore/src/test/java/com/google/cloud/datastore/execution/AggregationQueryExecutorTest.java

@@ -37,6 +37,7 @@
 import com.google.cloud.datastore.LongValue;
 import com.google.cloud.datastore.Query;
 import com.google.cloud.datastore.spi.v1.DatastoreRpc;
+import com.google.cloud.datastore.telemetry.MetricsRecorder;
 import com.google.common.collect.ImmutableMap;
 import com.google.datastore.v1.AggregationResultBatch;
 import com.google.datastore.v1.RunAggregationQueryRequest;
@@ -63,7 +64,11 @@ public void setUp() throws Exception {
     mockRpc = EasyMock.createStrictMock(DatastoreRpc.class);
     datastoreOptions =
         DatastoreOptions.newBuilder().setProjectId("project-id").setNamespace(NAMESPACE).build();
-    queryExecutor = new AggregationQueryExecutor(mockRpc, datastoreOptions);
+    queryExecutor =
+        new AggregationQueryExecutor(
+            mockRpc,
+            datastoreOptions,
+            MetricsRecorder.getInstance(datastoreOptions.getOpenTelemetryOptions()));
   }
 
   @Test