feat: add downstream write latency to bigtableproxy (#9790)

NOTE: this depends on and includes https://togithub.com/GoogleCloudPlatform/java-docs-samples/pull/9789

Author: igorbernstein2

bigtable/bigtable-proxy/README.md

@@ -45,6 +45,8 @@ in a project your choosing. The metrics will be published under the namespace
   Cloud Bigtable service.
 * `bigtableproxy.client.gfe.duration_missing.count` Count of calls missing gfe response headers
 * `bigtableproxy.client.call.duration` Total duration of how long the outbound call took
+* `bigtableproxy.server.write_wait.duration` Total amount of time spent waiting for the downstream
+  client to be ready for data.
 * `bigtableproxy.client.channel.count` Number of open channels
 * `bigtableproxy.client.channel_change_count` Number of channel transitions by previous and next
   states.

bigtable/bigtable-proxy/src/main/java/com/google/cloud/bigtable/examples/proxy/core/CallProxy.java

@@ -17,6 +17,7 @@
 package com.google.cloud.bigtable.examples.proxy.core;
 
 import com.google.cloud.bigtable.examples.proxy.metrics.Tracer;
+import com.google.common.base.Stopwatch;
 import io.grpc.ClientCall;
 import io.grpc.Metadata;
 import io.grpc.ServerCall;
@@ -30,6 +31,8 @@ class CallProxy<ReqT, RespT> {
   final RequestProxy serverCallListener;
   final ResponseProxy clientCallListener;
 
+  private final Stopwatch downstreamStopwatch = Stopwatch.createUnstarted();
+
   /**
    * @param tracer a lifecycle observer to publish metrics.
    * @param serverCall the incoming server call. This will be triggered a customer client.
@@ -157,6 +160,7 @@ public void onMessage(RespT message) {
           // The incoming call is not ready for more responses. Stop requesting additional data
           // and wait for it to catch up.
           needToRequest = true;
+          downstreamStopwatch.reset().start();
         }
       }
     }
@@ -169,6 +173,10 @@ public void onReady() {
     // Called from RequestProxy, which is a different thread than the ClientCall.Listener
     // callbacks.
     synchronized void onServerReady() {
+      if (downstreamStopwatch.isRunning()) {
+        tracer.onDownstreamLatency(downstreamStopwatch.elapsed());
+        downstreamStopwatch.stop();
+      }
       if (needToRequest) {
         serverCallListener.clientCall.request(1);
         needToRequest = false;

bigtable/bigtable-proxy/src/main/java/com/google/cloud/bigtable/examples/proxy/metrics/Metrics.java

@@ -48,5 +48,7 @@ public interface Metrics {
 
   void recordChannelStateChange(ConnectivityState prevState, ConnectivityState newState);
 
+  void recordDownstreamLatency(MetricsAttributes attrs, Duration latency);
+
   interface MetricsAttributes {}
 }

bigtable/bigtable-proxy/src/main/java/com/google/cloud/bigtable/examples/proxy/metrics/MetricsImpl.java

@@ -95,6 +95,7 @@ public class MetricsImpl implements Closeable, Metrics {
   private final DoubleHistogram clientQueueLatencies;
   private final DoubleHistogram clientCallLatencies;
   private final DoubleHistogram clientCallFirstByteLatencies;
+  private final DoubleHistogram downstreamLatencies;
   private final LongCounter serverCallsStarted;
   private final LongHistogram requestSizes;
   private final LongHistogram responseSizes;
@@ -211,6 +212,15 @@ private static SdkMeterProvider createMeterProvider(Credentials credentials, Str
             .setUnit("ms")
             .build();
 
+    downstreamLatencies =
+        meter
+            .histogramBuilder(METRIC_PREFIX + "server.write_wait.duration")
+            .setDescription(
+                "Total amount of time spent waiting for the downstream client to be"
+                    + " ready for data")
+            .setUnit("ms")
+            .build();
+
     channelCounter =
         meter
             .upDownCounterBuilder(METRIC_PREFIX + "client.channel.count")
@@ -352,6 +362,11 @@ public void recordChannelStateChange(ConnectivityState prevState, ConnectivitySt
     channelStateChangeCounter.add(1, attributes);
   }
 
+  @Override
+  public void recordDownstreamLatency(MetricsAttributes attrs, Duration latency) {
+    downstreamLatencies.record(toMs(latency), unwrap(attrs));
+  }
+
   private static double toMs(Duration duration) {
     return duration.toNanos() / 1_000_000.0;
   }

bigtable/bigtable-proxy/src/main/java/com/google/cloud/bigtable/examples/proxy/metrics/Tracer.java

@@ -51,6 +51,7 @@ public class Tracer extends ClientStreamTracer {
   private final Stopwatch stopwatch;
   private volatile Optional<Duration> grpcQueueDuration = Optional.empty();
   private final AtomicLong responseSize = new AtomicLong();
+  private volatile Duration downstreamLatency;
 
   public Tracer(Metrics metrics, CallLabels callLabels) {
     this.metrics = metrics;
@@ -116,6 +117,7 @@ public void inboundMessage(int seqNo) {
 
   public void onCallFinished(Status status) {
     grpcQueueDuration.ifPresent(d -> metrics.recordQueueLatency(attrs, d));
+    metrics.recordDownstreamLatency(attrs, downstreamLatency);
     metrics.recordResponseSize(attrs, responseSize.get());
     metrics.recordCallLatency(
         attrs, status, Duration.ofMillis(stopwatch.elapsed(TimeUnit.MILLISECONDS)));
@@ -128,4 +130,8 @@ public void onCredentialsFetch(Status status, Duration duration) {
   public CallLabels getCallLabels() {
     return callLabels;
   }
+
+  public void onDownstreamLatency(Duration latency) {
+    downstreamLatency = downstreamLatency.plus(latency);
+  }
 }

bigtable/bigtable-proxy/src/test/java/com/google/cloud/bigtable/examples/proxy/metrics/NoopMetrics.java

@@ -55,6 +55,9 @@ public void recordCallLatency(MetricsAttributes attrs, Status status, Duration d
   @Override
   public void recordFirstByteLatency(MetricsAttributes attrs, Duration duration) {}
 
+  @Override
+  public void recordDownstreamLatency(MetricsAttributes attrs, Duration latency) {}
+
   @Override
   public void updateChannelCount(int delta) {}