xds: refactor xDS Client Interceptor API to support CallOption propagation. (#9168)

This PR refactors the client-side xDS HTTP filter interceptor API to
support direct propagation of call options. This is a prerequisite for
implementing [gRFC
A83](https://github.com/grpc/proposal/blob/master/A83-xds-gcp-authn-filter.md#a83-xds-gcp-authentication-filter),
which requires filter interceptors to append new call credentials to the
existing call options during stream creation.

RELEASE NOTES: N/A

Author: Pranjali-2501

internal/resolver/config_selector.go

@@ -24,7 +24,6 @@ import (
 	"sync"
 
 	"google.golang.org/grpc/internal/serviceconfig"
-	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/resolver"
 )
 
@@ -52,82 +51,7 @@ type RPCConfig struct {
 	Context      context.Context
 	MethodConfig serviceconfig.MethodConfig // configuration to use for this RPC
 	OnCommitted  func()                     // Called when the RPC has been committed (retries no longer possible)
-	Interceptor  ClientInterceptor
-}
-
-// ClientStream is the same as grpc.ClientStream, but defined here for circular
-// dependency reasons.
-type ClientStream interface {
-	// Header returns the header metadata received from the server if there
-	// is any. It blocks if the metadata is not ready to read.
-	Header() (metadata.MD, error)
-	// Trailer returns the trailer metadata from the server, if there is any.
-	// It must only be called after stream.CloseAndRecv has returned, or
-	// stream.Recv has returned a non-nil error (including io.EOF).
-	Trailer() metadata.MD
-	// CloseSend closes the send direction of the stream. It closes the stream
-	// when non-nil error is met. It is also not safe to call CloseSend
-	// concurrently with SendMsg.
-	CloseSend() error
-	// Context returns the context for this stream.
-	//
-	// It should not be called until after Header or RecvMsg has returned. Once
-	// called, subsequent client-side retries are disabled.
-	Context() context.Context
-	// SendMsg is generally called by generated code. On error, SendMsg aborts
-	// the stream. If the error was generated by the client, the status is
-	// returned directly; otherwise, io.EOF is returned and the status of
-	// the stream may be discovered using RecvMsg.
-	//
-	// SendMsg blocks until:
-	//   - There is sufficient flow control to schedule m with the transport, or
-	//   - The stream is done, or
-	//   - The stream breaks.
-	//
-	// SendMsg does not wait until the message is received by the server. An
-	// untimely stream closure may result in lost messages. To ensure delivery,
-	// users should ensure the RPC completed successfully using RecvMsg.
-	//
-	// It is safe to have a goroutine calling SendMsg and another goroutine
-	// calling RecvMsg on the same stream at the same time, but it is not safe
-	// to call SendMsg on the same stream in different goroutines. It is also
-	// not safe to call CloseSend concurrently with SendMsg.
-	SendMsg(m any) error
-	// RecvMsg blocks until it receives a message into m or the stream is
-	// done. It returns io.EOF when the stream completes successfully. On
-	// any other error, the stream is aborted and the error contains the RPC
-	// status.
-	//
-	// It is safe to have a goroutine calling SendMsg and another goroutine
-	// calling RecvMsg on the same stream at the same time, but it is not
-	// safe to call RecvMsg on the same stream in different goroutines.
-	RecvMsg(m any) error
-}
-
-// ClientInterceptor is an interceptor for gRPC client streams.
-type ClientInterceptor interface {
-	// NewStream creates a ClientStream for an RPC.
-	//
-	// Implementations must delegate stream creation to the provided newStream
-	// function. To intercept or override stream behavior, implementations
-	// may wrap the ClientStream returned by the delegate.
-	//
-	// Note: RPCInfo.Context is currently unused and will be nil.
-	//
-	// The done function is invoked when the RPC has finished using its
-	// underlying connection or if a connection could not be assigned. Because
-	// interceptors operate at the application layer, RPC operations may
-	// continue on the ClientStream even after done has been called. The
-	// caller must ensure done is non-nil.
-	//
-	// To ensure RPC completion notifications propagate through the entire
-	// interceptor chain, implementations must ensure that the done function
-	// passed to the delegate newStream invokes the done function passed to
-	// NewStream.
-	NewStream(ctx context.Context, ri RPCInfo, done func(), newStream func(ctx context.Context, done func()) (ClientStream, error)) (ClientStream, error)
-	// Close closes the interceptor. Once called, no new calls to NewStream are
-	// accepted. Ongoing calls to NewStream are allowed to complete.
-	Close()
+	Interceptor  any
 }
 
 // ServerInterceptor is an interceptor for incoming RPC's on gRPC server side.

internal/transport/client_stream.go

@@ -39,9 +39,8 @@ const nonGRPCDataMaxLen = 1024
 type ClientStream struct {
 	Stream // Embed for common stream functionality.
 
-	ct       *http2Client
-	done     chan struct{} // closed at the end of stream to unblock writers.
-	doneFunc func()        // invoked at the end of stream.
+	ct   *http2Client
+	done chan struct{} // closed at the end of stream to unblock writers.
 
 	headerChan chan struct{} // closed to indicate the end of header metadata.
 	header     metadata.MD   // the received header metadata

internal/transport/http2_client.go

@@ -498,7 +498,6 @@ func (t *http2Client) newStream(ctx context.Context, callHdr *CallHdr, handler s
 		ct:           t,
 		done:         make(chan struct{}),
 		headerChan:   make(chan struct{}),
-		doneFunc:     callHdr.DoneFunc,
 		statsHandler: handler,
 	}
 	s.Stream.buf.init()
@@ -999,9 +998,6 @@ func (t *http2Client) closeStream(s *ClientStream, err error, rst bool, rstCode
 	t.controlBuf.executeAndPut(addBackStreamQuota, cleanup)
 	// This will unblock write.
 	close(s.done)
-	if s.doneFunc != nil {
-		s.doneFunc()
-	}
 }
 
 // Close kicks off the shutdown process of the transport. This should be called

internal/transport/transport.go

@@ -594,8 +594,6 @@ type CallHdr struct {
 
 	PreviousAttempts int // value of grpc-previous-rpc-attempts header to set
 
-	DoneFunc func() // called when the stream is finished
-
 	// Authority is used to explicitly override the `:authority` header.
 	//
 	// This value comes from one of two sources:

internal/xds/httpfilter/extproc/ext_proc.go

@@ -26,7 +26,6 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/internal/envconfig"
 	"google.golang.org/grpc/internal/optional"
-	"google.golang.org/grpc/internal/resolver"
 	"google.golang.org/grpc/internal/xds/httpfilter"
 	"google.golang.org/grpc/internal/xds/matcher"
 	"google.golang.org/grpc/internal/xds/xdsclient/xdsresource"
@@ -201,7 +200,7 @@ type clientFilter struct{}
 
 func (clientFilter) Close() {}
 
-func (clientFilter) BuildClientInterceptor(base, override httpfilter.FilterConfig) (resolver.ClientInterceptor, error) {
+func (clientFilter) BuildClientInterceptor(base, override httpfilter.FilterConfig) (httpfilter.ClientInterceptor, error) {
 	b, ok := base.(baseConfig)
 	if !ok {
 		return nil, fmt.Errorf("extproc: incorrect config type provided (%T): %v", base, base)
@@ -230,7 +229,7 @@ func (clientFilter) BuildClientInterceptor(base, override httpfilter.FilterConfi
 }
 
 type clientInterceptor struct {
-	resolver.ClientInterceptor
+	httpfilter.ClientInterceptor
 	config    baseConfig
 	extClient v3procservicegrpc.ExternalProcessorClient
 	cancel    func() error

internal/xds/httpfilter/fault/fault.go

@@ -29,6 +29,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	iresolver "google.golang.org/grpc/internal/resolver"
 	"google.golang.org/grpc/internal/xds/httpfilter"
@@ -114,7 +115,7 @@ type clientFilter struct{}
 
 func (clientFilter) Close() {}
 
-func (clientFilter) BuildClientInterceptor(cfg, override httpfilter.FilterConfig) (iresolver.ClientInterceptor, error) {
+func (clientFilter) BuildClientInterceptor(cfg, override httpfilter.FilterConfig) (httpfilter.ClientInterceptor, error) {
 	if cfg == nil {
 		return nil, fmt.Errorf("fault: nil config provided")
 	}
@@ -147,12 +148,12 @@ type interceptor struct {
 
 var activeFaults uint32 // global active faults; accessed atomically
 
-func (i *interceptor) NewStream(ctx context.Context, _ iresolver.RPCInfo, done func(), newStream func(ctx context.Context, done func()) (iresolver.ClientStream, error)) (iresolver.ClientStream, error) {
+func (i *interceptor) NewStream(ctx context.Context, _ iresolver.RPCInfo, newStream func(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStream, error), opts ...grpc.CallOption) (grpc.ClientStream, error) {
 	if maxAF := i.config.GetMaxActiveFaults(); maxAF != nil {
 		defer atomic.AddUint32(&activeFaults, ^uint32(0)) // decrement counter
 		if af := atomic.AddUint32(&activeFaults, 1); af > maxAF.GetValue() {
 			// Would exceed maximum active fault limit.
-			return newStream(ctx, done)
+			return newStream(ctx, opts...)
 		}
 	}
 
@@ -166,7 +167,7 @@ func (i *interceptor) NewStream(ctx context.Context, _ iresolver.RPCInfo, done f
 		}
 		return nil, err
 	}
-	return newStream(ctx, done)
+	return newStream(ctx, opts...)
 }
 
 func (i *interceptor) Close() {}

internal/xds/httpfilter/httpfilter.go

@@ -21,6 +21,9 @@
 package httpfilter
 
 import (
+	"context"
+
+	"google.golang.org/grpc"
 	iresolver "google.golang.org/grpc/internal/resolver"
 	"google.golang.org/protobuf/proto"
 )
@@ -66,6 +69,24 @@ type Builder interface {
 	IsTerminal() bool
 }
 
+// ClientInterceptor is an interceptor for gRPC client streams.
+type ClientInterceptor interface {
+	// NewStream creates a ClientStream for an RPC.
+	//
+	// Implementations may delegate stream creation to the provided newStream
+	// function, passing the provided CallOption slice along with any new
+	// CallOption instances they wish to add. To intercept or override stream
+	// behavior, implementations may wrap the ClientStream returned by the
+	// delegate.
+	//
+	// Note: RPCInfo.Context is currently unused and will be nil.
+	NewStream(ctx context.Context, ri iresolver.RPCInfo, newStream func(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStream, error), opts ...grpc.CallOption) (grpc.ClientStream, error)
+
+	// Close closes the interceptor. Once called, no new calls to NewStream are
+	// accepted. Ongoing calls to NewStream are allowed to complete.
+	Close()
+}
+
 // ClientFilterBuilder is an optional interface that a Builder can implement to
 // indicate its capability to build client-side filters.
 type ClientFilterBuilder interface {
@@ -84,7 +105,7 @@ type ClientFilter interface {
 	//
 	// It is valid for this method to return a nil Interceptor and a nil error.
 	// In this case, the RPC will not be intercepted by this filter.
-	BuildClientInterceptor(config, override FilterConfig) (iresolver.ClientInterceptor, error)
+	BuildClientInterceptor(config, override FilterConfig) (ClientInterceptor, error)
 
 	// Close is called when the filter is no longer needed.
 	Close()

internal/xds/httpfilter/router/router.go

@@ -91,7 +91,7 @@ type filter struct{}
 
 func (filter) Close() {}
 
-func (filter) BuildClientInterceptor(cfg, override httpfilter.FilterConfig) (iresolver.ClientInterceptor, error) {
+func (filter) BuildClientInterceptor(cfg, override httpfilter.FilterConfig) (httpfilter.ClientInterceptor, error) {
 	if _, ok := cfg.(config); !ok {
 		return nil, fmt.Errorf("router: incorrect config type provided (%T): %v", cfg, cfg)
 	}

internal/xds/resolver/cluster_specifier_plugin_test.go

@@ -25,6 +25,7 @@ import (
 	"testing"
 
 	"github.com/google/uuid"
+	"google.golang.org/grpc"
 	"google.golang.org/grpc/balancer"
 	iresolver "google.golang.org/grpc/internal/resolver"
 	"google.golang.org/grpc/internal/testutils"
@@ -438,13 +439,16 @@ func (s) TestResolverClusterSpecifierPlugin_WithFilters(t *testing.T) {
 	if res.Interceptor == nil {
 		t.Fatal("RPCInfo does not contain interceptors list")
 	}
-
-	newStream := func(context.Context, func()) (iresolver.ClientStream, error) {
+	newStream := func(context.Context, ...grpc.CallOption) (grpc.ClientStream, error) {
 		return nil, nil
 	}
 
-	if _, err = res.Interceptor.NewStream(ctx, iresolver.RPCInfo{Method: "/service/method", Context: ctx}, func() {}, newStream); err != nil {
-		t.Fatalf("NewStream() failed with error: %v", err)
+	if interceptor, ok := res.Interceptor.(httpfilter.ClientInterceptor); ok {
+		if _, err = interceptor.NewStream(ctx, iresolver.RPCInfo{Method: "/service/method", Context: ctx}, newStream); err != nil {
+			t.Fatalf("NewStream() failed with error: %v", err)
+		}
+	} else {
+		t.Fatalf("res.Interceptor is type %T, want httpfilter.ClientInterceptor", res.Interceptor)
 	}
 
 	// Verify that first filter receives the config.

internal/xds/resolver/serviceconfig.go

@@ -35,6 +35,7 @@ import (
 	"google.golang.org/grpc/internal/wrr"
 	"google.golang.org/grpc/internal/xds/balancer/clusterimpl"
 	"google.golang.org/grpc/internal/xds/balancer/clustermanager"
+	"google.golang.org/grpc/internal/xds/httpfilter"
 	"google.golang.org/grpc/internal/xds/xdsclient/xdsresource"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/status"
@@ -104,15 +105,15 @@ type virtualHost struct {
 
 // routeCluster holds information about a cluster as referenced by a route.
 type routeCluster struct {
-	name        string                      // Name of the cluster.
-	interceptor iresolver.ClientInterceptor // HTTP filters to run for RPCs matching this route.
+	name        string                       // Name of the cluster.
+	interceptor httpfilter.ClientInterceptor // HTTP filters to run for RPCs matching this route.
 }
 
 type route struct {
-	m                 *xdsresource.CompositeMatcher // converted from route matchers
-	actionType        xdsresource.RouteActionType   // holds route action type
-	clusters          wrr.WRR                       // holds *routeCluster entries
-	interceptors      []iresolver.ClientInterceptor // Interceptors across clusters belonging to this route
+	m                 *xdsresource.CompositeMatcher  // converted from route matchers
+	actionType        xdsresource.RouteActionType    // holds route action type
+	clusters          wrr.WRR                        // holds *routeCluster entries
+	interceptors      []httpfilter.ClientInterceptor // Interceptors across clusters belonging to this route
 	maxStreamDuration time.Duration
 	retryConfig       *xdsresource.RetryConfig
 	hashPolicies      []*xdsresource.HashPolicy