addressing comments

Author: Pranjali-2501

internal/internal.go

@@ -246,9 +246,9 @@ var (
 		return cleanup
 	}
 
-	// ServerStreamWrapper returns a ServerOption that sets the internal
-	// stream wrapper used for HTTP filters.
-	ServerStreamWrapper any // func(func(any) (any, error)) grpc.ServerOption
+	// XDSFilterWrapperOption returns a ServerOption that sets the internal
+	// stream wrapper used for server-side xDS HTTP filters.
+	XDSFilterWrapperOption any // func(func(grpc.ServerStream) (grpc.ServerStream, error)) grpc.ServerOption
 )
 
 // HealthChecker defines the signature of the client-side LB channel health

internal/xds/httpfilter/httpfilter.go

@@ -95,8 +95,8 @@ type ClientInterceptor interface {
 	// 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 closes the interceptor. No new RPCs will be dispatched to this
+	// interceptor, but ongoing calls to NewStream are allowed to complete.
 	Close()
 }
 
@@ -142,8 +142,8 @@ type ServerInterceptor interface {
 	// Implementations should never return (nil, nil).
 	InterceptRPC(ss grpc.ServerStream) (grpc.ServerStream, error)
 
-	// Close closes the interceptor. Once called, no new calls to InterceptRPC are
-	// accepted. Ongoing calls to InterceptRPC are allowed to complete.
+	// Close closes the interceptor. No new RPCs will be dispatched to this
+	// interceptor, but ongoing calls to InterceptRPC are allowed to complete.
 	Close()
 }
 

internal/xds/server/routing.go

@@ -20,15 +20,13 @@ package server
 
 import (
 	"errors"
-	"fmt"
 	"strings"
 
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/internal/transport"
 	"google.golang.org/grpc/internal/xds/xdsclient/xdsresource"
 	"google.golang.org/grpc/metadata"
-	"google.golang.org/grpc/status"
 )
 
 // RouteAndProcess routes the incoming RPC to a configured route in the route
@@ -50,16 +48,16 @@ func RouteAndProcess(ss grpc.ServerStream) (grpc.ServerStream, error) {
 		if logger.V(2) {
 			logger.Infof("RPC on connection with xDS Configuration error: %v", rc.err)
 		}
-		return nil, status.Error(codes.Unavailable, fmt.Sprintf("error from xDS configuration for matched route configuration: %v", rc.err))
+		return nil, rc.statusErrWithNodeID(codes.Unavailable, "error from xDS configuration for matched route configuration: %v", rc.err)
 	}
 
 	mn, ok := grpc.Method(ctx)
 	if !ok {
-		return nil, errors.New("missing method name in incoming context")
+		return nil, rc.statusErrWithNodeID(codes.Internal, "missing method name in incoming context")
 	}
 	md, ok := metadata.FromIncomingContext(ctx)
 	if !ok {
-		return nil, errors.New("missing metadata in incoming context")
+		return nil, rc.statusErrWithNodeID(codes.Internal, "missing metadata in incoming context")
 	}
 	// A41 added logic to the core grpc implementation to guarantee that once the
 	// RPC gets to this point, there will be a single, unambiguous authority

server.go

@@ -89,15 +89,7 @@ func init() {
 	internal.MetricsRecorderForServer = func(srv *Server) estats.MetricsRecorder {
 		return istats.NewMetricsRecorderList(srv.opts.statsHandlers)
 	}
-	internal.ServerStreamWrapper = func(w func(any) (any, error)) any {
-		return serverStreamWrapper(func(ss ServerStream) (ServerStream, error) {
-			res, err := w(ss)
-			if err != nil || res == nil {
-				return nil, err
-			}
-			return res.(ServerStream), nil
-		})
-	}
+	internal.XDSFilterWrapperOption = xdsFilterWrapperOption
 }
 
 var statusOK = status.New(codes.OK, "")
@@ -681,9 +673,9 @@ func bufferPool(bufferPool mem.BufferPool) ServerOption {
 	})
 }
 
-// serverStreamWrapper returns a ServerOption that sets the server-level
+// xdsFilterWrapperOption returns a ServerOption that sets the server-level
 // stream wrapper (used internally by xDS server filters).
