Skip to content

Commit 4561bb5

Browse files
ejona86ejona86
committed
Plumb target to load balancer
gRFC A78 has WRR and pick-first include a `grpc.target` label, defined in A66: > `grpc.target` : Canonicalized target URI used when creating gRPC > Channel, e.g. "dns:///pubsub.googleapis.com:443", > "xds:///helloworld-gke:8000". Canonicalized target URI is the form > with the scheme included if the user didn't mention the scheme > (`scheme://[authority]/path`). For channels such as inprocess channels > where a target URI is not available, implementations can synthesize a > target URI.
1 parent 27d5758 commit 4561bb5

File tree

5 files changed

+145
-101
lines changed

5 files changed

+145
-101
lines changed

‎api/src/main/java/io/grpc/LoadBalancer.java‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1185,6 +1185,13 @@ public ScheduledExecutorService getScheduledExecutorService() {
11851185
*/
11861186
public abstract String getAuthority();
11871187

1188+
/**
1189+
* Returns the target string of the channel, guaranteed to include its scheme.
1190+
*/
1191+
public String getChannelTarget() {
1192+
throw new UnsupportedOperationException();
1193+
}
1194+
11881195
/**
11891196
* Returns the ChannelCredentials used to construct the channel, without bearer tokens.
11901197
*

‎core/src/main/java/io/grpc/internal/ManagedChannelImpl.java‎

Lines changed: 34 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -166,6 +166,8 @@ public Result selectConfig(PickSubchannelArgs args) {
166166
@Nullable
167167
private final String authorityOverride;
168168
private final NameResolverRegistry nameResolverRegistry;
169+
private final URI targetUri;
170+
private final NameResolverProvider nameResolverProvider;
169171
private final NameResolver.Args nameResolverArgs;
170172
private final AutoConfiguredLoadBalancerFactory loadBalancerFactory;
171173
private final ClientTransportFactory originalTransportFactory;
@@ -383,8 +385,7 @@ private void shutdownNameResolverAndLoadBalancer(boolean channelIsActive) {
383385
nameResolverStarted = false;
384386
if (channelIsActive) {
385387
nameResolver = getNameResolver(
386-
target, authorityOverride, nameResolverRegistry, nameResolverArgs,
387-
transportFactory.getSupportedSocketAddressTypes());
388+
targetUri, authorityOverride, nameResolverProvider, nameResolverArgs);
388389
} else {
389390
nameResolver = null;
390391
}
@@ -621,6 +622,10 @@ ClientStream newSubstream(
621622
this.retryEnabled = builder.retryEnabled;
622623
this.loadBalancerFactory = new AutoConfiguredLoadBalancerFactory(builder.defaultLbPolicy);
623624
this.nameResolverRegistry = builder.nameResolverRegistry;
625+
ResolvedNameResolver resolvedResolver = getNameResolverProvider(
626+
target, nameResolverRegistry, transportFactory.getSupportedSocketAddressTypes());
627+
this.targetUri = resolvedResolver.targetUri;
628+
this.nameResolverProvider = resolvedResolver.provider;
624629
ScParser serviceConfigParser =
625630
new ScParser(
626631
retryEnabled,
@@ -640,8 +645,7 @@ ClientStream newSubstream(
640645
.setOverrideAuthority(this.authorityOverride)
641646
.build();
642647
this.nameResolver = getNameResolver(
643-
target, authorityOverride, nameResolverRegistry, nameResolverArgs,
644-
transportFactory.getSupportedSocketAddressTypes());
648+
targetUri, authorityOverride, nameResolverProvider, nameResolverArgs);
645649
this.balancerRpcExecutorPool = checkNotNull(balancerRpcExecutorPool, "balancerRpcExecutorPool");
646650
this.balancerRpcExecutorHolder = new ExecutorHolder(balancerRpcExecutorPool);
647651
this.delayedTransport = new DelayedClientTransport(this.executor, this.syncContext);
@@ -713,8 +717,20 @@ public CallTracer create() {
713717
}
714718
}
715719

716-
private static NameResolver getNameResolver(
717-
String target, NameResolverRegistry nameResolverRegistry, NameResolver.Args nameResolverArgs,
720+
@VisibleForTesting
721+
static class ResolvedNameResolver {
722+
public final URI targetUri;
723+
public final NameResolverProvider provider;
724+
725+
public ResolvedNameResolver(URI targetUri, NameResolverProvider provider) {
726+
this.targetUri = checkNotNull(targetUri, "targetUri");
727+
this.provider = checkNotNull(provider, "provider");
728+
}
729+
}
730+
731+
@VisibleForTesting
732+
static ResolvedNameResolver getNameResolverProvider(
733+
String target, NameResolverRegistry nameResolverRegistry,
718734
Collection<Class<? extends SocketAddress>> channelTransportSocketAddressTypes) {
719735
// Finding a NameResolver. Try using the target string as the URI. If that fails, try prepending
720736
// "dns:///".
@@ -761,23 +777,17 @@ private static NameResolver getNameResolver(
761777
}
762778
}
763779

764-
NameResolver resolver = provider.newNameResolver(targetUri, nameResolverArgs);
765-
if (resolver != null) {
766-
return resolver;
767-
}
768-
769-
throw new IllegalArgumentException(String.format(
770-
"cannot create a NameResolver for %s%s",
771-
target, uriSyntaxErrors.length() > 0 ? " (" + uriSyntaxErrors + ")" : ""));
780+
return new ResolvedNameResolver(targetUri, provider);
772781
}
773782

774783
@VisibleForTesting
775784
static NameResolver getNameResolver(
776-
String target, @Nullable final String overrideAuthority,
777-
NameResolverRegistry nameResolverRegistry, NameResolver.Args nameResolverArgs,
778-
Collection<Class<? extends SocketAddress>> channelTransportSocketAddressTypes) {
779-
NameResolver resolver = getNameResolver(target, nameResolverRegistry, nameResolverArgs,
780-
channelTransportSocketAddressTypes);
785+
URI targetUri, @Nullable final String overrideAuthority,
786+
NameResolverProvider provider, NameResolver.Args nameResolverArgs) {
787+
NameResolver resolver = provider.newNameResolver(targetUri, nameResolverArgs);
788+
if (resolver == null) {
789+
throw new IllegalArgumentException("cannot create a NameResolver for " + targetUri);
790+
}
781791

782792
// We wrap the name resolver in a RetryingNameResolver to give it the ability to retry failures.
783793
// TODO: After a transition period, all NameResolver implementations that need retry should use
@@ -1703,6 +1713,11 @@ public String getAuthority() {
17031713
return ManagedChannelImpl.this.authority();
17041714
}
17051715

1716+
@Override
1717+
public String getChannelTarget() {
1718+
return targetUri.toString();
1719+
}
1720+
17061721
@Override
17071722
public SynchronizationContext getSynchronizationContext() {
17081723
return syncContext;

‎core/src/test/java/io/grpc/internal/ManagedChannelImplGetNameResolverTest.java‎

Lines changed: 14 additions & 82 deletions
Original file line numberDiff line numberDiff line change
@@ -17,40 +17,22 @@
1717
package io.grpc.internal;
1818

1919
import static com.google.common.truth.Truth.assertThat;
20-
import static org.junit.Assert.assertEquals;
21-
import static org.junit.Assert.assertNotNull;
2220
import static org.junit.Assert.fail;
23-
import static org.mockito.Mockito.mock;
2421

25-
import io.grpc.ChannelLogger;
2622
import io.grpc.NameResolver;
27-
import io.grpc.NameResolver.Args;
28-
import io.grpc.NameResolver.ServiceConfigParser;
2923
import io.grpc.NameResolverProvider;
3024
import io.grpc.NameResolverRegistry;
31-
import io.grpc.ProxyDetector;
32-
import io.grpc.SynchronizationContext;
3325
import io.grpc.inprocess.InProcessSocketAddress;
34-
import java.lang.Thread.UncaughtExceptionHandler;
3526
import java.net.InetSocketAddress;
3627
import java.net.URI;
3728
import java.util.Collections;
3829
import org.junit.Test;
3930
import org.junit.runner.RunWith;
4031
import org.junit.runners.JUnit4;
4132

42-
/** Unit tests for ManagedChannelImpl#getNameResolver(). */
33+
/** Unit tests for ManagedChannelImpl#getNameResolverProvider(). */
4334
@RunWith(JUnit4.class)
4435
public class ManagedChannelImplGetNameResolverTest {
45-
private static final NameResolver.Args NAMERESOLVER_ARGS = NameResolver.Args.newBuilder()
46-
.setDefaultPort(447)
47-
.setProxyDetector(mock(ProxyDetector.class))
48-
.setSynchronizationContext(new SynchronizationContext(mock(UncaughtExceptionHandler.class)))
49-
.setServiceConfigParser(mock(ServiceConfigParser.class))
50-
.setChannelLogger(mock(ChannelLogger.class))
51-
.setScheduledExecutorService(new FakeClock().getScheduledExecutorService())
52-
.build();
53-
5436
@Test
5537
public void invalidUriTarget() {
5638
testInvalidTarget("defaultscheme:///[invalid]");
@@ -68,18 +50,6 @@ public void validAuthorityTarget() throws Exception {
6850
new URI("defaultscheme", "", "/foo.googleapis.com:8080", null));
6951
}
7052

71-
@Test
72-
public void validAuthorityTarget_overrideAuthority() throws Exception {
73-
String target = "foo.googleapis.com:8080";
74-
String overrideAuthority = "override.authority";
75-
URI expectedUri = new URI("defaultscheme", "", "/foo.googleapis.com:8080", null);
76-
NameResolverRegistry nameResolverRegistry = getTestRegistry(expectedUri.getScheme());
77-
NameResolver nameResolver = ManagedChannelImpl.getNameResolver(
78-
target, overrideAuthority, nameResolverRegistry, NAMERESOLVER_ARGS,
79-
Collections.singleton(InetSocketAddress.class));
80-
assertThat(nameResolver.getServiceAuthority()).isEqualTo(overrideAuthority);
81-
}
82-
8353
@Test
8454
public void validUriTarget() throws Exception {
8555
testValidTarget("scheme:///foo.googleapis.com:8080", "scheme:///foo.googleapis.com:8080",
@@ -121,47 +91,12 @@ public void validTargetStartingWithSlash() throws Exception {
12191
new URI("defaultscheme", "", "//target", null));
12292
}
12393

124-
@Test
125-
public void validTargetNoResolver() {
126-
NameResolverRegistry nameResolverRegistry = new NameResolverRegistry();
127-
NameResolverProvider nameResolverProvider = new NameResolverProvider() {
128-
@Override
129-
protected boolean isAvailable() {
130-
return true;
131-
}
132-
133-
@Override
134-
protected int priority() {
135-
return 5;
136-
}
137-
138-
@Override
139-
public NameResolver newNameResolver(URI targetUri, Args args) {
140-
return null;
141-
}
142-
143-
@Override
144-
public String getDefaultScheme() {
145-
return "defaultscheme";
146-
}
147-
};
148-
nameResolverRegistry.register(nameResolverProvider);
149-
try {
150-
ManagedChannelImpl.getNameResolver(
151-
"foo.googleapis.com:8080", null, nameResolverRegistry, NAMERESOLVER_ARGS,
152-
Collections.singleton(InetSocketAddress.class));
153-
fail("Should fail");
154-
} catch (IllegalArgumentException e) {
155-
// expected
156-
}
157-
}
158-
15994
@Test
16095
public void validTargetNoProvider() {
16196
NameResolverRegistry nameResolverRegistry = new NameResolverRegistry();
16297
try {
163-
ManagedChannelImpl.getNameResolver(
164-
"foo.googleapis.com:8080", null, nameResolverRegistry, NAMERESOLVER_ARGS,
98+
ManagedChannelImpl.getNameResolverProvider(
99+
"foo.googleapis.com:8080", nameResolverRegistry,
165100
Collections.singleton(InetSocketAddress.class));
166101
fail("Should fail");
167102
} catch (IllegalArgumentException e) {
@@ -173,8 +108,8 @@ public void validTargetNoProvider() {
173108
public void validTargetProviderAddrTypesNotSupported() {
174109
NameResolverRegistry nameResolverRegistry = getTestRegistry("testscheme");
175110
try {
176-
ManagedChannelImpl.getNameResolver(
177-
"testscheme:///foo.googleapis.com:8080", null, nameResolverRegistry, NAMERESOLVER_ARGS,
111+
ManagedChannelImpl.getNameResolverProvider(
112+
"testscheme:///foo.googleapis.com:8080", nameResolverRegistry,
178113
Collections.singleton(InProcessSocketAddress.class));
179114
fail("Should fail");
180115
} catch (IllegalArgumentException e) {
@@ -184,26 +119,23 @@ public void validTargetProviderAddrTypesNotSupported() {
184119
}
185120
}
186121

187-
188122
private void testValidTarget(String target, String expectedUriString, URI expectedUri) {
189123
NameResolverRegistry nameResolverRegistry = getTestRegistry(expectedUri.getScheme());
190-
FakeNameResolver nameResolver
191-
= (FakeNameResolver) ((RetryingNameResolver) ManagedChannelImpl.getNameResolver(
192-
target, null, nameResolverRegistry, NAMERESOLVER_ARGS,
193-
Collections.singleton(InetSocketAddress.class))).getRetriedNameResolver();
194-
assertNotNull(nameResolver);
195-
assertEquals(expectedUri, nameResolver.uri);
196-
assertEquals(expectedUriString, nameResolver.uri.toString());
124+
ManagedChannelImpl.ResolvedNameResolver resolved = ManagedChannelImpl.getNameResolverProvider(
125+
target, nameResolverRegistry, Collections.singleton(InetSocketAddress.class));
126+
assertThat(resolved.provider).isInstanceOf(FakeNameResolverProvider.class);
127+
assertThat(resolved.targetUri).isEqualTo(expectedUri);
128+
assertThat(resolved.targetUri.toString()).isEqualTo(expectedUriString);
197129
}
198130

199131
private void testInvalidTarget(String target) {
200132
NameResolverRegistry nameResolverRegistry = getTestRegistry("dns");
201133

202134
try {
203-
FakeNameResolver nameResolver = (FakeNameResolver) ManagedChannelImpl.getNameResolver(
204-
target, null, nameResolverRegistry, NAMERESOLVER_ARGS,
205-
Collections.singleton(InetSocketAddress.class));
206-
fail("Should have failed, but got resolver with " + nameResolver.uri);
135+
ManagedChannelImpl.ResolvedNameResolver resolved = ManagedChannelImpl.getNameResolverProvider(
136+
target, nameResolverRegistry, Collections.singleton(InetSocketAddress.class));
137+
FakeNameResolverProvider nameResolverProvider = (FakeNameResolverProvider) resolved.provider;
138+
fail("Should have failed, but got resolver provider " + nameResolverProvider);
207139
} catch (IllegalArgumentException e) {
208140
// expected
209141
}

‎core/src/test/java/io/grpc/internal/ManagedChannelImplTest.java‎

Lines changed: 85 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -104,6 +104,7 @@
104104
import io.grpc.NameResolver;
105105
import io.grpc.NameResolver.ConfigOrError;
106106
import io.grpc.NameResolver.ResolutionResult;
107+
import io.grpc.NameResolverProvider;
107108
import io.grpc.NameResolverRegistry;
108109
import io.grpc.ProxiedSocketAddress;
109110
import io.grpc.ProxyDetector;
@@ -112,6 +113,7 @@
112113
import io.grpc.Status;
113114
import io.grpc.Status.Code;
114115
import io.grpc.StringMarshaller;
116+
import io.grpc.SynchronizationContext;
115117
import io.grpc.internal.ClientTransportFactory.ClientTransportOptions;
116118
import io.grpc.internal.ClientTransportFactory.SwapChannelCredentialsResult;
117119
import io.grpc.internal.InternalSubchannel.TransportLogger;
@@ -188,6 +190,15 @@ public class ManagedChannelImplTest {
188190
.setUserAgent(USER_AGENT);
189191
private static final String TARGET = "fake://" + SERVICE_NAME;
190192
private static final String MOCK_POLICY_NAME = "mock_lb";
193+
private static final NameResolver.Args NAMERESOLVER_ARGS = NameResolver.Args.newBuilder()
194+
.setDefaultPort(447)
195+
.setProxyDetector(mock(ProxyDetector.class))
196+
.setSynchronizationContext(
197+
new SynchronizationContext(mock(Thread.UncaughtExceptionHandler.class)))
198+
.setServiceConfigParser(mock(NameResolver.ServiceConfigParser.class))
199+
.setScheduledExecutorService(new FakeClock().getScheduledExecutorService())
200+
.build();
201+
191202
private URI expectedUri;
192203
private final SocketAddress socketAddress =
193204
new SocketAddress() {
@@ -4306,6 +4317,80 @@ public void transportTerminated(Attributes transportAttrs) {
43064317
assertEquals(1, terminationCallbackCalled.get());
43074318
}
43084319

4320+
@Test
4321+
public void validAuthorityTarget_overrideAuthority() throws Exception {
4322+
String overrideAuthority = "override.authority";
4323+
String serviceAuthority = "fakeauthority";
4324+
NameResolverProvider nameResolverProvider = new NameResolverProvider() {
4325+
@Override protected boolean isAvailable() {
4326+
return true;
4327+
}
4328+
4329+
@Override protected int priority() {
4330+
return 5;
4331+
}
4332+
4333+
@Override public NameResolver newNameResolver(URI targetUri, NameResolver.Args args) {
4334+
return new NameResolver() {
4335+
@Override public String getServiceAuthority() {
4336+
return serviceAuthority;
4337+
}
4338+
4339+
@Override public void start(final Listener2 listener) {}
4340+
4341+
@Override public void shutdown() {}
4342+
};
4343+
}
4344+
4345+
@Override public String getDefaultScheme() {
4346+
return "defaultscheme";
4347+
}
4348+
};
4349+
4350+
URI targetUri = new URI("defaultscheme", "", "/foo.googleapis.com:8080", null);
4351+
NameResolver nameResolver = ManagedChannelImpl.getNameResolver(
4352+
targetUri, null, nameResolverProvider, NAMERESOLVER_ARGS);
4353+
assertThat(nameResolver.getServiceAuthority()).isEqualTo(serviceAuthority);
4354+
4355+
nameResolver = ManagedChannelImpl.getNameResolver(
4356+
targetUri, overrideAuthority, nameResolverProvider, NAMERESOLVER_ARGS);
4357+
assertThat(nameResolver.getServiceAuthority()).isEqualTo(overrideAuthority);
4358+
}
4359+
4360+
@Test
4361+
public void validTargetNoResolver_throws() {
4362+
NameResolverProvider nameResolverProvider = new NameResolverProvider() {
4363+
@Override
4364+
protected boolean isAvailable() {
4365+
return true;
4366+
}
4367+
4368+
@Override
4369+
protected int priority() {
4370+
return 5;
4371+
}
4372+
4373+
@Override
4374+
public NameResolver newNameResolver(URI targetUri, NameResolver.Args args) {
4375+
return null;
4376+
}
4377+
4378+
@Override
4379+
public String getDefaultScheme() {
4380+
return "defaultscheme";
4381+
}
4382+
};
4383+
try {
4384+
ManagedChannelImpl.getNameResolver(
4385+
URI.create("defaultscheme:///foo.gogoleapis.com:8080"),
4386+
null, nameResolverProvider, NAMERESOLVER_ARGS);
4387+
fail("Should fail");
4388+
} catch (IllegalArgumentException e) {
4389+
// expected
4390+
}
4391+
}
4392+
4393+
43094394
private static final class FakeBackoffPolicyProvider implements BackoffPolicy.Provider {
43104395
@Override
43114396
public BackoffPolicy get() {

0 commit comments

Comments
 (0)