xds/extproc: adds ClientInterceptor and ClientStream implementation for normal mode for gRFC A93 by eshitachandwani · Pull Request #9174 · grpc/grpc-go

eshitachandwani · 2026-06-08T11:34:55Z

This PR add implementation of NewStream and ClientStream for normal mode for A93: xds-ext-proc.
This PR does not include channel retention , metrics and observability mode.

#ext-proc-a93

RELEASE NOTES: None

codecov · 2026-06-08T11:38:03Z

Codecov Report

❌ Patch coverage is 69.11958% with 235 lines in your changes missing coverage. Please review.
✅ Project coverage is 83.00%. Comparing base (f8a85fa) to head (c45fb61).
⚠️ Report is 3 commits behind head on master.

Files with missing lines	Patch %	Lines
internal/xds/httpfilter/extproc/ext_proc.go	65.44%	164 Missing and 63 partials ⚠️
internal/xds/httpfilter/extconfig.go	95.95%	2 Missing and 2 partials ⚠️
...ternal/xds/httpfilter/extproc/internal/internal.go	0.00%	4 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##           master    #9174      +/-   ##
==========================================
- Coverage   83.30%   83.00%   -0.31%     
==========================================
  Files         421      422       +1     
  Lines       34061    34832     +771     
==========================================
+ Hits        28375    28911     +536     
- Misses       4261     4425     +164     
- Partials     1425     1496      +71

Files with missing lines	Coverage Δ
internal/resolver/config_selector.go	`100.00% <ø> (ø)`
internal/xds/httpfilter/extproc/config.go	`89.47% <ø> (+5.26%)`	⬆️
stream.go	`81.96% <100.00%> (ø)`
xds/xds.go	`40.90% <ø> (ø)`
internal/xds/httpfilter/extconfig.go	`95.16% <95.95%> (+3.16%)`	⬆️
...ternal/xds/httpfilter/extproc/internal/internal.go	`0.00% <0.00%> (ø)`
internal/xds/httpfilter/extproc/ext_proc.go	`68.11% <65.44%> (-8.82%)`	⬇️

... and 27 files with indirect coverage changes

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

….Done case becuase also cancel context incase of error to avoid calls to dataplane stream hanging.

easwars · 2026-07-07T18:41:12Z

