refactor channel health checking structure

Change-Id: I6857b40fa9a08d101b896470eaa7abdd9740c3e9

Author: nicholsl

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

@@ -61,8 +61,6 @@ public class BigtableChannelPool extends ManagedChannel {
   private final BigtableChannelPoolSettings settings;
   private final ChannelFactory channelFactory;
   private final ScheduledExecutorService executor;
-  private final ScheduledExecutorService channelHealthProbingExecutor =
-      Executors.newSingleThreadScheduledExecutor();
 
   private final Object entryWriteLock = new Object();
   @VisibleForTesting final AtomicReference<ImmutableList<Entry>> entries = new AtomicReference<>();
@@ -97,7 +95,7 @@ public static BigtableChannelPool create(
 
     for (int i = 0; i < settings.getInitialChannelCount(); i++) {
       initialListBuilder.add(
-          new Entry(channelFactory.createSingleChannel(), channelHealthProbingExecutor));
+          new Entry(channelFactory.createSingleChannel()));
     }
 
     entries.set(initialListBuilder.build());
@@ -155,9 +153,6 @@ public ManagedChannel shutdown() {
       // shutdownNow will cancel scheduled tasks
       executor.shutdownNow();
     }
-    if (channelHealthProbingExecutor != null) {
-      channelHealthProbingExecutor.shutdownNow();
-    }
     return this;
   }
 
@@ -198,9 +193,6 @@ public ManagedChannel shutdownNow() {
     if (executor != null) {
       executor.shutdownNow();
     }
-    if (channelHealthProbingExecutor != null) {
-      channelHealthProbingExecutor.shutdownNow();
-    }
     return this;
   }
 
@@ -328,7 +320,7 @@ private void expand(int desiredSize) {
     for (int i = 0; i < desiredSize - localEntries.size(); i++) {
       try {
         newEntries.add(
-            new Entry(channelFactory.createSingleChannel(), channelHealthProbingExecutor));
+            new Entry(channelFactory.createSingleChannel()));
       } catch (IOException e) {
         LOG.log(Level.WARNING, "Failed to add channel", e);
       }
@@ -367,7 +359,7 @@ void refresh() {
       for (int i = 0; i < newEntries.size(); i++) {
         try {
           newEntries.set(
-              i, new Entry(channelFactory.createSingleChannel(), channelHealthProbingExecutor));
+              i, new Entry(channelFactory.createSingleChannel()));
         } catch (IOException e) {
           LOG.log(Level.WARNING, "Failed to refresh channel, leaving old channel", e);
         }
@@ -445,26 +437,38 @@ static class Entry {
     private final AtomicInteger maxOutstanding = new AtomicInteger();
     private final AtomicInteger probesInFlight = new AtomicInteger(0);
 
+    /** Number of probes in flight plus number of probe results. (No-op stub) */
+    AtomicInteger recentProbesSent() {
+      return new AtomicInteger(0);
+    }
+
+    /** Number of recently failed probes. (No-op stub) */
+    AtomicInteger recentlyFailedProbes() {
+      return new AtomicInteger(0);
+    }
+
+    /**
+     * Determines if the channel is healthy. (No-op stub)
+     *
+     * @return A default value of true.
+     */
+    public boolean healthy() {
+      return true;
+    }
+
     // Flag that the channel should be closed once all of the outstanding RPC complete.
     private final AtomicBoolean shutdownRequested = new AtomicBoolean();
     // Flag that the channel has been closed.
     private final AtomicBoolean shutdownInitiated = new AtomicBoolean();
-    private final ChannelHealthChecker healthChecker;
 
-    private Entry(ManagedChannel channel, ScheduledExecutorService executor) {
+    private Entry(ManagedChannel channel) {
       this.channel = channel;
-      this.healthChecker = new ChannelHealthChecker(this, executor);
     }
 
     ManagedChannel getManagedChannel() {
       return this.channel;
     }
 
-    // Add a getter for the healthChecker
-    public ChannelHealthChecker getHealthChecker() {
-      return healthChecker;
-    }
-
     int getAndResetMaxOutstanding() {
       return maxOutstanding.getAndSet(outstandingRpcs.get());
     }

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

@@ -26,6 +26,28 @@
 /** Stub for a class that will manage the health checking in the BigtableChannelPool */
 public class ChannelPoolHealthChecker {
 
+  // Configuration constants
+  private static final int WINDOW_DURATION_MINUTES = 5;
+  private static final int PROBE_RATE_SECONDS = 30;
+  private static final int PROBE_DEADLINE_MILLISECONDS = 500;
+  private static final int MIN_PROBES_FOR_EVALUATION = 4;
+  private static final int FAILURE_PERCENT_THRESHOLD = 60;
+
+  /** Inner class to represent the result of a single probe. */
+  class ProbeResult {
+    final Instant startTime;
+    final boolean success;
+
+    ProbeResult(Instant startTime, boolean success) {
+      this.startTime = startTime;
+      this.success = success;
+    }
+
+    public boolean isSuccessful() {
+      return success;
+    }
+  }
+
   // Class fields
   private final Supplier<ImmutableList<Entry>> entrySupplier;
   private Instant lastEviction;
@@ -36,7 +58,26 @@ public ChannelPoolHealthChecker(Supplier<ImmutableList<Entry>> entrySupplier) {
     this.entrySupplier = entrySupplier;
     this.lastEviction = Instant.MIN;
     this.executor = Executors.newSingleThreadScheduledExecutor();
-    // Scheduling for detectAndRemoveOutlierChannels goes here
+    // Scheduling for runProbes and detectAndRemoveOutlierChannels goes here
+
+  }
+
+  /** Stop running health checking (No-op stub) */
+  public void stop() {
+    // Method stub, no operation.
+  }
+
+  /** Runs probes on all the channels in the pool. (No-op stub) */
+  private void runProbes() {
+    // Method stub, no operation.
+    for (Entry entry: this.entrySupplier.get()) {
+      //pingAndWarm
+    }
+  }
+
+  /** Callback that will update Entry data on probe complete. (No-op stub) */
+  private void onComplete() {
+
   }
 
   /**
@@ -52,5 +93,8 @@ private Entry findOutlierEntry() {
   /** Periodically detects and removes outlier channels from the pool. (No-op stub) */
   private void detectAndRemoveOutlierEntries() {
     // Method stub, no operation.
+    for (Entry entry: this.entrySupplier.get()) {
+      //if not healthy, enterIdle
+    }
   }
 }