Skip to content

Commit 1f25ba6

Browse files
kannanjgithubkannanjgithub
committed
Merge remote-tracking branch 'origin/master' into ext-proc-client-flow-control
# Conflicts: # xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java # xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java
2 parents a16d5de + 7fdcde1 commit 1f25ba6

7 files changed

Lines changed: 718 additions & 69 deletions

File tree

‎compiler/src/java_plugin/cpp/java_plugin.cpp‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -58,7 +58,7 @@ class JavaGrpcGenerator : public protobuf::compiler::CodeGenerator {
5858
return protobuf::Edition::EDITION_PROTO2;
5959
}
6060
protobuf::Edition GetMaximumEdition() const override {
61-
#if GOOGLE_PROTOBUF_VERSION >= 6036000
61+
#if GOOGLE_PROTOBUF_VERSION >= 7035000
6262
return protobuf::Edition::EDITION_2026;
6363
#elif GOOGLE_PROTOBUF_VERSION >= 6032000
6464
return protobuf::Edition::EDITION_2024;

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

Lines changed: 21 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -185,15 +185,14 @@ static synchronized void initMetricInstruments() {
185185
ExternalProcessorFilterConfig getFilterConfig() {
186186
return filterConfig;
187187
}
188-
188+
189189
@Override
190190
@SuppressWarnings("unchecked")
191191
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
192192
MethodDescriptor<ReqT, RespT> method,
193193
CallOptions callOptions,
194194
Channel next) {
195-
Executor callExecutor = callOptions.getExecutor();
196-
SerializingExecutor serializingExecutor = new SerializingExecutor(callExecutor);
195+
SerializingExecutor serializingExecutor = new SerializingExecutor(callOptions.getExecutor());
197196

198197
ExternalProcessorGrpc.ExternalProcessorStub extProcStub = ExternalProcessorGrpc.newStub(
199198
extProcChannel)
@@ -791,7 +790,7 @@ private void onExtProcStreamReady() {
791790

792791
void drainPendingRequests() {
793792
synchronized (streamLock) {
794-
if (config.getObservabilityMode()
793+
if (config.getObservabilityMode()
795794
|| currentProcessingMode.getResponseBodyMode() != ProcessingMode.BodySendMode.GRPC) {
796795
int toRequest = pendingRequests.getAndSet(0);
797796
if (toRequest > 0) {
@@ -894,6 +893,11 @@ public void request(int numMessages) {
894893
super.request(numMessages);
895894
return;
896895
}
896+
if (!config.getObservabilityMode()
897+
&& currentProcessingMode.getResponseBodyMode() != ProcessingMode.BodySendMode.GRPC) {
898+
super.request(numMessages);
899+
return;
900+
}
897901
synchronized (streamLock) {
898902
boolean sendResponseBodiesToExtProc = config.getObservabilityMode()
899903
|| currentProcessingMode.getResponseBodyMode() == ProcessingMode.BodySendMode.GRPC;
@@ -1043,6 +1047,18 @@ public void halfClose() {
10431047
}
10441048

10451049
if (extProcStreamState.get().isDraining()) {
1050+
boolean canProceed = false;
1051+
synchronized (streamLock) {
1052+
if (currentProcessingMode.getRequestBodyMode() == ProcessingMode.BodySendMode.NONE
1053+
|| (!bodyMessageSentToExtProc.get() && pendingDrainingMessages.isEmpty())) {
1054+
canProceed = true;
1055+
}
1056+
}
1057+
if (canProceed) {
1058+
if (requestSideClosed.compareAndSet(false, true)) {
1059+
proceedWithHalfClose();
1060+
}
1061+
}
10461062
return;
10471063
}
10481064

@@ -1392,7 +1408,7 @@ public void onHeaders(Metadata headers) {
13921408
return;
13931409
}
13941410

1395-
if (dataPlaneClientCall.getPassThroughMode().get()
1411+
if (dataPlaneClientCall.getPassThroughMode().get()
13961412
|| dataPlaneClientCall.getExtProcStreamState().get().isCompleted()
13971413
|| !sendResponseHeaders) {
13981414
proceedWithHeaders(headers);

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

Lines changed: 43 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -37,10 +37,14 @@
3737
import io.grpc.netty.InternalProtocolNegotiator;
3838
import io.grpc.netty.NettyServerBuilder;
3939
import io.grpc.xds.FilterChainMatchingProtocolNegotiators.FilterChainMatchingNegotiatorServerFactory;
40+
import java.net.InetSocketAddress;
41+
import java.net.SocketAddress;
4042
import java.util.Map;
4143
import java.util.concurrent.TimeUnit;
4244
import java.util.concurrent.atomic.AtomicBoolean;
45+
import java.util.function.Function;
4346
import java.util.logging.Logger;
47+
import javax.annotation.Nullable;
4448

4549
/**
4650
* A version of {@link ServerBuilder} to create xDS managed servers.
@@ -57,6 +61,7 @@ public final class XdsServerBuilder extends ForwardingServerBuilder<XdsServerBui
5761
private XdsClientPoolFactory xdsClientPoolFactory =
5862
SharedXdsClientPoolProvider.getDefaultProvider();
5963
private Map<String, ?> bootstrapOverride;
64+
@Nullable private Function<String, String> ldsResourceNameResolver;
6065
private long drainGraceTime = 10;
6166
private TimeUnit drainGraceTimeUnit = TimeUnit.MINUTES;
6267
private ChannelConfigurator channelConfigurator = builder -> { };
@@ -134,6 +139,25 @@ public static XdsServerBuilder forPort(int port, ServerCredentials serverCredent
134139
return new XdsServerBuilder(nettyDelegate, port);
135140
}
136141

142+
/** Creates a gRPC server builder for the given address. */
143+
public static XdsServerBuilder forAddress(
144+
SocketAddress address, ServerCredentials serverCredentials) {
145+
checkNotNull(address, "address");
146+
checkNotNull(serverCredentials, "serverCredentials");
147+
InternalProtocolNegotiator.ServerFactory originalNegotiatorFactory =
148+
InternalNettyServerCredentials.toNegotiator(serverCredentials);
149+
ServerCredentials wrappedCredentials =
150+
InternalNettyServerCredentials.create(
151+
new FilterChainMatchingNegotiatorServerFactory(originalNegotiatorFactory));
152+
NettyServerBuilder nettyDelegate = NettyServerBuilder.forAddress(address, wrappedCredentials);
153+
int port = 0;
154+
if (address instanceof InetSocketAddress) {
155+
InetSocketAddress inetSocketAddress = (InetSocketAddress) address;
156+
port = inetSocketAddress.getPort();
157+
}
158+
return new XdsServerBuilder(nettyDelegate, port);
159+
}
160+
137161
@Override
138162
public Server build() {
139163
checkState(isServerBuilt.compareAndSet(false, true), "Server already built!");
@@ -144,11 +168,28 @@ public Server build() {
144168
builder.set(ATTR_DRAIN_GRACE_NANOS, drainGraceTimeUnit.toNanos(drainGraceTime));
145169
}
146170
InternalNettyServerBuilder.eagAttributes(delegate, builder.build());
147-
return new XdsServerWrapper("0.0.0.0:" + port, delegate, xdsServingStatusListener,
148-
filterChainSelectorManager, xdsClientPoolFactory, bootstrapOverride, filterRegistry,
171+
return new XdsServerWrapper(
172+
"0.0.0.0:" + port,
173+
delegate,
174+
xdsServingStatusListener,
175+
filterChainSelectorManager,
176+
xdsClientPoolFactory,
177+
bootstrapOverride,
178+
ldsResourceNameResolver,
179+
filterRegistry,
149180
this.channelConfigurator);
150181
}
151182

183+
/**
184+
* Provides a function that takes the listening address and returns the LDS resource name. When
185+
* provided, this overrides the server_listener_resource_name_template in the bootstrap.
186+
*/
187+
public XdsServerBuilder ldsResourceNameResolver(
188+
Function<String, String> ldsResourceNameResolver) {
189+
this.ldsResourceNameResolver = checkNotNull(ldsResourceNameResolver, "ldsResourceNameResolver");
190+
return this;
191+
}
192+
152193
@VisibleForTesting
153194
XdsServerBuilder xdsClientPoolFactory(XdsClientPoolFactory xdsClientPoolFactory) {
154195
this.xdsClientPoolFactory = checkNotNull(xdsClientPoolFactory, "xdsClientPoolFactory");

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

Lines changed: 125 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -78,6 +78,7 @@
7878
import java.util.concurrent.TimeUnit;
7979
import java.util.concurrent.atomic.AtomicBoolean;
8080
import java.util.concurrent.atomic.AtomicReference;
81+
import java.util.function.Function;
8182
import java.util.logging.Level;
8283
import java.util.logging.Logger;
8384
import javax.annotation.Nullable;
@@ -108,6 +109,7 @@ public void uncaughtException(Thread t, Throwable e) {
108109
private final ThreadSafeRandom random = ThreadSafeRandomImpl.instance;
109110
private final XdsClientPoolFactory xdsClientPoolFactory;
110111
private final @Nullable Map<String, ?> bootstrapOverride;
112+
private final @Nullable Function<String, String> ldsResourceNameResolver;
111113
private final XdsServingStatusListener listener;
112114
private final FilterChainSelectorManager filterChainSelectorManager;
113115
private final AtomicBoolean started = new AtomicBoolean(false);
@@ -139,6 +141,7 @@ public void uncaughtException(Thread t, Throwable e) {
139141
FilterChainSelectorManager filterChainSelectorManager,
140142
XdsClientPoolFactory xdsClientPoolFactory,
141143
@Nullable Map<String, ?> bootstrapOverride,
144+
@Nullable Function<String, String> ldsResourceNameResolver,
142145
FilterRegistry filterRegistry,
143146
ChannelConfigurator channelConfigurator) {
144147
this(
@@ -148,6 +151,7 @@ public void uncaughtException(Thread t, Throwable e) {
148151
filterChainSelectorManager,
149152
xdsClientPoolFactory,
150153
bootstrapOverride,
154+
ldsResourceNameResolver,
151155
filterRegistry,
152156
SharedResourceHolder.get(GrpcUtil.TIMER_SERVICE),
153157
channelConfigurator);
@@ -161,6 +165,52 @@ public void uncaughtException(Thread t, Throwable e) {
161165
FilterChainSelectorManager filterChainSelectorManager,
162166
XdsClientPoolFactory xdsClientPoolFactory,
163167
@Nullable Map<String, ?> bootstrapOverride,
168+
FilterRegistry filterRegistry,
169+
ChannelConfigurator channelConfigurator) {
170+
this(
171+
listenerAddress,
172+
delegateBuilder,
173+
listener,
174+
filterChainSelectorManager,
175+
xdsClientPoolFactory,
176+
bootstrapOverride,
177+
null,
178+
filterRegistry,
179+
SharedResourceHolder.get(GrpcUtil.TIMER_SERVICE),
180+
channelConfigurator);
181+
sharedTimeService = true;
182+
}
183+
184+
XdsServerWrapper(
185+
String listenerAddress,
186+
ServerBuilder<?> delegateBuilder,
187+
XdsServingStatusListener listener,
188+
FilterChainSelectorManager filterChainSelectorManager,
189+
XdsClientPoolFactory xdsClientPoolFactory,
190+
@Nullable Map<String, ?> bootstrapOverride,
191+
FilterRegistry filterRegistry) {
192+
this(
193+
listenerAddress,
194+
delegateBuilder,
195+
listener,
196+
filterChainSelectorManager,
197+
xdsClientPoolFactory,
198+
bootstrapOverride,
199+
null,
200+
filterRegistry,
201+
SharedResourceHolder.get(GrpcUtil.TIMER_SERVICE),
202+
builder -> { });
203+
sharedTimeService = true;
204+
}
205+
206+
XdsServerWrapper(
207+
String listenerAddress,
208+
ServerBuilder<?> delegateBuilder,
209+
XdsServingStatusListener listener,
210+
FilterChainSelectorManager filterChainSelectorManager,
211+
XdsClientPoolFactory xdsClientPoolFactory,
212+
@Nullable Map<String, ?> bootstrapOverride,
213+
@Nullable Function<String, String> ldsResourceNameResolver,
164214
FilterRegistry filterRegistry) {
165215
this(
166216
listenerAddress,
@@ -169,8 +219,11 @@ public void uncaughtException(Thread t, Throwable e) {
169219
filterChainSelectorManager,
170220
xdsClientPoolFactory,
171221
bootstrapOverride,
222+
ldsResourceNameResolver,
172223
filterRegistry,
224+
SharedResourceHolder.get(GrpcUtil.TIMER_SERVICE),
173225
builder -> { });
226+
sharedTimeService = true;
174227
}
175228

176229
@VisibleForTesting
@@ -190,6 +243,31 @@ public void uncaughtException(Thread t, Throwable e) {
190243
filterChainSelectorManager,
191244
xdsClientPoolFactory,
192245
bootstrapOverride,
246+
null,
247+
filterRegistry,
248+
timeService,
249+
builder -> { });
250+
}
251+
252+
@VisibleForTesting
253+
XdsServerWrapper(
254+
String listenerAddress,
255+
ServerBuilder<?> delegateBuilder,
256+
XdsServingStatusListener listener,
257+
FilterChainSelectorManager filterChainSelectorManager,
258+
XdsClientPoolFactory xdsClientPoolFactory,
259+
@Nullable Map<String, ?> bootstrapOverride,
260+
@Nullable Function<String, String> ldsResourceNameResolver,
261+
FilterRegistry filterRegistry,
262+
ScheduledExecutorService timeService) {
263+
this(
264+
listenerAddress,
265+
delegateBuilder,
266+
listener,
267+
filterChainSelectorManager,
268+
xdsClientPoolFactory,
269+
bootstrapOverride,
270+
ldsResourceNameResolver,
193271
filterRegistry,
194272
timeService,
195273
builder -> { });
@@ -206,6 +284,31 @@ public void uncaughtException(Thread t, Throwable e) {
206284
FilterRegistry filterRegistry,
207285
ScheduledExecutorService timeService,
208286
ChannelConfigurator channelConfigurator) {
287+
this(
288+
listenerAddress,
289+
delegateBuilder,
290+
listener,
291+
filterChainSelectorManager,
292+
xdsClientPoolFactory,
293+
bootstrapOverride,
294+
null,
295+
filterRegistry,
296+
timeService,
297+
channelConfigurator);
298+
}
299+
300+
@VisibleForTesting
301+
XdsServerWrapper(
302+
String listenerAddress,
303+
ServerBuilder<?> delegateBuilder,
304+
XdsServingStatusListener listener,
305+
FilterChainSelectorManager filterChainSelectorManager,
306+
XdsClientPoolFactory xdsClientPoolFactory,
307+
@Nullable Map<String, ?> bootstrapOverride,
308+
@Nullable Function<String, String> ldsResourceNameResolver,
309+
FilterRegistry filterRegistry,
310+
ScheduledExecutorService timeService,
311+
ChannelConfigurator channelConfigurator) {
209312
this.listenerAddress = checkNotNull(listenerAddress, "listenerAddress");
210313
this.delegateBuilder = checkNotNull(delegateBuilder, "delegateBuilder");
211314
this.delegateBuilder.intercept(new ConfigApplyingInterceptor());
@@ -214,6 +317,7 @@ public void uncaughtException(Thread t, Throwable e) {
214317
= checkNotNull(filterChainSelectorManager, "filterChainSelectorManager");
215318
this.xdsClientPoolFactory = checkNotNull(xdsClientPoolFactory, "xdsClientPoolFactory");
216319
this.bootstrapOverride = bootstrapOverride;
320+
this.ldsResourceNameResolver = ldsResourceNameResolver;
217321
this.timeService = checkNotNull(timeService, "timeService");
218322
this.filterRegistry = checkNotNull(filterRegistry,"filterRegistry");
219323
this.delegate = delegateBuilder.build();
@@ -261,21 +365,28 @@ private void internalStart() {
261365
return;
262366
}
263367
xdsClient = xdsClientPool.getObject();
264-
String listenerTemplate = xdsClient.getBootstrapInfo().serverListenerResourceNameTemplate();
265-
if (listenerTemplate == null) {
266-
StatusException statusException =
267-
Status.UNAVAILABLE.withDescription(
268-
"Can only support xDS v3 with listener resource name template").asException();
269-
listener.onNotServing(statusException);
270-
initialStartFuture.set(statusException);
271-
xdsClient = xdsClientPool.returnObject(xdsClient);
272-
return;
273-
}
274-
String replacement = listenerAddress;
275-
if (listenerTemplate.startsWith(XDSTP_SCHEME)) {
276-
replacement = XdsClient.percentEncodePath(replacement);
368+
String resourceName;
369+
if (ldsResourceNameResolver != null) {
370+
resourceName = ldsResourceNameResolver.apply(listenerAddress);
371+
} else {
372+
String listenerTemplate = xdsClient.getBootstrapInfo().serverListenerResourceNameTemplate();
373+
if (listenerTemplate == null) {
374+
StatusException statusException =
375+
Status.UNAVAILABLE
376+
.withDescription("Can only support xDS v3 with listener resource name template")
377+
.asException();
378+
listener.onNotServing(statusException);
379+
initialStartFuture.set(statusException);
380+
xdsClient = xdsClientPool.returnObject(xdsClient);
381+
return;
382+
}
383+
String replacement = listenerAddress;
384+
if (listenerTemplate.startsWith(XDSTP_SCHEME)) {
385+
replacement = XdsClient.percentEncodePath(replacement);
386+
}
387+
resourceName = listenerTemplate.replaceAll("%s", replacement);
277388
}
278-
discoveryState = new DiscoveryState(listenerTemplate.replaceAll("%s", replacement));
389+
discoveryState = new DiscoveryState(resourceName);
279390
}
280391

281392
@Override

0 commit comments

Comments
 (0)