Skip to content

Commit 98e8d20

Browse files
kannanjgithubkannanjgithub
committed
xds: enforce directional and configuration ordering for ext_proc client interceptor
Updates ExternalProcessorClientInterceptor to enforce directional legal ordering and filter configuration rules according to the updated ext_proc gRFC specification: - Ensure legal ordering in client-to-server and server-to-client directions. - Fail with protocol error when ext_proc sends events the filter is not configured to send. - Allow ext_proc to insert messages before the filter has sent any. - Disallow events after end-of-stream. - Propagate protocol error cause as streamFailureCause in the RPC status. TAG=agy CONV=581f86a6-097d-44a2-9d6e-6f9d2e83cbc0
1 parent ca79e36 commit 98e8d20

2 files changed

Lines changed: 1037 additions & 27 deletions

File tree

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

Lines changed: 171 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -69,7 +69,6 @@
6969
import io.grpc.xds.ExternalProcessorFilter.ExternalProcessorFilterConfig;
7070
import io.grpc.xds.Filter.FilterContext;
7171
import io.grpc.xds.internal.extproc.DataPlaneCallState;
72-
import io.grpc.xds.internal.extproc.EventType;
7372
import io.grpc.xds.internal.extproc.ExtProcStreamState;
7473
import io.grpc.xds.internal.extproc.KnownLengthInputStream;
7574
import io.grpc.xds.internal.grpcservice.CachedChannelManager;
@@ -287,8 +286,40 @@ private static class DataPlaneClientCall
287286
private final DataPlaneDelayedCall<InputStream, InputStream> delayedCall;
288287
private final ScheduledExecutorService scheduler;
289288
final Object streamLock = new Object();
290-
@Nullable private volatile EventType expectedRequestResponse;
291-
@Nullable private volatile EventType expectedResponseResponse;
289+
private volatile boolean requestHeadersSent;
290+
private volatile boolean responseHeadersSent;
291+
private volatile boolean responseTrailersSent;
292+
private volatile boolean clientHeadersReceived;
293+
private volatile boolean clientEndOfStreamReceived;
294+
private volatile boolean serverHeadersReceived;
295+
private volatile boolean serverTrailersReceived;
296+
private volatile boolean serverEndOfStreamReceived;
297+
private volatile Throwable streamFailureCause;
298+
299+
private boolean isRequestHeaderConfigured() {
300+
ProcessingMode.HeaderSendMode mode = currentProcessingMode.getRequestHeaderMode();
301+
return mode == ProcessingMode.HeaderSendMode.SEND
302+
|| mode == ProcessingMode.HeaderSendMode.DEFAULT;
303+
}
304+
305+
private boolean isRequestBodyConfigured() {
306+
return currentProcessingMode.getRequestBodyMode() != ProcessingMode.BodySendMode.NONE;
307+
}
308+
309+
private boolean isResponseHeaderConfigured() {
310+
ProcessingMode.HeaderSendMode mode = currentProcessingMode.getResponseHeaderMode();
311+
return mode == ProcessingMode.HeaderSendMode.SEND
312+
|| mode == ProcessingMode.HeaderSendMode.DEFAULT;
313+
}
314+
315+
private boolean isResponseBodyConfigured() {
316+
return currentProcessingMode.getResponseBodyMode() != ProcessingMode.BodySendMode.NONE;
317+
}
318+
319+
private boolean isResponseTrailerConfigured() {
320+
return currentProcessingMode.getResponseTrailerMode() == ProcessingMode.HeaderSendMode.SEND;
321+
}
322+
292323
@Nullable private volatile ClientCallStreamObserver<ProcessingRequest>
293324
extProcClientCallRequestObserver;
294325
@GuardedBy("streamLock")
@@ -528,53 +559,151 @@ public void onNext(ProcessingResponse response) {
528559
}
529560

