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(