xds: External Processor Server Interceptor - #12889
Open
kannanjgithub wants to merge 469 commits into
Open
Conversation
…eful termination of the ext-proc stream.
…in the filter config update.
…ext-proc stream error.
…ived from ext-proc.
…ad create composite isReady and onReady behaviors for the calling client application to utilize them for applying flow control.
…rInterceptor use a raw bytes marshaller.
This commit introduces configuration objects for the external authorization (ExtAuthz) filter and the gRPC service it uses. These classes provide a structured, immutable representation of the configuration defined in the xDS protobuf messages. The main new classes are: - `ExtAuthzConfig`: Represents the configuration for the `ExtAuthz` filter, including settings for the gRPC service, header mutation rules, and other filter behaviors. - `GrpcServiceConfig`: Represents the configuration for a gRPC service, including the target URI, credentials, and other settings. - `HeaderMutationRulesConfig`: Represents the configuration for header mutation rules. This commit also includes parsers to create these configuration objects from the corresponding protobuf messages, as well as unit tests for the new classes.
… the updated requirements
…cService proto and the GrpcServiceXdsContextProvider. Also added GrpcServiceXdsContextProvider in the newInstance method for Filter in Filter.Provider.
…e GrpcService config.
Allow unit test to pass CacheChannelManager for in-process channel. Fix mock ext-proc service to handle all phases of the request-response events to avoid test hanging. The test failed with "too many messages" error due to sending an empty body message to the data plane server during the "half-close" phase . For a unary RPC, this was interpreted as a second request message, which is invalid. I've updated handleRequestBodyResponse and onExternalBody to only call super.sendMessage() or super.onMessage() if the body content is non-empty. This prevents the redundant empty message from being sent to the data plane while still allowing the external processor to signal the end of the stream. The stream between the filter and the external processor was never being closed on the client side, causing the InProcessChannel and InProcessServer to hang during shutdown while waiting for the active RPC to terminate. To fix this, I have updated ExternalProcessorFilter.java to ensure the control plane stream is gracefully closed when the data plane RPC completes or is cancelled. Changes made: 1. Closing on Completion: In ExtProcClientCall.onNext, once the ResponseTrailers handshake is finished and the application has been notified via proceedWithClose(), I now call extProcClientCallRequestObserver.onCompleted(). 2. Handling Cancellation: I overridden the cancel() method in ExtProcClientCall. If the data plane RPC is cancelled by the application, the filter now also cancels the external processor stream with an error, ensuring all resources are freed. 3. Observability Mode Fix: In observability mode, since we don't wait for a ResponseTrailers message from the server, I added logic to ExtProcListener.onClose() to close the external processor stream immediately after sending the final trailers. These changes ensure proper lifecycle management of the side-channel RPC.
…Shares backpressure logic with observability mode.
…companying message ever with half-close.
…by overriding ClientCall.requestMessages(int) in ExtProcClientCall. Also introduce null check for extProcClientCallRequestObserver in isReady since it may be called on the call even before start is called that initializes it.
…ation of the External Processor filter coordinates data across the application thread, the data plane response thread, and the external processor's response thread. To ensure thread safety and compliance with the gRPC contract, the
following synchronization measures were implemented:
1. Thread-Unsafe StreamObserver
* Challenge: The gRPC StreamObserver used to send messages to the external processor is not thread-safe. Concurrent calls to its onNext(), onCompleted(), and onError() methods from different threads can corrupt the internal state of the communication
channel. Additionally, calling isReady() on the observer while another thread is sending data can lead to race conditions.
* Fix: All interactions with the external processor's StreamObserver—including data transmission (onNext), terminal signals (onCompleted, onError), and readiness checks (isReady)—are now protected by the lock object.
2. ClientCall.Listener Serialization Contract
* Challenge: gRPC requires that all callbacks to an application's ClientCall.Listener (such as onHeaders, onMessage, and onReady) be strictly serialized. Because these events can be triggered by either the backend server or the external processor,
there was a risk of overlapping callbacks.
* Fix: The logic that delivers events to the application's Listener is now synchronized using the lock. This ensures that even if multiple threads attempt to "unblock" and deliver buffered metadata or status simultaneously, the application receives
them in a single, non-overlapping sequence.
3. Visibility and Consistency of Internal State
* Challenge: The filter maintains several internal state variables, such as buffers for response metadata and flags to track the lifecycle of the call. If these are accessed concurrently without synchronization, one thread might act on stale data,
potentially leading to duplicate headers or incorrect flow control decisions.
* Fix: Access to all internal state and control flags is now guarded by the lock. Furthermore, the flag indicating whether request headers have been processed was marked as volatile to ensure its state is immediately visible across threads during
high-frequency checks like sendMessage().
4. Synchronizing Terminal Signals
* Challenge: Closing or faulting the external processor's stream while another thread is still attempting to send data can cause crashes or undefined behavior in the gRPC transport.
* Fix: All terminal signals (onCompleted and onError) sent to the external processor's StreamObserver are synchronized with the same lock used for sending data. This ensures that the stream is only terminated after any ongoing data transfers have
safely finished.
Fix some incorrect handlings done using buffered messages, there should be no need to buffer messages except in the case of observability mode when headers have been not yet been sent.
Fix missing coordinated synchronizations between threads.
…rove concurrency in the External Processor filter is complete. The implementation now employs a more granular three-lock strategy:
1. streamLock: Guards all interactions with the extProcClientCallRequestObserver. This ensures the gRPC StreamObserver to the external processor is never accessed concurrently, protecting its internal state.
2. requestLock: Manages the outbound flow control. It guards the headersSent flag and the pendingActions queue, coordinating the transition from the initial buffering phase to active delivery to the backend server.
3. responseLock: Serializes all callbacks to the application's Listener (onHeaders, onMessage, onClose, onReady). It also guards shared response state like savedHeaders and savedStatus. This ensures strict compliance with the gRPC contract while
fixing a potential race condition in onExternalBody.
By decoupling the request and response data planes, the filter now supports full-duplex concurrency where outbound messages do not block inbound server responses. All lock acquisitions were carefully refactored to be sequential, maintaining a
consistent order and guaranteeing deadlock-free execution.
…tName (grpc#12644)" We want to synchronize the behavior across all gRPC languages, and also with envoy. This reverts commit 0ef1b39.
This change is a no-op. The create() form is clearer than positional arguments to a heavily overloaded constructor.
Guard this behavior change behind the RFC 3986 parser flag.
Google Play Services needs min Android API level of 23 (Android 6.0 Marshmallow). Fixes [grpc#11474](grpc#11474).
The backoff timer is only used when serializeRetries=true, and that exists to match the old/current pick_first's behavior as closely as possible. InternalSubchannel.updateAddresses() would take no action when in TRANSIENT_FAILURE; it would update the addresses and just wait for the backoff timer to expire. Note that this only impacts serializeRetries=true; in the other cases we do want to start trying to the new addresses immediately, because the backoff timers are in the subchannels. Note that this change was also important because requestConnection() can be directly triggered by the user with channel.getState(true), and that shouldn't defeat the backoff timer.
…-server and adaptations after that.
…rpc.xds.internal.extproc.
This adds triggerEvent/onEvent APIs to ServerCall and ServerCall.Listener, routing them through ServerStream transport to ensure thread-safety (especially for SerializeReentrantCallsDirectExecutor). TAG=agy CONV=e1bfa5a2-e855-4f79-abdd-ef2b264977be
Refactors ExternalProcessorServerInterceptor to use the new custom events framework (Hybrid Event Model), eliminating the syncContext lock and serializing all callbacks (including ext_proc stub responses) on the application executor. Also includes fixes for forwarding listeners to propagate custom events and corrections to onReady notification timing. TAG=agy CONV=e1bfa5a2-e855-4f79-abdd-ef2b264977be
… behavior. - Added unit tests in ServerImplTest for JumpToApplicationThreadServerStreamListener.triggerEvent. - Added serverStream_triggerEvent_afterClose in AbstractTransportTest to verify events are ignored after stream closure. - Updated Inbound.ServerInbound to check isClosed() before triggering events. TAG=agy CONV=e1bfa5a2-e855-4f79-abdd-ef2b264977be
…framework. - Added unit tests in AbstractServerStreamTest for triggerEvent propagation and close behavior. - Updated ContextsTest to cover onEvent propagation in ContextualizedServerCallListener. TAG=agy CONV=e1bfa5a2-e855-4f79-abdd-ef2b264977be
Wait for the server stream to be fully closed (via awaitClose) before calling triggerEvent, to ensure the transport has processed the cancellation and marked the listener as closed. This fixes flakiness in slower transports like Jetty. TAG=agy CONV=e1bfa5a2-e855-4f79-abdd-ef2b264977be
- Consolidated locking under streamLock. - Implemented queue-based flow control for request/response bodies and headers.
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 server interceptor as per gRFC A93.