Skip to content

xDS: Ext proc client interceptor implement grfc updates #13105

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
kannanjgithub wants to merge 10 commits into
base: master
Choose a base branch
from
Open

xDS: Ext proc client interceptor implement grfc updates #13105

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
85758e3
xds: Make RawMessageClientInterceptor conditional on ext_proc flags
kannanjgithub Aug 31, 2026
3350be4
Merge branch 'grpc:master' into master
kannanjgithub Sep 1, 2026
49786af
Merge remote-tracking branch 'origin/master'
kannanjgithub Sep 9, 2026
78e1e42
Merge remote-tracking branch 'refs/remotes/origin/master'
kannanjgithub Oct 6, 2026
7628b0d
xds: treat ext_proc stream OK termination without drain as non-OK status
kannanjgithub Oct 8, 2026
a2887c6
xds: cap flow control window at 64-bit signed integer limits to preve…
kannanjgithub Oct 8, 2026
7208ff8
xds: require processing_mode in ExternalProcessor filter config
kannanjgithub Oct 8, 2026
ca79e36
xds: default to UNKNOWN status when ImmediateResponse status is unset
kannanjgithub Oct 8, 2026
98e8d20
xds: enforce directional and configuration ordering for ext_proc clie…
kannanjgithub Oct 9, 2026
c371e3a
xds: improve test coverage for ext_proc client interceptor
kannanjgithub Oct 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
244 changes: 205 additions & 39 deletions xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.math.LongMath;
import com.google.protobuf.ByteString;
import com.google.protobuf.Struct;
import io.envoyproxy.envoy.extensions.filters.http.ext_proc.v3.ProcessingMode;
Expand Down Expand Up @@ -68,7 +69,6 @@
import io.grpc.xds.ExternalProcessorFilter.ExternalProcessorFilterConfig;
import io.grpc.xds.Filter.FilterContext;
import io.grpc.xds.internal.extproc.DataPlaneCallState;
import io.grpc.xds.internal.extproc.EventType;
import io.grpc.xds.internal.extproc.ExtProcStreamState;
import io.grpc.xds.internal.extproc.KnownLengthInputStream;
import io.grpc.xds.internal.grpcservice.CachedChannelManager;
Expand Down Expand Up @@ -286,8 +286,40 @@ private static class DataPlaneClientCall
private final DataPlaneDelayedCall<InputStream, InputStream> delayedCall;
private final ScheduledExecutorService scheduler;
final Object streamLock = new Object();
@Nullable private volatile EventType expectedRequestResponse;
@Nullable private volatile EventType expectedResponseResponse;
private volatile boolean requestHeadersSent;
private volatile boolean responseHeadersSent;
private volatile boolean responseTrailersSent;
private volatile boolean clientHeadersReceived;
private volatile boolean clientEndOfStreamReceived;
private volatile boolean serverHeadersReceived;
private volatile boolean serverTrailersReceived;
private volatile boolean serverEndOfStreamReceived;
private volatile Throwable streamFailureCause;

private boolean isRequestHeaderConfigured() {
ProcessingMode.HeaderSendMode mode = currentProcessingMode.getRequestHeaderMode();
return mode == ProcessingMode.HeaderSendMode.SEND
|| mode == ProcessingMode.HeaderSendMode.DEFAULT;
}

private boolean isRequestBodyConfigured() {
return currentProcessingMode.getRequestBodyMode() != ProcessingMode.BodySendMode.NONE;
}

private boolean isResponseHeaderConfigured() {
ProcessingMode.HeaderSendMode mode = currentProcessingMode.getResponseHeaderMode();
return mode == ProcessingMode.HeaderSendMode.SEND
|| mode == ProcessingMode.HeaderSendMode.DEFAULT;
}

private boolean isResponseBodyConfigured() {
return currentProcessingMode.getResponseBodyMode() != ProcessingMode.BodySendMode.NONE;
}

private boolean isResponseTrailerConfigured() {
return currentProcessingMode.getResponseTrailerMode() == ProcessingMode.HeaderSendMode.SEND;
}

