feat: idle channel eviction

google-cloud-bigtable/clirr-ignored-differences.xml

@@ -426,4 +426,34 @@
  <method>*create*</method>
  <to>*</to>
  </difference>
+ <difference>
+ <differenceType>4001</differenceType>
+ <className>com/google/cloud/bigtable/data/v2/stub/BigtableChannelPrimer</className>
+ <to>com/google/api/gax/grpc/ChannelPrimer</to>
+ </difference>
+ <difference>
+ <differenceType>4001</differenceType>
+ <className>com/google/cloud/bigtable/data/v2/stub/NoOpChannelPrimer</className>
+ <to>com/google/api/gax/grpc/ChannelPrimer</to>
+ </difference>
+ <difference>
+ <differenceType>7005</differenceType>
+ <className>com/google/cloud/bigtable/gaxx/grpc/BigtableChannelPool</className>
+ <method>*create*</method>
+ <to>*</to>
+ </difference>
+ <difference>
+ <!-- InternalApi was updated -->
+ <differenceType>7005</differenceType>
+ <className>com/google/cloud/bigtable/gaxx/grpc/BigtableTransportChannelProvider</className>
+ <method>*create*</method>
+ <to>*</to>
+ </difference>
+ <!-- change method return type is ok because BigtableChannelPrimer is InternalApi -->
+ <difference>
+ <differenceType>7006</differenceType>
+ <className>com/google/cloud/bigtable/data/v2/stub/BigtableChannelPrimer</className>
+ <method>*sendPrimeRequestsAsync*</method>
+ <to>com.google.api.core.ApiFuture</to>
+ </difference>
 </differences>

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

@@ -15,14 +15,15 @@
  */
 package com.google.cloud.bigtable.data.v2.stub;
 
+import com.google.api.core.ApiFuture;
 import com.google.api.core.InternalApi;
 import com.google.api.core.SettableApiFuture;
-import com.google.api.gax.grpc.ChannelPrimer;
 import com.google.auth.Credentials;
 import com.google.bigtable.v2.BigtableGrpc;
 import com.google.bigtable.v2.InstanceName;
 import com.google.bigtable.v2.PingAndWarmRequest;
 import com.google.bigtable.v2.PingAndWarmResponse;
+import com.google.cloud.bigtable.gaxx.grpc.ChannelPrimer;
 import io.grpc.CallCredentials;
 import io.grpc.CallOptions;
 import io.grpc.ClientCall;
@@ -110,8 +111,7 @@ private void sendPrimeRequestsBlocking(ManagedChannel managedChannel) {
  }
  }
 
