Repository navigation
xds: External Processor Server Interceptor - #12889
Open
kannanjgithub wants to merge 506 commits into
Open
kannanjgithub wants to merge 506 commits into
kannanjgithub wants to merge 506 commits into
Conversation
grpc#12700) ### Description This PR updates the "Outgoing Flow Control" section in the Manual Flow Control example's README. The previous documentation incorrectly implied that calling `onNext()` on a stream would block if the underlying Netty buffer was full, thereby limiting the send rate. This PR clarifies that `onNext()` does *not* block, but rather queues the messages in memory, which can ultimately lead to an `OutOfMemoryError` if messages are sent too quickly. The updated text correctly advises developers to use `CallStreamObserver.isReady()` to prevent this memory exhaustion, rather than to avoid blocking. Fixes grpc#12657 --------- Co-authored-by: Kannan J <kannanjgithub@google.com>
…pc#12718) This alignment resolves a version skew issue that caused NoClassDefFoundError crashes during instrumentation tests on Firebase Test Lab. Fixes grpc#12703 (comment)
grpc#12705) This PR addresses a race condition where ManagedChannelOrphanWrapper could incorrectly log a "not shutdown properly" warning during garbage collection when using directExecutor(). Changes: Reference Management: Moved phantom.clearSafely() to execute after the super.shutdown() calls to ensure the orphan tracker isn't detached prematurely. Reachability Fence: Added a reachability fence in shutdown() and shutdownNow() to ensure the wrapper remains alive until the methods return, preventing the JIT from marking it for early collection. Regression Test: Added a test case that simulates a reference being held on the stack to verify the fix and prevent future regressions. Testing: Verified with ./gradlew :grpc-core:test --tests ManagedChannelOrphanWrapperTest -PskipAndroid=true. Fixes grpc#12641 --------- Co-authored-by: Kannan J <kannanjgithub@google.com>
…ocessor filter. Eliminated responseLock by leveraging the safeguarded serializing executor from CallOptions, ensuring that all listener callbacks and internal state mutations are strictly serialized. Updated ExtProcClientCall to use this shared executor for the external processor RPC stub.
… executor null-checks.
Updated ExternalProcessorFilterTest to register the interceptor using ManagedChannelBuilder.intercept(), aligning the test environment with production behavior. Removed the manual fallback to directExecutor() in ExternalProcessorFilter, as the framework (ManagedChannelImpl) now guarantees a non-null, safeguarded executor in CallOptions for internal interceptors.
Fix: Resolve lifecycle and race condition issues in External Processor filter.
- Fixed IllegalStateException by moving initial request header transmission out of beforeStart().
- Fixed NullPointerException by providing a fallback to directExecutor() when CallOptions.getExecutor() is null.
- Resolved 'call was half-closed' state machine violation by tracking half-close state and skipping body mutations if the application has already half-closed the call.
- Corrected proto field access for BodyMutation and improved robustness of header mapping.
- Updated unit tests to verify fixes under simulated race conditions using async calls.
Summary of Fixes:
1. Resolved IllegalStateException: Not started:
* Root Cause: The filter was calling onNext() (which triggers sendMessage()) from within the beforeStart() callback of the ClientResponseObserver. In
gRPC-Java, beforeStart() is invoked before the underlying ClientCall.start() has completed, violating the call lifecycle.
* Fix: Moved the initial request headers transmission to immediately after the stub.process() call returns. This ensures the call has officially started
before any messages are sent.
2. Resolved NullPointerException: callExecutor:
* Root Cause: DelayedClientCall requires a non-null executor. When CallOptions.DEFAULT was used in tests, callOptions.getExecutor() returned null,
causing an NPE during filter initialization.
* Fix: Implemented a fallback to MoreExecutors.directExecutor() when the call options do not provide an executor.
3. Resolved IllegalStateException: call was half-closed (Race Condition):
* Root Cause: In unary calls (like blockingUnaryCall), gRPC half-closes the call immediately after sending the request. If the External Processor
returned header mutations after this half-close, the filter would attempt to send a sendMessage to the backend, causing a state machine violation.
* Fix: Added a halfClosed state tracker to the ExtProcClientCall. The filter now gracefully skips sending body mutations or empty messages if the
application has already half-closed the call.
4. Proto Field Correctness:
* Fixed incorrect proto access logic where hasStreamedResponse() and getStreamedResponse() were being called on CommonResponse instead of BodyMutation.
5. Environmental Stability:
* Bypassed Gradle instrumentation errors caused by JDK 25 by explicitly running the build and tests using JDK 21
(JAVA_HOME=/usr/lib/jvm/java-21-openjdk-amd64).
6. Unit Test Enhancements:
* Updated ExternalProcessorFilterTest.java to use asyncUnaryCall for better concurrency testing.
* Simulated a real-world race condition by introducing a delay in the mock External Processor, verifying that the halfClosed logic effectively prevents
state machine errors.
… and add test coverage
… bug Makes `allowedGrpcServices` to be a non-optional struct instead of an `Optional<Map<str,AllowedService>>` since it's essentially an immuatable hash map, making it preferable to use an empty instance instead of null. Change a small bug where we continued instead of return when parsing bootstrap credentials.
…fig parsing Extend the xDS Filter API to support injecting bootstrap information into filters during configuration parsing. This allows filters to access context information (e.g., allowed gRPC services) from the resource loading layer during configuration validation and parsing. - Update `Filter.Provider.parseFilterConfig` and `parseFilterConfigOverride` to accept a `FilterContext` parameter. - Introduce `BootstrapInfoGrpcServiceContextProvider` to encapsulate bootstrap info for context resolution. - Update `XdsListenerResource` and `XdsRouteConfigureResource` to construct and pass `FilterContext` during configuration parsing. - Update sub-filters (`FaultFilter`, `RbacFilter`, `GcpAuthenticationFilter`, `RouterFilter`) to match the updated `FilterContext` signature. Known Gaps & Limitations: 1. **MetricHolder**: Propagation of `MetricHolder` is not supported with this approach currently and is planned for support in a later phase. 2. **NameResolverRegistry**: Propagation is deferred for consistency. While it could be passed from `XdsNameResolver` on the client side, there is no equivalent mechanism on the server side. To ensure consistent behavior, `DefaultRegistry` is used when validating schemes and creating channels.
Merged changes from sauravzg-feat-bootstrap-filter-context branch and updated ExternalProcessorFilter and its test to work with the new Filter.Provider.parseFilterConfig API which now takes a FilterContext containing BootstrapInfo and ServerInfo.
…r filter.
- Fixed IllegalStateException by moving initial request header transmission out of beforeStart().
- Fixed NullPointerException by providing a fallback to directExecutor() when CallOptions.getExecutor() is null.
- Resolved 'call was half-closed' state machine violation by tracking half-close state and skipping body mutations if the application has already half-closed the call.
- Corrected proto field access for BodyMutation and improved robustness of header mapping.
- Updated unit tests to verify fixes under simulated race conditions using async calls.
Summary of Fixes:
1. Resolved IllegalStateException: Not started:
* Root Cause: The filter was calling onNext() (which triggers sendMessage()) from within the beforeStart() callback of the ClientResponseObserver. In
gRPC-Java, beforeStart() is invoked before the underlying ClientCall.start() has completed, violating the call lifecycle.
* Fix: Moved the initial request headers transmission to immediately after the stub.process() call returns. This ensures the call has officially started
before any messages are sent.
2. Resolved NullPointerException: callExecutor:
* Root Cause: DelayedClientCall requires a non-null executor. When CallOptions.DEFAULT was used in tests, callOptions.getExecutor() returned null,
causing an NPE during filter initialization.
* Fix: (Superceded - this fallback was moved to test code later) Implemented a fallback to MoreExecutors.directExecutor() when the call options do not provide an executor.
3. Resolved IllegalStateException: call was half-closed (Race Condition):
* Root Cause: In unary calls (like blockingUnaryCall), gRPC half-closes the call immediately after sending the request. If the External Processor
returned header mutations after this half-close, the filter would attempt to send a sendMessage to the backend, causing a state machine violation.
* Fix: Added a halfClosed state tracker to the ExtProcClientCall. The filter now gracefully skips sending body mutations or empty messages if the
application has already half-closed the call.
4. Proto Field Correctness:
* Fixed incorrect proto access logic where hasStreamedResponse() and getStreamedResponse() were being called on CommonResponse instead of BodyMutation.
5. Environmental Stability:
* Bypassed Gradle instrumentation errors caused by JDK 25 by explicitly running the build and tests using JDK 21
(JAVA_HOME=/usr/lib/jvm/java-21-openjdk-amd64).
6. Unit Test Enhancements:
* Updated ExternalProcessorFilterTest.java to use asyncUnaryCall for better concurrency testing.
* Simulated a real-world race condition by introducing a delay in the mock External Processor, verifying that the halfClosed logic effectively prevents
state machine errors.
This commit introduces a library for handling header mutations as specified by the xDS protocol. This library provides the core functionality for modifying request and response headers based on a set of rules. The main components of this library are: - `HeaderMutator`: Applies header mutations to `Metadata` objects. - `HeaderMutationFilter`: Filters header mutations based on a set of configurable rules, such as disallowing mutations of system headers. - `HeaderMutations`: A value class that represents the set of mutations to be applied to request and response headers. - `HeaderMutationDisallowedException`: An exception that is thrown when a disallowed header mutation is attempted. This commit also includes comprehensive unit tests for the new library.
…ntrol - Fix observability mode early close hang by proceeding with close immediately. - Fix flow control close deferral to correctly check for pending/outstanding messages. - Refine isRequestSideCompleted to only require half-close if request body is intercepted. - Add Test 16 and Test 17 to verify these behaviors. - Fix Test 18 regression by triggering failure on onCompleted. TAG=agy CONV=9ea901ca-a127-468c-a836-414f2154bf85
Implement Phase 4 of the coverage improvement plan. - Add tests for observability mode early close with no trailers. - Add tests for trailers-only close when headers are skipped (normal and observability modes). - Improve flow control test to use manual flow control to trigger request buffering and deferred half-close coverage. TAG=agy CONV=9ea901ca-a127-468c-a836-414f2154bf85
Remove redundant IDLE check from isReady and redundant isExtProcStreamCompleted check from isSidecarReady. Add tests and verify fallback behavior of isReady when stream is completed. TAG=agy CONV=9ea901ca-a127-468c-a836-414f2154bf85
TAG=agy CONV=9ea901ca-a127-468c-a836-414f2154bf85
TAG=agy CONV=9ea901ca-a127-468c-a836-414f2154bf85
Updated ServerInbound.triggerEvent to invoke the listener's triggerEvent callback inside the synchronized(this) block. This ensures that the check for isClosed() and the invocation of the listener are atomic relative to stream closure (which also runs under the same lock). This prevents a race where triggerEvent could be called on the listener after the stream has been closed, which would result in out-of-order events delivered to the application. This is consistent with how other listener callbacks (like closed and halfClosed) are delivered in Inbound.java. TAG=agy CONV=e1bfa5a2-e855-4f79-abdd-ef2b264977be
…ExceptionInterceptor, and OpenTelemetryTracingModule. - binder: Implement onEvent in PendingAuthListener to buffer and replay custom events to the delegate once auth completes, preventing events from being dropped. - util: Handle onEvent in TransmitStatusRuntimeExceptionInterceptor listener wrapper to catch StatusRuntimeException and close the call. Serialize triggerEvent on SerializingServerCall's executor. - opentelemetry: Implement onEvent in ContextServerCallListener to attach OpenTelemetry trace context and scope during delegate invocation. - Add unit tests for all updated implementations. TAG=agy CONV=e1bfa5a2-e855-4f79-abdd-ef2b264977be
…ceptor - In DataPlaneServerListener, buffer custom events received before the delegate is set (e.g. during initial ext-proc headers processing). - Forward non-ext-proc events to delegate.onEvent when delegate is active. - Drain buffered savedEvents upon delegate installation and during fail-open. - Add unit test verifying custom event buffering and propagation. TAG=agy CONV=e1bfa5a2-e855-4f79-abdd-ef2b264977be
…mListenerImpl triggerEvent Wrap ServerCallImpl.triggerEvent and ServerStreamListenerImpl.triggerEvent in PerfMark.traceTask with PerfMark.attachTag, aligning them with sendMessage, sendHeaders, close, request, and listener callbacks. TAG=agy CONV=e1bfa5a2-e855-4f79-abdd-ef2b264977be
…d is true Check closeCalled before dispatching triggerEvent to the transport stream, avoiding unnecessary task allocations and transport hops if the call has already been closed. TAG=agy CONV=e1bfa5a2-e855-4f79-abdd-ef2b264977be
…stom event changes - Test onEvent throwing StatusRuntimeException closes the call with status and trailers. - Test onEvent throwing StatusRuntimeException on an already closed call does not trigger duplicate close. - Test SerializingServerCall executes triggerEvent sequentially in FIFO order on serializingExecutor. TAG=agy CONV=e1bfa5a2-e855-4f79-abdd-ef2b264977be
…e stream cleanup in AbstractTransportTest
In MockServerTransportListener.streamCreated(), stream.setListener(listener)
was called after streams.add(StreamCreation(...)). This created a race
condition where a test thread calling takeStreamOrFail() could dequeue
the stream and call serverStream.triggerEvent() before stream.setListener()
was called by the transport/container thread. When this occurred (e.g. in
TomcatTransportTest on multi-core runners), ServletServerStream invoked
transportState.triggerEvent() on the test thread, saw a null listener,
threw a NullPointerException (swallowed by SerializingExecutor), and
never enqueued the event into the listener queue, leading to a timeout
and assertion failure:
expected:<...Object@...> but was:<null>
Setting stream.setListener(listener) before enqueuing to streams guarantees
that any thread consuming the StreamCreation will always observe a fully
initialized listener.
Additionally, in AbstractTransportTest.serverStream_triggerEvent(), replace
clientStream.cancel(Status.CANCELLED) with serverStream.close(Status.OK, ...)
for clean stream closure instead of leaving an uncoordinated client RST_STREAM
in flight during container tearDown.
TAG=agy
CONV=e1bfa5a2-e855-4f79-abdd-ef2b264977be
The ext_proc spec has been updated to drop adding the backend_service metric label. TAG=agy CONV=86c74ebe-9fd7-4876-b27c-a4c1b230d346
…without_message per ext_proc spec TAG=agy CONV=86c74ebe-9fd7-4876-b27c-a4c1b230d346
TAG=agy CONV=86c74ebe-9fd7-4876-b27c-a4c1b230d346
# Conflicts: # xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java # xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java # xds/third_party/envoy/src/main/proto/envoy/service/ext_proc/v3/external_processor.proto
…nterceptorTest TAG=agy CONV=86c74ebe-9fd7-4876-b27c-a4c1b230d346
…roc server interceptor Refactor ExternalProcessorServerInterceptor so that transport and delegate operations (super.request(), proceedWithClose(), proceedWithSendHeaders(), super.sendMessage(), and super.isReady()) are invoked outside streamLock. This prevents potential deadlock when using transports like Binder, where inbound callbacks (e.g. triggerEvent) hold transport locks (Inbound.this) while executing listener callbacks synchronously, while transport calls like request() and close() synchronize on Inbound.this. TAG=agy CONV=b39523be-9572-456e-a428-8b0e255067da
…tus in server interceptor Update ext_proc server interceptor to treat OK termination without drain as non-OK status unless in observability mode, in accordance with the updated gRFC specification. TAG=agy CONV=581f86a6-097d-44a2-9d6e-6f9d2e83cbc0
…nt overflow Use LongMath.saturatedAdd when processing ServerWindowUpdate and replenishing sidestream windows to prevent integer overflow from turning windows negative. TAG=agy CONV=581f86a6-097d-44a2-9d6e-6f9d2e83cbc0
Reject xDS filter configuration if processing_mode is unset and fix checkstyle issues. TAG=agy CONV=581f86a6-097d-44a2-9d6e-6f9d2e83cbc0
TAG=agy CONV=581f86a6-097d-44a2-9d6e-6f9d2e83cbc0
…rceptor ext_proc ProcessingResponses do not have a 1-to-1 relationship with ProcessingRequests (messages can be inserted, dropped, or chunked). Remove outstandingResponseBodyRequests tracking and instead defer close only while messages are waiting in pendingResponseBodyMessages. TAG=agy CONV=581f86a6-097d-44a2-9d6e-6f9d2e83cbc0
…ver interceptor - Replace expectedResponses queue with directional state flags and configuration checks. - Enforce strict directional ordering on ext_proc ProcessingResponse events. - Propagate protocol violation causes on stream failure. - Add Category 21 unit tests covering 17 protocol ordering and violation scenarios. TAG=agy CONV=581f86a6-097d-44a2-9d6e-6f9d2e83cbc0
# Conflicts: # api/src/main/java/io/grpc/ServerCall.java # core/src/main/java/io/grpc/internal/ServerCallImpl.java # core/src/test/java/io/grpc/internal/ServerCallImplTest.java
…st configs - Extract MetadataValueParser into top-level package-private interface to avoid javac 8 JDK-8024694 symbol resolution issue in GcpAuthenticationFilter. - Set default processing_mode in ExternalProcessorClientInterceptorTest to satisfy required processing_mode validation. TAG=agy CONV=581f86a6-097d-44a2-9d6e-6f9d2e83cbc0
xds: implement separate request and response draining in ExternalProcessorServerInterceptor Update ExternalProcessorServerInterceptor to support separate request-side and response-side draining per the updated gRFC and Envoy ext_proc proto: - Replace the unified DRAINING stream state with directional draining and pass-through state (requestDraining, requestDrainComplete, responseDraining, responseDrainComplete, requestPassThrough, and passThroughMode). - Split flow-control readiness into isRequestSidecarReady() and isResponseSidecarReady() so draining one direction does not block or prematurely unblock the other direction. - Implement the two-way drain_complete body handshake for request and response streams when body processing is configured, and ignore drain requests after end_of_stream has already been sent for that direction. - Close the ext_proc stream once both directions have completed draining (in non-observability mode), and treat premature OK termination from ext_proc before both directions are drained or at EOS as an error. - Update existing server interceptor draining tests and add unit tests for directional draining and EOS edge cases. TAG=agy CONV=2c1e4760-c239-4698-810a-162bf10fccc4 ```
This branch has not been deployed
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
14 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 server interceptor as per gRFC A93.