Skip to content

Commit f52bbd2

Browse files
kannanjgithubkannanjgithub
committed
xds: only install RawMessageClientInterceptor when a filter requires payload access
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. TAG=agy CONV=56c87717-243d-4f12-af9e-c532e509892a
1 parent b5266c1 commit f52bbd2

7 files changed

Lines changed: 291 additions & 133 deletions

File tree

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

Lines changed: 73 additions & 59 deletions
Large diffs are not rendered by default.

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

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

137+
@Override
138+
public boolean requiresPayloadAccess(
139+
FilterConfig filterConfig, @Nullable FilterConfig overrideConfig) {
140+
ExternalProcessorFilterConfig extProcFilterConfig =
141+
(ExternalProcessorFilterConfig) filterConfig;
142+
if (overrideConfig != null) {
143+
extProcFilterConfig = mergeConfigs(extProcFilterConfig,
144+
(ExternalProcessorFilterOverrideConfig) overrideConfig);
145+
}
146+
ProcessingMode mode = extProcFilterConfig.getExternalProcessor().getProcessingMode();
147+
return mode.getRequestBodyMode() != ProcessingMode.BodySendMode.NONE
148+
|| mode.getResponseBodyMode() != ProcessingMode.BodySendMode.NONE;
149+
}
150+
137151
private static ExternalProcessorFilterConfig mergeConfigs(
138152
ExternalProcessorFilterConfig extProcFilterConfig,
139153
ExternalProcessorFilterOverrideConfig extProcFilterConfigOverride) {

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

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

127+
/**
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.
133+
*/
134+
default boolean requiresPayloadAccess(
135+
FilterConfig config, @Nullable FilterConfig overrideConfig) {
136+
return false;
137+
}
138+
127139
/**
128140
* Releases filter resources like shared resources and remote connections.
129141
*

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

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -893,6 +893,7 @@ private ClientInterceptor createFilters(
893893
selectedOverrideConfigs.putAll(weightedCluster.filterConfigOverrides());
894894
}
895895

896+
boolean anyFilterRequiresPayload = false;
896897
ImmutableList.Builder<ClientInterceptor> filterInterceptors = ImmutableList.builder();
897898
for (NamedFilterConfig namedFilter : filterConfigs) {
898899
String name = namedFilter.name;
@@ -907,12 +908,14 @@ private ClientInterceptor createFilters(
907908

908909
if (interceptor != null) {
909910
filterInterceptors.add(interceptor);
911+
if (filter.requiresPayloadAccess(config, overrideConfig)) {
912+
anyFilterRequiresPayload = true;
913+
}
910914
}
911915
}
912916

913917
ImmutableList.Builder<ClientInterceptor> withRawMessage = ImmutableList.builder();
914-
if (GrpcUtil.getFlag("GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_CLIENT", false)
915-
|| GrpcUtil.getFlag("GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_SERVER", false)) {
918+
if (anyFilterRequiresPayload) {
916919
withRawMessage.add(new RawMessageClientInterceptor());
917920
}
918921
withRawMessage.addAll(filterInterceptors.build());
@@ -1138,6 +1141,10 @@ static final class RawMessageClientInterceptor implements ClientInterceptor {
11381141
new MethodDescriptor.Marshaller<InputStream>() {
11391142
@Override
11401143
public InputStream stream(InputStream value) {
1144+
// For retry attempts, RetriableStream calls stream(value) once per attempt.
1145+
// Returning a fresh KnownLengthInputStream wrapping the immutable ByteString ensures
1146+
// each retry attempt reads from the beginning of the payload rather than an already
1147+
// drained stream.
11411148
if (value instanceof KnownLengthInputStream) {
11421149
return new KnownLengthInputStream(((KnownLengthInputStream) value).getByteString());
11431150
}

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

Lines changed: 112 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import io.envoyproxy.envoy.extensions.filters.http.ext_proc.v3.ExtProcPerRoute;
2525
import io.envoyproxy.envoy.extensions.filters.http.ext_proc.v3.ExternalProcessor;
2626
import io.envoyproxy.envoy.extensions.filters.http.ext_proc.v3.ProcessingMode;
27+
import io.grpc.MetricRecorder;
2728
import io.grpc.NameResolver;
2829
import io.grpc.NameResolverProvider;
2930
import io.grpc.NameResolverRegistry;
@@ -341,4 +342,115 @@ public void givenInvalidProto_whenParseFilterConfigOverride_thenReturnsError() t
341342

342343
assertThat(result.errorDetail).contains("Invalid proto:");
343344
}
345+
346+
@Test
347+
public void requiresPayloadAccess_defaultProcessingMode_returnsFalse() throws Exception {
348+
ExternalProcessor proto = createBaseProto(extProcServerName).build();
349+
ExternalProcessorFilterConfig config =
350+
provider.parseFilterConfig(Any.pack(proto), filterContext).config;
351+
352+
try (ExternalProcessorFilter filter = provider.newInstance(
353+
Filter.FilterContext.create("test-filter", new MetricRecorder() {}))) {
354+
assertThat(filter.requiresPayloadAccess(config, null)).isFalse();
355+
}
356+
}
357+
358+
@Test
359+
public void requiresPayloadAccess_explicitNone_returnsFalse() throws Exception {
360+
ExternalProcessor proto = createBaseProto(extProcServerName)
361+
.setProcessingMode(ProcessingMode.newBuilder()
362+
.setRequestBodyMode(ProcessingMode.BodySendMode.NONE)
363+
.setResponseBodyMode(ProcessingMode.BodySendMode.NONE)
364+
.build())
365+
.build();
366+
ExternalProcessorFilterConfig config =
367+
provider.parseFilterConfig(Any.pack(proto), filterContext).config;
368+
369+
try (ExternalProcessorFilter filter = provider.newInstance(
370+
Filter.FilterContext.create("test-filter", new MetricRecorder() {}))) {
371+
assertThat(filter.requiresPayloadAccess(config, null)).isFalse();
372+
}
373+
}
374+
375+
@Test
376+
public void requiresPayloadAccess_requestBodyGrpc_returnsTrue() throws Exception {
377+
ExternalProcessor proto = createBaseProto(extProcServerName)
378+
.setProcessingMode(ProcessingMode.newBuilder()
379+
.setRequestBodyMode(ProcessingMode.BodySendMode.GRPC)
380+
.build())
381+
.build();
382+
ExternalProcessorFilterConfig config =
383+
provider.parseFilterConfig(Any.pack(proto), filterContext).config;
384+
385+
try (ExternalProcessorFilter filter = provider.newInstance(
386+
Filter.FilterContext.create("test-filter", new MetricRecorder() {}))) {
387+
assertThat(filter.requiresPayloadAccess(config, null)).isTrue();
388+
}
389+
}
390+
391+
@Test
392+
public void requiresPayloadAccess_responseBodyGrpc_returnsTrue() throws Exception {
393+
ExternalProcessor proto = createBaseProto(extProcServerName)
394+
.setProcessingMode(ProcessingMode.newBuilder()
395+
.setResponseBodyMode(ProcessingMode.BodySendMode.GRPC)
396+
.setResponseTrailerMode(ProcessingMode.HeaderSendMode.SEND)
397+
.build())
398+
.build();
399+
ExternalProcessorFilterConfig config =
400+
provider.parseFilterConfig(Any.pack(proto), filterContext).config;
401+
402+
try (ExternalProcessorFilter filter = provider.newInstance(
403+
Filter.FilterContext.create("test-filter", new MetricRecorder() {}))) {
404+
assertThat(filter.requiresPayloadAccess(config, null)).isTrue();
405+
}
406+
}
407+
408+
@Test
409+
public void requiresPayloadAccess_overrideTurnsOnPayloadAccess_returnsTrue() throws Exception {
410+
ExternalProcessor proto = createBaseProto(extProcServerName).build();
411+
ExternalProcessorFilterConfig config =
412+
provider.parseFilterConfig(Any.pack(proto), filterContext).config;
413+
414+
ExtProcPerRoute perRoute = ExtProcPerRoute.newBuilder()
415+
.setOverrides(ExtProcOverrides.newBuilder()
416+
.setProcessingMode(ProcessingMode.newBuilder()
417+
.setRequestBodyMode(ProcessingMode.BodySendMode.GRPC)
418+
.build())
419+
.build())
420+
.build();
421+
ExternalProcessorFilterOverrideConfig overrideConfig =
422+
provider.parseFilterConfigOverride(Any.pack(perRoute), filterContext).config;
423+
424+
try (ExternalProcessorFilter filter = provider.newInstance(
425+
Filter.FilterContext.create("test-filter", new MetricRecorder() {}))) {
426+
assertThat(filter.requiresPayloadAccess(config, overrideConfig)).isTrue();
427+
}
428+
}
429+
430+
@Test
431+
public void requiresPayloadAccess_overrideTurnsOffPayloadAccess_returnsFalse() throws Exception {
432+
ExternalProcessor proto = createBaseProto(extProcServerName)
433+
.setProcessingMode(ProcessingMode.newBuilder()
434+
.setRequestBodyMode(ProcessingMode.BodySendMode.GRPC)
435+
.build())
436+
.build();
437+
ExternalProcessorFilterConfig config =
438+
provider.parseFilterConfig(Any.pack(proto), filterContext).config;
439+
440+
ExtProcPerRoute perRoute = ExtProcPerRoute.newBuilder()
441+
.setOverrides(ExtProcOverrides.newBuilder()
442+
.setProcessingMode(ProcessingMode.newBuilder()
443+
.setRequestBodyMode(ProcessingMode.BodySendMode.NONE)
444+
.setResponseBodyMode(ProcessingMode.BodySendMode.NONE)
445+
.build())
446+
.build())
447+
.build();
448+
ExternalProcessorFilterOverrideConfig overrideConfig =
449+
provider.parseFilterConfigOverride(Any.pack(perRoute), filterContext).config;
450+
451+
try (ExternalProcessorFilter filter = provider.newInstance(
452+
Filter.FilterContext.create("test-filter", new MetricRecorder() {}))) {
453+
assertThat(filter.requiresPayloadAccess(config, overrideConfig)).isFalse();
454+
}
455+
}
344456
}

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

Lines changed: 45 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -21,10 +21,16 @@
2121

2222
import com.google.common.collect.ImmutableList;
2323
import com.google.protobuf.Message;
24+
import io.grpc.CallOptions;
25+
import io.grpc.Channel;
26+
import io.grpc.ClientCall;
27+
import io.grpc.ClientInterceptor;
28+
import io.grpc.MethodDescriptor;
2429
import io.grpc.ServerInterceptor;
2530
import java.util.ConcurrentModificationException;
2631
import java.util.concurrent.ConcurrentHashMap;
2732
import java.util.concurrent.ConcurrentMap;
33+
import java.util.concurrent.ScheduledExecutorService;
2834
import java.util.concurrent.atomic.AtomicBoolean;
2935
import java.util.stream.IntStream;
3036
import javax.annotation.Nullable;
@@ -38,10 +44,22 @@ class StatefulFilter implements Filter {
3844
private final AtomicBoolean shutdown = new AtomicBoolean();
3945

4046
final int idx;
47+
private final boolean requiresPayloadAccess;
4148
@Nullable volatile String lastCfg = null;
4249

4350
public StatefulFilter(int idx) {
51+
this(idx, false);
52+
}
53+
54+
public StatefulFilter(int idx, boolean requiresPayloadAccess) {
4455
this.idx = idx;
56+
this.requiresPayloadAccess = requiresPayloadAccess;
57+
}
58+
59+
@Override
60+
public boolean requiresPayloadAccess(
61+
FilterConfig config, @Nullable FilterConfig overrideConfig) {
62+
return requiresPayloadAccess;
4563
}
4664

4765
public boolean isShutdown() {
@@ -67,6 +85,21 @@ public ServerInterceptor buildServerInterceptor(
6785
return null;
6886
}
6987

88+
@Nullable
89+
@Override
90+
public ClientInterceptor buildClientInterceptor(
91+
FilterConfig config,
92+
@Nullable FilterConfig overrideConfig,
93+
ScheduledExecutorService scheduler) {
94+
return new ClientInterceptor() {
95+
@Override
96+
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
97+
MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
98+
return next.newCall(method, callOptions);
99+
}
100+
};
101+
}
102+
70103
@Override
71104
public String toString() {
72105
StringBuilder sb = new StringBuilder().append("StatefulFilter{")
@@ -80,16 +113,26 @@ public String toString() {
80113
static final class Provider implements Filter.Provider {
81114

82115
private final String typeUrl;
116+
private final boolean requiresPayloadAccess;
83117
private final ConcurrentMap<Integer, StatefulFilter> instances = new ConcurrentHashMap<>();
84118

85119
volatile int counter;
86120

87121
Provider() {
88-
this(DEFAULT_TYPE_URL);
122+
this(DEFAULT_TYPE_URL, false);
123+
}
124+
125+
Provider(boolean requiresPayloadAccess) {
126+
this(DEFAULT_TYPE_URL, requiresPayloadAccess);
89127
}
90128

91129
Provider(String typeUrl) {
130+
this(typeUrl, false);
131+
}
132+
133+
Provider(String typeUrl, boolean requiresPayloadAccess) {
92134
this.typeUrl = typeUrl;
135+
this.requiresPayloadAccess = requiresPayloadAccess;
93136
}
94137

95138
@Override
@@ -109,7 +152,7 @@ public boolean isServerFilter() {
109152

110153
@Override
111154
public synchronized StatefulFilter newInstance(FilterContext context) {
112-
StatefulFilter filter = new StatefulFilter(counter++);
155+
StatefulFilter filter = new StatefulFilter(counter++, requiresPayloadAccess);
113156
instances.put(filter.idx, filter);
114157
return filter;
115158
}

0 commit comments

Comments
 (0)