Skip to content

Commit d8f7094

Browse files
larry-safranlarry-safran
committed
XdsClient Fallback working
1 parent 25aa454 commit d8f7094

24 files changed

+1350
-237
lines changed

‎census/src/test/java/io/grpc/census/CensusModulesTest.java

+8-1
Original file line numberDiff line numberDiff line change
@@ -56,6 +56,7 @@
5656
import io.grpc.ClientInterceptors;
5757
import io.grpc.ClientStreamTracer;
5858
import io.grpc.Context;
59+
import io.grpc.KnownLength;
5960
import io.grpc.Metadata;
6061
import io.grpc.MethodDescriptor;
6162
import io.grpc.ServerCall;
@@ -99,6 +100,7 @@
99100
import io.opencensus.trace.Tracer;
100101
import io.opencensus.trace.propagation.BinaryFormat;
101102
import io.opencensus.trace.propagation.SpanContextParseException;
103+
import java.io.IOException;
102104
import java.io.InputStream;
103105
import java.util.HashSet;
104106
import java.util.List;
@@ -136,7 +138,7 @@ public class CensusModulesTest {
136138
ClientStreamTracer.StreamInfo.newBuilder()
137139
.setCallOptions(CallOptions.DEFAULT.withOption(NAME_RESOLUTION_DELAYED, 10L)).build();
138140

139-
private static class StringInputStream extends InputStream {
141+
private static class StringInputStream extends InputStream implements KnownLength {
140142
final String string;
141143

142144
StringInputStream(String string) {
@@ -149,6 +151,11 @@ public int read() {
149151
// passed to the InProcess server and consumed by MARSHALLER.parse().
150152
throw new UnsupportedOperationException("Should not be called");
151153
}
154+
155+
@Override
156+
public int available() throws IOException {
157+
return string == null ? 0 : string.length();
158+
}
152159
}
153160

154161
private static final MethodDescriptor.Marshaller<String> MARSHALLER =

‎xds/src/main/java/io/grpc/xds/CsdsService.java

+30-12
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,7 @@
3939
import io.grpc.xds.client.XdsClient.ResourceMetadata.ResourceMetadataStatus;
4040
import io.grpc.xds.client.XdsClient.ResourceMetadata.UpdateFailureState;
4141
import io.grpc.xds.client.XdsResourceType;
42-
import java.util.HashMap;
42+
import java.util.ArrayList;
4343
import java.util.List;
4444
import java.util.Map;
4545
import java.util.concurrent.ExecutionException;
@@ -126,20 +126,32 @@ private boolean handleRequest(
126126
Status.INVALID_ARGUMENT.withDescription("node_matchers not supported"));
127127
} else {
128128
List<String> targets = xdsClientPoolFactory.getTargets();
129+
List<ClientConfig> clientConfigs = new ArrayList<>(targets.size());
130+
129131
for (int i = 0; i < targets.size() && error == null; i++) {
130-
String target = targets.get(i);
131132
try {
132-
responseObserver.onNext(getConfigDumpForRequest(target, request));
133+
ClientConfig clientConfig = getConfigForRequest(targets.get(i));
134+
if (clientConfig != null) {
135+
clientConfigs.add(clientConfig);
136+
}
133137
} catch (InterruptedException e) {
134138
Thread.currentThread().interrupt();
135139
logger.log(Level.FINE, "Server interrupted while building CSDS config dump", e);
136140
error = Status.ABORTED.withDescription("Thread interrupted").withCause(e).asException();
137141
} catch (RuntimeException e) {
138142
logger.log(Level.WARNING, "Unexpected error while building CSDS config dump", e);
139-
error =
140-
Status.INTERNAL.withDescription("Unexpected internal error").withCause(e).asException();
143+
error = Status.INTERNAL.withDescription("Unexpected internal error").withCause(e)
144+
.asException();
141145
}
142146
}
147+
148+
try {
149+
responseObserver.onNext(getStatusResponse(clientConfigs));
150+
} catch (RuntimeException e) {
151+
logger.log(Level.WARNING, "Unexpected error while processing CSDS config dump", e);
152+
error = Status.INTERNAL.withDescription("Unexpected internal error").withCause(e)
153+
.asException();
154+
}
143155
}
144156

145157
if (error == null) {
@@ -149,29 +161,35 @@ private boolean handleRequest(
149161
return false;
150162
}
151163

152-
private ClientStatusResponse getConfigDumpForRequest(String target, ClientStatusRequest request)
153-
throws InterruptedException {
164+
private ClientConfig getConfigForRequest(String target) throws InterruptedException {
154165
ObjectPool<XdsClient> xdsClientPool = xdsClientPoolFactory.get(target);
155166
if (xdsClientPool == null) {
156-
return ClientStatusResponse.getDefaultInstance();
167+
return null;
157168
}
158169

159170
XdsClient xdsClient = null;
160171
try {
161172
xdsClient = xdsClientPool.getObject();
162-
return ClientStatusResponse.newBuilder()
163-
.addConfig(getClientConfigForXdsClient(xdsClient))
164-
.build();
173+
return getClientConfigForXdsClient(xdsClient, target);
165174
} finally {
166175
if (xdsClient != null) {
167176
xdsClientPool.returnObject(xdsClient);
168177
}
169178
}
170179
}
171180

181+
private ClientStatusResponse getStatusResponse(List<ClientConfig> clientConfigs) {
182+
if (clientConfigs.isEmpty()) {
183+
return ClientStatusResponse.getDefaultInstance();
184+
}
185+
return ClientStatusResponse.newBuilder().addAllConfig(clientConfigs).build();
186+
}
187+
172188
@VisibleForTesting
173-
static ClientConfig getClientConfigForXdsClient(XdsClient xdsClient) throws InterruptedException {
189+
static ClientConfig getClientConfigForXdsClient(XdsClient xdsClient, String target)
190+
throws InterruptedException {
174191
ClientConfig.Builder builder = ClientConfig.newBuilder()
192+
.setClientScope(target)
175193
.setNode(xdsClient.getBootstrapInfo().node().toEnvoyProtoNode());
176194

177195
Map<XdsResourceType<?>, Map<String, ResourceMetadata>> metadataByType =

‎xds/src/main/java/io/grpc/xds/GrpcXdsTransportFactory.java

+7
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
import io.grpc.CallOptions;
2323
import io.grpc.ChannelCredentials;
2424
import io.grpc.ClientCall;
25+
import io.grpc.ConnectivityState;
2526
import io.grpc.Context;
2627
import io.grpc.Grpc;
2728
import io.grpc.ManagedChannel;
@@ -84,6 +85,12 @@ public void shutdown() {
8485
channel.shutdown();
8586
}
8687

88+
@Override
89+
public boolean isConnected() {
90+
ConnectivityState state = channel.getState(false);
91+
return state == ConnectivityState.READY;
92+
}
93+
8794
private class XdsStreamingCall<ReqT, RespT> implements
8895
XdsTransportFactory.StreamingCall<ReqT, RespT> {
8996

‎xds/src/main/java/io/grpc/xds/SharedXdsClientPoolProvider.java

+26-15
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,6 @@
2121

2222
import com.google.common.annotations.VisibleForTesting;
2323
import com.google.common.collect.ImmutableList;
24-
import com.sun.javafx.UnmodifiableArrayList;
2524
import io.grpc.internal.ExponentialBackoffPolicy;
2625
import io.grpc.internal.GrpcUtil;
2726
import io.grpc.internal.ObjectPool;
@@ -33,7 +32,6 @@
3332
import io.grpc.xds.client.XdsClientImpl;
3433
import io.grpc.xds.client.XdsInitializationException;
3534
import io.grpc.xds.internal.security.TlsContextManagerImpl;
36-
import java.util.List;
3735
import java.util.Map;
3836
import java.util.concurrent.ConcurrentHashMap;
3937
import java.util.concurrent.ScheduledExecutorService;
@@ -57,8 +55,7 @@ final class SharedXdsClientPoolProvider implements XdsClientPoolFactory {
5755
private final Bootstrapper bootstrapper;
5856
private final Object lock = new Object();
5957
private final AtomicReference<Map<String, ?>> bootstrapOverride = new AtomicReference<>();
60-
// private volatile ObjectPool<XdsClient> xdsClientPool;
61-
private Map<String, ObjectPool<XdsClient>> targetToXdsClientMap = new ConcurrentHashMap<>();
58+
private final Map<String, ObjectPool<XdsClient>> targetToXdsClientMap = new ConcurrentHashMap<>();
6259

6360
SharedXdsClientPoolProvider() {
6461
this(new GrpcBootstrapperImpl());
@@ -101,7 +98,7 @@ public ObjectPool<XdsClient> getOrCreate(String target) throws XdsInitialization
10198
if (bootstrapInfo.servers().isEmpty()) {
10299
throw new XdsInitializationException("No xDS server provided");
103100
}
104-
ref = new RefCountedXdsClientObjectPool(bootstrapInfo);
101+
ref = new RefCountedXdsClientObjectPool(bootstrapInfo, target);
105102
targetToXdsClientMap.put(target, ref);
106103
}
107104
}
@@ -122,7 +119,11 @@ private static class SharedXdsClientPoolProviderHolder {
122119
@ThreadSafe
123120
@VisibleForTesting
124121
static class RefCountedXdsClientObjectPool implements ObjectPool<XdsClient> {
122+
123+
private static final ExponentialBackoffPolicy.Provider BACKOFF_POLICY_PROVIDER =
124+
new ExponentialBackoffPolicy.Provider();
125125
private final BootstrapInfo bootstrapInfo;
126+
private final String target; // The target associated with the xDS client.
126127
private final Object lock = new Object();
127128
@GuardedBy("lock")
128129
private ScheduledExecutorService scheduler;
@@ -132,8 +133,9 @@ static class RefCountedXdsClientObjectPool implements ObjectPool<XdsClient> {
132133
private int refCount;
133134

134135
@VisibleForTesting
135-
RefCountedXdsClientObjectPool(BootstrapInfo bootstrapInfo) {
136+
RefCountedXdsClientObjectPool(BootstrapInfo bootstrapInfo, String target) {
136137
this.bootstrapInfo = checkNotNull(bootstrapInfo);
138+
this.target = target;
137139
}
138140

139141
@Override
@@ -144,15 +146,19 @@ public XdsClient getObject() {
144146
log.log(Level.INFO, "xDS node ID: {0}", bootstrapInfo.node().getId());
145147
}
146148
scheduler = SharedResourceHolder.get(GrpcUtil.TIMER_SERVICE);
147-
xdsClient = new XdsClientImpl(
148-
DEFAULT_XDS_TRANSPORT_FACTORY,
149-
bootstrapInfo,
150-
scheduler,
151-
new ExponentialBackoffPolicy.Provider(),
152-
GrpcUtil.STOPWATCH_SUPPLIER,
153-
TimeProvider.SYSTEM_TIME_PROVIDER,
154-
MessagePrinter.INSTANCE,
155-
new TlsContextManagerImpl(bootstrapInfo));
149+
try {
150+
xdsClient = new XdsClientImpl(
151+
DEFAULT_XDS_TRANSPORT_FACTORY,
152+
bootstrapInfo,
153+
scheduler,
154+
BACKOFF_POLICY_PROVIDER,
155+
GrpcUtil.STOPWATCH_SUPPLIER,
156+
TimeProvider.SYSTEM_TIME_PROVIDER,
157+
MessagePrinter.INSTANCE,
158+
new TlsContextManagerImpl(bootstrapInfo));
159+
} catch (Exception e) {
160+
throw new RuntimeException(e);
161+
}
156162
}
157163
refCount++;
158164
return xdsClient;
@@ -179,5 +185,10 @@ XdsClient getXdsClientForTest() {
179185
return xdsClient;
180186
}
181187
}
188+
189+
public String getTarget() {
190+
return target;
191+
}
182192
}
193+
183194
}

‎xds/src/main/java/io/grpc/xds/XdsListenerResource.java

+5
Original file line numberDiff line numberDiff line change
@@ -114,6 +114,11 @@ protected LdsUpdate doParse(Args args, Message unpackedMessage)
114114
}
115115
}
116116

117+
@Override
118+
public boolean isSharedName() {
119+
return true;
120+
}
121+
117122
private LdsUpdate processClientSideListener(Listener listener)
118123
throws ResourceInvalidException {
119124
// Unpack HttpConnectionManager from the Listener.

‎xds/src/main/java/io/grpc/xds/XdsNameResolver.java

+3-3
Original file line numberDiff line numberDiff line change
@@ -139,9 +139,9 @@ final class XdsNameResolver extends NameResolver {
139139
ServiceConfigParser serviceConfigParser,
140140
SynchronizationContext syncContext, ScheduledExecutorService scheduler,
141141
@Nullable Map<String, ?> bootstrapOverride) {
142-
this(targetUri, targetUri.getAuthority(), name, overrideAuthority, serviceConfigParser, syncContext, scheduler,
143-
SharedXdsClientPoolProvider.getDefaultProvider(), ThreadSafeRandomImpl.instance,
144-
FilterRegistry.getDefaultRegistry(), bootstrapOverride);
142+
this(targetUri, targetUri.getAuthority(), name, overrideAuthority, serviceConfigParser,
143+
syncContext, scheduler, SharedXdsClientPoolProvider.getDefaultProvider(),
144+
ThreadSafeRandomImpl.instance, FilterRegistry.getDefaultRegistry(), bootstrapOverride);
145145
}
146146

147147
@VisibleForTesting

‎xds/src/main/java/io/grpc/xds/XdsServerWrapper.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -171,7 +171,7 @@ public void run() {
171171

172172
private void internalStart() {
173173
try {
174-
xdsClientPool = xdsClientPoolFactory.getOrCreate();
174+
xdsClientPool = xdsClientPoolFactory.getOrCreate("");
175175
} catch (Exception e) {
176176
StatusException statusException = Status.UNAVAILABLE.withDescription(
177177
"Failed to initialize xDS").withCause(e).asException();

‎xds/src/main/java/io/grpc/xds/client/BootstrapperImpl.java

+16
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,9 @@
4141
@Internal
4242
public abstract class BootstrapperImpl extends Bootstrapper {
4343

44+
public static final String GRPC_EXPERIMENTAL_XDS_FALLBACK =
45+
"GRPC_EXPERIMENTAL_XDS_FALLBACK";
46+
4447
// Client features.
4548
@VisibleForTesting
4649
public static final String CLIENT_FEATURE_DISABLE_OVERPROVISIONING =
@@ -59,11 +62,18 @@ protected BootstrapperImpl() {
5962
logger = XdsLogger.withLogId(InternalLogId.allocate("bootstrapper", null));
6063
}
6164

65+
// Delayed initialization of xdsFallbackEnabled to allow for flag initialization.
66+
public static boolean isEnabledXdsFallback() {
67+
return GrpcUtil.getFlag(GRPC_EXPERIMENTAL_XDS_FALLBACK, false);
68+
}
69+
6270
protected abstract String getJsonContent() throws IOException, XdsInitializationException;
6371

6472
protected abstract Object getImplSpecificConfig(Map<String, ?> serverConfig, String serverUri)
6573
throws XdsInitializationException;
6674

75+
76+
6777
/**
6878
* Reads and parses bootstrap config. The config is expected to be in JSON format.
6979
*/
@@ -102,6 +112,9 @@ protected BootstrapInfo.Builder bootstrapBuilder(Map<String, ?> rawData)
102112
throw new XdsInitializationException("Invalid bootstrap: 'xds_servers' does not exist.");
103113
}
104114
List<ServerInfo> servers = parseServerInfos(rawServerConfigs, logger);
115+
if (servers.size() > 1 && !isEnabledXdsFallback()) {
116+
servers = ImmutableList.of(servers.get(0));
117+
}
105118
builder.servers(servers);
106119

107120
Node.Builder nodeBuilder = Node.newBuilder();
@@ -208,6 +221,9 @@ protected BootstrapInfo.Builder bootstrapBuilder(Map<String, ?> rawData)
208221
if (rawAuthorityServers == null || rawAuthorityServers.isEmpty()) {
209222
authorityServers = servers;
210223
} else {
224+
if (rawAuthorityServers.size() > 1 && !isEnabledXdsFallback()) {
225+
rawAuthorityServers = ImmutableList.of(rawAuthorityServers.get(0));
226+
}
211227
authorityServers = parseServerInfos(rawAuthorityServers, logger);
212228
}
213229
authorityInfoMapBuilder.put(

0 commit comments

Comments
 (0)