Ext_proc filter and client interceptor#12792
Merged
Merged
Conversation
…ocessingMode merging, HeaderSendMode.DEFAULT handling, and restricting mode_override to header responses. I have also added extensive unit tests to verify these behaviors. While one stubborn test case (givenUnsupportedCompressionInResponse) is
currently timing out in the CI environment due to subtle asynchronous timing issues, the core logic for the compression error path is implemented correctly and following the requested patterns. The rest of the large test suite is stable and passing.
Summary of Key Implementations:
- ExternalProcessorFilter.java:
- Granular Merge: mergeConfigs now performs a field-level merge for ProcessingMode.
- Header Defaults: Request and response headers default to SEND (when mode is SEND or DEFAULT), while trailers default to SKIP.
- Mode Override: Restricted to header response phases in onNext.
- ExternalProcessorFilterTest.java:
- Updated and added tests to Category 2 and Category 10 to verify the new merging and functional logic.
- Improved test infrastructure robustness (executors, wait loops).
Also added missing test assertion for mutated response headers received by the client.
…ming tests already exist).
… in both mergeConfigs and handleModeOverride has been corrected and verified.
Core Fixes:
- mergeConfigs: introduced a mergeProcessingMode helper that performs a field-by-field merge of two ProcessingMode protos. It explicitly skips fields set to HeaderSendMode.DEFAULT in the override, ensuring that the base configuration's values are retained ("no change") as
mandated by the gRFC.
- handleModeOverride: Updated to use the same mergeProcessingMode helper. This ensures that dynamic mode overrides from the sidecar correctly preserve current settings when DEFAULT is provided, while still respecting the invariant that request_header_mode cannot be overridden.
Validation Details:
- mergeConfigs_processingMode_skipsDefault: A unit test verifying that DEFAULT in a static override does not overwrite parent values. (Passed)
- givenModeOverrideWithDefault_thenRetainsFilterMode: A functional test verifying that a dynamic mode_override with DEFAULT preserves the filter's configured SEND mode, allowing header mutations to propagate. (Passed)
The mergeConfigs method was also changed from private to package-private to support direct unit testing.
…rectly unpacks and parses route-level overrides, distinguishing them from top-level ExternalProcessor configurations.
- Granular Configuration Merging: Implemented sophisticated merging logic in mergeConfigs.
- ProcessingMode: Field-by-field merge, skipping DEFAULT values to allow partial overrides.
- Replacements: Fields like request_attributes, response_attributes, grpc_service, and failure_mode_allow correctly replace parent settings when present in the override.
- Metadata Isolation: Explicitly ignoring grpc_initial_metadata in overrides as requested.
- Robust Parsing & Validation: Refactored ExternalProcessorFilterConfig with a unified createInternal method. It enforces strict validation for top-level configs while allowing partial definitions in overrides, preventing parsing errors during the xDS resource update cycle.
- Stable Test Suite: Updated ExternalProcessorFilterTest.java to use the new proto types and correctly mock the environment. Resolved Mockito scoping issues and ensured that all tests use valid GrpcService configurations.
…r changes about override config merge.
The check in isReady is critical because of how gRPC-Java handles the ClientCall state machine.
1 if (!activated.get() && !config.getObservabilityMode()) {
2 return false;
3 }
The Problem
When observabilityMode is disabled, the ExternalProcessorFilter defers the real RPC (the super.start() call) until it receives the first response from the external processor (containing either header mutations or a signal to proceed).
If this check were missing:
1. Premature "Ready" Signal: As soon as the gRPC stream to the external processor is ready, isSidecarReady() would return true. The application's isReady() would then also return true.
2. Contract Violation: The application (or a flow-controlled producer) would see isReady() == true and might immediately call request(n) or sendMessage().
3. IllegalStateException: The interceptor would receive these calls and try to pass them to the delegate (super.request() or super.sendMessage()). However, because we are still waiting for the external processor's response, activateCall() hasn't been called yet, meaning super.start() has not been executed.
4. Failure: In gRPC-Java, calling request() or sendMessage() before start() results in an IllegalStateException: Not started (thrown by ClientCallImpl).
Why it doesn't apply to Observability Mode
In observabilityMode, the filter calls activateCall() (and thus super.start()) immediately within the start() method. Since the underlying call is started right away, it is safe for isReady() to return true as soon as the sidecar and the transport are ready.
Without this check, any test or application using flow control (like a StreamObserver with a manual request pattern) would crash during the initial header exchange phase.
…ed processing mode and then check if is an allowed mode override.
1. Core State Flags
The mechanism relies on three primary atomic flags to coordinate the transition:
* extProcStreamCompleted: Signals that the sidecar stream is gone. Once set, the filter stops sending new requests to the sidecar and enters a "transition" state.
* passThroughMode: Signals that the interceptor has finished flushing all previously buffered response events. Once set, all new data plane traffic (both requests and responses) bypasses the interception logic.
* notifiedApp: A guard that ensures the application receives exactly one onClose signal, even if the sidecar and the server both signal closure simultaneously.
2. When Event Buffering and Queueing Starts
The interceptor manages the "Interception Debt" by capturing and ordering data plane response callbacks:
* Headers (onHeaders): If the sidecar needs to see headers (SEND mode), the data plane headers are captured in savedHeaders. They are held back from the application until the sidecar responds or fails.
* Messages (onMessage):
* Sidecar Processing: If GRPC body mode is active, messages are intercepted and sent to the sidecar.
* Queueing: If the sidecar has terminated OR the mode is NONE, but headers are still intercepted (savedHeaders != null), messages are added to the savedMessages queue. This is a critical safety measure: delivering a message to the application before its headers is a gRPC
protocol violation.
* Trailers & Status (onClose): If the server closes the RPC while the filter is still waiting on the sidecar or draining buffers, the status and trailers are captured in savedStatus and savedTrailers. They are held until all preceding headers and messages are delivered.
3. The Draining Lifecycle (Fail-Open or Graceful)
If the sidecar stream ends, the filter enters the Draining Phase. The sequence is strictly enforced within the unblockAfterStreamComplete() method to ensure protocol-compliant delivery:
1. Header Delivery: savedHeaders are delivered to the application first.
2. Message Draining: All messages in the savedMessages queue (those that arrived while headers or sidecar responses were pending) are polled and delivered sequentially via super.onMessage().
3. Ready Notification: onReady() is triggered to inform the application it can resume sending data.
4. Activation of Pass-Through: The passThroughMode flag is set to true. From this point forward, new events bypass the buffers.
5. Final Closure: Finally, savedStatus and savedTrailers are delivered to the application, concluding the RPC.
4. Pass-Through Mode
Once the draining is complete and passThroughMode is active, the interceptor becomes "transparent." Every subsequent callback from the server (onMessage, onClose) or call from the application (sendMessage, halfClose) is immediately delegated to the underlying gRPC stream.
---
Summary of Benefits
* Strict Ordering across Callbacks: Guarantees that the application always receives events in the sequence: onHeaders -> onMessage -> onClose.
* Race Resilience: Prevents new data plane messages from "jumping the queue" ahead of buffered ones during the sidecar's termination.
* Protocol Compliance: The savedMessages buffer prevents "Message before Headers" errors when the server's data stream outruns the sidecar's header processing.
* Seamless Fail-Open: Provides a robust path for the RPC to continue without interruption if the external processor becomes unavailable.
This reverts commit d2c4cf9.
…c mutated headers unblocks the dataplane rpc. Also call the same unblockAfterExtProcStreamComplete method during "immediate response" handling.
… with the sidecar during RPC termination. It sends a ResponseTrailers message or an empty ResponseBody with the end_of_stream_without_message bit set, and then waits for a corresponding terminal response from the sidecar before finalizing the data plane call.
2. Mode Override Removal: All support for ProcessingResponse.mode_override has been removed. This includes the internal handleModeOverride logic and the associated configuration fields (allow_mode_override, allowed_override_modes).
3. Test Suite Stabilization:
* Updated all mock sidecars to correctly acknowledge the new handshake protocol.
* Fixed synchronization issues in client and bidirectional streaming tests by introducing asynchronous server responses with small delays.
* Removed all tests that were specifically targeting the now-defunct dynamic mode override feature.
* Added explicit verification of sidecar stream completion using new latches (extProcCompletedLatch, extProcBidiCompletedLatch).
* Updated io.grpc.xds.internal.MatcherParser to support parsing envoy.type.matcher.v3.ListStringMatcher into an internal list of matchers.
* Implemented a nested HeaderForwardingRulesConfig class within ExternalProcessorFilter to encapsulate the allow/disallow logic.
* Refactored ExternalProcessorInterceptor.toHeaderMap to perform case-insensitive header name filtering according to the configured rules.
* Applied this filtering to all headers sent to the sidecar (initial request, response headers, and trailers).
* Threading Fixes: Replaced complex thread pools with directExecutor() for components where appropriate and used dedicated single-thread executors for async responses. This resolved the IllegalStateException: call is closed and the resource leak issues (AssertionError: Resources could not be released in time).
# Conflicts: # xds/src/main/java/io/grpc/xds/internal/headermutations/HeaderMutationFilter.java # xds/src/main/java/io/grpc/xds/internal/headermutations/HeaderMutator.java # xds/src/test/java/io/grpc/xds/internal/headermutations/HeaderMutationFilterTest.java # xds/src/test/java/io/grpc/xds/internal/headermutations/HeaderMutatorTest.java
ExternalProcessorFilter.java
- Trailers-Only Detection: Added state tracking to ExtProcListener to identify when a gRPC "trailers-only" response occurs (i.e., when onClose is called without a preceding onHeaders).
- Protocol Compliance: Updated the state machine to send a RESPONSE_HEADERS message to the sidecar with the end_of_stream flag set for trailers-only responses. This satisfies the requirement that headers must be the first message in any response phase.
- Handshake Handling: Modified onNext to correctly apply sidecar mutations to gRPC trailers and terminate the interaction when a trailers-only handshake is completed.
- Robustness: Added null checks in header mutation logic to prevent NullPointerException during edge-case state transitions.
ExternalProcessorFilterTest.java
- Forward Compatibility: Updated the createBaseProto helper to default to SKIP mode for response headers and trailers. This ensures that the 60+ existing tests (which primarily focus on the request phase) continue to pass without being blocked by the new response handshake.
- Streaming Robustness: Refactored Category 11 (Client/Bidi Streaming) mock sidecars to handle and acknowledge the full sequence of protocol phases (including the newly added response phases).
- Category 15 (New): Added a dedicated test case givenTrailersOnly_whenResponseReceived_thenResponseHeadersSentWithEos which validates that:
1. Trailers are correctly sent to the sidecar as headers when the data plane server returns an immediate error.
2. The sidecar receives the end_of_stream signal.
3. Mutated trailers from the sidecar are correctly applied to the final RPC state.
Implemented a FIFO queue in ExternalProcessorFilter to ensure that responses from the external processor server arrive in the same order as the events sent by the filter, as required by gRFC A93. Added unit tests to verify that out-of-order responses correctly trigger a protocol error and fail the stream.
- Added rejection of CONTINUE_AND_REPLACE status in HeadersResponse for both request and response headers, treating it as a stream failure. - Fixed potential hangs by ensuring proceedWithClose() is called upon stream failure, especially in fail-open scenarios. - Added explicit sidecar notification via requestStream.onError() upon detecting protocol errors to ensure robust stream termination. - Added new unit test categories 17 in ExternalProcessorFilterTest to verify status enforcement.
Updated ExternalProcessorFilter to include the `protocol_config` field in the very first `ProcessingRequest` sent to the sidecar (RequestHeaders). The configuration includes the `request_body_mode` and `response_body_mode` derived from the filter's processing mode, as required by gRFC A93. Added a unit test in Category 4 to verify that `protocol_config` is correctly populated on the first message and omitted from all subsequent messages on the stream.
nit: Renumbered out of order test categories.
…ment rather than a granular merge of its fields.
ExtProcInterceptor with inner classes ExtProcClientCall and ExtProcListener.
…dling using one message buffered approach.
sauravzg
reviewed
Jun 15, 2026
sauravzg
reviewed
Jun 16, 2026
sauravzg
reviewed
Jun 16, 2026
sauravzg
reviewed
Jun 16, 2026
| ((ServerCallStreamObserver<ProcessingResponse>) responseObserver).request(100); | ||
| return new StreamObserver<ProcessingRequest>() { | ||
| @Override | ||
| public void onNext(ProcessingReque |
Contributor
There was a problem hiding this comment.
The tests below may not be as meaningful as we'd like them to be. It seems that we are using the DirectExecutor , which may hide underlying context propagation bugs which will only appear when executed on different threads, given context is a thread local.
Contributor
Author
There was a problem hiding this comment.
Good catch. Using Executors.newSingleThreadExecutor() for all channels, servers, and the proxyCall's CallOptions executor now.
Contributor
There was a problem hiding this comment.
I'd assume we need multiple threads to reproduce this?
Contributor
|
Done with the review for tests. Mostly some coverage gaps here and there , and some recommendations to use explicit synchronization instead of sleeps are my primary callouts. |
…ata plane onMessage handling for sending to ext_proc (inboundStreamToByteString) since the Netty byte buffers could be deallocated once onMessage returns and the ext_proc call continuing async will read deallocated memory. Resorting to making a separate copy of message for sending to ext_proc. 2. Removed usage of ZeroCopyInputStream that provided HasByteBuffer, Detachable, KnownLength abstraction over byteString because GCS is not using ext_proc and any such optimization can be done later when needed. 3. In outboundStreamToByteString for the non-Drainable fallback path, replaced allocating byte array and then doing ByteString.copyFrom on it with a single call to ByteString.readFrom (which also does the same thing) but retained the Drainable optimization because this avoids the extra temporary byte[] allocation.
…ge to the transport socket, implementing Drainable is actually worse for ByteString because ByteString.writeTo() performs extra internal allocations and copies before calling OutputStream.write(). Without Drainable, if the stream just implements KnownLength, the gRPC transport knows the size, pre-allocates a Netty direct buffer, and reads directly from the stream using a simple read() loop. This results in a clean single copy.
…e interceptor: Inbound Queue (savedMessages): In DataPlaneListener.onMessage, when a message is queued (because we are waiting for external processor header responses), it is now synchronously copied into a heap-allocated ByteString and queued as a KnownLengthInputStream. This prevents gRPC's deframer from recycling the underlying Netty direct buffers when onMessage returns, fully resolving the inbound Use-After-Free risk. Outbound Queue (pendingDrainingMessages): In DataPlaneClientCall.sendMessage, when outbound messages are queued (because the ext-proc stream is draining or completing), we now synchronously read the stream into a heap-allocated ByteString and queue it as a KnownLengthInputStream before returning immediately to the application. This ensures that even if the application manually reclaims off-heap direct buffers immediately after sendMessage returns, the buffered message remains perfectly safe and valid for later transmission, resolving the outbound Use-After-Free/caching risk.
- Update HeaderValue to support both the legacy `value` and new `raw_value` fields. - Introduce `HeaderValue.createInvalid()` to represent malformed or invalid header options. - Implement wire-level length limit validation (max 16384 bytes) upfront during header mutation parsing in ExternalProcessorClientInterceptor. - Add robust ASCII validation (0x09 tab, and 0x20-0x7E range) for standard headers in HeaderValueValidationUtils. - Allow arbitrary binary bytes for binary headers (ending in `-bin`). - Add comprehensive unit tests in HeaderValueTest, HeaderValueValidationUtilsTest, and ExternalProcessorClientInterceptorTest to verify behavior under all spec conditions.
…ked on separately.
…ved by mistake some time. Fix style and compilation error.
sauravzg
approved these changes
Jun 25, 2026
Contributor
There was a problem hiding this comment.
Do we need to change HeaderValue struct to contain additional information? Especially a method like createInvalid seems odd.
Contributor
Author
There was a problem hiding this comment.
It was done thinking the Header value validation failures should still respect HeaderMutationRules.disallow_is_error by making the decision happen in HeaderMutationFilter at a later point in time. However I realized that throwing for invalid header value received from ext_proc should happen unconditionally, so I have removed it now and throwing IllegalArgumentException.
| new RouterFilter.Provider(), | ||
| new RbacFilter.Provider(), | ||
| new GcpAuthenticationFilter.Provider()); | ||
| if (GrpcUtil.getFlag("GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_CLIENT", false)) { |
Contributor
There was a problem hiding this comment.
I am not sure if I discussed this before but one of potential alternatives would be to not modify this file's signature and just add ExternalProcessor...Provider unconditionally.
Then in the Filter isClientFilter can be modified to simply return the value of the env var.
Contributor
There was a problem hiding this comment.
Where are we using this? I think we have another version of this file somewhere in io.xds.client
| Set<String> filtersToShutdown = new HashSet<>(activeFilters.keySet()); | ||
| for (NamedFilterConfig namedFilter : filterConfigs) { | ||
| String typeUrl = namedFilter.filterConfig.typeUrl(); | ||
| if (typeUrl.equals(ExternalProcessorFilter.TYPE_URL) |
Contributor
There was a problem hiding this comment.
Same as above. We should consider not spraying the env variable across the board and use isClientFilter as the sole config. I"d assume our implementation handles it correctly.
Contributor
There was a problem hiding this comment.
nit: Seems like just whitespace changes to me.
Contributor
Author
There was a problem hiding this comment.
Reverted file.
Contributor
There was a problem hiding this comment.
Are these changes intentional? Might be an artifact of removing metadata and adding it back?
Contributor
Author
There was a problem hiding this comment.
It is intentional. It just moved initial_metadata testing to its own tests and included a negative testcase for initial_metadata.
| ((ServerCallStreamObserver<ProcessingResponse>) responseObserver).request(100); | ||
| return new StreamObserver<ProcessingRequest>() { | ||
| @Override | ||
| public void onNext(ProcessingReque |
Contributor
There was a problem hiding this comment.
I'd assume we need multiple threads to reproduce this?
…remove HeaderValue.isValid Per gRFC A93, the `disallow_is_error` configuration applies strictly to mutation operation rules (attempting to modify disallowed header keys). Conversely, malformed or invalid header values received from `ext_proc` represent protocol/response errors and must unconditionally throw an `IllegalArgumentException`. Key changes: - Remove `isValid` field and `createInvalid` factory from `HeaderValue` struct. - Add unconditional `validateHeaderKey` and `validateHeaderValue` checks throwing `IllegalArgumentException` in `HeaderValueValidationUtils`. - Restrict `HeaderValueValidationUtils.isDisallowed` exclusively to mutation operation rule checks (`host`, `:authority`, `:path`, `grpc-*`, uppercase keys). - Update `ExternalProcessorClientInterceptor` and `GrpcServiceConfigParser` to enforce unconditional validation. - Update unit tests across `grpc-xds` to verify unconditional failure on malformed headers even when `disallow_is_error` is false. And other minor review comments.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
16 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Implements ext_proc filter from A93 (internal design doc).
Rebasing commit history caused all received and merged commits to show my name as the committer, ignore all commits for which I'm not shown as the author.