Skip to content

Commit 7208ff8

Browse files
kannanjgithubkannanjgithub
committed
xds: require processing_mode in ExternalProcessor filter config
Reject xDS filter configuration if processing_mode is unset and fix checkstyle issues. TAG=agy CONV=581f86a6-097d-44a2-9d6e-6f9d2e83cbc0
1 parent a2887c6 commit 7208ff8

4 files changed

Lines changed: 86 additions & 29 deletions

File tree

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

Lines changed: 12 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -497,9 +497,11 @@ public void onNext(ProcessingResponse response) {
497497
synchronized (streamLock) {
498498
wasReady = isReady();
499499
downstreamToSidestreamWindow = LongMath.saturatedAdd(
500-
downstreamToSidestreamWindow, update.getWindowIncrementDownstreamToSidestream());
500+
downstreamToSidestreamWindow,
501+
update.getWindowIncrementDownstreamToSidestream());
501502
upstreamToSidestreamWindow = LongMath.saturatedAdd(
502-
upstreamToSidestreamWindow, update.getWindowIncrementUpstreamToSidestream());
503+
upstreamToSidestreamWindow,
504+
update.getWindowIncrementUpstreamToSidestream());
503505
drainPendingRequestBodyMessages();
504506
drainPendingRequests();
505507
if (wrappedListener != null) {
@@ -782,8 +784,10 @@ void mergeAccumulatedWindowUpdates(ProcessingRequest.Builder requestBuilder) {
782784
.build());
783785
accumulatedWindowUpdateSidestreamToUpstream -= incrementUpstream;
784786
accumulatedWindowUpdateSidestreamToDownstream -= incrementDownstream;
785-
sidestreamToUpstreamWindow = LongMath.saturatedAdd(sidestreamToUpstreamWindow, incrementUpstream);
786-
sidestreamToDownstreamWindow = LongMath.saturatedAdd(sidestreamToDownstreamWindow, incrementDownstream);
787+
sidestreamToUpstreamWindow =
788+
LongMath.saturatedAdd(sidestreamToUpstreamWindow, incrementUpstream);
789+
sidestreamToDownstreamWindow =
790+
LongMath.saturatedAdd(sidestreamToDownstreamWindow, incrementDownstream);
787791
}
788792
}
789793

