request_id

Author: surbhigarg92

google-cloud-spanner/src/main/java/com/google/cloud/spanner/BuiltInMetricsConstant.java

@@ -16,6 +16,8 @@
 
 package com.google.cloud.spanner;
 
+import static com.google.cloud.spanner.XGoogSpannerRequestId.REQUEST_ID;
+
 import com.google.api.core.InternalApi;
 import com.google.api.gax.tracing.OpenTelemetryMetricsRecorder;
 import com.google.common.collect.ImmutableList;
@@ -26,7 +28,9 @@
 import io.opentelemetry.sdk.metrics.InstrumentSelector;
 import io.opentelemetry.sdk.metrics.InstrumentType;
 import io.opentelemetry.sdk.metrics.View;
+import java.util.Arrays;
 import java.util.Collection;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -94,6 +98,10 @@ public class BuiltInMetricsConstant {
       AttributeKey.stringKey("directpath_enabled");
   public static final AttributeKey<String> DIRECT_PATH_USED_KEY =
       AttributeKey.stringKey("directpath_used");
+  public static final AttributeKey<String> REQUEST_ID_KEY =
+      AttributeKey.stringKey(REQUEST_ID);
+  public static Set<String> ALLOWED_EXEMPLARS_ATTRIBUTES = new HashSet<>(
+      Arrays.asList(REQUEST_ID));
 
   // IP address prefixes allocated for DirectPath backends.
   public static final String DP_IPV6_PREFIX = "2001:4860:8040";

google-cloud-spanner/src/main/java/com/google/cloud/spanner/BuiltInMetricsTracer.java

@@ -64,7 +64,7 @@ class BuiltInMetricsTracer extends MetricsTracer implements ApiTracer {
     // filtered_attributes.
     // Below testmetric attribute will be available in exemplar as we have added a attributefilter
     // for our metric views.
-    this.attributes.put("request_id", "test");
+    // this.attributes.put("request_id", "test");
     this.traceWrapper = traceWrapper;
     this.currentSpan = currentSpan;
   }

google-cloud-spanner/src/main/java/com/google/cloud/spanner/SpannerCloudMonitoringExporterUtils.java

@@ -22,6 +22,7 @@
 import static com.google.api.MetricDescriptor.ValueType.DISTRIBUTION;
 import static com.google.api.MetricDescriptor.ValueType.DOUBLE;
 import static com.google.api.MetricDescriptor.ValueType.INT64;
+import static com.google.cloud.spanner.BuiltInMetricsConstant.ALLOWED_EXEMPLARS_ATTRIBUTES;
 import static com.google.cloud.spanner.BuiltInMetricsConstant.GAX_METER_NAME;
 import static com.google.cloud.spanner.BuiltInMetricsConstant.GRPC_METER_NAME;
 import static com.google.cloud.spanner.BuiltInMetricsConstant.PROJECT_ID_KEY;
@@ -294,6 +295,12 @@ private static String makeSpanName(String projectId, String traceId, String span
   private static DroppedLabels mapFilteredAttributes(Attributes attributes) {
     DroppedLabels.Builder labels = DroppedLabels.newBuilder();
     attributes.forEach((k, v) -> labels.putLabel(cleanAttributeKey(k.getKey()), v.toString()));
+    attributes.forEach((k, v) -> {
+      String key = cleanAttributeKey(k.getKey());
+      if (ALLOWED_EXEMPLARS_ATTRIBUTES.contains(key)) {
+        labels.putLabel(key, v.toString());
+      }
+    });
     return labels.build();
   }
 

google-cloud-spanner/src/main/java/com/google/cloud/spanner/XGoogSpannerRequestId.java

@@ -35,8 +35,9 @@ public class XGoogSpannerRequestId {
   @VisibleForTesting
   static final String RAND_PROCESS_ID = XGoogSpannerRequestId.generateRandProcessId();
 
+  static String REQUEST_ID = "x-goog-spanner-request-id";
   public static final Metadata.Key<String> REQUEST_HEADER_KEY =
-      Metadata.Key.of("x-goog-spanner-request-id", Metadata.ASCII_STRING_MARSHALLER);
+      Metadata.Key.of(REQUEST_ID, Metadata.ASCII_STRING_MARSHALLER);
 
   @VisibleForTesting
   static final long VERSION = 1; // The version of the specification being implemented.

google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/HeaderInterceptor.java

@@ -128,6 +128,7 @@ public void onHeaders(Metadata metadata) {
                   Boolean isDirectPathUsed =
                       isDirectPathUsed(getAttributes().get(Grpc.TRANSPORT_ATTR_REMOTE_ADDR));
                   addDirectPathUsedAttribute(compositeTracer, isDirectPathUsed);
+                  addRequestIdAttribute(compositeTracer, "test");
                   processHeader(
                       metadata, tagContext, attributes, span, compositeTracer, isDirectPathUsed);
                   super.onHeaders(metadata);
@@ -308,6 +309,14 @@ private void addDirectPathUsedAttribute(
     }
   }
 
+  private void addRequestIdAttribute(
+      CompositeTracer compositeTracer, String requestId) {
+    if (compositeTracer != null) {
+      compositeTracer.addAttributes(
+          BuiltInMetricsConstant.REQUEST_ID_KEY.getKey(), requestId);
+    }
+  }
+
   private Boolean isDirectPathUsed(SocketAddress remoteAddr) {
     if (remoteAddr instanceof InetSocketAddress) {
       InetAddress inetAddress = ((InetSocketAddress) remoteAddr).getAddress();

google-cloud-spanner/src/test/java/com/google/cloud/spanner/it/ITBuiltInMetricsTest.java

@@ -35,6 +35,13 @@
 import java.io.IOException;
 import java.time.Duration;
 import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import org.junit.After;
 import org.junit.BeforeClass;
@@ -89,12 +96,41 @@ public void testBuiltinMetricsWithDefaultOTEL() throws Exception {
             .setEndTime(Timestamps.fromMillis(end.toEpochMilli()))
             .build();
 
-    for (int i = 0; i < 100; i++) {
-      client
-          .readWriteTransaction()
-          .run(transaction -> transaction.executeQuery(Statement.of("Select 1")));
+    int totalQueries = 1;
+    int batchSize = 1;
+    ExecutorService executor = Executors.newFixedThreadPool(100);
+
+    for (int batch = 0; batch < totalQueries / batchSize; batch++) {
+      List<Callable<Void>> tasks = new ArrayList<>();
+      for (int i = 0; i < batchSize; i++) {
+        tasks.add(
+            () -> {
+              client
+                  .readWriteTransaction()
+                  .run(
+                      transaction -> {
+                        transaction.executeQuery(Statement.of("SELECT 1"));
+                        return null;
+                      });
+              return null;
+            });
+      }
+
+      List<Future<Void>> futures = executor.invokeAll(tasks);
+      for (Future<Void> future : futures) {
+        try {
+          future.get(); // propagate exceptions
+        } catch (ExecutionException e) {
+          System.err.println("Error in query: " + e.getCause());
+        }
+      }
+
+      System.out.println("Completed batch " + (batch + 1));
     }
 
+    executor.shutdown();
+    executor.awaitTermination(1, TimeUnit.HOURS);
+
     for (String metric : METRICS) {
       String metricFilter =
           String.format(