-func serverStreamWrapper(w func(ServerStream) (ServerStream, error)) ServerOption {
+func xdsFilterWrapperOption(w func(ServerStream) (ServerStream, error)) ServerOption {
 	return newFuncServerOption(func(o *serverOptions) {
 		o.streamWrapper = w
 	})
@@ -1450,12 +1442,7 @@ func (s *Server) processRPC(ctx context.Context, stream *transport.ServerStream,
 	var appErr error
 	var wrappedStream ServerStream = ss
 	if s.opts.streamWrapper != nil {
-		ws, err := s.opts.streamWrapper(ss)
-		if err != nil {
-			appErr = err
-		} else if ws != nil {
-			wrappedStream = ws
-		}
+		wrappedStream, appErr = s.opts.streamWrapper(ss)
 	}
 
 	if appErr == nil {

test/server_test.go

@@ -490,20 +490,17 @@ type wrappedTestStream struct {
 	grpc.ServerStream
 }
 
-// Test verifies that an internal stream wrapper option configured on the
+// Test verifies that an internal xDS filter wrapper option configured on the
 // server gets invoked and can wrap the ServerStream for both Unary and
 // Streaming RPCs.
-func (s) TestServerStreamWrapper(t *testing.T) {
+func (s) TestXDSFilterWrapperOption(t *testing.T) {
 	var wrapperCalled atomic.Bool
-	wrapper := func(ss any) (any, error) {
+	wrapper := func(ss grpc.ServerStream) (grpc.ServerStream, error) {
 		wrapperCalled.Store(true)
-		if sstream, ok := ss.(grpc.ServerStream); ok && sstream != nil {
-			return &wrappedTestStream{ServerStream: sstream}, nil
-		}
-		return nil, nil
+		return &wrappedTestStream{ServerStream: ss}, nil
 	}
 
-	opt := internal.ServerStreamWrapper.(func(func(any) (any, error)) any)(wrapper).(grpc.ServerOption)
+	opt := internal.XDSFilterWrapperOption.(func(func(grpc.ServerStream) (grpc.ServerStream, error)) grpc.ServerOption)(wrapper)
 
 	ss := &stubserver.StubServer{
 		EmptyCallF: func(context.Context, *testpb.Empty) (*testpb.Empty, error) {
@@ -524,7 +521,7 @@ func (s) TestServerStreamWrapper(t *testing.T) {
 		t.Fatalf("EmptyCall failed: %v", err)
 	}
 	if !wrapperCalled.Load() {
-		t.Fatal("ServerStreamWrapper callback was not called for Unary RPC")
+		t.Fatal("XDSFilterWrapperOption callback was not called for Unary RPC")
 	}
 
 	wrapperCalled.Store(false)
@@ -536,19 +533,19 @@ func (s) TestServerStreamWrapper(t *testing.T) {
 		t.Fatalf("Recv failed: %v", err)
 	}
 	if !wrapperCalled.Load() {
-		t.Fatal("ServerStreamWrapper callback was not called for Streaming RPC")
+		t.Fatal("XDSFilterWrapperOption callback was not called for Streaming RPC")
 	}
 }
 
-// Test verifies that if an internal stream wrapper returns an error,
+// Test verifies that if an internal xDS filter wrapper returns an error,
 // the RPC is rejected early with that status error before executing
 // handlers.
-func (s) TestServerStreamWrapper_EarlyRejection(t *testing.T) {
-	wrapper := func(any) (any, error) {
+func (s) TestXDSFilterWrapperOption_EarlyRejection(t *testing.T) {
+	wrapper := func(grpc.ServerStream) (grpc.ServerStream, error) {
 		return nil, status.Error(codes.PermissionDenied, "early rejection by internal wrapper")
 	}
 
-	opt := internal.ServerStreamWrapper.(func(func(any) (any, error)) any)(wrapper).(grpc.ServerOption)
+	opt := internal.XDSFilterWrapperOption.(func(func(grpc.ServerStream) (grpc.ServerStream, error)) grpc.ServerOption)(wrapper)
 
 	ss := &stubserver.StubServer{
 		EmptyCallF: func(context.Context, *testpb.Empty) (*testpb.Empty, error) {

xds/server.go

@@ -75,15 +75,18 @@ func NewGRPCServer(opts ...grpc.ServerOption) (*GRPCServer, error) {
 	// xdsFilterWrapper is a callback passed to the underlying gRPC server.
 	// It adapts the generic stream parameter and delegates to RouteAndProcess
 	// to execute xDS HTTP filter interceptors for each incoming RPC.
-	xdsFilterWrapper := func(ss any) (any, error) {
-		return server.RouteAndProcess(ss.(grpc.ServerStream))
+	xDSFilterWrapper := func(ss grpc.ServerStream) (grpc.ServerStream, error) {
+		return server.RouteAndProcess(ss)
 	}
-	// Construct a grpc.ServerOption that registers xdsFilterWrapper on
+	// Construct a grpc.ServerOption that registers xDSFilterWrapper on
 	// the server.
-	xdsInternalOpt := internal.ServerStreamWrapper.(func(func(any) (any, error)) any)(xdsFilterWrapper).(grpc.ServerOption)
-	opts = append(opts, xdsInternalOpt)
+	xDSFilterWrapperOption := internal.XDSFilterWrapperOption.(func(func(grpc.ServerStream) (grpc.ServerStream, error)) grpc.ServerOption)
+	newOpts := []grpc.ServerOption{
+		xDSFilterWrapperOption(xDSFilterWrapper),
+	}
+	newOpts = append(newOpts, opts...)
 	s := &GRPCServer{
-		gs:   newGRPCServer(opts...),
+		gs:   newGRPCServer(newOpts...),
 		quit: grpcsync.NewEvent(),
 	}
 	s.handleServerOptions(opts)

xds/server_test.go

@@ -147,7 +147,7 @@ func (s) TestNewServer_Success(t *testing.T) {
 					t.Fatalf("%d ServerOptions passed to grpc.Server, want %d", got, wantServerOpts)
 				}
 				// Verify that the user passed ServerOptions are forwarded as is.
-				if !reflect.DeepEqual(opts[:len(test.serverOpts)], test.serverOpts) {
+				if !reflect.DeepEqual(opts[1:], test.serverOpts) {
 					t.Fatalf("got ServerOptions %v, want %v", opts[:len(test.serverOpts)], test.serverOpts)
 				}
 				return grpc.NewServer(opts...)