diff --git a/internal/xds/server/server_options.go b/internal/xds/server/server_options.go new file mode 100644 index 000000000000..e9e638c51385 --- /dev/null +++ b/internal/xds/server/server_options.go @@ -0,0 +1,69 @@ +/* + * + * Copyright 2026 gRPC authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package server + +import ( + "net" + + "google.golang.org/grpc" + "google.golang.org/grpc/internal/xds/xdsclient" +) + +// Options contains options used by an xDS-enabled gRPC server. +// +// This type is internal so that the public xds package can expose server +// options without also owning their application and storage. +type Options struct { + ModeCallback ServingModeCallback + ClientPoolForTesting *xdsclient.Pool + OverrideListenerResourceName func(net.Addr) string +} + +type serverOption struct { + grpc.EmptyServerOption + apply func(*Options) +} + +// NewServerOption returns a grpc.ServerOption which applies f to the internal +// xDS server options. +func NewServerOption(f func(*Options)) grpc.ServerOption { + return &serverOption{apply: f} +} + +// ApplyServerOptions applies all internal xDS server options in opts to so. +func ApplyServerOptions(opts []grpc.ServerOption, so *Options) { + for _, opt := range opts { + if o, ok := opt.(*serverOption); ok { + o.apply(so) + } + } +} + +// OverrideListenerResourceName returns a server option that overrides the LDS +// resource name selected for an xDS server listener. The supplied function is +// called by Serve with the address returned by the listener's Addr method, and +// its return value is used as-is as the LDS listener resource name. +// +// The function is called once for each Serve invocation that gets past listener +// validation and the server-stopped check. If Serve is called concurrently, the +// function may be called concurrently and must be safe for concurrent use. +func OverrideListenerResourceName(f func(net.Addr) string) grpc.ServerOption { + return NewServerOption(func(o *Options) { + o.OverrideListenerResourceName = f + }) +} diff --git a/xds/server.go b/xds/server.go index e48f809d02ba..789cae3829c0 100644 --- a/xds/server.go +++ b/xds/server.go @@ -64,7 +64,7 @@ type GRPCServer struct { gs grpcServer quit *grpcsync.Event logger *internalgrpclog.PrefixLogger - opts *serverOptions + opts *server.Options xdsC xdsclient.XDSClient xdsClientClose func() } @@ -94,8 +94,8 @@ func NewGRPCServer(opts ...grpc.ServerOption) (*GRPCServer, error) { // simplifies the code by eliminating the need for a mutex to protect the // xdsC and xdsClientClose fields. pool := xdsClientPool - if s.opts.clientPoolForTesting != nil { - pool = s.opts.clientPoolForTesting + if s.opts.ClientPoolForTesting != nil { + pool = s.opts.ClientPoolForTesting } xdsClient, xdsClientClose, err := pool.NewClient(xdsclient.NameForServer, mrl) if err != nil { @@ -104,11 +104,14 @@ func NewGRPCServer(opts ...grpc.ServerOption) (*GRPCServer, error) { // Validate the bootstrap configuration for server specific fields. - // Listener resource name template is mandatory on the server side. - cfg := xdsClient.BootstrapConfig() - if cfg.ServerListenerResourceNameTemplate() == "" { - xdsClientClose() - return nil, errors.New("missing server_listener_resource_name_template in the bootstrap configuration") + // Listener resource name template is mandatory on the server side unless a + // listener resource name override is provided. + if s.opts.OverrideListenerResourceName == nil { + cfg := xdsClient.BootstrapConfig() + if cfg.ServerListenerResourceNameTemplate() == "" { + xdsClientClose() + return nil, errors.New("missing server_listener_resource_name_template in the bootstrap configuration") + } } s.xdsC = xdsClient @@ -124,32 +127,28 @@ func NewGRPCServer(opts ...grpc.ServerOption) (*GRPCServer, error) { // the user, and handles the xDS server specific options. func (s *GRPCServer) handleServerOptions(opts []grpc.ServerOption) { so := s.defaultServerOptions() - for _, opt := range opts { - if o, ok := opt.(*serverOption); ok { - o.apply(so) - } - } + server.ApplyServerOptions(opts, so) s.opts = so } -func (s *GRPCServer) defaultServerOptions() *serverOptions { - return &serverOptions{ +func (s *GRPCServer) defaultServerOptions() *server.Options { + return &server.Options{ // A default serving mode change callback which simply logs at the // default-visible log level. This will be used if the application does not // register a mode change callback. // - // Note that this means that `s.opts.modeCallback` will never be nil and can + // Note that this means that `s.opts.ModeCallback` will never be nil and can // safely be invoked directly from `handleServingModeChanges`. - modeCallback: s.loggingServerModeChangeCallback, + ModeCallback: s.loggingServerModeChangeCallback, } } -func (s *GRPCServer) loggingServerModeChangeCallback(addr net.Addr, args ServingModeChangeArgs) { - switch args.Mode { +func (s *GRPCServer) loggingServerModeChangeCallback(addr net.Addr, mode connectivity.ServingMode, err error) { + switch mode { case connectivity.ServingModeServing: - s.logger.Errorf("Listener %q entering mode: %q", addr.String(), args.Mode) + s.logger.Errorf("Listener %q entering mode: %q", addr.String(), mode) case connectivity.ServingModeNotServing: - s.logger.Errorf("Listener %q entering mode: %q due to error: %v", addr.String(), args.Mode, args.Err) + s.logger.Errorf("Listener %q entering mode: %q due to error: %v", addr.String(), mode, err) } } @@ -188,8 +187,14 @@ func (s *GRPCServer) Serve(lis net.Listener) error { // to subscribe to for a gRPC server. If the token `%s` is present in the // string, it will be replaced with the server's listening "IP:port" (e.g., // "0.0.0.0:8080", "[::]:8080"). - cfg := s.xdsC.BootstrapConfig() - name := bootstrap.PopulateResourceTemplate(cfg.ServerListenerResourceNameTemplate(), lis.Addr().String()) + var name string + if s.opts.OverrideListenerResourceName != nil { + name = s.opts.OverrideListenerResourceName(lis.Addr()) + } else { + cfg := s.xdsC.BootstrapConfig() + name = cfg.ServerListenerResourceNameTemplate() + name = bootstrap.PopulateResourceTemplate(name, lis.Addr().String()) + } // Create a listenerWrapper which handles all functionality required by // this particular instance of Serve(). @@ -197,12 +202,7 @@ func (s *GRPCServer) Serve(lis net.Listener) error { Listener: lis, ListenerResourceName: name, XDSClient: s.xdsC, - ModeCallback: func(addr net.Addr, mode connectivity.ServingMode, err error) { - s.opts.modeCallback(addr, ServingModeChangeArgs{ - Mode: mode, - Err: err, - }) - }, + ModeCallback: s.opts.ModeCallback, }) return s.gs.Serve(lw) } diff --git a/xds/server_options.go b/xds/server_options.go index c72d93ec3f61..2afda6797b62 100644 --- a/xds/server_options.go +++ b/xds/server_options.go @@ -24,23 +24,21 @@ import ( "google.golang.org/grpc" "google.golang.org/grpc/connectivity" "google.golang.org/grpc/internal/xds/bootstrap" + internalserver "google.golang.org/grpc/internal/xds/server" "google.golang.org/grpc/internal/xds/xdsclient" ) -type serverOptions struct { - modeCallback ServingModeCallbackFunc - clientPoolForTesting *xdsclient.Pool -} - -type serverOption struct { - grpc.EmptyServerOption - apply func(*serverOptions) -} - // ServingModeCallback returns a grpc.ServerOption which allows users to // register a callback to get notified about serving mode changes. func ServingModeCallback(cb ServingModeCallbackFunc) grpc.ServerOption { - return &serverOption{apply: func(o *serverOptions) { o.modeCallback = cb }} + return internalserver.NewServerOption(func(o *internalserver.Options) { + o.ModeCallback = func(addr net.Addr, mode connectivity.ServingMode, err error) { + cb(addr, ServingModeChangeArgs{ + Mode: mode, + Err: err, + }) + } + }) } // ServingModeCallbackFunc is the callback that users can register to get @@ -77,7 +75,9 @@ func BootstrapContentsForTesting(bootstrapContents []byte) grpc.ServerOption { config, err := bootstrap.NewConfigFromContents(bootstrapContents) if err != nil { logger.Warningf("Failed to parse bootstrap contents %s for server options: %v", string(bootstrapContents), err) - return &serverOption{apply: func(o *serverOptions) { o.clientPoolForTesting = nil }} + return internalserver.NewServerOption(func(o *internalserver.Options) { + o.ClientPoolForTesting = nil + }) } return ClientPoolForTesting(xdsclient.NewPool(config)) } @@ -96,5 +96,7 @@ func BootstrapContentsForTesting(bootstrapContents []byte) grpc.ServerOption { // Notice: This API is EXPERIMENTAL and may be changed or removed in a // later release. func ClientPoolForTesting(pool *xdsclient.Pool) grpc.ServerOption { - return &serverOption{apply: func(o *serverOptions) { o.clientPoolForTesting = pool }} + return internalserver.NewServerOption(func(o *internalserver.Options) { + o.ClientPoolForTesting = pool + }) } diff --git a/xds/server_test.go b/xds/server_test.go index fc943567a6ec..9822656d41f8 100644 --- a/xds/server_test.go +++ b/xds/server_test.go @@ -42,6 +42,7 @@ import ( "google.golang.org/grpc/internal/testutils" "google.golang.org/grpc/internal/testutils/xds/e2e" "google.golang.org/grpc/internal/xds/bootstrap" + internalserver "google.golang.org/grpc/internal/xds/server" "google.golang.org/grpc/internal/xds/xdsclient" "google.golang.org/grpc/internal/xds/xdsclient/xdsresource/version" @@ -227,6 +228,84 @@ func (s) TestNewServer_Failure(t *testing.T) { } } +// TestServer_OverrideListenerResourceNameOverridesMissingTemplate verifies +// that an internal listener resource name override takes precedence over +// server_listener_resource_name_template from bootstrap, including when the +// template is absent, and that its returned name is used for the LDS watch. +func (s) TestServer_OverrideListenerResourceNameOverridesMissingTemplate(t *testing.T) { + const wantResourceName = "xdstp://foo/bar" + ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout) + defer cancel() + + ldsResourceNamesCh := make(chan []string, 1) + mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{ + OnStreamRequest: func(_ int64, req *v3discoverypb.DiscoveryRequest) error { + if req.GetTypeUrl() == version.V3ListenerURL { + select { + case ldsResourceNamesCh <- req.GetResourceNames(): + case <-ctx.Done(): + } + } + return nil + }, + }) + + bs, err := bootstrap.NewContentsForTesting(bootstrap.ConfigOptionsForTesting{ + Servers: []byte(fmt.Sprintf(`[{ + "server_uri": %q, + "channel_creds": [{"type": "insecure"}] + }]`, mgmtServer.Address)), + Node: []byte(fmt.Sprintf(`{"id": "%s"}`, uuid.New().String())), + }) + if err != nil { + t.Fatalf("Failed to create bootstrap configuration: %v", err) + } + + fs := newFakeGRPCServer() + origNewGRPCServer := newGRPCServer + newGRPCServer = func(...grpc.ServerOption) grpcServer { return fs } + defer func() { newGRPCServer = origNewGRPCServer }() + + lisAddrCh := make(chan net.Addr, 1) + resourceNameOpt := internalserver.OverrideListenerResourceName(func(addr net.Addr) string { + lisAddrCh <- addr + return wantResourceName + }) + srv, err := NewGRPCServer(resourceNameOpt, BootstrapContentsForTesting(bs)) + if err != nil { + t.Fatalf("NewGRPCServer() failed: %v", err) + } + defer srv.Stop() + + lis, err := testutils.LocalTCPListener() + if err != nil { + t.Fatalf("testutils.LocalTCPListener() failed: %v", err) + } + go func() { _ = srv.Serve(lis) }() + + // Verify that OverrideListenerResourceName receives the listener address. + select { + case gotAddr := <-lisAddrCh: + if gotAddr.String() != lis.Addr().String() { + t.Fatalf("OverrideListenerResourceName() called with address %q, want %q", gotAddr, lis.Addr()) + } + case <-ctx.Done(): + t.Fatal("Timeout waiting for OverrideListenerResourceName to be called") + } + + // Verify that the LDS watch uses the resource name returned by the override. + var gotResourceNames []string + select { + case gotResourceNames = <-ldsResourceNamesCh: + case <-ctx.Done(): + t.Fatal("Timeout waiting for an LDS request") + } + wantResourceNames := []string{wantResourceName} + if !cmp.Equal(gotResourceNames, wantResourceNames) { + t.Fatalf("LDS watch registered for names %v, want %v", gotResourceNames, wantResourceNames) + } +} + func (s) TestRegisterService(t *testing.T) { fs := newFakeGRPCServer()