@Nullable private volatile ClientCallStreamObserver<ProcessingRequest>
extProcClientCallRequestObserver;
@GuardedBy("streamLock")
Expand Down Expand Up @@ -495,8 +527,12 @@ public void onNext(ProcessingResponse response) {
boolean wasReady;
synchronized (streamLock) {
wasReady = isReady();
downstreamToSidestreamWindow += update.getWindowIncrementDownstreamToSidestream();
upstreamToSidestreamWindow += update.getWindowIncrementUpstreamToSidestream();
downstreamToSidestreamWindow = LongMath.saturatedAdd(
downstreamToSidestreamWindow,
update.getWindowIncrementDownstreamToSidestream());
upstreamToSidestreamWindow = LongMath.saturatedAdd(
upstreamToSidestreamWindow,
update.getWindowIncrementUpstreamToSidestream());
drainPendingRequestBodyMessages();
drainPendingRequests();
if (wrappedListener != null) {
Expand All @@ -523,53 +559,151 @@ public void onNext(ProcessingResponse response) {
}

if (response.hasRequestHeaders()) {
EventType expected = expectedRequestResponse;
if (expected == null || expected != EventType.REQUEST_HEADERS) {
if (!isRequestHeaderConfigured()) {
internalOnError(Status.UNAVAILABLE
.withDescription("Protocol error: received response out of order. "
+ "Filter is not configured to send client headers.")
.asRuntimeException());
return;
}
if (!requestHeadersSent) {
internalOnError(Status.UNAVAILABLE
.withDescription("Protocol error: received response out of order. "
+ "Received request_headers before request_headers was sent.")
.asRuntimeException());
return;
}
if (clientEndOfStreamReceived) {
internalOnError(Status.UNAVAILABLE
.withDescription("Protocol error: received response out of order. "
+ "Received request_headers after end-of-stream in "
+ "client-to-server direction.")
.asRuntimeException());
return;
}
if (clientHeadersReceived) {
internalOnError(Status.UNAVAILABLE
.withDescription("Protocol error: received response out of order. Expected: "
+ expected + ", Received: REQUEST_HEADERS")
.withDescription("Protocol error: received response out of order. "
+ "Duplicate request_headers response.")
.asRuntimeException());
return;
}
expectedRequestResponse = null;
clientHeadersReceived = true;
} else if (response.hasResponseHeaders()) {
EventType expected = expectedResponseResponse;
if (expected == null || expected != EventType.RESPONSE_HEADERS) {
if (!isResponseHeaderConfigured()) {
internalOnError(Status.UNAVAILABLE
.withDescription("Protocol error: received response out of order. "
+ "Filter is not configured to send server headers.")
.asRuntimeException());
return;
}
if (!responseHeadersSent) {
internalOnError(Status.UNAVAILABLE
.withDescription("Protocol error: received response out of order. "
+ "Received response_headers before response_headers was sent.")
.asRuntimeException());
return;
}
if (serverTrailersReceived || serverEndOfStreamReceived) {
internalOnError(Status.UNAVAILABLE
.withDescription("Protocol error: received response out of order. "
+ "Received response_headers after trailers or end-of-stream.")
.asRuntimeException());
return;
}
if (serverHeadersReceived) {
internalOnError(Status.UNAVAILABLE
.withDescription("Protocol error: received response out of order. Expected: "
+ expected + ", Received: RESPONSE_HEADERS")
.withDescription("Protocol error: received response out of order. "
+ "Duplicate response_headers response.")
.asRuntimeException());
return;
}
expectedResponseResponse = null;
serverHeadersReceived = true;
} else if (response.hasResponseTrailers()) {
EventType expected = expectedResponseResponse;
if (expected == null || expected != EventType.RESPONSE_TRAILERS) {
if (!isResponseTrailerConfigured()) {
internalOnError(Status.UNAVAILABLE
.withDescription("Protocol error: received response out of order. "
+ "Filter is not configured to send server trailers.")
.asRuntimeException());
return;
}
if (isResponseHeaderConfigured() && !serverHeadersReceived) {
internalOnError(Status.UNAVAILABLE
.withDescription("Protocol error: received response out of order. "
+ "Received response_trailers before response_headers response.")
.asRuntimeException());
return;
}
if (!responseTrailersSent) {
internalOnError(Status.UNAVAILABLE
.withDescription("Protocol error: received response out of order. "
+ "Received response_trailers before response_trailers was sent.")
.asRuntimeException());
return;
}
if (serverTrailersReceived) {
internalOnError(Status.UNAVAILABLE
.withDescription("Protocol error: received response out of order. "
+ "Duplicate response_trailers response.")
.asRuntimeException());
return;
}
if (serverEndOfStreamReceived) {
internalOnError(Status.UNAVAILABLE
.withDescription("Protocol error: received response out of order. Expected: "
+ expected + ", Received: RESPONSE_TRAILERS")
.withDescription("Protocol error: received response out of order. "
+ "Received response_trailers after end-of-stream.")
.asRuntimeException());
return;
}
expectedResponseResponse = null;
serverTrailersReceived = true;
serverEndOfStreamReceived = true;
} else if (response.hasRequestBody()) {
EventType expected = expectedRequestResponse;
if (expected == EventType.REQUEST_HEADERS) {
if (!isRequestBodyConfigured()) {
internalOnError(Status.UNAVAILABLE
.withDescription(
"Protocol error: received request_body when filter is not configured "
+ "to send client messages.")
.asRuntimeException());
return;
}
if (isRequestHeaderConfigured() && !clientHeadersReceived) {
internalOnError(Status.UNAVAILABLE
.withDescription(
"Protocol error: received request_body before request_headers response.")
.asRuntimeException());
return;
}
if (clientEndOfStreamReceived) {
internalOnError(Status.UNAVAILABLE
.withDescription(
"Protocol error: received request_body after end-of-stream in "
+ "client-to-server direction.")
.asRuntimeException());
return;
}
} else if (response.hasResponseBody()) {
EventType expected = expectedResponseResponse;
if (expected == EventType.RESPONSE_HEADERS) {
if (!isResponseBodyConfigured()) {
internalOnError(Status.UNAVAILABLE
.withDescription(
"Protocol error: received response_body when filter is not configured "
+ "to send server messages.")
.asRuntimeException());
return;
}
if (isResponseHeaderConfigured() && !serverHeadersReceived) {
internalOnError(Status.UNAVAILABLE
.withDescription(
"Protocol error: received response_body before headers response.")
.asRuntimeException());
return;
}
if (serverTrailersReceived || serverEndOfStreamReceived) {
internalOnError(Status.UNAVAILABLE
.withDescription(
"Protocol error: received response_body after trailers or end-of-stream.")
.asRuntimeException());
return;
}
}

if (response.getRequestDrain()) {
Expand Down Expand Up @@ -621,6 +755,7 @@ else if (response.hasResponseHeaders()) {
mutator);
}
if (wrappedListener.isTrailersOnly()) {
serverEndOfStreamReceived = true;
wrappedListener.proceedWithClose();
} else {
wrappedListener.proceedWithHeaders();
Expand Down Expand Up @@ -669,11 +804,18 @@ public void onError(Throwable t) {

@Override
public void onCompleted() {
if (markExtProcStreamCompleted(extProcStreamState)) {
synchronized (streamLock) {
extProcClientCallRequestObserver = null;
ExtProcStreamState state = extProcStreamState.get();
if (state == ExtProcStreamState.DRAINING || config.getObservabilityMode()) {
if (markExtProcStreamCompleted(extProcStreamState)) {
synchronized (streamLock) {
extProcClientCallRequestObserver = null;
}
handleFailOpen(wrappedListener);
}
handleFailOpen(wrappedListener);
} else if (state == ExtProcStreamState.ACTIVE) {
internalOnError(Status.UNAVAILABLE
.withDescription("External processor stream completed without drain")
.asRuntimeException());
}
}
});
Expand Down Expand Up @@ -707,11 +849,11 @@ private void sendToExtProc(ProcessingRequest request) {
}

if (request.hasRequestHeaders()) {
expectedRequestResponse = EventType.REQUEST_HEADERS;
requestHeadersSent = true;
} else if (request.hasResponseHeaders()) {
expectedResponseResponse = EventType.RESPONSE_HEADERS;
responseHeadersSent = true;
} else if (request.hasResponseTrailers()) {
expectedResponseResponse = EventType.RESPONSE_TRAILERS;
responseTrailersSent = true;
}

ProcessingRequest requestToSend = request;
Expand Down Expand Up @@ -772,8 +914,10 @@ void mergeAccumulatedWindowUpdates(ProcessingRequest.Builder requestBuilder) {
.build());
accumulatedWindowUpdateSidestreamToUpstream -= incrementUpstream;
accumulatedWindowUpdateSidestreamToDownstream -= incrementDownstream;
sidestreamToUpstreamWindow += incrementUpstream;
sidestreamToDownstreamWindow += incrementDownstream;
sidestreamToUpstreamWindow =
LongMath.saturatedAdd(sidestreamToUpstreamWindow, incrementUpstream);
sidestreamToDownstreamWindow =
LongMath.saturatedAdd(sidestreamToDownstreamWindow, incrementDownstream);
}
}

Expand All @@ -796,8 +940,10 @@ private void trySendAccumulatedWindowUpdates() {
if (shouldSend) {
accumulatedWindowUpdateSidestreamToUpstream -= incrementUpstream;
accumulatedWindowUpdateSidestreamToDownstream -= incrementDownstream;
sidestreamToUpstreamWindow += incrementUpstream;
sidestreamToDownstreamWindow += incrementDownstream;
sidestreamToUpstreamWindow =
LongMath.saturatedAdd(sidestreamToUpstreamWindow, incrementUpstream);
sidestreamToDownstreamWindow =
LongMath.saturatedAdd(sidestreamToDownstreamWindow, incrementDownstream);

sendToExtProc(ProcessingRequest.newBuilder()
.setClientWindowUpdate(ProcessingRequest.ClientWindowUpdate.newBuilder()
Expand Down Expand Up @@ -846,10 +992,15 @@ private void closeExtProcStream() {
}

private void internalOnError(Throwable t) {
this.streamFailureCause = t;
if (markExtProcStreamFailed(extProcStreamState)) {
synchronized (streamLock) {
if (extProcClientCallRequestObserver != null) {
extProcClientCallRequestObserver.onError(t);
try {
extProcClientCallRequestObserver.onError(t);
} catch (Throwable ignored) {
// Ignore exception if observer is already closed/completed
}
extProcClientCallRequestObserver = null;
}
}
Expand Down Expand Up @@ -1185,6 +1336,7 @@ private void handleRequestBodyResponse(BodyResponse bodyResponse) {
}
}
if (isEndOfStream) {
clientEndOfStreamReceived = true;
synchronized (streamLock) {
if (pendingUpstreamBodyMessages.isEmpty()) {
if (requestSideClosed.compareAndSet(false, true)) {
Expand Down Expand Up @@ -1309,7 +1461,9 @@ void drainPendingUpstreamBodyMessages() {

private void handleImmediateResponse(ImmediateResponse immediate, DataPlaneListener listener)
throws HeaderMutationDisallowedException {
Status status = Status.fromCodeValue(immediate.getGrpcStatus().getStatus());
Status status = immediate.hasGrpcStatus()
? Status.fromCodeValue(immediate.getGrpcStatus().getStatus())
: Status.UNKNOWN;
if (!immediate.getDetails().isEmpty()) {
status = status.withDescription(immediate.getDetails());
}
Expand Down Expand Up @@ -1619,7 +1773,9 @@ public void onClose(Status status, Metadata trailers) {
|| dataPlaneClientCall.bodyMessageSentToExtProc.get())) {
if (markDataPlaneCallClosed(dataPlaneClientCall.dataPlaneCallState)) {
proceedWithClose(Status.INTERNAL.withDescription("External processor stream failed")
.withCause(status.getCause()), new Metadata());
.withCause(dataPlaneClientCall.streamFailureCause != null
? dataPlaneClientCall.streamFailureCause
: status.getCause()), new Metadata());
}
return;
}
Expand Down Expand Up @@ -1691,7 +1847,17 @@ private void proceedWithHeaders(Metadata headers) {
void proceedWithClose() {
if (savedStatus != null) {
if (markDataPlaneCallClosed(dataPlaneClientCall.dataPlaneCallState)) {
proceedWithClose(savedStatus, savedTrailers);
Status toDeliver = savedStatus;
if (savedStatus.isOk()
&& dataPlaneClientCall.getExtProcStreamState().get().isFailed()
&& (!dataPlaneClientCall.getConfig().getFailureModeAllow()
|| dataPlaneClientCall.bodyMessageSentToExtProc.get())) {
toDeliver = Status.INTERNAL.withDescription("External processor stream failed")
.withCause(dataPlaneClientCall.streamFailureCause != null
? dataPlaneClientCall.streamFailureCause
: savedStatus.getCause());
}
proceedWithClose(toDeliver, savedTrailers);
}
savedStatus = null;
savedTrailers = null;
Expand Down
3 changes: 3 additions & 0 deletions xds/src/main/java/io/grpc/xds/ExternalProcessorFilter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -219,6 +219,9 @@ static final class ExternalProcessorFilterConfig implements FilterConfig {

static ConfigOrError<ExternalProcessorFilterConfig> create(
ExternalProcessor externalProcessor, FilterConfigParseContext context) {
if (!externalProcessor.hasProcessingMode()) {
return ConfigOrError.fromError("Missing required field: processing_mode");
}
ProcessingMode mode = externalProcessor.getProcessingMode();
GrpcService grpcService = externalProcessor.getGrpcService();
HeaderMutationRulesConfig mutationRulesConfig = null;
Expand Down
Loading
Loading