Skip to content

Commit cd2963c

Browse files
kannanjgithubkannanjgithub
committed
xds: align external processor interceptor with new fail-open body specification
- Update ExternalProcessorClientInterceptor to only allow fail-open continuation when no request or response body messages have been sent to the ext_proc stream. - Fail the data plane RPC with INTERNAL status by default when the ext_proc stream fails with a non-OK status. - Add bodyMessageSentToExtProc flag to trace client/server body messages sent to ext_proc. - Rename and update givenFailureModeAllowTrue_whenExtProcStreamFailsAfterRequestBodySent_thenCallFails. - Add new test givenFailureModeAllowTrue_whenExtProcStreamFailsAfterResponseBodySent_thenCallFails. - Update related test assertions to expect INTERNAL status on stream failures.
1 parent d173b6e commit cd2963c

2 files changed

Lines changed: 148 additions & 14 deletions

File tree

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

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -457,6 +457,7 @@ private enum EventType {
457457
final AtomicBoolean requestSideClosed = new AtomicBoolean(false);
458458
final AtomicBoolean isProcessingTrailers = new AtomicBoolean(false);
459459
final AtomicBoolean pendingHalfClose = new AtomicBoolean(false);
460+
final AtomicBoolean bodyMessageSentToExtProc = new AtomicBoolean(false);
460461

461462
protected DataPlaneClientCall(
462463
DataPlaneDelayedCall<InputStream, InputStream> delayedCall,
@@ -784,7 +785,7 @@ public void onError(Throwable t) {
784785
synchronized (streamLock) {
785786
extProcClientCallRequestObserver = null;
786787
}
787-
if (config.getFailureModeAllow()) {
788+
if (config.getFailureModeAllow() && !bodyMessageSentToExtProc.get()) {
788789
handleFailOpen(wrappedListener);
789790
} else {
790791
String message = "External processor stream failed";
@@ -905,7 +906,7 @@ private void internalOnError(Throwable t) {
905906
extProcClientCallRequestObserver = null;
906907
}
907908
}
908-
if (config.getFailureModeAllow()) {
909+
if (config.getFailureModeAllow() && !bodyMessageSentToExtProc.get()) {
909910
handleFailOpen(wrappedListener);
910911
} else {
911912
String message = "External processor stream failed";
@@ -1011,6 +1012,7 @@ public void sendMessage(InputStream message) {
10111012
.setEndOfStream(false)
10121013
.build())
10131014
.build());
1015+
bodyMessageSentToExtProc.set(true);
10141016

10151017
if (config.getObservabilityMode()) {
10161018
super.sendMessage(new OutboundZeroCopyInputStream(bodyByteString));
@@ -1321,6 +1323,7 @@ public void onMessage(InputStream message) {
13211323
try {
13221324
ByteString bodyByteString = inboundStreamToByteString(message);
13231325
sendResponseBodyToExtProc(bodyByteString, false);
1326+
dataPlaneClientCall.bodyMessageSentToExtProc.set(true);
13241327

13251328
if (dataPlaneClientCall.getConfig().getObservabilityMode()) {
13261329
dataPlaneClientCall.getCallContext().run(
@@ -1337,9 +1340,10 @@ public void onClose(Status status, Metadata trailers) {
13371340
DataPlaneClientCall.ExtProcStreamState extProcStreamState =
13381341
dataPlaneClientCall.getExtProcStreamState().get();
13391342
if (extProcStreamState.isFailed()
1340-
&& !dataPlaneClientCall.getConfig().getFailureModeAllow()) {
1343+
&& (!dataPlaneClientCall.getConfig().getFailureModeAllow()
1344+
|| dataPlaneClientCall.bodyMessageSentToExtProc.get())) {
13411345
if (dataPlaneClientCall.markDataPlaneCallClosed()) {
1342-
proceedWithClose(Status.UNAVAILABLE.withDescription("External processor stream failed")
1346+
proceedWithClose(Status.INTERNAL.withDescription("External processor stream failed")
13431347
.withCause(status.getCause()), new Metadata());
13441348
}
13451349
return;

‎xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java‎

Lines changed: 140 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -7748,9 +7748,9 @@ public void onClose(Status status, Metadata trailers) {
77487748
interceptCall(interceptor, METHOD_SAY_HELLO, callOptions, dataPlaneChannel);
77497749
proxyCall.start(appListener, new Metadata());
77507750

7751-
// Verify application receives UNAVAILABLE due to sidecar failure
7751+
// Verify application receives INTERNAL due to sidecar failure
77527752
assertThat(closedLatch.await(5, TimeUnit.SECONDS)).isTrue();
7753-
assertThat(closedStatus.get().getCode()).isEqualTo(Status.Code.UNAVAILABLE);
7753+
assertThat(closedStatus.get().getCode()).isEqualTo(Status.Code.INTERNAL);
77547754
assertThat(closedStatus.get().getDescription()).contains("External processor stream failed");
77557755

77567756
proxyCall.cancel("Cleanup", null);
@@ -8126,12 +8126,12 @@ public void onClose(Status status, Metadata trailers) {
81268126
proxyCall.sendMessage("test");
81278127
proxyCall.halfClose();
81288128

8129-
// Verify application receives UNAVAILABLE with correct description
8129+
// Verify application receives INTERNAL with correct description
81308130
for (int i = 0; i < 10000 && closedLatch.getCount() > 0; i++) {
81318131
fakeClock.forwardTime(1, TimeUnit.MILLISECONDS);
81328132
}
81338133
assertThat(closedLatch.await(5, TimeUnit.SECONDS)).isTrue();
8134-
assertThat(closedStatus.get().getCode()).isEqualTo(Status.Code.UNAVAILABLE);
8134+
assertThat(closedStatus.get().getCode()).isEqualTo(Status.Code.INTERNAL);
81358135
assertThat(closedStatus.get().getDescription()).contains("External processor stream failed");
81368136

81378137
proxyCall.cancel("Cleanup", null);
@@ -8271,9 +8271,9 @@ public void onCompleted() {
82718271
proxyCall.sendMessage("test");
82728272
proxyCall.halfClose();
82738273

8274-
// Verify application receives UNAVAILABLE with correct description
8274+
// Verify application receives INTERNAL with correct description
82758275
assertThat(closedLatch.await(5, TimeUnit.SECONDS)).isTrue();
8276-
assertThat(closedStatus.get().getCode()).isEqualTo(Status.Code.UNAVAILABLE);
8276+
assertThat(closedStatus.get().getCode()).isEqualTo(Status.Code.INTERNAL);
82778277
assertThat(closedStatus.get().getDescription()).contains("External processor stream failed");
82788278

82798279
proxyCall.cancel("Cleanup", null);
@@ -10623,9 +10623,9 @@ public void onCompleted() {
1062310623
assertThat(sidecarLatch.await(5, TimeUnit.SECONDS)).isTrue();
1062410624
assertThat(appCloseLatch.await(5, TimeUnit.SECONDS)).isTrue();
1062510625

10626-
// The call should fail with UNAVAILABLE status
10626+
// The call should fail with INTERNAL status
1062710627
// due to stream failure triggered by protocol error
10628-
assertThat(appStatus.get().getCode()).isEqualTo(Status.Code.UNAVAILABLE);
10628+
assertThat(appStatus.get().getCode()).isEqualTo(Status.Code.INTERNAL);
1062910629
assertThat(appStatus.get().getDescription()).contains("External processor stream failed");
1063010630

1063110631
channelManager.close();
@@ -11372,7 +11372,7 @@ public void onClose(Status status, Metadata trailers) {
1137211372
}
1137311373

1137411374
@Test
11375-
public void givenFailureModeAllowTrue_whenExtProcStreamFailsDuringBodyOrEos_thenCallUnblocked()
11375+
public void givenFailureModeAllowTrue_whenExtProcStreamFailsAfterRequestBodySent_thenCallFails()
1137611376
throws Exception {
1137711377
String uniqueExtProcServerName = InProcessServerBuilder.generateName();
1137811378
String uniqueDataPlaneServerName = InProcessServerBuilder.generateName();
@@ -11455,10 +11455,12 @@ public void onCompleted() {}
1145511455
DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()),
1145611456
dataPlaneChannel);
1145711457

11458+
final AtomicReference<Status> closedStatus = new AtomicReference<>();
1145811459
final CountDownLatch callCompletedLatch = new CountDownLatch(1);
1145911460
clientCall.start(new ClientCall.Listener<String>() {
1146011461
@Override
1146111462
public void onClose(Status status, Metadata trailers) {
11463+
closedStatus.set(status);
1146211464
callCompletedLatch.countDown();
1146311465
}
1146411466
}, new Metadata());
@@ -11474,9 +11476,137 @@ public void onClose(Status status, Metadata trailers) {
1147411476
responseObserverRef.get()
1147511477
.onError(new RuntimeException("Stream failure after sending body/EOS"));
1147611478

11477-
// Verify that the call is unblocked (onClose is called) instead of hanging.
11479+
// Verify that the call failed with INTERNAL status instead of succeeding.
11480+
boolean completed = callCompletedLatch.await(5, TimeUnit.SECONDS);
11481+
assertThat(completed).isTrue();
11482+
assertThat(closedStatus.get().getCode()).isEqualTo(Status.Code.INTERNAL);
11483+
assertThat(closedStatus.get().getDescription()).contains("External processor stream failed");
11484+
11485+
channelManager.close();
11486+
}
11487+
11488+
@Test
11489+
public void givenFailureModeAllowTrue_whenExtProcStreamFailsAfterResponseBodySent_thenCallFails()
11490+
throws Exception {
11491+
String uniqueExtProcServerName = InProcessServerBuilder.generateName();
11492+
String uniqueDataPlaneServerName = InProcessServerBuilder.generateName();
11493+
11494+
ExternalProcessor proto = createBaseProto(uniqueExtProcServerName)
11495+
.setFailureModeAllow(true)
11496+
.setProcessingMode(ProcessingMode.newBuilder()
11497+
.setRequestHeaderMode(ProcessingMode.HeaderSendMode.SEND)
11498+
.setRequestBodyMode(ProcessingMode.BodySendMode.NONE)
11499+
.setResponseHeaderMode(ProcessingMode.HeaderSendMode.SEND)
11500+
.setResponseBodyMode(ProcessingMode.BodySendMode.GRPC)
11501+
.setResponseTrailerMode(ProcessingMode.HeaderSendMode.SEND)
11502+
.build())
11503+
.build();
11504+
ConfigOrError<ExternalProcessorFilterConfig> configOrError =
11505+
provider.parseFilterConfig(Any.pack(proto), filterContext);
11506+
assertThat(configOrError.errorDetail).isNull();
11507+
ExternalProcessorFilterConfig filterConfig = configOrError.config;
11508+
11509+
final AtomicReference<StreamObserver<ProcessingResponse>> responseObserverRef =
11510+
new AtomicReference<>();
11511+
final CountDownLatch streamActiveLatch = new CountDownLatch(1);
11512+
final CountDownLatch streamFailedLatch = new CountDownLatch(1);
11513+
11514+
ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl =
11515+
new ExternalProcessorGrpc.ExternalProcessorImplBase() {
11516+
@Override
11517+
public StreamObserver<ProcessingRequest> process(
11518+
StreamObserver<ProcessingResponse> responseObserver) {
11519+
responseObserverRef.set(responseObserver);
11520+
streamActiveLatch.countDown();
11521+
return new StreamObserver<ProcessingRequest>() {
11522+
@Override
11523+
public void onNext(ProcessingRequest request) {
11524+
if (request.hasRequestHeaders()) {
11525+
responseObserver.onNext(ProcessingResponse.newBuilder()
11526+
.setRequestHeaders(HeadersResponse.newBuilder().build())
11527+
.build());
11528+
} else if (request.hasResponseHeaders()) {
11529+
responseObserver.onNext(ProcessingResponse.newBuilder()
11530+
.setResponseHeaders(HeadersResponse.newBuilder().build())
11531+
.build());
11532+
} else if (request.hasResponseBody()) {
11533+
// Fail the stream once we see response body message
11534+
responseObserver.onError(
11535+
Status.INTERNAL.withDescription("Stream failure after response body")
11536+
.asRuntimeException());
11537+
streamFailedLatch.countDown();
11538+
}
11539+
}
11540+
11541+
@Override
11542+
public void onError(Throwable t) {}
11543+
11544+
@Override
11545+
public void onCompleted() {}
11546+
};
11547+
}
11548+
};
11549+
11550+
grpcCleanup.register(InProcessServerBuilder.forName(uniqueExtProcServerName)
11551+
.addService(extProcImpl)
11552+
.directExecutor()
11553+
.build().start());
11554+
11555+
CachedChannelManager channelManager = new CachedChannelManager(config -> {
11556+
return grpcCleanup.register(
11557+
InProcessChannelBuilder.forName(uniqueExtProcServerName)
11558+
.directExecutor()
11559+
.build());
11560+
});
11561+
11562+
ExternalProcessorClientInterceptor interceptor = new ExternalProcessorClientInterceptor(
11563+
filterConfig, channelManager, scheduler, FAKE_CONTEXT);
11564+
11565+
MutableHandlerRegistry dataPlaneRegistry = new MutableHandlerRegistry();
11566+
grpcCleanup.register(InProcessServerBuilder.forName(uniqueDataPlaneServerName)
11567+
.fallbackHandlerRegistry(dataPlaneRegistry)
11568+
.directExecutor()
11569+
.build().start());
11570+
11571+
dataPlaneRegistry.addService(ServerServiceDefinition.builder("test.TestService")
11572+
.addMethod(METHOD_SAY_HELLO, ServerCalls.asyncUnaryCall((request, responseObserver) -> {
11573+
responseObserver.onNext("response-body-msg");
11574+
responseObserver.onCompleted();
11575+
})).build());
11576+
11577+
ManagedChannel dataPlaneChannel = grpcCleanup.register(
11578+
InProcessChannelBuilder.forName(uniqueDataPlaneServerName)
11579+
.directExecutor()
11580+
.build());
11581+
11582+
ClientCall<String, String> clientCall = interceptCall(
11583+
interceptor, METHOD_SAY_HELLO,
11584+
DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()),
11585+
dataPlaneChannel);
11586+
11587+
final AtomicReference<Status> closedStatus = new AtomicReference<>();
11588+
final CountDownLatch callCompletedLatch = new CountDownLatch(1);
11589+
clientCall.start(new ClientCall.Listener<String>() {
11590+
@Override
11591+
public void onClose(Status status, Metadata trailers) {
11592+
closedStatus.set(status);
11593+
callCompletedLatch.countDown();
11594+
}
11595+
}, new Metadata());
11596+
clientCall.request(1);
11597+
11598+
boolean active = streamActiveLatch.await(5, TimeUnit.SECONDS);
11599+
assertThat(active).isTrue();
11600+
11601+
// Since request body mode is NONE, this sendMessage is NOT sent to ext_proc
11602+
clientCall.sendMessage("app-msg");
11603+
clientCall.halfClose();
11604+
11605+
// Verify call completed and failed with INTERNAL status
1147811606
boolean completed = callCompletedLatch.await(5, TimeUnit.SECONDS);
1147911607
assertThat(completed).isTrue();
11608+
assertThat(closedStatus.get().getCode()).isEqualTo(Status.Code.INTERNAL);
11609+
assertThat(closedStatus.get().getDescription()).contains("External processor stream failed");
1148011610

1148111611
channelManager.close();
1148211612
}

0 commit comments

Comments
 (0)