From 9366622b382b2f0759d929295301c5da41504e1e Mon Sep 17 00:00:00 2001 From: Kannan J Date: Fri, 21 Aug 2026 12:04:27 +0000 Subject: [PATCH 1/5] xds: improve test coverage for ExternalProcessorClientInterceptor Remove redundant isCompleted check from isSidecarReady and null checks from recordDuration. TAG=agy CONV=9ea901ca-a127-468c-a836-414f2154bf85 --- .../xds/ExternalProcessorClientInterceptor.java | 17 ++++++----------- 1 file changed, 6 insertions(+), 11 deletions(-) diff --git a/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java b/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java index 80cfb0ac2ce..70038e44baf 100644 --- a/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java +++ b/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java @@ -368,14 +368,12 @@ private void activateCall() { } private void recordDuration(DoubleHistogramMetricInstrument instrument, long durationNanos) { - if (instrument != null) { - double durationSecs = (double) durationNanos / 1_000_000_000.0; - metricsRecorder.recordDoubleHistogram( - instrument, - durationSecs, - ImmutableList.of(target), - ImmutableList.of(backendService)); - } + double durationSecs = (double) durationNanos / 1_000_000_000.0; + metricsRecorder.recordDoubleHistogram( + instrument, + durationSecs, + ImmutableList.of(target), + ImmutableList.of(backendService)); } /** @@ -733,9 +731,6 @@ private void onReadyNotify() { private boolean isSidecarReady() { ExtProcStreamState state = extProcStreamState.get(); - if (state.isCompleted()) { - return true; - } if (state.isDraining()) { return false; } From 6505b9992047735636287cdd563450796a5d8327 Mon Sep 17 00:00:00 2001 From: Kannan J Date: Fri, 4 Sep 2026 06:22:36 +0000 Subject: [PATCH 2/5] xds: drop backend_service metric label from ext_proc client metrics The ext_proc spec has been updated to drop adding the backend_service metric label. TAG=agy CONV=86c74ebe-9fd7-4876-b27c-a4c1b230d346 --- .../ExternalProcessorClientInterceptor.java | 18 ++++++------- ...xternalProcessorClientInterceptorTest.java | 25 ++++++++----------- 2 files changed, 18 insertions(+), 25 deletions(-) diff --git a/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java b/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java index 70038e44baf..61594074d83 100644 --- a/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java +++ b/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java @@ -128,7 +128,7 @@ static synchronized void initMetricInstruments() { "s", LATENCY_BUCKETS, ImmutableList.of("grpc.target"), - ImmutableList.of("grpc.lb.backend_service"), + ImmutableList.of(), true); clientHalfCloseDuration = registry.registerDoubleHistogram( @@ -138,7 +138,7 @@ static synchronized void initMetricInstruments() { "s", LATENCY_BUCKETS, ImmutableList.of("grpc.target"), - ImmutableList.of("grpc.lb.backend_service"), + ImmutableList.of(), true); serverHeadersDuration = registry.registerDoubleHistogram( @@ -148,7 +148,7 @@ static synchronized void initMetricInstruments() { "s", LATENCY_BUCKETS, ImmutableList.of("grpc.target"), - ImmutableList.of("grpc.lb.backend_service"), + ImmutableList.of(), true); serverTrailersDuration = registry.registerDoubleHistogram( @@ -158,7 +158,7 @@ static synchronized void initMetricInstruments() { "s", LATENCY_BUCKETS, ImmutableList.of("grpc.target"), - ImmutableList.of("grpc.lb.backend_service"), + ImmutableList.of(), true); } } @@ -248,8 +248,7 @@ public ClientCall interceptCall( DataPlaneClientCall dataPlaneCall = new DataPlaneClientCall( delayedCall, rawCall, extProcStub, filterConfig, filterConfig.getMutationRulesConfig(), - scheduler, rawMethod, next, metricsRecorder, next.authority(), - callOptions.getOption(XdsNameResolver.CLUSTER_SELECTION_KEY)); + scheduler, rawMethod, next, metricsRecorder, next.authority()); return (ClientCall) (ClientCall) dataPlaneCall; } @@ -294,7 +293,6 @@ private static class DataPlaneClientCall private final Channel channel; private final MetricRecorder metricsRecorder; private final String target; - private final String backendService; private volatile Context callContext = Context.ROOT; private long clientHeadersStartNanos; @@ -327,8 +325,7 @@ protected DataPlaneClientCall( MethodDescriptor method, Channel channel, MetricRecorder metricsRecorder, - String target, - String backendService) { + String target) { super(delayedCall); this.delayedCall = delayedCall; this.rawCall = rawCall; @@ -341,7 +338,6 @@ protected DataPlaneClientCall( this.channel = channel; this.metricsRecorder = checkNotNull(metricsRecorder, "metricsRecorder"); this.target = checkNotNull(target, "target"); - this.backendService = checkNotNull(backendService, "backendService"); } @@ -373,7 +369,7 @@ private void recordDuration(DoubleHistogramMetricInstrument instrument, long dur instrument, durationSecs, ImmutableList.of(target), - ImmutableList.of(backendService)); + ImmutableList.of()); } /** diff --git a/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java b/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java index cd6de138a48..c0b410cf8d2 100644 --- a/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java +++ b/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java @@ -128,8 +128,7 @@ public class ExternalProcessorClientInterceptorTest { private ExternalProcessorFilter.Provider provider; private static final Filter.FilterContext FAKE_CONTEXT = Filter.FilterContext.create( "test-filter", new io.grpc.MetricRecorder() {}); - private static final CallOptions DEFAULT_CALL_OPTIONS = CallOptions.DEFAULT - .withOption(XdsNameResolver.CLUSTER_SELECTION_KEY, "backend-service-metric"); + private static final CallOptions DEFAULT_CALL_OPTIONS = CallOptions.DEFAULT; private Filter.FilterConfigParseContext filterContext; private Bootstrapper.BootstrapInfo bootstrapInfo; private Bootstrapper.ServerInfo serverInfo; @@ -13523,8 +13522,7 @@ public void onCompleted() { ClientCall proxyCall = interceptCall(interceptor, METHOD_SAY_HELLO, - DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()) - .withOption(XdsNameResolver.CLUSTER_SELECTION_KEY, "backend-service-metric"), + DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()), dataPlaneChannel); proxyCall.start(new ClientCall.Listener() { @@ -13555,25 +13553,25 @@ public void onCompleted() { Mockito.eq(ExternalProcessorClientInterceptor.clientHeadersDuration), Mockito.anyDouble(), Mockito.eq(com.google.common.collect.ImmutableList.of("xds:///target-service-metric")), - Mockito.eq(com.google.common.collect.ImmutableList.of("backend-service-metric"))); + Mockito.eq(com.google.common.collect.ImmutableList.of())); Mockito.verify(mockMetricRecorder, Mockito.times(1)).recordDoubleHistogram( Mockito.eq(ExternalProcessorClientInterceptor.clientHalfCloseDuration), Mockito.anyDouble(), Mockito.eq(com.google.common.collect.ImmutableList.of("xds:///target-service-metric")), - Mockito.eq(com.google.common.collect.ImmutableList.of("backend-service-metric"))); + Mockito.eq(com.google.common.collect.ImmutableList.of())); Mockito.verify(mockMetricRecorder, Mockito.times(1)).recordDoubleHistogram( Mockito.eq(ExternalProcessorClientInterceptor.serverHeadersDuration), Mockito.anyDouble(), Mockito.eq(com.google.common.collect.ImmutableList.of("xds:///target-service-metric")), - Mockito.eq(com.google.common.collect.ImmutableList.of("backend-service-metric"))); + Mockito.eq(com.google.common.collect.ImmutableList.of())); Mockito.verify(mockMetricRecorder, Mockito.times(1)).recordDoubleHistogram( Mockito.eq(ExternalProcessorClientInterceptor.serverTrailersDuration), Mockito.anyDouble(), Mockito.eq(com.google.common.collect.ImmutableList.of("xds:///target-service-metric")), - Mockito.eq(com.google.common.collect.ImmutableList.of("backend-service-metric"))); + Mockito.eq(com.google.common.collect.ImmutableList.of())); channelManager.close(); realScheduler.shutdown(); @@ -13690,8 +13688,7 @@ public ServerCall.Listener interceptCall( ClientCall proxyCall = interceptCall(interceptor, METHOD_SAY_HELLO, - DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()) - .withOption(XdsNameResolver.CLUSTER_SELECTION_KEY, "backend-service-metric-fail"), + DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()), dataPlaneChannel); proxyCall.start(new ClientCall.Listener() { @@ -13725,25 +13722,25 @@ public ServerCall.Listener interceptCall( Mockito.eq(ExternalProcessorClientInterceptor.clientHeadersDuration), Mockito.anyDouble(), Mockito.eq(com.google.common.collect.ImmutableList.of("xds:///target-service-metric-fail")), - Mockito.eq(com.google.common.collect.ImmutableList.of("backend-service-metric-fail"))); + Mockito.eq(com.google.common.collect.ImmutableList.of())); Mockito.verify(mockMetricRecorder, Mockito.times(1)).recordDoubleHistogram( Mockito.eq(ExternalProcessorClientInterceptor.clientHalfCloseDuration), Mockito.anyDouble(), Mockito.eq(com.google.common.collect.ImmutableList.of("xds:///target-service-metric-fail")), - Mockito.eq(com.google.common.collect.ImmutableList.of("backend-service-metric-fail"))); + Mockito.eq(com.google.common.collect.ImmutableList.of())); Mockito.verify(mockMetricRecorder, Mockito.times(1)).recordDoubleHistogram( Mockito.eq(ExternalProcessorClientInterceptor.serverHeadersDuration), Mockito.anyDouble(), Mockito.eq(com.google.common.collect.ImmutableList.of("xds:///target-service-metric-fail")), - Mockito.eq(com.google.common.collect.ImmutableList.of("backend-service-metric-fail"))); + Mockito.eq(com.google.common.collect.ImmutableList.of())); Mockito.verify(mockMetricRecorder, Mockito.times(1)).recordDoubleHistogram( Mockito.eq(ExternalProcessorClientInterceptor.serverTrailersDuration), Mockito.anyDouble(), Mockito.eq(com.google.common.collect.ImmutableList.of("xds:///target-service-metric-fail")), - Mockito.eq(com.google.common.collect.ImmutableList.of("backend-service-metric-fail"))); + Mockito.eq(com.google.common.collect.ImmutableList.of())); channelManager.close(); realScheduler.shutdown(); From 61b6fec1c29baa5101a0de9fa35131e514756e04 Mon Sep 17 00:00:00 2001 From: Kannan J Date: Fri, 4 Sep 2026 07:38:47 +0000 Subject: [PATCH 3/5] xds: set end_of_stream on client half-close and handle end_of_stream_without_message per ext_proc spec --- .../ExternalProcessorClientInterceptor.java | 8 +- ...xternalProcessorClientInterceptorTest.java | 543 +++++++++++++++++- 2 files changed, 544 insertions(+), 7 deletions(-) diff --git a/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java b/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java index 61594074d83..167a67a6001 100644 --- a/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java +++ b/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java @@ -886,6 +886,7 @@ public void halfClose() { // Mode is GRPC sendToExtProc(ProcessingRequest.newBuilder() .setRequestBody(HttpBody.newBuilder() + .setEndOfStream(true) .setEndOfStreamWithoutMessage(true) .build()) .build()); @@ -916,10 +917,13 @@ private void handleRequestBodyResponse(BodyResponse bodyResponse) { BodyMutation mutation = bodyResponse.getResponse().getBodyMutation(); if (mutation.hasStreamedResponse()) { StreamedBodyResponse streamed = mutation.getStreamedResponse(); - if (!streamed.getEndOfStreamWithoutMessage()) { + boolean isEndOfStream = streamed.getEndOfStream(); + boolean isEndOfStreamWithoutMessage = + isEndOfStream && streamed.getEndOfStreamWithoutMessage(); + if (!isEndOfStreamWithoutMessage) { super.sendMessage(new KnownLengthInputStream(streamed.getBody())); } - if (streamed.getEndOfStream() || streamed.getEndOfStreamWithoutMessage()) { + if (isEndOfStream) { if (requestSideClosed.compareAndSet(false, true)) { proceedWithHalfClose(); } diff --git a/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java b/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java index c0b410cf8d2..17b3f7d1171 100644 --- a/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java +++ b/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java @@ -2361,6 +2361,7 @@ public void onCompleted() { .isEqualTo("test-message"); assertThat(capturedRequests.get(1).hasRequestHeaders()).isFalse(); assertThat(capturedRequests.get(1).hasRequestBody()).isTrue(); + assertThat(capturedRequests.get(1).getRequestBody().getEndOfStream()).isTrue(); assertThat(capturedRequests.get(1).getRequestBody().getEndOfStreamWithoutMessage()).isTrue(); proxyCall.cancel("Cleanup", null); @@ -2421,11 +2422,13 @@ public void onNext(ProcessingRequest request) { } BodyResponse.Builder bodyResponse = BodyResponse.newBuilder(); if (request.getRequestBody().getBody().isEmpty() + && request.getRequestBody().getEndOfStream() && request.getRequestBody().getEndOfStreamWithoutMessage()) { bodyResponse.setResponse(CommonResponse.newBuilder() .setBodyMutation(BodyMutation.newBuilder() .setStreamedResponse(StreamedBodyResponse.newBuilder() .setEndOfStream(true) + .setEndOfStreamWithoutMessage(true) .build()) .build()) .build()); @@ -2555,6 +2558,7 @@ public void onNext(ProcessingRequest request) { if (request.hasRequestBody()) { BodyResponse.Builder bodyResponse = BodyResponse.newBuilder(); if (request.getRequestBody().getBody().isEmpty() + && request.getRequestBody().getEndOfStream() && request.getRequestBody().getEndOfStreamWithoutMessage()) { bodyResponse.setResponse(CommonResponse.newBuilder() .setBodyMutation(BodyMutation.newBuilder() @@ -2684,6 +2688,7 @@ public void onNext(ProcessingRequest request) { if (request.hasRequestBody()) { BodyResponse.Builder bodyResponse = BodyResponse.newBuilder(); if (request.getRequestBody().getBody().isEmpty() + && request.getRequestBody().getEndOfStream() && request.getRequestBody().getEndOfStreamWithoutMessage()) { bodyResponse.setResponse(CommonResponse.newBuilder() .setBodyMutation(BodyMutation.newBuilder() @@ -2832,6 +2837,7 @@ public void onNext(ProcessingRequest request) { request.getRequestBody().getBody().toStringUtf8().equals("Trigger EOS"); BodyResponse.Builder bodyResponse = BodyResponse.newBuilder(); if (triggerEos || (request.getRequestBody().getBody().isEmpty() + && request.getRequestBody().getEndOfStream() && request.getRequestBody().getEndOfStreamWithoutMessage())) { bodyResponse.setResponse(CommonResponse.newBuilder() .setBodyMutation(BodyMutation.newBuilder() @@ -4741,6 +4747,7 @@ public StreamObserver process( @Override public void onNext(ProcessingRequest request) { if (request.hasRequestBody() + && request.getRequestBody().getEndOfStream() && request.getRequestBody().getEndOfStreamWithoutMessage()) { halfCloseLatch.countDown(); } @@ -4800,6 +4807,93 @@ public void onCompleted() { channelManager.close(); } + @Test + @SuppressWarnings("unchecked") + public void clientInterceptor_halfClose_sendsBothEndOfStreamAndEndOfStreamWithoutMessage() + throws Exception { + String uniqueExtProcServerName = InProcessServerBuilder.generateName(); + String uniqueDataPlaneServerName = InProcessServerBuilder.generateName(); + ExternalProcessor proto = ExternalProcessor.newBuilder() + .setGrpcService(GrpcService.newBuilder() + .setGoogleGrpc(GrpcService.GoogleGrpc.newBuilder() + .setTargetUri("in-process:///" + uniqueExtProcServerName) + .addChannelCredentialsPlugin(Any.newBuilder() + .setTypeUrl("type.googleapis.com/envoy.extensions.grpc_service." + + "channel_credentials.insecure.v3.InsecureCredentials") + .build()) + .build()) + .build()) + .setProcessingMode(ProcessingMode.newBuilder() + .setRequestHeaderMode(ProcessingMode.HeaderSendMode.SKIP) + .setRequestBodyMode(ProcessingMode.BodySendMode.GRPC).build()) + .build(); + ConfigOrError configOrError = + provider.parseFilterConfig(Any.pack(proto), filterContext); + assertThat(configOrError.errorDetail).isNull(); + ExternalProcessorFilterConfig filterConfig = configOrError.config; + + final CountDownLatch halfCloseLatch = new CountDownLatch(1); + final AtomicReference capturedHalfCloseReq = new AtomicReference<>(); + ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl = + new ExternalProcessorGrpc.ExternalProcessorImplBase() { + @Override + public StreamObserver process( + StreamObserver responseObserver) { + ((ServerCallStreamObserver) responseObserver).request(100); + return new StreamObserver() { + @Override + public void onNext(ProcessingRequest request) { + if (request.hasRequestBody() + && request.getRequestBody().getEndOfStream() + && request.getRequestBody().getEndOfStreamWithoutMessage()) { + capturedHalfCloseReq.set(request); + halfCloseLatch.countDown(); + } + } + + @Override + public void onError(Throwable t) {} + + @Override + public void onCompleted() {} + }; + } + }; + grpcCleanup.register(InProcessServerBuilder.forName(uniqueExtProcServerName) + .addService(extProcImpl) + .directExecutor() + .build().start()); + + CachedChannelManager channelManager = new CachedChannelManager(config -> { + return grpcCleanup.register( + InProcessChannelBuilder.forName(uniqueExtProcServerName).directExecutor().build()); + }); + + ExternalProcessorClientInterceptor interceptor = new ExternalProcessorClientInterceptor( + filterConfig, channelManager, scheduler, FAKE_CONTEXT); + + ManagedChannel dataPlaneChannel = grpcCleanup.register( + InProcessChannelBuilder.forName(uniqueDataPlaneServerName).directExecutor().build()); + + CallOptions callOptions = DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()); + ClientCall proxyCall = + interceptCall(interceptor, METHOD_CLIENT_STREAMING, callOptions, dataPlaneChannel); + + proxyCall.start(new ClientCall.Listener() {}, new Metadata()); + proxyCall.request(1); + proxyCall.halfClose(); + + assertThat(halfCloseLatch.await(5, TimeUnit.SECONDS)).isTrue(); + ProcessingRequest req = capturedHalfCloseReq.get(); + assertThat(req).isNotNull(); + assertThat(req.getRequestBody().getEndOfStream()).isTrue(); + assertThat(req.getRequestBody().getEndOfStreamWithoutMessage()).isTrue(); + assertThat(req.getRequestBody().getBody().isEmpty()).isTrue(); + + proxyCall.cancel("Cleanup", null); + channelManager.close(); + } + @Test @SuppressWarnings("unchecked") public void deferredHalfClose_whenExtProcRespondsWithEosWithoutMessage_thenSuperHalfCloseCalled() @@ -4837,7 +4931,8 @@ public StreamObserver process( @Override public void onNext(ProcessingRequest request) { if (request.hasRequestBody()) { - if (request.getRequestBody().getEndOfStreamWithoutMessage()) { + if (request.getRequestBody().getEndOfStream() + && request.getRequestBody().getEndOfStreamWithoutMessage()) { responseObserver.onNext(ProcessingResponse.newBuilder() .setRequestBody(BodyResponse.newBuilder() .setResponse(CommonResponse.newBuilder() @@ -4865,6 +4960,7 @@ public void onNext(ProcessingRequest request) { .setResponse(CommonResponse.newBuilder() .setBodyMutation(BodyMutation.newBuilder() .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setEndOfStream(true) .setEndOfStreamWithoutMessage(true) .build()) .build()) @@ -4966,6 +5062,150 @@ public void halfClose() { channelManager.close(); } + @Test + @SuppressWarnings("unchecked") + public void clientInterceptor_extProcResponse_eosTrue_eosWithoutMessageTrue_bodyIgnored() + throws Exception { + String uniqueExtProcServerName = InProcessServerBuilder.generateName(); + String uniqueDataPlaneServerName = InProcessServerBuilder.generateName(); + ExternalProcessor proto = ExternalProcessor.newBuilder() + .setGrpcService(GrpcService.newBuilder() + .setGoogleGrpc(GrpcService.GoogleGrpc.newBuilder() + .setTargetUri("in-process:///" + uniqueExtProcServerName) + .addChannelCredentialsPlugin(Any.newBuilder() + .setTypeUrl("type.googleapis.com/envoy.extensions.grpc_service." + + "channel_credentials.insecure.v3.InsecureCredentials") + .build()) + .build()) + .build()) + .setProcessingMode(ProcessingMode.newBuilder() + .setRequestHeaderMode(ProcessingMode.HeaderSendMode.SKIP) + .setRequestBodyMode(ProcessingMode.BodySendMode.GRPC).build()) + .build(); + ConfigOrError configOrError = + provider.parseFilterConfig(Any.pack(proto), filterContext); + assertThat(configOrError.errorDetail).isNull(); + ExternalProcessorFilterConfig filterConfig = configOrError.config; + + ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl = + new ExternalProcessorGrpc.ExternalProcessorImplBase() { + @Override + public StreamObserver process( + final StreamObserver responseObserver) { + ((ServerCallStreamObserver) responseObserver).request(100); + return new StreamObserver() { + @Override + public void onNext(ProcessingRequest request) { + if (request.hasRequestBody()) { + if (request.getRequestBody().getEndOfStream() + && request.getRequestBody().getEndOfStreamWithoutMessage()) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setBody(ByteString.copyFromUtf8("unexpected-ignored-body")) + .setEndOfStream(true) + .setEndOfStreamWithoutMessage(true) + .build()) + .build()) + .build()) + .build()) + .build()); + } + } + } + + @Override + public void onError(Throwable t) {} + + @Override + public void onCompleted() { + responseObserver.onCompleted(); + } + }; + } + }; + grpcCleanup.register(InProcessServerBuilder.forName(uniqueExtProcServerName) + .addService(extProcImpl) + .directExecutor() + .build().start()); + + CachedChannelManager channelManager = new CachedChannelManager(config -> { + return grpcCleanup.register( + InProcessChannelBuilder.forName(uniqueExtProcServerName).directExecutor().build()); + }); + + ExternalProcessorClientInterceptor interceptor = new ExternalProcessorClientInterceptor( + filterConfig, channelManager, scheduler, FAKE_CONTEXT); + + final List serverReceivedMessages = new java.util.concurrent.CopyOnWriteArrayList<>(); + MutableHandlerRegistry uniqueRegistry = new MutableHandlerRegistry(); + grpcCleanup.register(InProcessServerBuilder.forName(uniqueDataPlaneServerName) + .fallbackHandlerRegistry(uniqueRegistry) + .directExecutor() + .build().start()); + uniqueRegistry.addService(ServerServiceDefinition.builder("test.TestService") + .addMethod(METHOD_CLIENT_STREAMING, ServerCalls.asyncClientStreamingCall( + new ServerCalls.ClientStreamingMethod() { + @Override + public StreamObserver invoke(StreamObserver responseObserver) { + return new StreamObserver() { + @Override + public void onNext(String value) { + serverReceivedMessages.add(value); + } + + @Override + public void onError(Throwable t) {} + + @Override + public void onCompleted() { + responseObserver.onNext("Ack"); + responseObserver.onCompleted(); + } + }; + } + })) + .build()); + + final java.util.concurrent.CountDownLatch dataPlaneHalfClosedLatch = + new java.util.concurrent.CountDownLatch(1); + ManagedChannel dataPlaneChannel = grpcCleanup.register( + InProcessChannelBuilder.forName(uniqueDataPlaneServerName) + .intercept(new ClientInterceptor() { + @Override + public ClientCall interceptCall( + MethodDescriptor method, CallOptions callOptions, Channel next) { + return new io.grpc.ForwardingClientCall.SimpleForwardingClientCall( + next.newCall(method, callOptions)) { + @Override + public void halfClose() { + dataPlaneHalfClosedLatch.countDown(); + super.halfClose(); + } + }; + } + }) + .directExecutor() + .build()); + + CallOptions callOptions = DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()); + ClientCall proxyCall = + interceptCall(interceptor, METHOD_CLIENT_STREAMING, callOptions, dataPlaneChannel); + proxyCall.start(new ClientCall.Listener() {}, new Metadata()); + + proxyCall.request(1); + proxyCall.halfClose(); + + assertThat(dataPlaneHalfClosedLatch.await(5, java.util.concurrent.TimeUnit.SECONDS)).isTrue(); + // Since both end_of_stream and end_of_stream_without_message are true, the body must NOT be delivered + assertThat(serverReceivedMessages).isEmpty(); + + proxyCall.cancel("Cleanup", null); + channelManager.close(); + } + @Test @SuppressWarnings("unchecked") public void givenDeferredHalfClose_whenExtProcRespondsWithEndOfStream_thenSuperHalfCloseCalled() @@ -5003,7 +5243,8 @@ public StreamObserver process( @Override public void onNext(ProcessingRequest request) { if (request.hasRequestBody()) { - if (request.getRequestBody().getEndOfStreamWithoutMessage()) { + if (request.getRequestBody().getEndOfStream() + && request.getRequestBody().getEndOfStreamWithoutMessage()) { responseObserver.onNext(ProcessingResponse.newBuilder() .setRequestBody(BodyResponse.newBuilder() .setResponse(CommonResponse.newBuilder() @@ -5122,6 +5363,297 @@ public void halfClose() { channelManager.close(); } + @Test + @SuppressWarnings("unchecked") + public void + clientInterceptor_extProcResponse_eosTrue_eosWithoutMessageFalse_emptyBody_deliveredAsMessage() + throws Exception { + String uniqueExtProcServerName = InProcessServerBuilder.generateName(); + String uniqueDataPlaneServerName = InProcessServerBuilder.generateName(); + ExternalProcessor proto = ExternalProcessor.newBuilder() + .setGrpcService(GrpcService.newBuilder() + .setGoogleGrpc(GrpcService.GoogleGrpc.newBuilder() + .setTargetUri("in-process:///" + uniqueExtProcServerName) + .addChannelCredentialsPlugin(Any.newBuilder() + .setTypeUrl("type.googleapis.com/envoy.extensions.grpc_service." + + "channel_credentials.insecure.v3.InsecureCredentials") + .build()) + .build()) + .build()) + .setProcessingMode(ProcessingMode.newBuilder() + .setRequestHeaderMode(ProcessingMode.HeaderSendMode.SKIP) + .setRequestBodyMode(ProcessingMode.BodySendMode.GRPC).build()) + .build(); + ConfigOrError configOrError = + provider.parseFilterConfig(Any.pack(proto), filterContext); + assertThat(configOrError.errorDetail).isNull(); + ExternalProcessorFilterConfig filterConfig = configOrError.config; + + ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl = + new ExternalProcessorGrpc.ExternalProcessorImplBase() { + @Override + public StreamObserver process( + final StreamObserver responseObserver) { + ((ServerCallStreamObserver) responseObserver).request(100); + return new StreamObserver() { + @Override + public void onNext(ProcessingRequest request) { + if (request.hasRequestBody()) { + if (request.getRequestBody().getEndOfStream() + && request.getRequestBody().getEndOfStreamWithoutMessage()) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setBody(ByteString.EMPTY) + .setEndOfStream(true) + .setEndOfStreamWithoutMessage(false) + .build()) + .build()) + .build()) + .build()) + .build()); + } + } + } + + @Override + public void onError(Throwable t) {} + + @Override + public void onCompleted() { + responseObserver.onCompleted(); + } + }; + } + }; + grpcCleanup.register(InProcessServerBuilder.forName(uniqueExtProcServerName) + .addService(extProcImpl) + .directExecutor() + .build().start()); + + CachedChannelManager channelManager = new CachedChannelManager(config -> { + return grpcCleanup.register( + InProcessChannelBuilder.forName(uniqueExtProcServerName).directExecutor().build()); + }); + + ExternalProcessorClientInterceptor interceptor = new ExternalProcessorClientInterceptor( + filterConfig, channelManager, scheduler, FAKE_CONTEXT); + + final List serverReceivedMessages = new java.util.concurrent.CopyOnWriteArrayList<>(); + MutableHandlerRegistry uniqueRegistry = new MutableHandlerRegistry(); + grpcCleanup.register(InProcessServerBuilder.forName(uniqueDataPlaneServerName) + .fallbackHandlerRegistry(uniqueRegistry) + .directExecutor() + .build().start()); + uniqueRegistry.addService(ServerServiceDefinition.builder("test.TestService") + .addMethod(METHOD_CLIENT_STREAMING, ServerCalls.asyncClientStreamingCall( + new ServerCalls.ClientStreamingMethod() { + @Override + public StreamObserver invoke(StreamObserver responseObserver) { + return new StreamObserver() { + @Override + public void onNext(String value) { + serverReceivedMessages.add(value); + } + + @Override + public void onError(Throwable t) {} + + @Override + public void onCompleted() { + responseObserver.onNext("Ack"); + responseObserver.onCompleted(); + } + }; + } + })) + .build()); + + final java.util.concurrent.CountDownLatch dataPlaneHalfClosedLatch = + new java.util.concurrent.CountDownLatch(1); + ManagedChannel dataPlaneChannel = grpcCleanup.register( + InProcessChannelBuilder.forName(uniqueDataPlaneServerName) + .intercept(new ClientInterceptor() { + @Override + public ClientCall interceptCall( + MethodDescriptor method, CallOptions callOptions, Channel next) { + return new io.grpc.ForwardingClientCall.SimpleForwardingClientCall( + next.newCall(method, callOptions)) { + @Override + public void halfClose() { + dataPlaneHalfClosedLatch.countDown(); + super.halfClose(); + } + }; + } + }) + .directExecutor() + .build()); + + CallOptions callOptions = DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()); + ClientCall proxyCall = + interceptCall(interceptor, METHOD_CLIENT_STREAMING, callOptions, dataPlaneChannel); + proxyCall.start(new ClientCall.Listener() {}, new Metadata()); + + proxyCall.request(1); + proxyCall.halfClose(); + + assertThat(dataPlaneHalfClosedLatch.await(5, java.util.concurrent.TimeUnit.SECONDS)).isTrue(); + // Since end_of_stream is true and end_of_stream_without_message is false, empty body is delivered as a message + assertThat(serverReceivedMessages).containsExactly(""); + + proxyCall.cancel("Cleanup", null); + channelManager.close(); + } + + @Test + @SuppressWarnings("unchecked") + public void + clientInterceptor_extProcResponse_eosFalse_eosWithoutMessageTrue_ignoredEosWithoutMessage() + throws Exception { + String uniqueExtProcServerName = InProcessServerBuilder.generateName(); + String uniqueDataPlaneServerName = InProcessServerBuilder.generateName(); + ExternalProcessor proto = ExternalProcessor.newBuilder() + .setGrpcService(GrpcService.newBuilder() + .setGoogleGrpc(GrpcService.GoogleGrpc.newBuilder() + .setTargetUri("in-process:///" + uniqueExtProcServerName) + .addChannelCredentialsPlugin(Any.newBuilder() + .setTypeUrl("type.googleapis.com/envoy.extensions.grpc_service." + + "channel_credentials.insecure.v3.InsecureCredentials") + .build()) + .build()) + .build()) + .setProcessingMode(ProcessingMode.newBuilder() + .setRequestHeaderMode(ProcessingMode.HeaderSendMode.SKIP) + .setRequestBodyMode(ProcessingMode.BodySendMode.GRPC).build()) + .build(); + ConfigOrError configOrError = + provider.parseFilterConfig(Any.pack(proto), filterContext); + assertThat(configOrError.errorDetail).isNull(); + ExternalProcessorFilterConfig filterConfig = configOrError.config; + + ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl = + new ExternalProcessorGrpc.ExternalProcessorImplBase() { + @Override + public StreamObserver process( + final StreamObserver responseObserver) { + ((ServerCallStreamObserver) responseObserver).request(100); + return new StreamObserver() { + @Override + public void onNext(ProcessingRequest request) { + if (request.hasRequestBody()) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setBody(ByteString.copyFromUtf8("normal-body")) + .setEndOfStream(false) + .setEndOfStreamWithoutMessage(true) + .build()) + .build()) + .build()) + .build()) + .build()); + } + } + + @Override + public void onError(Throwable t) {} + + @Override + public void onCompleted() { + responseObserver.onCompleted(); + } + }; + } + }; + grpcCleanup.register(InProcessServerBuilder.forName(uniqueExtProcServerName) + .addService(extProcImpl) + .directExecutor() + .build().start()); + + CachedChannelManager channelManager = new CachedChannelManager(config -> { + return grpcCleanup.register( + InProcessChannelBuilder.forName(uniqueExtProcServerName).directExecutor().build()); + }); + + ExternalProcessorClientInterceptor interceptor = new ExternalProcessorClientInterceptor( + filterConfig, channelManager, scheduler, FAKE_CONTEXT); + + final List serverReceivedMessages = new java.util.concurrent.CopyOnWriteArrayList<>(); + MutableHandlerRegistry uniqueRegistry = new MutableHandlerRegistry(); + grpcCleanup.register(InProcessServerBuilder.forName(uniqueDataPlaneServerName) + .fallbackHandlerRegistry(uniqueRegistry) + .directExecutor() + .build().start()); + final java.util.concurrent.CountDownLatch messageDeliveredLatch = + new java.util.concurrent.CountDownLatch(1); + uniqueRegistry.addService(ServerServiceDefinition.builder("test.TestService") + .addMethod(METHOD_CLIENT_STREAMING, ServerCalls.asyncClientStreamingCall( + new ServerCalls.ClientStreamingMethod() { + @Override + public StreamObserver invoke(StreamObserver responseObserver) { + return new StreamObserver() { + @Override + public void onNext(String value) { + serverReceivedMessages.add(value); + messageDeliveredLatch.countDown(); + } + + @Override + public void onError(Throwable t) {} + + @Override + public void onCompleted() { + responseObserver.onNext("Ack"); + responseObserver.onCompleted(); + } + }; + } + })) + .build()); + + final java.util.concurrent.CountDownLatch dataPlaneHalfClosedLatch = + new java.util.concurrent.CountDownLatch(1); + ManagedChannel dataPlaneChannel = grpcCleanup.register( + InProcessChannelBuilder.forName(uniqueDataPlaneServerName) + .intercept(new ClientInterceptor() { + @Override + public ClientCall interceptCall( + MethodDescriptor method, CallOptions callOptions, Channel next) { + return new io.grpc.ForwardingClientCall.SimpleForwardingClientCall( + next.newCall(method, callOptions)) { + @Override + public void halfClose() { + dataPlaneHalfClosedLatch.countDown(); + super.halfClose(); + } + }; + } + }) + .directExecutor() + .build()); + + CallOptions callOptions = DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()); + ClientCall proxyCall = + interceptCall(interceptor, METHOD_CLIENT_STREAMING, callOptions, dataPlaneChannel); + proxyCall.start(new ClientCall.Listener() {}, new Metadata()); + + proxyCall.request(1); + proxyCall.sendMessage("msg1"); + + assertThat(messageDeliveredLatch.await(5, java.util.concurrent.TimeUnit.SECONDS)).isTrue(); + assertThat(serverReceivedMessages).containsExactly("normal-body"); + // Since end_of_stream was false, end_of_stream_without_message is ignored, so data plane must NOT be half-closed + assertThat(dataPlaneHalfClosedLatch.getCount()).isEqualTo(1); + + proxyCall.cancel("Cleanup", null); + channelManager.close(); + } + @Test @SuppressWarnings("unchecked") public void @@ -5190,6 +5722,7 @@ public void onNext(ProcessingRequest request) { .setResponse(CommonResponse.newBuilder() .setBodyMutation(BodyMutation.newBuilder() .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setEndOfStream(true) .setEndOfStreamWithoutMessage(true) .build()) .build()) @@ -11178,7 +11711,7 @@ public void onNext(ProcessingRequest request) { .build()); } else if (request.hasRequestBody()) { if (request.getRequestBody().getEndOfStream() - || request.getRequestBody().getEndOfStreamWithoutMessage()) { + && request.getRequestBody().getEndOfStreamWithoutMessage()) { receivedPhases.add("REQ_BODY_EOS"); resp.setRequestBody( BodyResponse.newBuilder() @@ -11456,7 +11989,7 @@ public void onNext(ProcessingRequest request) { .build()); } else if (request.hasRequestBody()) { if (request.getRequestBody().getEndOfStream() - || request.getRequestBody().getEndOfStreamWithoutMessage()) { + && request.getRequestBody().getEndOfStreamWithoutMessage()) { receivedPhases.add("REQ_BODY_EOS"); resp.setRequestBody( BodyResponse.newBuilder() @@ -13040,7 +13573,7 @@ public StreamObserver process( public void onNext(ProcessingRequest request) { if (request.hasRequestBody()) { if (request.getRequestBody().getEndOfStream() - || request.getRequestBody().getEndOfStreamWithoutMessage()) { + && request.getRequestBody().getEndOfStreamWithoutMessage()) { // This is the half-close request! observerRef.get().onNext(ProcessingResponse.newBuilder() .setRequestBody(BodyResponse.newBuilder() From 6d2335de71d87bccdc5ed2b343c23b443192923f Mon Sep 17 00:00:00 2001 From: Kannan J Date: Fri, 4 Sep 2026 08:34:21 +0000 Subject: [PATCH 4/5] xds: redo ext_proc client half-close and metric label changes after master merge - Drop backend_service metric label from ExternalProcessorClientInterceptor. - Set end_of_stream on client half-close alongside end_of_stream_without_message. - Handle end_of_stream_without_message per ext_proc spec on StreamedBodyResponse. - Update tests and mocks to reflect the updated half-close semantics. TAG=agy CONV=86c74ebe-9fd7-4876-b27c-a4c1b230d346 --- .../ExternalProcessorClientInterceptor.java | 26 +- ...xternalProcessorClientInterceptorTest.java | 719 ++++++++++++++++-- 2 files changed, 651 insertions(+), 94 deletions(-) diff --git a/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java b/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java index 04830b29fa7..1b68a3fd835 100644 --- a/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java +++ b/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java @@ -132,7 +132,7 @@ static synchronized void initMetricInstruments() { "s", LATENCY_BUCKETS, ImmutableList.of("grpc.target"), - ImmutableList.of("grpc.lb.backend_service"), + ImmutableList.of(), true); clientHalfCloseDuration = registry.registerDoubleHistogram( @@ -142,7 +142,7 @@ static synchronized void initMetricInstruments() { "s", LATENCY_BUCKETS, ImmutableList.of("grpc.target"), - ImmutableList.of("grpc.lb.backend_service"), + ImmutableList.of(), true); serverHeadersDuration = registry.registerDoubleHistogram( @@ -152,7 +152,7 @@ static synchronized void initMetricInstruments() { "s", LATENCY_BUCKETS, ImmutableList.of("grpc.target"), - ImmutableList.of("grpc.lb.backend_service"), + ImmutableList.of(), true); serverTrailersDuration = registry.registerDoubleHistogram( @@ -162,7 +162,7 @@ static synchronized void initMetricInstruments() { "s", LATENCY_BUCKETS, ImmutableList.of("grpc.target"), - ImmutableList.of("grpc.lb.backend_service"), + ImmutableList.of(), true); } } @@ -246,8 +246,7 @@ public ClientCall interceptCall( DataPlaneClientCall dataPlaneCall = new DataPlaneClientCall( delayedCall, rawCall, extProcStub, filterConfig, filterConfig.getMutationRulesConfig(), - scheduler, rawMethod, next, metricsRecorder, next.authority(), - callOptions.getOption(XdsNameResolver.CLUSTER_SELECTION_KEY)); + scheduler, rawMethod, next, metricsRecorder, next.authority()); return (ClientCall) (ClientCall) dataPlaneCall; } @@ -340,7 +339,6 @@ private static class DataPlaneClientCall private final Channel channel; private final MetricRecorder metricsRecorder; private final String target; - private final String backendService; private volatile Context callContext = Context.ROOT; private volatile long clientHeadersStartNanos; @@ -374,8 +372,7 @@ protected DataPlaneClientCall( MethodDescriptor method, Channel channel, MetricRecorder metricsRecorder, - String target, - String backendService) { + String target) { super(delayedCall); this.delayedCall = delayedCall; this.rawCall = rawCall; @@ -388,7 +385,6 @@ protected DataPlaneClientCall( this.channel = channel; this.metricsRecorder = checkNotNull(metricsRecorder, "metricsRecorder"); this.target = checkNotNull(target, "target"); - this.backendService = checkNotNull(backendService, "backendService"); } private boolean activateCall() { @@ -420,7 +416,7 @@ private void recordDuration(DoubleHistogramMetricInstrument instrument, long dur instrument, durationSecs, ImmutableList.of(target), - ImmutableList.of(backendService)); + ImmutableList.of()); } } @@ -1124,6 +1120,7 @@ public void halfClose() { ProcessingRequest.Builder builder = ProcessingRequest.newBuilder() .setRequestBody(HttpBody.newBuilder() + .setEndOfStream(true) .setEndOfStreamWithoutMessage(true) .build()); mergeAccumulatedWindowUpdates(builder); @@ -1156,7 +1153,10 @@ private void handleRequestBodyResponse(BodyResponse bodyResponse) { BodyMutation mutation = bodyResponse.getResponse().getBodyMutation(); if (mutation.hasStreamedResponse()) { StreamedBodyResponse streamed = mutation.getStreamedResponse(); - if (!streamed.getEndOfStreamWithoutMessage()) { + boolean isEndOfStream = streamed.getEndOfStream(); + boolean isEndOfStreamWithoutMessage = + isEndOfStream && streamed.getEndOfStreamWithoutMessage(); + if (!isEndOfStreamWithoutMessage) { ByteString body = streamed.getBody(); boolean sendImmediately = false; synchronized (streamLock) { @@ -1173,7 +1173,7 @@ private void handleRequestBodyResponse(BodyResponse bodyResponse) { trySendAccumulatedWindowUpdates(); } } - if (streamed.getEndOfStream() || streamed.getEndOfStreamWithoutMessage()) { + if (isEndOfStream) { synchronized (streamLock) { if (pendingUpstreamBodyMessages.isEmpty()) { if (requestSideClosed.compareAndSet(false, true)) { diff --git a/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java b/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java index 0701b670b5d..c9d051edb90 100644 --- a/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java +++ b/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java @@ -130,8 +130,7 @@ public class ExternalProcessorClientInterceptorTest { private ExternalProcessorFilter.Provider provider; private static final Filter.FilterContext FAKE_CONTEXT = Filter.FilterContext.create( "test-filter", new io.grpc.MetricRecorder() {}); - private static final CallOptions DEFAULT_CALL_OPTIONS = CallOptions.DEFAULT - .withOption(XdsNameResolver.CLUSTER_SELECTION_KEY, "backend-service-metric"); + private static final CallOptions DEFAULT_CALL_OPTIONS = CallOptions.DEFAULT; private Filter.FilterConfigParseContext filterContext; private Bootstrapper.BootstrapInfo bootstrapInfo; private Bootstrapper.ServerInfo serverInfo; @@ -2363,6 +2362,7 @@ public void onCompleted() { .isEqualTo("test-message"); assertThat(capturedRequests.get(1).hasRequestHeaders()).isFalse(); assertThat(capturedRequests.get(1).hasRequestBody()).isTrue(); + assertThat(capturedRequests.get(1).getRequestBody().getEndOfStream()).isTrue(); assertThat(capturedRequests.get(1).getRequestBody().getEndOfStreamWithoutMessage()).isTrue(); proxyCall.cancel("Cleanup", null); @@ -2423,11 +2423,13 @@ public void onNext(ProcessingRequest request) { } BodyResponse.Builder bodyResponse = BodyResponse.newBuilder(); if (request.getRequestBody().getBody().isEmpty() + && request.getRequestBody().getEndOfStream() && request.getRequestBody().getEndOfStreamWithoutMessage()) { bodyResponse.setResponse(CommonResponse.newBuilder() .setBodyMutation(BodyMutation.newBuilder() .setStreamedResponse(StreamedBodyResponse.newBuilder() .setEndOfStream(true) + .setEndOfStreamWithoutMessage(true) .build()) .build()) .build()); @@ -2557,6 +2559,7 @@ public void onNext(ProcessingRequest request) { if (request.hasRequestBody()) { BodyResponse.Builder bodyResponse = BodyResponse.newBuilder(); if (request.getRequestBody().getBody().isEmpty() + && request.getRequestBody().getEndOfStream() && request.getRequestBody().getEndOfStreamWithoutMessage()) { bodyResponse.setResponse(CommonResponse.newBuilder() .setBodyMutation(BodyMutation.newBuilder() @@ -2686,6 +2689,7 @@ public void onNext(ProcessingRequest request) { if (request.hasRequestBody()) { BodyResponse.Builder bodyResponse = BodyResponse.newBuilder(); if (request.getRequestBody().getBody().isEmpty() + && request.getRequestBody().getEndOfStream() && request.getRequestBody().getEndOfStreamWithoutMessage()) { bodyResponse.setResponse(CommonResponse.newBuilder() .setBodyMutation(BodyMutation.newBuilder() @@ -2834,6 +2838,7 @@ public void onNext(ProcessingRequest request) { request.getRequestBody().getBody().toStringUtf8().equals("Trigger EOS"); BodyResponse.Builder bodyResponse = BodyResponse.newBuilder(); if (triggerEos || (request.getRequestBody().getBody().isEmpty() + && request.getRequestBody().getEndOfStream() && request.getRequestBody().getEndOfStreamWithoutMessage())) { bodyResponse.setResponse(CommonResponse.newBuilder() .setBodyMutation(BodyMutation.newBuilder() @@ -4797,6 +4802,109 @@ public void onCompleted() { channelManager.close(); } + @Test + @SuppressWarnings("unchecked") + public void clientInterceptor_halfClose_sendsBothEndOfStreamAndEndOfStreamWithoutMessage() + throws Exception { + String uniqueExtProcServerName = InProcessServerBuilder.generateName(); + String uniqueDataPlaneServerName = InProcessServerBuilder.generateName(); + ExternalProcessor proto = ExternalProcessor.newBuilder() + .setGrpcService(GrpcService.newBuilder() + .setGoogleGrpc(GrpcService.GoogleGrpc.newBuilder() + .setTargetUri("in-process:///" + uniqueExtProcServerName) + .addChannelCredentialsPlugin(Any.newBuilder() + .setTypeUrl("type.googleapis.com/envoy.extensions.grpc_service." + + "channel_credentials.insecure.v3.InsecureCredentials") + .build()) + .build()) + .build()) + .setProcessingMode(ProcessingMode.newBuilder() + .setRequestHeaderMode(ProcessingMode.HeaderSendMode.SKIP) + .setRequestBodyMode(ProcessingMode.BodySendMode.GRPC).build()) + .build(); + ConfigOrError configOrError = + provider.parseFilterConfig(Any.pack(proto), filterContext); + assertThat(configOrError.errorDetail).isNull(); + ExternalProcessorFilterConfig filterConfig = configOrError.config; + + final CountDownLatch halfCloseLatch = new CountDownLatch(1); + final AtomicReference capturedHalfCloseReq = new AtomicReference<>(); + ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl = + new ExternalProcessorGrpc.ExternalProcessorImplBase() { + @Override + public StreamObserver process( + StreamObserver responseObserver) { + ((ServerCallStreamObserver) responseObserver).request(100); + return new StreamObserver() { + @Override + public void onNext(ProcessingRequest request) { + if (request.hasRequestBody() + && request.getRequestBody().getEndOfStream() + && request.getRequestBody().getEndOfStreamWithoutMessage()) { + capturedHalfCloseReq.set(request); + halfCloseLatch.countDown(); + } + } + + @Override + public void onError(Throwable t) {} + + @Override + public void onCompleted() {} + }; + } + }; + grpcCleanup.register(InProcessServerBuilder.forName(uniqueExtProcServerName) + .addService(extProcImpl) + .directExecutor() + .build().start()); + + CachedChannelManager channelManager = new CachedChannelManager(config -> { + return grpcCleanup.register( + InProcessChannelBuilder.forName(uniqueExtProcServerName).directExecutor().build()); + }); + + ExternalProcessorClientInterceptor interceptor = new ExternalProcessorClientInterceptor( + filterConfig, channelManager, scheduler, FAKE_CONTEXT); + + final CountDownLatch dataPlaneHalfCloseLatch = new CountDownLatch(1); + ManagedChannel dataPlaneChannel = grpcCleanup.register( + InProcessChannelBuilder.forName(uniqueDataPlaneServerName) + .intercept(new ClientInterceptor() { + @Override + public ClientCall interceptCall( + MethodDescriptor method, CallOptions callOptions, Channel next) { + return new io.grpc.ForwardingClientCall.SimpleForwardingClientCall( + next.newCall(method, callOptions)) { + @Override + public void halfClose() { + dataPlaneHalfCloseLatch.countDown(); + super.halfClose(); + } + }; + } + }) + .directExecutor() + .build()); + + CallOptions callOptions = DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()); + ClientCall proxyCall = + interceptCall(interceptor, METHOD_SAY_HELLO, callOptions, dataPlaneChannel); + proxyCall.start(new ClientCall.Listener() {}, new Metadata()); + + proxyCall.halfClose(); + + // Verify sidecar received both end_of_stream AND end_of_stream_without_message + assertThat(halfCloseLatch.await(5, TimeUnit.SECONDS)).isTrue(); + ProcessingRequest req = capturedHalfCloseReq.get(); + assertThat(req.getRequestBody().getEndOfStream()).isTrue(); + assertThat(req.getRequestBody().getEndOfStreamWithoutMessage()).isTrue(); + assertThat(req.getRequestBody().getBody().isEmpty()).isTrue(); + + proxyCall.cancel("Cleanup", null); + channelManager.close(); + } + @Test @SuppressWarnings("unchecked") public void deferredHalfClose_whenExtProcRespondsWithEosWithoutMessage_thenSuperHalfCloseCalled() @@ -4834,7 +4942,320 @@ public StreamObserver process( @Override public void onNext(ProcessingRequest request) { if (request.hasRequestBody()) { - if (request.getRequestBody().getEndOfStreamWithoutMessage()) { + if (request.getRequestBody().getEndOfStream() + && request.getRequestBody().getEndOfStreamWithoutMessage()) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setBody(ByteString.copyFromUtf8("mutated1")) + .build()) + .build()) + .build()) + .build()) + .build()); + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setBody(ByteString.copyFromUtf8("mutated2")) + .build()) + .build()) + .build()) + .build()) + .build()); + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setEndOfStream(true) + .setEndOfStreamWithoutMessage(true) + .build()) + .build()) + .build()) + .build()) + .build()); + } + } + } + + @Override + public void onError(Throwable t) { + } + + @Override + public void onCompleted() { + responseObserver.onCompleted(); + } + }; + } + }; + grpcCleanup.register(InProcessServerBuilder.forName(uniqueExtProcServerName) + .addService(extProcImpl) + .directExecutor() + .build().start()); + + CachedChannelManager channelManager = new CachedChannelManager(config -> { + return grpcCleanup.register( + InProcessChannelBuilder.forName(uniqueExtProcServerName).directExecutor().build()); + }); + + ExternalProcessorClientInterceptor interceptor = new ExternalProcessorClientInterceptor( + filterConfig, channelManager, scheduler, FAKE_CONTEXT); + + final List serverReceivedMessages = new java.util.concurrent.CopyOnWriteArrayList<>(); + MutableHandlerRegistry uniqueRegistry = new MutableHandlerRegistry(); + grpcCleanup.register(InProcessServerBuilder.forName(uniqueDataPlaneServerName) + .fallbackHandlerRegistry(uniqueRegistry) + .directExecutor() + .build().start()); + uniqueRegistry.addService(ServerServiceDefinition.builder("test.TestService") + .addMethod(METHOD_CLIENT_STREAMING, ServerCalls.asyncClientStreamingCall( + new ServerCalls.ClientStreamingMethod() { + @Override + public StreamObserver invoke(StreamObserver responseObserver) { + return new StreamObserver() { + @Override + public void onNext(String value) { + serverReceivedMessages.add(value); + } + + @Override + public void onError(Throwable t) { + } + + @Override + public void onCompleted() { + responseObserver.onNext("Ack"); + responseObserver.onCompleted(); + } + }; + } + })) + .build()); + + final java.util.concurrent.CountDownLatch dataPlaneHalfClosedLatch = + new java.util.concurrent.CountDownLatch(1); + ManagedChannel dataPlaneChannel = grpcCleanup.register( + InProcessChannelBuilder.forName(uniqueDataPlaneServerName) + .intercept(new ClientInterceptor() { + @Override + public ClientCall interceptCall( + MethodDescriptor method, CallOptions callOptions, Channel next) { + return new io.grpc.ForwardingClientCall.SimpleForwardingClientCall( + next.newCall(method, callOptions)) { + @Override + public void halfClose() { + dataPlaneHalfClosedLatch.countDown(); + super.halfClose(); + } + }; + } + }) + .directExecutor() + .build()); + + CallOptions callOptions = DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()); + ClientCall proxyCall = + interceptCall(interceptor, METHOD_CLIENT_STREAMING, callOptions, dataPlaneChannel); + proxyCall.start(new ClientCall.Listener() {}, new Metadata()); + + proxyCall.request(1); + proxyCall.halfClose(); + + assertThat(dataPlaneHalfClosedLatch.await(5, java.util.concurrent.TimeUnit.SECONDS)).isTrue(); + assertThat(serverReceivedMessages).containsExactly("mutated1", "mutated2"); + + proxyCall.cancel("Cleanup", null); + channelManager.close(); + } + + @Test + @SuppressWarnings("unchecked") + public void clientInterceptor_extProcResponse_eosTrue_eosWithoutMessageTrue_bodyIgnored() + throws Exception { + String uniqueExtProcServerName = InProcessServerBuilder.generateName(); + String uniqueDataPlaneServerName = InProcessServerBuilder.generateName(); + ExternalProcessor proto = ExternalProcessor.newBuilder() + .setGrpcService(GrpcService.newBuilder() + .setGoogleGrpc(GrpcService.GoogleGrpc.newBuilder() + .setTargetUri("in-process:///" + uniqueExtProcServerName) + .addChannelCredentialsPlugin(Any.newBuilder() + .setTypeUrl("type.googleapis.com/envoy.extensions.grpc_service." + + "channel_credentials.insecure.v3.InsecureCredentials") + .build()) + .build()) + .build()) + .setProcessingMode(ProcessingMode.newBuilder() + .setRequestHeaderMode(ProcessingMode.HeaderSendMode.SKIP) + .setRequestBodyMode(ProcessingMode.BodySendMode.GRPC).build()) + .build(); + ConfigOrError configOrError = + provider.parseFilterConfig(Any.pack(proto), filterContext); + assertThat(configOrError.errorDetail).isNull(); + ExternalProcessorFilterConfig filterConfig = configOrError.config; + + ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl = + new ExternalProcessorGrpc.ExternalProcessorImplBase() { + @Override + public StreamObserver process( + final StreamObserver responseObserver) { + ((ServerCallStreamObserver) responseObserver).request(100); + return new StreamObserver() { + @Override + public void onNext(ProcessingRequest request) { + if (request.hasRequestBody()) { + if (request.getRequestBody().getEndOfStream() + && request.getRequestBody().getEndOfStreamWithoutMessage()) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setBody(ByteString.copyFromUtf8("unexpected-ignored-body")) + .setEndOfStream(true) + .setEndOfStreamWithoutMessage(true) + .build()) + .build()) + .build()) + .build()) + .build()); + } + } + } + + @Override + public void onError(Throwable t) {} + + @Override + public void onCompleted() { + responseObserver.onCompleted(); + } + }; + } + }; + grpcCleanup.register(InProcessServerBuilder.forName(uniqueExtProcServerName) + .addService(extProcImpl) + .directExecutor() + .build().start()); + + CachedChannelManager channelManager = new CachedChannelManager(config -> { + return grpcCleanup.register( + InProcessChannelBuilder.forName(uniqueExtProcServerName).directExecutor().build()); + }); + + ExternalProcessorClientInterceptor interceptor = new ExternalProcessorClientInterceptor( + filterConfig, channelManager, scheduler, FAKE_CONTEXT); + + final List serverReceivedMessages = new java.util.concurrent.CopyOnWriteArrayList<>(); + MutableHandlerRegistry uniqueRegistry = new MutableHandlerRegistry(); + grpcCleanup.register(InProcessServerBuilder.forName(uniqueDataPlaneServerName) + .fallbackHandlerRegistry(uniqueRegistry) + .directExecutor() + .build().start()); + uniqueRegistry.addService(ServerServiceDefinition.builder("test.TestService") + .addMethod(METHOD_CLIENT_STREAMING, ServerCalls.asyncClientStreamingCall( + new ServerCalls.ClientStreamingMethod() { + @Override + public StreamObserver invoke(StreamObserver responseObserver) { + return new StreamObserver() { + @Override + public void onNext(String value) { + serverReceivedMessages.add(value); + } + + @Override + public void onError(Throwable t) {} + + @Override + public void onCompleted() { + responseObserver.onNext("Ack"); + responseObserver.onCompleted(); + } + }; + } + })) + .build()); + + final java.util.concurrent.CountDownLatch dataPlaneHalfClosedLatch = + new java.util.concurrent.CountDownLatch(1); + ManagedChannel dataPlaneChannel = grpcCleanup.register( + InProcessChannelBuilder.forName(uniqueDataPlaneServerName) + .intercept(new ClientInterceptor() { + @Override + public ClientCall interceptCall( + MethodDescriptor method, CallOptions callOptions, Channel next) { + return new io.grpc.ForwardingClientCall.SimpleForwardingClientCall( + next.newCall(method, callOptions)) { + @Override + public void halfClose() { + dataPlaneHalfClosedLatch.countDown(); + super.halfClose(); + } + }; + } + }) + .directExecutor() + .build()); + + CallOptions callOptions = DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()); + ClientCall proxyCall = + interceptCall(interceptor, METHOD_CLIENT_STREAMING, callOptions, dataPlaneChannel); + proxyCall.start(new ClientCall.Listener() {}, new Metadata()); + + proxyCall.request(1); + proxyCall.halfClose(); + + assertThat(dataPlaneHalfClosedLatch.await(5, java.util.concurrent.TimeUnit.SECONDS)).isTrue(); + // Since both end_of_stream and end_of_stream_without_message are true, the body must NOT be delivered + assertThat(serverReceivedMessages).isEmpty(); + + proxyCall.cancel("Cleanup", null); + channelManager.close(); + } + + @Test + @SuppressWarnings("unchecked") + public void givenDeferredHalfClose_whenExtProcRespondsWithEndOfStream_thenSuperHalfCloseCalled() + throws Exception { + String uniqueExtProcServerName = InProcessServerBuilder.generateName(); + String uniqueDataPlaneServerName = InProcessServerBuilder.generateName(); + ExternalProcessor proto = ExternalProcessor.newBuilder() + .setGrpcService(GrpcService.newBuilder() + .setGoogleGrpc(GrpcService.GoogleGrpc.newBuilder() + .setTargetUri("in-process:///" + uniqueExtProcServerName) + .addChannelCredentialsPlugin(Any.newBuilder() + .setTypeUrl("type.googleapis.com/envoy.extensions.grpc_service." + + "channel_credentials.insecure.v3.InsecureCredentials") + .build()) + .build()) + .build()) + .setProcessingMode(ProcessingMode.newBuilder() + .setRequestHeaderMode(ProcessingMode.HeaderSendMode.SKIP) + .setRequestBodyMode(ProcessingMode.BodySendMode.GRPC).build()) + .build(); + ConfigOrError configOrError = + provider.parseFilterConfig(Any.pack(proto), filterContext); + assertThat(configOrError.errorDetail).isNull(); + ExternalProcessorFilterConfig filterConfig = configOrError.config; + + // External Processor Server + ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl; + extProcImpl = new ExternalProcessorGrpc.ExternalProcessorImplBase() { + @Override + @SuppressWarnings("unchecked") + public StreamObserver process( + final StreamObserver responseObserver) { + ((ServerCallStreamObserver) responseObserver).request(100); + return new StreamObserver() { + @Override + public void onNext(ProcessingRequest request) { + if (request.hasRequestBody()) { + if (request.getRequestBody().getEndOfStream() + && request.getRequestBody().getEndOfStreamWithoutMessage()) { responseObserver.onNext(ProcessingResponse.newBuilder() .setRequestBody(BodyResponse.newBuilder() .setResponse(CommonResponse.newBuilder() @@ -4852,17 +5273,7 @@ public void onNext(ProcessingRequest request) { .setBodyMutation(BodyMutation.newBuilder() .setStreamedResponse(StreamedBodyResponse.newBuilder() .setBody(ByteString.copyFromUtf8("mutated2")) - .build()) - .build()) - .build()) - .build()) - .build()); - responseObserver.onNext(ProcessingResponse.newBuilder() - .setRequestBody(BodyResponse.newBuilder() - .setResponse(CommonResponse.newBuilder() - .setBodyMutation(BodyMutation.newBuilder() - .setStreamedResponse(StreamedBodyResponse.newBuilder() - .setEndOfStreamWithoutMessage(true) + .setEndOfStream(true) .build()) .build()) .build()) @@ -4965,7 +5376,8 @@ public void halfClose() { @Test @SuppressWarnings("unchecked") - public void givenDeferredHalfClose_whenExtProcRespondsWithEndOfStream_thenSuperHalfCloseCalled() + public void + clientInterceptor_extProcResponse_eosTrue_eosWithoutMessageFalse_emptyBody_deliveredAsMessage() throws Exception { String uniqueExtProcServerName = InProcessServerBuilder.generateName(); String uniqueDataPlaneServerName = InProcessServerBuilder.generateName(); @@ -4974,8 +5386,8 @@ public void givenDeferredHalfClose_whenExtProcRespondsWithEndOfStream_thenSuperH .setGoogleGrpc(GrpcService.GoogleGrpc.newBuilder() .setTargetUri("in-process:///" + uniqueExtProcServerName) .addChannelCredentialsPlugin(Any.newBuilder() - .setTypeUrl("type.googleapis.com/envoy.extensions.grpc_service." - + "channel_credentials.insecure.v3.InsecureCredentials") + .setTypeUrl("type.googleapis.com/envoy.extensions.grpc_service." + + "channel_credentials.insecure.v3.InsecureCredentials") .build()) .build()) .build()) @@ -4988,57 +5400,45 @@ public void givenDeferredHalfClose_whenExtProcRespondsWithEndOfStream_thenSuperH assertThat(configOrError.errorDetail).isNull(); ExternalProcessorFilterConfig filterConfig = configOrError.config; - // External Processor Server - ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl; - extProcImpl = new ExternalProcessorGrpc.ExternalProcessorImplBase() { - @Override - @SuppressWarnings("unchecked") - public StreamObserver process( - final StreamObserver responseObserver) { - ((ServerCallStreamObserver) responseObserver).request(100); - return new StreamObserver() { + ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl = + new ExternalProcessorGrpc.ExternalProcessorImplBase() { @Override - public void onNext(ProcessingRequest request) { - if (request.hasRequestBody()) { - if (request.getRequestBody().getEndOfStreamWithoutMessage()) { - responseObserver.onNext(ProcessingResponse.newBuilder() - .setRequestBody(BodyResponse.newBuilder() - .setResponse(CommonResponse.newBuilder() - .setBodyMutation(BodyMutation.newBuilder() - .setStreamedResponse(StreamedBodyResponse.newBuilder() - .setBody(ByteString.copyFromUtf8("mutated1")) - .build()) - .build()) - .build()) - .build()) - .build()); - responseObserver.onNext(ProcessingResponse.newBuilder() - .setRequestBody(BodyResponse.newBuilder() - .setResponse(CommonResponse.newBuilder() - .setBodyMutation(BodyMutation.newBuilder() - .setStreamedResponse(StreamedBodyResponse.newBuilder() - .setBody(ByteString.copyFromUtf8("mutated2")) - .setEndOfStream(true) + public StreamObserver process( + final StreamObserver responseObserver) { + ((ServerCallStreamObserver) responseObserver).request(100); + return new StreamObserver() { + @Override + public void onNext(ProcessingRequest request) { + if (request.hasRequestBody()) { + if (request.getRequestBody().getEndOfStream() + && request.getRequestBody().getEndOfStreamWithoutMessage()) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setBody(ByteString.EMPTY) + .setEndOfStream(true) + .setEndOfStreamWithoutMessage(false) + .build()) .build()) .build()) .build()) - .build()) - .build()); + .build()); + } + } } - } - } - @Override - public void onError(Throwable t) { - } + @Override + public void onError(Throwable t) {} - @Override - public void onCompleted() { - responseObserver.onCompleted(); + @Override + public void onCompleted() { + responseObserver.onCompleted(); + } + }; } }; - } - }; grpcCleanup.register(InProcessServerBuilder.forName(uniqueExtProcServerName) .addService(extProcImpl) .directExecutor() @@ -5070,8 +5470,7 @@ public void onNext(String value) { } @Override - public void onError(Throwable t) { - } + public void onError(Throwable t) {} @Override public void onCompleted() { @@ -5113,7 +5512,154 @@ public void halfClose() { proxyCall.halfClose(); assertThat(dataPlaneHalfClosedLatch.await(5, java.util.concurrent.TimeUnit.SECONDS)).isTrue(); - assertThat(serverReceivedMessages).containsExactly("mutated1", "mutated2"); + // Since end_of_stream is true and end_of_stream_without_message is false, empty body is delivered as a message + assertThat(serverReceivedMessages).containsExactly(""); + + proxyCall.cancel("Cleanup", null); + channelManager.close(); + } + + @Test + @SuppressWarnings("unchecked") + public void + clientInterceptor_extProcResponse_eosFalse_eosWithoutMessageTrue_ignoredEosWithoutMessage() + throws Exception { + String uniqueExtProcServerName = InProcessServerBuilder.generateName(); + String uniqueDataPlaneServerName = InProcessServerBuilder.generateName(); + ExternalProcessor proto = ExternalProcessor.newBuilder() + .setGrpcService(GrpcService.newBuilder() + .setGoogleGrpc(GrpcService.GoogleGrpc.newBuilder() + .setTargetUri("in-process:///" + uniqueExtProcServerName) + .addChannelCredentialsPlugin(Any.newBuilder() + .setTypeUrl("type.googleapis.com/envoy.extensions.grpc_service." + + "channel_credentials.insecure.v3.InsecureCredentials") + .build()) + .build()) + .build()) + .setProcessingMode(ProcessingMode.newBuilder() + .setRequestHeaderMode(ProcessingMode.HeaderSendMode.SKIP) + .setRequestBodyMode(ProcessingMode.BodySendMode.GRPC).build()) + .build(); + ConfigOrError configOrError = + provider.parseFilterConfig(Any.pack(proto), filterContext); + assertThat(configOrError.errorDetail).isNull(); + ExternalProcessorFilterConfig filterConfig = configOrError.config; + + ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl = + new ExternalProcessorGrpc.ExternalProcessorImplBase() { + @Override + public StreamObserver process( + final StreamObserver responseObserver) { + ((ServerCallStreamObserver) responseObserver).request(100); + return new StreamObserver() { + @Override + public void onNext(ProcessingRequest request) { + if (request.hasRequestBody()) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setBody(ByteString.copyFromUtf8("normal-body")) + .setEndOfStream(false) + .setEndOfStreamWithoutMessage(true) + .build()) + .build()) + .build()) + .build()) + .build()); + } + } + + @Override + public void onError(Throwable t) {} + + @Override + public void onCompleted() { + responseObserver.onCompleted(); + } + }; + } + }; + grpcCleanup.register(InProcessServerBuilder.forName(uniqueExtProcServerName) + .addService(extProcImpl) + .directExecutor() + .build().start()); + + CachedChannelManager channelManager = new CachedChannelManager(config -> { + return grpcCleanup.register( + InProcessChannelBuilder.forName(uniqueExtProcServerName).directExecutor().build()); + }); + + ExternalProcessorClientInterceptor interceptor = new ExternalProcessorClientInterceptor( + filterConfig, channelManager, scheduler, FAKE_CONTEXT); + + final List serverReceivedMessages = new java.util.concurrent.CopyOnWriteArrayList<>(); + MutableHandlerRegistry uniqueRegistry = new MutableHandlerRegistry(); + grpcCleanup.register(InProcessServerBuilder.forName(uniqueDataPlaneServerName) + .fallbackHandlerRegistry(uniqueRegistry) + .directExecutor() + .build().start()); + final java.util.concurrent.CountDownLatch messageDeliveredLatch = + new java.util.concurrent.CountDownLatch(1); + uniqueRegistry.addService(ServerServiceDefinition.builder("test.TestService") + .addMethod(METHOD_CLIENT_STREAMING, ServerCalls.asyncClientStreamingCall( + new ServerCalls.ClientStreamingMethod() { + @Override + public StreamObserver invoke(StreamObserver responseObserver) { + return new StreamObserver() { + @Override + public void onNext(String value) { + serverReceivedMessages.add(value); + messageDeliveredLatch.countDown(); + } + + @Override + public void onError(Throwable t) {} + + @Override + public void onCompleted() { + responseObserver.onNext("Ack"); + responseObserver.onCompleted(); + } + }; + } + })) + .build()); + + final java.util.concurrent.CountDownLatch dataPlaneHalfClosedLatch = + new java.util.concurrent.CountDownLatch(1); + ManagedChannel dataPlaneChannel = grpcCleanup.register( + InProcessChannelBuilder.forName(uniqueDataPlaneServerName) + .intercept(new ClientInterceptor() { + @Override + public ClientCall interceptCall( + MethodDescriptor method, CallOptions callOptions, Channel next) { + return new io.grpc.ForwardingClientCall.SimpleForwardingClientCall( + next.newCall(method, callOptions)) { + @Override + public void halfClose() { + dataPlaneHalfClosedLatch.countDown(); + super.halfClose(); + } + }; + } + }) + .directExecutor() + .build()); + + CallOptions callOptions = DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()); + ClientCall proxyCall = + interceptCall(interceptor, METHOD_CLIENT_STREAMING, callOptions, dataPlaneChannel); + proxyCall.start(new ClientCall.Listener() {}, new Metadata()); + + proxyCall.request(1); + proxyCall.sendMessage("msg1"); + + assertThat(messageDeliveredLatch.await(5, java.util.concurrent.TimeUnit.SECONDS)).isTrue(); + assertThat(serverReceivedMessages).containsExactly("normal-body"); + // Since end_of_stream was false, end_of_stream_without_message is ignored, so data plane must NOT be half-closed + assertThat(dataPlaneHalfClosedLatch.getCount()).isEqualTo(1); proxyCall.cancel("Cleanup", null); channelManager.close(); @@ -5187,6 +5733,7 @@ public void onNext(ProcessingRequest request) { .setResponse(CommonResponse.newBuilder() .setBodyMutation(BodyMutation.newBuilder() .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setEndOfStream(true) .setEndOfStreamWithoutMessage(true) .build()) .build()) @@ -12203,12 +12750,14 @@ public void onNext(ProcessingRequest request) { .build()); firstBodyLatch.countDown(); } else if (request.hasRequestBody()) { - if (request.getRequestBody().getEndOfStreamWithoutMessage()) { + if (request.getRequestBody().getEndOfStream() + && request.getRequestBody().getEndOfStreamWithoutMessage()) { responseObserver.onNext(ProcessingResponse.newBuilder() .setRequestBody(BodyResponse.newBuilder() .setResponse(CommonResponse.newBuilder() .setBodyMutation(BodyMutation.newBuilder() .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setEndOfStream(true) .setEndOfStreamWithoutMessage(true) .build()) .build()) @@ -12378,6 +12927,7 @@ public void onClose(Status status, Metadata trailers) { assertThat(receivedRequests.get(3).hasRequestBody()).isTrue(); assertThat(receivedRequests.get(3).getRequestBody().getBody().size()).isEqualTo(30000); assertThat(receivedRequests.get(4).hasRequestBody()).isTrue(); + assertThat(receivedRequests.get(4).getRequestBody().getEndOfStream()).isTrue(); assertThat(receivedRequests.get(4).getRequestBody().getEndOfStreamWithoutMessage()).isTrue(); channelManager.close(); @@ -12917,7 +13467,7 @@ public void onNext(ProcessingRequest request) { ByteString original = request.getRequestBody().getBody(); boolean eos = request.getRequestBody().getEndOfStream() - || request.getRequestBody().getEndOfStreamWithoutMessage(); + && request.getRequestBody().getEndOfStreamWithoutMessage(); responseObserver.onNext( ProcessingResponse.newBuilder() .setRequestBody( @@ -12936,6 +13486,7 @@ public void onNext(ProcessingRequest request) { + original .toStringUtf8())) .setEndOfStream(eos) + .setEndOfStreamWithoutMessage(eos) .build()) .build()) .build()) @@ -14853,7 +15404,7 @@ public void onNext(ProcessingRequest request) { .build()); } else if (request.hasRequestBody()) { HttpBody requestBody = request.getRequestBody(); - if (requestBody.getEndOfStreamWithoutMessage()) { + if (requestBody.getEndOfStream() && requestBody.getEndOfStreamWithoutMessage()) { responseObserver.onNext( ProcessingResponse.newBuilder() .setRequestBody( @@ -14865,6 +15416,7 @@ public void onNext(ProcessingRequest request) { .setStreamedResponse( StreamedBodyResponse.newBuilder() .setEndOfStream(true) + .setEndOfStreamWithoutMessage(true) .build()) .build()) .build()) @@ -15136,9 +15688,11 @@ public void onCompleted() {} // Verify halfClose sends EOF request piggybacking the accumulated 12 bytes update. assertThat(filterClientRequests(receivedRequests)).hasSize(3); assertThat(filterClientRequests(receivedRequests).get(2).hasRequestBody()).isTrue(); + assertThat(filterClientRequests(receivedRequests).get(2) + .getRequestBody().getEndOfStream()) + .isTrue(); assertThat(filterClientRequests(receivedRequests).get(2) .getRequestBody().getEndOfStreamWithoutMessage()) - .isTrue(); assertThat(filterClientRequests(receivedRequests).get(2).hasClientWindowUpdate()).isTrue(); assertThat(filterClientRequests(receivedRequests).get(2) @@ -15299,6 +15853,8 @@ public void onCompleted() {} // Verify EOF request content assertThat(filterClientRequests(receivedRequests).get(3).hasRequestBody()).isTrue(); + assertThat(filterClientRequests(receivedRequests).get(3).getRequestBody() + .getEndOfStream()).isTrue(); assertThat(filterClientRequests(receivedRequests).get(3).getRequestBody() .getEndOfStreamWithoutMessage()).isTrue(); @@ -16631,7 +17187,8 @@ public void onNext(ProcessingRequest request) { .build()) .build()); } else if (request.hasRequestBody()) { - if (request.getRequestBody().getEndOfStreamWithoutMessage()) { + if (request.getRequestBody().getEndOfStream() + && request.getRequestBody().getEndOfStreamWithoutMessage()) { receivedPhases.add("REQ_BODY_EOS"); resp.setRequestBody( BodyResponse.newBuilder() @@ -16929,7 +17486,8 @@ public void onNext(ProcessingRequest request) { .build()) .build()); } else if (request.hasRequestBody()) { - if (request.getRequestBody().getEndOfStreamWithoutMessage()) { + if (request.getRequestBody().getEndOfStream() + && request.getRequestBody().getEndOfStreamWithoutMessage()) { receivedPhases.add("REQ_BODY_EOS"); resp.setRequestBody( BodyResponse.newBuilder() @@ -18534,7 +19092,7 @@ public StreamObserver process( public void onNext(ProcessingRequest request) { if (request.hasRequestBody()) { if (request.getRequestBody().getEndOfStream() - || request.getRequestBody().getEndOfStreamWithoutMessage()) { + && request.getRequestBody().getEndOfStreamWithoutMessage()) { // This is the half-close request! observerRef.get().onNext(ProcessingResponse.newBuilder() .setRequestBody(BodyResponse.newBuilder() @@ -18542,6 +19100,7 @@ public void onNext(ProcessingRequest request) { .setBodyMutation(BodyMutation.newBuilder() .setStreamedResponse(StreamedBodyResponse.newBuilder() .setEndOfStream(true) + .setEndOfStreamWithoutMessage(true) .build()) .build()) .build()) @@ -19016,8 +19575,7 @@ public void onCompleted() { ClientCall proxyCall = interceptCall(interceptor, METHOD_SAY_HELLO, - DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()) - .withOption(XdsNameResolver.CLUSTER_SELECTION_KEY, "backend-service-metric"), + DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()), dataPlaneChannel); proxyCall.start(new ClientCall.Listener() { @@ -19048,25 +19606,25 @@ public void onCompleted() { Mockito.eq(ExternalProcessorClientInterceptor.clientHeadersDuration), Mockito.anyDouble(), Mockito.eq(com.google.common.collect.ImmutableList.of("xds:///target-service-metric")), - Mockito.eq(com.google.common.collect.ImmutableList.of("backend-service-metric"))); + Mockito.eq(com.google.common.collect.ImmutableList.of())); Mockito.verify(mockMetricRecorder, Mockito.times(1)).recordDoubleHistogram( Mockito.eq(ExternalProcessorClientInterceptor.clientHalfCloseDuration), Mockito.anyDouble(), Mockito.eq(com.google.common.collect.ImmutableList.of("xds:///target-service-metric")), - Mockito.eq(com.google.common.collect.ImmutableList.of("backend-service-metric"))); + Mockito.eq(com.google.common.collect.ImmutableList.of())); Mockito.verify(mockMetricRecorder, Mockito.times(1)).recordDoubleHistogram( Mockito.eq(ExternalProcessorClientInterceptor.serverHeadersDuration), Mockito.anyDouble(), Mockito.eq(com.google.common.collect.ImmutableList.of("xds:///target-service-metric")), - Mockito.eq(com.google.common.collect.ImmutableList.of("backend-service-metric"))); + Mockito.eq(com.google.common.collect.ImmutableList.of())); Mockito.verify(mockMetricRecorder, Mockito.times(1)).recordDoubleHistogram( Mockito.eq(ExternalProcessorClientInterceptor.serverTrailersDuration), Mockito.anyDouble(), Mockito.eq(com.google.common.collect.ImmutableList.of("xds:///target-service-metric")), - Mockito.eq(com.google.common.collect.ImmutableList.of("backend-service-metric"))); + Mockito.eq(com.google.common.collect.ImmutableList.of())); channelManager.close(); realScheduler.shutdown(); @@ -19183,8 +19741,7 @@ public ServerCall.Listener interceptCall( ClientCall proxyCall = interceptCall(interceptor, METHOD_SAY_HELLO, - DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()) - .withOption(XdsNameResolver.CLUSTER_SELECTION_KEY, "backend-service-metric-fail"), + DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()), dataPlaneChannel); proxyCall.start(new ClientCall.Listener() { @@ -19218,25 +19775,25 @@ public ServerCall.Listener interceptCall( Mockito.eq(ExternalProcessorClientInterceptor.clientHeadersDuration), Mockito.anyDouble(), Mockito.eq(com.google.common.collect.ImmutableList.of("xds:///target-service-metric-fail")), - Mockito.eq(com.google.common.collect.ImmutableList.of("backend-service-metric-fail"))); + Mockito.eq(com.google.common.collect.ImmutableList.of())); Mockito.verify(mockMetricRecorder, Mockito.times(1)).recordDoubleHistogram( Mockito.eq(ExternalProcessorClientInterceptor.clientHalfCloseDuration), Mockito.anyDouble(), Mockito.eq(com.google.common.collect.ImmutableList.of("xds:///target-service-metric-fail")), - Mockito.eq(com.google.common.collect.ImmutableList.of("backend-service-metric-fail"))); + Mockito.eq(com.google.common.collect.ImmutableList.of())); Mockito.verify(mockMetricRecorder, Mockito.times(1)).recordDoubleHistogram( Mockito.eq(ExternalProcessorClientInterceptor.serverHeadersDuration), Mockito.anyDouble(), Mockito.eq(com.google.common.collect.ImmutableList.of("xds:///target-service-metric-fail")), - Mockito.eq(com.google.common.collect.ImmutableList.of("backend-service-metric-fail"))); + Mockito.eq(com.google.common.collect.ImmutableList.of())); Mockito.verify(mockMetricRecorder, Mockito.times(1)).recordDoubleHistogram( Mockito.eq(ExternalProcessorClientInterceptor.serverTrailersDuration), Mockito.anyDouble(), Mockito.eq(com.google.common.collect.ImmutableList.of("xds:///target-service-metric-fail")), - Mockito.eq(com.google.common.collect.ImmutableList.of("backend-service-metric-fail"))); + Mockito.eq(com.google.common.collect.ImmutableList.of())); channelManager.close(); realScheduler.shutdown(); From 7233b3ca77e7f1516c805725d7be4519a559f76f Mon Sep 17 00:00:00 2001 From: Kannan J Date: Fri, 4 Sep 2026 10:00:40 +0000 Subject: [PATCH 5/5] xds: fix checkstyle violations in ExternalProcessorClientInterceptorTest TAG=agy CONV=86c74ebe-9fd7-4876-b27c-a4c1b230d346 --- .../xds/ExternalProcessorClientInterceptorTest.java | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java b/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java index c9d051edb90..83614089dce 100644 --- a/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java +++ b/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java @@ -5210,7 +5210,8 @@ public void halfClose() { proxyCall.halfClose(); assertThat(dataPlaneHalfClosedLatch.await(5, java.util.concurrent.TimeUnit.SECONDS)).isTrue(); - // Since both end_of_stream and end_of_stream_without_message are true, the body must NOT be delivered + // Since both end_of_stream and end_of_stream_without_message are true, + // the body must NOT be delivered assertThat(serverReceivedMessages).isEmpty(); proxyCall.cancel("Cleanup", null); @@ -5377,7 +5378,7 @@ public void halfClose() { @Test @SuppressWarnings("unchecked") public void - clientInterceptor_extProcResponse_eosTrue_eosWithoutMessageFalse_emptyBody_deliveredAsMessage() + clientInterceptor_extProcResponse_eosTrue_eosWithoutMsgFalse_emptyBody_deliveredAsMessage() throws Exception { String uniqueExtProcServerName = InProcessServerBuilder.generateName(); String uniqueDataPlaneServerName = InProcessServerBuilder.generateName(); @@ -5512,7 +5513,8 @@ public void halfClose() { proxyCall.halfClose(); assertThat(dataPlaneHalfClosedLatch.await(5, java.util.concurrent.TimeUnit.SECONDS)).isTrue(); - // Since end_of_stream is true and end_of_stream_without_message is false, empty body is delivered as a message + // Since end_of_stream is true and end_of_stream_without_message is false, + // empty body is delivered as a message assertThat(serverReceivedMessages).containsExactly(""); proxyCall.cancel("Cleanup", null); @@ -5658,7 +5660,8 @@ public void halfClose() { assertThat(messageDeliveredLatch.await(5, java.util.concurrent.TimeUnit.SECONDS)).isTrue(); assertThat(serverReceivedMessages).containsExactly("normal-body"); - // Since end_of_stream was false, end_of_stream_without_message is ignored, so data plane must NOT be half-closed + // Since end_of_stream was false, end_of_stream_without_message is ignored, + // so data plane must NOT be half-closed assertThat(dataPlaneHalfClosedLatch.getCount()).isEqualTo(1); proxyCall.cancel("Cleanup", null);