feat: add internal "deadline remaining" client side metric #2341

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

@@ -104,6 +104,7 @@
 import com.google.cloud.bigtable.data.v2.stub.changestream.GenerateInitialChangeStreamPartitionsUserCallable;
 import com.google.cloud.bigtable.data.v2.stub.changestream.ReadChangeStreamResumptionStrategy;
 import com.google.cloud.bigtable.data.v2.stub.changestream.ReadChangeStreamUserCallable;
+import com.google.cloud.bigtable.data.v2.stub.metrics.BigtableTracer;
 import com.google.cloud.bigtable.data.v2.stub.metrics.BigtableTracerStreamingCallable;
 import com.google.cloud.bigtable.data.v2.stub.metrics.BigtableTracerUnaryCallable;
 import com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsTracerFactory;
@@ -539,7 +540,12 @@ public <RowT> ServerStreamingCallable<Query, RowT> createReadRowsCallable(
         new TracedServerStreamingCallable<>(
             readRowsUserCallable, clientContext.getTracerFactory(), span);
 
-    return traced.withDefaultCallContext(clientContext.getDefaultCallContext());
+    return traced.withDefaultCallContext(
+        clientContext
+            .getDefaultCallContext()
+            .withOption(
+                BigtableTracer.OPERATION_TIMEOUT_KEY,
+                settings.readRowsSettings().getRetrySettings().getTotalTimeout()));
   }
 
   /**
@@ -575,7 +581,12 @@ public <RowT> UnaryCallable<Query, RowT> createReadRowCallable(RowAdapter<RowT>
         new TracedUnaryCallable<>(
             firstRow, clientContext.getTracerFactory(), getSpanName("ReadRow"));
 
-    return traced.withDefaultCallContext(clientContext.getDefaultCallContext());
+    return traced.withDefaultCallContext(
+        clientContext
+            .getDefaultCallContext()
+            .withOption(
+                BigtableTracer.OPERATION_TIMEOUT_KEY,
+                settings.readRowSettings().getRetrySettings().getTotalTimeout()));
   }
 
   /**
@@ -694,7 +705,12 @@ private <RowT> UnaryCallable<Query, List<RowT>> createBulkReadRowsCallable(
     UnaryCallable<Query, List<RowT>> traced =
         new TracedUnaryCallable<>(tracedBatcher, clientContext.getTracerFactory(), span);
 
-    return traced.withDefaultCallContext(clientContext.getDefaultCallContext());
+    return traced.withDefaultCallContext(
+        clientContext
+            .getDefaultCallContext()
+            .withOption(
+                BigtableTracer.OPERATION_TIMEOUT_KEY,
+                settings.bulkReadRowsSettings().getRetrySettings().getTotalTimeout()));
   }
 
   /**
@@ -766,7 +782,14 @@ private UnaryCallable<String, List<KeyOffset>> createSampleRowKeysCallable() {
     UnaryCallable<com.google.bigtable.v2.SampleRowKeysRequest, List<SampleRowKeysResponse>>
         baseCallable = createSampleRowKeysBaseCallable();
     return createUserFacingUnaryCallable(
-        methodName, new SampleRowKeysCallable(baseCallable, requestContext));
+        methodName,
+        new SampleRowKeysCallable(baseCallable, requestContext)
+            .withDefaultCallContext(
+                clientContext
+                    .getDefaultCallContext()
+                    .withOption(
+                        BigtableTracer.OPERATION_TIMEOUT_KEY,
+                        settings.sampleRowKeysSettings().getRetrySettings().getTotalTimeout())));
   }
 
   /**
@@ -789,7 +812,14 @@ private UnaryCallable<String, List<KeyOffset>> createSampleRowKeysCallable() {
     UnaryCallable<com.google.bigtable.v2.SampleRowKeysRequest, List<SampleRowKeysResponse>>
         baseCallable = createSampleRowKeysBaseCallable();
     return createUserFacingUnaryCallable(
-        methodName, new SampleRowKeysCallableWithRequest(baseCallable, requestContext));
+        methodName,
+        new SampleRowKeysCallableWithRequest(baseCallable, requestContext)
+            .withDefaultCallContext(
+                clientContext
+                    .getDefaultCallContext()
+                    .withOption(
+                        BigtableTracer.OPERATION_TIMEOUT_KEY,
+                        settings.sampleRowKeysSettings().getRetrySettings().getTotalTimeout())));
   }
 
   /**
@@ -836,7 +866,14 @@ public Map<String, String> extract(MutateRowRequest mutateRowRequest) {
         withRetries(withBigtableTracer, settings.mutateRowSettings());
 
     return createUserFacingUnaryCallable(
-        methodName, new MutateRowCallable(retrying, requestContext));
+        methodName,
+        new MutateRowCallable(retrying, requestContext)
+            .withDefaultCallContext(
+                clientContext
+                    .getDefaultCallContext()
+                    .withOption(
+                        BigtableTracer.OPERATION_TIMEOUT_KEY,
+                        settings.mutateRowSettings().getRetrySettings().getTotalTimeout())));
   }
 
   /**
@@ -953,7 +990,12 @@ public Map<String, String> extract(MutateRowsRequest mutateRowsRequest) {
         new TracedUnaryCallable<>(
             tracedBatcherUnaryCallable, clientContext.getTracerFactory(), spanName);
 
-    return traced.withDefaultCallContext(clientContext.getDefaultCallContext());
+    return traced.withDefaultCallContext(
+        clientContext
+            .getDefaultCallContext()
+            .withOption(
+                BigtableTracer.OPERATION_TIMEOUT_KEY,
+                settings.bulkMutateRowsSettings().getRetrySettings().getTotalTimeout()));
   }
 
   /**
@@ -1093,7 +1135,17 @@ public Map<String, String> extract(
         withRetries(withBigtableTracer, settings.checkAndMutateRowSettings());
 
     return createUserFacingUnaryCallable(
-        methodName, new CheckAndMutateRowCallable(retrying, requestContext));
+        methodName,
+        new CheckAndMutateRowCallable(retrying, requestContext)
+            .withDefaultCallContext(
+                clientContext
+                    .getDefaultCallContext()
+                    .withOption(
+                        BigtableTracer.OPERATION_TIMEOUT_KEY,
+                        settings
+                            .checkAndMutateRowSettings()
+                            .getRetrySettings()
+                            .getTotalTimeout())));
   }
 
   /**
@@ -1139,7 +1191,17 @@ public Map<String, String> extract(ReadModifyWriteRowRequest request) {
         withRetries(withBigtableTracer, settings.readModifyWriteRowSettings());
 
     return createUserFacingUnaryCallable(
-        methodName, new ReadModifyWriteRowCallable(retrying, requestContext));
+        methodName,
+        new ReadModifyWriteRowCallable(retrying, requestContext)
+            .withDefaultCallContext(
+                clientContext
+                    .getDefaultCallContext()
+                    .withOption(
+                        BigtableTracer.OPERATION_TIMEOUT_KEY,
+                        settings
+                            .readModifyWriteRowSettings()
+                            .getRetrySettings()
+                            .getTotalTimeout())));
   }
 
   /**
@@ -1222,7 +1284,15 @@ public Map<String, String> extract(
     ServerStreamingCallable<String, ByteStringRange> traced =
         new TracedServerStreamingCallable<>(retrying, clientContext.getTracerFactory(), span);
 
-    return traced.withDefaultCallContext(clientContext.getDefaultCallContext());
+    return traced.withDefaultCallContext(
+        clientContext
+            .getDefaultCallContext()
+            .withOption(
+                BigtableTracer.OPERATION_TIMEOUT_KEY,
+                settings
+                    .generateInitialChangeStreamPartitionsSettings()
+                    .getRetrySettings()
+                    .getTotalTimeout()));
   }
 
   /**
@@ -1302,7 +1372,12 @@ public Map<String, String> extract(
         new TracedServerStreamingCallable<>(
             readChangeStreamUserCallable, clientContext.getTracerFactory(), span);
 
-    return traced.withDefaultCallContext(clientContext.getDefaultCallContext());
+    return traced.withDefaultCallContext(
+        clientContext
+            .getDefaultCallContext()
+            .withOption(
+                BigtableTracer.OPERATION_TIMEOUT_KEY,
+                settings.readChangeStreamSettings().getRetrySettings().getTotalTimeout()));
   }
 
   /**
@@ -1388,7 +1463,13 @@ public Map<String, String> extract(ExecuteQueryRequest executeQueryRequest) {
         new TracedServerStreamingCallable<>(retries, clientContext.getTracerFactory(), span);
 
     return new ExecuteQueryCallable(
-        traced.withDefaultCallContext(clientContext.getDefaultCallContext()), requestContext);
+        traced.withDefaultCallContext(
+            clientContext
+                .getDefaultCallContext()
+                .withOption(
+                    BigtableTracer.OPERATION_TIMEOUT_KEY,
+                    settings.executeQuerySettings().getRetrySettings().getTotalTimeout())),
+        requestContext);
   }
 
   /**
@@ -1420,7 +1501,12 @@ public Map<String, String> extract(PingAndWarmRequest request) {
                     })
                 .build(),
             Collections.emptySet());
-    return pingAndWarm.withDefaultCallContext(clientContext.getDefaultCallContext());
+    return pingAndWarm.withDefaultCallContext(
+        clientContext
+            .getDefaultCallContext()
+            .withOption(
+                BigtableTracer.OPERATION_TIMEOUT_KEY,
+                settings.pingAndWarmSettings().getRetrySettings().getTotalTimeout()));
   }
 
   private <RequestT, ResponseT> UnaryCallable<RequestT, ResponseT> withRetries(

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

@@ -23,6 +23,7 @@
 import static com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsConstants.METER_NAME;
 import static com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsConstants.OPERATION_LATENCIES_NAME;
 import static com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsConstants.PER_CONNECTION_ERROR_COUNT_NAME;
+import static com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsConstants.REMAINING_DEADLINE_NAME;
 import static com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsConstants.RETRY_COUNT_NAME;
 import static com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsConstants.SERVER_LATENCIES_NAME;
 
@@ -115,7 +116,8 @@ public final class BigtableCloudMonitoringExporter implements MetricExporter {
               CLIENT_BLOCKING_LATENCIES_NAME,
               APPLICATION_BLOCKING_LATENCIES_NAME,
               RETRY_COUNT_NAME,
-              CONNECTIVITY_ERROR_COUNT_NAME)
+              CONNECTIVITY_ERROR_COUNT_NAME,
+              REMAINING_DEADLINE_NAME)
           .stream()
           .map(m -> METER_NAME + m)
           .collect(ImmutableList.toImmutableList());

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

@@ -16,10 +16,12 @@
 package com.google.cloud.bigtable.data.v2.stub.metrics;
 
 import com.google.api.core.BetaApi;
+import com.google.api.core.InternalApi;
 import com.google.api.gax.rpc.ApiCallContext;
 import com.google.api.gax.tracing.ApiTracer;
 import com.google.api.gax.tracing.BaseApiTracer;
 import javax.annotation.Nullable;
+import org.threeten.bp.Duration;
 
 /**
  * A Bigtable specific {@link ApiTracer} that includes additional contexts. This class is a base
@@ -30,6 +32,10 @@ public class BigtableTracer extends BaseApiTracer {
 
   private volatile int attempt = 0;
 
+  @InternalApi("for internal use only")
+  public static final ApiCallContext.Key<Duration> OPERATION_TIMEOUT_KEY =
+      ApiCallContext.Key.create("OPERATION_TIMEOUT");
+
   @Override
   public void attemptStarted(int attemptNumber) {
     this.attempt = attemptNumber;
@@ -93,4 +99,12 @@ public void grpcChannelQueuedLatencies(long queuedTimeMs) {
   public void grpcMessageSent() {
     // noop
   }
+
+  /**
+   * Record the operation timeout from user settings for calculating remaining deadline. This will
+   * be called in BuiltinMetricsTracer.
+   */
+  public void setOperationTimeout(Duration operationTimeout) {
+    // noop
+  }
 }

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