@@ -806,8 +810,10 @@ private void trySendAccumulatedWindowUpdates() {
806810
if (shouldSend) {
807811
accumulatedWindowUpdateSidestreamToUpstream -= incrementUpstream;
808812
accumulatedWindowUpdateSidestreamToDownstream -= incrementDownstream;
809-
sidestreamToUpstreamWindow = LongMath.saturatedAdd(sidestreamToUpstreamWindow, incrementUpstream);
810-
sidestreamToDownstreamWindow = LongMath.saturatedAdd(sidestreamToDownstreamWindow, incrementDownstream);
813+
sidestreamToUpstreamWindow =
814+
LongMath.saturatedAdd(sidestreamToUpstreamWindow, incrementUpstream);
815+
sidestreamToDownstreamWindow =
816+
LongMath.saturatedAdd(sidestreamToDownstreamWindow, incrementDownstream);
811817

812818
sendToExtProc(ProcessingRequest.newBuilder()
813819
.setClientWindowUpdate(ProcessingRequest.ClientWindowUpdate.newBuilder()

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

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -219,6 +219,9 @@ static final class ExternalProcessorFilterConfig implements FilterConfig {
219219

220220
static ConfigOrError<ExternalProcessorFilterConfig> create(
221221
ExternalProcessor externalProcessor, FilterConfigParseContext context) {
222+
if (!externalProcessor.hasProcessingMode()) {
223+
return ConfigOrError.fromError("Missing required field: processing_mode");
224+
}
222225
ProcessingMode mode = externalProcessor.getProcessingMode();
223226
GrpcService grpcService = externalProcessor.getGrpcService();
224227
HeaderMutationRulesConfig mutationRulesConfig = null;

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

Lines changed: 47 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -274,7 +274,8 @@ private ExternalProcessor.Builder createBaseProto(String targetName) {
274274
+ "channel_credentials.insecure.v3.InsecureCredentials")
275275
.build())
276276
.build())
277-
.build());
277+
.build())
278+
.setProcessingMode(ProcessingMode.getDefaultInstance());
278279
}
279280

280281
// --- Category 1: Configuration Override ---
@@ -644,6 +645,7 @@ public void givenInterceptor_whenCallIntercepted_thenExtProcStubUsesSerializingE
644645
.build())
645646
.build())
646647
.build())
648+
.setProcessingMode(ProcessingMode.getDefaultInstance())
647649
.build();
648650
ConfigOrError<ExternalProcessorFilterConfig> configOrError =
649651
provider.parseFilterConfig(Any.pack(proto), filterContext);
@@ -739,6 +741,7 @@ public void givenGrpcServiceWithTimeout_whenCallIntercepted_thenExtProcStubHasCo
739741
.build())
740742
.setTimeout(com.google.protobuf.Duration.newBuilder().setSeconds(5).build())
741743
.build())
744+
.setProcessingMode(ProcessingMode.getDefaultInstance())
742745
.build();
743746
ConfigOrError<ExternalProcessorFilterConfig> configOrError =
744747
provider.parseFilterConfig(Any.pack(proto), filterContext);
@@ -6066,6 +6069,7 @@ public void givenObservabilityTrue_whenExtProcBusy_thenIsReadyReturnsFalse()
60666069
.build())
60676070
.build())
60686071
.setObservabilityMode(true)
6072+
.setProcessingMode(ProcessingMode.getDefaultInstance())
60696073
.build();
60706074
ConfigOrError<ExternalProcessorFilterConfig> configOrError =
60716075
provider.parseFilterConfig(Any.pack(proto), filterContext);
@@ -6188,6 +6192,7 @@ public void givenObservabilityMode_whenUpstreamBusy_thenIsReadyReturnsFalse()
61886192
.build())
61896193
.build())
61906194
.setObservabilityMode(true)
6195+
.setProcessingMode(ProcessingMode.getDefaultInstance())
61916196
.build();
61926197
ConfigOrError<ExternalProcessorFilterConfig> configOrError =
61936198
provider.parseFilterConfig(Any.pack(proto), filterContext);
@@ -6312,6 +6317,7 @@ public void givenNormalMode_whenUpstreamBusy_thenIsReadyReturnsTrue()
63126317
.build())
63136318
.build())
63146319
.setObservabilityMode(false)
6320+
.setProcessingMode(ProcessingMode.getDefaultInstance())
63156321
.build();
63166322
ConfigOrError<ExternalProcessorFilterConfig> configOrError =
63176323
provider.parseFilterConfig(Any.pack(proto), filterContext);
@@ -6425,6 +6431,7 @@ public void givenCongestionInExtProc_whenExtProcBecomesReady_thenTriggersOnReady
64256431
.build())
64266432
.build())
64276433
.setObservabilityMode(true)
6434+
.setProcessingMode(ProcessingMode.getDefaultInstance())
64286435
.build();
64296436
ConfigOrError<ExternalProcessorFilterConfig> configOrError =
64306437
provider.parseFilterConfig(Any.pack(proto), filterContext);
@@ -7605,6 +7612,7 @@ public void givenRequestDrainActive_whenIsReadyCalled_thenReturnsFalse() throws
76057612
.build())
76067613
.build())
76077614
.build())
7615+
.setProcessingMode(ProcessingMode.getDefaultInstance())
76087616
.build();
76097617
ConfigOrError<ExternalProcessorFilterConfig> configOrError =
76107618
provider.parseFilterConfig(Any.pack(proto), filterContext);
@@ -7705,6 +7713,7 @@ public void givenDrainingStream_whenExtProcStreamCompletes_thenOnReady() throws
77057713
.build())
77067714
.build())
77077715
.build())
7716+
.setProcessingMode(ProcessingMode.getDefaultInstance())
77087717
.build();
77097718
ConfigOrError<ExternalProcessorFilterConfig> configOrError =
77107719
provider.parseFilterConfig(Any.pack(proto), filterContext);
@@ -7866,6 +7875,7 @@ public void givenDrainingStream_whenObserverIsNull_thenSendMessageDoesNotQueue()
78667875
.build())
78677876
.build())
78687877
.build())
7878+
.setProcessingMode(ProcessingMode.getDefaultInstance())
78697879
.build();
78707880
ConfigOrError<ExternalProcessorFilterConfig> configOrError =
78717881
provider.parseFilterConfig(Any.pack(proto), filterContext);
@@ -9523,6 +9533,7 @@ public void givenObservabilityTrue_whenExtProcBusy_thenAppRequestsBuffered()
95239533
.build())
95249534
.build())
95259535
.setObservabilityMode(true)
9536+
.setProcessingMode(ProcessingMode.getDefaultInstance())
95269537
.build();
95279538
ConfigOrError<ExternalProcessorFilterConfig> configOrError =
95289539
provider.parseFilterConfig(Any.pack(proto), filterContext);
@@ -9795,6 +9806,7 @@ public void givenBufferedRequests_whenExtProcStreamBecomesReady_thenDataPlaneDra
97959806
.build())
97969807
.build())
97979808
.setObservabilityMode(true)
9809+
.setProcessingMode(ProcessingMode.getDefaultInstance())
97989810
.build();
97999811
ConfigOrError<ExternalProcessorFilterConfig> configOrError =
98009812
provider.parseFilterConfig(Any.pack(proto), filterContext);
@@ -9939,6 +9951,7 @@ public void givenNoRequestDrain_whenExtProcStreamCompletesNormally_thenTreatsAsN
99399951
.build())
99409952
.build())
99419953
.build())
9954+
.setProcessingMode(ProcessingMode.getDefaultInstance())
99429955
.build();
99439956
ConfigOrError<ExternalProcessorFilterConfig> configOrError =
99449957
provider.parseFilterConfig(Any.pack(proto), filterContext);
@@ -9948,30 +9961,30 @@ public void givenNoRequestDrain_whenExtProcStreamCompletesNormally_thenTreatsAsN
99489961
// External Processor Server completes normally without sending request_drain = true
99499962
ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl =
99509963
new ExternalProcessorGrpc.ExternalProcessorImplBase() {
9951-
@Override
9952-
@SuppressWarnings("unchecked")
9953-
public StreamObserver<ProcessingRequest> process(
9954-
final StreamObserver<ProcessingResponse> responseObserver) {
9955-
((ServerCallStreamObserver<ProcessingResponse>) responseObserver).request(100);
9956-
return new StreamObserver<ProcessingRequest>() {
99579964
@Override
9958-
public void onNext(ProcessingRequest request) {
9959-
if (request.hasRequestHeaders()) {
9960-
// Immediately complete the stream from server side without drain
9961-
responseObserver.onCompleted();
9962-
}
9963-
}
9965+
@SuppressWarnings("unchecked")
9966+
public StreamObserver<ProcessingRequest> process(
9967+
final StreamObserver<ProcessingResponse> responseObserver) {
9968+
((ServerCallStreamObserver<ProcessingResponse>) responseObserver).request(100);
9969+
return new StreamObserver<ProcessingRequest>() {
9970+
@Override
9971+
public void onNext(ProcessingRequest request) {
9972+
if (request.hasRequestHeaders()) {
9973+
// Immediately complete the stream from server side without drain
9974+
responseObserver.onCompleted();
9975+
}
9976+
}
99649977

9965-
@Override
9966-
public void onError(Throwable t) {
9967-
}
9978+
@Override
9979+
public void onError(Throwable t) {
9980+
}
99689981

9969-
@Override
9970-
public void onCompleted() {
9982+
@Override
9983+
public void onCompleted() {
9984+
}
9985+
};
99719986
}
99729987
};
9973-
}
9974-
};
99759988
grpcCleanup.register(InProcessServerBuilder.forName(extProcServerName)
99769989
.addService(extProcImpl)
99779990
.directExecutor()
@@ -10393,6 +10406,7 @@ public void givenFailureModeAllowFalse_whenExtProcStreamFails_thenDataPlaneCallC
1039310406
.build())
1039410407
.build())
1039510408
.setFailureModeAllow(false) // Fail Closed
10409+
.setProcessingMode(ProcessingMode.getDefaultInstance())
1039610410
.build();
1039710411
ConfigOrError<ExternalProcessorFilterConfig> configOrError =
1039810412
provider.parseFilterConfig(Any.pack(proto), filterContext);
@@ -10484,6 +10498,7 @@ public void givenFailureModeAllowTrue_whenExtProcStreamFails_thenCallFailsOpen()
1048410498
.build())
1048510499
.build())
1048610500
.setFailureModeAllow(true) // Fail Open
10501+
.setProcessingMode(ProcessingMode.getDefaultInstance())
1048710502
.build();
1048810503
ConfigOrError<ExternalProcessorFilterConfig> configOrError =
1048910504
provider.parseFilterConfig(Any.pack(proto), filterContext);
@@ -10621,6 +10636,7 @@ public void givenObservabilityMode_whenDataPlaneClosed_thenSidecarCloseIsDeferre
1062110636
.setObservabilityMode(true)
1062210637
.setDeferredCloseTimeout(
1062310638
com.google.protobuf.Duration.newBuilder().setSeconds(10).build())
10639+
.setProcessingMode(ProcessingMode.getDefaultInstance())
1062410640
.build();
1062510641
ConfigOrError<ExternalProcessorFilterConfig> configOrError =
1062610642
provider.parseFilterConfig(Any.pack(proto), filterContext);
@@ -11198,6 +11214,7 @@ public void testObservabilityMode_ProceedsWithoutBlockingOnExtProcResponseHeader
1119811214
.build())
1119911215
.build())
1120011216
.setObservabilityMode(true)
11217+
.setProcessingMode(ProcessingMode.getDefaultInstance())
1120111218
.build();
1120211219
ConfigOrError<ExternalProcessorFilterConfig> configOrError =
1120311220
provider.parseFilterConfig(Any.pack(proto), filterContext);
@@ -11325,6 +11342,7 @@ public void givenImmediateResponse_whenReceived_thenDataPlaneCallCancelled()
1132511342
.build())
1132611343
.build())
1132711344
.build())
11345+
.setProcessingMode(ProcessingMode.getDefaultInstance())
1132811346
.build();
1132911347
ConfigOrError<ExternalProcessorFilterConfig> configOrError =
1133011348
provider.parseFilterConfig(Any.pack(proto), filterContext);
@@ -11436,6 +11454,7 @@ public void givenImmediateResponseAndObservabilityTrue_whenReceived_thenImmediat
1143611454
.build())
1143711455
.build())
1143811456
.setObservabilityMode(true)
11457+
.setProcessingMode(ProcessingMode.getDefaultInstance())
1143911458
.build();
1144011459
ConfigOrError<ExternalProcessorFilterConfig> configOrError =
1144111460
provider.parseFilterConfig(Any.pack(proto), filterContext);
@@ -11547,6 +11566,7 @@ public void givenImmediateResponseDisabled_whenReceivedBeforeActivation_thenSide
1154711566
.build())
1154811567
.build())
1154911568
.setDisableImmediateResponse(true)
11569+
.setProcessingMode(ProcessingMode.getDefaultInstance())
1155011570
.build();
1155111571
ConfigOrError<ExternalProcessorFilterConfig> configOrError =
1155211572
provider.parseFilterConfig(Any.pack(proto), filterContext);
@@ -11657,6 +11677,7 @@ public void givenImmediateResponseDisabled_whenReceivedAfterActivation_thenSidec
1165711677
.build())
1165811678
.build())
1165911679
.setDisableImmediateResponse(true)
11680+
.setProcessingMode(ProcessingMode.getDefaultInstance())
1166011681
.build();
1166111682
ConfigOrError<ExternalProcessorFilterConfig> configOrError =
1166211683
provider.parseFilterConfig(Any.pack(proto), filterContext);
@@ -11948,6 +11969,7 @@ public void givenActiveRpc_whenDataPlaneCallCancelled_thenExtProcStreamIsErrored
1194811969
.build())
1194911970
.build())
1195011971
.build())
11972+
.setProcessingMode(ProcessingMode.getDefaultInstance())
1195111973
.build();
1195211974
ConfigOrError<ExternalProcessorFilterConfig> configOrError =
1195311975
provider.parseFilterConfig(Any.pack(proto), filterContext);
@@ -12045,6 +12067,7 @@ public void givenObservabilityModeFalse_whenExtProcBusy_thenIsReadyReturnsFalse(
1204512067
.build())
1204612068
.build())
1204712069
.setObservabilityMode(false)
12070+
.setProcessingMode(ProcessingMode.getDefaultInstance())
1204812071
.build();
1204912072
ConfigOrError<ExternalProcessorFilterConfig> configOrError =
1205012073
provider.parseFilterConfig(Any.pack(proto), filterContext);
@@ -12815,7 +12838,8 @@ public void onCompleted() {
1281512838
assertThat(bodyLatch.await(5, TimeUnit.SECONDS)).isTrue();
1281612839
assertThat(receivedRequests.size()).isAtLeast(2);
1281712840
assertThat(receivedRequests.get(1).hasRequestBody()).isTrue();
12818-
assertThat(receivedRequests.get(1).getRequestBody().getBody().toStringUtf8()).isEqualTo("hello");
12841+
assertThat(receivedRequests.get(1).getRequestBody().getBody().toStringUtf8())
12842+
.isEqualTo("hello");
1281912843

1282012844
proxyCall.cancel("Cleanup", null);
1282112845
channelManager.close();
@@ -12951,7 +12975,8 @@ public void onCompleted() {}
1295112975
assertThat(responseBodyLatch.await(5, TimeUnit.SECONDS)).isTrue();
1295212976
assertThat(receivedRequests.size()).isAtLeast(2);
1295312977
assertThat(receivedRequests.get(1).hasResponseBody()).isTrue();
12954-
assertThat(receivedRequests.get(1).getResponseBody().getBody().toStringUtf8()).isEqualTo("server message");
12978+
assertThat(receivedRequests.get(1).getResponseBody().getBody().toStringUtf8())
12979+
.isEqualTo("server message");
1295512980

1295612981
proxyCall.cancel("Cleanup", null);
1295712982
channelManager.close();

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

Lines changed: 24 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -138,7 +138,8 @@ private ExternalProcessor.Builder createBaseProto(String targetName) {
138138
+ "channel_credentials.insecure.v3.InsecureCredentials")
139139
.build())
140140
.build())
141-
.build());
141+
.build())
142+
.setProcessingMode(ProcessingMode.getDefaultInstance());
142143
}
143144

