fix(spanner): remove directpath path and only use cloudpath admin clients #13157
+199
−33
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -103,12 +103,17 @@ | |
| import io.opentelemetry.sdk.trace.SdkTracerProvider; | ||
| import io.opentelemetry.sdk.trace.samplers.Sampler; | ||
| import java.io.IOException; | ||
| import java.lang.reflect.Array; | ||
| import java.lang.reflect.Modifier; | ||
| import java.net.InetSocketAddress; | ||
| import java.time.Duration; | ||
| import java.util.Collection; | ||
| import java.util.Collections; | ||
| import java.util.HashMap; | ||
| import java.util.IdentityHashMap; | ||
| import java.util.Map; | ||
| import java.util.Objects; | ||
| import java.util.Set; | ||
| import java.util.concurrent.Executor; | ||
| import java.util.concurrent.ScheduledExecutorService; | ||
| import java.util.concurrent.TimeUnit; | ||
|
|
@@ -172,6 +177,25 @@ public class GapicSpannerRpcTest { | |
| new java.util.Date( | ||
| System.currentTimeMillis() + TimeUnit.MILLISECONDS.convert(1L, TimeUnit.DAYS)))); | ||
|
|
||
| private static final String GRPC_GCP_CHANNEL_REF_CLASS_NAME = | ||
| "com.google.cloud.grpc.GcpManagedChannel$ChannelRef"; | ||
|
|
||
| private static final class GrpcGcpObjectCounts { | ||
| int gcpManagedChannels; | ||
| int channelRefs; | ||
|
|
||
| GrpcGcpObjectCounts minus(GrpcGcpObjectCounts other) { | ||
| GrpcGcpObjectCounts difference = new GrpcGcpObjectCounts(); | ||
| difference.gcpManagedChannels = gcpManagedChannels - other.gcpManagedChannels; | ||
| difference.channelRefs = channelRefs - other.channelRefs; | ||
| return difference; | ||
| } | ||
|
|
||
| String debugString() { | ||
| return "GcpManagedChannel=" + gcpManagedChannels + ", ChannelRef=" + channelRefs; | ||
| } | ||
| } | ||
|
|
||
| private static MockSpannerServiceImpl mockSpanner; | ||
| private static Server server; | ||
| private static InetSocketAddress address; | ||
|
|
@@ -1409,6 +1433,134 @@ private SpannerOptions createSpannerOptions() { | |
| .build(); | ||
| } | ||
|
|
||
| @Test | ||
| public void testDirectPathFallbackUsesCloudPathForAdminClients() { | ||
| SpannerOptions.useEnvironment(new SpannerOptions.SpannerEnvironment() {}); | ||
| GapicSpannerRpc rpc = null; | ||
| try { | ||
| SpannerOptions options = createDirectPathFallbackObjectCountOptions().build(); | ||
| assumeTrue( | ||
| "GCP fallback must be enabled for this DirectPath fallback test", | ||
| options.isEnableGcpFallback()); | ||
| GrpcGcpObjectCounts before = countGrpcGcpObjectsFromChannelz(); | ||
| rpc = new GapicSpannerRpc(options); | ||
| GrpcGcpObjectCounts counts = countGrpcGcpObjectsFromChannelz().minus(before); | ||
| assertEquals(counts.debugString(), 4, counts.gcpManagedChannels); | ||
| assertEquals(counts.debugString(), 32, counts.channelRefs); | ||
| } finally { | ||
| if (rpc != null) { | ||
| rpc.shutdown(); | ||
| } | ||
| SpannerOptions.useDefaultEnvironment(); | ||
| } | ||
| } | ||
|
|
||
| @Test | ||
| public void testDirectPathFallbackWithGaxChannelPoolDoesNotCreateGrpcGcpChannelRefs() { | ||
| SpannerOptions.useEnvironment(new SpannerOptions.SpannerEnvironment() {}); | ||
| GapicSpannerRpc rpc = null; | ||
| try { | ||
| SpannerOptions options = | ||
| createDirectPathFallbackObjectCountOptions().disableGrpcGcpExtension().build(); | ||
| assumeTrue( | ||
| "GCP fallback must be enabled for this DirectPath fallback test", | ||
| options.isEnableGcpFallback()); | ||
| GrpcGcpObjectCounts before = countGrpcGcpObjectsFromChannelz(); | ||
| rpc = new GapicSpannerRpc(options); | ||
| GrpcGcpObjectCounts counts = countGrpcGcpObjectsFromChannelz().minus(before); | ||
| assertEquals(counts.debugString(), 0, counts.gcpManagedChannels); | ||
| assertEquals(counts.debugString(), 0, counts.channelRefs); | ||
| } finally { | ||
| if (rpc != null) { | ||
| rpc.shutdown(); | ||
| } | ||
| SpannerOptions.useDefaultEnvironment(); | ||
| } | ||
| } | ||
|
|
||
| private SpannerOptions.Builder createDirectPathFallbackObjectCountOptions() { | ||
| return SpannerOptions.newBuilder() | ||
| .setProjectId("test-project") | ||
| .setEnableDirectAccess(true) | ||
| .setHost("http://localhost:1") | ||
| .setCredentials(NoCredentials.getInstance()); | ||
| } | ||
|
|
||
| private static GrpcGcpObjectCounts countGrpcGcpObjectsFromChannelz() { | ||
| GrpcGcpObjectCounts counts = new GrpcGcpObjectCounts(); | ||
| Object channelz = io.grpc.InternalChannelz.instance(); | ||
| Set<Object> visited = Collections.newSetFromMap(new IdentityHashMap<>()); | ||
| countGrpcGcpObjectsFromChannelzField(channelz, "rootChannels", visited, counts); | ||
| countGrpcGcpObjectsFromChannelzField(channelz, "subchannels", visited, counts); | ||
| return counts; | ||
| } | ||
|
|
||
| private static void countGrpcGcpObjectsFromChannelzField( | ||
| Object channelz, String fieldName, Set<Object> visited, GrpcGcpObjectCounts counts) { | ||
| try { | ||
| java.lang.reflect.Field field = channelz.getClass().getDeclaredField(fieldName); | ||
| field.setAccessible(true); | ||
| countGrpcGcpObjects(field.get(channelz), visited, counts); | ||
| } catch (RuntimeException | ReflectiveOperationException ignored) { | ||
| // Ignore fields that are not reflectively accessible in this runtime. | ||
| } | ||
| } | ||
|
|
||
| private static void countGrpcGcpObjects( | ||
| Object object, Set<Object> visited, GrpcGcpObjectCounts counts) { | ||
| if (object == null || !visited.add(object)) { | ||
| return; | ||
| } | ||
| if (object instanceof GcpManagedChannel) { | ||
| counts.gcpManagedChannels++; | ||
| } | ||
| Class<?> clazz = object.getClass(); | ||
| if (clazz.getName().equals(GRPC_GCP_CHANNEL_REF_CLASS_NAME)) { | ||
| counts.channelRefs++; | ||
| } | ||
| if (object instanceof Collection<?>) { | ||
| for (Object value : (Collection<?>) object) { | ||
| countGrpcGcpObjects(value, visited, counts); | ||
| } | ||
| return; | ||
| } | ||
| if (object instanceof Map<?, ?>) { | ||
| for (Map.Entry<?, ?> entry : ((Map<?, ?>) object).entrySet()) { | ||
| countGrpcGcpObjects(entry.getKey(), visited, counts); | ||
| countGrpcGcpObjects(entry.getValue(), visited, counts); | ||
| } | ||
| return; | ||
| } | ||
| if (clazz.isArray()) { | ||
| int length = Array.getLength(object); | ||
| for (int i = 0; i < length; i++) { | ||
| countGrpcGcpObjects(Array.get(object, i), visited, counts); | ||
| } | ||
| return; | ||
| } | ||
| if (!shouldInspectFields(clazz)) { | ||
| return; | ||
| } | ||
| for (Class<?> current = clazz; current != null; current = current.getSuperclass()) { | ||
| for (java.lang.reflect.Field field : current.getDeclaredFields()) { | ||
| if (Modifier.isStatic(field.getModifiers())) { | ||
| continue; | ||
| } | ||
| try { | ||
| field.setAccessible(true); | ||
| countGrpcGcpObjects(field.get(object), visited, counts); | ||
| } catch (RuntimeException | IllegalAccessException ignored) { | ||
| // Ignore fields that are not reflectively accessible in this runtime. | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private static boolean shouldInspectFields(Class<?> clazz) { | ||
| String name = clazz.getName(); | ||
| return name.startsWith("com.google.") || name.startsWith("io.grpc."); | ||
| } | ||
|
|
||
| static class TestableGapicSpannerRpc extends GapicSpannerRpc { | ||
| public TestableGapicSpannerRpc(SpannerOptions options) { | ||
| super(options); | ||
|
|
||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
When the GCP extension is enabled, the GAX pool size should be set to 1 to avoid redundant connection pooling, as the extension manages its own internal pool. This ensures consistency with the logic used when fallback is enabled (see line 701).