xds: External Processor Server Interceptor - #12889
Open
kannanjgithub wants to merge 496 commits into
Open
Conversation
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.
Since we're only supporting API levels 23+, all the supported Android versions handle multidex natively, and without any bugs to workaround. Also bump some minSdkVersion that didn't get updated in fa7b52b so that multiDex is actually enabled by default. See also b/476359563
DnsNameResolver discards refresh requests if it has been too soon after the last refresh, because the result is assumed to be identical to the previous fetch. Android itself will adhere to the RR's TTL, so requesting too frequently shouldn't have been causing too much I/O, but it could be causing extra CPU usage. Having some lower limit will reduce the number of useless address updates into the LB tree. 30 seconds is the same as regular Java and Go/C++ (which copied Java as a "reasonable" value). Note that other languages _delay_ the refresh instead of _discarding_ the refresh, but there's no reason why the existing discard behavior would cause much problem on Android vs normal Java. Chrome apparently uses 1 minute, so this really looks like it shouldn't cause problems as long as AndroidChannelBuilder is being used.
The internal result was needed before 90d0fab allowed addresses to fail yet still provide attributes and service config. Now the code can just use the regular API. This does cause a behavior change where TXT records are looked up even if address lookup failed, however that's actually what we wanted to allow in 90d0fab by adding the new API. Also, the TXT code was added in 2017 and it's now 2026 yet it is still disabled, so it's unlikely to matter soon.
…#12697) 4de4718 upgraded android-interop-testing to SDK version 23, but this had previously been avoided because it triggered a Gradle or AGP bug. The race happened to not trigger locally or for the PR's CI and the change was merged. But the problem still was present, and the CI is failing to build very frequently. This works around the problem by explicitly adding a dependency from mergeExtDexDebug. I didn't see any other mergeExtDex tasks created, in particular mergeExtDexRelease. Hopefully we can remove this after upgrading AGP or Gradle, but at least we can move forward with newer Android API levels again.
…DS server by ref-counting This PR implements reusing the gRPC xDS transport (and underlying gRPC channel) to the same xDS server by ref-counting, which is already implemented in gRPC C++ ([link](https://github.com/grpc/grpc/blob/5a3a5d53145b94895610825e783a8896a61a3c73/src/core/xds/grpc/xds_transport_grpc.cc#L399-L414)) and gRPC Go ([link](https://github.com/grpc/grpc-go/blob/81c7924ec9f5f4a01c18b82c9d67691c1cd93bd5/internal/xds/clients/grpctransport/grpc_transport.go#L78-L120)). This optimization is expected to reduce memory footprint of the xDS management server and xDS enabled clients as channel establishment and lifecycle management of the connection is expensive. * Implemented a map to store `GrpcXdsTransport` instances keyed by the `Bootstrapper.ServerInfo` and each `GrpcXdsTransport` has a ref count. Note, the map cannot be simply keyed by the xDS server address as the client could have different channel credentials to the same xDS server, which should be counted as different transport instances. * When `GrpcXdsTransportFactory.create()` is called, the existing transport is reused if it already exists in the map and increment its ref count, otherwise create a new transport, store it in the map, and increment its ref count. * When `GrpcXdsTransport.shutdown()` is called, its ref count is decremented and the underlying gRPC channel is shut down when its ref count reaches zero. * Note this ref-counting of the `GrpcXdsTransport` is different and orthogonal to the ref-counting of the xDS client keyed by the xDS server target name to allow for xDS-based fallback per [gRFC A71](https://github.com/grpc/proposal/blob/master/A71-xds-fallback.md). Prod risk level: Low * Reusing the underlying gRPC channel to the xDS server would not affect the gRPC xDS (ADS/LRS) streams which would be multiplexed on the same channel, however, this means new xDS (ADS/LRS) streams and RPCs may fail due to hitting the limit of `MAX_CONCURRENT_STREAMS`. Tested: * Verified end-to-end with a xDS enabled gRPC Java client communicating to multiple different gRPC backend servers behind *different targets* using the xDS management server for name resolution and endpoint discovery. Verified gRPC xDS transport creation, ref-counting, reuse, shutdown, deletion from map when ref count is zero all worked as expected. Implementation details / context: * Used `java.util.concurrent.ConcurrentHashMap` APIs `compute` and `computeIfPresent` where the entire method invocation is performed atomically to achieve a concurrent and thread-safe solution which follows Java best practices. Alternatives considered: * Write own synchronization logic with synchronized block and locks. After discussion internally, it was preferred to use existing concurrency libraries which is less error-prone and should offer better performance.
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.
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.
# Conflicts: # xds/src/main/java/io/grpc/xds/XdsServerWrapper.java
…from ext_proc were getting stuck in pendingMutatedRequestBodies because they were waiting for App demand to be drained. I fixed this by allowing end_of_stream_without_message to be drained immediately without requiring App demand.
… onCompleted due to client half-close) is handled correctly by the double-close protection and still propagates the original error status and not overwritten by Ok from the client half-close. Added metrics tests.
ExternalProcessorServerInterceptor.java : 1. Deferred Half-Close Propagation Bug (Null Delegate) Issue: In proceedWithHalfClose(), if the delegate (app listener) was not set yet (null), the interceptor set requestSideClosed to true and returned. When the delegate was finally set via SetDelegateEvent, the subsequent call to proceedWithHalfClose() returned early because requestSideClosed was already true, preventing the half-close from ever reaching the application. Fix: Modified proceedWithHalfClose() to return early without setting requestSideClosed if delegate is null, allowing the subsequent call to successfully propagate the half-close. 2. Observability Mode Close Hang Bug Issue: In close(), the interceptor checked if there were outstanding response body requests. In observability mode, we return early from response processing, so outstandingResponseBodyRequests is never decremented. This caused close() to wait indefinitely for it to reach 0, hanging the call. Fix: Bypassed the check for outstanding/pending messages in close() if observabilityMode is enabled. 3. Observability Mode Half-Close Propagation Bug Issue: In onHalfClose(), if the mode was GRPC, we sent EOS to ext proc and returned without calling proceedWithHalfClose(). In observability mode, we don't process responses, so proceedWithHalfClose() was never called, and the app never saw the half-close. Fix: Updated onHalfClose() to call proceedWithHalfClose() immediately if observabilityMode is enabled.
…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
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.