feat(grpc-gcp): penalize retryable channel errors

Author: rahul2393

grpc-gcp-java/src/main/java/com/google/cloud/grpc/GcpManagedChannel.java

@@ -87,6 +87,11 @@ public class GcpManagedChannel extends ManagedChannel {
   private static final Logger logger = Logger.getLogger(GcpManagedChannel.class.getName());
   static final AtomicInteger channelPoolIndex = new AtomicInteger();
 
+  @FunctionalInterface
+  interface NanoClock {
+    long get();
+  }
+
   // Counter for tracking channel ids.
   final AtomicInteger nextChannelId = new AtomicInteger();
   static final int DEFAULT_MAX_CHANNEL = 10;
@@ -173,6 +178,8 @@ private static int stateFromChannelId(int channelId) {
   private int maxScaleUpPercent = 30;
   private int maxScaleDownChannels = 2;
   private Duration drainIdleGrace = Duration.ofMinutes(1);
+  private int errorPenaltyStep = 5;
+  private Duration errorPenaltyDuration = Duration.ofSeconds(5);
   private boolean isDynamicScalingEnabled = false;
   private int maxConcurrentStreamsLowWatermark = DEFAULT_MAX_STREAM;
   private GcpManagedChannelOptions.ChannelPickStrategy channelPickStrategy =
@@ -202,6 +209,7 @@ private static int stateFromChannelId(int channelId) {
   // One-slot scale-up signal. At most one worker mutates pool size at a time.
   private final AtomicBoolean scaleUpSignalPending = new AtomicBoolean();
   private final AtomicBoolean scaleUpWorkerRunning = new AtomicBoolean();
+  private final AtomicLong totalErrorPenaltyLoad = new AtomicLong();
 
   private volatile long lastScaleUpNanos = Long.MIN_VALUE;
   private int consecutiveLowLoadChecks;
@@ -295,14 +303,14 @@ private static int stateFromChannelId(int channelId) {
   private AtomicLong scaleDownCount = new AtomicLong();
 
   // Clock supplier for nanoTime, injectable for testing.
-  private Supplier<Long> nanoClock = System::nanoTime;
+  private NanoClock nanoClock = System::nanoTime;
   private IntUnaryOperator candidateIndexPicker =
       bound -> ThreadLocalRandom.current().nextInt(bound);
   @Nullable private volatile Consumer<ChannelRef> pickerValidationHookForTest;
   @Nullable private volatile Runnable inactiveMappingRemovalHookForTest;
 
   @VisibleForTesting
-  void setNanoClock(Supplier<Long> nanoClock) {
+  void setNanoClock(NanoClock nanoClock) {
     this.nanoClock = nanoClock;
   }
 
@@ -467,6 +475,9 @@ void checkScaleDown() {
       int removeCount = Math.min(maxScaleDownChannels, Math.max(0, channelCount - desiredSize));
       removedChannels = removeChannels(removeCount);
     }
+    for (ChannelRef channelRef : removedChannels) {
+      channelRef.clearErrorPenalty();
+    }
     List<String> keysToUnbind =
         affinityKeyToChannelRef.entrySet().stream()
             .filter(entry -> removedChannels.contains(entry.getValue()))
@@ -646,11 +657,12 @@ private long activeLoad(List<ChannelRef> refs) {
     return load;
   }
 
-  private long pickerLoad(List<ChannelRef> refs) {
+  private long pickerLoad(List<ChannelRef> refs, long now) {
+    // Expiry accounting is swept before dynamicUpscale acquires the pool monitor.
     long load = 0;
     for (ChannelRef channelRef : refs) {
       if (channelRef.isActive()) {
-        load += channelRef.getPickerLoad();
+        load += channelRef.getPickerLoad(now);
       }
     }
     return load;
@@ -689,6 +701,8 @@ private void initOptions() {
       maxScaleUpPercent = poolOptions.getMaxScaleUpPercent();
       maxScaleDownChannels = poolOptions.getMaxScaleDownChannels();
       drainIdleGrace = poolOptions.getDrainIdleGrace();
+      errorPenaltyStep = poolOptions.getErrorPenaltyStep();
+      errorPenaltyDuration = poolOptions.getErrorPenaltyDuration();
       isDynamicScalingEnabled =
           minRpcPerChannel > 0 && maxRpcPerChannel > 0 && !scaleDownInterval.isZero();
       channelPickStrategy = poolOptions.getChannelPickStrategy();
@@ -2058,11 +2072,12 @@ private ChannelRef pickLeastBusyChannelDifferentFrom(@Nullable ChannelRef exclud
     }
     ChannelRef leastBusyChannelRef = null;
     int leastBusyStreams = Integer.MAX_VALUE;
+    long now = nanoClock.get();
     for (ChannelRef candidate : channelRefs) {
       if (candidate == excludedChannelRef || !candidate.isActive()) {
         continue;
       }
-      int streams = candidate.getPickerLoad();
+      int streams = candidate.getPickerLoad(now);
       if (leastBusyChannelRef == null || streams < leastBusyStreams) {
         leastBusyChannelRef = candidate;
         leastBusyStreams = streams;
@@ -2131,8 +2146,10 @@ private void maybeSignalScaleUp(ChannelRef selectedChannel) {
         || activeChannels >= maxSize) {
       return;
     }
-    if (selectedChannel.getActiveStreamsCount() <= maxRpcPerChannel
-        && ((double) totalActiveStreams.get() / activeChannels) <= maxRpcPerChannel) {
+    int selectedLoad = selectedChannel.getPickerLoad();
+    long totalLoad = (long) totalActiveStreams.get() + totalErrorPenaltyLoad.get();
+    if (selectedLoad <= maxRpcPerChannel
+        && ((double) totalLoad / activeChannels) <= maxRpcPerChannel) {
       return;
     }
     signalScaleUp();
@@ -2171,12 +2188,15 @@ private void runScaleUpWorker() {
   }
 
   private void dynamicUpscale() {
+    long now = nanoClock.get();
+    for (ChannelRef channelRef : channelRefs) {
+      channelRef.currentErrorPenalty(now);
+    }
     final int channelsToBuild;
     synchronized (this) {
       if (!isDynamicScalingEnabled || shuttingDown || channelRefs.size() >= maxSize) {
         return;
       }
-      long now = nanoClock.get();
       if (lastScaleUpNanos != Long.MIN_VALUE
           && now - lastScaleUpNanos < scaleUpCooldown.toNanos()) {
         return;
@@ -2187,7 +2207,7 @@ private void dynamicUpscale() {
       if (active == 0) {
         return;
       }
-      int desired = ceilDiv(pickerLoad(activeChannels), targetRpcPerChannel());
+      int desired = ceilDiv(pickerLoad(activeChannels, now), targetRpcPerChannel());
       int add = desired - active;
       // Small pools may add two channels per event before percentage growth dominates.
       int percentCap = Math.max(2, ceilDiv((long) active * maxScaleUpPercent, 100));
@@ -2304,12 +2324,13 @@ private ChannelRef pickLeastBusyWithFallback(boolean forFallback) {
     ChannelRef overallCandidate = null;
     int overallMinStreams = Integer.MAX_VALUE;
     int readyMaxStreams = 0;
+    long now = nanoClock.get();
 
     for (ChannelRef channelRef : channelRefs) {
       if (!channelRef.isActive()) {
         continue;
       }
-      int cnt = channelRef.getPickerLoad();
+      int cnt = channelRef.getPickerLoad(now);
       if (overallCandidate == null || cnt < overallMinStreams) {
         overallMinStreams = cnt;
         overallCandidate = channelRef;
@@ -2323,7 +2344,7 @@ private ChannelRef pickLeastBusyWithFallback(boolean forFallback) {
     }
 
     if (overallCandidate == null) {
-      return leastLoadedActiveChannel(channelRefs);
+      return leastLoadedActiveChannel(channelRefs, now);
     }
 
     // For scale-up, use maxStreams among ready channels (consistent with non-fallback path).
@@ -2396,11 +2417,20 @@ ChannelRef pickFromCandidates(List<ChannelRef> candidates) {
   }
 
   private ChannelRef leastLoadedActiveChannel(List<ChannelRef> candidates) {
+    return leastLoadedActiveChannel(candidates, nanoClock.get());
+  }
+
+  private ChannelRef leastLoadedActiveChannel(List<ChannelRef> candidates, long now) {
     ChannelRef best = null;
+    int bestLoad = Integer.MAX_VALUE;
     for (ChannelRef candidate : candidates) {
-      if (candidate.isActive()
-          && (best == null || candidate.getPickerLoad() < best.getPickerLoad())) {
+      if (!candidate.isActive()) {
+        continue;
+      }
+      int candidateLoad = candidate.getPickerLoad(now);
+      if (best == null || candidateLoad < bestLoad) {
         best = candidate;
+        bestLoad = candidateLoad;
       }
     }
     if (best != null) {
@@ -2822,6 +2852,10 @@ protected class ChannelRef {
     private final long createdNanos = nanoClock.get();
     private volatile long lastActivityNanos = createdNanos;
     private long lastResponseNanos = createdNanos;
+
+    private volatile int errorPenaltyLoad;
+    private volatile long errorPenaltyExpiresAtNanos;
+
     private final AtomicInteger deadlineExceededCount = new AtomicInteger();
     private final AtomicLong okCalls = new AtomicLong();
     private final AtomicLong errCalls = new AtomicLong();
@@ -2938,6 +2972,7 @@ protected void activeStreamsCountDecr(long startNanos, Status status, boolean fr
       if (unresponsiveDetectionEnabled) {
         detectUnresponsiveConnection(startNanos, status, fromClientSide);
       }
+      applyErrorPenalty(status);
       if (actStreams == 0 && !isActive()) {
         scheduleDrain(this);
       }
@@ -2957,7 +2992,98 @@ protected int getActiveStreamsCount() {
     }
 
     protected int getPickerLoad() {
-      return getActiveStreamsCount();
+      return (int)
+          Math.min(Integer.MAX_VALUE, (long) getActiveStreamsCount() + currentErrorPenalty());
+    }
+
+    @VisibleForTesting
+    int currentErrorPenalty() {
+      int penalty = errorPenaltyLoad;
+      if (penalty == 0) {
+        return 0;
+      }
+      return currentErrorPenalty(nanoClock.get());
+    }
+
+    private int currentErrorPenalty(long now) {
+      int penalty = errorPenaltyLoad;
+      if (penalty == 0) {
+        return 0;
+      }
+      if (now - errorPenaltyExpiresAtNanos < 0) {
+        return penalty;
+      }
+      int expiredPenalty;
+      synchronized (this) {
+        penalty = errorPenaltyLoad;
+        if (penalty == 0) {
+          return 0;
+        }
+        if (now - errorPenaltyExpiresAtNanos < 0) {
+          return penalty;
+        }
+        errorPenaltyLoad = 0;
+        errorPenaltyExpiresAtNanos = 0;
+        expiredPenalty = penalty;
+      }
+      totalErrorPenaltyLoad.addAndGet(-expiredPenalty);
+      return 0;
+    }
+
+    private int getPickerLoad(long now) {
+      int penalty = errorPenaltyLoad;
+      if (penalty != 0 && now - errorPenaltyExpiresAtNanos >= 0) {
+        penalty = 0;
+      }
+      return (int) Math.min(Integer.MAX_VALUE, (long) getActiveStreamsCount() + penalty);
+    }
+
+    private void applyErrorPenalty(Status status) {
+      if (!isDynamicScalingEnabled
+          || !active
+          || errorPenaltyStep == 0
+          || errorPenaltyDuration.isZero()
+          || errorPenaltyDuration.isNegative()
+          || (status.getCode() != Code.UNAVAILABLE
+              && status.getCode() != Code.RESOURCE_EXHAUSTED)) {
+        return;
+      }
+      long now = nanoClock.get();
+      long durationNanos = errorPenaltyDuration.toNanos();
+      long addedPenalty;
+      synchronized (this) {
+        if (!active) {
+          return;
+        }
+        int previousContribution = errorPenaltyLoad;
+        int current =
+            previousContribution != 0 && now - errorPenaltyExpiresAtNanos < 0
+                ? previousContribution
+                : 0;
+        int next = (int) Math.min(maxRpcPerChannel, (long) current + errorPenaltyStep);
+        errorPenaltyExpiresAtNanos = now + durationNanos;
+        errorPenaltyLoad = next;
+        addedPenalty = (long) next - previousContribution;
+      }
+      if (addedPenalty != 0) {
+        totalErrorPenaltyLoad.addAndGet(addedPenalty);
+      }
+      if (addedPenalty > 0) {
+        maybeSignalScaleUp(this);
+      }
+    }
+
+    private void clearErrorPenalty() {
+      int clearedPenalty;
+      synchronized (this) {
+        clearedPenalty = errorPenaltyLoad;
+        if (clearedPenalty == 0) {
+          return;
+        }
+        errorPenaltyLoad = 0;
+        errorPenaltyExpiresAtNanos = 0;
+      }
+      totalErrorPenaltyLoad.addAndGet(-clearedPenalty);
     }
 
     protected long getAndResetOkCalls() {