- public SettableApiFuture<PingAndWarmResponse> sendPrimeRequestsAsync(
- ManagedChannel managedChannel) {
+ public ApiFuture<PingAndWarmResponse> sendPrimeRequestsAsync(ManagedChannel managedChannel) {
  ClientCall<PingAndWarmRequest, PingAndWarmResponse> clientCall =
  managedChannel.newCall(
  BigtableGrpc.getPingAndWarmMethod(),

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

@@ -20,7 +20,6 @@
 import com.google.api.gax.core.BackgroundResource;
 import com.google.api.gax.core.CredentialsProvider;
 import com.google.api.gax.core.FixedCredentialsProvider;
-import com.google.api.gax.grpc.ChannelPrimer;
 import com.google.api.gax.grpc.InstantiatingGrpcChannelProvider;
 import com.google.api.gax.rpc.ClientContext;
 import com.google.auth.Credentials;
@@ -34,6 +33,7 @@
 import com.google.cloud.bigtable.data.v2.stub.metrics.MetricsProvider;
 import com.google.cloud.bigtable.data.v2.stub.metrics.NoopMetricsProvider;
 import com.google.cloud.bigtable.gaxx.grpc.BigtableTransportChannelProvider;
+import com.google.cloud.bigtable.gaxx.grpc.ChannelPrimer;
 import io.grpc.ManagedChannelBuilder;
 import io.grpc.opentelemetry.GrpcOpenTelemetry;
 import io.opentelemetry.api.OpenTelemetry;

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

@@ -15,8 +15,11 @@
  */
 package com.google.cloud.bigtable.data.v2.stub;
 
+import com.google.api.core.ApiFuture;
 import com.google.api.core.InternalApi;
-import com.google.api.gax.grpc.ChannelPrimer;
+import com.google.api.core.SettableApiFuture;
+import com.google.bigtable.v2.PingAndWarmResponse;
+import com.google.cloud.bigtable.gaxx.grpc.ChannelPrimer;
 import io.grpc.ManagedChannel;
 
 @InternalApi
@@ -28,7 +31,14 @@ static NoOpChannelPrimer create() {
  private NoOpChannelPrimer() {}
 
  @Override
- public void primeChannel(ManagedChannel managedChannel) {
+ public void primeChannel(ManagedChannel channel) {
  // No op
  }
+
+ @Override
+ public ApiFuture<PingAndWarmResponse> sendPrimeRequestsAsync(ManagedChannel channel) {
+ SettableApiFuture future = SettableApiFuture.create();
+ future.set(PingAndWarmResponse.getDefaultInstance());
+ return future;
+ }
 }

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/gaxx/grpc/BigtableChannelPool.java

@@ -17,7 +17,7 @@
 
 import com.google.api.core.InternalApi;
 import com.google.api.gax.grpc.ChannelFactory;
-import com.google.api.gax.grpc.ChannelPrimer;
+import com.google.cloud.bigtable.gaxx.grpc.ChannelPoolHealthChecker.ProbeResult;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
@@ -31,9 +31,11 @@
 import io.grpc.MethodDescriptor;
 import io.grpc.Status;
 import java.io.IOException;
+import java.time.Clock;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CancellationException;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
@@ -64,9 +66,9 @@ public class BigtableChannelPool extends ManagedChannel {
 
  private final ChannelPrimer channelPrimer;
  private final ScheduledExecutorService executor;
-
  private final Object entryWriteLock = new Object();
  @VisibleForTesting final AtomicReference<ImmutableList<Entry>> entries = new AtomicReference<>();
+ private final ChannelPoolHealthChecker channelPoolHealthChecker;
  private final AtomicInteger indexTicker = new AtomicInteger();
  private final String authority;
 
@@ -96,6 +98,10 @@ public static BigtableChannelPool create(
  this.settings = settings;
  this.channelFactory = channelFactory;
  this.channelPrimer = channelPrimer;
+ Clock systemClock = Clock.systemUTC();
+ this.channelPoolHealthChecker =
+ new ChannelPoolHealthChecker(entries::get, channelPrimer, executor, systemClock);
+ this.channelPoolHealthChecker.start();
 
  ImmutableList.Builder<Entry> initialListBuilder = ImmutableList.builder();
 
@@ -445,15 +451,32 @@ static class Entry {
 
  private final AtomicInteger maxOutstanding = new AtomicInteger();
 
- // Flag that the channel should be closed once all of the outstanding RPC complete.
+ /** Queue storing the last 5 minutes of probe results */
+ @VisibleForTesting
+ final ConcurrentLinkedQueue<ProbeResult> probeHistory = new ConcurrentLinkedQueue<>();
+
+ /**
+ * Keep both # of failed and # of successful probes so that we don't have to check size() on the
+ * ConcurrentLinkedQueue all the time
+ */
+ final AtomicInteger failedProbesInWindow = new AtomicInteger();
+
+ final AtomicInteger successfulProbesInWindow = new AtomicInteger();
+
+ // Flag that the channel should be closed once all the outstanding RPCs complete.
  private final AtomicBoolean shutdownRequested = new AtomicBoolean();
  // Flag that the channel has been closed.
  private final AtomicBoolean shutdownInitiated = new AtomicBoolean();
 
- private Entry(ManagedChannel channel) {
+ @VisibleForTesting
+ Entry(ManagedChannel channel) {
  this.channel = channel;
  }
 
+ ManagedChannel getManagedChannel() {
+ return this.channel;
+ }
+
  int getAndResetMaxOutstanding() {
  return maxOutstanding.getAndSet(outstandingRpcs.get());
  }
@@ -468,7 +491,7 @@ private boolean retain() {
  // register desire to start RPC
  int currentOutstanding = outstandingRpcs.incrementAndGet();
 
- // Rough book keeping
+ // Rough bookkeeping
  int prevMax = maxOutstanding.get();
  if (currentOutstanding > prevMax) {
  maxOutstanding.incrementAndGet();

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/gaxx/grpc/BigtableTransportChannelProvider.java

@@ -18,7 +18,6 @@
 import com.google.api.core.InternalApi;
 import com.google.api.gax.grpc.ChannelFactory;
 import com.google.api.gax.grpc.ChannelPoolSettings;
-import com.google.api.gax.grpc.ChannelPrimer;
 import com.google.api.gax.grpc.GrpcTransportChannel;
 import com.google.api.gax.grpc.InstantiatingGrpcChannelProvider;
 import com.google.api.gax.rpc.TransportChannel;