Feat: client side metrics

google-cloud-firestore/clirr-ignored-differences.xml

@@ -314,4 +314,19 @@
     <method>com.google.cloud.firestore.Query getQuery()</method>
   </difference>
 
+    <!-- Relocate constants in TraceUtil -->
+  <difference>
+    <differenceType>6011</differenceType>
+    <className>com/google/cloud/firestore/telemetry/TraceUtil</className>
+    <field>SPAN_NAME_*</field>
+  </difference>
+
+  <!-- Change the parameter type of internalStream function -->
+  <difference>
+    <differenceType>7005</differenceType>
+    <className>com/google/cloud/firestore/StreamableQuery</className>
+    <method>* internalStream(com.google.api.gax.rpc.ApiStreamObserver, long, com.google.protobuf.ByteString, com.google.cloud.Timestamp, com.google.cloud.firestore.ExplainOptions, boolean)</method>
+    <to>*</to>
+  </difference>
+
 </differences>

google-cloud-firestore/pom.xml

@@ -119,6 +119,18 @@
       <groupId>io.opentelemetry.instrumentation</groupId>
       <artifactId>opentelemetry-grpc-1.6</artifactId>
     </dependency>
+    <dependency>
+      <groupId>io.opentelemetry</groupId>
+      <artifactId>opentelemetry-sdk</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>io.opentelemetry</groupId>
+      <artifactId>opentelemetry-sdk-metrics</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>com.google.cloud.opentelemetry</groupId>
+      <artifactId>exporter-metrics</artifactId>
+    </dependency>
     <!-- END OpenTelemetry -->
 
     <!-- Test dependencies -->
@@ -181,12 +193,6 @@
       <scope>test</scope>
     </dependency>
     <!-- OpenTelemetry -->
-    <dependency>
-      <groupId>io.opentelemetry</groupId>
-      <artifactId>opentelemetry-sdk</artifactId>
-      <version>${opentelemetry.version}</version>
-      <scope>test</scope>
-    </dependency>
     <dependency>
       <groupId>io.opentelemetry</groupId>
       <artifactId>opentelemetry-sdk-testing</artifactId>
@@ -225,20 +231,6 @@
       <version>2.52.0</version>
       <scope>test</scope>
     </dependency>
-    <dependency>
-      <groupId>com.google.cloud.opentelemetry</groupId>
-      <artifactId>exporter-trace</artifactId>
-      <version>0.33.0</version>
-      <scope>test</scope>
-    </dependency>
-    <!-- END OpenTelemetry -->
-    <!-- Cloud Ops -->
-    <dependency>
-      <groupId>com.google.api.grpc</groupId>
-      <artifactId>proto-google-cloud-trace-v1</artifactId>
-      <version>2.52.0</version>
-      <scope>test</scope>
-    </dependency>
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-trace</artifactId>

google-cloud-firestore/src/main/java/com/google/cloud/firestore/AggregateQuery.java

@@ -16,8 +16,8 @@
 
 package com.google.cloud.firestore;
 
+import static com.google.cloud.firestore.telemetry.TelemetryConstants.METHOD_NAME_RUN_AGGREGATION_QUERY;
 import static com.google.cloud.firestore.telemetry.TraceUtil.ATTRIBUTE_KEY_ATTEMPT;
-import static com.google.cloud.firestore.telemetry.TraceUtil.SPAN_NAME_RUN_AGGREGATION_QUERY;
 
 import com.google.api.core.ApiFuture;
 import com.google.api.core.InternalExtensionOnly;
@@ -27,6 +27,8 @@
 import com.google.api.gax.rpc.StatusCode;
 import com.google.api.gax.rpc.StreamController;
 import com.google.cloud.Timestamp;
+import com.google.cloud.firestore.telemetry.MetricsUtil.MetricsContext;
+import com.google.cloud.firestore.telemetry.TelemetryConstants;
 import com.google.cloud.firestore.telemetry.TraceUtil;
 import com.google.cloud.firestore.telemetry.TraceUtil.Scope;
 import com.google.cloud.firestore.v1.FirestoreSettings;