144145
// --- Category 1: Filter Provider registration based on flag ---
@@ -214,9 +215,30 @@ public void givenUnsupportedResponseBodyMode_whenParsed_thenReturnsError() throw
214215
assertThat(result.errorDetail).contains("Invalid response_body_mode");
215216
}
216217

218+
@Test
219+
public void givenMissingProcessingMode_whenParsed_thenReturnsError() throws Exception {
220+
ExternalProcessor proto = ExternalProcessor.newBuilder()
221+
.setGrpcService(GrpcService.newBuilder()
222+
.setGoogleGrpc(GrpcService.GoogleGrpc.newBuilder()
223+
.setTargetUri("in-process:///" + extProcServerName)
224+
.addChannelCredentialsPlugin(Any.newBuilder()
225+
.setTypeUrl("type.googleapis.com/envoy.extensions.grpc_service."
226+
+ "channel_credentials.insecure.v3.InsecureCredentials")
227+
.build())
228+
.build())
229+
.build())
230+
.build();
231+
232+
ConfigOrError<ExternalProcessorFilterConfig> result =
233+
provider.parseFilterConfig(Any.pack(proto), filterContext);
234+
235+
assertThat(result.errorDetail).contains("Missing required field: processing_mode");
236+
}
237+
217238
@Test
218239
public void givenNonGoogleGrpcService_whenParsed_thenReturnsError() throws Exception {
219240
ExternalProcessor proto = ExternalProcessor.newBuilder()
241+
.setProcessingMode(ProcessingMode.getDefaultInstance())
220242
.setGrpcService(GrpcService.newBuilder().build()) // Invalid: no GoogleGrpc
221243
.build();
222244

@@ -229,6 +251,7 @@ public void givenNonGoogleGrpcService_whenParsed_thenReturnsError() throws Excep
229251
@Test
230252
public void givenInvalidGrpcService_whenParsed_thenReturnsError() throws Exception {
231253
ExternalProcessor proto = ExternalProcessor.newBuilder()
254+
.setProcessingMode(ProcessingMode.getDefaultInstance())
232255
.setGrpcService(GrpcService.newBuilder()
233256
.setGoogleGrpc(GrpcService.GoogleGrpc.newBuilder()
234257
.setTargetUri("in-process:///" + extProcServerName)

0 commit comments

Comments
 (0)