Skip to content

Commit 29d07cc

Browse files
kannanjgithubkannanjgithub
authored
xDS: Support retries in RawMessageClientInterceptor and conditionally add it to the filter chain (#13042)
### 1. Supporting retries Previously, `RawMessageClientInterceptor.sendMessage()` forwarded a single-use InputStream directly to `rawCall.sendMessage()`. When retries were triggered by RetriableStream, attempt 1 drained the InputStream to EOF. Subsequent retry attempts would then serialize the already-drained stream, sending an empty (0-byte) request payload. Fix this by: 1. Buffering the request stream into an immutable `ByteString` (using `Drainable` when supported) in `RawMessageClientInterceptor.sendMessage()` and wrapping it in a `KnownLengthInputStream`. 2. Exposing `getByteString()` on `KnownLengthInputStream`. 3. Having `RAW_MARSHALLER.stream()` return a fresh `KnownLengthInputStream` wrapping the underlying `ByteString` on each attempt, allowing retries to read the payload from byte 0. ### 2. Conditional addition of `RawMessageClientInterceptor` Also we need to only install `RawMessageClientInterceptor` when a filter requires payload access to avoid the performance hit for non ext_proc cases. As suggested in the [grfc discussion](grpc/proposal#484 (comment)), introduce `Filter.requiresPayloadAccess(config, overrideConfig)` to allow filters to declare whether they need access to message payloads. Update `ExternalProcessorFilter` to implement this method, returning false when body send mode is NONE for both request and response. Update `XdsNameResolver` to only install `RawMessageClientInterceptor` when at least one configured filter requires payload access. Generalize `ExternalProcessorClientInterceptor` to support generic request/response types when payload interception is not needed. #### Generalizing `ExternalProcessorClientInterceptor` to Support Pass-through Types `ExternalProcessorClientInterceptor` previously hardcoded `ClientCall<InputStream, InputStream>` and `SimpleForwardingClientCallListener<InputStream>`, assuming `RawMessageClientInterceptor` was always present earlier in the chain to convert application messages to `InputStream`. Now that `RawMessageClientInterceptor` is omitted when body send mode is `NONE`, `ExternalProcessorClientInterceptor` directly handles the application's request and response types (`ReqT`, `RespT`). `DataPlaneClientCall` and `DataPlaneListener` were generalized to `<ReqT, RespT>` to avoid compiler-generated synthetic bridge casts to `InputStream`, which would otherwise result in a runtime `ClassCastException` when sending or receiving application messages. Fixes #13010
1 parent ecbe959 commit 29d07cc

10 files changed

Lines changed: 1239 additions & 192 deletions

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

Lines changed: 93 additions & 62 deletions
Large diffs are not rendered by default.

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

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -134,6 +134,31 @@ public ClientInterceptor buildClientInterceptor(FilterConfig filterConfig,
134134
extProcFilterConfig, cachedChannelManager, scheduler, context);
135135
}
136136