530561
if (response.hasRequestHeaders()) {
531-
EventType expected = expectedRequestResponse;
532-
if (expected == null || expected != EventType.REQUEST_HEADERS) {
562+
if (!isRequestHeaderConfigured()) {
563+
internalOnError(Status.UNAVAILABLE
564+
.withDescription("Protocol error: received response out of order. "
565+
+ "Filter is not configured to send client headers.")
566+
.asRuntimeException());
567+
return;
568+
}
569+
if (!requestHeadersSent) {
570+
internalOnError(Status.UNAVAILABLE
571+
.withDescription("Protocol error: received response out of order. "
572+
+ "Received request_headers before request_headers was sent.")
573+
.asRuntimeException());
574+
return;
575+
}
576+
if (clientHeadersReceived) {
577+
internalOnError(Status.UNAVAILABLE
578+
.withDescription("Protocol error: received response out of order. "
579+
+ "Duplicate request_headers response.")
580+
.asRuntimeException());
581+
return;
582+
}
583+
if (clientEndOfStreamReceived) {
533584
internalOnError(Status.UNAVAILABLE
534-
.withDescription("Protocol error: received response out of order. Expected: "
535-
+ expected + ", Received: REQUEST_HEADERS")
585+
.withDescription("Protocol error: received response out of order. "
586+
+ "Received request_headers after end-of-stream in "
587+
+ "client-to-server direction.")
536588
.asRuntimeException());
537589
return;
538590
}
539-
expectedRequestResponse = null;
591+
clientHeadersReceived = true;
540592
} else if (response.hasResponseHeaders()) {
541-
EventType expected = expectedResponseResponse;
542-
if (expected == null || expected != EventType.RESPONSE_HEADERS) {
593+
if (!isResponseHeaderConfigured()) {
594+
internalOnError(Status.UNAVAILABLE
595+
.withDescription("Protocol error: received response out of order. "
596+
+ "Filter is not configured to send server headers.")
597+
.asRuntimeException());
598+
return;
599+
}
600+
if (!responseHeadersSent) {
543601
internalOnError(Status.UNAVAILABLE
544-
.withDescription("Protocol error: received response out of order. Expected: "
545-
+ expected + ", Received: RESPONSE_HEADERS")
602+
.withDescription("Protocol error: received response out of order. "
603+
+ "Received response_headers before response_headers was sent.")
546604
.asRuntimeException());
547605
return;
548606
}
549-
expectedResponseResponse = null;
607+
if (serverHeadersReceived) {
608+
internalOnError(Status.UNAVAILABLE
609+
.withDescription("Protocol error: received response out of order. "
610+
+ "Duplicate response_headers response.")
611+
.asRuntimeException());
612+
return;
613+
}
614+
if (serverTrailersReceived || serverEndOfStreamReceived) {
615+
internalOnError(Status.UNAVAILABLE
616+
.withDescription("Protocol error: received response out of order. "
617+
+ "Received response_headers after trailers or end-of-stream.")
618+
.asRuntimeException());
619+
return;
620+
}
621+
serverHeadersReceived = true;
550622
} else if (response.hasResponseTrailers()) {
551-
EventType expected = expectedResponseResponse;
552-
if (expected == null || expected != EventType.RESPONSE_TRAILERS) {
623+
if (!isResponseTrailerConfigured()) {
624+
internalOnError(Status.UNAVAILABLE
625+
.withDescription("Protocol error: received response out of order. "
626+
+ "Filter is not configured to send server trailers.")
627+
.asRuntimeException());
628+
return;
629+
}
630+
if (isResponseHeaderConfigured() && !serverHeadersReceived) {
631+
internalOnError(Status.UNAVAILABLE
632+
.withDescription("Protocol error: received response out of order. "
633+
+ "Received response_trailers before response_headers response.")
634+
.asRuntimeException());
635+
return;
636+
}
637+
if (!responseTrailersSent) {
638+
internalOnError(Status.UNAVAILABLE
639+
.withDescription("Protocol error: received response out of order. "
640+
+ "Received response_trailers before response_trailers was sent.")
641+
.asRuntimeException());
642+
return;
643+
}
644+
if (serverTrailersReceived) {
553645
internalOnError(Status.UNAVAILABLE
554-
.withDescription("Protocol error: received response out of order. Expected: "
555-
+ expected + ", Received: RESPONSE_TRAILERS")
646+
.withDescription("Protocol error: received response out of order. "
647+
+ "Duplicate response_trailers response.")
556648
.asRuntimeException());
557649
return;
558650
}
559-
expectedResponseResponse = null;
651+
if (serverEndOfStreamReceived) {
652+
internalOnError(Status.UNAVAILABLE
653+
.withDescription("Protocol error: received response out of order. "
654+
+ "Received response_trailers after end-of-stream.")
655+
.asRuntimeException());
656+
return;
657+
}
658+
serverTrailersReceived = true;
659+
serverEndOfStreamReceived = true;
560660
} else if (response.hasRequestBody()) {
561-
EventType expected = expectedRequestResponse;
562-
if (expected == EventType.REQUEST_HEADERS) {
661+
if (!isRequestBodyConfigured()) {
662+
internalOnError(Status.UNAVAILABLE
663+
.withDescription(
664+
"Protocol error: received request_body when filter is not configured "
665+
+ "to send client messages.")
666+
.asRuntimeException());
667+
return;
668+
}
669+
if (isRequestHeaderConfigured() && !clientHeadersReceived) {
563670
internalOnError(Status.UNAVAILABLE
564671
.withDescription(
565672
"Protocol error: received request_body before request_headers response.")
566673
.asRuntimeException());
567674
return;
568675
}
676+
if (clientEndOfStreamReceived) {
677+
internalOnError(Status.UNAVAILABLE
678+
.withDescription(
679+
"Protocol error: received request_body after end-of-stream in "
680+
+ "client-to-server direction.")
681+
.asRuntimeException());
682+
return;
683+
}
569684
} else if (response.hasResponseBody()) {
570-
EventType expected = expectedResponseResponse;
571-
if (expected == EventType.RESPONSE_HEADERS) {
685+
if (!isResponseBodyConfigured()) {
686+
internalOnError(Status.UNAVAILABLE
687+
.withDescription(
688+
"Protocol error: received response_body when filter is not configured "
689+
+ "to send server messages.")
690+
.asRuntimeException());
691+
return;
692+
}
693+
if (isResponseHeaderConfigured() && !serverHeadersReceived) {
572694
internalOnError(Status.UNAVAILABLE
573695
.withDescription(
574696
"Protocol error: received response_body before headers response.")
575697
.asRuntimeException());
576698
return;
577699
}
700+
if (serverTrailersReceived || serverEndOfStreamReceived) {
701+
internalOnError(Status.UNAVAILABLE
702+
.withDescription(
703+
"Protocol error: received response_body after trailers or end-of-stream.")
704+
.asRuntimeException());
705+
return;
706+
}
578707
}
579708

580709
if (response.getRequestDrain()) {
@@ -626,6 +755,7 @@ else if (response.hasResponseHeaders()) {
626755
mutator);
627756
}
628757
if (wrappedListener.isTrailersOnly()) {
758+
serverEndOfStreamReceived = true;
629759
wrappedListener.proceedWithClose();
630760
} else {
631761
wrappedListener.proceedWithHeaders();
@@ -719,11 +849,11 @@ private void sendToExtProc(ProcessingRequest request) {
719849
}
720850

721851
if (request.hasRequestHeaders()) {
722-
expectedRequestResponse = EventType.REQUEST_HEADERS;
852+
requestHeadersSent = true;
723853
} else if (request.hasResponseHeaders()) {
724-
expectedResponseResponse = EventType.RESPONSE_HEADERS;
854+
responseHeadersSent = true;
725855
} else if (request.hasResponseTrailers()) {
726-
expectedResponseResponse = EventType.RESPONSE_TRAILERS;
856+
responseTrailersSent = true;
727857
}
728858

729859
ProcessingRequest requestToSend = request;
@@ -862,6 +992,7 @@ private void closeExtProcStream() {
862992
}
863993

864994
private void internalOnError(Throwable t) {
995+
this.streamFailureCause = t;
865996
if (markExtProcStreamFailed(extProcStreamState)) {
866997
synchronized (streamLock) {
867998
if (extProcClientCallRequestObserver != null) {
@@ -1205,6 +1336,7 @@ private void handleRequestBodyResponse(BodyResponse bodyResponse) {
12051336
}
12061337
}
12071338
if (isEndOfStream) {
1339+
clientEndOfStreamReceived = true;
12081340
synchronized (streamLock) {
12091341
if (pendingUpstreamBodyMessages.isEmpty()) {
12101342
if (requestSideClosed.compareAndSet(false, true)) {
@@ -1641,7 +1773,9 @@ public void onClose(Status status, Metadata trailers) {
16411773
|| dataPlaneClientCall.bodyMessageSentToExtProc.get())) {
16421774
if (markDataPlaneCallClosed(dataPlaneClientCall.dataPlaneCallState)) {
16431775
proceedWithClose(Status.INTERNAL.withDescription("External processor stream failed")
1644-
.withCause(status.getCause()), new Metadata());
1776+
.withCause(dataPlaneClientCall.streamFailureCause != null
1777+
? dataPlaneClientCall.streamFailureCause
1778+
: status.getCause()), new Metadata());
16451779
}
16461780
return;
16471781
}
@@ -1713,7 +1847,17 @@ private void proceedWithHeaders(Metadata headers) {
17131847
void proceedWithClose() {
17141848
if (savedStatus != null) {
17151849
if (markDataPlaneCallClosed(dataPlaneClientCall.dataPlaneCallState)) {
1716-
proceedWithClose(savedStatus, savedTrailers);
1850+
Status toDeliver = savedStatus;
1851+
if (savedStatus.isOk()
1852+
&& dataPlaneClientCall.getExtProcStreamState().get().isFailed()
1853+
&& (!dataPlaneClientCall.getConfig().getFailureModeAllow()
1854+
|| dataPlaneClientCall.bodyMessageSentToExtProc.get())) {
1855+
toDeliver = Status.INTERNAL.withDescription("External processor stream failed")
1856+
.withCause(dataPlaneClientCall.streamFailureCause != null
1857+
? dataPlaneClientCall.streamFailureCause
1858+
: savedStatus.getCause());
1859+
}
1860+
proceedWithClose(toDeliver, savedTrailers);
17171861
}
17181862
savedStatus = null;
17191863
savedTrailers = null;

0 commit comments

Comments
 (0)