+	req := cs.newProcessingRequest(true)
+	req.Request = &v3procservicepb.ProcessingRequest_RequestBody{
+		RequestBody: &v3procservicepb.HttpBody{
+			EndOfStream:               true,


In the "Messages Sent To the ext_proc Server" section of the gRFC, this field is documented as follows:

end_of_stream: For client messages, may be true if the client sent a half-close at the same time as the last message. For server messages, will always be false.

Given the above, shouldn't this be set to false (i.e., not set at all) because our CloseSend does not provide a way to half-close with a message at the same time.

That is what I had earlier , but then the gRFC changed to say that for response , we will consider EOS without message only if EOS is also true , so I thought that should be true for request too.

end_of_stream_without_message : Will be set to true to indicate a half-close with no message to send. Ignored if end_of_stream is false.

And I have a comment on the gRFC to clarify the same : grpc/proposal#484 (comment)
Let me know what you think.

Can we ping that comment thread on the gRFC please?

easwars · 2026-07-14T18:51:23Z

+				cs.failProcStream(fmt.Errorf("extproc: external processor returned invalid status instead of CONTINUE for response headers"))
+				return
+			}
+			if err = cs.config.mutationRules.ApplyAdditions(header.GetResponse().GetHeaderMutation().GetSetHeaders(), cs.responseHeader); err != nil {


Any update here?

easwars · 2026-07-16T06:33:14Z

+	// type to send the data to the dataplane server.
+	if !cs.reqForwardingStarted {
+		cs.reqForwardingStarted = true
+		go cs.requestForwardingToDataplaneLoop(msg.ProtoReflect().Type())


Are we certain that this is always the same type for the life of an RPC?

(Less important): It is possible to get this type from some descriptor to avoid this?

Are we certain that this is always the same type for the life of an RPC?

I am not sure what you mean, but I dont think it is possible to change the proto mid RPC.

(Less important): It is possible to get this type from some descriptor to avoid this?

I dont think there is an easy or direct way to get this from any descriptor, I dont see this message type being stored anywhere inside grpc go.

I am not sure what you mean, but I dont think it is possible to change the proto mid RPC.

What I meant only applies when someone is not using code generated using our protoc-gen-go-grpc. Then, they could call SendMsg with different type of proto messages over the life of a stream.

easwars · 2026-07-16T21:45:29Z

+	req := cs.newProcessingRequest(true)
+	req.Request = &v3procservicepb.ProcessingRequest_RequestBody{
+		RequestBody: &v3procservicepb.HttpBody{
+			EndOfStream:               true,


Can we ping that comment thread on the gRFC please?

easwars · 2026-07-16T21:51:25Z

+				if strings.HasSuffix(key, "-bin") {
+					rawValue = []byte(base64.StdEncoding.EncodeToString(rawValue))
+				}
+				headerMap.Headers = append(headerMap.Headers, &v3corepb.HeaderValue{
+					Key:      key,
+					RawValue: rawValue,
+				})
+			}


Couple of simplifications here:

Make this block into a helper and call it from both here and down below and pass it a key of type []byte

When having to do base64 encoding, use the base64.StdEncoding.Encode function which works on byte slices instead of string and therefore you can avoid one conversion back from a string into a []byte

easwars · 2026-07-16T21:54:47Z

+	// type to send the data to the dataplane server.
+	if !cs.reqForwardingStarted {
+		cs.reqForwardingStarted = true
+		go cs.requestForwardingToDataplaneLoop(msg.ProtoReflect().Type())


I am not sure what you mean, but I dont think it is possible to change the proto mid RPC.

What I meant only applies when someone is not using code generated using our protoc-gen-go-grpc. Then, they could call SendMsg with different type of proto messages over the life of a stream.

easwars · 2026-07-16T22:18:10Z

+			}
+			cs.mutatedReqBuffer.Load()
+
+			if streamedResp.GetEndOfStream() && streamedResp.GetEndOfStreamWithoutMessage() {


Can we only check streamedResp.GetEndOfStreamWithoutMessage here? The check for streamedResp.GetEndOfStream down below will catch the case where this entry in the buffer represents a message+EOS.

easwars · 2026-07-16T22:19:40Z

+}
+
+// createDataplaneStream initializes the underlying gRPC dataplane stream using
+// the provided newStream function.


Nit: Can we document that if dataplane stream creation fails, the top-level context used by the proc filter is cancelled and thereby all spawned goroutines will terminate.

easwars · 2026-07-16T23:09:15Z

+			// Signal that the response header is modified and ready to be sent to the
+			// client, so that if there is any buffered response body, it can be sent
+			// after the header.
+			cs.responseHeaderModified.Fire()


By firing this event from here (and similarly for the trailers case below), how are we guaranteeing that ordering is maintained for response messages?

Other response messages to be sent to the client application are added to the mutatedRespBuffer and read from there when the application calls Recv. Shouldn't the headers and trailers go through that buffer as well?

easwars · 2026-07-16T23:19:26Z

+	case <-cs.ctx.Done():
+		return nil
+	}
+	return cs.responseTrailers


Nit: Can we move this return to be under the case <-cs.responseTrailerModified.Done() to be more explicit.

easwars · 2026-07-16T23:30:11Z

+	if cs.trailersOnly {
+		return nil, nil
+	}
+	return cs.responseHeader, nil


Nit: Can we move this block to be under the case <-cs.responseHeaderModified.Done() to be more explicit.

easwars · 2026-07-16T23:33:00Z

+			trailers := dataplaneStream.Trailer()
+			if len(trailers) > 0 {
+				header = trailers
+				cs.trailersOnly = true
+			}


Nit: Possible simplification:

if header == nil { // A trailers-only response returns nil from dataplaneStream.Header(). headers = dataplaneStream.Trailer() if len(headers) > 0 { cs.trailersOnly = true } }

easwars · 2026-07-16T23:39:19Z

+			}
+			cs.mutatedRespBuffer.Put(streamedResp)
+
+		case resp.GetResponseHeaders() != nil:


We probably need a little bit more safetey checks here I feel. What if the proc server already sent us a response headers message and we sent cs.responseHeader to the application. Now, the proc server sends one more response headers (clearly a misbehaving proc server), then we end up mutating the same metadata.MD that we've given the client application and that will lead to a data race.

easwars · 2026-07-17T00:02:07Z

+}
+
+func (cs *clientStream) RecvMsg(m any) error {
+	if cs.procStreamFailed.HasFired() {


Based on my other comment where I requested the use of an atomic boolean to guard the inside of initiateResponseHeaderProcessing, can you also check if this condition procStreamFailed can be taken care of by that method. If we can do that (possibly by calling waitForDataplaneStream outside of the boolean guard), then we can directly call initiateResponseHeaderProcessing. Something to consider.

easwars · 2026-07-17T00:06:58Z

+			if val := cs.procStreamErr.Load(); val != nil {
+				return val.(error)
+			}


It is interesting that you use this pattern instead of the more commonly used pattern:

if cs.procStreamFailed.HasFired() { return cs.procStreamErr.Load().(error) }

Any specific reason for this?

easwars · 2026-07-17T00:09:43Z

+			cs.initiateResponseTrailerProcessing()
+			return cs.waitForTrailerProcessing(err)


Why are we blocking the application's RecvMsg call on trailers? Shouldn't we return the error we received from the dataplane stream and have them call Trailer if they want to see the trailers?

easwars · 2026-07-17T00:11:59Z

+
+	// Start the background receiving loop on the first RecvMsg call to capture
+	// the type of message to be received.
+	if !cs.responseRecvStarted {


Nit: should this be called responseForwardingStarted or respForwardingStarted, similar to reqForwardingStarted?

easwars · 2026-07-17T00:15:36Z

+
+		newMsg := msgType.New().Interface()
+		if err := dataplaneStream.RecvMsg(newMsg); err != nil {
+			cs.initiateResponseTrailerProcessing()


Same concern about why we need to block for trailers here?

easwars · 2026-07-17T00:22:57Z

+			if val := cs.procStreamErr.Load(); val != nil {
+				return val.(error)
+			}
+			s, err := cs.waitForDataplaneStream(cs.ctx)
+			if err != nil {
+				return err
+			}
+			if err := s.RecvMsg(m); err != nil {
+				cs.initiateResponseTrailerProcessing()
+				return cs.waitForTrailerProcessing(err)
+			}


This block of code is very similar to the one up top where we handle receiving directly from the dataplane stream when some conditions are met. Maybe, we can find a way to share this code?

eshitachandwani added 4 commits June 8, 2026 09:41

all changes without proofread

1d2c398

merge master

b1a0e64

formatting changes

37df0bb

formatting changes

f7b5a43

eshitachandwani added this to the 1.83 Release milestone Jun 8, 2026

eshitachandwani added Type: Feature New features or improvements in behavior Area: xDS Includes everything xDS related, including LB policies used with xDS. labels Jun 8, 2026

eshitachandwani added 4 commits June 8, 2026 17:13

formatting changes

3e971ad

formatting changes

fb113e6

test changes

c72d1e7

fix comments

b7cb5df

eshitachandwani requested review from easwars and mbissa June 8, 2026 15:43

eshitachandwani assigned easwars and mbissa Jun 8, 2026

eshitachandwani added 2 commits June 11, 2026 02:22

correct FMA, grfc change , add sequence check

e2d1249

vet fix

d4cc4e8

easwars reviewed Jun 11, 2026

View reviewed changes

Comment thread internal/resolver/config_selector.go

Comment thread internal/xds/httpfilter/extproc/ext_proc.go

Comment thread internal/xds/httpfilter/extproc/ext_proc.go Outdated

Comment thread internal/xds/httpfilter/extproc/ext_proc.go Outdated

comments

079a068

easwars reviewed Jun 11, 2026

View reviewed changes

easwars assigned eshitachandwani and unassigned easwars Jun 11, 2026

eshitachandwani mentioned this pull request Jun 12, 2026

xds/extproc: adds ClientInterceptor and ClientStream implementation for observability mode for gRFC A93 #9180

Open

eshitachandwani and others added 3 commits June 15, 2026 14:42

remove failstream call from Send and add extStream error check in ctx…

f2813ec

….Done case becuase also cancel context incase of error to avoid calls to dataplane stream hanging.

Merge branch 'master' into newStream

0859c1b

change the interface replace done with opts

20543fc

eshitachandwani requested a review from easwars June 15, 2026 09:56

eshitachandwani assigned easwars and unassigned eshitachandwani Jun 15, 2026

eshitachandwani removed their assignment Jul 1, 2026

eshitachandwani added 2 commits July 6, 2026 19:30

add trailer only handling and tests and remove dead code

8eee206

fix test race

2971c45

easwars reviewed Jul 7, 2026

View reviewed changes

Comment thread internal/xds/httpfilter/extproc/ext_proc.go Outdated

Comment thread internal/xds/httpfilter/extproc/ext_proc.go

Comment thread internal/xds/httpfilter/extproc/ext_proc.go

Comment thread internal/xds/httpfilter/extproc/ext_proc.go

easwars reviewed Jul 7, 2026

View reviewed changes

easwars reviewed Jul 8, 2026

View reviewed changes

mbissa removed their assignment Jul 8, 2026

review comment

641ad7a

eshitachandwani requested a review from easwars July 8, 2026 21:30

fix errors

5b49e7c

easwars modified the milestones: 1.83 Release, 1.84 Release Jul 14, 2026

easwars reviewed Jul 14, 2026

View reviewed changes

eshitachandwani added 3 commits July 15, 2026 22:03

correct comments and trailer behaviour

6f17e5c

Merge remote-tracking branch 'upstream' into newStream

b8511f3

bounded buffer correct

26293f3

eshitachandwani requested a review from easwars July 15, 2026 18:39

easwars reviewed Jul 16, 2026

View reviewed changes

Comment thread internal/xds/httpfilter/extproc/ext_proc.go Outdated

Comment thread internal/xds/httpfilter/extproc/ext_proc.go Outdated

Comment thread internal/xds/httpfilter/extproc/ext_proc_ext_test.go Outdated

easwars reviewed Jul 16, 2026

View reviewed changes

easwars assigned eshitachandwani and unassigned easwars Jul 16, 2026

eshitachandwani added 2 commits July 16, 2026 21:12

review comments

ebb4c35

bin header

c45fb61

eshitachandwani requested a review from easwars July 16, 2026 18:39

eshitachandwani assigned easwars and unassigned eshitachandwani Jul 16, 2026

easwars reviewed Jul 16, 2026

View reviewed changes

easwars reviewed Jul 17, 2026

View reviewed changes

easwars assigned eshitachandwani and unassigned easwars Jul 17, 2026

		cs.initiateResponseTrailerProcessing()
		return cs.waitForTrailerProcessing(err)

Uh oh!

Conversation

eshitachandwani commented Jun 8, 2026

Uh oh!

codecov Bot commented Jun 8, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

codecov Bot commented Jun 8, 2026 •

edited

Loading