137+
@Override
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(
152+
FilterConfig filterConfig, @Nullable FilterConfig overrideConfig) {
153+
ExternalProcessorFilterConfig extProcFilterConfig =
154+
(ExternalProcessorFilterConfig) filterConfig;
155+
if (overrideConfig != null) {
156+
extProcFilterConfig = mergeConfigs(extProcFilterConfig,
157+
(ExternalProcessorFilterOverrideConfig) overrideConfig);
158+
}
159+
return extProcFilterConfig.getExternalProcessor().getProcessingMode();
160+
}
161+
137162
private static ExternalProcessorFilterConfig mergeConfigs(
138163
ExternalProcessorFilterConfig extProcFilterConfig,
139164
ExternalProcessorFilterOverrideConfig extProcFilterConfigOverride) {

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

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -124,6 +124,24 @@ default ServerInterceptor buildServerInterceptor(
124124
return null;
125125
}
126126

127+
/**
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.
139+
*/
140+
default boolean requiresResponsePayloadAccess(
141+
FilterConfig config, @Nullable FilterConfig overrideConfig) {
142+
return false;
143+
}
144+
127145
/**
128146
* Releases filter resources like shared resources and remote connections.
129147
*

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

Lines changed: 168 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
import com.google.common.collect.ImmutableMap;
2828
import com.google.common.collect.Sets;
2929
import com.google.gson.Gson;
30+
import com.google.protobuf.ByteString;
3031
import com.google.protobuf.util.Durations;
3132
import io.grpc.Attributes;
3233
import io.grpc.CallOptions;
@@ -69,6 +70,9 @@
6970
import io.grpc.xds.client.XdsInitializationException;
7071
import io.grpc.xds.client.XdsLogger;
7172
import io.grpc.xds.client.XdsLogger.XdsLogLevel;
73+
import io.grpc.xds.internal.extproc.ExternalProcessorUtil;
74+
import io.grpc.xds.internal.extproc.KnownLengthInputStream;
75+
import java.io.IOException;
7276
import java.io.InputStream;
7377
import java.util.ArrayList;
7478
import java.util.Collections;
@@ -889,6 +893,8 @@ private ClientInterceptor createFilters(
889893
selectedOverrideConfigs.putAll(weightedCluster.filterConfigOverrides());
890894
}
891895

896+
boolean anyFilterRequiresRequestPayload = false;
897+
boolean anyFilterRequiresResponsePayload = false;
892898
ImmutableList.Builder<ClientInterceptor> filterInterceptors = ImmutableList.builder();
893899
for (NamedFilterConfig namedFilter : filterConfigs) {
894900
String name = namedFilter.name;
@@ -903,13 +909,19 @@ private ClientInterceptor createFilters(
903909

904910
if (interceptor != null) {
905911
filterInterceptors.add(interceptor);
912+
if (filter.requiresRequestPayloadAccess(config, overrideConfig)) {
913+
anyFilterRequiresRequestPayload = true;
914+
}
915+
if (filter.requiresResponsePayloadAccess(config, overrideConfig)) {
916+
anyFilterRequiresResponsePayload = true;
917+
}
906918
}
907919
}
908920

909921
ImmutableList.Builder<ClientInterceptor> withRawMessage = ImmutableList.builder();
910-
if (GrpcUtil.getFlag("GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_CLIENT", false)
911-
|| GrpcUtil.getFlag("GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_SERVER", false)) {
912-
withRawMessage.add(new RawMessageClientInterceptor());
922+
if (anyFilterRequiresRequestPayload || anyFilterRequiresResponsePayload) {
923+
withRawMessage.add(new RawMessageClientInterceptor(
924+
anyFilterRequiresRequestPayload, anyFilterRequiresResponsePayload));
913925
}
914926
withRawMessage.addAll(filterInterceptors.build());
915927
return combineInterceptors(withRawMessage.build());
@@ -1134,6 +1146,13 @@ static final class RawMessageClientInterceptor implements ClientInterceptor {
11341146
new MethodDescriptor.Marshaller<InputStream>() {
11351147
@Override
11361148
public InputStream stream(InputStream value) {
1149+
// For retry attempts, RetriableStream calls stream(value) once per attempt.
1150+
// Returning a fresh KnownLengthInputStream wrapping the immutable ByteString ensures
1151+
// each retry attempt reads from the beginning of the payload rather than an already
1152+
// drained stream.
1153+
if (value instanceof KnownLengthInputStream) {
1154+
return new KnownLengthInputStream(((KnownLengthInputStream) value).getByteString());
1155+
}
11371156
return value;
11381157
}
11391158

@@ -1143,73 +1162,186 @@ public InputStream parse(InputStream stream) {
11431162
}
11441163
};
11451164

1165+
private final boolean interceptRequest;
1166+
private final boolean interceptResponse;
1167+
1168+
RawMessageClientInterceptor() {
1169+
this(true, true);
1170+
}
1171+
1172+
RawMessageClientInterceptor(boolean interceptRequest, boolean interceptResponse) {
1173+
this.interceptRequest = interceptRequest;
1174+
this.interceptResponse = interceptResponse;
1175+
}
1176+
11461177
@Override
11471178
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
11481179
final MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
1180+
if (interceptRequest && interceptResponse) {
1181+
return interceptBoth(method, callOptions, next);
1182+
} else if (interceptRequest) {
1183+
return interceptRequestOnly(method, callOptions, next);
1184+
} else if (interceptResponse) {
1185+
return interceptResponseOnly(method, callOptions, next);
1186+
} else {
1187+
return next.newCall(method, callOptions);
1188+
}
1189+
}
1190+
1191+
private static <ReqT, RespT> ClientCall<ReqT, RespT> interceptBoth(
1192+
final MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
11491193
MethodDescriptor<InputStream, InputStream> rawMethod =
11501194
method.toBuilder(RAW_MARSHALLER, RAW_MARSHALLER).build();
11511195
final ClientCall<InputStream, InputStream> rawCall = next.newCall(rawMethod, callOptions);
1152-
return new ClientCall<ReqT, RespT>() {
1196+
return new PartialForwardingClientCall<ReqT, RespT>() {
11531197
@Override
1154-
public void start(final Listener<RespT> responseListener, Metadata headers) {
1155-
rawCall.start(new Listener<InputStream>() {
1156-
@Override
1157-
public void onHeaders(Metadata headers) {
1158-
responseListener.onHeaders(headers);
1159-
}
1198+
protected ClientCall<?, ?> delegate() {
1199+
return rawCall;
1200+
}
11601201

1161-
@Override
1162-
public void onMessage(InputStream message) {
1163-
responseListener.onMessage(method.getResponseMarshaller().parse(message));
1164-
}
1202+
@Override
1203+
public void start(Listener<RespT> responseListener, Metadata headers) {
1204+
rawCall.start(createRawListener(method, responseListener), headers);
1205+
}
11651206

1166-
@Override
1167-
public void onClose(Status status, Metadata trailers) {
1168-
responseListener.onClose(status, trailers);
1169-
}
1207+
@Override
1208+
public void sendMessage(ReqT message) {
1209+
sendSerializedMessage(rawCall, method, message);
1210+
}
1211+
};
1212+
}
11701213

1171-
@Override
1172-
public void onReady() {
1173-
responseListener.onReady();
1174-
}
1175-
}, headers);
1214+
private static <ReqT, RespT> ClientCall<ReqT, RespT> interceptRequestOnly(
1215+
final MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
1216+
MethodDescriptor<InputStream, RespT> rawReqMethod =
1217+
method.toBuilder(RAW_MARSHALLER, method.getResponseMarshaller()).build();
1218+
final ClientCall<InputStream, RespT> rawCall = next.newCall(rawReqMethod, callOptions);
1219+
return new PartialForwardingClientCall<ReqT, RespT>() {
1220+
@Override
1221+
protected ClientCall<?, ?> delegate() {
1222+
return rawCall;
1223+
}
1224+
1225+
@Override
1226+
public void start(Listener<RespT> responseListener, Metadata headers) {
1227+
rawCall.start(responseListener, headers);
11761228
}
11771229

11781230
@Override
1179-
public void request(int numMessages) {
1180-
rawCall.request(numMessages);
1231+
public void sendMessage(ReqT message) {
1232+
sendSerializedMessage(rawCall, method, message);
11811233
}
1234+
};
1235+
}
11821236

1237+
private static <ReqT, RespT> ClientCall<ReqT, RespT> interceptResponseOnly(
1238+
final MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
1239+
MethodDescriptor<ReqT, InputStream> rawRespMethod =
1240+
method.toBuilder(method.getRequestMarshaller(), RAW_MARSHALLER).build();
1241+
final ClientCall<ReqT, InputStream> rawCall = next.newCall(rawRespMethod, callOptions);
1242+
return new PartialForwardingClientCall<ReqT, RespT>() {
11831243
@Override
1184-
public void cancel(@Nullable String message, @Nullable Throwable cause) {
1185-
rawCall.cancel(message, cause);
1244+
protected ClientCall<?, ?> delegate() {
1245+
return rawCall;
11861246
}
11871247

11881248
@Override
1189-
public void halfClose() {
1190-
rawCall.halfClose();
1249+
public void start(Listener<RespT> responseListener, Metadata headers) {
1250+
rawCall.start(createRawListener(method, responseListener), headers);
11911251
}
11921252

11931253
@Override
11941254
public void sendMessage(ReqT message) {
1195-
rawCall.sendMessage(method.getRequestMarshaller().stream(message));
1255+
rawCall.sendMessage(message);
1256+
}
1257+
};
1258+
}
1259+
1260+
private static <RespT> ClientCall.Listener<InputStream> createRawListener(
1261+
final MethodDescriptor<?, RespT> method,
1262+
final ClientCall.Listener<RespT> responseListener) {
1263+
return new ClientCall.Listener<InputStream>() {
1264+
@Override
1265+
public void onHeaders(Metadata headers) {
1266+
responseListener.onHeaders(headers);
11961267
}
11971268

11981269
@Override
1199-
public boolean isReady() {
1200-
return rawCall.isReady();
1270+
public void onMessage(InputStream message) {
1271+
responseListener.onMessage(method.getResponseMarshaller().parse(message));
12011272
}
12021273

12031274
@Override
1204-
public void setMessageCompression(boolean enabled) {
1205-
rawCall.setMessageCompression(enabled);
1275+
public void onClose(Status status, Metadata trailers) {
1276+
responseListener.onClose(status, trailers);
12061277
}
12071278

12081279
@Override
1209-
public Attributes getAttributes() {
1210-
return rawCall.getAttributes();
1280+
public void onReady() {
1281+
responseListener.onReady();
12111282
}
12121283
};
12131284
}
1285+
1286+
private static <ReqT> void sendSerializedMessage(
1287+
ClientCall<InputStream, ?> rawCall,
1288+
MethodDescriptor<ReqT, ?> method,
1289+
ReqT message) {
1290+
ByteString byteString;
1291+
try {
1292+
InputStream stream = method.getRequestMarshaller().stream(message);
1293+
try {
1294+
byteString = ExternalProcessorUtil.outboundStreamToByteString(stream);
1295+
} finally {
1296+
try {
1297+
stream.close();
1298+
} catch (IOException ignored) {
1299+
// ignore
1300+
}
1301+
}
1302+
} catch (IOException | RuntimeException e) {
1303+
rawCall.cancel("Failed to read message for raw message interceptor", e);
1304+
return;
1305+
} catch (Error e) {
1306+
rawCall.cancel("Client sendMessage() failed with Error", e);
1307+
throw e;
1308+
}
1309+
rawCall.sendMessage(new KnownLengthInputStream(byteString));
1310+
}
1311+
1312+
private abstract static class PartialForwardingClientCall<ReqT, RespT>
1313+
extends ClientCall<ReqT, RespT> {
1314+
protected abstract ClientCall<?, ?> delegate();
1315+
1316+
@Override
1317+
public void request(int numMessages) {
1318+
delegate().request(numMessages);
1319+
}
1320+
1321+
@Override
1322+
public void cancel(@Nullable String message, @Nullable Throwable cause) {
1323+
delegate().cancel(message, cause);
1324+
}
1325+
1326+
@Override
1327+
public void halfClose() {
1328+
delegate().halfClose();
1329+
}
1330+
1331+
@Override
1332+
public boolean isReady() {
1333+
return delegate().isReady();
1334+
}
1335+
1336+
@Override
1337+
public void setMessageCompression(boolean enabled) {
1338+
delegate().setMessageCompression(enabled);
1339+
}
1340+
1341+
@Override
1342+
public Attributes getAttributes() {
1343+
return delegate().getAttributes();
1344+
}
1345+
}
12141346
}
12151347
}

‎xds/src/main/java/io/grpc/xds/internal/extproc/ExternalProcessorUtil.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -55,6 +55,9 @@ private ExternalProcessorUtil() {}
5555
* Reads an InputStream into a ByteString.
5656
*/
5757
public static ByteString outboundStreamToByteString(InputStream message) throws IOException {
58+
if (message instanceof KnownLengthInputStream) {
59+
return ((KnownLengthInputStream) message).getByteString();
60+
}
5861
if (message instanceof Drainable) {
5962
int size = message.available();
6063
ByteString.Output output =

‎xds/src/main/java/io/grpc/xds/internal/extproc/KnownLengthInputStream.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,12 +25,18 @@
2525
* An {@link InputStream} backed by a {@link ByteString} that implements {@link KnownLength}.
2626
*/
2727
public final class KnownLengthInputStream extends InputStream implements KnownLength {
28+
private final ByteString byteString;
2829
private final InputStream delegate;
2930

3031
public KnownLengthInputStream(ByteString byteString) {
32+
this.byteString = byteString;
3133
this.delegate = byteString.newInput();
3234
}
3335

36+
public ByteString getByteString() {
37+
return byteString;
38+
}
39+
3440
@Override
3541
public int read() throws IOException {
3642
return delegate.read();

0 commit comments

Comments
 (0)