diff --git a/grpc-gcp-java/src/main/java/com/google/cloud/grpc/fallback/GcpFallbackChannel.java b/grpc-gcp-java/src/main/java/com/google/cloud/grpc/fallback/GcpFallbackChannel.java index b991234848c1..671562b5aad4 100644 --- a/grpc-gcp-java/src/main/java/com/google/cloud/grpc/fallback/GcpFallbackChannel.java +++ b/grpc-gcp-java/src/main/java/com/google/cloud/grpc/fallback/GcpFallbackChannel.java @@ -20,7 +20,6 @@ import static java.util.concurrent.TimeUnit.MILLISECONDS; import static java.util.concurrent.TimeUnit.NANOSECONDS; -import com.google.cloud.grpc.GcpThreadFactory; import com.google.common.annotations.VisibleForTesting; import io.grpc.CallOptions; import io.grpc.Channel; @@ -31,9 +30,10 @@ import io.grpc.ManagedChannelBuilder; import io.grpc.MethodDescriptor; import io.grpc.Status; -import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; import java.util.logging.Logger; import javax.annotation.Nullable; @@ -50,14 +50,17 @@ public class GcpFallbackChannel extends ManagedChannel { private final Channel primaryChannel; // Wrapped fallback channel to be used for RPCs. private final Channel fallbackChannel; - private final AtomicLong primarySuccesses = new AtomicLong(0); - private final AtomicLong primaryFailures = new AtomicLong(0); - private final AtomicLong fallbackSuccesses = new AtomicLong(0); - private final AtomicLong fallbackFailures = new AtomicLong(0); - private boolean inFallbackMode = false; + private final GcpFallbackState fallbackState; + private final boolean ownsFallbackState; private final GcpFallbackOpenTelemetry openTelemetry; + private final AtomicBoolean localInFallbackMode = new AtomicBoolean(false); + private final AtomicLong localProbeSuccesses = new AtomicLong(0); + private final AtomicLong localFirstPrimaryProbeSuccessNanos = new AtomicLong(0); + private final ScheduledExecutorService execService; + private volatile ScheduledFuture primaryProbeFuture = null; + private volatile ScheduledFuture fallbackProbeFuture = null; public GcpFallbackChannel( GcpFallbackChannelOptions options, @@ -82,13 +85,15 @@ public GcpFallbackChannel( checkNotNull(options); checkNotNull(primaryChannelBuilder); checkNotNull(fallbackChannelBuilder); - if (execService != null) { - this.execService = execService; + this.options = options; + if (options.getSharedState() != null) { + this.fallbackState = options.getSharedState(); + this.ownsFallbackState = false; } else { - this.execService = - Executors.newScheduledThreadPool(3, GcpThreadFactory.newThreadFactory("gcp-fallback-%d")); + this.fallbackState = new GcpFallbackState(); // Private state for backward compatibility + this.ownsFallbackState = true; } - this.options = options; + this.execService = fallbackState.getOrCreateExecutorService(execService, options); if (options.getGcpOpenTelemetry() != null) { this.openTelemetry = options.getGcpOpenTelemetry(); } else { @@ -149,13 +154,15 @@ public GcpFallbackChannel( checkNotNull(options); checkNotNull(primaryChannel); checkNotNull(fallbackChannel); - if (execService != null) { - this.execService = execService; + this.options = options; + if (options.getSharedState() != null) { + this.fallbackState = options.getSharedState(); + this.ownsFallbackState = false; } else { - this.execService = - Executors.newScheduledThreadPool(3, GcpThreadFactory.newThreadFactory("gcp-fallback-%d")); + this.fallbackState = new GcpFallbackState(); // Private state for backward compatibility + this.ownsFallbackState = true; } - this.options = options; + this.execService = fallbackState.getOrCreateExecutorService(execService, options); if (options.getGcpOpenTelemetry() != null) { this.openTelemetry = options.getGcpOpenTelemetry(); } else { @@ -175,105 +182,135 @@ public GcpFallbackChannel( } public boolean isInFallbackMode() { - return inFallbackMode || primaryChannel == null; + if (fallbackState.getInFallbackMode().get()) { + if (localInFallbackMode.compareAndSet(false, true)) { + localProbeSuccesses.set(0); + localFirstPrimaryProbeSuccessNanos.set(0); + } + } else if (!options.isEnablePerChannelRecovery()) { + if (localInFallbackMode.compareAndSet(true, false)) { + localProbeSuccesses.set(0); + localFirstPrimaryProbeSuccessNanos.set(0); + } + } + if (options.isEnablePerChannelRecovery()) { + return (localInFallbackMode.get() && fallbackChannel != null) || primaryChannel == null; + } + return (fallbackState.getInFallbackMode().get() && fallbackChannel != null) + || primaryChannel == null; + } + + @VisibleForTesting + GcpFallbackState getFallbackState() { + return fallbackState; + } + + @VisibleForTesting + AtomicBoolean getLocalInFallbackMode() { + return localInFallbackMode; + } + + @VisibleForTesting + AtomicLong getLocalProbeSuccesses() { + return localProbeSuccesses; } private void init() { if (options.getPrimaryProbingFunction() != null) { - execService.scheduleAtFixedRate( - this::probePrimary, - options.getPrimaryProbingInterval().toMillis(), - options.getPrimaryProbingInterval().toMillis(), - MILLISECONDS); + this.primaryProbeFuture = + fallbackState.scheduleTask( + this::probePrimary, + options.getPrimaryProbingInterval().toMillis(), + options.getPrimaryProbingInterval().toMillis(), + MILLISECONDS); } if (options.getFallbackProbingFunction() != null) { - execService.scheduleAtFixedRate( - this::probeFallback, - options.getFallbackProbingInterval().toMillis(), - options.getFallbackProbingInterval().toMillis(), - MILLISECONDS); - } - - if (options.isEnableFallback() - && options.getPeriod() != null - && options.getPeriod().toMillis() > 0) { - execService.scheduleAtFixedRate( - this::checkErrorRates, - options.getPeriod().toMillis(), - options.getPeriod().toMillis(), - MILLISECONDS); + this.fallbackProbeFuture = + fallbackState.scheduleTask( + this::probeFallback, + options.getFallbackProbingInterval().toMillis(), + options.getFallbackProbingInterval().toMillis(), + MILLISECONDS); } + + fallbackState.startPeriodicEvaluation(options, execService); } private void checkErrorRates() { - long successes = primarySuccesses.getAndSet(0); - long failures = primaryFailures.getAndSet(0); - float errRate = 0f; - if (failures + successes > 0) { - errRate = (float) failures / (failures + successes); - } - // Report primary error rate. - openTelemetry.getModule().reportErrorRate(options.getPrimaryChannelName(), errRate); - - if (!isInFallbackMode() && options.isEnableFallback() && fallbackChannel != null) { - if (failures >= options.getMinFailedCalls() && errRate >= options.getErrorRateThreshold()) { - if (inFallbackMode != true) { - openTelemetry - .getModule() - .reportFallback(options.getPrimaryChannelName(), options.getFallbackChannelName()); - } - inFallbackMode = true; - } - } - successes = fallbackSuccesses.getAndSet(0); - failures = fallbackFailures.getAndSet(0); - errRate = 0f; - if (failures + successes > 0) { - errRate = (float) failures / (failures + successes); - } - // Report fallback error rate. - openTelemetry.getModule().reportErrorRate(options.getFallbackChannelName(), errRate); - - openTelemetry - .getModule() - .reportCurrentChannel(options.getPrimaryChannelName(), inFallbackMode == false); - openTelemetry - .getModule() - .reportCurrentChannel(options.getFallbackChannelName(), inFallbackMode == true); + fallbackState.checkErrorRates(options, openTelemetry); } private void processPrimaryStatusCode(Status.Code statusCode) { if (options.getErroneousStates().contains(statusCode)) { - // Count error. - primaryFailures.incrementAndGet(); + fallbackState.getPrimaryFailures().incrementAndGet(); } else { - // Count success. - primarySuccesses.incrementAndGet(); + fallbackState.getPrimarySuccesses().incrementAndGet(); } - // Report status code. openTelemetry.getModule().reportStatus(options.getPrimaryChannelName(), statusCode); } private void processFallbackStatusCode(Status.Code statusCode) { if (options.getErroneousStates().contains(statusCode)) { - // Count error. - fallbackFailures.incrementAndGet(); + fallbackState.getFallbackFailures().incrementAndGet(); } else { - // Count success. - fallbackSuccesses.incrementAndGet(); + fallbackState.getFallbackSuccesses().incrementAndGet(); } - // Report status code. openTelemetry.getModule().reportStatus(options.getFallbackChannelName(), statusCode); } private void probePrimary() { + if (fallbackState.getInFallbackMode().get()) { + if (localInFallbackMode.compareAndSet(false, true)) { + localProbeSuccesses.set(0); + localFirstPrimaryProbeSuccessNanos.set(0); + } + } else if (!options.isEnablePerChannelRecovery()) { + if (localInFallbackMode.compareAndSet(true, false)) { + localProbeSuccesses.set(0); + localFirstPrimaryProbeSuccessNanos.set(0); + } + } + boolean inFallback = + options.isEnablePerChannelRecovery() + ? localInFallbackMode.get() + : fallbackState.getInFallbackMode().get(); + if (!inFallback && primaryChannel != null) { + return; + } String result = ""; if (primaryDelegateChannel == null) { result = INIT_FAILURE_REASON; } else { result = options.getPrimaryProbingFunction().apply(primaryDelegateChannel); } + if ("OK".equals(result)) { + long nowNanos = System.nanoTime(); + long firstSuccessNanos = + localFirstPrimaryProbeSuccessNanos.updateAndGet(prev -> prev == 0 ? nowNanos : prev); + long primaryProbeSuccessCount = localProbeSuccesses.incrementAndGet(); + + boolean durationSatisfied = true; + if (options.getMinPrimaryProbeSuccessDuration() != null + && !options.getMinPrimaryProbeSuccessDuration().isZero() + && !options.getMinPrimaryProbeSuccessDuration().isNegative()) { + long elapsedNanos = nowNanos - firstSuccessNanos; + durationSatisfied = elapsedNanos >= options.getMinPrimaryProbeSuccessDuration().toNanos(); + } + + if (options.isEnableRecovery() + && primaryProbeSuccessCount >= options.getMinPrimaryProbeSuccessCount() + && durationSatisfied) { + fallbackState.getInFallbackMode().set(false); + localInFallbackMode.set(false); + localProbeSuccesses.set(0); + localFirstPrimaryProbeSuccessNanos.set(0); + } + } else { + localInFallbackMode.set(true); + localProbeSuccesses.set(0); + localFirstPrimaryProbeSuccessNanos.set(0); + } // Report metric based on result. openTelemetry.getModule().reportProbeResult(options.getPrimaryChannelName(), result); } @@ -308,27 +345,71 @@ public String authority() { return primaryChannel.authority(); } + @Override + public io.grpc.ConnectivityState getState(boolean requestConnection) { + if (isInFallbackMode()) { + if (fallbackDelegateChannel != null) { + return fallbackDelegateChannel.getState(requestConnection); + } + return io.grpc.ConnectivityState.SHUTDOWN; + } + + if (primaryDelegateChannel != null) { + return primaryDelegateChannel.getState(requestConnection); + } + return io.grpc.ConnectivityState.SHUTDOWN; + } + + @Override + public void notifyWhenStateChanged(io.grpc.ConnectivityState source, Runnable callback) { + if (isInFallbackMode()) { + if (fallbackDelegateChannel != null) { + fallbackDelegateChannel.notifyWhenStateChanged(source, callback); + } + } else { + if (primaryDelegateChannel != null) { + primaryDelegateChannel.notifyWhenStateChanged(source, callback); + } + } + } + @Override public ManagedChannel shutdown() { + if (primaryProbeFuture != null) { + primaryProbeFuture.cancel(false); + } + if (fallbackProbeFuture != null) { + fallbackProbeFuture.cancel(false); + } if (primaryDelegateChannel != null) { primaryDelegateChannel.shutdown(); } if (fallbackDelegateChannel != null) { fallbackDelegateChannel.shutdown(); } - execService.shutdown(); + if (ownsFallbackState) { + fallbackState.shutdown(); + } return this; } @Override public ManagedChannel shutdownNow() { + if (primaryProbeFuture != null) { + primaryProbeFuture.cancel(true); + } + if (fallbackProbeFuture != null) { + fallbackProbeFuture.cancel(true); + } if (primaryDelegateChannel != null) { primaryDelegateChannel.shutdownNow(); } if (fallbackDelegateChannel != null) { fallbackDelegateChannel.shutdownNow(); } - execService.shutdownNow(); + if (ownsFallbackState) { + fallbackState.shutdownNow(); + } return this; } diff --git a/grpc-gcp-java/src/main/java/com/google/cloud/grpc/fallback/GcpFallbackChannelOptions.java b/grpc-gcp-java/src/main/java/com/google/cloud/grpc/fallback/GcpFallbackChannelOptions.java index 31d5cd981907..0fcc21fd4e34 100644 --- a/grpc-gcp-java/src/main/java/com/google/cloud/grpc/fallback/GcpFallbackChannelOptions.java +++ b/grpc-gcp-java/src/main/java/com/google/cloud/grpc/fallback/GcpFallbackChannelOptions.java @@ -25,6 +25,7 @@ import java.time.Duration; import java.util.EnumSet; import java.util.Set; +import java.util.concurrent.ScheduledExecutorService; import java.util.function.Function; public class GcpFallbackChannelOptions { @@ -37,9 +38,15 @@ public class GcpFallbackChannelOptions { private final Function fallbackProbingFunction; private final Duration primaryProbingInterval; private final Duration fallbackProbingInterval; + private final int minPrimaryProbeSuccessCount; + private final Duration minPrimaryProbeSuccessDuration; + private final boolean enableRecovery; + private final boolean enablePerChannelRecovery; private final String primaryChannelName; private final String fallbackChannelName; private final GcpFallbackOpenTelemetry openTelemetry; + private final ScheduledExecutorService sharedExecutorService; + private final GcpFallbackState sharedState; public GcpFallbackChannelOptions(Builder builder) { this.enableFallback = builder.enableFallback; @@ -51,9 +58,15 @@ public GcpFallbackChannelOptions(Builder builder) { this.fallbackProbingFunction = builder.fallbackProbingFunction; this.primaryProbingInterval = builder.primaryProbingInterval; this.fallbackProbingInterval = builder.fallbackProbingInterval; + this.minPrimaryProbeSuccessCount = builder.minPrimaryProbeSuccessCount; + this.minPrimaryProbeSuccessDuration = builder.minPrimaryProbeSuccessDuration; + this.enableRecovery = builder.enableRecovery; + this.enablePerChannelRecovery = builder.enablePerChannelRecovery; this.primaryChannelName = builder.primaryChannelName; this.fallbackChannelName = builder.fallbackChannelName; this.openTelemetry = builder.openTelemetry; + this.sharedExecutorService = builder.sharedExecutorService; + this.sharedState = builder.sharedState; } public static Builder newBuilder() { @@ -96,6 +109,22 @@ public Duration getFallbackProbingInterval() { return fallbackProbingInterval; } + public int getMinPrimaryProbeSuccessCount() { + return minPrimaryProbeSuccessCount; + } + + public Duration getMinPrimaryProbeSuccessDuration() { + return minPrimaryProbeSuccessDuration; + } + + public boolean isEnableRecovery() { + return enableRecovery; + } + + public boolean isEnablePerChannelRecovery() { + return enablePerChannelRecovery; + } + public String getPrimaryChannelName() { return primaryChannelName; } @@ -108,6 +137,14 @@ public GcpFallbackOpenTelemetry getGcpOpenTelemetry() { return openTelemetry; } + public ScheduledExecutorService getSharedExecutorService() { + return sharedExecutorService; + } + + public GcpFallbackState getSharedState() { + return sharedState; + } + public static class Builder { private boolean enableFallback = true; private float errorRateThreshold = 1f; @@ -122,10 +159,17 @@ public static class Builder { private Duration primaryProbingInterval = Duration.ofMinutes(1); private Duration fallbackProbingInterval = Duration.ofMinutes(15); + private int minPrimaryProbeSuccessCount = 10; + private Duration minPrimaryProbeSuccessDuration = Duration.ZERO; + private boolean enableRecovery = false; + private boolean enablePerChannelRecovery = false; + private String primaryChannelName = "primary"; private String fallbackChannelName = "fallback"; private GcpFallbackOpenTelemetry openTelemetry = null; + private ScheduledExecutorService sharedExecutorService = null; + private GcpFallbackState sharedState = null; public Builder() {} @@ -184,6 +228,26 @@ public Builder setFallbackProbingInterval(Duration fallbackProbingInterval) { return this; } + public Builder setMinPrimaryProbeSuccessCount(int minPrimaryProbeSuccessCount) { + this.minPrimaryProbeSuccessCount = minPrimaryProbeSuccessCount; + return this; + } + + public Builder setMinPrimaryProbeSuccessDuration(Duration minPrimaryProbeSuccessDuration) { + this.minPrimaryProbeSuccessDuration = minPrimaryProbeSuccessDuration; + return this; + } + + public Builder setEnableRecovery(boolean enableRecovery) { + this.enableRecovery = enableRecovery; + return this; + } + + public Builder setEnablePerChannelRecovery(boolean enablePerChannelRecovery) { + this.enablePerChannelRecovery = enablePerChannelRecovery; + return this; + } + public Builder setPrimaryChannelName(String primaryChannelName) { this.primaryChannelName = primaryChannelName; return this; @@ -199,6 +263,16 @@ public Builder setGcpFallbackOpenTelemetry(GcpFallbackOpenTelemetry openTelemetr return this; } + public Builder setSharedExecutorService(ScheduledExecutorService sharedExecutorService) { + this.sharedExecutorService = sharedExecutorService; + return this; + } + + public Builder setSharedState(GcpFallbackState sharedState) { + this.sharedState = sharedState; + return this; + } + public GcpFallbackChannelOptions build() { return new GcpFallbackChannelOptions(this); } diff --git a/grpc-gcp-java/src/main/java/com/google/cloud/grpc/fallback/GcpFallbackState.java b/grpc-gcp-java/src/main/java/com/google/cloud/grpc/fallback/GcpFallbackState.java new file mode 100644 index 000000000000..266d0176947d --- /dev/null +++ b/grpc-gcp-java/src/main/java/com/google/cloud/grpc/fallback/GcpFallbackState.java @@ -0,0 +1,209 @@ +/* + * 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.fallback; + +import com.google.cloud.grpc.GcpThreadFactory; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; + +/** + * Shared thread-safe state container for coordinated pool-wide failover, recovery, and background + * tasks. + * + *

All channels in a pool share this state instance and its background executor service, + * consolidating probing and error evaluation threads across the entire channel pool. + */ +public class GcpFallbackState { + private final AtomicLong primarySuccesses = new AtomicLong(0); + private final AtomicLong primaryFailures = new AtomicLong(0); + private final AtomicLong fallbackSuccesses = new AtomicLong(0); + private final AtomicLong fallbackFailures = new AtomicLong(0); + private final AtomicBoolean inFallbackMode = new AtomicBoolean(false); + private final AtomicBoolean evaluationStarted = new AtomicBoolean(false); + + private ScheduledExecutorService execService = null; + private boolean ownsExecutor = false; + private volatile ScheduledFuture scheduledEvaluationFuture = null; + + public AtomicLong getPrimarySuccesses() { + return primarySuccesses; + } + + public AtomicLong getPrimaryFailures() { + return primaryFailures; + } + + public AtomicLong getFallbackSuccesses() { + return fallbackSuccesses; + } + + public AtomicLong getFallbackFailures() { + return fallbackFailures; + } + + public AtomicBoolean getInFallbackMode() { + return inFallbackMode; + } + + /** + * Retrieves or lazily initializes the shared background executor service. + * + * @param externalExec optional external executor service to use (e.g. from test or options). + * @param options optional fallback channel configuration options. + * @return the active ScheduledExecutorService. + */ + public synchronized ScheduledExecutorService getOrCreateExecutorService( + ScheduledExecutorService externalExec, GcpFallbackChannelOptions options) { + if (this.execService != null) { + return this.execService; + } + if (externalExec != null) { + this.execService = externalExec; + this.ownsExecutor = + (options == null + || options.getSharedState() == null + || options.getSharedExecutorService() == null); + } else if (options != null && options.getSharedExecutorService() != null) { + this.execService = options.getSharedExecutorService(); + this.ownsExecutor = false; + } else { + this.execService = + Executors.newScheduledThreadPool( + 3, GcpThreadFactory.newThreadFactory("gcp-fallback-state-%d")); + this.ownsExecutor = true; + } + return this.execService; + } + + /** Schedules a periodic task (e.g., probe) on the shared background executor service. */ + public synchronized ScheduledFuture scheduleTask( + Runnable command, long initialDelay, long period, TimeUnit unit) { + if (this.execService == null || this.execService.isShutdown()) { + return null; + } + return this.execService.scheduleAtFixedRate(command, initialDelay, period, unit); + } + + /** + * Starts the periodic error rate evaluation loop exactly once across all channels sharing this + * state. + * + * @param options the fallback channel configuration options. + * @param externalExec optional executor service to use if not yet initialized. + */ + public void startPeriodicEvaluation( + GcpFallbackChannelOptions options, ScheduledExecutorService externalExec) { + if (options == null + || !options.isEnableFallback() + || options.getPeriod() == null + || options.getPeriod().toMillis() <= 0) { + return; + } + if (evaluationStarted.compareAndSet(false, true)) { + ScheduledExecutorService executor = getOrCreateExecutorService(externalExec, options); + GcpFallbackOpenTelemetry openTelemetry = + options.getGcpOpenTelemetry() != null + ? options.getGcpOpenTelemetry() + : GcpFallbackOpenTelemetry.newBuilder().build(); + + scheduledEvaluationFuture = + executor.scheduleAtFixedRate( + () -> checkErrorRates(options, openTelemetry), + options.getPeriod().toMillis(), + options.getPeriod().toMillis(), + TimeUnit.MILLISECONDS); + } + } + + /** + * Evaluates error rates across all channels sharing this state and updates fallback mode. + * + * @param options the fallback channel configuration options. + * @param openTelemetry telemetry module for recording error metrics. + */ + public void checkErrorRates( + GcpFallbackChannelOptions options, GcpFallbackOpenTelemetry openTelemetry) { + long successes = primarySuccesses.getAndSet(0); + long failures = primaryFailures.getAndSet(0); + float errRate = 0f; + if (failures + successes > 0) { + errRate = (float) failures / (failures + successes); + } + if (openTelemetry != null && openTelemetry.getModule() != null) { + openTelemetry.getModule().reportErrorRate(options.getPrimaryChannelName(), errRate); + } + + if (!inFallbackMode.get() && options.isEnableFallback()) { + if (failures >= options.getMinFailedCalls() && errRate >= options.getErrorRateThreshold()) { + inFallbackMode.set(true); + if (openTelemetry != null && openTelemetry.getModule() != null) { + openTelemetry + .getModule() + .reportFallback(options.getPrimaryChannelName(), options.getFallbackChannelName()); + } + } + } + + successes = fallbackSuccesses.getAndSet(0); + failures = fallbackFailures.getAndSet(0); + errRate = 0f; + if (failures + successes > 0) { + errRate = (float) failures / (failures + successes); + } + if (openTelemetry != null && openTelemetry.getModule() != null) { + openTelemetry.getModule().reportErrorRate(options.getFallbackChannelName(), errRate); + openTelemetry + .getModule() + .reportCurrentChannel(options.getPrimaryChannelName(), !inFallbackMode.get()); + openTelemetry + .getModule() + .reportCurrentChannel(options.getFallbackChannelName(), inFallbackMode.get()); + } + } + + /** Stops any running scheduled evaluation. */ + public synchronized void stopPeriodicEvaluation() { + if (scheduledEvaluationFuture != null) { + scheduledEvaluationFuture.cancel(false); + scheduledEvaluationFuture = null; + } + evaluationStarted.set(false); + } + + /** Shuts down the state, cancelling evaluation and shutting down internal executor if owned. */ + public synchronized void shutdown() { + stopPeriodicEvaluation(); + if (ownsExecutor && execService != null && !execService.isShutdown()) { + execService.shutdown(); + } + } + + /** + * Shuts down the state immediately, cancelling evaluation and terminating internal executor if + * owned. + */ + public synchronized void shutdownNow() { + stopPeriodicEvaluation(); + if (ownsExecutor && execService != null && !execService.isShutdown()) { + execService.shutdownNow(); + } + } +} diff --git a/grpc-gcp-java/src/test/java/com/google/cloud/grpc/fallback/GcpFallbackChannelTest.java b/grpc-gcp-java/src/test/java/com/google/cloud/grpc/fallback/GcpFallbackChannelTest.java index 210c482d588f..898220a26887 100644 --- a/grpc-gcp-java/src/test/java/com/google/cloud/grpc/fallback/GcpFallbackChannelTest.java +++ b/grpc-gcp-java/src/test/java/com/google/cloud/grpc/fallback/GcpFallbackChannelTest.java @@ -41,6 +41,8 @@ import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.ArgumentMatchers.isNull; +import static org.mockito.Mockito.atLeast; +import static org.mockito.Mockito.atLeastOnce; import static org.mockito.Mockito.clearInvocations; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; @@ -82,6 +84,7 @@ import java.util.List; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; import java.util.function.Function; import javax.annotation.Nonnull; @@ -1082,6 +1085,7 @@ public void testProbingTasksScheduled_ifConfigured() { .build(); initializeChannelAndCaptureTasks(options); + gcpFallbackChannel.getFallbackState().getInFallbackMode().set(true); assertNotNull(primaryProbingTask); assertNotNull(fallbackProbingTask); @@ -1118,6 +1122,7 @@ public void testProbing_reportsMetrics() throws InterruptedException { .build(); initializeChannelAndCaptureTasks(options); + gcpFallbackChannel.getFallbackState().getInFallbackMode().set(true); assertNotNull(primaryProbingTask); assertNotNull(fallbackProbingTask); @@ -1212,6 +1217,7 @@ public void testProbing_reportsInitFailureForFallback() throws InterruptedExcept .build(); initializeChannelWithInvalidFallbackBuilderAndCaptureTasks(options); + gcpFallbackChannel.getFallbackState().getInFallbackMode().set(true); assertNotNull(primaryProbingTask); assertNotNull(fallbackProbingTask); @@ -1286,4 +1292,568 @@ public void testConstructor_failsWhenBothBuildersFail() { new GcpFallbackChannel( getDefaultOptions(), mockPrimaryInvalidBuilder, mockFallbackInvalidBuilder)); } + + @SuppressWarnings({"unchecked"}) + private void simulateCallOnChannel( + GcpFallbackChannel channel, + Status statusToReturn, + ManagedChannel primaryDelegate, + ManagedChannel fallbackDelegate, + ClientCall primaryCall, + ClientCall fallbackCall, + boolean expectFallbackRouting) { + final ClientCall.Listener dummyCallListener = mock(ClientCall.Listener.class); + final Metadata requestHeaders = new Metadata(); + + ClientCall testCall = channel.newCall(methodDescriptor, callOptions); + assertNotNull(testCall); + + ClientCall targetCall; + if (expectFallbackRouting) { + verify(fallbackDelegate).newCall(methodDescriptor, callOptions); + verify(primaryDelegate, never()).newCall(methodDescriptor, callOptions); + targetCall = fallbackCall; + } else { + verify(primaryDelegate).newCall(methodDescriptor, callOptions); + verify(fallbackDelegate, never()).newCall(methodDescriptor, callOptions); + targetCall = primaryCall; + } + + testCall.start(dummyCallListener, requestHeaders); + + ArgumentCaptor> delegateListenerCaptor = + ArgumentCaptor.forClass(ClientCall.Listener.class); + verify(targetCall).start(delegateListenerCaptor.capture(), eq(requestHeaders)); + delegateListenerCaptor.getValue().onClose(statusToReturn, new Metadata()); + + clearInvocations(primaryDelegate, fallbackDelegate, targetCall); + } + + @Test + public void testSharedState_singleEvaluationScheduled() { + GcpFallbackState sharedState = new GcpFallbackState(); + GcpFallbackChannelOptions options = + getDefaultOptionsBuilder().setSharedState(sharedState).build(); + + ScheduledExecutorService mockExec1 = mock(ScheduledExecutorService.class); + ScheduledExecutorService mockExec2 = mock(ScheduledExecutorService.class); + + GcpFallbackChannel channel1 = + new GcpFallbackChannel(options, mockPrimaryBuilder, mockFallbackBuilder, mockExec1); + GcpFallbackChannel channel2 = + new GcpFallbackChannel(options, mockPrimaryBuilder, mockFallbackBuilder, mockExec2); + + try { + // Periodic evaluation was started on mockExec1 by the shared state + verify(mockExec1) + .scheduleAtFixedRate( + any(Runnable.class), + eq(options.getPeriod().toMillis()), + eq(options.getPeriod().toMillis()), + eq(MILLISECONDS)); + + // Channel 2 sharing the same state did NOT schedule a duplicate evaluation loop + verify(mockExec2, never()) + .scheduleAtFixedRate( + any(Runnable.class), + eq(options.getPeriod().toMillis()), + eq(options.getPeriod().toMillis()), + eq(MILLISECONDS)); + } finally { + channel1.shutdownNow(); + channel2.shutdownNow(); + } + } + + @Test + public void testSharedState_coordinatedFailover() { + GcpFallbackState sharedState = new GcpFallbackState(); + GcpFallbackChannelOptions options = + getDefaultOptionsBuilder() + .setSharedState(sharedState) + .setMinFailedCalls(3) + .setErrorRateThreshold(0.5f) + .build(); + + ScheduledExecutorService mockExec1 = mock(ScheduledExecutorService.class); + ScheduledExecutorService mockExec2 = mock(ScheduledExecutorService.class); + + ArgumentCaptor taskCaptor = ArgumentCaptor.forClass(Runnable.class); + + GcpFallbackChannel channel1 = + new GcpFallbackChannel(options, mockPrimaryBuilder, mockFallbackBuilder, mockExec1); + GcpFallbackChannel channel2 = + new GcpFallbackChannel(options, mockPrimaryBuilder, mockFallbackBuilder, mockExec2); + + try { + verify(mockExec1) + .scheduleAtFixedRate( + taskCaptor.capture(), + eq(options.getPeriod().toMillis()), + eq(options.getPeriod().toMillis()), + eq(MILLISECONDS)); + Runnable checkErrorRates = taskCaptor.getValue(); + + // Both channels initially in primary mode + assertFalse(channel1.isInFallbackMode()); + assertFalse(channel2.isInFallbackMode()); + + // Channel 1 processes 2 failures, Channel 2 processes 1 failure (total 3 failures on shared + // state) + simulateCallOnChannel( + channel1, + Status.UNAVAILABLE, + mockPrimaryDelegateChannel, + mockFallbackDelegateChannel, + mockPrimaryClientCall, + mockFallbackClientCall, + false); + simulateCallOnChannel( + channel1, + Status.UNAVAILABLE, + mockPrimaryDelegateChannel, + mockFallbackDelegateChannel, + mockPrimaryClientCall, + mockFallbackClientCall, + false); + simulateCallOnChannel( + channel2, + Status.UNAVAILABLE, + mockPrimaryDelegateChannel, + mockFallbackDelegateChannel, + mockPrimaryClientCall, + mockFallbackClientCall, + false); + + // Run checkErrorRates on shared state + checkErrorRates.run(); + + // Both channels must now be in fallback mode + assertTrue(channel1.isInFallbackMode()); + assertTrue(channel2.isInFallbackMode()); + + // Subsequent call on Channel 2 routes to fallback channel + simulateCallOnChannel( + channel2, + Status.OK, + mockPrimaryDelegateChannel, + mockFallbackDelegateChannel, + mockPrimaryClientCall, + mockFallbackClientCall, + true); + } finally { + channel1.shutdownNow(); + channel2.shutdownNow(); + } + } + + @Test + public void testSharedState_channelShutdownLeavesSiblingChannelsFunctional() { + GcpFallbackState sharedState = new GcpFallbackState(); + GcpFallbackChannelOptions options = + getDefaultOptionsBuilder().setSharedState(sharedState).build(); + + ScheduledExecutorService mockExec1 = mock(ScheduledExecutorService.class); + ScheduledExecutorService mockExec2 = mock(ScheduledExecutorService.class); + + GcpFallbackChannel channel1 = + new GcpFallbackChannel(options, mockPrimaryBuilder, mockFallbackBuilder, mockExec1); + GcpFallbackChannel channel2 = + new GcpFallbackChannel(options, mockPrimaryBuilder, mockFallbackBuilder, mockExec2); + + try { + // Shutting down channel1 does not shut down the shared executor while sibling channels are + // active + channel1.shutdown(); + verify(mockExec1, never()).shutdown(); + + // Channel 2 can still transition and read shared fallback state + sharedState.getInFallbackMode().set(true); + assertTrue(channel2.isInFallbackMode()); + } finally { + channel2.shutdownNow(); + sharedState.shutdown(); + verify(mockExec1).shutdown(); + } + } + + @Test + public void testSharedState_probingRequiresBothCountAndDurationToRecover() + throws InterruptedException { + GcpFallbackState sharedState = new GcpFallbackState(); + sharedState.getInFallbackMode().set(true); + + GcpFallbackChannelOptions options = + getDefaultOptionsBuilder() + .setSharedState(sharedState) + .setEnableRecovery(true) + .setPrimaryProbingFunction(channel -> "OK") + .setMinPrimaryProbeSuccessCount(2) + .setMinPrimaryProbeSuccessDuration(Duration.ofMillis(50)) + .build(); + + ScheduledExecutorService mockExec = mock(ScheduledExecutorService.class); + ArgumentCaptor taskCaptor = ArgumentCaptor.forClass(Runnable.class); + + GcpFallbackChannel channel = + new GcpFallbackChannel(options, mockPrimaryBuilder, mockFallbackBuilder, mockExec); + + try { + verify(mockExec) + .scheduleAtFixedRate( + taskCaptor.capture(), + eq(options.getPrimaryProbingInterval().toMillis()), + eq(options.getPrimaryProbingInterval().toMillis()), + eq(MILLISECONDS)); + Runnable probeTask = taskCaptor.getValue(); + + // Probe 1: Success, but count < 2 and duration not yet met + probeTask.run(); + assertTrue(channel.isInFallbackMode()); + assertEquals(1, channel.getLocalProbeSuccesses().get()); + + // Probe 2 immediately: count == 2, but duration (50ms) not elapsed yet! + probeTask.run(); + assertTrue(channel.isInFallbackMode()); + assertEquals(2, channel.getLocalProbeSuccesses().get()); + + // Wait for duration window to pass + Thread.sleep(60); + + // Probe 3: count >= 2 and duration >= 50ms satisfied -> Recover! + probeTask.run(); + assertFalse(channel.isInFallbackMode()); + assertEquals(0, channel.getLocalProbeSuccesses().get()); + } finally { + channel.shutdownNow(); + sharedState.shutdown(); + } + } + + @Test + public void testSharedState_probingFailureResetsDurationTimer() { + GcpFallbackState sharedState = new GcpFallbackState(); + sharedState.getInFallbackMode().set(true); + + AtomicBoolean probeOk = new AtomicBoolean(true); + GcpFallbackChannelOptions options = + getDefaultOptionsBuilder() + .setSharedState(sharedState) + .setEnableRecovery(true) + .setPrimaryProbingFunction(channel -> probeOk.get() ? "OK" : "UNAVAILABLE") + .setMinPrimaryProbeSuccessCount(5) + .setMinPrimaryProbeSuccessDuration(Duration.ofMinutes(10)) + .build(); + + ScheduledExecutorService mockExec = mock(ScheduledExecutorService.class); + ArgumentCaptor taskCaptor = ArgumentCaptor.forClass(Runnable.class); + + GcpFallbackChannel channel = + new GcpFallbackChannel(options, mockPrimaryBuilder, mockFallbackBuilder, mockExec); + + try { + verify(mockExec) + .scheduleAtFixedRate( + taskCaptor.capture(), + eq(options.getPrimaryProbingInterval().toMillis()), + eq(options.getPrimaryProbingInterval().toMillis()), + eq(MILLISECONDS)); + Runnable probeTask = taskCaptor.getValue(); + + // Successful probe initializes local probe counter + probeTask.run(); + assertEquals(1, channel.getLocalProbeSuccesses().get()); + + // Failing probe resets local probe count to 0 + probeOk.set(false); + probeTask.run(); + assertEquals(0, channel.getLocalProbeSuccesses().get()); + assertTrue(channel.isInFallbackMode()); + } finally { + channel.shutdownNow(); + sharedState.shutdown(); + } + } + + @Test + public void testProbePrimary_skippedWhenNotInFallbackMode() { + AtomicLong probeCalls = new AtomicLong(0); + GcpFallbackState sharedState = new GcpFallbackState(); + sharedState.getInFallbackMode().set(false); + + GcpFallbackChannelOptions options = + getDefaultOptionsBuilder() + .setSharedState(sharedState) + .setPrimaryProbingFunction( + channel -> { + probeCalls.incrementAndGet(); + return "OK"; + }) + .build(); + + ScheduledExecutorService mockExec = mock(ScheduledExecutorService.class); + ArgumentCaptor taskCaptor = ArgumentCaptor.forClass(Runnable.class); + + GcpFallbackChannel channel = + new GcpFallbackChannel(options, mockPrimaryBuilder, mockFallbackBuilder, mockExec); + + try { + verify(mockExec) + .scheduleAtFixedRate( + taskCaptor.capture(), + eq(options.getPrimaryProbingInterval().toMillis()), + eq(options.getPrimaryProbingInterval().toMillis()), + eq(MILLISECONDS)); + Runnable probeTask = taskCaptor.getValue(); + + // Channel is NOT in fallback mode -> probe task should immediately return without probing! + assertFalse(channel.isInFallbackMode()); + probeTask.run(); + assertEquals(0, probeCalls.get()); + assertEquals(0, channel.getLocalProbeSuccesses().get()); + } finally { + channel.shutdownNow(); + sharedState.shutdown(); + } + } + + @Test + public void testPerChannelIndependentRecovery_oneChannelRecoversWhileOtherStaysInFallback() { + GcpFallbackState sharedState = new GcpFallbackState(); + sharedState.getInFallbackMode().set(true); // Pool-wide fallback active + + GcpFallbackChannelOptions options1 = + getDefaultOptionsBuilder() + .setSharedState(sharedState) + .setEnableRecovery(true) + .setEnablePerChannelRecovery(true) + .setPrimaryProbingFunction(channel -> "OK") + .setMinPrimaryProbeSuccessCount(1) + .setMinPrimaryProbeSuccessDuration(Duration.ZERO) + .build(); + + GcpFallbackChannelOptions options2 = + getDefaultOptionsBuilder() + .setSharedState(sharedState) + .setEnableRecovery(true) + .setEnablePerChannelRecovery(true) + .setPrimaryProbingFunction(channel -> "UNAVAILABLE") + .setMinPrimaryProbeSuccessCount(1) + .setMinPrimaryProbeSuccessDuration(Duration.ZERO) + .build(); + + ScheduledExecutorService mockExec1 = mock(ScheduledExecutorService.class); + ScheduledExecutorService mockExec2 = mock(ScheduledExecutorService.class); + ArgumentCaptor taskCaptor1 = ArgumentCaptor.forClass(Runnable.class); + + GcpFallbackChannel channel1 = + new GcpFallbackChannel(options1, mockPrimaryBuilder, mockFallbackBuilder, mockExec1); + GcpFallbackChannel channel2 = + new GcpFallbackChannel(options2, mockPrimaryBuilder, mockFallbackBuilder, mockExec2); + + try { + verify(mockExec1, atLeastOnce()) + .scheduleAtFixedRate( + taskCaptor1.capture(), + eq(options1.getPrimaryProbingInterval().toMillis()), + eq(options1.getPrimaryProbingInterval().toMillis()), + eq(MILLISECONDS)); + Runnable probeTask1 = taskCaptor1.getAllValues().get(0); + + assertTrue(channel1.isInFallbackMode()); + assertTrue(channel2.isInFallbackMode()); + + // Run probe on channel 1 -> channel 1 recovers to DirectPath + probeTask1.run(); + + assertFalse("Channel 1 should recover to DirectPath", channel1.isInFallbackMode()); + assertTrue("Channel 2 should remain in CloudPath fallback mode", channel2.isInFallbackMode()); + assertFalse("Global fallback should be unlatched", sharedState.getInFallbackMode().get()); + } finally { + channel1.shutdownNow(); + channel2.shutdownNow(); + sharedState.shutdown(); + } + } + + @Test + public void testPoolLevelRecovery_whenPerChannelRecoveryDisabled_allChannelsRecoverTogether() { + GcpFallbackState sharedState = new GcpFallbackState(); + sharedState.getInFallbackMode().set(true); // Pool-wide fallback active + + GcpFallbackChannelOptions options1 = + getDefaultOptionsBuilder() + .setSharedState(sharedState) + .setEnableRecovery(true) + .setEnablePerChannelRecovery(false) // Pool-level recovery + .setPrimaryProbingFunction(channel -> "OK") + .setMinPrimaryProbeSuccessCount(1) + .setMinPrimaryProbeSuccessDuration(Duration.ZERO) + .build(); + + GcpFallbackChannelOptions options2 = + getDefaultOptionsBuilder() + .setSharedState(sharedState) + .setEnableRecovery(true) + .setEnablePerChannelRecovery(false) // Pool-level recovery + .setPrimaryProbingFunction(channel -> "UNAVAILABLE") + .setMinPrimaryProbeSuccessCount(1) + .setMinPrimaryProbeSuccessDuration(Duration.ZERO) + .build(); + + ScheduledExecutorService mockExec1 = mock(ScheduledExecutorService.class); + ScheduledExecutorService mockExec2 = mock(ScheduledExecutorService.class); + ArgumentCaptor taskCaptor1 = ArgumentCaptor.forClass(Runnable.class); + + GcpFallbackChannel channel1 = + new GcpFallbackChannel(options1, mockPrimaryBuilder, mockFallbackBuilder, mockExec1); + GcpFallbackChannel channel2 = + new GcpFallbackChannel(options2, mockPrimaryBuilder, mockFallbackBuilder, mockExec2); + + try { + verify(mockExec1, atLeastOnce()) + .scheduleAtFixedRate( + taskCaptor1.capture(), + eq(options1.getPrimaryProbingInterval().toMillis()), + eq(options1.getPrimaryProbingInterval().toMillis()), + eq(MILLISECONDS)); + Runnable probeTask1 = taskCaptor1.getAllValues().get(0); + + assertTrue(channel1.isInFallbackMode()); + assertTrue(channel2.isInFallbackMode()); + + // Run probe on channel 1 -> channel 1 recovers to DirectPath and unlatches global fallback + probeTask1.run(); + + // With enablePerChannelRecovery=false, both channel 1 and channel 2 recover to DirectPath + // together + assertFalse("Channel 1 should recover to DirectPath", channel1.isInFallbackMode()); + assertFalse( + "Channel 2 should also recover to DirectPath with pool", channel2.isInFallbackMode()); + assertFalse("Global fallback should be unlatched", sharedState.getInFallbackMode().get()); + } finally { + channel1.shutdownNow(); + channel2.shutdownNow(); + sharedState.shutdown(); + } + } + + @Test + public void testRecoveryDisabled_probingSucceedsButChannelRemainsInFallback() { + GcpFallbackState sharedState = new GcpFallbackState(); + sharedState.getInFallbackMode().set(true); // Pool-wide fallback active + + GcpFallbackChannelOptions options = + getDefaultOptionsBuilder() + .setSharedState(sharedState) + .setEnableRecovery(false) // Recovery disabled by default + .setPrimaryProbingFunction(channel -> "OK") + .setMinPrimaryProbeSuccessCount(1) + .setMinPrimaryProbeSuccessDuration(Duration.ZERO) + .build(); + + ScheduledExecutorService mockExec = mock(ScheduledExecutorService.class); + ArgumentCaptor taskCaptor = ArgumentCaptor.forClass(Runnable.class); + + GcpFallbackChannel channel = + new GcpFallbackChannel(options, mockPrimaryBuilder, mockFallbackBuilder, mockExec); + + try { + verify(mockExec) + .scheduleAtFixedRate( + taskCaptor.capture(), + eq(options.getPrimaryProbingInterval().toMillis()), + eq(options.getPrimaryProbingInterval().toMillis()), + eq(MILLISECONDS)); + Runnable probeTask = taskCaptor.getValue(); + + assertTrue(channel.isInFallbackMode()); + + // Run probe -> probe succeeds, but enableRecovery is false + probeTask.run(); + + // Channel must remain in fallback mode + assertTrue(channel.isInFallbackMode()); + assertTrue(sharedState.getInFallbackMode().get()); + } finally { + channel.shutdownNow(); + sharedState.shutdown(); + } + } + + @Test + public void testPoolLevelRecovery_multipleFailoverCyclesResetProbeStatistics() { + GcpFallbackState sharedState = new GcpFallbackState(); + sharedState.getInFallbackMode().set(true); // Cycle 1: Fallback active + + GcpFallbackChannelOptions options1 = + getDefaultOptionsBuilder() + .setSharedState(sharedState) + .setEnableRecovery(true) + .setEnablePerChannelRecovery(false) + .setPrimaryProbingFunction(channel -> "OK") + .setMinPrimaryProbeSuccessCount(2) + .setMinPrimaryProbeSuccessDuration(Duration.ZERO) + .build(); + + GcpFallbackChannelOptions options2 = + getDefaultOptionsBuilder() + .setSharedState(sharedState) + .setEnableRecovery(true) + .setEnablePerChannelRecovery(false) + .setPrimaryProbingFunction(channel -> "OK") + .setMinPrimaryProbeSuccessCount(2) + .setMinPrimaryProbeSuccessDuration(Duration.ZERO) + .build(); + + ScheduledExecutorService mockExec = mock(ScheduledExecutorService.class); + ArgumentCaptor taskCaptor = ArgumentCaptor.forClass(Runnable.class); + + GcpFallbackChannel channel1 = + new GcpFallbackChannel(options1, mockPrimaryBuilder, mockFallbackBuilder, mockExec); + GcpFallbackChannel channel2 = + new GcpFallbackChannel(options2, mockPrimaryBuilder, mockFallbackBuilder, mockExec); + + try { + verify(mockExec, atLeast(2)) + .scheduleAtFixedRate( + taskCaptor.capture(), + eq(options1.getPrimaryProbingInterval().toMillis()), + eq(options1.getPrimaryProbingInterval().toMillis()), + eq(MILLISECONDS)); + Runnable probeTask1 = taskCaptor.getAllValues().get(0); + Runnable probeTask2 = taskCaptor.getAllValues().get(1); + + // Both channels enter fallback + assertTrue(channel1.isInFallbackMode()); + assertTrue(channel2.isInFallbackMode()); + + // Channel 1 probes twice -> recovers pool + probeTask1.run(); + probeTask1.run(); + assertFalse(channel1.isInFallbackMode()); + assertFalse(channel2.isInFallbackMode()); + + // Now Cycle 2: Incident occurs again, pool enters fallback + sharedState.getInFallbackMode().set(true); + assertTrue(channel2.isInFallbackMode()); + + // Channel 2's localProbeSuccesses must be reset to 0 in new cycle + assertEquals(0, channel2.getLocalProbeSuccesses().get()); + + // Probe once: count is 1 (< 2 required), must still be in fallback + probeTask2.run(); + assertEquals(1, channel2.getLocalProbeSuccesses().get()); + assertTrue(channel2.isInFallbackMode()); + + // Probe second time: count is 2 (>= 2 required) -> recovers! + probeTask2.run(); + assertFalse(channel2.isInFallbackMode()); + assertFalse(sharedState.getInFallbackMode().get()); + } finally { + channel1.shutdownNow(); + channel2.shutdownNow(); + sharedState.shutdown(); + } + } }