chore(x-goog-spanner-request-id): set span attribute x_goog_spanner_request_id

This sets the span attribute x_goog_spanner_request_id so as to enable
correlation with distributed tracing with OpenTelemetry.

Updates #3537

Author: odeke-em

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

@@ -198,4 +198,8 @@ XGoogSpannerRequestId withNthClientId(long replacementClientId) {
     return XGoogSpannerRequestId.of(
         replacementClientId, this.nthChannelId, this.nthRequest, this.attempt);
   }
+
+  public static String getRequestIdFromMetadata(Metadata md) {
+    return md.get(REQUEST_HEADER_KEY);
+  }
 }

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

@@ -28,6 +28,7 @@
 import com.google.cloud.spanner.CompositeTracer;
 import com.google.cloud.spanner.SpannerExceptionFactory;
 import com.google.cloud.spanner.SpannerRpcMetrics;
+import com.google.cloud.spanner.XGoogSpannerRequestId;
 import com.google.common.cache.Cache;
 import com.google.common.cache.CacheBuilder;
 import com.google.spanner.admin.database.v1.DatabaseName;
@@ -174,6 +175,10 @@ private void processHeader(
         }
         if (span != null) {
           span.setAttribute("gfe_latency", String.valueOf(gfeLatency));
+          String reqId = XGoogSpannerRequestId.getRequestIdFromMetadata(metadata);
+          if (reqId != null) {
+            span.setAttribute("x_goog_spanner_request_id", reqId);
+          }
         }
       } else {
         measureMap.put(SPANNER_GFE_HEADER_MISSING_COUNT, 1L).record(tagContext);

google-cloud-spanner/src/test/java/com/google/cloud/spanner/OpenTelemetrySpanTest.java

@@ -18,6 +18,7 @@
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 
@@ -748,6 +749,17 @@ public void testTransactionRunnerWithRetryOnBeginTransaction() {
         beginTransactionSpan.toString(),
         beginTransactionSpan.getEvents().stream()
             .anyMatch(event -> event.getName().equals("Starting RPC retry 1")));
+    verifyAtLeast1SpanHasXGoogSpannerRequestIdAttribute(finishedSpans);
+  }
+
+  private void verifyAtLeast1SpanHasXGoogSpannerRequestIdAttribute(List<SpanData> finishedSpans) {
+    AttributeKey<String> attributeKey = AttributeKey.stringKey("x_goog_spanner_request_id");
+    SpanData matchedSpan =
+        finishedSpans.stream()
+            .filter(span -> !span.getAttributes().get(attributeKey).isEmpty())
+            .findAny()
+            .orElseThrow(IllegalStateException::new);
+    assertNotNull(matchedSpan);
   }
 
   @Test
@@ -798,6 +810,7 @@ public void testSingleUseRetryOnExecuteStreamingSql() {
         executeStreamingQuery.toString(),
         executeStreamingQuery.getEvents().stream()
             .anyMatch(event -> event.getName().contains("Stream broken. Safe to retry")));
+    verifyAtLeast1SpanHasXGoogSpannerRequestIdAttribute(finishedSpans);
   }
 
   @Test
@@ -845,6 +858,7 @@ public void testRetryOnExecuteSql() {
         executeSqlSpan.toString(),
         executeSqlSpan.getEvents().stream()
             .anyMatch(event -> event.getName().equals("Starting RPC retry 1")));
+    verifyAtLeast1SpanHasXGoogSpannerRequestIdAttribute(finishedSpans);
   }
 
   @Test
@@ -866,12 +880,14 @@ public void testTableAttributes() {
               }
               return null;
             });
+    List<SpanData> finishedSpans = spanExporter.getFinishedSpanItems();
     SpanData spanData =
-        spanExporter.getFinishedSpanItems().stream()
+        finishedSpans.stream()
             .filter(x -> x.getName().equals("CloudSpannerOperation.ExecuteStreamingRead"))
             .findFirst()
             .get();
     verifyTableAttributes(spanData);
+    verifyAtLeast1SpanHasXGoogSpannerRequestIdAttribute(finishedSpans);
   }
 
   private void waitForFinishedSpans(int numExpectedSpans) {