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..83614089dce 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,321 @@ 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 +5274,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 +5377,8 @@ public void halfClose() { @Test @SuppressWarnings("unchecked") - public void givenDeferredHalfClose_whenExtProcRespondsWithEndOfStream_thenSuperHalfCloseCalled() + public void + clientInterceptor_extProcResponse_eosTrue_eosWithoutMsgFalse_emptyBody_deliveredAsMessage() throws Exception { String uniqueExtProcServerName = InProcessServerBuilder.generateName(); String uniqueDataPlaneServerName = InProcessServerBuilder.generateName(); @@ -4974,8 +5387,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 +5401,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 +5471,7 @@ public void onNext(String value) { } @Override - public void onError(Throwable t) { - } + public void onError(Throwable t) {} @Override public void onCompleted() { @@ -5113,7 +5513,156 @@ 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 +5736,7 @@ public void onNext(ProcessingRequest request) { .setResponse(CommonResponse.newBuilder() .setBodyMutation(BodyMutation.newBuilder() .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setEndOfStream(true) .setEndOfStreamWithoutMessage(true) .build()) .build()) @@ -12203,12 +12753,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 +12930,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 +13470,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 +13489,7 @@ public void onNext(ProcessingRequest request) { + original .toStringUtf8())) .setEndOfStream(eos) + .setEndOfStreamWithoutMessage(eos) .build()) .build()) .build()) @@ -14853,7 +15407,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 +15419,7 @@ public void onNext(ProcessingRequest request) { .setStreamedResponse( StreamedBodyResponse.newBuilder() .setEndOfStream(true) + .setEndOfStreamWithoutMessage(true) .build()) .build()) .build()) @@ -15136,9 +15691,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 +15856,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 +17190,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 +17489,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 +19095,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 +19103,7 @@ public void onNext(ProcessingRequest request) { .setBodyMutation(BodyMutation.newBuilder() .setStreamedResponse(StreamedBodyResponse.newBuilder() .setEndOfStream(true) + .setEndOfStreamWithoutMessage(true) .build()) .build()) .build()) @@ -19016,8 +19578,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 +19609,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 +19744,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 +19778,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();