Skip to content

Commit 68532f5

Browse files
kannanjgithubkannanjgithub
committed
xds: align server processor interceptor with new fail-open body specification
- Update ExternalProcessorServerInterceptor 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. - Add test givenFailureModeAllowTrue_whenExtProcStreamFailsAfterRequestBodySent_thenCallFails under Category 15. - Add test givenFailureModeAllowTrue_whenExtProcStreamFailsAfterResponseBodySent_thenCallFails under Category 15. - Update related test assertions to expect INTERNAL status on stream failures.
1 parent 35b4c32 commit 68532f5

2 files changed

Lines changed: 221 additions & 11 deletions

File tree

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

Lines changed: 12 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -336,6 +336,7 @@ private enum EventType {
336336
final AtomicBoolean halfClosed = new AtomicBoolean(false);
337337
final AtomicBoolean requestSideClosed = new AtomicBoolean(false);
338338
final AtomicBoolean dataPlaneCallClosed = new AtomicBoolean(false);
339+
final AtomicBoolean bodyMessageSentToExtProc = new AtomicBoolean(false);
339340

340341
final AtomicBoolean isProcessingTrailers = new AtomicBoolean(false);
341342
final AtomicBoolean responseHeadersSent = new AtomicBoolean(false);
@@ -425,7 +426,8 @@ boolean markDataPlaneCallClosed() {
425426
}
426427

427428
private void activateCall() {
428-
if ((extProcStreamState.get() == ExtProcStreamState.FAILED && !config.getFailureModeAllow())
429+
if ((extProcStreamState.get() == ExtProcStreamState.FAILED
430+
&& (!config.getFailureModeAllow() || bodyMessageSentToExtProc.get()))
429431
|| !dataPlaneCallState.compareAndSet(
430432
DataPlaneCallState.IDLE, DataPlaneCallState.ACTIVE)) {
431433
return;
@@ -645,10 +647,10 @@ public void onError(Throwable t) {
645647
synchronized (streamLock) {
646648
extProcClientCallRequestObserver = null;
647649
}
648-
if (config.getFailureModeAllow()) {
650+
if (config.getFailureModeAllow() && !bodyMessageSentToExtProc.get()) {
649651
handleFailOpen();
650652
} else {
651-
rawCall.close(Status.UNAVAILABLE.withDescription("External processor stream failed").withCause(t), new Metadata());
653+
rawCall.close(Status.INTERNAL.withDescription("External processor stream failed").withCause(t), new Metadata());
652654
}
653655
}
654656
}
@@ -783,10 +785,10 @@ private void internalOnError(Throwable t) {
783785
}
784786
}
785787
expectedResponses.clear();
786-
if (config.getFailureModeAllow()) {
788+
if (config.getFailureModeAllow() && !bodyMessageSentToExtProc.get()) {
787789
handleFailOpen();
788790
} else {
789-
rawCall.close(Status.UNAVAILABLE.withDescription("External processor stream failed").withCause(t), new Metadata());
791+
rawCall.close(Status.INTERNAL.withDescription("External processor stream failed").withCause(t), new Metadata());
790792
}
791793
}
792794
}
@@ -961,9 +963,10 @@ public void sendMessage(InputStream message) {
961963
@Override
962964
public void close(Status status, Metadata trailers) {
963965
serverTrailersStartNanos = System.nanoTime();
964-
if (isExtProcStreamFailed() && !config.getFailureModeAllow()) {
966+
if (isExtProcStreamFailed()
967+
&& (!config.getFailureModeAllow() || bodyMessageSentToExtProc.get())) {
965968
if (markDataPlaneCallClosed()) {
966-
proceedWithClose(Status.UNAVAILABLE.withDescription("External processor stream failed").withCause(status.getCause()), new Metadata());
969+
proceedWithClose(Status.INTERNAL.withDescription("External processor stream failed").withCause(status.getCause()), new Metadata());
967970
}
968971
return;
969972
}
@@ -1088,6 +1091,7 @@ private void sendResponseBodyToExtProc(@Nullable ByteString bodyByteString, bool
10881091
HttpBody.Builder bodyBuilder = HttpBody.newBuilder();
10891092
if (bodyByteString != null) {
10901093
bodyBuilder.setBody(bodyByteString);
1094+
bodyMessageSentToExtProc.set(true);
10911095
}
10921096
bodyBuilder.setEndOfStream(endOfStream);
10931097

@@ -1390,6 +1394,7 @@ private void sendRequestBodyToExtProc(@Nullable ByteString bodyByteString, boole
13901394
if (bodyByteString != null) {
13911395
bodyBuilder.setBody(bodyByteString);
13921396
bodyBuilder.setEndOfStream(endOfStream);
1397+
dataPlaneServerCall.bodyMessageSentToExtProc.set(true);
13931398
} else {
13941399
bodyBuilder.setEndOfStreamWithoutMessage(true);
13951400
}

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

Lines changed: 209 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -6717,8 +6717,213 @@ public void onClose(Status status, Metadata trailers) {
67176717
clientCall.sendMessage(new ByteArrayInputStream("hello".getBytes(StandardCharsets.UTF_8)));
67186718
clientCall.halfClose();
67196719

6720+
assertThat(closedStatus.get().getCode()).isEqualTo(Status.Code.INTERNAL);
6721+
assertThat(closedStatus.get().getDescription()).contains("External processor stream failed");
6722+
}
6723+
6724+
@Test
6725+
public void givenFailureModeAllowTrue_whenExtProcStreamFailsAfterRequestBodySent_thenCallFails()
6726+
throws Exception {
6727+
ExternalProcessor proto = createBaseProto(extProcServerName)
6728+
.setFailureModeAllow(true)
6729+
.setProcessingMode(ProcessingMode.newBuilder()
6730+
.setRequestHeaderMode(ProcessingMode.HeaderSendMode.SEND)
6731+
.setRequestBodyMode(ProcessingMode.BodySendMode.GRPC)
6732+
.setResponseHeaderMode(ProcessingMode.HeaderSendMode.SKIP)
6733+
.setResponseBodyMode(ProcessingMode.BodySendMode.NONE)
6734+
.setResponseTrailerMode(ProcessingMode.HeaderSendMode.SKIP)
6735+
.build())
6736+
.build();
6737+
ConfigOrError<ExternalProcessorFilterConfig> configOrError =
6738+
provider.parseFilterConfig(Any.pack(proto), filterContext);
6739+
assertThat(configOrError.errorDetail).isNull();
6740+
ExternalProcessorFilterConfig filterConfig = configOrError.config;
6741+
6742+
final CountDownLatch extProcLatch = new CountDownLatch(1);
6743+
final AtomicReference<StreamObserver<ProcessingResponse>> responseObserverRef =
6744+
new AtomicReference<>();
6745+
ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl =
6746+
new ExternalProcessorGrpc.ExternalProcessorImplBase() {
6747+
@Override
6748+
public StreamObserver<ProcessingRequest> process(
6749+
final StreamObserver<ProcessingResponse> responseObserver) {
6750+
responseObserverRef.set(responseObserver);
6751+
((ServerCallStreamObserver<ProcessingResponse>) responseObserver).request(100);
6752+
return new StreamObserver<ProcessingRequest>() {
6753+
@Override
6754+
public void onNext(ProcessingRequest request) {
6755+
if (request.hasRequestHeaders()) {
6756+
responseObserver.onNext(ProcessingResponse.newBuilder()
6757+
.setRequestHeaders(HeadersResponse.newBuilder().build())
6758+
.build());
6759+
} else if (request.hasRequestBody()) {
6760+
// Fail the stream after receiving the request body
6761+
responseObserver.onError(
6762+
Status.INTERNAL.withDescription("Simulated stream failure").asRuntimeException());
6763+
extProcLatch.countDown();
6764+
}
6765+
}
6766+
6767+
@Override
6768+
public void onError(Throwable t) {}
6769+
6770+
@Override
6771+
public void onCompleted() {}
6772+
};
6773+
}
6774+
};
6775+
6776+
grpcCleanup.register(InProcessServerBuilder.forName(extProcServerName)
6777+
.addService(extProcImpl)
6778+
.directExecutor()
6779+
.build().start());
6780+
6781+
CachedChannelManager channelManager = new CachedChannelManager(config -> {
6782+
return grpcCleanup.register(
6783+
InProcessChannelBuilder.forName(extProcServerName)
6784+
.directExecutor()
6785+
.build());
6786+
});
6787+
6788+
ExternalProcessorServerInterceptor interceptor = new ExternalProcessorServerInterceptor(
6789+
filterConfig, channelManager, FAKE_CONTEXT);
6790+
6791+
final AtomicBoolean callStarted = new AtomicBoolean(false);
6792+
dataPlaneHandler = new DataPlaneServiceHandler() {
6793+
@Override
6794+
public void sayHello(InputStream request, StreamObserver<InputStream> responseObserver) {
6795+
callStarted.set(true);
6796+
responseObserver.onNext(request);
6797+
responseObserver.onCompleted();
6798+
}
6799+
};
6800+
6801+
startDataPlane(interceptor);
6802+
6803+
io.grpc.ClientCall<InputStream, InputStream> clientCall = dataPlaneChannel.newCall(
6804+
METHOD_SAY_HELLO_RAW, io.grpc.CallOptions.DEFAULT);
6805+
6806+
final CountDownLatch callCompletedLatch = new CountDownLatch(1);
6807+
final AtomicReference<Status> closedStatus = new AtomicReference<>();
6808+
clientCall.start(new io.grpc.ClientCall.Listener<InputStream>() {
6809+
@Override
6810+
public void onClose(Status status, Metadata trailers) {
6811+
closedStatus.set(status);
6812+
callCompletedLatch.countDown();
6813+
}
6814+
}, new Metadata());
6815+
6816+
clientCall.request(1);
6817+
clientCall.sendMessage(new ByteArrayInputStream("hello".getBytes(StandardCharsets.UTF_8)));
6818+
clientCall.halfClose();
6819+
6820+
// Verify stream failed
6821+
assertThat(extProcLatch.await(5, TimeUnit.SECONDS)).isTrue();
6822+
// Verify call completed and failed with INTERNAL status (not fail-open)
67206823
assertThat(callCompletedLatch.await(5, TimeUnit.SECONDS)).isTrue();
6721-
assertThat(closedStatus.get().getCode()).isEqualTo(Status.Code.UNAVAILABLE);
6824+
assertThat(callStarted.get()).isFalse();
6825+
assertThat(closedStatus.get().getCode()).isEqualTo(Status.Code.INTERNAL);
6826+
assertThat(closedStatus.get().getDescription()).contains("External processor stream failed");
6827+
}
6828+
6829+
@Test
6830+
public void givenFailureModeAllowTrue_whenExtProcStreamFailsAfterResponseBodySent_thenCallFails()
6831+
throws Exception {
6832+
ExternalProcessor proto = createBaseProto(extProcServerName)
6833+
.setFailureModeAllow(true)
6834+
.setProcessingMode(ProcessingMode.newBuilder()
6835+
.setRequestHeaderMode(ProcessingMode.HeaderSendMode.SKIP)
6836+
.setRequestBodyMode(ProcessingMode.BodySendMode.NONE)
6837+
.setResponseHeaderMode(ProcessingMode.HeaderSendMode.SKIP)
6838+
.setResponseBodyMode(ProcessingMode.BodySendMode.GRPC)
6839+
.setResponseTrailerMode(ProcessingMode.HeaderSendMode.SEND)
6840+
.build())
6841+
.build();
6842+
ConfigOrError<ExternalProcessorFilterConfig> configOrError =
6843+
provider.parseFilterConfig(Any.pack(proto), filterContext);
6844+
assertThat(configOrError.errorDetail).isNull();
6845+
ExternalProcessorFilterConfig filterConfig = configOrError.config;
6846+
6847+
final CountDownLatch extProcLatch = new CountDownLatch(1);
6848+
final AtomicReference<StreamObserver<ProcessingResponse>> responseObserverRef =
6849+
new AtomicReference<>();
6850+
ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl =
6851+
new ExternalProcessorGrpc.ExternalProcessorImplBase() {
6852+
@Override
6853+
public StreamObserver<ProcessingRequest> process(
6854+
final StreamObserver<ProcessingResponse> responseObserver) {
6855+
responseObserverRef.set(responseObserver);
6856+
((ServerCallStreamObserver<ProcessingResponse>) responseObserver).request(100);
6857+
return new StreamObserver<ProcessingRequest>() {
6858+
@Override
6859+
public void onNext(ProcessingRequest request) {
6860+
if (request.hasResponseBody()) {
6861+
// Fail the stream after receiving the response body
6862+
responseObserver.onError(
6863+
Status.INTERNAL.withDescription("Simulated stream failure").asRuntimeException());
6864+
extProcLatch.countDown();
6865+
}
6866+
}
6867+
6868+
@Override
6869+
public void onError(Throwable t) {}
6870+
6871+
@Override
6872+
public void onCompleted() {}
6873+
};
6874+
}
6875+
};
6876+
6877+
grpcCleanup.register(InProcessServerBuilder.forName(extProcServerName)
6878+
.addService(extProcImpl)
6879+
.directExecutor()
6880+
.build().start());
6881+
6882+
CachedChannelManager channelManager = new CachedChannelManager(config -> {
6883+
return grpcCleanup.register(
6884+
InProcessChannelBuilder.forName(extProcServerName)
6885+
.directExecutor()
6886+
.build());
6887+
});
6888+
6889+
ExternalProcessorServerInterceptor interceptor = new ExternalProcessorServerInterceptor(
6890+
filterConfig, channelManager, FAKE_CONTEXT);
6891+
6892+
final AtomicBoolean callStarted = new AtomicBoolean(false);
6893+
dataPlaneHandler = new DataPlaneServiceHandler() {
6894+
@Override
6895+
public void sayHello(InputStream request, StreamObserver<InputStream> responseObserver) {
6896+
callStarted.set(true);
6897+
responseObserver.onNext(request);
6898+
responseObserver.onCompleted();
6899+
}
6900+
};
6901+
6902+
startDataPlane(interceptor);
6903+
6904+
io.grpc.ClientCall<InputStream, InputStream> clientCall = dataPlaneChannel.newCall(
6905+
METHOD_SAY_HELLO_RAW, io.grpc.CallOptions.DEFAULT);
6906+
6907+
final CountDownLatch callCompletedLatch = new CountDownLatch(1);
6908+
final AtomicReference<Status> closedStatus = new AtomicReference<>();
6909+
clientCall.start(new io.grpc.ClientCall.Listener<InputStream>() {
6910+
@Override
6911+
public void onClose(Status status, Metadata trailers) {
6912+
closedStatus.set(status);
6913+
callCompletedLatch.countDown();
6914+
}
6915+
}, new Metadata());
6916+
6917+
clientCall.request(1);
6918+
clientCall.sendMessage(new ByteArrayInputStream("hello".getBytes(StandardCharsets.UTF_8)));
6919+
clientCall.halfClose();
6920+
6921+
// Verify stream failed
6922+
assertThat(extProcLatch.await(5, TimeUnit.SECONDS)).isTrue();
6923+
// Verify call completed and failed with INTERNAL status (not fail-open)
6924+
assertThat(callCompletedLatch.await(5, TimeUnit.SECONDS)).isTrue();
6925+
assertThat(callStarted.get()).isTrue();
6926+
assertThat(closedStatus.get().getCode()).isEqualTo(Status.Code.INTERNAL);
67226927
assertThat(closedStatus.get().getDescription()).contains("External processor stream failed");
67236928
}
67246929

@@ -8167,7 +8372,7 @@ public void onClose(Status status, Metadata trailers) {
81678372

81688373
assertThat(sidecarLatch.await(5, TimeUnit.SECONDS)).isTrue();
81698374
assertThat(callCompletedLatch.await(5, TimeUnit.SECONDS)).isTrue();
8170-
assertThat(receivedStatus.get().getCode()).isEqualTo(Status.Code.UNAVAILABLE);
8375+
assertThat(receivedStatus.get().getCode()).isEqualTo(Status.Code.INTERNAL);
81718376
assertThat(receivedStatus.get().getDescription()).contains("External processor stream failed");
81728377
}
81738378

@@ -8351,7 +8556,7 @@ public void onClose(Status status, Metadata trailers) {
83518556

83528557
assertThat(sidecarLatch.await(5, TimeUnit.SECONDS)).isTrue();
83538558
assertThat(callCompletedLatch.await(5, TimeUnit.SECONDS)).isTrue();
8354-
assertThat(receivedStatus.get().getCode()).isEqualTo(Status.Code.UNAVAILABLE);
8559+
assertThat(receivedStatus.get().getCode()).isEqualTo(Status.Code.INTERNAL);
83558560
assertThat(receivedStatus.get().getDescription()).contains("External processor stream failed");
83568561
}
83578562

@@ -8445,7 +8650,7 @@ public void onClose(Status status, Metadata trailers) {
84458650

84468651
assertThat(sidecarLatch.await(5, TimeUnit.SECONDS)).isTrue();
84478652
assertThat(callCompletedLatch.await(5, TimeUnit.SECONDS)).isTrue();
8448-
assertThat(receivedStatus.get().getCode()).isEqualTo(Status.Code.UNAVAILABLE);
8653+
assertThat(receivedStatus.get().getCode()).isEqualTo(Status.Code.INTERNAL);
84498654
assertThat(receivedStatus.get().getDescription()).contains("External processor stream failed");
84508655
}
84518656

0 commit comments

Comments
 (0)