@@ -16,6 +16,7 @@
 package com.google.cloud.bigtable.data.v2.stub.metrics;
 
 import com.google.api.core.InternalApi;
+import com.google.api.gax.grpc.GrpcCallContext;
 import com.google.api.gax.grpc.GrpcResponseMetadata;
 import com.google.api.gax.rpc.ApiCallContext;
 import com.google.api.gax.rpc.ResponseObserver;
@@ -26,6 +27,7 @@
 import com.google.common.base.Stopwatch;
 import java.util.concurrent.TimeUnit;
 import javax.annotation.Nonnull;
+import org.threeten.bp.Duration;
 
 /**
  * This callable will
@@ -62,6 +64,9 @@ public void call(
       BigtableTracerResponseObserver<ResponseT> innerObserver =
           new BigtableTracerResponseObserver<>(
               responseObserver, (BigtableTracer) context.getTracer(), responseMetadata);
+      GrpcCallContext callContext = (GrpcCallContext) context;
+      Duration deadline = callContext.getOption(BigtableTracer.OPERATION_TIMEOUT_KEY);
+      ((BigtableTracer) context.getTracer()).setOperationTimeout(deadline);
       innerCallable.call(
           request,
           innerObserver,

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

@@ -19,12 +19,14 @@
 import com.google.api.core.ApiFutureCallback;
 import com.google.api.core.ApiFutures;
 import com.google.api.core.InternalApi;
+import com.google.api.gax.grpc.GrpcCallContext;
 import com.google.api.gax.grpc.GrpcResponseMetadata;
 import com.google.api.gax.rpc.ApiCallContext;
 import com.google.api.gax.rpc.UnaryCallable;
 import com.google.common.base.Preconditions;
 import com.google.common.util.concurrent.MoreExecutors;
 import javax.annotation.Nonnull;
+import org.threeten.bp.Duration;
 
 /**
  * This callable will:
@@ -58,6 +60,9 @@ public ApiFuture<ResponseT> futureCall(RequestT request, ApiCallContext context)
       BigtableTracerUnaryCallback<ResponseT> callback =
           new BigtableTracerUnaryCallback<ResponseT>(
               (BigtableTracer) context.getTracer(), responseMetadata);
+      GrpcCallContext callContext = (GrpcCallContext) context;
+      Duration deadline = callContext.getOption(BigtableTracer.OPERATION_TIMEOUT_KEY);
+      ((BigtableTracer) context.getTracer()).setOperationTimeout(deadline);
       ApiFuture<ResponseT> future =
           innerCallable.futureCall(
               request,

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

@@ -58,6 +58,7 @@ public class BuiltinMetricsConstants {
   static final String SERVER_LATENCIES_NAME = "server_latencies";
   static final String FIRST_RESPONSE_LATENCIES_NAME = "first_response_latencies";
   static final String APPLICATION_BLOCKING_LATENCIES_NAME = "application_latencies";
+  static final String REMAINING_DEADLINE_NAME = "remaining_deadline";
   static final String CLIENT_BLOCKING_LATENCIES_NAME = "throttling_latencies";
   static final String PER_CONNECTION_ERROR_COUNT_NAME = "per_connection_error_count";
 
@@ -214,6 +215,16 @@ public static Map<InstrumentSelector, View> getAllViews() {
         ImmutableSet.<AttributeKey>builder()
             .add(BIGTABLE_PROJECT_ID_KEY, INSTANCE_ID_KEY, APP_PROFILE_KEY, CLIENT_NAME_KEY)
             .build());
+    defineView(
+        views,
+        REMAINING_DEADLINE_NAME,
+        AGGREGATION_WITH_MILLIS_HISTOGRAM,
+        InstrumentType.HISTOGRAM,
+        "ms",
+        ImmutableSet.<AttributeKey>builder()
+            .addAll(COMMON_ATTRIBUTES)
+            .add(STREAMING_KEY, STATUS_KEY)
+            .build());
 
     return views.build();
   }