@@ -70,6 +72,11 @@ private TraceUtil getTraceUtil() {
     return query.getFirestore().getOptions().getTraceUtil();
   }
 
+  @Nonnull
+  private MetricsContext createMetricsContext(String method) {
+    return query.getFirestore().getOptions().getMetricsUtil().createMetricsContext(method);
+  }
+
   /** Returns the query whose aggregations will be calculated by this object. */
   @Nonnull
   public Query getQuery() {
@@ -96,14 +103,20 @@ public ApiFuture<AggregateQuerySnapshot> get() {
    */
   @Nonnull
   public ApiFuture<ExplainResults<AggregateQuerySnapshot>> explain(ExplainOptions options) {
-    TraceUtil.Span span = getTraceUtil().startSpan(TraceUtil.SPAN_NAME_AGGREGATION_QUERY_GET);
+    TraceUtil.Span span =
+        getTraceUtil().startSpan(TelemetryConstants.METHOD_NAME_AGGREGATION_QUERY_GET);
+
+    MetricsContext metricsContext =
+        createMetricsContext(TelemetryConstants.METHOD_NAME_RUN_AGGREGATION_QUERY_EXPLAIN);
+
     try (Scope ignored = span.makeCurrent()) {
       AggregateQueryExplainResponseDeliverer responseDeliverer =
           new AggregateQueryExplainResponseDeliverer(
               /* transactionId= */ null,
               /* readTime= */ null,
               /* startTimeNanos= */ query.rpcContext.getClock().nanoTime(),
-              /* explainOptions= */ options);
+              /* explainOptions= */ options,
+              metricsContext);
       runQuery(responseDeliverer, /* attempt */ 0);
       ApiFuture<ExplainResults<AggregateQuerySnapshot>> result = responseDeliverer.getFuture();
       span.endAtFuture(result);
@@ -121,14 +134,22 @@ ApiFuture<AggregateQuerySnapshot> get(
         getTraceUtil()
             .startSpan(
                 transactionId == null
-                    ? TraceUtil.SPAN_NAME_AGGREGATION_QUERY_GET
-                    : TraceUtil.SPAN_NAME_TRANSACTION_GET_AGGREGATION_QUERY);
+                    ? TelemetryConstants.METHOD_NAME_AGGREGATION_QUERY_GET
+                    : TelemetryConstants.METHOD_NAME_TRANSACTION_GET_AGGREGATION_QUERY);
+
+    MetricsContext metricsContext =
+        createMetricsContext(
+            transactionId == null
+                ? TelemetryConstants.METHOD_NAME_RUN_AGGREGATION_QUERY_GET
+                : TelemetryConstants.METHOD_NAME_RUN_AGGREGATION_QUERY_TRANSACTIONAL);
+
     try (Scope ignored = span.makeCurrent()) {
       AggregateQueryResponseDeliverer responseDeliverer =
           new AggregateQueryResponseDeliverer(
               transactionId,
               readTime,
-              /* startTimeNanos= */ query.rpcContext.getClock().nanoTime());
+              /* startTimeNanos= */ query.rpcContext.getClock().nanoTime(),
+              metricsContext);
       runQuery(responseDeliverer, /* attempt= */ 0);
       ApiFuture<AggregateQuerySnapshot> result = responseDeliverer.getFuture();
       span.endAtFuture(result);
@@ -165,14 +186,17 @@ private abstract static class ResponseDeliverer<T> {
     private final @Nullable com.google.protobuf.Timestamp readTime;
     private final long startTimeNanos;
     private final SettableApiFuture<T> future = SettableApiFuture.create();
+    private MetricsContext metricsContext;
 
     ResponseDeliverer(
         @Nullable ByteString transactionId,
         @Nullable com.google.protobuf.Timestamp readTime,
-        long startTimeNanos) {
+        long startTimeNanos,
+        MetricsContext metricsContext) {
       this.transactionId = transactionId;
       this.readTime = readTime;
       this.startTimeNanos = startTimeNanos;
+      this.metricsContext = metricsContext;
     }
 
     @Nullable
@@ -198,15 +222,29 @@ ApiFuture<T> getFuture() {
       return future;
     }
 
-    protected void setFuture(T value) {
-      future.set(value);
+    void deliverFirstResponse() {
+      metricsContext.recordFirstResponseLatency();
     }
 
     void deliverError(Throwable throwable) {
       future.setException(throwable);
+      metricsContext.recordEndToEndLatency(throwable);
+    }
+
+    void deliverResult(
+        @Nullable Map<String, Value> serverData,
+        Timestamp readTime,
+        @Nullable ExplainMetrics metrics) {
+      try {
+        T result = processResult(serverData, readTime, metrics);
+        future.set(result);
+        metricsContext.recordEndToEndLatency();
+      } catch (Exception error) {
+        deliverError(error);
+      }
     }
 
-    abstract void deliverResult(
+    abstract T processResult(
         @Nullable Map<String, Value> serverData,
         Timestamp readTime,
         @Nullable ExplainMetrics metrics);
@@ -216,24 +254,23 @@ private class AggregateQueryResponseDeliverer extends ResponseDeliverer<Aggregat
     AggregateQueryResponseDeliverer(
         @Nullable ByteString transactionId,
         @Nullable com.google.protobuf.Timestamp readTime,
-        long startTimeNanos) {
-      super(transactionId, readTime, startTimeNanos);
+        long startTimeNanos,
+        MetricsContext metricsContext) {
+      super(transactionId, readTime, startTimeNanos, metricsContext);
     }
 
     @Override
-    void deliverResult(
+    AggregateQuerySnapshot processResult(
         @Nullable Map<String, Value> serverData,
         Timestamp readTime,
         @Nullable ExplainMetrics metrics) {
       if (serverData == null) {
-        deliverError(new RuntimeException("Did not receive any aggregate query results."));
-        return;
+        throw new RuntimeException("Did not receive any aggregate query results.");
       }
-      setFuture(
-          new AggregateQuerySnapshot(
-              AggregateQuery.this,
-              readTime,
-              convertServerAggregateFieldsMapToClientAggregateFieldsMap(serverData)));
+      return new AggregateQuerySnapshot(
+          AggregateQuery.this,
+          readTime,
+          convertServerAggregateFieldsMapToClientAggregateFieldsMap(serverData));
     }
   }
 
@@ -245,8 +282,9 @@ private final class AggregateQueryExplainResponseDeliverer
         @Nullable ByteString transactionId,
         @Nullable com.google.protobuf.Timestamp readTime,
         long startTimeNanos,
-        @Nullable ExplainOptions explainOptions) {
-      super(transactionId, readTime, startTimeNanos);
+        @Nullable ExplainOptions explainOptions,
+        MetricsContext metricsContext) {
+      super(transactionId, readTime, startTimeNanos, metricsContext);
       this.explainOptions = explainOptions;
     }
 
@@ -257,14 +295,13 @@ ExplainOptions getExplainOptions() {
     }
 
     @Override
-    void deliverResult(
+    ExplainResults<AggregateQuerySnapshot> processResult(
         @Nullable Map<String, Value> serverData,
         Timestamp readTime,
         @Nullable ExplainMetrics metrics) {
       // The server is required to provide ExplainMetrics for explain queries.
       if (metrics == null) {
-        deliverError(new RuntimeException("Did not receive any metrics for explain query."));
-        return;
+        throw new RuntimeException("Did not receive any metrics for explain query.");
       }
       AggregateQuerySnapshot snapshot =
           serverData == null
@@ -273,7 +310,7 @@ void deliverResult(
                   AggregateQuery.this,
                   readTime,
                   convertServerAggregateFieldsMapToClientAggregateFieldsMap(serverData));
-      setFuture(new ExplainResults<>(metrics, snapshot));
+      return new ExplainResults<>(metrics, snapshot);
     }
   }
 
@@ -284,6 +321,7 @@ private final class AggregateQueryResponseObserver<T>
     @Nullable private Map<String, Value> aggregateFieldsMap = null;
     @Nullable private ExplainMetrics metrics = null;
     private int attempt;
+    private boolean firstResponse = false;
 
     AggregateQueryResponseObserver(ResponseDeliverer<T> responseDeliverer, int attempt) {
       this.responseDeliverer = responseDeliverer;
@@ -302,15 +340,20 @@ private boolean isExplainQuery() {
     public void onStart(StreamController streamController) {
       getTraceUtil()
           .currentSpan()
-          .addEvent(SPAN_NAME_RUN_AGGREGATION_QUERY + " Stream started.", getAttemptAttributes());
+          .addEvent(METHOD_NAME_RUN_AGGREGATION_QUERY + " Stream started.", getAttemptAttributes());
     }
 
     @Override
     public void onResponse(RunAggregationQueryResponse response) {
+      if (!firstResponse) {
+        firstResponse = true;
+        responseDeliverer.deliverFirstResponse();
+      }
+
       getTraceUtil()
           .currentSpan()
           .addEvent(
-              SPAN_NAME_RUN_AGGREGATION_QUERY + " Response Received.", getAttemptAttributes());
+              METHOD_NAME_RUN_AGGREGATION_QUERY + " Response Received.", getAttemptAttributes());
       if (response.hasReadTime()) {
         readTime = Timestamp.fromProto(response.getReadTime());
       }
@@ -339,15 +382,14 @@ public void onError(Throwable throwable) {
         getTraceUtil()
             .currentSpan()
             .addEvent(
-                SPAN_NAME_RUN_AGGREGATION_QUERY + ": Retryable Error",
+                METHOD_NAME_RUN_AGGREGATION_QUERY + ": Retryable Error",
                 Collections.singletonMap("error.message", throwable.getMessage()));
-
         runQuery(responseDeliverer, attempt + 1);
       } else {
         getTraceUtil()
             .currentSpan()
             .addEvent(
-                SPAN_NAME_RUN_AGGREGATION_QUERY + ": Error",
+                METHOD_NAME_RUN_AGGREGATION_QUERY + ": Error",
                 Collections.singletonMap("error.message", throwable.getMessage()));
         responseDeliverer.deliverError(throwable);
       }

google-cloud-firestore/src/main/java/com/google/cloud/firestore/BulkWriter.java

@@ -26,6 +26,8 @@
 import com.google.api.core.SettableApiFuture;
 import com.google.api.gax.rpc.ApiException;
 import com.google.api.gax.rpc.StatusCode.Code;
+import com.google.cloud.firestore.telemetry.MetricsUtil.MetricsContext;
+import com.google.cloud.firestore.telemetry.TelemetryConstants;
 import com.google.cloud.firestore.telemetry.TraceUtil;
 import com.google.cloud.firestore.telemetry.TraceUtil.Context;
 import com.google.cloud.firestore.telemetry.TraceUtil.Scope;
@@ -912,8 +914,15 @@ private void sendBatchLocked(final BulkCommitBatch batch, final boolean flush) {
           firestore
               .getOptions()
               .getTraceUtil()
-              .startSpan(TraceUtil.SPAN_NAME_BULK_WRITER_COMMIT, traceContext)
+              .startSpan(TelemetryConstants.METHOD_NAME_BULK_WRITER_COMMIT, traceContext)
               .setAttribute(ATTRIBUTE_KEY_DOC_COUNT, batch.getMutationsSize());
+
+      MetricsContext metricsContext =
+          firestore
+              .getOptions()
+              .getMetricsUtil()
+              .createMetricsContext(TelemetryConstants.METHOD_NAME_BULK_WRITER_COMMIT);
+
       try (Scope ignored = span.makeCurrent()) {
         ApiFuture<Void> result = batch.bulkCommit();
         result.addListener(
@@ -926,8 +935,10 @@ private void sendBatchLocked(final BulkCommitBatch batch, final boolean flush) {
             },
             bulkWriterExecutor);
         span.endAtFuture(result);
+        metricsContext.recordEndToEndLatencyAtFuture(result);
       } catch (Exception error) {
         span.end(error);
+        metricsContext.recordEndToEndLatency(error);
         throw error;
       }
     } else {