Merge branch 'main' into compute_disk_create_with_snapshot_schedule

Author: TetyanaYahodska

.github/CODEOWNERS

@@ -49,8 +49,6 @@
 # DEE Platform Ops (DEEPO)
 /errorreporting                  @GoogleCloudPlatform/java-samples-reviewers @yoshi-approver @GoogleCloudPlatform/cloud-samples-reviewers
 /monitoring                      @GoogleCloudPlatform/java-samples-reviewers @yoshi-approver @GoogleCloudPlatform/cloud-samples-reviewers
-/opencensus                      @GoogleCloudPlatform/java-samples-reviewers @yoshi-approver @GoogleCloudPlatform/cloud-samples-reviewers
-/trace                           @GoogleCloudPlatform/java-samples-reviewers @yoshi-approver @GoogleCloudPlatform/cloud-samples-reviewers
 
 # Cloud SDK Databases & Data Analytics teams
 # ---* Cloud Native DB

bigtable/bigtable-proxy/README.md

@@ -45,7 +45,11 @@ 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.
 * `bigtableproxy.client.call.max_outstanding_count` Maximum number of concurrent RPCs in a single
   minute window
 * `bigtableproxy.presence` Counts number of proxy processes (emit 1 per process).

bigtable/bigtable-proxy/src/main/java/com/google/cloud/bigtable/examples/proxy/channelpool/ChannelPool.java

@@ -47,7 +47,7 @@
  * A {@link ManagedChannel} that will send requests round-robin via a set of channels.
  *
  * <p>In addition to spreading requests over a set of child connections, the pool will also actively
- * manage the lifecycle of the channels. Currently lifecycle management is limited to pre-emptively
+ * manage the lifecycle of the channels. Currently, lifecycle management is limited to pre-emptively
  * replacing channels every hour. In the future it will dynamically size the pool based on number of
  * outstanding requests.
  *

bigtable/bigtable-proxy/src/main/java/com/google/cloud/bigtable/examples/proxy/channelpool/DataChannel.java

@@ -19,7 +19,8 @@
 import com.google.bigtable.v2.BigtableGrpc;
 import com.google.bigtable.v2.PingAndWarmRequest;
 import com.google.bigtable.v2.PingAndWarmResponse;
-import com.google.cloud.bigtable.examples.proxy.metrics.CallLabels;
+import com.google.cloud.bigtable.examples.proxy.core.CallLabels;
+import com.google.cloud.bigtable.examples.proxy.core.CallLabels.PrimingKey;
 import com.google.cloud.bigtable.examples.proxy.metrics.Metrics;
 import com.google.cloud.bigtable.examples.proxy.metrics.Tracer;
 import com.google.common.util.concurrent.ListenableFuture;
@@ -34,13 +35,13 @@
 import io.grpc.ManagedChannel;
 import io.grpc.ManagedChannelBuilder;
 import io.grpc.Metadata;
+import io.grpc.Metadata.Key;
 import io.grpc.MethodDescriptor;
 import io.grpc.Status;
-import io.grpc.StatusRuntimeException;
-import java.net.URLEncoder;
-import java.nio.charset.StandardCharsets;
+import java.time.Duration;
 import java.util.List;
 import java.util.Optional;
