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 671562b5aad4..547fdc8bca10 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 @@ -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); } 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 index 266d0176947d..17c7ee60bbc9 100644 --- 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 @@ -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 @@ -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); @@ -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() diff --git a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/GapicSpannerRpc.java b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/GapicSpannerRpc.java index 7d969d39fc59..5057c1025d21 100644 --- a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/GapicSpannerRpc.java +++ b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/GapicSpannerRpc.java @@ -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; @@ -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); } @@ -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 { @@ -594,15 +602,26 @@ public UnaryCallable 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(); } @@ -700,9 +719,17 @@ public ClientCall 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 = @@ -710,25 +737,13 @@ public ClientCall interceptCall( 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; }); } @@ -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 @@ -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 @@ -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); @@ -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(); @@ -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 { + private final java.util.concurrent.atomic.AtomicReference 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> extraHeaders = + metadataProvider.newExtraHeaders(name, projectName); + + io.grpc.Metadata metadata = new io.grpc.Metadata(); + for (java.util.Map.Entry> entry : extraHeaders.entrySet()) { + io.grpc.Metadata.Key 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 capturedAttributes = + new java.util.concurrent.atomic.AtomicReference<>(); + + io.grpc.ClientInterceptor attributeInterceptor = + new io.grpc.ClientInterceptor() { + @Override + public io.grpc.ClientCall interceptCall( + io.grpc.MethodDescriptor method, + io.grpc.CallOptions callOptions, + io.grpc.Channel next) { + io.grpc.ClientCall call = next.newCall(method, callOptions); + return new io.grpc.ForwardingClientCall.SimpleForwardingClientCall( + call) { + @Override + public void start(Listener 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 { diff --git a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/spi/v1/GapicSpannerRpcTest.java b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/spi/v1/GapicSpannerRpcTest.java index 72c0f47243be..779c20686a31 100644 --- a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/spi/v1/GapicSpannerRpcTest.java +++ b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/spi/v1/GapicSpannerRpcTest.java @@ -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(); @@ -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) @@ -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)