diff --git a/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java b/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java index 0bc79e5ec5d..ef417ed349e 100644 --- a/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java +++ b/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java @@ -316,6 +316,14 @@ private static class DataPlaneClientCall final AtomicBoolean pendingHalfClose = new AtomicBoolean(false); final AtomicBoolean bodyMessageSentToExtProc = new AtomicBoolean(false); private final AtomicBoolean downstreamCancelled = new AtomicBoolean(false); + final AtomicBoolean requestDraining = new AtomicBoolean(false); + final AtomicBoolean responseDraining = new AtomicBoolean(false); + final AtomicBoolean requestDrainComplete = new AtomicBoolean(false); + final AtomicBoolean responseDrainComplete = new AtomicBoolean(false); + final AtomicBoolean requestEosSent = new AtomicBoolean(false); + final AtomicBoolean responseTrailersSent = new AtomicBoolean(false); + final AtomicBoolean requestBodySentToExtProc = new AtomicBoolean(false); + final AtomicBoolean responseBodySentToExtProc = new AtomicBoolean(false); protected DataPlaneClientCall( DataPlaneDelayedCall delayedCall, @@ -502,10 +510,25 @@ public void onNext(ProcessingResponse response) { } } - if (response.getRequestDrain()) { - extProcStreamState.set(ExtProcStreamState.DRAINING); - halfCloseExtProcStream(); - activateCall(); + if (response.getRequestDrainRequests()) { + if (!requestEosSent.get() && requestDraining.compareAndSet(false, true)) { + activateCall(); + sendToExtProc(ProcessingRequest.newBuilder() + .setRequestBody(HttpBody.newBuilder() + .setDrainComplete(true) + .build()) + .build()); + } + } + if (response.getRequestDrainResponses()) { + if (!responseTrailersSent.get() && responseDraining.compareAndSet(false, true)) { + activateCall(); + sendToExtProc(ProcessingRequest.newBuilder() + .setResponseBody(HttpBody.newBuilder() + .setDrainComplete(true) + .build()) + .build()); + } } // 1. Client Headers @@ -599,6 +622,33 @@ public void onError(Throwable t) { @Override public void onCompleted() { + boolean requestDrainRequired = + currentProcessingMode.getRequestBodyMode() == ProcessingMode.BodySendMode.GRPC + && requestBodySentToExtProc.get() + && !requestDrainComplete.get() + && !requestEosSent.get(); + + boolean responseDrainRequired = + currentProcessingMode.getResponseBodyMode() == ProcessingMode.BodySendMode.GRPC + && responseBodySentToExtProc.get() + && !responseDrainComplete.get() + && !responseTrailersSent.get(); + + if (!config.getObservabilityMode() && (requestDrainRequired || responseDrainRequired)) { + if (markExtProcStreamFailed(extProcStreamState)) { + synchronized (streamLock) { + extProcClientCallRequestObserver = null; + } + cancelDownstream( + "External processor stream completed without drain", + Status.INTERNAL + .withDescription("External processor stream completed without drain") + .asRuntimeException()); + wrappedListener.proceedWithClose(); + } + return; + } + if (markExtProcStreamCompleted(extProcStreamState)) { handleFailOpen(wrappedListener); } @@ -680,6 +730,9 @@ private void onExtProcStreamReady() { } private void drainPendingRequests() { + if (!isResponseSidecarReady()) { + return; + } int toRequest = pendingRequests.getAndSet(0); if (toRequest > 0) { super.request(toRequest); @@ -719,24 +772,32 @@ private void internalOnError(Throwable t) { } } - private void halfCloseExtProcStream() { - synchronized (streamLock) { - if (!extProcStreamState.get().isCompleted() && extProcClientCallRequestObserver != null) { - extProcClientCallRequestObserver.onCompleted(); - } - } - } + private void onReadyNotify() { wrappedListener.onReadyNotify(); } - private boolean isSidecarReady() { + private boolean isRequestSidecarReady() { ExtProcStreamState state = extProcStreamState.get(); if (state.isCompleted()) { return true; } - if (state.isDraining()) { + if (requestDraining.get()) { + return false; + } + synchronized (streamLock) { + ClientCallStreamObserver observer = extProcClientCallRequestObserver; + return observer != null && observer.isReady(); + } + } + + private boolean isResponseSidecarReady() { + ExtProcStreamState state = extProcStreamState.get(); + if (state.isCompleted()) { + return true; + } + if (responseDraining.get()) { return false; } synchronized (streamLock) { @@ -756,7 +817,7 @@ public boolean isReady() { if (dataPlaneCallState.get() == DataPlaneCallState.IDLE && !config.getObservabilityMode()) { return false; } - boolean sidecarReady = isSidecarReady(); + boolean sidecarReady = isRequestSidecarReady(); if (config.getObservabilityMode()) { return super.isReady() && sidecarReady; } @@ -774,7 +835,11 @@ public void request(int numMessages) { super.request(numMessages); return; } - if (!isSidecarReady()) { + if (dataPlaneCallState.get() == DataPlaneCallState.IDLE) { + pendingRequests.addAndGet(numMessages); + return; + } + if (!isResponseSidecarReady()) { pendingRequests.addAndGet(numMessages); return; } @@ -800,7 +865,7 @@ public void sendMessage(InputStream message) { } ExtProcStreamState state = extProcStreamState.get(); - if (state.isDraining() || state.isCompleted()) { + if (requestDraining.get() || state.isCompleted()) { if (currentProcessingMode.getRequestBodyMode() == ProcessingMode.BodySendMode.NONE) { super.sendMessage(message); return; @@ -829,6 +894,7 @@ public void sendMessage(InputStream message) { .setEndOfStream(false) .build()) .build()); + requestBodySentToExtProc.set(true); bodyMessageSentToExtProc.set(true); if (config.getObservabilityMode()) { @@ -858,46 +924,53 @@ public void halfClose() { return; } - pendingHalfClose.set(true); - - if (extProcStreamState.get().isCompleted()) { + synchronized (streamLock) { if (passThroughMode.get()) { if (requestSideClosed.compareAndSet(false, true)) { proceedWithHalfClose(); } + return; } - return; - } - if (extProcStreamState.get().isDraining()) { - boolean canProceed = false; - synchronized (streamLock) { - if (currentProcessingMode.getRequestBodyMode() == ProcessingMode.BodySendMode.NONE - || (!bodyMessageSentToExtProc.get() && pendingDrainingMessages.isEmpty())) { + pendingHalfClose.set(true); + + if (extProcStreamState.get().isCompleted()) { + return; + } + + if (requestDraining.get() || requestDrainComplete.get()) { + boolean canProceed = false; + if (currentProcessingMode.getRequestBodyMode() == ProcessingMode.BodySendMode.NONE) { canProceed = true; + } else if (pendingDrainingMessages.isEmpty()) { + if (requestDrainComplete.get() || !bodyMessageSentToExtProc.get()) { + canProceed = true; + } + } + + if (canProceed) { + if (requestSideClosed.compareAndSet(false, true)) { + proceedWithHalfClose(); + } } + return; } - if (canProceed) { + + if (currentProcessingMode.getRequestBodyMode() == ProcessingMode.BodySendMode.NONE) { if (requestSideClosed.compareAndSet(false, true)) { proceedWithHalfClose(); } + return; } - return; - } - if (currentProcessingMode.getRequestBodyMode() == ProcessingMode.BodySendMode.NONE) { - if (requestSideClosed.compareAndSet(false, true)) { - proceedWithHalfClose(); - } - return; + // Mode is GRPC + sendToExtProc(ProcessingRequest.newBuilder() + .setRequestBody(HttpBody.newBuilder() + .setEndOfStreamWithoutMessage(true) + .build()) + .build()); + requestEosSent.set(true); } - - // Mode is GRPC - sendToExtProc(ProcessingRequest.newBuilder() - .setRequestBody(HttpBody.newBuilder() - .setEndOfStreamWithoutMessage(true) - .build()) - .build()); } private void cancelDownstream(@Nullable String message, @Nullable Throwable cause) { @@ -925,7 +998,9 @@ private void handleRequestBodyResponse(BodyResponse bodyResponse) { BodyMutation mutation = bodyResponse.getResponse().getBodyMutation(); if (mutation.hasStreamedResponse()) { StreamedBodyResponse streamed = mutation.getStreamedResponse(); - if (!streamed.getEndOfStreamWithoutMessage()) { + if (streamed.getDrainComplete()) { + handleRequestDrainComplete(); + } else if (!streamed.getEndOfStreamWithoutMessage()) { super.sendMessage(new KnownLengthInputStream(streamed.getBody())); } if (streamed.getEndOfStream() || streamed.getEndOfStreamWithoutMessage()) { @@ -943,7 +1018,11 @@ private void handleResponseBodyResponse( BodyMutation mutation = bodyResponse.getResponse().getBodyMutation(); if (mutation.hasStreamedResponse()) { StreamedBodyResponse streamed = mutation.getStreamedResponse(); - listener.onExternalBody(streamed.getBody()); + if (streamed.getDrainComplete()) { + handleResponseDrainComplete(); + } else { + listener.onExternalBody(streamed.getBody()); + } } } } @@ -991,6 +1070,23 @@ private void drainPendingDrainingMessages() { } } + private void handleRequestDrainComplete() { + if (requestDraining.compareAndSet(true, false)) { + requestDrainComplete.set(true); + drainPendingDrainingMessages(); + if (wrappedListener != null) { + wrappedListener.onReadyNotify(); + } + } + } + + private void handleResponseDrainComplete() { + if (responseDraining.compareAndSet(true, false)) { + responseDrainComplete.set(true); + wrappedListener.unblockAfterResponseDrain(); + } + } + private void handleFailOpen(DataPlaneListener listener) { activateCall(); drainPendingRequests(); @@ -1111,7 +1207,8 @@ public void onHeaders(Metadata headers) { || dataPlaneClientCall.getCurrentProcessingMode().getResponseHeaderMode() == ProcessingMode.HeaderSendMode.DEFAULT; - if (dataPlaneClientCall.getExtProcStreamState().get().isDraining() && sendResponseHeaders) { + + if (dataPlaneClientCall.responseDraining.get() && sendResponseHeaders) { this.savedHeaders = headers; return; } @@ -1144,7 +1241,7 @@ public void onMessage(InputStream message) { return; } - boolean checkDrain = dataPlaneClientCall.getExtProcStreamState().get().isDraining() + boolean checkDrain = dataPlaneClientCall.responseDraining.get() && dataPlaneClientCall.getCurrentProcessingMode().getResponseBodyMode() == ProcessingMode.BodySendMode.GRPC; @@ -1174,6 +1271,7 @@ public void onMessage(InputStream message) { try { ByteString bodyByteString = ByteString.readFrom(message); sendResponseBodyToExtProc(bodyByteString, false); + dataPlaneClientCall.responseBodySentToExtProc.set(true); dataPlaneClientCall.bodyMessageSentToExtProc.set(true); if (dataPlaneClientCall.getConfig().getObservabilityMode()) { @@ -1198,8 +1296,13 @@ public void onClose(Status status, Metadata trailers) { && (!dataPlaneClientCall.getConfig().getFailureModeAllow() || dataPlaneClientCall.bodyMessageSentToExtProc.get())) { if (markDataPlaneCallClosed(dataPlaneClientCall.dataPlaneCallState)) { - proceedWithClose(Status.INTERNAL.withDescription("External processor stream failed") - .withCause(status.getCause()), new Metadata()); + Status finalStatus = Status.INTERNAL.withCause(status.getCause()); + if (status.getDescription() != null) { + finalStatus = finalStatus.withDescription(status.getDescription()); + } else { + finalStatus = finalStatus.withDescription("External processor stream failed"); + } + proceedWithClose(finalStatus, new Metadata()); } return; } @@ -1225,7 +1328,7 @@ public void onClose(Status status, Metadata trailers) { dataPlaneClientCall.getCurrentProcessingMode().getResponseTrailerMode() == ProcessingMode.HeaderSendMode.SEND; - if (dataPlaneClientCall.getExtProcStreamState().get().isDraining() && sendResponseTrailers) { + if (dataPlaneClientCall.responseDraining.get() && sendResponseTrailers) { return; } @@ -1237,7 +1340,9 @@ public void onClose(Status status, Metadata trailers) { } void onReadyNotify() { - dataPlaneClientCall.getCallContext().run(() -> delegate().onReady()); + if (dataPlaneClientCall.isReady()) { + dataPlaneClientCall.getCallContext().run(() -> delegate().onReady()); + } } void proceedWithHeaders() { @@ -1245,7 +1350,7 @@ void proceedWithHeaders() { proceedWithHeaders(savedHeaders); synchronized (savedMessages) { savedHeaders = null; - if (!dataPlaneClientCall.getExtProcStreamState().get().isDraining()) { + if (!dataPlaneClientCall.responseDraining.get()) { InputStream msg; while ((msg = savedMessages.poll()) != null) { onMessage(msg); @@ -1299,6 +1404,17 @@ void unblockAfterStreamComplete() { proceedWithHeaders(); proceedWithSavedMessages(); dataPlaneClientCall.drainPendingDrainingMessages(); + onReadyNotify(); + proceedWithClose(); + } + + void unblockAfterResponseDrain() { + synchronized (savedMessages) { + inboundPassThrough = true; + } + proceedWithHeaders(); + proceedWithSavedMessages(); + dataPlaneClientCall.drainPendingRequests(); proceedWithClose(); } @@ -1319,57 +1435,63 @@ private void triggerCloseHandshake() { return; } - boolean sendResponseHeaders = - dataPlaneClientCall.getCurrentProcessingMode().getResponseHeaderMode() - == ProcessingMode.HeaderSendMode.SEND - || dataPlaneClientCall.getCurrentProcessingMode().getResponseHeaderMode() - == ProcessingMode.HeaderSendMode.DEFAULT; - - boolean sendResponseTrailers = - dataPlaneClientCall.getCurrentProcessingMode().getResponseTrailerMode() - == ProcessingMode.HeaderSendMode.SEND; + boolean closeNow = dataPlaneClientCall.getConfig().getObservabilityMode(); - if (trailersOnly.get()) { - if (sendResponseHeaders) { + if (dataPlaneClientCall.responseDrainComplete.get()) { + closeNow = true; + } else { + boolean sendResponseHeaders = + dataPlaneClientCall.getCurrentProcessingMode().getResponseHeaderMode() + == ProcessingMode.HeaderSendMode.SEND + || dataPlaneClientCall.getCurrentProcessingMode().getResponseHeaderMode() + == ProcessingMode.HeaderSendMode.DEFAULT; + + boolean sendResponseTrailers = + dataPlaneClientCall.getCurrentProcessingMode().getResponseTrailerMode() + == ProcessingMode.HeaderSendMode.SEND; + + if (trailersOnly.get()) { + if (sendResponseHeaders) { + dataPlaneClientCall.sendToExtProc(ProcessingRequest.newBuilder() + .setResponseHeaders(HttpHeaders.newBuilder() + .setHeaders( + toHeaderMap( + savedTrailers, + dataPlaneClientCall.getConfig().getForwardRulesConfig())) + .setEndOfStream(true) + .build()) + .build()); + dataPlaneClientCall.responseTrailersSent.set(true); + } else { + closeNow = true; + } + } else if (sendResponseTrailers) { + dataPlaneClientCall.getIsProcessingTrailers().set(true); dataPlaneClientCall.sendToExtProc(ProcessingRequest.newBuilder() - .setResponseHeaders(HttpHeaders.newBuilder() - .setHeaders( + .setResponseTrailers(HttpTrailers.newBuilder() + .setTrailers( toHeaderMap( savedTrailers, dataPlaneClientCall.getConfig().getForwardRulesConfig())) - .setEndOfStream(true) .build()) .build()); + dataPlaneClientCall.responseTrailersSent.set(true); } else { - proceedWithClose(); - if (!dataPlaneClientCall.getConfig().getObservabilityMode()) { - dataPlaneClientCall.closeExtProcStream(); - } - } - } else if (sendResponseTrailers) { - dataPlaneClientCall.getIsProcessingTrailers().set(true); - dataPlaneClientCall.sendToExtProc(ProcessingRequest.newBuilder() - .setResponseTrailers(HttpTrailers.newBuilder() - .setTrailers( - toHeaderMap( - savedTrailers, - dataPlaneClientCall.getConfig().getForwardRulesConfig())) - .build()) - .build()); - } else { - proceedWithClose(); - if (!dataPlaneClientCall.getConfig().getObservabilityMode()) { - dataPlaneClientCall.closeExtProcStream(); + closeNow = true; } } - if (dataPlaneClientCall.getConfig().getObservabilityMode()) { + if (closeNow) { proceedWithClose(); - @SuppressWarnings("unused") - ScheduledFuture unused = dataPlaneClientCall.getScheduler().schedule( - dataPlaneClientCall::closeExtProcStream, - dataPlaneClientCall.getConfig().getDeferredCloseTimeoutNanos(), - TimeUnit.NANOSECONDS); + if (dataPlaneClientCall.getConfig().getObservabilityMode()) { + @SuppressWarnings("unused") + ScheduledFuture unused = dataPlaneClientCall.getScheduler().schedule( + dataPlaneClientCall::closeExtProcStream, + dataPlaneClientCall.getConfig().getDeferredCloseTimeoutNanos(), + TimeUnit.NANOSECONDS); + } else { + dataPlaneClientCall.closeExtProcStream(); + } } } diff --git a/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java b/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java index cd6de138a48..61d0b647cfc 100644 --- a/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java +++ b/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java @@ -6070,11 +6070,11 @@ public boolean isReady() { // 4. Assert that proxyCall.isReady() delegates directly to the downstream call downstreamReady.set(true); assertThat(proxyCall.isReady()).isTrue(); - assertThat(downstreamIsReadyCallCount.get()).isEqualTo(1); + assertThat(downstreamIsReadyCallCount.get()).isEqualTo(2); downstreamReady.set(false); assertThat(proxyCall.isReady()).isFalse(); - assertThat(downstreamIsReadyCallCount.get()).isEqualTo(2); + assertThat(downstreamIsReadyCallCount.get()).isEqualTo(3); proxyCall.cancel("cleanup", null); channelManager.close(); @@ -6159,7 +6159,7 @@ public void onNext(ProcessingRequest request) { if (request.hasRequestHeaders()) { responseObserver.onNext(ProcessingResponse.newBuilder() .setRequestHeaders(HeadersResponse.newBuilder().build()) - .setRequestDrain(true) // Trigger Request Drain + .setRequestDrainRequests(true) // Trigger Request Drain .build()); sidecarActionLatch.countDown(); } @@ -6343,7 +6343,7 @@ public void onNext(ProcessingRequest request) { if (request.hasRequestHeaders()) { responseObserver.onNext(ProcessingResponse.newBuilder() .setRequestHeaders(HeadersResponse.newBuilder().build()) - .setRequestDrain(true) + .setRequestDrainRequests(true) .build()); drainSentLatch.countDown(); } @@ -6456,7 +6456,7 @@ public void onNext(ProcessingRequest request) { if (request.hasRequestHeaders()) { responseObserver.onNext(ProcessingResponse.newBuilder() .setRequestHeaders(HeadersResponse.newBuilder().build()) - .setRequestDrain(true) + .setRequestDrainRequests(true) .build()); drainSentLatch.countDown(); } @@ -6555,6 +6555,7 @@ public void testHalfCloseDeferredWhenDrainingAndMessagesSent() throws Exception final CountDownLatch headersReceivedLatch = new CountDownLatch(1); final CountDownLatch bodyReceivedLatch = new CountDownLatch(1); + final CountDownLatch filterSentDrainLatch = new CountDownLatch(1); final AtomicReference> extProcResponseObserverRef = new AtomicReference<>(); ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl = @@ -6568,17 +6569,24 @@ public StreamObserver process( @Override public void onNext(ProcessingRequest request) { if (request.hasRequestHeaders()) { - responseObserver.onNext(ProcessingResponse.newBuilder() - .setRequestHeaders(HeadersResponse.newBuilder().build()) - .build()); + synchronized (responseObserver) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestHeaders(HeadersResponse.newBuilder().build()) + .build()); + } headersReceivedLatch.countDown(); } else if (request.hasRequestBody()) { - // Respond to body with request_drain = true - responseObserver.onNext(ProcessingResponse.newBuilder() - .setRequestBody(BodyResponse.newBuilder().build()) - .setRequestDrain(true) - .build()); - bodyReceivedLatch.countDown(); + if (request.getRequestBody().getDrainComplete()) { + filterSentDrainLatch.countDown(); + } else { + synchronized (responseObserver) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestBody(BodyResponse.newBuilder().build()) + .setRequestDrainRequests(true) + .build()); + } + bodyReceivedLatch.countDown(); + } } } @@ -6649,14 +6657,34 @@ public void onCompleted() { // Verify downstream server has NOT received half-close yet. assertThat(serverHalfClosedLatch.await(1, TimeUnit.SECONDS)).isFalse(); - // Now complete the ext_proc stream. - if (extProcResponseObserverRef.get() != null) { - extProcResponseObserverRef.get().onCompleted(); + // Wait for filter to send drain_complete + assertThat(filterSentDrainLatch.await(5, TimeUnit.SECONDS)).isTrue(); + + // Now complete the draining handshake by echoing drain_complete + StreamObserver extProcRespObserver = extProcResponseObserverRef.get(); + assertThat(extProcRespObserver).isNotNull(); + synchronized (extProcRespObserver) { + extProcRespObserver.onNext(ProcessingResponse.newBuilder() + .setRequestBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setDrainComplete(true) + .build()) + .build()) + .build()) + .build()) + .build()); } // Verify downstream server now receives half-close. assertThat(serverHalfClosedLatch.await(5, TimeUnit.SECONDS)).isTrue(); + // Cleanup ext_proc stream + synchronized (extProcRespObserver) { + extProcRespObserver.onCompleted(); + } + channelManager.close(); } @@ -6697,7 +6725,7 @@ public void onNext(ProcessingRequest request) { if (request.hasRequestHeaders()) { responseObserver.onNext(ProcessingResponse.newBuilder() .setRequestHeaders(HeadersResponse.newBuilder().build()) - .setRequestDrain(true) // Trigger Request Drain + .setRequestDrainResponses(true) // Trigger Request Drain .build()); sidecarActionLatch.countDown(); } @@ -6824,7 +6852,7 @@ public void onNext(ProcessingRequest request) { if (request.hasRequestHeaders()) { responseObserver.onNext(ProcessingResponse.newBuilder() .setRequestHeaders(HeadersResponse.newBuilder().build()) - .setRequestDrain(true) // Trigger Request Drain + .setRequestDrainResponses(true) // Trigger Request Drain .build()); sidecarActionLatch.countDown(); } @@ -6945,7 +6973,7 @@ public void onNext(ProcessingRequest request) { if (request.hasRequestHeaders()) { responseObserver.onNext(ProcessingResponse.newBuilder() .setRequestHeaders(HeadersResponse.newBuilder().build()) - .setRequestDrain(true) // Trigger Request Drain + .setRequestDrainResponses(true) // Trigger Request Drain .build()); sidecarActionLatch.countDown(); } @@ -7063,7 +7091,7 @@ public StreamObserver process( public void onNext(ProcessingRequest request) { if (request.hasRequestHeaders()) { responseObserver.onNext(ProcessingResponse.newBuilder() - .setRequestDrain(true) + .setRequestDrainRequests(true) .build()); drainLatch.countDown(); } @@ -7169,7 +7197,7 @@ public void onNext(ProcessingRequest request) { new Thread(() -> { synchronized (responseObserver) { responseObserver.onNext(ProcessingResponse.newBuilder() - .setRequestDrain(true) + .setRequestDrainRequests(true) .build()); } sidecarOnNextLatch.countDown(); @@ -7328,7 +7356,7 @@ public StreamObserver process( public void onNext(ProcessingRequest request) { if (request.hasRequestHeaders()) { responseObserver.onNext(ProcessingResponse.newBuilder() - .setRequestDrain(true) + .setRequestDrainRequests(true) .build()); drainLatch.countDown(); } @@ -7437,7 +7465,7 @@ public void onNext(ProcessingRequest request) { new Thread(() -> { synchronized (responseObserver) { responseObserver.onNext(ProcessingResponse.newBuilder() - .setRequestDrain(true) + .setRequestDrainRequests(true) .build()); } try { @@ -7582,8 +7610,9 @@ public void onMessage(String message) { ExternalProcessorFilterConfig filterConfig = configOrError.config; // External Processor Server - final CountDownLatch sidecarFinishLatch = new CountDownLatch(1); - final CountDownLatch drainCompletedLatch = new CountDownLatch(1); + final AtomicReference> extProcResponseObserverRef = + new AtomicReference<>(); + final CountDownLatch filterSentDrainLatch = new CountDownLatch(1); final AtomicInteger extProcReceivedBodyCount = new AtomicInteger(0); ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl; extProcImpl = new ExternalProcessorGrpc.ExternalProcessorImplBase() { @@ -7591,6 +7620,7 @@ public void onMessage(String message) { @SuppressWarnings("unchecked") public StreamObserver process( final StreamObserver responseObserver) { + extProcResponseObserverRef.set(responseObserver); ((ServerCallStreamObserver) responseObserver).request(100); return new StreamObserver() { @Override @@ -7600,21 +7630,16 @@ public void onNext(ProcessingRequest request) { synchronized (responseObserver) { responseObserver.onNext(ProcessingResponse.newBuilder() .setRequestHeaders(HeadersResponse.newBuilder().build()) - .setRequestDrain(true) + .setRequestDrainRequests(true) .build()); } - try { - if (sidecarFinishLatch.await(5, TimeUnit.SECONDS)) { - synchronized (responseObserver) { - responseObserver.onCompleted(); - } - } - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } }).start(); } else if (request.hasRequestBody()) { - extProcReceivedBodyCount.incrementAndGet(); + if (request.getRequestBody().getDrainComplete()) { + filterSentDrainLatch.countDown(); + } else { + extProcReceivedBodyCount.incrementAndGet(); + } } } @@ -7624,7 +7649,6 @@ public void onError(Throwable t) { @Override public void onCompleted() { - drainCompletedLatch.countDown(); } }; } @@ -7690,7 +7714,7 @@ public void onMessage(String message) { proxyCall.start(appListener, new Metadata()); // Wait for drain to be processed - assertThat(drainCompletedLatch.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(filterSentDrainLatch.await(5, TimeUnit.SECONDS)).isTrue(); assertThat(proxyCall.isReady()).isFalse(); // Send message and half-close concurrently during drain state @@ -7707,7 +7731,23 @@ public void onMessage(String message) { assertThat(extProcReceivedBodyCount.get()).isEqualTo(0); // Now let sidecar complete - sidecarFinishLatch.countDown(); + System.out.println("JetskiTest: completing drain from test"); + StreamObserver extProcRespObserver = extProcResponseObserverRef.get(); + assertThat(extProcRespObserver).isNotNull(); + synchronized (extProcRespObserver) { + extProcRespObserver.onNext(ProcessingResponse.newBuilder() + .setRequestBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setDrainComplete(true) + .build()) + .build()) + .build()) + .build()) + .build()); + extProcRespObserver.onCompleted(); + } // Request response from data plane proxyCall.request(1); @@ -7753,7 +7793,9 @@ public void drainingStartsAfterRequestHeaders_whenAppSendsAndHalfCloses_thenBuff ExternalProcessorFilterConfig filterConfig = configOrError.config; // External Processor Server - final CountDownLatch sidecarFinishLatch = new CountDownLatch(1); + final AtomicReference> extProcResponseObserverRef = + new AtomicReference<>(); + final CountDownLatch filterSentDrainLatch = new CountDownLatch(1); final CountDownLatch drainCompletedLatch = new CountDownLatch(1); final CountDownLatch extProcReceivedBodyLatch = new CountDownLatch(1); final CountDownLatch mutatedBodyDeliveredLatch = new CountDownLatch(1); @@ -7764,6 +7806,7 @@ public void drainingStartsAfterRequestHeaders_whenAppSendsAndHalfCloses_thenBuff @SuppressWarnings("unchecked") public StreamObserver process( final StreamObserver responseObserver) { + extProcResponseObserverRef.set(responseObserver); ((ServerCallStreamObserver) responseObserver).request(100); return new StreamObserver() { @Override @@ -7775,32 +7818,27 @@ public void onNext(ProcessingRequest request) { .build()); } } else if (request.hasRequestBody()) { - extProcReceivedBodyLatch.countDown(); - new Thread(() -> { - synchronized (responseObserver) { - responseObserver.onNext(ProcessingResponse.newBuilder() - .setRequestBody(BodyResponse.newBuilder() - .setResponse(CommonResponse.newBuilder() - .setBodyMutation(BodyMutation.newBuilder() - .setStreamedResponse(StreamedBodyResponse.newBuilder() - .setBody(ByteString.copyFromUtf8("Mutated Message 1")) - .build()) - .build()) - .build()) - .build()) - .setRequestDrain(true) - .build()); - } - try { - if (sidecarFinishLatch.await(5, TimeUnit.SECONDS)) { - synchronized (responseObserver) { - responseObserver.onCompleted(); - } + if (request.getRequestBody().getDrainComplete()) { + filterSentDrainLatch.countDown(); + } else { + extProcReceivedBodyLatch.countDown(); + new Thread(() -> { + synchronized (responseObserver) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setBody(ByteString.copyFromUtf8("Mutated Message 1")) + .build()) + .build()) + .build()) + .build()) + .setRequestDrainRequests(true) + .build()); } - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } - }).start(); + }).start(); + } } } @@ -7810,7 +7848,6 @@ public void onError(Throwable t) { @Override public void onCompleted() { - drainCompletedLatch.countDown(); } }; } @@ -7887,6 +7924,7 @@ public void onMessage(String message) { assertThat(mutatedBodyDeliveredLatch.await(5, TimeUnit.SECONDS)).isTrue(); // Verify proxyCall is in draining (i.e. isReady is false) + assertThat(filterSentDrainLatch.await(5, TimeUnit.SECONDS)).isTrue(); assertThat(proxyCall.isReady()).isFalse(); // Send message and half-close concurrently during drain state @@ -7903,7 +7941,24 @@ public void onMessage(String message) { assertThat(dataPlaneReceivedMessages).containsExactly("Mutated Message 1"); // Now let sidecar complete - sidecarFinishLatch.countDown(); + System.out.println("JetskiTest: completing drain from test"); + StreamObserver extProcRespObserver = extProcResponseObserverRef.get(); + assertThat(extProcRespObserver).isNotNull(); + synchronized (extProcRespObserver) { + extProcRespObserver.onNext(ProcessingResponse.newBuilder() + .setRequestBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setDrainComplete(true) + .build()) + .build()) + .build()) + .build()) + .build()); + extProcRespObserver.onCompleted(); + drainCompletedLatch.countDown(); + } // Wait for the control stream drain to be fully completed assertThat(drainCompletedLatch.await(5, TimeUnit.SECONDS)).isTrue(); @@ -7952,36 +8007,36 @@ public void drainingStartsBeforeResponseHeaders_whenUpstreamResponds_thenBuffere ExternalProcessorFilterConfig filterConfig = configOrError.config; // External Processor Server - final CountDownLatch sidecarFinishLatch = new CountDownLatch(1); - final CountDownLatch drainCompletedLatch = new CountDownLatch(1); + final AtomicReference> extProcResponseObserverRef = + new AtomicReference<>(); + final CountDownLatch filterSentDrainLatch = new CountDownLatch(1); ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl; extProcImpl = new ExternalProcessorGrpc.ExternalProcessorImplBase() { @Override @SuppressWarnings("unchecked") public StreamObserver process( final StreamObserver responseObserver) { + extProcResponseObserverRef.set(responseObserver); ((ServerCallStreamObserver) responseObserver).request(100); return new StreamObserver() { @Override public void onNext(ProcessingRequest request) { + System.out.println("JetskiMock: Received request: " + request); if (request.hasRequestHeaders()) { + System.out.println("JetskiMock: Received request headers, triggering drain"); new Thread(() -> { synchronized (responseObserver) { responseObserver.onNext(ProcessingResponse.newBuilder() .setRequestHeaders(HeadersResponse.newBuilder().build()) - .setRequestDrain(true) + .setRequestDrainResponses(true) .build()); } - try { - if (sidecarFinishLatch.await(5, TimeUnit.SECONDS)) { - synchronized (responseObserver) { - responseObserver.onCompleted(); - } - } - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } }).start(); + } else if (request.hasResponseBody()) { + if (request.getResponseBody().getDrainComplete()) { + System.out.println("JetskiMock: Received drain_complete from filter"); + filterSentDrainLatch.countDown(); + } } } @@ -7991,7 +8046,6 @@ public void onError(Throwable t) { @Override public void onCompleted() { - drainCompletedLatch.countDown(); } }; } @@ -8071,27 +8125,49 @@ public void onClose(Status status, Metadata trailers) { // Request messages from server proxyCall.request(10); - // Wait for drain to be processed and sidecar's client stream to finish - assertThat(drainCompletedLatch.await(5, TimeUnit.SECONDS)).isTrue(); - assertThat(proxyCall.isReady()).isFalse(); + // Wait for drain to be processed + System.out.println("JetskiTest: waiting for filterSentDrainLatch"); + assertThat(filterSentDrainLatch.await(5, TimeUnit.SECONDS)).isTrue(); + System.out.println("JetskiTest: filterSentDrainLatch counted down"); + assertThat(proxyCall.isReady()).isTrue(); // Verify the data plane call has started + System.out.println("JetskiTest: waiting for dataPlaneCallStartedLatch"); assertThat(dataPlaneCallStartedLatch.await(5, TimeUnit.SECONDS)).isTrue(); + System.out.println("JetskiTest: dataPlaneCallStartedLatch counted down"); StreamObserver dataPlaneResponseObserver = dataPlaneResponseObserverRef.get(); assertThat(dataPlaneResponseObserver).isNotNull(); // Upstream server sends response headers, response message, and closes call during drain + System.out.println("JetskiTest: sending response headers/message"); dataPlaneResponseObserver.onNext("Response Message During Drain"); dataPlaneResponseObserver.onCompleted(); // Verify app listener has NOT received headers, messages, or close yet because the drain // is active + System.out.println("JetskiTest: verifying app hasn't received anything"); assertThat(appReceivedHeaders.get()).isNull(); assertThat(appReceivedMessages).isEmpty(); assertThat(appReceivedStatus.get()).isNull(); // Now let sidecar complete the drain - sidecarFinishLatch.countDown(); + System.out.println("JetskiTest: completing drain from test"); + StreamObserver extProcRespObserver = extProcResponseObserverRef.get(); + assertThat(extProcRespObserver).isNotNull(); + synchronized (extProcRespObserver) { + extProcRespObserver.onNext(ProcessingResponse.newBuilder() + .setResponseBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setDrainComplete(true) + .build()) + .build()) + .build()) + .build()) + .build()); + extProcRespObserver.onCompleted(); + } // Wait for the call to close on application side assertThat(appCloseLatch.await(5, TimeUnit.SECONDS)).isTrue(); @@ -8136,8 +8212,10 @@ public void drainingStartsAfterResponseHeaders_whenUpstreamResponds_thenBuffered final CountDownLatch respBody1Latch = new CountDownLatch(1); final CountDownLatch respBody2Latch = new CountDownLatch(1); final CountDownLatch m3SentLatch = new CountDownLatch(1); - final CountDownLatch sidecarFinishLatch = new CountDownLatch(1); final CountDownLatch drainCompletedLatch = new CountDownLatch(1); + final CountDownLatch filterSentDrainLatch = new CountDownLatch(1); + final AtomicReference> extProcResponseObserverRef = + new AtomicReference<>(); // External Processor Server ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl = @@ -8146,6 +8224,7 @@ public void drainingStartsAfterResponseHeaders_whenUpstreamResponds_thenBuffered @SuppressWarnings("unchecked") public StreamObserver process( final StreamObserver responseObserver) { + extProcResponseObserverRef.set(responseObserver); ((ServerCallStreamObserver) responseObserver).request(100); return new StreamObserver() { @Override @@ -8165,64 +8244,63 @@ public void onNext(ProcessingRequest request) { } respHeadersLatch.countDown(); } else if (request.hasResponseBody()) { - String msgStr = request.getResponseBody().getBody().toStringUtf8(); - if ("Original Message 1".equals(msgStr)) { - new Thread(() -> { - try { - // Wait until M2 is received by sidecar so both M1 and M2 are in flight - if (m2ReceivedLatch.await(5, TimeUnit.SECONDS)) { - synchronized (responseObserver) { - responseObserver.onNext(ProcessingResponse.newBuilder() - .setResponseBody(BodyResponse.newBuilder() - .setResponse(CommonResponse.newBuilder() - .setBodyMutation(BodyMutation.newBuilder() - .setStreamedResponse(StreamedBodyResponse.newBuilder() - .setBody(ByteString.copyFromUtf8( - "Mutated Message 1")) - .build()) - .build()) - .build()) - .build()) - .setRequestDrain(true) - .build()); - } - respBody1Latch.countDown(); - } - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } - }).start(); - } else if ("Original Message 2".equals(msgStr)) { - m2ReceivedLatch.countDown(); - new Thread(() -> { - try { - // Wait until M3 is sent by upstream concurrently during drain - if (m3SentLatch.await(5, TimeUnit.SECONDS)) { - synchronized (responseObserver) { - responseObserver.onNext(ProcessingResponse.newBuilder() - .setResponseBody(BodyResponse.newBuilder() - .setResponse(CommonResponse.newBuilder() - .setBodyMutation(BodyMutation.newBuilder() - .setStreamedResponse(StreamedBodyResponse.newBuilder() - .setBody(ByteString.copyFromUtf8( - "Mutated Message 2")) - .build()) - .build()) - .build()) - .build()) - .build()); + if (request.getResponseBody().getDrainComplete()) { + filterSentDrainLatch.countDown(); + } else { + String msgStr = request.getResponseBody().getBody().toStringUtf8(); + if ("Original Message 1".equals(msgStr)) { + new Thread(() -> { + try { + // Wait until M2 is received by sidecar so both M1 and M2 are in flight + if (m2ReceivedLatch.await(5, TimeUnit.SECONDS)) { + synchronized (responseObserver) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setResponseBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setBody(ByteString.copyFromUtf8( + "Mutated Message 1")) + .build()) + .build()) + .build()) + .build()) + .setRequestDrainResponses(true) + .build()); + } + respBody1Latch.countDown(); } - respBody2Latch.countDown(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); } - if (sidecarFinishLatch.await(5, TimeUnit.SECONDS)) { - synchronized (responseObserver) { - responseObserver.onCompleted(); + }).start(); + } else if ("Original Message 2".equals(msgStr)) { + m2ReceivedLatch.countDown(); + new Thread(() -> { + try { + // Wait until M3 is sent by upstream concurrently during drain + if (m3SentLatch.await(5, TimeUnit.SECONDS)) { + synchronized (responseObserver) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setResponseBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setBody(ByteString.copyFromUtf8( + "Mutated Message 2")) + .build()) + .build()) + .build()) + .build()) + .build()); + } + respBody2Latch.countDown(); } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); } - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } - }).start(); + }).start(); + } } } } @@ -8339,6 +8417,7 @@ public void onClose(Status status, Metadata trailers) { // Verify the stream is currently in DRAINING state and app has only received Mutated // Message 1 so far assertThat(appReceivedMessages).containsExactly("Mutated Message 1"); + assertThat(filterSentDrainLatch.await(5, TimeUnit.SECONDS)).isTrue(); // 3. Upstream concurrently sends M3 and completes the call during draining dataPlaneResponseObserver.onNext("Original Message 3"); @@ -8360,7 +8439,23 @@ public void onClose(Status status, Metadata trailers) { assertThat(appReceivedMessages).containsExactly("Mutated Message 1", "Mutated Message 2"); // 5. Complete sidecar stream to finish the drain - sidecarFinishLatch.countDown(); + System.out.println("JetskiTest: completing drain from test"); + StreamObserver extProcRespObserver = extProcResponseObserverRef.get(); + assertThat(extProcRespObserver).isNotNull(); + synchronized (extProcRespObserver) { + extProcRespObserver.onNext(ProcessingResponse.newBuilder() + .setResponseBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setDrainComplete(true) + .build()) + .build()) + .build()) + .build()) + .build()); + extProcRespObserver.onCompleted(); + } // Wait for the call to close on application side assertThat(appCloseLatch.await(5, TimeUnit.SECONDS)).isTrue(); @@ -8378,12 +8473,9 @@ public void onClose(Status status, Metadata trailers) { channelManager.close(); } - // --- Category 15: Inbound Backpressure (request(n) / pendingRequests) --- - @Test @SuppressWarnings("unchecked") - public void givenObservabilityTrue_whenExtProcBusy_thenAppRequestsBuffered() - throws Exception { + public void drainIgnored_whenRequestEosSent() throws Exception { ExternalProcessor proto = ExternalProcessor.newBuilder() .setGrpcService(GrpcService.newBuilder() .setGoogleGrpc(GrpcService.GoogleGrpc.newBuilder() @@ -8394,90 +8486,95 @@ public void givenObservabilityTrue_whenExtProcBusy_thenAppRequestsBuffered() .build()) .build()) .build()) - .setObservabilityMode(true) + .setProcessingMode(ProcessingMode.newBuilder() + .setRequestBodyMode(ProcessingMode.BodySendMode.GRPC) + .setResponseBodyMode(ProcessingMode.BodySendMode.GRPC) + .setResponseHeaderMode(ProcessingMode.HeaderSendMode.SEND) + .setResponseTrailerMode(ProcessingMode.HeaderSendMode.SEND) + .build()) .build(); ConfigOrError configOrError = provider.parseFilterConfig(Any.pack(proto), filterContext); assertThat(configOrError.errorDetail).isNull(); ExternalProcessorFilterConfig filterConfig = configOrError.config; + final CountDownLatch filterSentDrainLatch = new CountDownLatch(1); + final CountDownLatch reqEosReceivedLatch = new CountDownLatch(1); + final AtomicReference> extProcResponseObserverRef = + new AtomicReference<>(); + // External Processor Server - ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl; - extProcImpl = new ExternalProcessorGrpc.ExternalProcessorImplBase() { - @Override - @SuppressWarnings("unchecked") - public StreamObserver process( - StreamObserver responseObserver) { - ((ServerCallStreamObserver) responseObserver).request(100); - return new StreamObserver() { + ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl = + new ExternalProcessorGrpc.ExternalProcessorImplBase() { @Override - public void onNext(ProcessingRequest request) { - } + @SuppressWarnings("unchecked") + public StreamObserver process( + final StreamObserver responseObserver) { + extProcResponseObserverRef.set(responseObserver); + ((ServerCallStreamObserver) responseObserver).request(100); + return new StreamObserver() { + @Override + public void onNext(ProcessingRequest request) { + System.out.println("JetskiMock1: Received request: " + request); + if (request.hasRequestHeaders()) { + System.out.println("JetskiMock1: Received request headers"); + synchronized (responseObserver) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestHeaders(HeadersResponse.newBuilder().build()) + .build()); + } + } else if (request.hasRequestBody()) { + if (request.getRequestBody().getEndOfStreamWithoutMessage() + || request.getRequestBody().getEndOfStream()) { + System.out.println("JetskiMock1: Received request EOS"); + reqEosReceivedLatch.countDown(); + } + if (request.getRequestBody().getDrainComplete()) { + System.out.println("JetskiMock1: Received drain_complete"); + filterSentDrainLatch.countDown(); + } + } + } - @Override - public void onError(Throwable t) { - } + @Override + public void onError(Throwable t) {} - @Override - public void onCompleted() { + @Override + public void onCompleted() {} + }; } }; - } - }; + grpcCleanup.register(InProcessServerBuilder.forName(extProcServerName) .addService(extProcImpl) .directExecutor() .build().start()); - final AtomicBoolean sidecarReady = new AtomicBoolean(true); - final AtomicReference> sidecarListenerRef = - new AtomicReference<>(); CachedChannelManager channelManager = new CachedChannelManager(config -> { return grpcCleanup.register( - InProcessChannelBuilder.forName(extProcServerName) - .directExecutor() - .intercept(new ClientInterceptor() { - @Override - public ClientCall interceptCall( - MethodDescriptor method, CallOptions callOptions, Channel next) { - return new io.grpc.ForwardingClientCall.SimpleForwardingClientCall< - ReqT, RespT>(next.newCall(method, callOptions)) { - @Override - public void start(Listener responseListener, Metadata headers) { - sidecarListenerRef.set((Listener) responseListener); - super.start(responseListener, headers); - } - - @Override - public boolean isReady() { - return sidecarReady.get(); - } - }; - } - }) - .build()); + InProcessChannelBuilder.forName(extProcServerName).directExecutor().build()); }); ExternalProcessorClientInterceptor interceptor = new ExternalProcessorClientInterceptor( filterConfig, channelManager, scheduler, FAKE_CONTEXT); - final AtomicInteger dataPlaneRequestCount = new AtomicInteger(0); + final CountDownLatch dataPlaneCallStartedLatch = new CountDownLatch(1); dataPlaneServiceRegistry.addService(ServerServiceDefinition.builder("test.TestService") - .addMethod(METHOD_SAY_HELLO, ServerCalls.asyncBidiStreamingCall( - new ServerCalls.BidiStreamingMethod() { + .addMethod(METHOD_CLIENT_STREAMING, ServerCalls.asyncClientStreamingCall( + new ServerCalls.ClientStreamingMethod() { @Override public StreamObserver invoke(StreamObserver responseObserver) { + dataPlaneCallStartedLatch.countDown(); return new StreamObserver() { @Override - public void onNext(String value) { - } + public void onNext(String value) {} @Override - public void onError(Throwable t) { - } + public void onError(Throwable t) {} @Override public void onCompleted() { + responseObserver.onNext("Direct Response"); responseObserver.onCompleted(); } }; @@ -8486,11 +8583,372 @@ public void onCompleted() { .build()); ManagedChannel dataPlaneChannel = grpcCleanup.register( - InProcessChannelBuilder.forName(dataPlaneServerName) - .directExecutor() - .intercept(new ClientInterceptor() { - @Override - public ClientCall interceptCall( + InProcessChannelBuilder.forName(dataPlaneServerName).directExecutor().build()); + + final AtomicReference appReceivedMessage = new AtomicReference<>(); + final CountDownLatch appLatch = new CountDownLatch(1); + ClientCall.Listener appListener = new ClientCall.Listener() { + @Override + public void onMessage(String message) { + appReceivedMessage.set(message); + } + + @Override + public void onClose(Status status, Metadata trailers) { + appLatch.countDown(); + } + }; + + CallOptions callOptions = DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()); + ClientCall proxyCall = + interceptCall(interceptor, METHOD_CLIENT_STREAMING, callOptions, dataPlaneChannel); + proxyCall.start(appListener, new Metadata()); + + // Send request headers by requesting message (starts call) + proxyCall.request(1); + + // Wait for the data plane call to start + assertThat(dataPlaneCallStartedLatch.await(5, TimeUnit.SECONDS)).isTrue(); + + // Send app message and half-close (sends EOS) + proxyCall.sendMessage("App Message"); + proxyCall.halfClose(); + + // Wait for mock server to receive EOS + assertThat(reqEosReceivedLatch.await(5, TimeUnit.SECONDS)).isTrue(); + + // Now, mock server sends drain request AFTER EOS was already sent by client + StreamObserver extProcRespObserver = extProcResponseObserverRef.get(); + assertThat(extProcRespObserver).isNotNull(); + synchronized (extProcRespObserver) { + extProcRespObserver.onNext(ProcessingResponse.newBuilder() + .setRequestDrainRequests(true) + .build()); + } + + // Verify that filter does NOT send drain_complete (latch should timeout) + assertThat(filterSentDrainLatch.await(2, TimeUnit.SECONDS)).isFalse(); + + // Complete the call + synchronized (extProcRespObserver) { + extProcRespObserver.onCompleted(); + } + + assertThat(appLatch.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(appReceivedMessage.get()).isEqualTo("Direct Response"); + + proxyCall.cancel("Cleanup", null); + channelManager.close(); + } + + @Test + @SuppressWarnings("unchecked") + public void drainIgnored_whenResponseTrailersSent() throws Exception { + ExternalProcessor proto = ExternalProcessor.newBuilder() + .setGrpcService(GrpcService.newBuilder() + .setGoogleGrpc(GrpcService.GoogleGrpc.newBuilder() + .setTargetUri("in-process:///" + extProcServerName) + .addChannelCredentialsPlugin(Any.newBuilder() + .setTypeUrl("type.googleapis.com/envoy.extensions.grpc_service." + + "channel_credentials.insecure.v3.InsecureCredentials") + .build()) + .build()) + .build()) + .setProcessingMode(ProcessingMode.newBuilder() + .setRequestBodyMode(ProcessingMode.BodySendMode.GRPC) + .setResponseBodyMode(ProcessingMode.BodySendMode.GRPC) + .setResponseHeaderMode(ProcessingMode.HeaderSendMode.SEND) + .setResponseTrailerMode(ProcessingMode.HeaderSendMode.SEND) + .build()) + .build(); + ConfigOrError configOrError = + provider.parseFilterConfig(Any.pack(proto), filterContext); + assertThat(configOrError.errorDetail).isNull(); + ExternalProcessorFilterConfig filterConfig = configOrError.config; + + final CountDownLatch filterSentDrainLatch = new CountDownLatch(1); + final CountDownLatch trailersReceivedLatch = new CountDownLatch(1); + final AtomicReference> extProcResponseObserverRef = + new AtomicReference<>(); + + // External Processor Server + ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl = + new ExternalProcessorGrpc.ExternalProcessorImplBase() { + @Override + @SuppressWarnings("unchecked") + public StreamObserver process( + final StreamObserver responseObserver) { + extProcResponseObserverRef.set(responseObserver); + ((ServerCallStreamObserver) responseObserver).request(100); + return new StreamObserver() { + @Override + public void onNext(ProcessingRequest request) { + System.out.println("JetskiMock2: Received request: " + request); + if (request.hasRequestHeaders()) { + System.out.println("JetskiMock2: Received request headers"); + synchronized (responseObserver) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestHeaders(HeadersResponse.newBuilder().build()) + .build()); + } + } else if (request.hasResponseHeaders()) { + System.out.println("JetskiMock2: Received response headers"); + synchronized (responseObserver) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setResponseHeaders(HeadersResponse.newBuilder().build()) + .build()); + } + } else if (request.hasResponseTrailers()) { + System.out.println("JetskiMock2: Received response trailers"); + trailersReceivedLatch.countDown(); + } else if (request.hasResponseBody()) { + if (request.getResponseBody().getDrainComplete()) { + System.out.println("JetskiMock2: Received drain_complete"); + filterSentDrainLatch.countDown(); + } else { + System.out.println("JetskiMock2: Received response body chunk, echoing"); + synchronized (responseObserver) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setResponseBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setBody(request.getResponseBody().getBody()) + .build()) + .build()) + .build()) + .build()) + .build()); + } + } + } + } + + @Override + public void onError(Throwable t) {} + + @Override + public void onCompleted() {} + }; + } + }; + + grpcCleanup.register(InProcessServerBuilder.forName(extProcServerName) + .addService(extProcImpl) + .directExecutor() + .build().start()); + + CachedChannelManager channelManager = new CachedChannelManager(config -> { + return grpcCleanup.register( + InProcessChannelBuilder.forName(extProcServerName).directExecutor().build()); + }); + + ExternalProcessorClientInterceptor interceptor = new ExternalProcessorClientInterceptor( + filterConfig, channelManager, scheduler, FAKE_CONTEXT); + + final CountDownLatch dataPlaneCallStartedLatch = new CountDownLatch(1); + final AtomicReference> dataPlaneResponseObserverRef = + new AtomicReference<>(); + dataPlaneServiceRegistry.addService(ServerServiceDefinition.builder("test.TestService") + .addMethod(METHOD_BIDI_STREAMING, ServerCalls.asyncBidiStreamingCall( + new ServerCalls.BidiStreamingMethod() { + @Override + public StreamObserver invoke(StreamObserver responseObserver) { + dataPlaneResponseObserverRef.set(responseObserver); + dataPlaneCallStartedLatch.countDown(); + return new StreamObserver() { + @Override + public void onNext(String value) {} + + @Override + public void onError(Throwable t) {} + + @Override + public void onCompleted() {} + }; + } + })) + .build()); + + ManagedChannel dataPlaneChannel = grpcCleanup.register( + InProcessChannelBuilder.forName(dataPlaneServerName).directExecutor().build()); + + final List appReceivedMessages = new java.util.concurrent.CopyOnWriteArrayList<>(); + final CountDownLatch appCloseLatch = new CountDownLatch(1); + ClientCall.Listener appListener = new ClientCall.Listener() { + @Override + public void onHeaders(Metadata headers) {} + + @Override + public void onMessage(String message) { + appReceivedMessages.add(message); + } + + @Override + public void onClose(Status status, Metadata trailers) { + appCloseLatch.countDown(); + } + }; + + CallOptions callOptions = DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()); + ClientCall proxyCall = + interceptCall(interceptor, METHOD_BIDI_STREAMING, callOptions, dataPlaneChannel); + proxyCall.start(appListener, new Metadata()); + + proxyCall.request(10); + + // Wait for the data plane call to start + assertThat(dataPlaneCallStartedLatch.await(5, TimeUnit.SECONDS)).isTrue(); + StreamObserver dataPlaneResponseObserver = dataPlaneResponseObserverRef.get(); + assertThat(dataPlaneResponseObserver).isNotNull(); + + // Upstream server sends response and closes call (sends trailers) + dataPlaneResponseObserver.onNext("Response Message"); + dataPlaneResponseObserver.onCompleted(); + + // Wait for mock ext_proc to receive and process response trailers + assertThat(trailersReceivedLatch.await(5, TimeUnit.SECONDS)).isTrue(); + + // Now, mock server sends drain request AFTER trailers were already sent/processed + StreamObserver extProcRespObserver = extProcResponseObserverRef.get(); + assertThat(extProcRespObserver).isNotNull(); + synchronized (extProcRespObserver) { + extProcRespObserver.onNext(ProcessingResponse.newBuilder() + .setRequestDrainResponses(true) + .build()); + } + + // Verify that filter does NOT send drain_complete (latch should timeout) + assertThat(filterSentDrainLatch.await(2, TimeUnit.SECONDS)).isFalse(); + + // Complete the call + synchronized (extProcRespObserver) { + extProcRespObserver.onNext(ProcessingResponse.newBuilder() + .setResponseTrailers(TrailersResponse.newBuilder().build()) + .build()); + extProcRespObserver.onCompleted(); + } + + assertThat(appCloseLatch.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(appReceivedMessages).containsExactly("Response Message"); + + proxyCall.cancel("Cleanup", null); + channelManager.close(); + } + + // --- Category 15: Inbound Backpressure (request(n) / pendingRequests) --- + + @Test + @SuppressWarnings("unchecked") + public void givenObservabilityTrue_whenExtProcBusy_thenAppRequestsBuffered() + throws Exception { + ExternalProcessor proto = ExternalProcessor.newBuilder() + .setGrpcService(GrpcService.newBuilder() + .setGoogleGrpc(GrpcService.GoogleGrpc.newBuilder() + .setTargetUri("in-process:///" + extProcServerName) + .addChannelCredentialsPlugin(Any.newBuilder() + .setTypeUrl("type.googleapis.com/envoy.extensions.grpc_service." + + "channel_credentials.insecure.v3.InsecureCredentials") + .build()) + .build()) + .build()) + .setObservabilityMode(true) + .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( + StreamObserver responseObserver) { + ((ServerCallStreamObserver) responseObserver).request(100); + return new StreamObserver() { + @Override + public void onNext(ProcessingRequest request) { + } + + @Override + public void onError(Throwable t) { + } + + @Override + public void onCompleted() { + } + }; + } + }; + grpcCleanup.register(InProcessServerBuilder.forName(extProcServerName) + .addService(extProcImpl) + .directExecutor() + .build().start()); + + final AtomicBoolean sidecarReady = new AtomicBoolean(true); + final AtomicReference> sidecarListenerRef = + new AtomicReference<>(); + CachedChannelManager channelManager = new CachedChannelManager(config -> { + return grpcCleanup.register( + InProcessChannelBuilder.forName(extProcServerName) + .directExecutor() + .intercept(new ClientInterceptor() { + @Override + public ClientCall interceptCall( + MethodDescriptor method, CallOptions callOptions, Channel next) { + return new io.grpc.ForwardingClientCall.SimpleForwardingClientCall< + ReqT, RespT>(next.newCall(method, callOptions)) { + @Override + public void start(Listener responseListener, Metadata headers) { + sidecarListenerRef.set((Listener) responseListener); + super.start(responseListener, headers); + } + + @Override + public boolean isReady() { + return sidecarReady.get(); + } + }; + } + }) + .build()); + }); + + ExternalProcessorClientInterceptor interceptor = new ExternalProcessorClientInterceptor( + filterConfig, channelManager, scheduler, FAKE_CONTEXT); + + final AtomicInteger dataPlaneRequestCount = new AtomicInteger(0); + dataPlaneServiceRegistry.addService(ServerServiceDefinition.builder("test.TestService") + .addMethod(METHOD_SAY_HELLO, ServerCalls.asyncBidiStreamingCall( + new ServerCalls.BidiStreamingMethod() { + @Override + public StreamObserver invoke(StreamObserver responseObserver) { + return new StreamObserver() { + @Override + public void onNext(String value) { + } + + @Override + public void onError(Throwable t) { + } + + @Override + public void onCompleted() { + responseObserver.onCompleted(); + } + }; + } + })) + .build()); + + ManagedChannel dataPlaneChannel = grpcCleanup.register( + InProcessChannelBuilder.forName(dataPlaneServerName) + .directExecutor() + .intercept(new ClientInterceptor() { + @Override + public ClientCall interceptCall( MethodDescriptor method, CallOptions callOptions, Channel next) { return new io.grpc.ForwardingClientCall.SimpleForwardingClientCall( next.newCall(method, callOptions)) { @@ -8539,7 +8997,7 @@ public void request(int numMessages) { @Test @SuppressWarnings("unchecked") - public void givenRequestDrainActive_whenAppRequestsMessages_thenRequestsBuffered() + public void givenResponseDrainActive_whenAppRequestsMessages_thenRequestsBuffered() throws Exception { ExternalProcessor proto = ExternalProcessor.newBuilder() .setGrpcService(GrpcService.newBuilder() @@ -8561,6 +9019,12 @@ public void givenRequestDrainActive_whenAppRequestsMessages_thenRequestsBuffered assertThat(configOrError.errorDetail).isNull(); ExternalProcessorFilterConfig filterConfig = configOrError.config; + final CountDownLatch drainSentLatch = new CountDownLatch(1); + final CountDownLatch headersReceivedLatch = new CountDownLatch(1); + final CountDownLatch sendDrainLatch = new CountDownLatch(1); + final CountDownLatch filterSentDrainCompleteLatch = new CountDownLatch(1); + final AtomicReference> responseObserverRef = + new AtomicReference<>(); // External Processor Server ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl; extProcImpl = new ExternalProcessorGrpc.ExternalProcessorImplBase() { @@ -8568,14 +9032,30 @@ public void givenRequestDrainActive_whenAppRequestsMessages_thenRequestsBuffered @SuppressWarnings("unchecked") public StreamObserver process( final StreamObserver responseObserver) { + responseObserverRef.set(responseObserver); ((ServerCallStreamObserver) responseObserver).request(100); return new StreamObserver() { @Override public void onNext(ProcessingRequest request) { if (request.hasRequestHeaders()) { - responseObserver.onNext(ProcessingResponse.newBuilder() - .setRequestDrain(true) - .build()); + headersReceivedLatch.countDown(); + new Thread(() -> { + try { + sendDrainLatch.await(); + synchronized (responseObserver) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestDrainResponses(true) + .build()); + } + drainSentLatch.countDown(); + } catch (Exception e) { + System.out.println("JetskiDebug: error in mock server: " + e); + } + }).start(); + } else if (request.hasResponseBody()) { + if (request.getResponseBody().getDrainComplete()) { + filterSentDrainCompleteLatch.countDown(); + } } } @@ -8634,22 +9114,213 @@ public void request(int numMessages) { ClientCall proxyCall = interceptCall(interceptor, METHOD_SAY_HELLO, callOptions, dataPlaneChannel); proxyCall.start(new ClientCall.Listener() {}, new Metadata()); + + // Wait for headers to reach mock server + assertThat(headersReceivedLatch.await(5, TimeUnit.SECONDS)).isTrue(); + + // App requests messages early (before drain is received) + proxyCall.request(3); + + // Now trigger drain + sendDrainLatch.countDown(); // Wait for drain to be processed - long startTime = System.currentTimeMillis(); - while (proxyCall.isReady() && System.currentTimeMillis() - startTime < 5000) { - Thread.sleep(10); + assertThat(drainSentLatch.await(5, TimeUnit.SECONDS)).isTrue(); + // proxyCall.isReady() should remain true during response drain (request path is NOT draining) + assertThat(proxyCall.isReady()).isTrue(); + + // Verify requests are buffered and not sent to data plane + assertThat(dataPlaneRequestCount.get()).isEqualTo(0); + + // Wait for filter to send drain_complete to mock server + assertThat(filterSentDrainCompleteLatch.await(5, TimeUnit.SECONDS)).isTrue(); + + // Echo drain_complete back to filter to complete handshake + synchronized (responseObserverRef.get()) { + responseObserverRef.get().onNext(ProcessingResponse.newBuilder() + .setResponseBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setDrainComplete(true) + .build()) + .build()) + .build()) + .build()) + .build()); } - assertThat(proxyCall.isReady()).isFalse(); - // App requests more messages + // Verify requests are now drained to data plane + assertThat(dataPlaneRequestCount.get()).isEqualTo(3); + + proxyCall.cancel("Cleanup", null); + channelManager.close(); + } + + @Test + @SuppressWarnings("unchecked") + public void givenRequestDrainActive_whenAppRequestsMessages_thenRequestsDrained() + throws Exception { + ExternalProcessor proto = ExternalProcessor.newBuilder() + .setGrpcService(GrpcService.newBuilder() + .setGoogleGrpc(GrpcService.GoogleGrpc.newBuilder() + .setTargetUri("in-process:///" + extProcServerName) + .addChannelCredentialsPlugin(Any.newBuilder() + .setTypeUrl("type.googleapis.com/envoy.extensions.grpc_service." + + "channel_credentials.insecure.v3.InsecureCredentials") + .build()) + .build()) + .build()) + .setProcessingMode(ProcessingMode.newBuilder() + .setRequestBodyMode(ProcessingMode.BodySendMode.GRPC) + .setRequestTrailerMode(ProcessingMode.HeaderSendMode.SEND) + .build()) + .build(); + ConfigOrError configOrError = + provider.parseFilterConfig(Any.pack(proto), filterContext); + assertThat(configOrError.errorDetail).isNull(); + ExternalProcessorFilterConfig filterConfig = configOrError.config; + + final CountDownLatch drainSentLatch = new CountDownLatch(1); + final CountDownLatch headersReceivedLatch = new CountDownLatch(1); + final CountDownLatch sendDrainLatch = new CountDownLatch(1); + final CountDownLatch filterSentDrainCompleteLatch = new CountDownLatch(1); + final AtomicReference> responseObserverRef = + new AtomicReference<>(); + // External Processor Server + ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl; + extProcImpl = new ExternalProcessorGrpc.ExternalProcessorImplBase() { + @Override + @SuppressWarnings("unchecked") + public StreamObserver process( + final StreamObserver responseObserver) { + responseObserverRef.set(responseObserver); + ((ServerCallStreamObserver) responseObserver).request(100); + return new StreamObserver() { + @Override + public void onNext(ProcessingRequest request) { + if (request.hasRequestHeaders()) { + headersReceivedLatch.countDown(); + new Thread(() -> { + try { + sendDrainLatch.await(); + synchronized (responseObserver) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestHeaders(HeadersResponse.newBuilder().build()) + .setRequestDrainRequests(true) + .build()); + } + drainSentLatch.countDown(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }).start(); + } else if (request.hasRequestBody()) { + if (request.getRequestBody().getDrainComplete()) { + filterSentDrainCompleteLatch.countDown(); + } + } + } + + @Override + public void onError(Throwable t) { + } + + @Override + public void onCompleted() { + } + }; + } + }; + grpcCleanup.register(InProcessServerBuilder.forName(extProcServerName) + .addService(extProcImpl) + .directExecutor() + .build().start()); + + CachedChannelManager channelManager = new CachedChannelManager(config -> { + return grpcCleanup.register( + InProcessChannelBuilder.forName(extProcServerName).directExecutor().build()); + }); + + dataPlaneServiceRegistry.addService(ServerServiceDefinition.builder("test.TestService") + .addMethod(METHOD_SAY_HELLO, ServerCalls.asyncUnaryCall( + (request, responseObserver) -> { + responseObserver.onNext("Hello " + request); + responseObserver.onCompleted(); + })) + .build()); + + final AtomicInteger dataPlaneRequestCount = new AtomicInteger(); + ExternalProcessorClientInterceptor interceptor = new ExternalProcessorClientInterceptor( + filterConfig, channelManager, scheduler, FAKE_CONTEXT); + + ManagedChannel dataPlaneChannel = grpcCleanup.register( + InProcessChannelBuilder.forName(dataPlaneServerName) + .directExecutor() + .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 request(int numMessages) { + dataPlaneRequestCount.addAndGet(numMessages); + super.request(numMessages); + } + }; + } + }) + .build()); + + CallOptions callOptions = DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()); + ClientCall proxyCall = + interceptCall(interceptor, METHOD_SAY_HELLO, callOptions, dataPlaneChannel); + ClientCall.Listener mockListener = Mockito.mock(ClientCall.Listener.class); + proxyCall.start(mockListener, new Metadata()); + + // Wait for headers to reach mock server + assertThat(headersReceivedLatch.await(5, TimeUnit.SECONDS)).isTrue(); + + // App requests messages early (before drain is received) proxyCall.request(3); - // Verify requests are buffered and not sent to data plane + // Verify requests are buffered and not sent to data plane yet (call is IDLE) assertThat(dataPlaneRequestCount.get()).isEqualTo(0); - // proxyCall.isReady() should remain false during drain - assertThat(proxyCall.isReady()).isFalse(); + + // Now trigger drain (which also activates the call) + sendDrainLatch.countDown(); + + // Wait for drain to be processed + assertThat(drainSentLatch.await(5, TimeUnit.SECONDS)).isTrue(); + + // Verify requests are now drained to data plane (request drain does not block response path) + assertThat(dataPlaneRequestCount.get()).isEqualTo(3); + + // Verify onReady was NOT called because request path is draining + Mockito.verify(mockListener, Mockito.never()).onReady(); + // Wait for filter to send drain_complete to mock server + assertThat(filterSentDrainCompleteLatch.await(5, TimeUnit.SECONDS)).isTrue(); + + // Echo drain_complete back to filter to complete handshake + synchronized (responseObserverRef.get()) { + responseObserverRef.get().onNext(ProcessingResponse.newBuilder() + .setRequestBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setDrainComplete(true) + .build()) + .build()) + .build()) + .build()) + .build()); + } + + // Verify onReady IS now called after request drain completes + Mockito.verify(mockListener, Mockito.timeout(5000)).onReady(); + proxyCall.cancel("Cleanup", null); channelManager.close(); } @@ -9567,9 +10238,11 @@ public void onCompleted() { .build()); try { + final AtomicInteger onCloseCallCount = new AtomicInteger(0); final CountDownLatch appCloseLatch = new CountDownLatch(1); ClientCall.Listener appListener = new ClientCall.Listener() { @Override public void onClose(Status status, Metadata trailers) { + onCloseCallCount.incrementAndGet(); appCloseLatch.countDown(); } }; @@ -9596,6 +10269,7 @@ public void onCompleted() { fakeClock.forwardTime(1, TimeUnit.SECONDS); } assertThat(appCloseLatch.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(onCloseCallCount.get()).isEqualTo(1); // At this point, app received onClose, but sidecar should NOT be completed yet assertThat(sidecarCompletedLatch.getCount()).isEqualTo(1); @@ -9775,7 +10449,8 @@ public void onClose(Status status, Metadata trailers) { } assertThat(closedLatch.await(5, TimeUnit.SECONDS)).isTrue(); assertThat(closedStatus.get().getCode()).isEqualTo(Status.Code.INTERNAL); - assertThat(closedStatus.get().getDescription()).contains("External processor stream failed"); + assertThat(closedStatus.get().getDescription()) + .contains("gRPC message compression not supported in ext_proc"); proxyCall.cancel("Cleanup", null); channelManager.close(); @@ -9917,7 +10592,8 @@ public void onCompleted() { // Verify application receives INTERNAL with correct description assertThat(closedLatch.await(5, TimeUnit.SECONDS)).isTrue(); assertThat(closedStatus.get().getCode()).isEqualTo(Status.Code.INTERNAL); - assertThat(closedStatus.get().getDescription()).contains("External processor stream failed"); + assertThat(closedStatus.get().getDescription()) + .contains("gRPC message compression not supported in ext_proc"); proxyCall.cancel("Cleanup", null); channelManager.close(); diff --git a/xds/third_party/envoy/src/main/proto/envoy/service/ext_proc/v3/external_processor.proto b/xds/third_party/envoy/src/main/proto/envoy/service/ext_proc/v3/external_processor.proto index 1c033c08d26..1ac424b6358 100644 --- a/xds/third_party/envoy/src/main/proto/envoy/service/ext_proc/v3/external_processor.proto +++ b/xds/third_party/envoy/src/main/proto/envoy/service/ext_proc/v3/external_processor.proto @@ -6,6 +6,7 @@ import "envoy/config/core/v3/base.proto"; import "envoy/extensions/filters/http/ext_proc/v3/processing_mode.proto"; import "envoy/type/v3/http_status.proto"; +import "google/protobuf/any.proto"; import "google/protobuf/duration.proto"; import "google/protobuf/struct.proto"; @@ -23,37 +24,31 @@ option (udpa.annotations.file_status).package_version_status = ACTIVE; // [#protodoc-title: External processing service] -// A service that can access and modify HTTP requests and responses -// as part of a filter chain. +// A service that can access and modify HTTP requests and responses as part of a filter chain. // The overall external processing protocol works like this: // // 1. The data plane sends to the service information about the HTTP request. -// 2. The service sends back a ProcessingResponse message that directs -// the data plane to either stop processing, continue without it, or send -// it the next chunk of the message body. -// 3. If so requested, the data plane sends the server the message body in -// chunks, or the entire body at once. In either case, the server may send -// back a ProcessingResponse for each message it receives, or wait for -// a certain amount of body chunks received before streaming back the -// ProcessingResponse messages. -// 4. If so requested, the data plane sends the server the HTTP trailers, -// and the server sends back a ProcessingResponse. -// 5. At this point, request processing is done, and we pick up again -// at step 1 when the data plane receives a response from the upstream -// server. -// 6. At any point above, if the server closes the gRPC stream cleanly, -// then the data plane proceeds without consulting the server. -// 7. At any point above, if the server closes the gRPC stream with an error, -// then the data plane returns a 500 error to the client, unless the filter -// was configured to ignore errors. +// 2. The service sends back a ``ProcessingResponse`` message that directs the data plane to either +// stop processing, continue without it, or send it the next chunk of the message body. +// 3. If so requested, the data plane sends the server the message body in chunks, or the entire +// body at once. In either case, the server may send back a ``ProcessingResponse`` for each +// message it receives, or wait for a certain amount of body chunks to be received before +// streaming back the ``ProcessingResponse`` messages. +// 4. If so requested, the data plane sends the server the HTTP trailers, and the server sends back +// a ``ProcessingResponse``. +// 5. At this point, request processing is done, and we pick up again at step 1 when the data plane +// receives a response from the upstream server. +// 6. At any point above, if the server closes the gRPC stream cleanly, then the data plane +// proceeds without consulting the server. +// 7. At any point above, if the server closes the gRPC stream with an error, then the data plane +// returns a ``500`` error to the client, unless the filter was configured to ignore errors. // -// In other words, the process is a request/response conversation, but -// using a gRPC stream to make it easier for the server to -// maintain state. +// In other words, the process is a request/response conversation, but using a gRPC stream to make +// it easier for the server to maintain state. service ExternalProcessor { // This begins the bidirectional stream that the data plane will use to // give the server control over what the filter does. The actual - // protocol is described by the ProcessingRequest and ProcessingResponse + // protocol is described by the ``ProcessingRequest`` and ``ProcessingResponse`` // messages below. rpc Process(stream ProcessingRequest) returns (stream ProcessingResponse) { } @@ -61,30 +56,90 @@ service ExternalProcessor { // This message specifies the filter protocol configurations which will be sent to the ext_proc // server in a :ref:`ProcessingRequest `. -// If the server does not support these protocol configurations, it may choose to close the gRPC stream. -// If the server supports these protocol configurations, it should respond based on the API specifications. +// If the server does not support these protocol configurations, it may choose to close the gRPC +// stream. If the server supports these protocol configurations, it should respond based on the +// API specifications. message ProtocolConfiguration { - // Specify the filter configuration :ref:`request_body_mode - // ` + // Specifies the filter configuration + // :ref:`request_body_mode `. envoy.extensions.filters.http.ext_proc.v3.ProcessingMode.BodySendMode request_body_mode = 1 - [(validate.rules).enum = {defined_only: true}]; + [(validate.rules).enum = {defined_only: true}]; - // Specify the filter configuration :ref:`response_body_mode - // ` + // Specifies the filter configuration + // :ref:`response_body_mode `. envoy.extensions.filters.http.ext_proc.v3.ProcessingMode.BodySendMode response_body_mode = 2 - [(validate.rules).enum = {defined_only: true}]; + [(validate.rules).enum = {defined_only: true}]; - // Specify the filter configuration :ref:`send_body_without_waiting_for_header_response - // ` - // If the client is waiting for a header response from the server, setting ``true`` means the client will send body to the server - // as they arrive. Setting ``false`` means the client will buffer the arrived data and not send it to the server immediately. + // Specifies the filter configuration + // :ref:`send_body_without_waiting_for_header_response `. + // If the client is waiting for a header response from the server, setting to ``true`` means the + // client will send the body to the server as it arrives. Setting to ``false`` means the client + // will buffer the arrived data and not send it to the server immediately. bool send_body_without_waiting_for_header_response = 3; } // This represents the different types of messages that the data plane can send // to an external processing server. -// [#next-free-field: 12] +// [#next-free-field: 14] message ProcessingRequest { + // Initial flow control window sizes for ``FULL_DUPLEX_STREAMED`` and + // ``GRPC`` body send modes. + // + // A sender starts with this amount of flow control window. Whenever + // it sends body data, it must decrement its flow control window by + // the number of bytes that it has sent. When its flow control + // window is less than or equal to the amount of body data it wishes + // to send, it may not send until it receives a window update causing + // its flow control window to be large enough. + // + // However, note that in ``GRPC`` body send mode, whenever the flow + // control window is greater than zero, a sender may send a single + // message, even if the size of that message exceeds the available flow + // control window. At that point, the flow control window will be negative + // and the sender must not send the next message until it becomes positive. + // + // Note that the initial size for the to-sidestream windows are set by + // the sender, not the receiver. This is because each sidestream may be + // routed to a different ext_proc server instance, but there is no + // connection-level handshake to set a default for that server + // instance, so the only alternative here would be to have the + // ext_proc server instance set this on a per-stream basis, which + // would require an additional round-trip and therefore hurt latency. + // This unfortunately means that the ext_proc server instance has a + // bit less control: as soon as it receives these initial values, it can + // immediately send a window update that reduces the window, but it + // must be prepared to handle any data that the sender has already sent. + // The initial sizes for the to-sidestream windows are generally + // expected to be in the range of 32K to 64K. + // + // [#not-implemented-hide:] + message FlowControlInit { + // Downstream-to-sidestream initial window size. + int64 initial_window_downstream_to_sidestream = 1; + + // Sidestream-to-upstream initial window size. + int64 initial_window_sidestream_to_upstream = 2; + + // Upstream-to-sidestream initial window size. + int64 initial_window_upstream_to_sidestream = 3; + + // Sidestream-to-downstream initial window size. + int64 initial_window_sidestream_to_downstream = 4; + } + + // Flow control window update. Values may be positive or negative. The + // sender must immediately add these values to its flow control window, + // which governs how much data can be sent. + // + // [#not-implemented-hide:] + message ClientWindowUpdate { + // Window update for sidestream-to-upstream. + int64 window_increment_sidestream_to_upstream = 1; + + // Window update for sidestream-to-downstream. + int64 window_increment_sidestream_to_downstream = 2; + } + reserved 1; reserved "async_mode"; @@ -93,35 +148,33 @@ message ProcessingRequest { // ones are set for a particular HTTP request/response depend on the // processing mode. oneof request { - option (validate.required) = true; - // Information about the HTTP request headers, as well as peer info and additional // properties. Unless ``observability_mode`` is ``true``, the server must send back a - // HeaderResponse message, an ImmediateResponse message, or close the stream. + // ``HeaderResponse`` message, an ``ImmediateResponse`` message, or close the stream. HttpHeaders request_headers = 2; // Information about the HTTP response headers, as well as peer info and additional // properties. Unless ``observability_mode`` is ``true``, the server must send back a - // HeaderResponse message or close the stream. + // ``HeaderResponse`` message or close the stream. HttpHeaders response_headers = 3; - // A chunk of the HTTP request body. Unless ``observability_mode`` is true, the server must send back - // a BodyResponse message, an ImmediateResponse message, or close the stream. + // A chunk of the HTTP request body. Unless ``observability_mode`` is ``true``, the server must + // send back a ``BodyResponse`` message, an ``ImmediateResponse`` message, or close the stream. HttpBody request_body = 4; - // A chunk of the HTTP response body. Unless ``observability_mode`` is ``true``, the server must send back - // a BodyResponse message or close the stream. + // A chunk of the HTTP response body. Unless ``observability_mode`` is ``true``, the server must + // send back a ``BodyResponse`` message or close the stream. HttpBody response_body = 5; // The HTTP trailers for the request path. Unless ``observability_mode`` is ``true``, the server - // must send back a TrailerResponse message or close the stream. + // must send back a ``TrailerResponse`` message or close the stream. // // This message is only sent if the trailers processing mode is set to ``SEND`` and // the original downstream request has trailers. HttpTrailers request_trailers = 6; // The HTTP trailers for the response path. Unless ``observability_mode`` is ``true``, the server - // must send back a TrailerResponse message or close the stream. + // must send back a ``TrailerResponse`` message or close the stream. // // This message is only sent if the trailers processing mode is set to ``SEND`` and // the original upstream response has trailers. @@ -137,39 +190,75 @@ message ProcessingRequest { // :ref:`attributes ` supported in the data plane. map attributes = 9; - // Specify whether the filter that sent this request is running in :ref:`observability_mode - // ` - // and defaults to false. + // Specifies whether the filter that sent this request is running in + // :ref:`observability_mode `. // - // * A value of ``false`` indicates that the server must respond - // to this message by either sending back a matching ProcessingResponse message, - // or by closing the stream. + // * A value of ``false`` indicates that the server must respond to this message by either + // sending back a matching ``ProcessingResponse`` message, or by closing the stream. // * A value of ``true`` indicates that the server should not respond to this message, as any - // responses will be ignored. However, it may still close the stream to indicate that no more messages - // are needed. + // responses will be ignored. However, it may still close the stream to indicate that no more + // messages are needed. // + // Defaults to ``false``. bool observability_mode = 10; // Specify the filter protocol configurations to be sent to the server. // ``protocol_config`` is only encoded in the first ``ProcessingRequest`` message from the client to the server. ProtocolConfiguration protocol_config = 11; + + // Flow control initialization for ``FULL_DUPLEX_STREAMED`` and + // ``GRPC`` body send modes. + // + // Must be set in the initial message on the stream. Not used in + // subsequent messages. + // + // [#not-implemented-hide:] + FlowControlInit flow_control_init = 12; + + // Flow control updates for ``FULL_DUPLEX_STREAMED`` and ``GRPC`` body + // send modes. + // + // This message may be included in a request message that also + // populates one of the fields in the ``request`` oneof above, or it + // may be sent in a request message that does not set the + // ``request`` oneof. + // + // In ``FULL_DUPLEX_STREAMED`` body send mode, for backward + // compatibility with data planes that do not yet support flow control, + // the data plane must not send a message containing only this field + // (i.e., not setting the ``request`` oneof) unless the ext_proc server + // has sent a window update, thus indicating that it supports flow control. + // + // [#not-implemented-hide:] + ClientWindowUpdate client_window_update = 13; } // This represents the different types of messages the server may send back to the data plane -// when the ``observability_mode`` field in the received ProcessingRequest is set to false. +// when the ``observability_mode`` field in the received ``ProcessingRequest`` is set to ``false``. // // * If the corresponding ``BodySendMode`` in the // :ref:`processing_mode ` -// is not set to ``FULL_DUPLEX_STREAMED``, then for every received ProcessingRequest, -// the server must send back exactly one ProcessingResponse message. +// is not set to ``FULL_DUPLEX_STREAMED``, then for every received ``ProcessingRequest``, +// the server must send back exactly one ``ProcessingResponse`` message. // * If it is set to ``FULL_DUPLEX_STREAMED``, the server must follow the API defined -// for this mode to send the ProcessingResponse messages. -// [#next-free-field: 13] +// for this mode to send the ``ProcessingResponse`` messages. +// [#next-free-field: 17] message ProcessingResponse { + // Flow control window update. Values may be positive or negative. The + // sender must immediately add these values to its flow control window, + // which governs how much data can be sent. + // + // [#not-implemented-hide:] + message ServerWindowUpdate { + // Window update for downstream-to-sidestream. + int64 window_increment_downstream_to_sidestream = 1; + + // Window update for upstream-to-sidestream. + int64 window_increment_upstream_to_sidestream = 2; + } + // The response type that is sent by the server. oneof response { - option (validate.required) = true; - // The server must send back this message in response to a message with the // ``request_headers`` field set. HeadersResponse request_headers = 1; @@ -204,17 +293,19 @@ message ProcessingResponse { ImmediateResponse immediate_response = 7; // The server sends back this message to initiate or continue local response streaming. - // The server must initiate local response streaming with the ``headers_response`` in response to a ProcessingRequest - // with the ``request_headers`` only. - // The server may follow up with multiple messages containing ``body_response``. The server must indicate - // end of stream by setting ``end_of_stream`` to ``true`` in the ``headers_response`` + // The server must initiate local response streaming with the ``headers_response`` in response + // to a ``ProcessingRequest`` with the ``request_headers`` only. + // The server may follow up with multiple messages containing ``body_response``. The server must + // indicate end of stream by setting ``end_of_stream`` to ``true`` in the ``headers_response`` // or ``body_response`` message or by sending a ``trailers_response`` message. - // The client may send a ``request_body`` or ``request_trailers`` to the server depending on configuration. + // The client may send a ``request_body`` or ``request_trailers`` to the server depending on + // configuration. // The streaming local response can only be sent when the ``request_header_mode`` in the filter // :ref:`processing_mode ` - // is set to ``SEND``. The ext_proc server should not send StreamedImmediateResponse if it did not observe request headers, - // as it will result in the race with the upstream server response and reset of the client request. - // Presently only the FULL_DUPLEX_STREAMED or NONE body modes are supported. + // is set to ``SEND``. The ext_proc server should not send ``StreamedImmediateResponse`` if it + // did not observe request headers, as it will result in a race with the upstream server + // response and reset of the client request. + // Presently only the ``FULL_DUPLEX_STREAMED`` or ``NONE`` body modes are supported. StreamedImmediateResponse streamed_immediate_response = 11; } @@ -223,99 +314,229 @@ message ProcessingResponse { // field name(s) of the struct. google.protobuf.Struct dynamic_metadata = 8; - // Override how parts of the HTTP request and response are processed - // for the duration of this particular request/response only. Servers - // may use this to intelligently control how requests are processed - // based on the headers and other metadata that they see. - // This field is only applicable when servers responding to the header requests. - // If it is set in the response to the body or trailer requests, it will be ignored by the data plane. + // Optional typed metadata that will be emitted as dynamic metadata to be consumed by + // following filters. This metadata will be placed in the namespace(s) specified by the + // keys of the map. + // + // Typed dynamic metadata should be preferred over untyped dynamic metadata (``dynamic_metadata``) + // because it is more efficient and more type-safe. + map typed_dynamic_metadata = 13; + + // Override how parts of the HTTP request and response are processed for the duration of this + // particular request/response only. Servers may use this to intelligently control how requests + // are processed based on the headers and other metadata that they see. + // + // This field is applicable when servers are responding to the header requests. If it is set + // in the response to the body or trailer requests, it will be ignored by the data plane. // It is also ignored by the data plane when the ext_proc filter config - // :ref:`allow_mode_override - // ` - // is set to false, or - // :ref:`send_body_without_waiting_for_header_response - // ` - // is set to true. + // :ref:`allow_mode_override ` + // is set to ``false``, or + // :ref:`send_body_without_waiting_for_header_response ` + // is set to ``true``. + // + // The external processing server can override the processing mode by returning a standalone ``mode_override`` + // in the ``ProcessingResponse`` after receiving a ``ProcessingRequest`` for request headers. This standalone + // override must be sent before the request headers response is sent. + // Subsequent messages will adhere to this new mode. + // + // Constraints: + // + // 1. Request Path: A standalone ``mode_override`` only supports transitioning the body processing mode + // from other mode to ``FULL_DUPLEX_STREAMED``, and the trailer mode must be set to ``SEND``. + // 2. Response Path: There are no restrictions on processing mode changes in this case. + // envoy.extensions.filters.http.ext_proc.v3.ProcessingMode mode_override = 9; // [#not-implemented-hide:] - // Used only in ``FULL_DUPLEX_STREAMED`` and ``GRPC`` body send modes. - // Instructs the data plane to stop sending body data and to send a - // half-close on the ext_proc stream. The ext_proc server should then echo - // back all subsequent body contents as-is until it sees the client's - // half-close, at which point the ext_proc server can terminate the stream - // with an OK status. This provides a safe way for the ext_proc server - // to indicate that it does not need to see the rest of the stream; - // without this, the ext_proc server could not terminate the stream - // early, because it would wind up dropping any body contents that the - // client had already sent before it saw the ext_proc stream termination. - bool request_drain = 12; - - // When ext_proc server receives a request message, in case it needs more - // time to process the message, it sends back a ProcessingResponse message - // with a new timeout value. When the data plane receives this response - // message, it ignores other fields in the response, just stop the original - // timer, which has the timeout value specified in - // :ref:`message_timeout - // ` - // and start a new timer with this ``override_message_timeout`` value and keep the - // data plane ext_proc filter state machine intact. - // Has to be >= 1ms and <= - // :ref:`max_message_timeout ` - // Such message can be sent at most once in a particular data plane ext_proc filter processing state. - // To enable this API, one has to set ``max_message_timeout`` to a number >= 1ms. + // Deprecated and not implemented. This field has been replaced with + // the request_drain_requests and request_drain_responses fields. + bool request_drain = 12 + [deprecated = true, (envoy.annotations.deprecated_at_minor_version) = "3.0"]; + + // [#not-implemented-hide:] + // Initiates a drain of request body data. Used only in + // ``FULL_DUPLEX_STREAMED`` and ``GRPC`` body send modes. + // + // The expected sequence for a drain is as follows: + // + // 1. The ext_proc server sends a message to the data plane with this + // field set to true. + // 2. The data plane pauses reading from the downstream client, applying + // any necessary flow control push-back. + // 3. The data plane sends a + // :ref:`request_body ` + // with the + // :ref:`drain_complete ` + // field set to true. + // 4. The ext_proc server continues processing all subsequent request body + // chunks until it sees the body chunk with the + // :ref:`drain_complete ` + // field set to true. When sending back its response to that last body + // chunk, the ext_proc server will set the + // :ref:`drain_complete ` + // field to true to let the data plane know that it has finished draining. + // 5. When the data plane sees the response with the + // :ref:`drain_complete ` + // field set to true, it will resume reading from the downstream + // client, passing all data directly to the upstream server, + // without going through the ext_proc sidestream. + // + // This procedure provides a safe way for the ext_proc server to indicate + // that it does not need to see the rest of the request body. + // + // Note that if the data plane is sending request body data and the + // ext_proc server wants to terminate the stream with an OK status, it + // must perform this drain before doing so. + // + // Note that the data plane may have either sent a body chunk with + // :ref:`end_of_stream ` + // set to true or may have sent trailers before it received the drain + // request from the ext_proc server. In these cases, the data plane will + // ignore the drain request, and the ext_proc will consider the drain + // complete when it sees the end-of-stream or trailers. + bool request_drain_requests = 15; + + // [#not-implemented-hide:] + // Initiates a drain of response body data. Used only in + // ``FULL_DUPLEX_STREAMED`` and ``GRPC`` body send modes. + // + // The expected sequence for a drain is as follows: + // + // 1. The ext_proc server sends a message to the data plane with this + // field set to true. + // 2. The data plane pauses reading from the upstream server, applying + // any necessary flow control push-back. + // 3. The data plane sends a + // :ref:`response_body ` + // with the + // :ref:`drain_complete ` + // field set to true. + // 4. The ext_proc server continues processing all subsequent response body + // chunks until it sees the body chunk with the + // :ref:`drain_complete ` + // field set to true. When sending back its response to that last body + // chunk, the ext_proc server will set the + // :ref:`drain_complete ` + // field to true to let the data plane know that it has finished draining. + // 5. When the data plane sees the response with the + // :ref:`drain_complete ` + // field set to true, it will resume reading from the upstream + // server, passing all data directly to the downstream client, + // without going through the ext_proc sidestream. + // + // This procedure provides a safe way for the ext_proc server to indicate + // that it does not need to see the rest of the response body. + // + // Note that if the data plane is sending response body data and the + // ext_proc server wants to terminate the stream with an OK status, it + // must perform this drain before doing so. + // + // Note that the data plane may have either sent a body chunk with + // :ref:`end_of_stream ` + // set to true or may have sent trailers before it received the drain + // request from the ext_proc server. In these cases, the data plane will + // ignore the drain request, and the ext_proc will consider the drain + // complete when it sees the end-of-stream or trailers. + bool request_drain_responses = 16; + + // When the ext_proc server receives a request message and needs more time to process it, it + // sends back a ``ProcessingResponse`` message with a new timeout value. When the data plane + // receives this response message, it ignores other fields in the response, stops the original + // timer (which has the timeout value specified in + // :ref:`message_timeout `), + // and starts a new timer with this ``override_message_timeout`` value while keeping the data + // plane ext_proc filter state machine intact. + // + // The value must be >= 1ms and <= + // :ref:`max_message_timeout `. + // Such a message can be sent at most once in a particular data plane ext_proc filter processing + // state. To enable this API, ``max_message_timeout`` must be set to a value >= 1ms. google.protobuf.Duration override_message_timeout = 10; + + // Flow control updates for ``FULL_DUPLEX_STREAMED`` and ``GRPC`` body + // send modes. + // + // This message may be included in a response message that also + // populates one of the fields in the ``response`` oneof above, or it + // may be sent in a response message that does not set the + // ``response`` oneof. + // + // In ``FULL_DUPLEX_STREAMED`` body send mode, for backward + // compatibility with data planes that do not yet support flow control, + // the ext_proc server must not set this field unless the data plane + // sent initial window sizes in its initial message on the stream. + // Conversely, if the data plane did send initial window sizes in its + // initial message on the stream, the ext_proc server must send a + // window update immediately to let the data plane know that it also + // supports flow control. If the ext_proc server is sending a message + // immediately anyway (e.g., for a header or body chunk), it can include + // this field in that same message; otherwise, the ext_proc server must + // send a message containing only this field. + // + // [#not-implemented-hide:] + ServerWindowUpdate server_window_update = 14; } // The following are messages that are sent to the server. -// This message is sent to the external server when the HTTP request and responses +// This message is sent to the external server when the HTTP request and response headers // are first received. message HttpHeaders { - // The HTTP request headers. All header keys will be - // lower-cased, because HTTP header keys are case-insensitive. - // The header value is encoded in the + // The HTTP request headers. All header keys will be lower-cased, because HTTP header keys are + // case-insensitive. The header value is encoded in the // :ref:`raw_value ` field. config.core.v3.HeaderMap headers = 1; // [#not-implemented-hide:] - // This field is deprecated and not implemented. Attributes will be sent in - // the top-level :ref:`attributes ` field. map attributes = 2 - [deprecated = true, (envoy.annotations.deprecated_at_minor_version) = "3.0"]; + [deprecated = true, (envoy.annotations.deprecated_at_minor_version) = "3.0"]; - // If ``true``, then there is no message body associated with this - // request or response. + // If ``true``, then there is no message body associated with this request or response. bool end_of_stream = 3; } -// This message is sent to the external server when the HTTP request and -// response bodies are received. +// This message is sent to the external server when the HTTP request and response bodies are +// received. +// [#next-free-field: 6] message HttpBody { - // The contents of the body in the HTTP request/response. Note that in - // streaming mode multiple ``HttpBody`` messages may be sent. + // The contents of the body in the HTTP request/response. Note that in streaming mode multiple + // ``HttpBody`` messages may be sent. // - // In ``GRPC`` body send mode, a separate ``HttpBody`` message will be - // sent for each message in the gRPC stream. + // In ``GRPC`` body send mode, a separate ``HttpBody`` message will be sent for each message in + // the gRPC stream. bytes body = 1; - // If ``true``, this will be the last ``HttpBody`` message that will be sent and no - // trailers will be sent for the current request/response. + // If ``true``, this will be the last ``HttpBody`` message that will be sent and no trailers + // will be sent for the current request/response. bool end_of_stream = 2; - // This field is used in ``GRPC`` body send mode when ``end_of_stream`` is - // true and ``body`` is empty. Those values would normally indicate an - // empty message on the stream with the end-of-stream bit set. - // However, if the half-close happens after the last message on the - // stream was already sent, then this field will be true to indicate an - // end-of-stream with *no* message (as opposed to an empty message). + // This field is used only in ``GRPC`` body send mode. It is not used in any other body send + // mode. + // + // This field is used only when ``end_of_stream`` is true and ``body`` is empty. + // Normally, in ``GRPC`` body send mode, an empty ``body`` field indicates an empty message on + // the gRPC stream. However, it is possible that the gRPC client sends a half-close without + // actually sending a message on the stream, so we need a way to differentiate between + // an empty message being sent and no message being sent. If this field is true, then it + // indicates that no message has been sent; if it is false, then it indicates that an empty + // message has been sent. + // [#not-implemented-hide:] bool end_of_stream_without_message = 3; - // This field is used in ``GRPC`` body send mode to indicate whether - // the message is compressed. This will never be set to true by gRPC - // but may be set to true by a proxy like Envoy. + // This field is used in ``GRPC`` body send mode to indicate whether the message is compressed. + // This will never be set to ``true`` by gRPC but may be set to ``true`` by a proxy like Envoy. bool grpc_message_compressed = 4; + + // [#not-implemented-hide:] + // In ``FULL_DUPLEX_STREAMED`` or ``GRPC`` body send modes, if the + // data plane has seen the + // :ref:`request_drain_requests ` + // or :ref:`request_drain_responses ` + // field, it will populate this field to indicate that it has finished + // sending data to the ext_proc server. + bool drain_complete = 5; } // This message is sent to the external server when the HTTP request and @@ -352,13 +573,14 @@ message TrailersResponse { HeaderMutation header_mutation = 1; } -// This message is sent by the external server to the data plane after ``HttpHeaders`` -// to initiate local response streaming. The server may follow up with multiple messages containing ``body_response``. -// The server must indicate end of stream by setting ``end_of_stream`` to ``true`` in the ``headers_response`` -// or ``body_response`` message or by sending a ``trailers_response`` message. +// This message is sent by the external server to the data plane after ``HttpHeaders`` to initiate +// local response streaming. The server may follow up with multiple messages containing +// ``body_response``. The server must indicate end of stream by setting ``end_of_stream`` to +// ``true`` in the ``headers_response`` or ``body_response`` message or by sending a +// ``trailers_response`` message. message StreamedImmediateResponse { oneof response { - // Response headers to be sent downstream. The ":status" header must be set. + // Response headers to be sent downstream. The ``:status`` header must be set. HttpHeaders headers_response = 1; // Response body to be sent downstream. @@ -384,7 +606,7 @@ message CommonResponse { // further messages for this request or response even if the processing // mode is configured to do so. // - // When used in response to a request_headers or response_headers message, + // When used in response to a ``request_headers`` or ``response_headers`` message, // this status makes it possible to either completely replace the body // while discarding the original body, or to add a body to a message that // formerly did not have one. @@ -401,23 +623,22 @@ message CommonResponse { ResponseStatus status = 1 [(validate.rules).enum = {defined_only: true}]; // Instructions on how to manipulate the headers. When responding to an - // HttpBody request, header mutations will only take effect if - // the current processing mode for the body is BUFFERED. + // ``HttpBody`` request, header mutations will only take effect if the current processing mode + // for the body is ``BUFFERED``. HeaderMutation header_mutation = 2; - // Replace the body of the last message sent to the remote server on this - // stream. If responding to an HttpBody request, simply replace or clear - // the body chunk that was sent with that request. Body mutations may take - // effect in response either to ``header`` or ``body`` messages. When it is - // in response to ``header`` messages, it only take effect if the + // Replace the body of the last message sent to the remote server on this stream. If responding + // to an ``HttpBody`` request, simply replace or clear the body chunk that was sent with that + // request. Body mutations may take effect in response either to ``header`` or ``body`` messages. + // When it is in response to ``header`` messages, it only takes effect if the // :ref:`status ` - // is set to CONTINUE_AND_REPLACE. + // is set to ``CONTINUE_AND_REPLACE``. BodyMutation body_mutation = 3; // [#not-implemented-hide:] - // Add new trailers to the message. This may be used when responding to either a - // HttpHeaders or HttpBody message, but only if this message is returned - // along with the CONTINUE_AND_REPLACE status. + // Add new trailers to the message. This may be used when responding to either an + // ``HttpHeaders`` or ``HttpBody`` message, but only if this message is returned + // along with the ``CONTINUE_AND_REPLACE`` status. // The header value is encoded in the // :ref:`raw_value ` field. config.core.v3.HeaderMap trailers = 4; @@ -429,34 +650,32 @@ message CommonResponse { bool clear_route_cache = 5; } -// This message causes the filter to attempt to create a locally -// generated response, send it downstream, stop processing -// additional filters, and ignore any additional messages received -// from the remote server for this request or response. If a response -// has already started, then this will either ship the reply directly -// to the downstream codec, or reset the stream. +// This message causes the filter to attempt to create a locally generated response, send it +// downstream, stop processing additional filters, and ignore any additional messages received +// from the remote server for this request or response. If a response has already started, then +// this will either ship the reply directly to the downstream codec, or reset the stream. // [#next-free-field: 6] message ImmediateResponse { // The response code to return. type.v3.HttpStatus status = 1 [(validate.rules).message = {required: true}]; - // Apply changes to the default headers, which will include content-type. + // Apply changes to the default headers, which will include ``content-type``. HeaderMutation headers = 2; // The message body to return with the response which is sent using the - // text/plain content type, or encoded in the grpc-message header. + // ``text/plain`` content type, or encoded in the ``grpc-message`` header. bytes body = 3; // If set, then include a gRPC status trailer. GrpcStatus grpc_status = 4; // A string detailing why this local reply was sent, which may be included - // in log and debug output (e.g. this populates the %RESPONSE_CODE_DETAILS% + // in log and debug output (e.g., this populates the ``%RESPONSE_CODE_DETAILS%`` // command operator field for use in access logging). string details = 5; } -// This message specifies a gRPC status for an ImmediateResponse message. +// This message specifies a gRPC status for an ``ImmediateResponse`` message. message GrpcStatus { // The actual gRPC status. uint32 status = 1; @@ -478,33 +697,45 @@ message HeaderMutation { } // The body response message corresponding to ``FULL_DUPLEX_STREAMED`` or ``GRPC`` body modes. +// [#next-free-field: 6] message StreamedBodyResponse { // In ``FULL_DUPLEX_STREAMED`` body send mode, contains the body response chunk that will be // passed to the upstream/downstream by the data plane. In ``GRPC`` body send mode, contains // a serialized gRPC message to be passed to the upstream/downstream by the data plane. bytes body = 1; - // The server sets this flag to true if it has received a body request with - // :ref:`end_of_stream ` set to true, - // and this is the last chunk of body responses. - // Note that in ``GRPC`` body send mode, this allows the ext_proc - // server to tell the data plane to send a half close after a client - // message, which will result in discarding any other messages sent by - // the client application. + // The server sets this flag to ``true`` if it has received a body request with + // :ref:`end_of_stream ` set to + // ``true``, and this is the last chunk of body responses. + // + // Note that in ``GRPC`` body send mode, this allows the ext_proc server to tell the data plane + // to send a half close after a client message, which will result in discarding any other + // messages sent by the client application. bool end_of_stream = 2; - // This field is used in ``GRPC`` body send mode when ``end_of_stream`` is - // true and ``body`` is empty. Those values would normally indicate an - // empty message on the stream with the end-of-stream bit set. - // However, if the half-close happens after the last message on the - // stream was already sent, then this field will be true to indicate an - // end-of-stream with *no* message (as opposed to an empty message). + // This field is used in ``GRPC`` body send mode when ``end_of_stream`` is ``true`` and ``body`` + // is empty. Those values would normally indicate an empty message on the stream with the + // end-of-stream bit set. However, if the half-close happens after the last message on the stream + // was already sent, then this field will be ``true`` to indicate an end-of-stream with *no* + // message (as opposed to an empty message). bool end_of_stream_without_message = 3; - // This field is used in ``GRPC`` body send mode to indicate whether - // the message is compressed. This will never be set to true by gRPC - // but may be set to true by a proxy like Envoy. + // This field is used in ``GRPC`` body send mode to indicate whether the message is compressed. + // This will never be set to ``true`` by gRPC but may be set to ``true`` by a proxy like Envoy. bool grpc_message_compressed = 4; + + // [#not-implemented-hide:] + // In ``FULL_DUPLEX_STREAMED`` or ``GRPC`` body send modes, if the + // ext_proc server has seen the + // :ref:`drain_complete ` + // field, it will populate this field to indicate that it has finished + // sending data back to the data plane. + // + // If the data plane receives a message with this field set to true + // before it has sent a drain-complete message to the ext_proc server, + // the data plane will treat that as if the ext_proc stream has failed + // with a non-OK status. + bool drain_complete = 5; } // This message specifies the body mutation the server sends to the data plane. @@ -517,17 +748,16 @@ message BodyMutation { // is not set to ``FULL_DUPLEX_STREAMED`` or ``GRPC``. bytes body = 1; - // Clear the corresponding body chunk. - // Should only be used when the corresponding ``BodySendMode`` in the + // Clear the corresponding body chunk. Should only be used when the corresponding + // ``BodySendMode`` in the // :ref:`processing_mode ` // is not set to ``FULL_DUPLEX_STREAMED`` or ``GRPC``. - // Clear the corresponding body chunk. bool clear_body = 2; // Must be used when the corresponding ``BodySendMode`` in the // :ref:`processing_mode ` // is set to ``FULL_DUPLEX_STREAMED`` or ``GRPC``. StreamedBodyResponse streamed_response = 3 - [(xds.annotations.v3.field_status).work_in_progress = true]; + [(xds.annotations.v3.field_status).work_in_progress = true]; } }