From f2e9cae2c68da5b7ff72cec9f0d62ed3ec2fa07b Mon Sep 17 00:00:00 2001 From: Rahul Yadav Date: Mon, 31 Aug 2026 23:12:26 +0530 Subject: [PATCH] feat(grpc-gcp): penalize retryable channel errors --- .../google/cloud/grpc/GcpManagedChannel.java | 241 +++++++- .../cloud/grpc/GcpManagedChannelOptions.java | 71 ++- .../GcpManagedChannelErrorPenaltyTest.java | 569 ++++++++++++++++++ .../grpc/GcpManagedChannelOptionsTest.java | 25 + 4 files changed, 871 insertions(+), 35 deletions(-) create mode 100644 grpc-gcp-java/src/test/java/com/google/cloud/grpc/GcpManagedChannelErrorPenaltyTest.java diff --git a/grpc-gcp-java/src/main/java/com/google/cloud/grpc/GcpManagedChannel.java b/grpc-gcp-java/src/main/java/com/google/cloud/grpc/GcpManagedChannel.java index 54cf48920cfc..64cc532ab69f 100644 --- a/grpc-gcp-java/src/main/java/com/google/cloud/grpc/GcpManagedChannel.java +++ b/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,9 @@ 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 long errorPenaltyDurationNanos = Duration.ofSeconds(5).toNanos(); private boolean isDynamicScalingEnabled = false; private int maxConcurrentStreamsLowWatermark = DEFAULT_MAX_STREAM; private GcpManagedChannelOptions.ChannelPickStrategy channelPickStrategy = @@ -202,6 +210,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 +304,14 @@ private static int stateFromChannelId(int channelId) { private AtomicLong scaleDownCount = new AtomicLong(); // Clock supplier for nanoTime, injectable for testing. - private Supplier nanoClock = System::nanoTime; + private NanoClock nanoClock = System::nanoTime; private IntUnaryOperator candidateIndexPicker = bound -> ThreadLocalRandom.current().nextInt(bound); @Nullable private volatile Consumer pickerValidationHookForTest; @Nullable private volatile Runnable inactiveMappingRemovalHookForTest; @VisibleForTesting - void setNanoClock(Supplier nanoClock) { + void setNanoClock(NanoClock nanoClock) { this.nanoClock = nanoClock; } @@ -467,6 +476,9 @@ void checkScaleDown() { int removeCount = Math.min(maxScaleDownChannels, Math.max(0, channelCount - desiredSize)); removedChannels = removeChannels(removeCount); } + for (ChannelRef channelRef : removedChannels) { + channelRef.clearErrorPenalty(); + } List keysToUnbind = affinityKeyToChannelRef.entrySet().stream() .filter(entry -> removedChannels.contains(entry.getValue())) @@ -632,6 +644,43 @@ private static int ceilDiv(long numerator, int denominator) { return (int) Math.min(Integer.MAX_VALUE, 1 + ((numerator - 1) / denominator)); } + private static int ceilMultiplyDivide(int factor, long numerator, long denominator) { + if (factor <= 0 || numerator <= 0 || denominator <= 0) { + return 0; + } + if (numerator >= denominator) { + return factor; + } + if (numerator <= Long.MAX_VALUE / factor) { + long product = factor * numerator; + return (int) (product / denominator + (product % denominator == 0 ? 0 : 1)); + } + + // Overflow-safe binary long division for unusually large configured durations or penalties. + long quotient = 0; + long remainder = 0; + for (int bit = Integer.highestOneBit(factor); bit != 0; bit >>>= 1) { + quotient <<= 1; + long denominatorMinusRemainder = denominator - remainder; + if (remainder >= denominatorMinusRemainder) { + remainder -= denominatorMinusRemainder; + quotient++; + } else { + remainder += remainder; + } + if ((factor & bit) != 0) { + long denominatorMinusNumerator = denominator - numerator; + if (remainder >= denominatorMinusNumerator) { + remainder -= denominatorMinusNumerator; + quotient++; + } else { + remainder += numerator; + } + } + } + return (int) (quotient + (remainder == 0 ? 0 : 1)); + } + private int targetRpcPerChannel() { return Math.max(1, (minRpcPerChannel + maxRpcPerChannel) / 2); } @@ -646,11 +695,12 @@ private long activeLoad(List refs) { return load; } - private long pickerLoad(List refs) { + private long pickerLoad(List 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 +739,9 @@ private void initOptions() { maxScaleUpPercent = poolOptions.getMaxScaleUpPercent(); maxScaleDownChannels = poolOptions.getMaxScaleDownChannels(); drainIdleGrace = poolOptions.getDrainIdleGrace(); + errorPenaltyStep = poolOptions.getErrorPenaltyStep(); + errorPenaltyDuration = poolOptions.getErrorPenaltyDuration(); + errorPenaltyDurationNanos = errorPenaltyDuration.toNanos(); isDynamicScalingEnabled = minRpcPerChannel > 0 && maxRpcPerChannel > 0 && !scaleDownInterval.isZero(); channelPickStrategy = poolOptions.getChannelPickStrategy(); @@ -1933,12 +1986,12 @@ protected synchronized ChannelRef getChannelRefRoundRobin() { /** * Pick a {@link ChannelRef} (and create a new one if necessary). If notReadyFallbackEnabled is * true in the {@link GcpResiliencyOptions} then instead of a channel in a non-READY state another - * channel in the READY state and having fewer than maximum allowed number of active streams will - * be provided if available. Subsequent calls with the same affinity key will provide the same + * channel in the READY state and having picker load below the maximum allowed threshold will be + * provided if available. Subsequent calls with the same affinity key will provide the same * fallback channel as long as the fallback channel is in the READY state. * * @param key affinity key. If it is specified, pick the ChannelRef bound with the affinity key. - * Otherwise pick the one with the smallest number of streams. + * Otherwise pick using the lowest picker load: active streams plus active error penalty. */ protected ChannelRef getChannelRef(@Nullable String key) { if (key == null || key.isEmpty()) { @@ -2058,11 +2111,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 +2185,11 @@ private void maybeSignalScaleUp(ChannelRef selectedChannel) { || activeChannels >= maxSize) { return; } - if (selectedChannel.getActiveStreamsCount() <= maxRpcPerChannel - && ((double) totalActiveStreams.get() / activeChannels) <= maxRpcPerChannel) { + long now = nanoClock.get(); + int selectedLoad = selectedChannel.getPickerLoad(now); + long totalLoad = (long) totalActiveStreams.get() + totalErrorPenaltyLoad.get(); + // A pool at the average cap has no spare capacity, including when penalty represents loss. + if (selectedLoad <= maxRpcPerChannel && totalLoad < (long) activeChannels * maxRpcPerChannel) { return; } signalScaleUp(); @@ -2171,12 +2228,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 +2247,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)); @@ -2248,8 +2308,8 @@ private boolean shouldScaleUp(int minStreams) { /** * Pick a {@link ChannelRef} (and create a new one if necessary). If notReadyFallbackEnabled is * true in the {@link GcpResiliencyOptions} then instead of a channel in a non-READY state another - * channel in the READY state and having fewer than maximum allowed number of active streams will - * be provided if available. + * channel in the READY state and having picker load below the maximum allowed threshold will be + * provided if available. */ private ChannelRef pickLeastBusyChannel(boolean forFallback) { // Retries cover post-snapshot deactivation, not draining density. @@ -2272,7 +2332,8 @@ private ChannelRef pickLeastBusyChannel(boolean forFallback) { * GcpManagedChannelOptions.ChannelPickStrategy}. */ private ChannelRef pickLeastBusyNoFallback() { - ChannelRef channelCandidate = pickFromCandidates(channelRefs); + long now = nanoClock.get(); + ChannelRef channelCandidate = pickFromCandidates(channelRefs, now); if (!isDynamicScalingEnabled && channelRefs.size() < maxSize) { // With power-of-two, streams distribute approximately (not exactly) evenly. // Use max streams for scale-up: if ANY channel hits the watermark, it's overloaded now @@ -2282,7 +2343,7 @@ private ChannelRef pickLeastBusyNoFallback() { int streams = channelPickStrategy == GcpManagedChannelOptions.ChannelPickStrategy.POWER_OF_TWO ? getMaxActiveStreams() - : channelCandidate.getPickerLoad(); + : channelCandidate.getPickerLoad(now); if (streams >= maxConcurrentStreamsLowWatermark) { ChannelRef newChannel = tryCreateNewChannel(); if (newChannel != null) { @@ -2304,12 +2365,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 +2385,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). @@ -2344,7 +2406,7 @@ private ChannelRef pickLeastBusyWithFallback(boolean forFallback) { if (!readyCandidates.isEmpty()) { // Apply power-of-two among eligible channels to avoid thundering herd. - ChannelRef readyCandidate = pickFromCandidates(readyCandidates); + ChannelRef readyCandidate = pickFromCandidates(readyCandidates, now); if (!forFallback && readyCandidate.getId() != overallCandidate.getId()) { if (logger.isLoggable(Level.FINEST)) { logger.finest( @@ -2377,6 +2439,10 @@ private ChannelRef pickLeastBusyWithFallback(boolean forFallback) { */ @VisibleForTesting ChannelRef pickFromCandidates(List candidates) { + return pickFromCandidates(candidates, nanoClock.get()); + } + + private ChannelRef pickFromCandidates(List candidates, long now) { Object[] snapshot = candidates.toArray(); int size = snapshot.length; if (channelPickStrategy == GcpManagedChannelOptions.ChannelPickStrategy.POWER_OF_TWO) { @@ -2386,21 +2452,30 @@ ChannelRef pickFromCandidates(List candidates) { if (!first.isActive() || !second.isActive()) { continue; } - ChannelRef picked = pickLessBusy(first, second); + ChannelRef picked = pickLessBusy(first, second, now); if (picked.isActive()) { return picked; } } } - return leastLoadedActiveChannel(candidates); + return leastLoadedActiveChannel(candidates, now); } private ChannelRef leastLoadedActiveChannel(List candidates) { + return leastLoadedActiveChannel(candidates, nanoClock.get()); + } + + private ChannelRef leastLoadedActiveChannel(List 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) { @@ -2410,8 +2485,8 @@ private ChannelRef leastLoadedActiveChannel(List candidates) { } @VisibleForTesting - ChannelRef pickLessBusy(ChannelRef first, ChannelRef second) { - return first.getPickerLoad() <= second.getPickerLoad() ? first : second; + ChannelRef pickLessBusy(ChannelRef first, ChannelRef second, long now) { + return first.getPickerLoad(now) <= second.getPickerLoad(now) ? first : second; } @Override @@ -2822,6 +2897,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 +3017,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 +3037,116 @@ 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; + long expiresAtNanos = errorPenaltyExpiresAtNanos; + if (penalty == 0 || expiresAtNanos == 0) { + return 0; + } + if (now - expiresAtNanos < 0) { + return decayedErrorPenalty(penalty, now, expiresAtNanos); + } + int expiredPenalty; + synchronized (this) { + penalty = errorPenaltyLoad; + expiresAtNanos = errorPenaltyExpiresAtNanos; + if (penalty == 0 || expiresAtNanos == 0) { + return 0; + } + if (now - expiresAtNanos < 0) { + return decayedErrorPenalty(penalty, now, expiresAtNanos); + } + errorPenaltyLoad = 0; + errorPenaltyExpiresAtNanos = 0; + expiredPenalty = penalty; + } + totalErrorPenaltyLoad.addAndGet(-expiredPenalty); + return 0; + } + + private int getPickerLoad(long now) { + int penalty = errorPenaltyLoad; + long expiresAtNanos = errorPenaltyExpiresAtNanos; + if (penalty != 0) { + penalty = decayedErrorPenalty(penalty, now, expiresAtNanos); + } + return (int) Math.min(Integer.MAX_VALUE, (long) getActiveStreamsCount() + penalty); + } + + private int decayedErrorPenalty(int penalty, long now, long expiresAtNanos) { + if (expiresAtNanos == 0) { + return 0; + } + long remainingNanos = expiresAtNanos - now; + if (remainingNanos <= 0) { + return 0; + } + // Picker steering decays smoothly. Aggregate accounting deliberately retains the full + // contribution until expiry or clear, making scale-up load a conservative upper bound while + // preserving one atomic net delta per aggregate transition. + return ceilMultiplyDivide( + penalty, Math.min(remainingNanos, errorPenaltyDurationNanos), errorPenaltyDurationNanos); + } + + private void applyErrorPenalty(Status status) { + if (!isDynamicScalingEnabled + || !active + || errorPenaltyStep == 0 + || (status.getCode() != Code.UNAVAILABLE + && status.getCode() != Code.RESOURCE_EXHAUSTED)) { + return; + } + long now = nanoClock.get(); + long addedPenalty; + synchronized (this) { + if (!active) { + return; + } + int previousContribution = errorPenaltyLoad; + long previousExpiry = errorPenaltyExpiresAtNanos; + int current = + previousContribution != 0 && previousExpiry != 0 && now - previousExpiry < 0 + ? previousContribution + : 0; + int next = (int) Math.min(maxRpcPerChannel, (long) current + errorPenaltyStep); + long nextExpiry = now + errorPenaltyDurationNanos; + // Zero is reserved as the cleared expiry sentinel. + errorPenaltyExpiresAtNanos = nextExpiry == 0 ? 1 : nextExpiry; + 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() { diff --git a/grpc-gcp-java/src/main/java/com/google/cloud/grpc/GcpManagedChannelOptions.java b/grpc-gcp-java/src/main/java/com/google/cloud/grpc/GcpManagedChannelOptions.java index 60d18b16982c..0b6145afa685 100644 --- a/grpc-gcp-java/src/main/java/com/google/cloud/grpc/GcpManagedChannelOptions.java +++ b/grpc-gcp-java/src/main/java/com/google/cloud/grpc/GcpManagedChannelOptions.java @@ -39,8 +39,9 @@ public class GcpManagedChannelOptions { */ public enum ChannelPickStrategy { /** - * Scans all channels and picks the one with the fewest active streams. Ties are broken by - * iteration order (lowest index wins). This is the legacy behavior. + * Scans all channels and picks the one with the lowest picker load: active streams plus active + * error penalty. Ties are broken by iteration order (lowest index wins). This is the legacy + * behavior. * *

This strategy finds the global minimum but is susceptible to the thundering herd problem: * under burst traffic, all concurrent callers observe the same minimum and pile onto the same @@ -49,12 +50,13 @@ public enum ChannelPickStrategy { LINEAR_SCAN, /** - * Samples two channels at random with replacement and returns the one with fewer active - * streams. The first sample wins ties. Inactive samples are retried. + * Samples two channels at random with replacement and returns the one with lower picker load. + * The first sample wins ties, with no channel-warmth preference. Inactive or draining samples + * are retried. * - *

This is the default strategy. It avoids the thundering herd problem without preferring - * channel warmth. The trade-off is that it may not always find the global minimum, but in - * practice the difference is negligible because stream counts are inherently racy. + *

This is the default strategy. It avoids the thundering herd problem. The trade-off is that + * it may not always find the global minimum, but in practice the difference is negligible + * because picker load is inherently racy. */ POWER_OF_TWO, } @@ -220,6 +222,8 @@ public static class GcpChannelPoolOptions { // Maximum channels removed in one scale-down check. private final int maxScaleDownChannels; private final Duration drainIdleGrace; + private final int errorPenaltyStep; + private final Duration errorPenaltyDuration; // Use round-robin channel selection for affinity binding calls. private final boolean useRoundRobinOnBind; @@ -242,6 +246,8 @@ public GcpChannelPoolOptions(Builder builder) { maxScaleUpPercent = builder.maxScaleUpPercent; maxScaleDownChannels = builder.maxScaleDownChannels; drainIdleGrace = builder.drainIdleGrace; + errorPenaltyStep = builder.errorPenaltyStep; + errorPenaltyDuration = builder.errorPenaltyDuration; concurrentStreamsLowWatermark = builder.concurrentStreamsLowWatermark; useRoundRobinOnBind = builder.useRoundRobinOnBind; affinityKeyLifetime = builder.affinityKeyLifetime; @@ -293,6 +299,14 @@ public Duration getDrainIdleGrace() { return drainIdleGrace; } + public int getErrorPenaltyStep() { + return errorPenaltyStep; + } + + public Duration getErrorPenaltyDuration() { + return errorPenaltyDuration; + } + public int getConcurrentStreamsLowWatermark() { return concurrentStreamsLowWatermark; } @@ -329,7 +343,8 @@ public String toString() { "{maxSize: %d, minSize: %d, initSize: %d, minRpcPerChannel: %d, " + "maxRpcPerChannel: %d, scaleDownInterval: %s, scaleUpCooldown: %s, " + "scaleDownConsecutiveLowLoadChecks: %d, maxScaleUpPercent: %d, " - + "maxScaleDownChannels: %d, drainIdleGrace: %s, " + + "maxScaleDownChannels: %d, drainIdleGrace: %s, errorPenaltyStep: %d, " + + "errorPenaltyDuration: %s, " + "concurrentStreamsLowWatermark: %d, useRoundRobinOnBind: %s, " + "affinityKeyLifetime: %s, cleanupInterval: %s, channelPickStrategy: %s}", getMaxSize(), @@ -343,6 +358,8 @@ public String toString() { getMaxScaleUpPercent(), getMaxScaleDownChannels(), getDrainIdleGrace(), + getErrorPenaltyStep(), + getErrorPenaltyDuration(), getConcurrentStreamsLowWatermark(), isUseRoundRobinOnBind(), getAffinityKeyLifetime(), @@ -362,6 +379,8 @@ public static class Builder { 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 int concurrentStreamsLowWatermark = GcpManagedChannel.DEFAULT_MAX_STREAM; private boolean useRoundRobinOnBind = false; private Duration affinityKeyLifetime = Duration.ZERO; @@ -386,6 +405,8 @@ public Builder(GcpChannelPoolOptions options) { this.maxScaleUpPercent = options.getMaxScaleUpPercent(); this.maxScaleDownChannels = options.getMaxScaleDownChannels(); this.drainIdleGrace = options.getDrainIdleGrace(); + this.errorPenaltyStep = options.getErrorPenaltyStep(); + this.errorPenaltyDuration = options.getErrorPenaltyDuration(); this.concurrentStreamsLowWatermark = options.getConcurrentStreamsLowWatermark(); this.useRoundRobinOnBind = options.isUseRoundRobinOnBind(); this.affinityKeyLifetime = options.getAffinityKeyLifetime(); @@ -528,6 +549,38 @@ public Builder setDrainIdleGrace(Duration drainIdleGrace) { return this; } + /** + * Sets the load penalty added after each retryable channel error. A value of 0 disables error + * penalties. Each step accumulates on the stored undecayed penalty, capped at {@code + * maxRpcPerChannel}. Must not be negative. Defaults to 5. + */ + public Builder setErrorPenaltyStep(int errorPenaltyStep) { + Preconditions.checkArgument( + errorPenaltyStep >= 0, "Error penalty step must not be negative."); + this.errorPenaltyStep = errorPenaltyStep; + return this; + } + + /** + * Sets how long an applied penalty takes to decay linearly to zero. Each new retryable error + * resets the decay timer to a full window. Must be positive. Defaults to 5 seconds. + */ + public Builder setErrorPenaltyDuration(Duration errorPenaltyDuration) { + Preconditions.checkNotNull( + errorPenaltyDuration, "Error penalty duration must not be null."); + Preconditions.checkArgument( + !errorPenaltyDuration.isNegative() && !errorPenaltyDuration.isZero(), + "Error penalty duration must be positive."); + try { + errorPenaltyDuration.toNanos(); + } catch (ArithmeticException failure) { + throw new IllegalArgumentException( + "Error penalty duration must fit in nanoseconds.", failure); + } + this.errorPenaltyDuration = errorPenaltyDuration; + return this; + } + /** * Sets the concurrent streams low watermark. If every channel in the pool has at least this * amount of concurrent streams then a new channel will be created in the pool unless the pool @@ -590,7 +643,7 @@ public Builder setCleanupInterval(Duration cleanupInterval) { * sample wins ties, with no channel-warmth preference. Inactive samples are retried. * *

Use {@link ChannelPickStrategy#LINEAR_SCAN} to restore the legacy behavior of scanning - * all channels and always picking the one with the fewest active streams. + * all channels and always picking the one with the lowest picker load. * * @param strategy the channel pick strategy to use. */ diff --git a/grpc-gcp-java/src/test/java/com/google/cloud/grpc/GcpManagedChannelErrorPenaltyTest.java b/grpc-gcp-java/src/test/java/com/google/cloud/grpc/GcpManagedChannelErrorPenaltyTest.java new file mode 100644 index 000000000000..6f8867b6f4f5 --- /dev/null +++ b/grpc-gcp-java/src/test/java/com/google/cloud/grpc/GcpManagedChannelErrorPenaltyTest.java @@ -0,0 +1,569 @@ +/* + * Copyright 2026 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.google.cloud.grpc; + +import static com.google.common.truth.Truth.assertThat; +import static org.awaitility.Awaitility.await; +import static org.junit.Assert.assertThrows; + +import com.google.cloud.grpc.GcpManagedChannel.ChannelRef; +import com.google.cloud.grpc.GcpManagedChannelOptions.GcpChannelPoolOptions; +import com.google.common.util.concurrent.MoreExecutors; +import io.grpc.Status; +import java.lang.reflect.Field; +import java.lang.reflect.Method; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.After; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +@RunWith(JUnit4.class) +public final class GcpManagedChannelErrorPenaltyTest { + private final ExecutorService executor = MoreExecutors.newDirectExecutorService(); + private GcpManagedChannel pool; + + @After + public void tearDown() { + if (pool != null) { + pool.shutdownNow(); + } + executor.shutdownNow(); + } + + @Test + public void retryableErrorsAddBoundedDecayingPickerLoad() { + AtomicLong clock = new AtomicLong(1); + pool = newPool(2, 2, 2, 20, 10, Duration.ofSeconds(10)); + pool.setNanoClock(clock::get); + ChannelRef penalized = pool.channelRefs.get(0); + ChannelRef healthy = pool.channelRefs.get(1); + healthy.setActiveStreamsForTest(12); + + completeWithError(penalized, clock.get(), Status.UNAVAILABLE); + completeWithError(penalized, clock.get(), Status.RESOURCE_EXHAUSTED); + assertThat(penalized.currentErrorPenalty()).isEqualTo(20); + assertThat(totalErrorPenaltyLoad()).isEqualTo(20); + + AtomicInteger sample = new AtomicInteger(); + pool.setCandidateIndexPickerForTest(ignored -> sample.getAndIncrement() % 2); + assertThat(pool.pickFromCandidates(pool.channelRefs)).isSameInstanceAs(healthy); + + int previousPenalty = penalized.currentErrorPenalty(); + clock.addAndGet(Duration.ofSeconds(2).toNanos()); + assertThat(penalized.currentErrorPenalty()).isEqualTo(16); + assertThat(penalized.currentErrorPenalty()).isAtMost(previousPenalty); + previousPenalty = penalized.currentErrorPenalty(); + + clock.addAndGet(Duration.ofSeconds(3).toNanos()); + assertThat(penalized.currentErrorPenalty()).isEqualTo(10); + assertThat(penalized.currentErrorPenalty()).isAtMost(previousPenalty); + assertThat(totalErrorPenaltyLoad()).isEqualTo(20); + + sample.set(0); + assertThat(pool.pickFromCandidates(pool.channelRefs)).isSameInstanceAs(penalized); + pool.setCandidateIndexPickerForTest(ignored -> 1); + assertThat(pool.pickFromCandidates(pool.channelRefs)).isSameInstanceAs(healthy); + + previousPenalty = penalized.currentErrorPenalty(); + clock.addAndGet(Duration.ofSeconds(4).toNanos()); + assertThat(penalized.currentErrorPenalty()).isEqualTo(2); + assertThat(penalized.currentErrorPenalty()).isAtMost(previousPenalty); + assertThat(totalErrorPenaltyLoad()).isEqualTo(20); + + clock.addAndGet(Duration.ofSeconds(1).toNanos()); + assertThat(penalized.currentErrorPenalty()).isEqualTo(0); + assertThat(totalErrorPenaltyLoad()).isEqualTo(0); + } + + @Test + public void linearDecayHandlesMultiplicationOverflow() { + AtomicLong clock = new AtomicLong(1); + pool = newPool(2, Integer.MAX_VALUE, Integer.MAX_VALUE, Duration.ofNanos(Long.MAX_VALUE)); + pool.setNanoClock(clock::get); + ChannelRef channel = pool.channelRefs.get(0); + completeWithError(channel, clock.get(), Status.UNAVAILABLE); + + clock.addAndGet(Long.MAX_VALUE / 2); + + assertThat(channel.currentErrorPenalty()).isEqualTo(1_073_741_824); + assertThat(totalErrorPenaltyLoad()).isEqualTo(Integer.MAX_VALUE); + + clock.addAndGet(Long.MAX_VALUE - Long.MAX_VALUE / 2 - 1); + assertThat(channel.currentErrorPenalty()).isEqualTo(1); + assertThat(totalErrorPenaltyLoad()).isEqualTo(Integer.MAX_VALUE); + + clock.incrementAndGet(); + assertThat(channel.currentErrorPenalty()).isEqualTo(0); + assertThat(totalErrorPenaltyLoad()).isEqualTo(0); + } + + @Test + public void retryableErrorDuringDecayRestoresAccumulatedPenalty() { + AtomicLong clock = new AtomicLong(1); + pool = newPool(2, 10, 5, Duration.ofSeconds(10)); + pool.setNanoClock(clock::get); + ChannelRef channel = pool.channelRefs.get(0); + completeWithError(channel, clock.get(), Status.UNAVAILABLE); + clock.addAndGet(Duration.ofSeconds(5).toNanos()); + assertThat(channel.currentErrorPenalty()).isEqualTo(3); + + completeWithError(channel, clock.get(), Status.UNAVAILABLE); + + assertThat(channel.currentErrorPenalty()).isEqualTo(10); + assertThat(totalErrorPenaltyLoad()).isEqualTo(10); + } + + @Test + public void penaltyStepSaturatesWithoutIntegerOverflow() { + AtomicLong clock = new AtomicLong(1); + pool = newPool(2, Integer.MAX_VALUE, Integer.MAX_VALUE, Duration.ofSeconds(5)); + pool.setNanoClock(clock::get); + ChannelRef channel = pool.channelRefs.get(0); + + completeWithError(channel, clock.get(), Status.UNAVAILABLE); + completeWithError(channel, clock.get(), Status.RESOURCE_EXHAUSTED); + + assertThat(channel.currentErrorPenalty()).isEqualTo(Integer.MAX_VALUE); + assertThat(totalErrorPenaltyLoad()).isEqualTo(Integer.MAX_VALUE); + } + + @Test + public void pickerLoadSaturatesWithoutIntegerOverflow() { + AtomicLong clock = new AtomicLong(1); + pool = newPool(2, Integer.MAX_VALUE, Integer.MAX_VALUE, Duration.ofSeconds(5)); + pool.setNanoClock(clock::get); + ChannelRef channel = pool.channelRefs.get(0); + completeWithError(channel, clock.get(), Status.UNAVAILABLE); + channel.setActiveStreamsForTest(1); + + assertThat(channel.getPickerLoad()).isEqualTo(Integer.MAX_VALUE); + } + + @Test + public void aggregatePenaltyUsesLongWithoutOverflow() { + AtomicLong clock = new AtomicLong(1); + pool = newPool(2, Integer.MAX_VALUE, Integer.MAX_VALUE, Duration.ofSeconds(5)); + pool.setNanoClock(clock::get); + + completeWithError(pool.channelRefs.get(0), clock.get(), Status.UNAVAILABLE); + completeWithError(pool.channelRefs.get(1), clock.get(), Status.UNAVAILABLE); + + assertThat(totalErrorPenaltyLoad()).isEqualTo(2L * Integer.MAX_VALUE); + } + + @Test + public void expiredPenaltyReplacementUsesOneNetAggregateDelta() { + AtomicLong clock = new AtomicLong(1); + pool = newPool(2, 10, 5, Duration.ofSeconds(5)); + pool.setNanoClock(clock::get); + ChannelRef channel = pool.channelRefs.get(0); + completeWithError(channel, clock.get(), Status.UNAVAILABLE); + completeWithError(channel, clock.get(), Status.UNAVAILABLE); + assertThat(totalErrorPenaltyLoad()).isEqualTo(10); + + clock.addAndGet(Duration.ofSeconds(6).toNanos()); + completeWithError(channel, clock.get(), Status.UNAVAILABLE); + + assertThat(channel.currentErrorPenalty()).isEqualTo(5); + assertThat(totalErrorPenaltyLoad()).isEqualTo(5); + } + + @Test + public void concurrentExpirySubtractsAggregateExactlyOnce() throws Exception { + AtomicLong clock = new AtomicLong(1); + pool = newPool(); + pool.setNanoClock(clock::get); + ChannelRef channel = pool.channelRefs.get(0); + completeWithError(channel, clock.get(), Status.UNAVAILABLE); + clock.addAndGet(Duration.ofSeconds(6).toNanos()); + CountDownLatch start = new CountDownLatch(1); + ExecutorService readers = Executors.newFixedThreadPool(8); + try { + List> results = new ArrayList<>(); + for (int i = 0; i < 100; i++) { + results.add( + readers.submit( + () -> { + start.await(); + return channel.currentErrorPenalty(); + })); + } + start.countDown(); + for (Future result : results) { + assertThat(result.get()).isEqualTo(0); + } + assertThat(totalErrorPenaltyLoad()).isEqualTo(0); + } finally { + start.countDown(); + readers.shutdownNow(); + } + } + + @Test + public void nonRetryableErrorsAddNoPenalty() { + AtomicLong clock = new AtomicLong(1); + pool = newPool(); + pool.setNanoClock(clock::get); + ChannelRef channel = pool.channelRefs.get(0); + + completeWithError(channel, clock.get(), Status.OK); + completeWithError(channel, clock.get(), Status.INTERNAL); + completeWithError(channel, clock.get(), Status.DEADLINE_EXCEEDED); + completeWithError(channel, clock.get(), Status.CANCELLED); + + assertThat(channel.currentErrorPenalty()).isEqualTo(0); + assertThat(totalErrorPenaltyLoad()).isEqualTo(0); + } + + @Test + public void negativeAndZeroDurationRejected() { + assertThrows( + IllegalArgumentException.class, + () -> GcpChannelPoolOptions.newBuilder().setErrorPenaltyDuration(Duration.ofSeconds(-1))); + assertThrows( + IllegalArgumentException.class, + () -> GcpChannelPoolOptions.newBuilder().setErrorPenaltyDuration(Duration.ZERO)); + } + + @Test + public void penaltyContributesToScaleUpSignal() { + AtomicLong clock = new AtomicLong(1); + pool = newPool(2, 2, 3, 2, 2, Duration.ofSeconds(5)); + pool.setNanoClock(clock::get); + ChannelRef channel = pool.channelRefs.get(0); + channel.activeStreamsCountIncr(); + channel.activeStreamsCountIncr(); + + channel.activeStreamsCountDecr(clock.get(), Status.UNAVAILABLE, false); + + await().atMost(Duration.ofSeconds(5)).until(() -> pool.getNumberOfChannels() == 3); + } + + @Test + public void fullyPenalizedIdlePoolTriggersOneBoundedScaleUp() { + AtomicLong clock = new AtomicLong(1); + pool = newPool(2, 2, 6, 2, 2, Duration.ofMinutes(1)); + pool.setNanoClock(clock::get); + ChannelRef first = pool.channelRefs.get(0); + ChannelRef second = pool.channelRefs.get(1); + + completeWithError(first, clock.get(), Status.UNAVAILABLE); + completeWithError(second, clock.get(), Status.UNAVAILABLE); + + await().atMost(Duration.ofSeconds(5)).until(() -> pool.getNumberOfChannels() == 4); + assertThat(first.getActiveStreamsCount()).isEqualTo(0); + assertThat(second.getActiveStreamsCount()).isEqualTo(0); + assertThat(totalErrorPenaltyLoad()).isEqualTo(4); + + clock.addAndGet(Duration.ofSeconds(30).toNanos()); + assertThat(first.currentErrorPenalty()).isEqualTo(1); + assertThat(second.currentErrorPenalty()).isEqualTo(1); + first.activeStreamsCountIncr(); + first.activeStreamsCountDecr(clock.get(), Status.OK, false); + await() + .during(Duration.ofMillis(200)) + .atMost(Duration.ofSeconds(2)) + .until(() -> pool.getNumberOfChannels() == 4); + + clock.addAndGet(Duration.ofSeconds(30).toNanos()); + assertThat(first.currentErrorPenalty()).isEqualTo(0); + assertThat(second.currentErrorPenalty()).isEqualTo(0); + assertThat(totalErrorPenaltyLoad()).isEqualTo(0); + second.activeStreamsCountIncr(); + second.activeStreamsCountDecr(clock.get(), Status.OK, false); + await() + .during(Duration.ofMillis(200)) + .atMost(Duration.ofSeconds(2)) + .until(() -> pool.getNumberOfChannels() == 4); + } + + @Test + public void scaleDownClearsRemovedChannelPenalty() { + AtomicLong clock = new AtomicLong(1); + pool = newPool(1, 10, 5, Duration.ofSeconds(5)); + pool.setNanoClock(clock::get); + ChannelRef removed = pool.channelRefs.get(0); + completeWithError(removed, clock.get(), Status.UNAVAILABLE); + assertThat(totalErrorPenaltyLoad()).isEqualTo(5); + + pool.checkScaleDown(); + pool.checkScaleDown(); + pool.checkScaleDown(); + + assertThat(removed.isActive()).isFalse(); + assertThat(removed.currentErrorPenalty()).isEqualTo(0); + assertThat(totalErrorPenaltyLoad()).isEqualTo(0); + } + + @Test + public void rpcCompletionAfterScaleDownDoesNotLeakPenalty() throws Exception { + AtomicLong clock = new AtomicLong(1); + pool = newPool(1, 10, 5, Duration.ofSeconds(5)); + pool.setNanoClock(clock::get); + ChannelRef victim = pool.channelRefs.get(0); + victim.activeStreamsCountIncr(); + CountDownLatch completionStarted = new CountDownLatch(1); + AtomicReference completionThread = new AtomicReference<>(); + ExecutorService completionExecutor = + Executors.newSingleThreadExecutor( + command -> { + Thread thread = new Thread(command, "penalty-completion"); + completionThread.set(thread); + return thread; + }); + try { + Future completion; + synchronized (victim) { + completion = + completionExecutor.submit( + () -> { + completionStarted.countDown(); + victim.activeStreamsCountDecr(clock.get(), Status.UNAVAILABLE, false); + }); + await().atMost(Duration.ofSeconds(5)).until(() -> completionStarted.getCount() == 0); + await() + .atMost(Duration.ofSeconds(5)) + .until( + () -> + completionThread.get() != null + && completionThread.get().getState() == Thread.State.BLOCKED); + + pool.checkScaleDown(); + pool.checkScaleDown(); + pool.checkScaleDown(); + + assertThat(victim.isActive()).isFalse(); + assertThat(pool.channelRefs).doesNotContain(victim); + assertThat(totalErrorPenaltyLoad()).isEqualTo(0); + } + + completion.get(5, TimeUnit.SECONDS); + assertThat(victim.currentErrorPenalty()).isEqualTo(0); + assertThat(totalErrorPenaltyLoad()).isEqualTo(0); + + clock.addAndGet(Duration.ofHours(1).toNanos()); + pool.checkScaleDown(); + assertThat(totalErrorPenaltyLoad()).isEqualTo(0); + } finally { + completionExecutor.shutdownNow(); + } + } + + @Test + public void leastLoadedActiveChannelReadsClockOncePerScan() throws Exception { + AtomicLong clockReads = new AtomicLong(); + pool = newPool(4, 2, 4, 10, 5, Duration.ofSeconds(5)); + pool.setNanoClock( + () -> { + clockReads.incrementAndGet(); + return 1_000; + }); + ChannelRef penalized = pool.channelRefs.get(0); + completeWithError(penalized, 1_000, Status.UNAVAILABLE); + for (ChannelRef channelRef : pool.channelRefs) { + if (channelRef != penalized) { + channelRef.setActiveStreamsForTest(10); + } + } + clockReads.set(0); + Method leastLoaded = + GcpManagedChannel.class.getDeclaredMethod("leastLoadedActiveChannel", List.class); + leastLoaded.setAccessible(true); + + leastLoaded.invoke(pool, pool.channelRefs); + + assertThat(clockReads.get()).isEqualTo(1); + } + + @Test + public void powerOfTwoReadsClockOncePerPick() { + AtomicLong clockReads = new AtomicLong(); + pool = newPool(); + pool.setNanoClock( + () -> { + clockReads.incrementAndGet(); + return 1_000; + }); + completeWithError(pool.channelRefs.get(0), 1_000, Status.UNAVAILABLE); + completeWithError(pool.channelRefs.get(1), 1_000, Status.UNAVAILABLE); + AtomicInteger sample = new AtomicInteger(); + pool.setCandidateIndexPickerForTest(ignored -> sample.getAndIncrement() % 2); + clockReads.set(0); + + pool.pickFromCandidates(pool.channelRefs); + + assertThat(clockReads.get()).isEqualTo(1); + } + + @Test + public void pickerAndScaleUpReadsDoNotClearPenaltyUnderPoolMonitor() throws Exception { + AtomicLong clock = new AtomicLong(1); + pool = newPool(2, 2, 3, 10, 5, Duration.ofSeconds(5)); + pool.setNanoClock(clock::get); + ChannelRef channel = pool.channelRefs.get(0); + ChannelRef other = pool.channelRefs.get(1); + completeWithError(channel, clock.get(), Status.UNAVAILABLE); + clock.addAndGet(Duration.ofSeconds(6).toNanos()); + channel.deactivateForTest(); + other.deactivateForTest(); + Field active = ChannelRef.class.getDeclaredField("active"); + active.setAccessible(true); + AtomicInteger sample = new AtomicInteger(); + pool.setCandidateIndexPickerForTest( + ignored -> { + int index = sample.getAndIncrement(); + if (index == 0) { + try { + active.setBoolean(channel, true); + active.setBoolean(other, true); + } catch (IllegalAccessException failure) { + throw new AssertionError(failure); + } + } + return index % 2; + }); + Method pickerLoad = + GcpManagedChannel.class.getDeclaredMethod("pickerLoad", List.class, long.class); + pickerLoad.setAccessible(true); + + synchronized (pool) { + assertThat((long) pickerLoad.invoke(pool, pool.channelRefs, clock.get())).isEqualTo(0); + assertThat(totalErrorPenaltyLoad()).isEqualTo(5); + pool.getChannelRefRoundRobin(); + assertThat(totalErrorPenaltyLoad()).isEqualTo(5); + channel.activeStreamsCountIncr(); + assertThat(totalErrorPenaltyLoad()).isEqualTo(5); + channel.activeStreamsCountDecr(clock.get(), Status.OK, false); + } + + assertThat(channel.currentErrorPenalty()).isEqualTo(0); + assertThat(totalErrorPenaltyLoad()).isEqualTo(0); + } + + @Test + public void clearedExpirySentinelWinsWithNegativeClock() throws Exception { + pool = newPool(); + pool.setNanoClock(() -> -1); + ChannelRef channel = pool.channelRefs.get(0); + Field penaltyLoad = ChannelRef.class.getDeclaredField("errorPenaltyLoad"); + penaltyLoad.setAccessible(true); + penaltyLoad.setInt(channel, 5); + Field expiresAt = ChannelRef.class.getDeclaredField("errorPenaltyExpiresAtNanos"); + expiresAt.setAccessible(true); + expiresAt.setLong(channel, 0); + + assertThat(channel.currentErrorPenalty()).isEqualTo(0); + assertThat(channel.getPickerLoad()).isEqualTo(0); + assertThat(totalErrorPenaltyLoad()).isEqualTo(0); + } + + @Test + public void appliedPenaltyAvoidsClearedExpirySentinel() { + long durationNanos = Duration.ofSeconds(5).toNanos(); + AtomicLong clock = new AtomicLong(-durationNanos); + pool = newPool(); + pool.setNanoClock(clock::get); + ChannelRef channel = pool.channelRefs.get(0); + + completeWithError(channel, clock.get(), Status.UNAVAILABLE); + + assertThat(channel.currentErrorPenalty()).isEqualTo(5); + assertThat(channel.getPickerLoad()).isEqualTo(5); + assertThat(totalErrorPenaltyLoad()).isEqualTo(5); + + clock.set(0); + assertThat(channel.currentErrorPenalty()).isEqualTo(1); + assertThat(totalErrorPenaltyLoad()).isEqualTo(5); + + clock.set(1); + assertThat(channel.currentErrorPenalty()).isEqualTo(0); + assertThat(totalErrorPenaltyLoad()).isEqualTo(0); + } + + @Test + public void zeroPenaltyStepDisablesPenalties() { + AtomicLong clock = new AtomicLong(1); + pool = newPool(2, 10, 0, Duration.ofSeconds(5)); + pool.setNanoClock(clock::get); + ChannelRef channel = pool.channelRefs.get(0); + + completeWithError(channel, clock.get(), Status.UNAVAILABLE); + + assertThat(channel.currentErrorPenalty()).isEqualTo(0); + assertThat(totalErrorPenaltyLoad()).isEqualTo(0); + } + + private GcpManagedChannel newPool() { + return newPool(2, 10, 5, Duration.ofSeconds(5)); + } + + private GcpManagedChannel newPool( + int minimum, int maximumRpc, int penaltyStep, Duration penaltyDuration) { + return newPool(2, minimum, 2, maximumRpc, penaltyStep, penaltyDuration); + } + + private GcpManagedChannel newPool( + int initial, + int minimum, + int maximum, + int maximumRpc, + int penaltyStep, + Duration penaltyDuration) { + GcpChannelPoolOptions options = + GcpChannelPoolOptions.newBuilder() + .setInitSize(initial) + .setMinSize(minimum) + .setMaxSize(maximum) + .setDynamicScaling(1, maximumRpc, Duration.ofMinutes(1)) + .setErrorPenaltyStep(penaltyStep) + .setErrorPenaltyDuration(penaltyDuration) + .build(); + return (GcpManagedChannel) + GcpManagedChannelBuilder.forDelegateBuilder( + new GcpManagedChannelTest.FakeManagedChannelBuilder( + () -> new GcpManagedChannelTest.FakeManagedChannel(executor))) + .withOptions( + GcpManagedChannelOptions.newBuilder().withChannelPoolOptions(options).build()) + .build(); + } + + private static void completeWithError(ChannelRef channel, long now, Status status) { + channel.activeStreamsCountIncr(); + channel.activeStreamsCountDecr(now, status, false); + } + + private long totalErrorPenaltyLoad() { + try { + Field field = GcpManagedChannel.class.getDeclaredField("totalErrorPenaltyLoad"); + field.setAccessible(true); + return ((AtomicLong) field.get(pool)).get(); + } catch (ReflectiveOperationException failure) { + throw new AssertionError(failure); + } + } +} diff --git a/grpc-gcp-java/src/test/java/com/google/cloud/grpc/GcpManagedChannelOptionsTest.java b/grpc-gcp-java/src/test/java/com/google/cloud/grpc/GcpManagedChannelOptionsTest.java index 2a9aa1b2b6e8..27651fcbf377 100644 --- a/grpc-gcp-java/src/test/java/com/google/cloud/grpc/GcpManagedChannelOptionsTest.java +++ b/grpc-gcp-java/src/test/java/com/google/cloud/grpc/GcpManagedChannelOptionsTest.java @@ -215,16 +215,31 @@ public void testDynamicScalingKnobsHaveDefaultsAndSurviveCopy() { assertThat(defaults.getScaleUpCooldown()).isEqualTo(Duration.ofSeconds(10)); assertThat(defaults.getMaxScaleUpPercent()).isEqualTo(30); + assertThat(defaults.getErrorPenaltyStep()).isEqualTo(5); + assertThat(defaults.getErrorPenaltyDuration()).isEqualTo(Duration.ofSeconds(5)); + + GcpChannelPoolOptions copiedDefaults = GcpChannelPoolOptions.newBuilder(defaults).build(); + assertThat(copiedDefaults.getErrorPenaltyStep()).isEqualTo(5); + assertThat(copiedDefaults.getErrorPenaltyDuration()).isEqualTo(Duration.ofSeconds(5)); + assertThat(copiedDefaults.toString()).contains("errorPenaltyStep: 5"); GcpChannelPoolOptions configured = GcpChannelPoolOptions.newBuilder(defaults) .setScaleUpCooldown(Duration.ofSeconds(1)) .setMaxScaleUpPercent(40) + .setErrorPenaltyStep(2) + .setErrorPenaltyDuration(Duration.ofSeconds(3)) .build(); GcpChannelPoolOptions copied = GcpChannelPoolOptions.newBuilder(configured).build(); assertThat(copied.getScaleUpCooldown()).isEqualTo(Duration.ofSeconds(1)); assertThat(copied.getMaxScaleUpPercent()).isEqualTo(40); + assertThat(copied.getErrorPenaltyStep()).isEqualTo(2); + assertThat(copied.getErrorPenaltyDuration()).isEqualTo(Duration.ofSeconds(3)); + + GcpChannelPoolOptions disabled = + GcpChannelPoolOptions.newBuilder().setErrorPenaltyStep(0).build(); + assertThat(disabled.getErrorPenaltyStep()).isEqualTo(0); } @Test @@ -238,6 +253,14 @@ public void dynamicScalingDefaultKnobsRejectNegativeAndDefaultZeroCooldown() { .build() .getScaleUpCooldown()) .isEqualTo(Duration.ofSeconds(10)); + assertThrows( + IllegalArgumentException.class, + () -> GcpChannelPoolOptions.newBuilder().setErrorPenaltyStep(-1)); + assertThrows( + IllegalArgumentException.class, + () -> + GcpChannelPoolOptions.newBuilder() + .setErrorPenaltyDuration(Duration.ofSeconds(Long.MAX_VALUE))); } @Test @@ -255,6 +278,8 @@ public void channelPoolOptionsToStringIncludesEveryKnob() { assertThat(options).contains("maxScaleUpPercent:"); assertThat(options).contains("maxScaleDownChannels:"); assertThat(options).contains("drainIdleGrace:"); + assertThat(options).contains("errorPenaltyStep:"); + assertThat(options).contains("errorPenaltyDuration:"); assertThat(options).contains("concurrentStreamsLowWatermark:"); assertThat(options).contains("useRoundRobinOnBind:"); assertThat(options).contains("affinityKeyLifetime:");