+import java.util.Random;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
@@ -50,16 +51,31 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+/**
+ * Decorator for a Bigtable data plane connection to add channel warming via PingAndWarm. Channel
+ * warming will happen on creation and then every 3 minutes (with jitter).
+ */
 public class DataChannel extends ManagedChannel {
   private static final Logger LOGGER = LoggerFactory.getLogger(DataChannel.class);
 
+  private static final Metadata.Key<String> GFE_DEBUG_REQ_HEADER =
+      Key.of("X-Return-Encrypted-Headers", Metadata.ASCII_STRING_MARSHALLER);
+  private static final Metadata.Key<String> GFE_DEBUG_RESP_HEADER =
+      Key.of("X-Encrypted-Debug-Headers", Metadata.ASCII_STRING_MARSHALLER);
+
+  private static final Duration WARM_PERIOD = Duration.ofMinutes(3);
+  private static final Duration MAX_JITTER = Duration.ofSeconds(10);
+
+  private final Random random = new Random();
   private final ManagedChannel inner;
   private final Metrics metrics;
   private final ResourceCollector resourceCollector;
   private final CallCredentials callCredentials;
-  private final ScheduledFuture<?> antiIdleTask;
+  private final ScheduledExecutorService warmingExecutor;
+  private volatile ScheduledFuture<?> antiIdleTask;
 
   private final AtomicBoolean closed = new AtomicBoolean();
+  private final Object scheduleLock = new Object();
 
   public DataChannel(
       ResourceCollector resourceCollector,
@@ -80,8 +96,12 @@ public DataChannel(
             .keepAliveTime(30, TimeUnit.SECONDS)
             .keepAliveTimeout(10, TimeUnit.SECONDS)
             .build();
+
+    this.warmingExecutor = warmingExecutor;
     this.metrics = metrics;
 
+    new StateTransitionWatcher().run();
+
     try {
       warm();
     } catch (RuntimeException e) {
@@ -93,40 +113,57 @@ public DataChannel(
       throw e;
     }
 
-    antiIdleTask = warmingExecutor.scheduleAtFixedRate(this::warmQuietly, 3, 3, TimeUnit.MINUTES);
+    antiIdleTask =
+        warmingExecutor.schedule(this::warmTask, nextWarmup().toMillis(), TimeUnit.MILLISECONDS);
     metrics.updateChannelCount(1);
   }
 
-  private void warmQuietly() {
+  private Duration nextWarmup() {
+    return WARM_PERIOD.minus(
+        Duration.ofMillis((long) (MAX_JITTER.toMillis() * random.nextDouble())));
+  }
+
+  private void warmTask() {
     try {
       warm();
     } catch (RuntimeException e) {
       LOGGER.warn("anti idle ping failed, forcing reconnect", e);
       inner.enterIdle();
+    } finally {
+      synchronized (scheduleLock) {
+        if (!closed.get()) {
+          antiIdleTask =
+              warmingExecutor.schedule(
+                  this::warmTask, nextWarmup().toMillis(), TimeUnit.MILLISECONDS);
+        }
+      }
     }
   }
 
   private void warm() {
-    List<PingAndWarmRequest> requests = resourceCollector.getRequests();
-    if (requests.isEmpty()) {
+    List<PrimingKey> primingKeys = resourceCollector.getPrimingKeys();
+    if (primingKeys.isEmpty()) {
       return;
     }
 
+    LOGGER.debug("Warming channel {} with: {}", inner, primingKeys);
+
     List<ListenableFuture<PingAndWarmResponse>> futures =
-        requests.stream().map(this::sendPingAndWarm).collect(Collectors.toList());
+        primingKeys.stream().map(this::sendPingAndWarm).collect(Collectors.toList());
 
     int successCount = 0;
     int failures = 0;
     for (ListenableFuture<PingAndWarmResponse> future : futures) {
-      PingAndWarmRequest request = requests.get(successCount + failures);
+      PrimingKey request = primingKeys.get(successCount + failures);
       try {
         future.get();
         successCount++;
       } catch (ExecutionException e) {
-        // All permenant errors are ignored and treated as a success
+        // All permanent errors are ignored and treated as a success
         // The priming request for that generated the error will be dropped
-        if (e.getCause() instanceof StatusRuntimeException) {
-          StatusRuntimeException se = (StatusRuntimeException) e.getCause();
+        if (e.getCause() instanceof PingAndWarmException) {
+          PingAndWarmException se = (PingAndWarmException) e.getCause();
+
           switch (se.getStatus().getCode()) {
             case INTERNAL:
             case PERMISSION_DENIED:
@@ -139,8 +176,15 @@ private void warm() {
             default:
               // noop
           }
+          LOGGER.warn(
+              "Failed to prime channel with request: {}, status: {}, debug response headers: {}",
+              request,
+              se.getStatus(),
+              Optional.ofNullable(se.getDebugHeaders()).orElse("<missing>"));
+        } else {
+          LOGGER.warn("Unexpected failure priming channel with request: {}", request, e.getCause());
         }
-        LOGGER.warn("Failed to prime channel with request: {}", request, e.getCause());
+
         failures++;
       } catch (InterruptedException e) {
         throw new RuntimeException("Interrupted while priming channel with request: " + request, e);
@@ -151,13 +195,13 @@ private void warm() {
     }
   }
 
-  private ListenableFuture<PingAndWarmResponse> sendPingAndWarm(PingAndWarmRequest request) {
-    CallLabels callLabels =
-        CallLabels.create(
-            BigtableGrpc.getPingAndWarmMethod(),
-            Optional.of("bigtableproxy"),
-            Optional.of(request.getName()),
-            Optional.of(request.getAppProfileId()));
+  private ListenableFuture<PingAndWarmResponse> sendPingAndWarm(PrimingKey primingKey) {
+    Metadata metadata = primingKey.composeMetadata();
+    metadata.put(GFE_DEBUG_REQ_HEADER, "gfe_response_only");
+    PingAndWarmRequest request = primingKey.composeProto();
+    request = request.toBuilder().setName(request.getName()).build();
+
+    CallLabels callLabels = CallLabels.create(BigtableGrpc.getPingAndWarmMethod(), metadata);
     Tracer tracer = new Tracer(metrics, callLabels);
 
     CallOptions callOptions =
@@ -169,17 +213,11 @@ private ListenableFuture<PingAndWarmResponse> sendPingAndWarm(PingAndWarmRequest
     ClientCall<PingAndWarmRequest, PingAndWarmResponse> call =
         inner.newCall(BigtableGrpc.getPingAndWarmMethod(), callOptions);
 
-    Metadata metadata = new Metadata();
-    metadata.put(
-        CallLabels.REQUEST_PARAMS,
-        String.format(
-            "name=%s&app_profile_id=%s",
-            URLEncoder.encode(request.getName(), StandardCharsets.UTF_8),
-            URLEncoder.encode(request.getAppProfileId(), StandardCharsets.UTF_8)));
-
     SettableFuture<PingAndWarmResponse> f = SettableFuture.create();
     call.start(
         new Listener<>() {
+          String debugHeaders = null;
+
           @Override
           public void onMessage(PingAndWarmResponse response) {
             if (!f.set(response)) {
@@ -188,14 +226,22 @@ public void onMessage(PingAndWarmResponse response) {
             }
           }
 
+          @Override
+          public void onHeaders(Metadata headers) {
+            debugHeaders = headers.get(GFE_DEBUG_RESP_HEADER);
+          }
+
           @Override
           public void onClose(Status status, Metadata trailers) {
             tracer.onCallFinished(status);
 
             if (status.isOk()) {
-              f.setException(new IllegalStateException("PingAndWarm was missing a response"));
+              f.setException(
+                  new PingAndWarmException(
+                      "PingAndWarm was missing a response", debugHeaders, trailers, status));
             } else {
-              f.setException(status.asRuntimeException());
+              f.setException(
+                  new PingAndWarmException("PingAndWarm failed", debugHeaders, trailers, status));
             }
           }
         },
@@ -207,12 +253,45 @@ public void onClose(Status status, Metadata trailers) {
     return f;
   }
 
+  static class PingAndWarmException extends RuntimeException {
+
+    private final String debugHeaders;
+    private final Metadata trailers;
+    private final Status status;
+
+    public PingAndWarmException(
+        String message, String debugHeaders, Metadata trailers, Status status) {
+      super(String.format("PingAndWarm failed, status: " + status));
+      this.debugHeaders = debugHeaders;
+      this.trailers = trailers;
+      this.status = status;
+    }
+
+    public String getDebugHeaders() {
+      return debugHeaders;
+    }
+
+    public Metadata getTrailers() {
+      return trailers;
+    }
+
+    public Status getStatus() {
+      return status;
+    }
+  }
+
   @Override
   public ManagedChannel shutdown() {
-    if (closed.compareAndSet(false, true)) {
+    final boolean closing;
+
+    synchronized (scheduleLock) {
+      closing = closed.compareAndSet(false, true);
+      antiIdleTask.cancel(true);
+    }
+    if (closing) {
       metrics.updateChannelCount(-1);
     }
-    antiIdleTask.cancel(true);
+
     return inner.shutdown();
   }
 
@@ -228,10 +307,17 @@ public boolean isTerminated() {
 
   @Override
   public ManagedChannel shutdownNow() {
-    if (closed.compareAndSet(false, true)) {
+    final boolean closing;
+
+    synchronized (scheduleLock) {
+      closing = closed.compareAndSet(false, true);
+      antiIdleTask.cancel(true);
+    }
+
+    if (closing) {
       metrics.updateChannelCount(-1);
     }
-    antiIdleTask.cancel(true);
+
     return inner.shutdownNow();
   }
 
@@ -282,4 +368,20 @@ public <RequestT, ResponseT> ClientCall<RequestT, ResponseT> newCall(
   public String authority() {
     return inner.authority();
   }
+
+  class StateTransitionWatcher implements Runnable {
+    private ConnectivityState prevState = null;
+
+    @Override
+    public void run() {
+      if (closed.get()) {
+        return;
+      }
+
+      ConnectivityState newState = inner.getState(false);
+      metrics.recordChannelStateChange(prevState, newState);
+      prevState = newState;
+      inner.notifyWhenStateChanged(prevState, this);
+    }
+  }
 }