feat(bigtable): populate attempt latencies via peer info

.kokoro/presubmit/graalvm-native-a.cfg

@@ -3,7 +3,7 @@
 # Configure the docker image for kokoro-trampoline.
 env_vars: {
   key: "TRAMPOLINE_IMAGE"
-  value: "gcr.io/cloud-devrel-public-resources/graalvm_sdk_platform_a:3.53.0" # {x-version-update:google-cloud-shared-dependencies:current}
+  value: "gcr.io/cloud-devrel-public-resources/graalvm_sdk_platform_a:3.54.1" # {x-version-update:google-cloud-shared-dependencies:current}
 }
 
 env_vars: {

.kokoro/presubmit/graalvm-native-b.cfg

@@ -3,7 +3,7 @@
 # Configure the docker image for kokoro-trampoline.
 env_vars: {
   key: "TRAMPOLINE_IMAGE"
-  value: "gcr.io/cloud-devrel-public-resources/graalvm_sdk_platform_b:3.53.0" # {x-version-update:google-cloud-shared-dependencies:current}
+  value: "gcr.io/cloud-devrel-public-resources/graalvm_sdk_platform_b:3.54.1" # {x-version-update:google-cloud-shared-dependencies:current}
 }
 
 env_vars: {

.kokoro/presubmit/graalvm-native-c.cfg

@@ -3,7 +3,7 @@
 # Configure the docker image for kokoro-trampoline.
 env_vars: {
   key: "TRAMPOLINE_IMAGE"
-  value: "gcr.io/cloud-devrel-public-resources/graalvm_sdk_platform_c:3.53.0" # {x-version-update:google-cloud-shared-dependencies:current}
+  value: "gcr.io/cloud-devrel-public-resources/graalvm_sdk_platform_c:3.54.1" # {x-version-update:google-cloud-shared-dependencies:current}
 }
 
 env_vars: {

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStubSettings.java

@@ -935,7 +935,8 @@ private Builder() {
               .setReverseScans(true)
               .setLastScannedRowResponses(true)
               .setDirectAccessRequested(DIRECT_PATH_ENABLED)
-              .setTrafficDirectorEnabled(DIRECT_PATH_ENABLED);
+              .setTrafficDirectorEnabled(DIRECT_PATH_ENABLED)
+              .setPeerInfo(true);
     }
 
     private Builder(EnhancedBigtableStubSettings settings) {

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/metrics/BigtableTracer.java

@@ -91,6 +91,14 @@ public void setLocations(String zone, String cluster) {
     // noop
   }
 
+  /**
+   * Set the Bigtable transport zone, subzone and transport type so metrics can be tagged with
+   * location information. This will be called in BuiltinMetricsTracer.
+   */
+  public void setPeerInfo(String transportZone, String transportSubZone, String transportType) {
+    // noop
+  }
+
   /** Set the underlying transport used to process the attempt */
   public void setTransportAttrs(BuiltinMetricsTracer.TransportAttrs attrs) {}
 

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/metrics/BuiltinMetricsConstants.java

@@ -50,10 +50,12 @@ public class BuiltinMetricsConstants {
   static final AttributeKey<String> STATUS_KEY = AttributeKey.stringKey("status");
   static final AttributeKey<String> CLIENT_UID_KEY = AttributeKey.stringKey("client_uid");
 
-  static final AttributeKey<String> TRANSPORT_TYPE = AttributeKey.stringKey("transport_type");
-  static final AttributeKey<String> TRANSPORT_REGION = AttributeKey.stringKey("transport_region");
-  static final AttributeKey<String> TRANSPORT_ZONE = AttributeKey.stringKey("transport_zone");
-  static final AttributeKey<String> TRANSPORT_SUBZONE = AttributeKey.stringKey("transport_subzone");
+  static final AttributeKey<String> TRANSPORT_TYPE_KEY = AttributeKey.stringKey("transport_type");
+  static final AttributeKey<String> TRANSPORT_REGION_KEY =
+      AttributeKey.stringKey("transport_region");
+  static final AttributeKey<String> TRANSPORT_ZONE_KEY = AttributeKey.stringKey("transport_zone");
+  static final AttributeKey<String> TRANSPORT_SUBZONE_KEY =
+      AttributeKey.stringKey("transport_subzone");
 
   // gRPC attribute keys
   // Note that these attributes keys from transformed from
@@ -310,10 +312,10 @@ public static Map<InstrumentSelector, View> getAllViews() {
             .add(
                 STREAMING_KEY,
                 STATUS_KEY,
-                TRANSPORT_TYPE,
-                TRANSPORT_REGION,
-                TRANSPORT_ZONE,
-                TRANSPORT_SUBZONE)
+                TRANSPORT_TYPE_KEY,
+                TRANSPORT_REGION_KEY,
+                TRANSPORT_ZONE_KEY,
+                TRANSPORT_SUBZONE_KEY)
             .build());
     defineView(
         views,

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/metrics/BuiltinMetricsTracer.java

@@ -23,10 +23,9 @@
 import static com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsConstants.STATUS_KEY;
 import static com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsConstants.STREAMING_KEY;
 import static com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsConstants.TABLE_ID_KEY;
-import static com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsConstants.TRANSPORT_REGION;
-import static com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsConstants.TRANSPORT_SUBZONE;
-import static com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsConstants.TRANSPORT_TYPE;
-import static com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsConstants.TRANSPORT_ZONE;
+import static com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsConstants.TRANSPORT_SUBZONE_KEY;
+import static com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsConstants.TRANSPORT_TYPE_KEY;
+import static com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsConstants.TRANSPORT_ZONE_KEY;
 import static com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsConstants.ZONE_ID_KEY;
 
 import com.google.api.core.ObsoleteApi;
@@ -35,9 +34,7 @@
 import com.google.auto.value.AutoValue;
 import com.google.cloud.bigtable.Version;
 import com.google.common.base.Stopwatch;
-import com.google.common.base.Strings;
 import com.google.common.math.IntMath;
-import com.google.gson.Gson;
 import com.google.gson.reflect.TypeToken;
 import io.grpc.Deadline;
 import io.opentelemetry.api.common.Attributes;
@@ -50,7 +47,6 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
-import java.util.logging.Level;
 import java.util.logging.Logger;
 import javax.annotation.Nullable;
 
@@ -73,7 +69,6 @@ static TransportAttrs create(@Nullable String locality, @Nullable String backend
   }
 
   private static final Logger logger = Logger.getLogger(BuiltinMetricsTracer.class.getName());
-  private static final Gson GSON = new Gson();
   private static final TypeToken<Map<String, String>> LOCALITY_TYPE =
       new TypeToken<Map<String, String>>() {};
 
@@ -110,6 +105,11 @@ static TransportAttrs create(@Nullable String locality, @Nullable String backend
   private String zone = "global";
   private String cluster = "<unspecified>";
 
+  // peer info fields
+  private String transportZone = "global";
+  private String transportSubzone = "global";
+  private String transportType = "";
+
   private final AtomicLong totalClientBlockingTime = new AtomicLong(0);
 
   private final Attributes baseAttributes;
@@ -331,6 +331,13 @@ public void setLocations(String zone, String cluster) {
     this.cluster = cluster;
   }
 
+  @Override
+  public void setPeerInfo(String transportZone, String transportSubZone, String transportType) {
+    this.transportZone = transportZone;
+    this.transportType = transportType;
+    this.transportSubzone = transportSubZone;
+  }
+
   @Override
   public void setTransportAttrs(TransportAttrs attrs) {
     this.transportAttrs = attrs;
@@ -449,33 +456,13 @@ private void recordAttemptCompletion(@Nullable Throwable status) {
 
     attemptLatenciesHistogram.record(
         convertToMs(attemptTimer.elapsed(TimeUnit.NANOSECONDS)), attributes);
-
-    String transportType = "cloudpath";
-    String transportRegion = "";
-    String transportZone = "";
-    String transportSubzone = "";
-
-    try {
-      if (transportAttrs != null && !Strings.isNullOrEmpty(transportAttrs.getLocality())) {
-        // only directpath has locality
-        transportType = "directpath";
-        Map<String, String> localityMap =
-            GSON.fromJson(transportAttrs.getLocality(), LOCALITY_TYPE);
-        transportRegion = localityMap.getOrDefault("region", "");
-        transportZone = localityMap.getOrDefault("zone", "");
-        transportSubzone = localityMap.getOrDefault("sub_zone", "");
-      }
-    } catch (RuntimeException e) {
-      logger.log(
-          Level.WARNING, "Failed to parse transport locality: " + transportAttrs.getLocality(), e);
-    }
+    // TODO: move this to attemptLatencies
     attemptLatencies2Histogram.record(
         convertToMs(attemptTimer.elapsed(TimeUnit.NANOSECONDS)),
         attributes.toBuilder()
-            .put(TRANSPORT_TYPE, transportType)
-            .put(TRANSPORT_REGION, transportRegion)
-            .put(TRANSPORT_ZONE, transportZone)
-            .put(TRANSPORT_SUBZONE, transportSubzone)
+            .put(TRANSPORT_TYPE_KEY, transportType)
+            .put(TRANSPORT_ZONE_KEY, transportZone)
+            .put(TRANSPORT_SUBZONE_KEY, transportSubzone)
             .build());
 
     // When operationDeadline is set, it's possible that the deadline is passed by the time we send

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/metrics/CompositeTracer.java

@@ -225,6 +225,13 @@ public void setTransportAttrs(BuiltinMetricsTracer.TransportAttrs attrs) {
     }
   }
 
+  @Override
+  public void setPeerInfo(String transportZone, String transportSubZone, String transportType) {
+    for (BigtableTracer tracer : bigtableTracers) {
+      tracer.setPeerInfo(transportZone, transportSubZone, transportType);
+    }
+  }
+
   @Override
   public void onRequest(int requestCount) {
     for (BigtableTracer tracer : bigtableTracers) {

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/metrics/Util.java

@@ -29,6 +29,7 @@
 import com.google.bigtable.v2.MaterializedViewName;
 import com.google.bigtable.v2.MutateRowRequest;
 import com.google.bigtable.v2.MutateRowsRequest;
+import com.google.bigtable.v2.PeerInfo;
 import com.google.bigtable.v2.ReadChangeStreamRequest;
 import com.google.bigtable.v2.ReadModifyWriteRowRequest;
 import com.google.bigtable.v2.ReadRowsRequest;
@@ -55,6 +56,7 @@
 import java.time.Instant;
 import java.time.temporal.ChronoUnit;
 import java.util.Arrays;
+import java.util.Base64;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CancellationException;
@@ -78,6 +80,9 @@ public class Util {
   static final Metadata.Key<byte[]> LOCATION_METADATA_KEY =
       Metadata.Key.of("x-goog-ext-425905942-bin", Metadata.BINARY_BYTE_MARSHALLER);
 
+  static final Metadata.Key<String> BIGTABLE_PEER_INFO_KEY =
+      Metadata.Key.of("bigtable-peer-info", Metadata.ASCII_STRING_MARSHALLER);
+
   /** Convert an exception into a value that can be used to create an OpenCensus tag value. */
   static String extractStatus(@Nullable Throwable error) {
     final String statusString;
@@ -208,13 +213,29 @@ private static ResponseParams getResponseParams(@Nullable Metadata metadata) {
     return null;
   }
 
+  // fetch the PeerInfo from grpc initial metadata
+  private static PeerInfo getPeerInfo(@Nullable Metadata metadata) {
+    if (metadata == null) {
+      return null;
+    }
+    String encodedStr = metadata.get(Util.BIGTABLE_PEER_INFO_KEY);
+    if (encodedStr != null) {
+      try {
+        byte[] decoded = Base64.getUrlDecoder().decode(encodedStr);
+        return PeerInfo.parseFrom(decoded);
+      } catch (InvalidProtocolBufferException e) {
+      }
+    }
+    return null;
+  }
+
   static void recordMetricsFromMetadata(
       GrpcResponseMetadata responseMetadata, BigtableTracer tracer, Throwable throwable) {
-    Metadata metadata = responseMetadata.getMetadata();
+    Metadata initialMetadata = responseMetadata.getMetadata();
 
-    // Get the response params from the metadata. Check both headers and trailers
+    // Get the response params from the initial metadata. Check both headers and trailers
     // because in different environments the metadata could be returned in headers or trailers
-    @Nullable ResponseParams responseParams = getResponseParams(responseMetadata.getMetadata());
+    @Nullable ResponseParams responseParams = getResponseParams(initialMetadata);
     if (responseParams == null) {
       responseParams = getResponseParams(responseMetadata.getTrailingMetadata());
     }
@@ -223,9 +244,19 @@ static void recordMetricsFromMetadata(
       tracer.setLocations(responseParams.getZoneId(), responseParams.getClusterId());
     }
 
+    @Nullable PeerInfo peerInfo = getPeerInfo(initialMetadata);
+
+    // only fetch from grpc_initial_metadata
+    if (peerInfo != null) {
+      tracer.setPeerInfo(
+          peerInfo.getApplicationFrontendZone(),
+          peerInfo.getApplicationFrontendSubzone(),
+          peerInfo.getTransportType().toString());
+    }
+
     // server-timing metric will be added through GrpcResponseMetadata#onHeaders(Metadata),
     // so it's not checking trailing metadata here.
-    @Nullable Long latency = getGfeLatency(metadata);
+    @Nullable Long latency = getGfeLatency(initialMetadata);
     // For direct path, we won't see GFE server-timing header. However, if we received the
     // location info, we know that there isn't a connectivity issue. Set the latency to
     // 0 so gfe missing header won't get incremented.

google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/stub/metrics/BuiltinMetricsTracerTest.java

@@ -16,6 +16,7 @@
 package com.google.cloud.bigtable.data.v2.stub.metrics;
 
 import static com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsConstants.APPLICATION_BLOCKING_LATENCIES_NAME;
+import static com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsConstants.ATTEMPT_LATENCIES2_NAME;
 import static com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsConstants.ATTEMPT_LATENCIES_NAME;
 import static com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsConstants.CLIENT_BLOCKING_LATENCIES_NAME;
 import static com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsConstants.CLIENT_NAME_KEY;
@@ -30,6 +31,9 @@
 import static com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsConstants.STATUS_KEY;
 import static com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsConstants.STREAMING_KEY;
 import static com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsConstants.TABLE_ID_KEY;
+import static com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsConstants.TRANSPORT_SUBZONE_KEY;
+import static com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsConstants.TRANSPORT_TYPE_KEY;
+import static com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsConstants.TRANSPORT_ZONE_KEY;
 import static com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsConstants.ZONE_ID_KEY;
 import static com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsTestUtils.getAggregatedValue;
 import static com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsTestUtils.getMetricData;
@@ -55,6 +59,7 @@
 import com.google.bigtable.v2.MutateRowResponse;
 import com.google.bigtable.v2.MutateRowsRequest;
 import com.google.bigtable.v2.MutateRowsResponse;
+import com.google.bigtable.v2.PeerInfo;
 import com.google.bigtable.v2.ReadRowsRequest;
 import com.google.bigtable.v2.ReadRowsResponse;
 import com.google.bigtable.v2.ResponseParams;
@@ -104,6 +109,7 @@
 import java.time.Instant;
 import java.time.temporal.ChronoUnit;
 import java.util.ArrayList;
+import java.util.Base64;
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
@@ -134,6 +140,10 @@ public class BuiltinMetricsTracerTest {
   private static final String FIRST_RESPONSE_TABLE_ID = "first-response";
   private static final String ZONE = "us-west-1";
   private static final String CLUSTER = "cluster-0";
+  private static final String TRANSPORT_ZONE = "us-west-1-a";
+  private static final String TRANSPORT_SUBZONE = "xy";
+  private static final PeerInfo.TransportType TRANSPORT_TYPE =
+      PeerInfo.TransportType.TRANSPORT_TYPE_CLOUD_PATH;
   private static final long FAKE_SERVER_TIMING = 50;
   private static final long SERVER_LATENCY = 100;
   private static final long APPLICATION_LATENCY = 200;
@@ -200,6 +210,15 @@ public void sendHeaders(Metadata headers) {
                     byte[] byteArray = params.toByteArray();
                     headers.put(Util.LOCATION_METADATA_KEY, byteArray);
 
+                    PeerInfo peerInfo =
+                        PeerInfo.newBuilder()
+                            .setTransportType(TRANSPORT_TYPE)
+                            .setApplicationFrontendZone(TRANSPORT_ZONE)
+                            .setApplicationFrontendSubzone(TRANSPORT_SUBZONE)
+                            .build();
+
+                    String encoded = Base64.getUrlEncoder().encodeToString(peerInfo.toByteArray());
+                    headers.put(Util.BIGTABLE_PEER_INFO_KEY, encoded);
                     super.sendHeaders(headers);
                   }
                 },
@@ -303,6 +322,48 @@ public void testReadRowsOperationLatencies() {
     assertThat(value).isIn(Range.closed(SERVER_LATENCY, elapsed));
   }
 
+  @Test
+  public void testReadRowsAttemptLatencies2() {
+    Lists.newArrayList(stub.readRowsCallable().call(Query.create(TABLE)).iterator());
+
+    // Common attributes expected for ATTEMPT_LATENCIES2_NAME, assuming it
+    // does not get zone/cluster from response headers.
+    Attributes expectedAttributes1 =
+        baseAttributes.toBuilder()
+            .put(TABLE_ID_KEY, TABLE)
+            .put(ZONE_ID_KEY, "global") // Default value
+            .put(CLUSTER_ID_KEY, "<unspecified>") // Default value
+            .put(METHOD_KEY, "Bigtable.ReadRows")
+            .put(STREAMING_KEY, true)
+            .put(STATUS_KEY, "UNAVAILABLE")
+            .put(TRANSPORT_SUBZONE_KEY, "global")
+            .put(TRANSPORT_ZONE_KEY, "global")
+            .put(TRANSPORT_TYPE_KEY, "")
+            .put(CLIENT_NAME_KEY, CLIENT_NAME)
+            .build();
+
+    Attributes expectedAttributes2 =
+        baseAttributes.toBuilder()
+            .put(TABLE_ID_KEY, TABLE)
+            .put(ZONE_ID_KEY, ZONE) // Default value
+            .put(CLUSTER_ID_KEY, CLUSTER) // Default value
+            .put(METHOD_KEY, "Bigtable.ReadRows")
+            .put(STREAMING_KEY, true)
+            .put(STATUS_KEY, "OK")
+            .put(TRANSPORT_SUBZONE_KEY, TRANSPORT_SUBZONE)
+            .put(TRANSPORT_ZONE_KEY, TRANSPORT_ZONE)
+            .put(TRANSPORT_TYPE_KEY, TRANSPORT_TYPE.toString())
+            .put(CLIENT_NAME_KEY, CLIENT_NAME)
+            .build();
+
+    MetricData metricData = getMetricData(metricReader, ATTEMPT_LATENCIES2_NAME);
+
+    // Verify that metric data exists for both the UNAVAILABLE and OK attempts
+    // with the default zone and cluster.
+    verifyAttributes(metricData, expectedAttributes1);
+    verifyAttributes(metricData, expectedAttributes2);
+  }
+
   @Test
   public void testReadRowsOperationLatenciesOnAuthorizedView() {
     String authorizedViewId = "test-authorized-view-id";