feat: add custom attempt latency metric (#2882)

* feat: add custom attempt latency metric

* license header

Change-Id: Ia2df0e6bf15fe7bbd4327baa080c73009412924f

* address comments

Change-Id: I21f8d0f49749722d9b22b0db949da35df7672d22

* update

Change-Id: I391f6118d55885de7f1a2a695f27aff7a02eb426

* small fix

Change-Id: Ifdfc1f5845a8dd757dcaf0c36d03bd441882e1a4

Author: mutianf

google-cloud-bigtable/pom.xml

@@ -253,6 +253,12 @@
       <artifactId>proto-google-cloud-monitoring-v3</artifactId>
     </dependency>
 
+    <!-- export custom metrics to cloud console -->
+    <dependency>
+      <groupId>com.google.cloud.opentelemetry</groupId>
+      <artifactId>exporter-metrics</artifactId>
+    </dependency>
+
     <!-- Test dependencies -->
     <dependency>
       <groupId>com.google.api.grpc</groupId>

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/csm/MetricRegistry.java

@@ -27,6 +27,7 @@
 import com.google.cloud.bigtable.data.v2.internal.csm.metrics.ClientSessionOpenLatency;
 import com.google.cloud.bigtable.data.v2.internal.csm.metrics.ClientSessionUptime;
 import com.google.cloud.bigtable.data.v2.internal.csm.metrics.ClientTransportLatency;
+import com.google.cloud.bigtable.data.v2.internal.csm.metrics.CustomAttemptLatency;
 import com.google.cloud.bigtable.data.v2.internal.csm.metrics.GrpcMetric;
 import com.google.cloud.bigtable.data.v2.internal.csm.metrics.MetricWrapper;
 import com.google.cloud.bigtable.data.v2.internal.csm.metrics.PacemakerDelay;
@@ -87,6 +88,8 @@ public class MetricRegistry {
 
   final ClientChannelPoolFallbackCount channelFallbackCountMetric;
 
+  final CustomAttemptLatency customAttemptLatencyMetric;
+
   private final Map<String, MetricWrapper<?>> metrics = new HashMap<>();
   private final List<String> grpcMetricNames = new ArrayList<>();
 
@@ -117,6 +120,8 @@ public MetricRegistry() {
 
     channelFallbackCountMetric = register(new ClientChannelPoolFallbackCount());
 
+    customAttemptLatencyMetric = register(new CustomAttemptLatency());
+
     // From
     // https://github.com/grpc/grpc-java/blob/31fdb6c2268b4b1c8ba6c995ee46c58e84a831aa/rls/src/main/java/io/grpc/rls/CachingRlsLbClient.java#L138-L165
     registerGrpcMetric(
@@ -226,6 +231,8 @@ public class RecorderRegistry {
 
     public final ClientChannelPoolFallbackCount.Recorder channelFallbackCount;
 
+    public final CustomAttemptLatency.Recorder customAttemptLatency;
+
     private RecorderRegistry(Meter meter, boolean disableInternalMetrics) {
       // Public metrics
       operationLatency = operationLatencyMetric.newRecorder(meter);
@@ -260,6 +267,8 @@ private RecorderRegistry(Meter meter, boolean disableInternalMetrics) {
       pacemakerDelay = pacemakerDelayMetric.newRecorder(meter);
 
       channelFallbackCount = channelFallbackCountMetric.newRecorder(meter);
+
+      customAttemptLatency = customAttemptLatencyMetric.newRecorder(meter);
     }
   }
 }

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/csm/MetricsImpl.java

@@ -30,6 +30,7 @@
 import com.google.cloud.bigtable.data.v2.internal.csm.attributes.ClientInfo;
 import com.google.cloud.bigtable.data.v2.internal.csm.attributes.EnvInfo;
 import com.google.cloud.bigtable.data.v2.internal.csm.exporter.BigtableCloudMonitoringExporter;
+import com.google.cloud.bigtable.data.v2.internal.csm.exporter.BigtableFilteringExporter;
 import com.google.cloud.bigtable.data.v2.internal.csm.exporter.BigtablePeriodicReader;
 import com.google.cloud.bigtable.data.v2.internal.csm.opencensus.MetricsTracerFactory;
 import com.google.cloud.bigtable.data.v2.internal.csm.opencensus.RpcMeasureConstants;
@@ -52,6 +53,8 @@
 import com.google.cloud.bigtable.data.v2.internal.session.SessionPoolInfo;
 import com.google.cloud.bigtable.data.v2.internal.session.VRpcDescriptor;
 import com.google.cloud.bigtable.data.v2.stub.metrics.NoopMetricsProvider;
+import com.google.cloud.opentelemetry.metric.GoogleCloudMetricExporter;
+import com.google.cloud.opentelemetry.metric.MetricConfiguration;
 import com.google.common.base.Preconditions;
 import com.google.common.base.Splitter;
 import com.google.common.base.Suppliers;
@@ -68,17 +71,28 @@
 import io.opentelemetry.sdk.OpenTelemetrySdk;
 import io.opentelemetry.sdk.metrics.SdkMeterProvider;
 import io.opentelemetry.sdk.metrics.SdkMeterProviderBuilder;
+import io.opentelemetry.sdk.metrics.export.PeriodicMetricReader;
 import java.io.Closeable;
 import java.io.IOException;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Optional;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import javax.annotation.Nullable;
 import javax.annotation.concurrent.GuardedBy;
 
 public class MetricsImpl implements Metrics, Closeable {
+
+  public static final String CUSTOM_METRIC = "bigtable.internal.enable-custom-metric";
+
+  private static final boolean enableCustomMetric =
+      Optional.ofNullable(System.getProperty(CUSTOM_METRIC))
+          .map(Boolean::parseBoolean)
+          .orElse(false);
+
   private final ApiTracerFactory userTracerFactory;
   private final @Nullable OpenTelemetrySdk internalOtel;
   private final @Nullable MetricRegistry.RecorderRegistry internalRecorder;
@@ -196,7 +210,8 @@ public VRpcTracer newTableTracer(
     }
     ImmutableList.Builder<VRpcTracer> builder = ImmutableList.builder();
     builder.add(
-        new VRpcTracerImpl(internalRecorder, poolInfo, descriptor.getMethodInfo(), deadline));
+        new VRpcTracerImpl(
+            internalRecorder, poolInfo, descriptor.getMethodInfo(), deadline, enableCustomMetric));
     if (userRecorder != null) {
       builder.add(new VRpcTracerImpl(userRecorder, poolInfo, descriptor.getMethodInfo(), deadline));
     }
@@ -300,8 +315,28 @@ public static OpenTelemetrySdk createBuiltinOtel(
             metricsEndpoint,
             universeDomain);
 
-    meterProvider.registerMetricReader(new BigtablePeriodicReader(exporter, executor));
-
+    meterProvider.registerMetricReader(
+        new BigtablePeriodicReader(
+            new BigtableFilteringExporter(
+                exporter,
+                input -> input.getName().startsWith("bigtable.googleapis.com/internal/client")),
+            executor));
+
+    if (enableCustomMetric) {
+      // Monitored resource and project id are detected at export time
+      MetricConfiguration metricConfig =
+          MetricConfiguration.builder()
+              .setCredentials(credentials)
+              .setInstrumentationLibraryLabelsEnabled(false)
+              .build();
+      meterProvider.registerMetricReader(
+          PeriodicMetricReader.builder(
+                  new BigtableFilteringExporter(
+                      GoogleCloudMetricExporter.createWithConfiguration(metricConfig),
+                      input -> input.getName().startsWith("bigtable.custom")))
+              .setInterval(Duration.ofMinutes(1))
+              .build());
+    }
     return OpenTelemetrySdk.builder().setMeterProvider(meterProvider.build()).build();
   }
 

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/csm/attributes/Util.java

@@ -79,6 +79,10 @@ public static String formatTransportType(@Nullable PeerInfo peerInfo) {
             .orElse(TransportType.TRANSPORT_TYPE_UNKNOWN));
   }
 
+  public static long formatAfeId(@Nullable PeerInfo peerInfo) {
+    return Optional.ofNullable(peerInfo).map(PeerInfo::getApplicationFrontendId).orElse(0L);
+  }
+
   public static String transportTypeToString(TransportType transportType) {
     String label = transportTypeToStringWithoutFallback(transportType);
     if (label != null) {

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/csm/exporter/BigtableFilteringExporter.java

@@ -0,0 +1,64 @@
+/*
+ * Copyright 2026 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
+ *
+ *     https://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.bigtable.data.v2.internal.csm.exporter;
+
+import io.opentelemetry.sdk.common.CompletableResultCode;
+import io.opentelemetry.sdk.metrics.InstrumentType;
+import io.opentelemetry.sdk.metrics.data.AggregationTemporality;
+import io.opentelemetry.sdk.metrics.data.MetricData;
+import io.opentelemetry.sdk.metrics.export.MetricExporter;
+import java.util.Collection;
+import java.util.List;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+
+public class BigtableFilteringExporter implements MetricExporter {
+
+  private MetricExporter delegate;
+  private Predicate<MetricData> filter;
+
+  public BigtableFilteringExporter(MetricExporter exporter, Predicate<MetricData> filter) {
+    this.delegate = exporter;
+    this.filter = filter;
+  }
+
+  @Override
+  public CompletableResultCode export(Collection<MetricData> metrics) {
+    List<MetricData> filtered = metrics.stream().filter(filter).collect(Collectors.toList());
+    return delegate.export(filtered);
+  }
+
+  @Override
+  public CompletableResultCode flush() {
+    return delegate.flush();
+  }
+
+  @Override
+  public CompletableResultCode shutdown() {
+    return delegate.shutdown();
+  }
+
+  public void prepareForShutdown() {
+    if (delegate instanceof BigtableCloudMonitoringExporter) {
+      ((BigtableCloudMonitoringExporter) delegate).prepareForShutdown();
+    }
+  }
+
+  @Override
+  public AggregationTemporality getAggregationTemporality(InstrumentType instrumentType) {
+    return delegate.getAggregationTemporality(instrumentType);
+  }
+}

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/csm/exporter/BigtablePeriodicReader.java

@@ -34,10 +34,10 @@
  */
 public class BigtablePeriodicReader implements MetricReader {
   private final MetricReader delegate;
-  private final BigtableCloudMonitoringExporter exporter;
+  private final BigtableFilteringExporter exporter;
 
   public BigtablePeriodicReader(
-      BigtableCloudMonitoringExporter exporter, ScheduledExecutorService executor) {
+      BigtableFilteringExporter exporter, ScheduledExecutorService executor) {
     delegate = PeriodicMetricReader.builder(exporter).setExecutor(executor).build();
     this.exporter = exporter;
   }

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/csm/metrics/CustomAttemptLatency.java

@@ -0,0 +1,89 @@
+/*
+ * Copyright 2026 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
+ *
+ *     https://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.bigtable.data.v2.internal.csm.metrics;
+
+import com.google.bigtable.v2.ClusterInformation;
+import com.google.bigtable.v2.PeerInfo;
+import com.google.cloud.bigtable.data.v2.internal.csm.attributes.ClientInfo;
+import com.google.cloud.bigtable.data.v2.internal.csm.attributes.MethodInfo;
+import com.google.cloud.bigtable.data.v2.internal.csm.attributes.Util;
+import com.google.cloud.bigtable.data.v2.internal.csm.schema.CustomSchema;
+import io.grpc.Status;
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.metrics.DoubleHistogram;
+import io.opentelemetry.api.metrics.Meter;
+import java.time.Duration;
+import javax.annotation.Nullable;
+
+/**
+ * Custom attempt latencies with afe id metric label. This metric is high cardinality and is
+ * exported as a custom metric.
+ */
+public class CustomAttemptLatency extends MetricWrapper<CustomSchema> {
+  private static final String NAME = "bigtable.custom.attempt_latencies";
+
+  public CustomAttemptLatency() {
+    super(CustomSchema.INSTANCE, NAME);
+  }
+
+  public Recorder newRecorder(Meter meter) {
+    return new Recorder(meter);
+  }
+
+  public static class Recorder {
+    private final DoubleHistogram instrument;
+
+    private Recorder(Meter meter) {
+      instrument =
+          meter
+              .histogramBuilder(NAME)
+              .setDescription("Client observed latency per RPC attempt.")
+              .setUnit(Constants.Units.MILLISECOND)
+              .setExplicitBucketBoundariesAdvice(
+                  Constants.Buckets.AGGREGATION_WITH_MILLIS_HISTOGRAM)
+              .build();
+    }
+
+    public void record(
+        ClientInfo clientInfo,
+        String tableId,
+        @Nullable PeerInfo peerInfo,
+        @Nullable ClusterInformation clusterInfo,
+        MethodInfo methodInfo,
+        Status.Code code,
+        Duration latency) {
+
+      Attributes attributes =
+          Attributes.builder()
+              .put(
+                  Constants.MetricLabels.BIGTABLE_PROJECT_ID_KEY,
+                  clientInfo.getInstanceName().getProjectId())
+              .put(
+                  Constants.MetricLabels.INSTANCE_ID_KEY,
+                  clientInfo.getInstanceName().getInstanceId())
+              .put("table", tableId)
+              .put(Constants.MetricLabels.APP_PROFILE_KEY, clientInfo.getAppProfileId())
+              .put("cluster", Util.formatClusterIdMetricLabel(clusterInfo))
+              .put("zone", Util.formatZoneIdMetricLabel(clusterInfo))
+              .put(Constants.MetricLabels.STATUS_KEY, code.name())
+              .put(Constants.MetricLabels.METHOD_KEY, methodInfo.getName())
+              .put("afe_id", Util.formatAfeId(peerInfo))
+              .build();
+
+      instrument.record(toMillis(latency), attributes);
+    }
+  }
+}

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/csm/schema/CustomSchema.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2026 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
+ *
+ *     https://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.bigtable.data.v2.internal.csm.schema;
+
+import com.google.cloud.bigtable.data.v2.internal.csm.attributes.ClientInfo;
+import com.google.cloud.bigtable.data.v2.internal.csm.attributes.EnvInfo;
+import com.google.common.collect.ImmutableList;
+import com.google.monitoring.v3.ProjectName;
+import io.opentelemetry.api.common.Attributes;
+
+/** Placeholder schema for exporting custom metrics */
+public class CustomSchema extends Schema {
+  private CustomSchema() {
+    super("custom_schema", ImmutableList.of());
+  }
+
+  public static final CustomSchema INSTANCE = new CustomSchema();
+
+  @Override
+  public ProjectName extractProjectName(Attributes attrs, EnvInfo envInfo, ClientInfo clientInfo) {
+    return ProjectName.of(clientInfo.getInstanceName().getProjectId());
+  }
+}