Skip to content

Commit ca5c574

Browse files
kannanjgithubkannanjgithub
committed
xds: decouple request/response payload interception and address review feedback
- Split Filter.requiresPayloadAccess into requiresRequestPayloadAccess and requiresResponsePayloadAccess so RawMessageClientInterceptor only intercepts the direction(s) that require payload access. - Short-circuit ExternalProcessorUtil.outboundStreamToByteString when the input stream is already a KnownLengthInputStream, and reuse it in RawMessageClientInterceptor. - Cancel the call in RawMessageClientInterceptor.sendMessage when reading the request stream fails with IOException or RuntimeException. - Fail the RPC as a protocol error if ext_proc returns a request_body or response_body response when the corresponding body mode is not GRPC. - Guard DataPlaneListener.onMessage and drainSavedMessages against casting response messages to InputStream when responseBodyMode != GRPC. TAG=agy CONV=56c87717-243d-4f12-af9e-c532e509892a
1 parent f52bbd2 commit ca5c574

9 files changed

Lines changed: 821 additions & 170 deletions

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

Lines changed: 40 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -549,6 +549,15 @@ public void onNext(ProcessingResponse response) {
549549
}
550550
expectedResponseResponse = null;
551551
} else if (response.hasRequestBody()) {
552+
if (currentProcessingMode.getRequestBodyMode()
553+
!= ProcessingMode.BodySendMode.GRPC) {
554+
internalOnError(Status.UNAVAILABLE
555+
.withDescription(
556+
"Protocol error: received unexpected request_body response when"
557+
+ " request_body_mode is not GRPC.")
558+
.asRuntimeException());
559+
return;
560+
}
552561
EventType expected = expectedRequestResponse;
553562
if (expected == EventType.REQUEST_HEADERS) {
554563
internalOnError(Status.UNAVAILABLE
@@ -558,6 +567,15 @@ public void onNext(ProcessingResponse response) {
558567
return;
559568
}
560569
} else if (response.hasResponseBody()) {
570+
if (currentProcessingMode.getResponseBodyMode()
571+
!= ProcessingMode.BodySendMode.GRPC) {
572+
internalOnError(Status.UNAVAILABLE
573+
.withDescription(
574+
"Protocol error: received unexpected response_body response when"
575+
+ " response_body_mode is not GRPC.")
576+
.asRuntimeException());
577+
return;
578+
}
561579
EventType expected = expectedResponseResponse;
562580
if (expected == EventType.RESPONSE_HEADERS) {
563581
internalOnError(Status.UNAVAILABLE
@@ -999,7 +1017,7 @@ public void sendMessage(ReqT message) {
9991017
return;
10001018
}
10011019
try {
1002-
ByteString copiedBody = ByteString.readFrom((InputStream) message);
1020+
ByteString copiedBody = outboundStreamToByteString((InputStream) message);
10031021
pendingDrainingMessages.add((ReqT) new KnownLengthInputStream(copiedBody));
10041022
} catch (IOException e) {
10051023
cancelDownstream("Failed to copy outbound message for buffering", e);
@@ -1539,39 +1557,29 @@ public void onMessage(RespT message) {
15391557
return;
15401558
}
15411559

1542-
if (dataPlaneClientCall.getCurrentProcessingMode().getResponseBodyMode()
1543-
!= ProcessingMode.BodySendMode.GRPC) {
1544-
if (savedHeaders != null) {
1545-
savedMessages.add(message);
1546-
return;
1547-
}
1548-
if (dataPlaneClientCall.getPassThroughMode().get()
1549-
|| dataPlaneClientCall.getExtProcStreamState().get().isCompleted()) {
1550-
dataPlaneClientCall.getCallContext().run(() -> delegate().onMessage(message));
1551-
return;
1552-
}
1553-
dataPlaneClientCall.getCallContext().run(() -> delegate().onMessage(message));
1554-
return;
1555-
}
1556-
1557-
boolean checkDrain = dataPlaneClientCall.getExtProcStreamState().get().isDraining();
1560+
boolean checkDrain = dataPlaneClientCall.getExtProcStreamState().get().isDraining()
1561+
&& dataPlaneClientCall.getCurrentProcessingMode().getResponseBodyMode()
1562+
== ProcessingMode.BodySendMode.GRPC;
15581563

15591564
if (savedHeaders != null || checkDrain) {
1560-
try {
1561-
ByteString copiedBody = ByteString.readFrom((InputStream) message);
1562-
savedMessages.add((RespT) new KnownLengthInputStream(copiedBody));
1563-
} catch (IOException e) {
1564-
dataPlaneClientCall.cancelDownstream("Failed to copy inbound message for buffering", e);
1565+
if (message instanceof InputStream) {
1566+
try {
1567+
ByteString copiedBody = ByteString.readFrom((InputStream) message);
1568+
savedMessages.add((RespT) new KnownLengthInputStream(copiedBody));
1569+
} catch (IOException e) {
1570+
dataPlaneClientCall.cancelDownstream(
1571+
"Failed to copy inbound message for buffering", e);
1572+
}
1573+
} else {
1574+
savedMessages.add(message);
15651575
}
15661576
return;
15671577
}
15681578

1569-
if (dataPlaneClientCall.getPassThroughMode().get()) {
1570-
dataPlaneClientCall.getCallContext().run(() -> delegate().onMessage(message));
1571-
return;
1572-
}
1573-
1574-
if (dataPlaneClientCall.getExtProcStreamState().get().isCompleted()) {
1579+
if (dataPlaneClientCall.getPassThroughMode().get()
1580+
|| dataPlaneClientCall.getExtProcStreamState().get().isCompleted()
1581+
|| dataPlaneClientCall.getCurrentProcessingMode().getResponseBodyMode()
1582+
!= ProcessingMode.BodySendMode.GRPC) {
15751583
dataPlaneClientCall.getCallContext().run(() -> delegate().onMessage(message));
15761584
return;
15771585
}
@@ -1602,6 +1610,10 @@ public void onMessage(RespT message) {
16021610

16031611
void drainSavedMessages() {
16041612
synchronized (dataPlaneClientCall.streamLock) {
1613+
if (dataPlaneClientCall.getCurrentProcessingMode().getResponseBodyMode()
1614+
!= ProcessingMode.BodySendMode.GRPC) {
1615+
return;
1616+
}
16051617
while (dataPlaneClientCall.isSidecarReady()
16061618
&& dataPlaneClientCall.upstreamToSidestreamWindow > 0
16071619
&& !savedMessages.isEmpty()) {

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

Lines changed: 15 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -135,17 +135,28 @@ public ClientInterceptor buildClientInterceptor(FilterConfig filterConfig,
135135
}
136136

137137
@Override
138-
public boolean requiresPayloadAccess(
138+
public boolean requiresRequestPayloadAccess(
139+
FilterConfig filterConfig, @Nullable FilterConfig overrideConfig) {
140+
ProcessingMode mode = getProcessingMode(filterConfig, overrideConfig);
141+
return mode.getRequestBodyMode() != ProcessingMode.BodySendMode.NONE;
142+
}
143+
144+
@Override
145+
public boolean requiresResponsePayloadAccess(
146+
FilterConfig filterConfig, @Nullable FilterConfig overrideConfig) {
147+
ProcessingMode mode = getProcessingMode(filterConfig, overrideConfig);
148+
return mode.getResponseBodyMode() != ProcessingMode.BodySendMode.NONE;
149+
}
150+
151+
private static ProcessingMode getProcessingMode(
139152
FilterConfig filterConfig, @Nullable FilterConfig overrideConfig) {
140153
ExternalProcessorFilterConfig extProcFilterConfig =
141154
(ExternalProcessorFilterConfig) filterConfig;
142155
if (overrideConfig != null) {
143156
extProcFilterConfig = mergeConfigs(extProcFilterConfig,
144157
(ExternalProcessorFilterOverrideConfig) overrideConfig);
145158
}
146-
ProcessingMode mode = extProcFilterConfig.getExternalProcessor().getProcessingMode();
147-
return mode.getRequestBodyMode() != ProcessingMode.BodySendMode.NONE
148-
|| mode.getResponseBodyMode() != ProcessingMode.BodySendMode.NONE;
159+
return extProcFilterConfig.getExternalProcessor().getProcessingMode();
149160
}
150161

151162
private static ExternalProcessorFilterConfig mergeConfigs(

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

Lines changed: 12 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -125,13 +125,19 @@ default ServerInterceptor buildServerInterceptor(
125125
}
126126

127127
/**
128-
* Returns true if this filter requires access to the request or response message payloads for
129-
* the given configuration.
130-
*
131-
* <p>When true, interceptors that provide raw message payload access (e.g. {@code
132-
* RawMessageClientInterceptor}) will be installed in the client interceptor chain.
128+
* Returns true if this filter requires access to the request message payloads for the given
129+
* configuration.
130+
*/
131+
default boolean requiresRequestPayloadAccess(
132+
FilterConfig config, @Nullable FilterConfig overrideConfig) {
133+
return false;
134+
}
135+
136+
/**
137+
* Returns true if this filter requires access to the response message payloads for the given
138+
* configuration.
133139
*/
134-
default boolean requiresPayloadAccess(
140+
default boolean requiresResponsePayloadAccess(
135141
FilterConfig config, @Nullable FilterConfig overrideConfig) {
136142
return false;
137143
}

0 commit comments

Comments
 (0)