Skip to content

feat(grpc-gcp): Add shared fallback state and probing recovery options to GcpFallbackChannel #14013

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
kinsaurralde wants to merge 20 commits into
base: main
Choose a base branch
from
Open

feat(grpc-gcp): Add shared fallback state and probing recovery options to GcpFallbackChannel #14013

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
20 commits
Select commit Hold shift + click to select a range
ee9825c
feat(grpc-gcp): add GcpFallbackState and probing recovery to GcpFallb…
kinsaurralde Aug 7, 2026
5321aa2
feat(grpc-gcp): add enableRecovery and enablePerChannelRecovery optio…
kinsaurralde Aug 7, 2026
945a856
fixes
kinsaurralde Aug 7, 2026
78a0a95
Merge branch 'googleapis:main' into gcp_fallback_shared_state
kinsaurralde Sep 3, 2026
4f0d6ac
fixes
kinsaurralde Sep 3, 2026
dfc89ca
fix test executor
kinsaurralde Sep 4, 2026
132ee86
fix
kinsaurralde Sep 4, 2026
f5ada91
formatting
kinsaurralde Sep 4, 2026
4173210
Merge branch 'main' into gcp_fallback_shared_state
rahul2393 Sep 4, 2026
075dc59
remove old code
kinsaurralde Sep 4, 2026
2416c26
fix evaluationStarted
kinsaurralde Sep 4, 2026
172503d
Merge branch 'main' into gcp_fallback_shared_state
rahul2393 Sep 4, 2026
509453e
Merge branch 'main' into gcp_fallback_shared_state
kinsaurralde Sep 9, 2026
6247c02
Merge branch 'main' into gcp_fallback_shared_state
kinsaurralde Sep 14, 2026
e3e21ae
add fallback generation
kinsaurralde Sep 15, 2026
5731104
fixes
kinsaurralde Sep 16, 2026
bfce1d5
refactor(grpc-gcp): defer per channel recovery
kinsaurralde Sep 17, 2026
2cde341
formatting
kinsaurralde Sep 17, 2026
187dea1
address comments
kinsaurralde Sep 17, 2026
7f20786
fixes
kinsaurralde Sep 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
245 changes: 154 additions & 91 deletions grpc-gcp-java/src/main/java/com/google/cloud/grpc/fallback/GcpFallbackChannel.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -31,8 +30,8 @@
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.AtomicLong;
import java.util.logging.Logger;
Expand All @@ -50,14 +49,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 AtomicLong localProbeGeneration = new AtomicLong(0);
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,
Expand All @@ -82,13 +84,16 @@ 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 =
execService != null ? new GcpFallbackState(execService) : new GcpFallbackState();
this.ownsFallbackState = true;
}
this.options = options;
this.execService = fallbackState.getOrCreateExecutorService(options);
if (options.getGcpOpenTelemetry() != null) {
this.openTelemetry = options.getGcpOpenTelemetry();
} else {
Expand Down Expand Up @@ -149,13 +154,16 @@ 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 =
execService != null ? new GcpFallbackState(execService) : new GcpFallbackState();
this.ownsFallbackState = true;
}
this.options = options;
this.execService = fallbackState.getOrCreateExecutorService(options);
if (options.getGcpOpenTelemetry() != null) {
this.openTelemetry = options.getGcpOpenTelemetry();
} else {
Expand All @@ -174,106 +182,108 @@ public GcpFallbackChannel(
init();
}

private void resetProbeStatsIfGenerationChanged() {
long currentGen = fallbackState.getGeneration();
if (localProbeGeneration.getAndSet(currentGen) != currentGen) {
localProbeSuccesses.set(0);
localFirstPrimaryProbeSuccessNanos.set(0);
}
}
Comment thread
kinsaurralde marked this conversation as resolved.

public boolean isInFallbackMode() {
return inFallbackMode || primaryChannel == null;
return (fallbackState.isInFallbackMode() && fallbackChannel != null) || primaryChannel == null;
}

@VisibleForTesting
GcpFallbackState getFallbackState() {
return fallbackState;
}

@VisibleForTesting
AtomicLong getLocalProbeSuccesses() {
resetProbeStatsIfGenerationChanged();
return localProbeSuccesses;
}

private void init() {
localProbeGeneration.set(fallbackState.getGeneration());
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);
}
}

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.startPeriodicEvaluation(options);
}

private void processPrimaryStatusCode(Status.Code statusCode) {
if (options.getErroneousStates().contains(statusCode)) {
// Count error.
primaryFailures.incrementAndGet();
} else {
// Count success.
primarySuccesses.incrementAndGet();
if (fallbackChannel != null && !fallbackState.isInFallbackMode()) {
if (options.getErroneousStates().contains(statusCode)) {
fallbackState.getPrimaryFailures().incrementAndGet();
} else {
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.isInFallbackMode() && primaryChannel != null) {
return;
}
resetProbeStatsIfGenerationChanged();
long probeStartGen = localProbeGeneration.get();
String result = "";
if (primaryDelegateChannel == null) {
result = INIT_FAILURE_REASON;
} else {
result = options.getPrimaryProbingFunction().apply(primaryDelegateChannel);
}
if ("".equals(result) && fallbackState.getGeneration() == probeStartGen) {
if (options.isEnableRecovery() && fallbackChannel != null) {
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 (primaryProbeSuccessCount >= options.getMinPrimaryProbeSuccessCount()
&& durationSatisfied) {
fallbackState.recordRecovery(probeStartGen);
}
}
} else {
localProbeSuccesses.set(0);
localFirstPrimaryProbeSuccessNanos.set(0);
}
// Report metric based on result.
openTelemetry.getModule().reportProbeResult(options.getPrimaryChannelName(), result);
}
Comment thread
kinsaurralde marked this conversation as resolved.
Expand Down Expand Up @@ -308,27 +318,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);
}
}
}
Comment thread
kinsaurralde marked this conversation as resolved.
Comment thread
kinsaurralde marked this conversation as resolved.
Comment thread
kinsaurralde marked this conversation as resolved.

@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;
}

Expand All @@ -342,7 +396,10 @@ public boolean isShutdown() {
return false;
}

return execService.isShutdown();
if (ownsFallbackState && options.getSharedExecutorService() == null) {
return execService.isShutdown();
}
return true;
}

@Override
Expand All @@ -355,7 +412,10 @@ public boolean isTerminated() {
return false;
}

return execService.isTerminated();
if (ownsFallbackState && options.getSharedExecutorService() == null) {
return execService.isTerminated();
}
return true;
}

@Override
Expand All @@ -377,6 +437,9 @@ public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedE
awaitTimeNanos = endTimeNanos - System.nanoTime();
}

return execService.awaitTermination(awaitTimeNanos, NANOSECONDS);
if (ownsFallbackState && options.getSharedExecutorService() == null) {
return execService.awaitTermination(awaitTimeNanos, NANOSECONDS);
}
return true;
}
}
Loading
Loading