Skip to content

Eef probing recovery #1

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

Draft
kinsaurralde wants to merge 1 commit into
base: gcp_fallback_shared_state
Choose a base branch
from
Draft

Eef probing recovery #1

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
1 change: 1 addition & 0 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 @@ -303,6 +303,7 @@ private void probePrimary() {
&& durationSatisfied) {
fallbackState.getInFallbackMode().set(false);
localInFallbackMode.set(false);
logger.info("[kinsaurralde] Primary channel recovered to DirectPath.");
localProbeSuccesses.set(0);
localFirstPrimaryProbeSuccessNanos.set(0);
}
Expand Down
3 changes: 3 additions & 0 deletions grpc-gcp-java/src/main/java/com/google/cloud/grpc/fallback/GcpFallbackState.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.logging.Logger;

/**
* Shared thread-safe state container for coordinated pool-wide failover, recovery, and background
Expand All @@ -32,6 +33,7 @@
* consolidating probing and error evaluation threads across the entire channel pool.
*/
public class GcpFallbackState {
private static final Logger logger = Logger.getLogger(GcpFallbackState.class.getName());
private final AtomicLong primarySuccesses = new AtomicLong(0);
private final AtomicLong primaryFailures = new AtomicLong(0);
private final AtomicLong fallbackSuccesses = new AtomicLong(0);
Expand Down Expand Up @@ -154,6 +156,7 @@ public void checkErrorRates(
if (!inFallbackMode.get() && options.isEnableFallback()) {
if (failures >= options.getMinFailedCalls() && errRate >= options.getErrorRateThreshold()) {
inFallbackMode.set(true);
logger.info("[kinsaurralde] Primary error rate threshold crossed. Switching pool to fallback mode.");
if (openTelemetry != null && openTelemetry.getModule() != null) {
openTelemetry
.getModule()
Expand Down
184 changes: 164 additions & 20 deletions ...r/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/GapicSpannerRpc.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -268,6 +268,9 @@ public class GapicSpannerRpc implements SpannerRpc {

private static final String API_FILE = "grpc-gcp-apiconfig.json";

private final com.google.cloud.grpc.fallback.GcpFallbackState sharedFallbackState =
new com.google.cloud.grpc.fallback.GcpFallbackState();

private final RequestIdCreator requestIdCreator = new RequestIdCreatorImpl();
private boolean rpcIsClosed;
private final SpannerStub spannerStub;
Expand Down Expand Up @@ -315,6 +318,9 @@ public class GapicSpannerRpc implements SpannerRpc {

private final GrpcCallContext baseGrpcCallContext;

// Inside GapicSpannerRpc class fields
private final GetSessionProbe getSessionProbe = new GetSessionProbe();

public static GapicSpannerRpc create(SpannerOptions options) {
return new GapicSpannerRpc(options);
}
Expand All @@ -324,6 +330,8 @@ public GapicSpannerRpc(final SpannerOptions options) {
}

GapicSpannerRpc(final SpannerOptions options, boolean initializeStubs) {
System.out.println(
"[kinsaurralde] #################2 Using Local Cloud Spanner Client #################^2");
this.projectId = options.getProjectId();
String projectNameStr = PROJECT_NAME_TEMPLATE.instantiate("project", this.projectId);
try {
Expand Down Expand Up @@ -594,15 +602,26 @@ public <RequestT, ResponseT> UnaryCallable<RequestT, ResponseT> createUnaryCalla
}
}

@VisibleForTesting
com.google.cloud.grpc.fallback.GcpFallbackState getSharedFallbackState() {
return sharedFallbackState;
}

@VisibleForTesting
GcpFallbackChannelOptions createFallbackChannelOptions(
GcpFallbackOpenTelemetry fallbackTelemetry, int minFailedCalls) {
return GcpFallbackChannelOptions.newBuilder()
.setSharedState(sharedFallbackState)
.setPrimaryChannelName("directpath")
.setFallbackChannelName("cloudpath")
.setMinFailedCalls(minFailedCalls)
.setPeriod(Duration.ofMinutes(3))
.setGcpFallbackOpenTelemetry(fallbackTelemetry)
.setPrimaryProbingFunction(getSessionProbe)
.setPrimaryProbingInterval(Duration.ofSeconds(10))
.setMinPrimaryProbeSuccessCount(50)
.setMinPrimaryProbeSuccessDuration(Duration.ofMinutes(10))
.setEnableRecovery(true)
.build();
}

Expand Down Expand Up @@ -700,35 +719,31 @@ public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
if (existingConfigurator != null) {
builder = existingConfigurator.apply(builder);
}
GcpFallbackOpenTelemetry fallbackTelemetry =
GcpFallbackOpenTelemetry.newBuilder()
.withSdk(getFallbackOpenTelemetry(options))
.disableAllMetrics()
.enableMetrics(Arrays.asList("fallback_count", "call_status"))
.build();

FallbackChannelBuilder fallbackDelegate =
new FallbackChannelBuilder(
builder, cloudPathBuilder, createFallbackChannelOptions(fallbackTelemetry, 1));

ManagedChannelBuilder<?> primaryBuilder = builder;
ManagedChannelBuilder<?> fallbackBuilder = cloudPathBuilder;
if (options.isGrpcGcpExtensionEnabled()) {
String jsonApiConfig = parseGrpcGcpApiConfig();
GcpManagedChannelOptions gcpOptions =
grpcGcpOptionsWithMetricsAndDcp(options, channelPrimer);
if (gcpOptions == null) {
gcpOptions = GcpManagedChannelOptions.newBuilder().build();
}
primaryBuilder =
GcpManagedChannelBuilder.forDelegateBuilder(builder)
.withApiConfigJsonString(jsonApiConfig)
.withOptions(gcpOptions);
fallbackBuilder =
GcpManagedChannelBuilder.forDelegateBuilder(cloudPathBuilder)
.withApiConfigJsonString(jsonApiConfig)
.withOptions(gcpOptions);
}

GcpFallbackOpenTelemetry fallbackTelemetry =
GcpFallbackOpenTelemetry.newBuilder()
.withSdk(getFallbackOpenTelemetry(options))
.disableAllMetrics()
.enableMetrics(Arrays.asList("fallback_count", "call_status"))
.build();
return GcpManagedChannelBuilder.forDelegateBuilder(fallbackDelegate)
.withApiConfigJsonString(jsonApiConfig)
.withOptions(gcpOptions);
}

return new FallbackChannelBuilder(
primaryBuilder, fallbackBuilder, createFallbackChannelOptions(fallbackTelemetry, 1));
return fallbackDelegate;
});
}

Expand Down Expand Up @@ -759,6 +774,7 @@ private InstantiatingGrpcChannelProvider.Builder createBaseChannelProviderBuilde
// commit GRPC calls succeed
.setKeepAliveTimeDuration(options.getGrpcKeepAliveTime())
.setKeepAliveTimeoutDuration(options.getGrpcKeepAliveTimeout())
.setKeepAliveWithoutCalls(true)

// Then check if SpannerOptions provides an InterceptorProvider. Create a default
// SpannerInterceptorProvider if none is provided
Expand Down Expand Up @@ -2057,7 +2073,15 @@ public Session createSession(
CreateSessionRequest request = requestBuilder.build();
GrpcCallContext context =
newCallContext(options, databaseName, request, SpannerGrpc.getCreateSessionMethod(), true);
return get(spannerStub.createSessionCallable().futureCall(request, context));
// Execute the call
Session session = get(spannerStub.createSessionCallable().futureCall(request, context));

// Capture the multiplexed session name for the prober
if (isMultiplexed && session != null) {
getSessionProbe.setSessionName(session.getName());
}

return session;
}

@Override
Expand Down Expand Up @@ -2548,6 +2572,7 @@ public void shutdown() {
this.instanceAdminStub.close();
this.databaseAdminStub.close();
this.spannerWatchdog.shutdown();
this.sharedFallbackState.shutdown();

try {
this.spannerStub.awaitTermination(10L, TimeUnit.SECONDS);
Expand All @@ -2569,6 +2594,7 @@ public void shutdownNow() {
this.instanceAdminStub.close();
this.databaseAdminStub.close();
this.spannerWatchdog.shutdown();
this.sharedFallbackState.shutdownNow();

this.spannerStub.shutdownNow();
this.partitionedDmlStub.shutdownNow();
Expand Down Expand Up @@ -2680,6 +2706,124 @@ private static Duration systemProperty(String name, int defaultValue) {
return Duration.ofSeconds(stringValue.isEmpty() ? defaultValue : Integer.parseInt(stringValue));
}

private class GetSessionProbe
implements com.google.common.base.Function<io.grpc.Channel, String> {
private final java.util.concurrent.atomic.AtomicReference<String> sessionName =
new java.util.concurrent.atomic.AtomicReference<>(null);

public void setSessionName(String name) {
System.out.println(
"[kinsaurralde] @@@@@@@@@@@@@@ Setting session name for GetSession Probe @@@@@@@@@@@@@@");
this.sessionName.set(name);
}

@Override
public String apply(io.grpc.Channel channel) {
String name = sessionName.get();
if (name == null) {
return "WAITING_FOR_REAL_SESSION";
}

try {
com.google.spanner.v1.GetSessionRequest request =
com.google.spanner.v1.GetSessionRequest.newBuilder().setName(name).build();

java.util.Map<String, java.util.List<String>> extraHeaders =
metadataProvider.newExtraHeaders(name, projectName);

io.grpc.Metadata metadata = new io.grpc.Metadata();
for (java.util.Map.Entry<String, java.util.List<String>> entry : extraHeaders.entrySet()) {
io.grpc.Metadata.Key<String> key =
io.grpc.Metadata.Key.of(entry.getKey(), io.grpc.Metadata.ASCII_STRING_MARSHALLER);
if (entry.getValue() != null) {
for (String value : entry.getValue()) {
metadata.put(key, value);
}
}
}

final java.util.concurrent.atomic.AtomicReference<io.grpc.Attributes> capturedAttributes =
new java.util.concurrent.atomic.AtomicReference<>();

io.grpc.ClientInterceptor attributeInterceptor =
new io.grpc.ClientInterceptor() {
@Override
public <ReqT, RespT> io.grpc.ClientCall<ReqT, RespT> interceptCall(
io.grpc.MethodDescriptor<ReqT, RespT> method,
io.grpc.CallOptions callOptions,
io.grpc.Channel next) {
io.grpc.ClientCall<ReqT, RespT> call = next.newCall(method, callOptions);
return new io.grpc.ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(
call) {
@Override
public void start(Listener<RespT> responseListener, io.grpc.Metadata headers) {
super.start(
new io.grpc.ForwardingClientCallListener.SimpleForwardingClientCallListener<
RespT>(responseListener) {
@Override
public void onHeaders(io.grpc.Metadata headers) {
capturedAttributes.set(call.getAttributes());
super.onHeaders(headers);
}
},
headers);
}
};
}
};

// Direct synchronous probe RPC executed on the specific underlying channel.
com.google.spanner.v1.SpannerGrpc.SpannerBlockingStub stub =
com.google.spanner.v1.SpannerGrpc.newBlockingStub(channel)
.withInterceptors(
io.grpc.stub.MetadataUtils.newAttachHeadersInterceptor(metadata),
attributeInterceptor)
.withDeadlineAfter(10, java.util.concurrent.TimeUnit.SECONDS);

if (callCredentialsProvider != null) {
io.grpc.CallCredentials creds = callCredentialsProvider.getCallCredentials();
if (creds != null) {
stub = stub.withCallCredentials(creds);
}
}

stub.getSession(request);

io.grpc.Attributes attrs = capturedAttributes.get();
if (attrs != null) {
java.net.SocketAddress remoteAddr = attrs.get(io.grpc.Grpc.TRANSPORT_ATTR_REMOTE_ADDR);
if (!isDirectPathTransport(remoteAddr, attrs)) {
System.out.println(
"[kinsaurralde] Probe succeeded but NOT using DirectPath transport (remoteAddr: "
+ remoteAddr
+ ")");
return "NOT_DIRECTPATH";
}
}

System.out.println(
"[kinsaurralde] GetSession Probe OK for single channel (DirectPath verified)!");
return "OK";
} catch (io.grpc.StatusRuntimeException e) {
System.out.println("[kinsaurralde] Probe FAILED with " + e.getStatus().getCode().name());
return e.getStatus().getCode().name();
} catch (Exception e) {
return "ERROR";
}
}

private boolean isDirectPathTransport(
java.net.SocketAddress remoteAddr, io.grpc.Attributes attrs) {
// DirectPath (both IPv4 and IPv6) uses ALTS security handshaking.
// CloudPath (GFE) uses standard TLS/SSL certificates.
Object sslSession = attrs.get(io.grpc.Grpc.TRANSPORT_ATTR_SSL_SESSION);
if (sslSession != null) {
return false; // Standard TLS session indicates GFE (CloudPath)
}
return true; // ALTS transport indicates DirectPath (IPv4 or IPv6)
}
}

// Wrapper class to build the GcpFallbackChannel using GAX's configuration
private static class FallbackChannelBuilder
extends ForwardingChannelBuilder2<FallbackChannelBuilder> {
Expand Down
6 changes: 4 additions & 2 deletions ...ogle-cloud-spanner/src/test/java/com/google/cloud/spanner/spi/v1/GapicSpannerRpcTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1880,8 +1880,8 @@ public void testDirectPathFallbackCreatesOneGrpcGcpLayerPerPath() {
GrpcGcpObjectCounts before = countGrpcGcpObjectsFromChannelz();
rpc = new GapicSpannerRpc(options);
GrpcGcpObjectCounts counts = countGrpcGcpObjectsFromChannelz().minus(before);
assertEquals(counts.debugString(), 6, counts.gcpManagedChannels);
assertEquals(counts.debugString(), 48, counts.channelRefs);
assertEquals(counts.debugString(), 3, counts.gcpManagedChannels);
assertEquals(counts.debugString(), 24, counts.channelRefs);
} finally {
if (rpc != null) {
rpc.shutdown();
Expand Down Expand Up @@ -2047,6 +2047,7 @@ GcpFallbackChannelOptions createFallbackChannelOptions(
GcpFallbackOpenTelemetry fallbackTelemetry, int minFailedCalls) {
// Override default 1-minute period to 10ms for instant testing
return GcpFallbackChannelOptions.newBuilder()
.setSharedState(getSharedFallbackState())
.setPrimaryChannelName("directpath")
.setFallbackChannelName("cloudpath")
.setMinFailedCalls(10)
Expand Down Expand Up @@ -2126,6 +2127,7 @@ GcpFallbackChannelOptions createFallbackChannelOptions(
GcpFallbackOpenTelemetry fallbackTelemetry, int minFailedCalls) {
// Override default 1-minute period to 10ms for instant testing
return GcpFallbackChannelOptions.newBuilder()
.setSharedState(getSharedFallbackState())
.setPrimaryChannelName("directpath")
.setFallbackChannelName("cloudpath")
.setMinFailedCalls(1)
Expand Down
Loading