Mila/client side metrics demo2

README.md

@@ -19,7 +19,7 @@ If you are using Maven with [BOM][libraries-bom], add this to your pom.xml file:
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>libraries-bom</artifactId>
-      <version>26.47.0</version>
+      <version>26.48.0</version>
       <type>pom</type>
       <scope>import</scope>
     </dependency>
@@ -42,7 +42,7 @@ If you are using Maven without the BOM, add this to your dependencies:
 <dependency>
   <groupId>com.google.cloud</groupId>
   <artifactId>google-cloud-firestore</artifactId>
-  <version>3.26.5</version>
+  <version>3.27.0</version>
 </dependency>
 
 ```

google-cloud-firestore/pom.xml

@@ -119,6 +119,15 @@
       <groupId>io.opentelemetry.instrumentation</groupId>
       <artifactId>opentelemetry-grpc-1.6</artifactId>
     </dependency>
+    <dependency>
+      <groupId>io.opentelemetry</groupId>
+      <artifactId>opentelemetry-sdk</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>com.google.cloud.opentelemetry</groupId>
+      <artifactId>exporter-metrics</artifactId>
+      <version>0.31.0</version>
+    </dependency>
     <!-- END OpenTelemetry -->
 
     <!-- Test dependencies -->
@@ -181,12 +190,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>

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,9 @@
 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;
+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 +73,11 @@ private TraceUtil getTraceUtil() {
     return query.getFirestore().getOptions().getTraceUtil();
   }
 
+  @Nonnull
+  private MetricsUtil getMetricsUtil() {
+    return query.getFirestore().getOptions().getMetricsUtil();
+  }
+
   /** Returns the query whose aggregations will be calculated by this object. */
   @Nonnull
   public Query getQuery() {
@@ -96,7 +104,14 @@ 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 =
+        getMetricsUtil()
+            .createMetricsContext(
+                TelemetryConstants.METHOD_NAME_RUN_QUERY_EXPLAIN_AGGREGATION_QUERY);
+
     try (Scope ignored = span.makeCurrent()) {
       AggregateQueryExplainResponseDeliverer responseDeliverer =
           new AggregateQueryExplainResponseDeliverer(
@@ -106,10 +121,14 @@ public ApiFuture<ExplainResults<AggregateQuerySnapshot>> explain(ExplainOptions
               /* explainOptions= */ options);
       runQuery(responseDeliverer, /* attempt */ 0);
       ApiFuture<ExplainResults<AggregateQuerySnapshot>> result = responseDeliverer.getFuture();
+
       span.endAtFuture(result);
+      metricsContext.recordEndToEndLatencyAtFuture(result);
+
       return result;
     } catch (Exception error) {
       span.end(error);
+      metricsContext.recordEndToEndLatency(error);
       throw error;
     }
   }
@@ -121,8 +140,16 @@ 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 =
+        getMetricsUtil()
+            .createMetricsContext(
+                transactionId == null
+                    ? TelemetryConstants.METHOD_NAME_AGGREGATION_QUERY_GET
+                    : TelemetryConstants.METHOD_NAME_TRANSACTION_GET_AGGREGATION_QUERY);
+
     try (Scope ignored = span.makeCurrent()) {
       AggregateQueryResponseDeliverer responseDeliverer =
           new AggregateQueryResponseDeliverer(
@@ -132,9 +159,11 @@ ApiFuture<AggregateQuerySnapshot> get(
       runQuery(responseDeliverer, /* attempt= */ 0);
       ApiFuture<AggregateQuerySnapshot> result = responseDeliverer.getFuture();
       span.endAtFuture(result);
+      metricsContext.recordEndToEndLatencyAtFuture(result);
       return result;
     } catch (Exception error) {
       span.end(error);
+      metricsContext.recordEndToEndLatency(error);
       throw error;
     }
   }
@@ -302,15 +331,15 @@ 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) {
       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 +368,15 @@ 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,11 @@ 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 {

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

@@ -21,6 +21,8 @@
 import com.google.api.gax.rpc.ApiException;
 import com.google.api.gax.rpc.ApiExceptions;
 import com.google.api.gax.rpc.ApiStreamObserver;
+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.FirestoreClient.PartitionQueryPagedResponse;
@@ -109,7 +111,15 @@ public ApiFuture<List<QueryPartition>> getPartitions(long desiredPartitionCount)
               .getFirestore()
               .getOptions()
               .getTraceUtil()
-              .startSpan(TraceUtil.SPAN_NAME_PARTITION_QUERY);
+              .startSpan(TelemetryConstants.METHOD_NAME_PARTITION_QUERY);
+
+      MetricsContext metricsContext =
+          rpcContext
+              .getFirestore()
+              .getOptions()
+              .getMetricsUtil()
+              .createMetricsContext(TelemetryConstants.METHOD_NAME_PARTITION_QUERY);
+
       try (Scope ignored = span.makeCurrent()) {
         ApiFuture<List<QueryPartition>> result =
             ApiFutures.transform(
@@ -127,12 +137,18 @@ public ApiFuture<List<QueryPartition>> getPartitions(long desiredPartitionCount)
                 },
                 MoreExecutors.directExecutor());
         span.endAtFuture(result);
+        metricsContext.recordEndToEndLatencyAtFuture(result);
+
         return result;
       } catch (ApiException exception) {
         span.end(exception);
+        metricsContext.recordEndToEndLatency(exception);
+
         throw FirestoreException.forApiException(exception);
       } catch (Throwable throwable) {
         span.end(throwable);
+        metricsContext.recordEndToEndLatency(throwable);
+
         throw throwable;
       }
     }

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

@@ -24,6 +24,8 @@
 import com.google.api.gax.rpc.UnaryCallable;
 import com.google.cloud.firestore.encoding.CustomClassMapper;
 import com.google.cloud.firestore.spi.v1.FirestoreRpc;
+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.FirestoreClient.ListDocumentsPagedResponse;
@@ -134,7 +136,15 @@ public Iterable<DocumentReference> listDocuments() {
             .getFirestore()
             .getOptions()
             .getTraceUtil()
-            .startSpan(TraceUtil.SPAN_NAME_COL_REF_LIST_DOCUMENTS);
+            .startSpan(TelemetryConstants.METHOD_NAME_COL_REF_LIST_DOCUMENTS);
+
+    MetricsContext metricsContext =
+        rpcContext
+            .getFirestore()
+            .getOptions()
+            .getMetricsUtil()
+            .createMetricsContext(TelemetryConstants.METHOD_NAME_COL_REF_LIST_DOCUMENTS);
+
     try (Scope ignored = span.makeCurrent()) {
       ListDocumentsRequest.Builder request = ListDocumentsRequest.newBuilder();
       request.setParent(options.getParentPath().toString());
@@ -174,12 +184,18 @@ public void remove() {
             }
           };
       span.end();
+      metricsContext.recordEndToEndLatency();
+
       return result;
     } catch (ApiException exception) {
       span.end(exception);
+      metricsContext.recordEndToEndLatency(exception);
+
       throw FirestoreException.forApiException(exception);
     } catch (Throwable throwable) {
       span.end(throwable);
+      metricsContext.recordEndToEndLatency(throwable);
+
       throw throwable;
     }
   }
@@ -200,17 +216,28 @@ public ApiFuture<DocumentReference> add(@Nonnull final Map<String, Object> field
             .getFirestore()
             .getOptions()
             .getTraceUtil()
-            .startSpan(TraceUtil.SPAN_NAME_COL_REF_ADD);
+            .startSpan(TelemetryConstants.METHOD_NAME_COL_REF_ADD);
+
+    MetricsContext metricsContext =
+        rpcContext
+            .getFirestore()
+            .getOptions()
+            .getMetricsUtil()
+            .createMetricsContext(TelemetryConstants.METHOD_NAME_COL_REF_ADD);
+
     try (Scope ignored = span.makeCurrent()) {
       final DocumentReference documentReference = document();
       ApiFuture<WriteResult> createFuture = documentReference.create(fields);
       ApiFuture<DocumentReference> result =
           ApiFutures.transform(
               createFuture, writeResult -> documentReference, MoreExecutors.directExecutor());
       span.endAtFuture(result);
+      metricsContext.recordEndToEndLatencyAtFuture(result);
+
       return result;
     } catch (Exception error) {
       span.end(error);
+      metricsContext.recordEndToEndLatency(error);
       throw error;
     }
   }

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

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2024 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.firestore;
+
+import com.google.api.gax.tracing.ApiTracer;
+import com.google.api.gax.tracing.ApiTracerFactory;
+import com.google.api.gax.tracing.BaseApiTracerFactory;
+import com.google.api.gax.tracing.SpanName;
+import com.google.common.collect.ImmutableList;
+import java.util.ArrayList;
+import java.util.List;
+
+class CompositeApiTracerFactory extends BaseApiTracerFactory {
+
+  private final List<ApiTracerFactory> apiTracerFactories;
+
+  public CompositeApiTracerFactory(List<ApiTracerFactory> apiTracerFactories) {
+    this.apiTracerFactories = ImmutableList.copyOf(apiTracerFactories);
+  }
+
+  @Override
+  public ApiTracer newTracer(ApiTracer parent, SpanName spanName, OperationType operationType) {
+    List<ApiTracer> children = new ArrayList<>(apiTracerFactories.size());
+
+    for (ApiTracerFactory factory : apiTracerFactories) {
+      children.add(factory.newTracer(parent, spanName, operationType));
+    }
+    return new CompositeTracer(children);
+  }
+}