xds: pass options by value to helper routines which setup the managem…

…ent server in tests (#5833)

internal/testutils/xds/e2e/server.go

@@ -104,18 +104,16 @@ type ManagementServerOptions struct {
 // resource snapshot held by the management server, as required by the test
 // logic. When the test is done, it should call the Stop() method to cleanup
 // resources allocated by the management server.
-func StartManagementServer(opts *ManagementServerOptions) (*ManagementServer, error) {
+func StartManagementServer(opts ManagementServerOptions) (*ManagementServer, error) {
 	// Create a snapshot cache. The first parameter to NewSnapshotCache()
 	// controls whether the server should wait for all resources to be
 	// explicitly named in the request before responding to any of them.
-	wait := opts == nil || !opts.AllowResourceSubset
+	wait := !opts.AllowResourceSubset
 	cache := v3cache.NewSnapshotCache(wait, v3cache.IDHash{}, serverLogger{})
 	logger.Infof("Created new snapshot cache...")
 
-	var lis net.Listener
-	if opts != nil && opts.Listener != nil {
-		lis = opts.Listener
-	} else {
+	lis := opts.Listener
+	if lis == nil {
 		var err error
 		lis, err = net.Listen("tcp", "localhost:0")
 		if err != nil {
@@ -126,14 +124,11 @@ func StartManagementServer(opts *ManagementServerOptions) (*ManagementServer, er
 	// Cancelling the context passed to the server is the only way of stopping it
 	// at the end of the test.
 	ctx, cancel := context.WithCancel(context.Background())
-	callbacks := v3server.CallbackFuncs{}
-	if opts != nil {
-		callbacks = v3server.CallbackFuncs{
-			StreamOpenFunc:     opts.OnStreamOpen,
-			StreamClosedFunc:   opts.OnStreamClosed,
-			StreamRequestFunc:  opts.OnStreamRequest,
-			StreamResponseFunc: opts.OnStreamResponse,
-		}
+	callbacks := v3server.CallbackFuncs{
+		StreamOpenFunc:     opts.OnStreamOpen,
+		StreamClosedFunc:   opts.OnStreamClosed,
+		StreamRequestFunc:  opts.OnStreamRequest,
+		StreamResponseFunc: opts.OnStreamResponse,
 	}
 
 	// Create an xDS management server and register the ADS implementation

internal/testutils/xds/e2e/setup_management_server.go

@@ -41,7 +41,7 @@ import (
 // - bootstrap contents to be used by the client
 // - xDS resolver builder to be used by the client
 // - a cleanup function to be invoked at the end of the test
-func SetupManagementServer(t *testing.T, opts *ManagementServerOptions) (*ManagementServer, string, []byte, resolver.Builder, func()) {
+func SetupManagementServer(t *testing.T, opts ManagementServerOptions) (*ManagementServer, string, []byte, resolver.Builder, func()) {
 	t.Helper()
 
 	// Spin up an xDS management server on a local port.

test/xds/xds_client_ack_nack_test.go

@@ -81,7 +81,7 @@ func (s) TestClientResourceVersionAfterStreamRestart(t *testing.T) {
 
 	// Map from stream id to a map of resource type to resource version.
 	ackVersionsMap := make(map[int64]map[string]string)
-	managementServer, nodeID, _, resolver, cleanup1 := e2e.SetupManagementServer(t, &e2e.ManagementServerOptions{
+	managementServer, nodeID, _, resolver, cleanup1 := e2e.SetupManagementServer(t, e2e.ManagementServerOptions{
 		Listener: lis,
 		OnStreamRequest: func(id int64, req *v3discoverypb.DiscoveryRequest) error {
 			// Return early under the following circumstances:

test/xds/xds_client_affinity_test.go

@@ -88,7 +88,7 @@ func (s) TestClientSideAffinitySanityCheck(t *testing.T) {
 		return func() { envconfig.XDSRingHash = old }
 	}()()
 
-	managementServer, nodeID, _, resolver, cleanup1 := e2e.SetupManagementServer(t, nil)
+	managementServer, nodeID, _, resolver, cleanup1 := e2e.SetupManagementServer(t, e2e.ManagementServerOptions{})
 	defer cleanup1()
 
 	port, cleanup2 := startTestService(t, nil)

test/xds/xds_client_federation_test.go

@@ -54,15 +54,15 @@ func (s) TestClientSideFederation(t *testing.T) {
 	defer func() { envconfig.XDSFederation = oldXDSFederation }()
 
 	// Start a management server as the default authority.
-	serverDefaultAuth, err := e2e.StartManagementServer(nil)
+	serverDefaultAuth, err := e2e.StartManagementServer(e2e.ManagementServerOptions{})
 	if err != nil {
 		t.Fatalf("Failed to spin up the xDS management server: %v", err)
 	}
 	t.Cleanup(serverDefaultAuth.Stop)
 
 	// Start another management server as the other authority.
 	const nonDefaultAuth = "non-default-auth"
-	serverAnotherAuth, err := e2e.StartManagementServer(nil)
+	serverAnotherAuth, err := e2e.StartManagementServer(e2e.ManagementServerOptions{})
 	if err != nil {
 		t.Fatalf("Failed to spin up the xDS management server: %v", err)
 	}

test/xds/xds_client_integration_test.go

@@ -74,7 +74,7 @@ func startTestService(t *testing.T, server *stubserver.StubServer) (uint32, func
 }
 
 func (s) TestClientSideXDS(t *testing.T) {
-	managementServer, nodeID, _, resolver, cleanup1 := e2e.SetupManagementServer(t, nil)
+	managementServer, nodeID, _, resolver, cleanup1 := e2e.SetupManagementServer(t, e2e.ManagementServerOptions{})
 	defer cleanup1()
 
 	port, cleanup2 := startTestService(t, nil)

test/xds/xds_client_outlier_detection_test.go

@@ -49,7 +49,7 @@ import (
 // Detection balancer. This test verifies that an RPC is able to proceed
 // normally with this configuration.
 func (s) TestOutlierDetection_NoopConfig(t *testing.T) {
-	managementServer, nodeID, _, resolver, cleanup1 := e2e.SetupManagementServer(t, nil)
+	managementServer, nodeID, _, resolver, cleanup1 := e2e.SetupManagementServer(t, e2e.ManagementServerOptions{})
 	defer cleanup1()
 
 	port, cleanup2 := startTestService(t, nil)
@@ -166,7 +166,7 @@ func checkRoundRobinRPCs(ctx context.Context, client testpb.TestServiceClient, a
 // the unhealthy upstream is ejected, RPC's should regularly round robin across
 // all three upstreams.
 func (s) TestOutlierDetectionWithOutlier(t *testing.T) {
-	managementServer, nodeID, _, r, cleanup := e2e.SetupManagementServer(t, nil)
+	managementServer, nodeID, _, r, cleanup := e2e.SetupManagementServer(t, e2e.ManagementServerOptions{})
 	defer cleanup()
 
 	// Working backend 1.

test/xds/xds_client_retry_test.go

@@ -49,7 +49,7 @@ func (s) TestClientSideRetry(t *testing.T) {
 		},
 	}
 
-	managementServer, nodeID, _, resolver, cleanup1 := e2e.SetupManagementServer(t, nil)
+	managementServer, nodeID, _, resolver, cleanup1 := e2e.SetupManagementServer(t, e2e.ManagementServerOptions{})
 	defer cleanup1()
 
 	port, cleanup2 := startTestService(t, ss)

test/xds/xds_rls_clusterspecifier_plugin_test.go

@@ -130,7 +130,7 @@ func testRLSinxDS(t *testing.T, lbPolicy e2e.LoadBalancingPolicy) {
 	// Set up all components and configuration necessary - management server,
 	// xDS resolver, fake RLS Server, and xDS configuration which specifies an
 	// RLS Balancer that communicates to this set up fake RLS Server.
-	managementServer, nodeID, _, resolver, cleanup1 := e2e.SetupManagementServer(t, nil)
+	managementServer, nodeID, _, resolver, cleanup1 := e2e.SetupManagementServer(t, e2e.ManagementServerOptions{})
 	defer cleanup1()
 	port, cleanup2 := startTestService(t, nil)
 	defer cleanup2()

test/xds/xds_security_config_nack_test.go

@@ -41,7 +41,7 @@ func (s) TestUnmarshalListener_WithUpdateValidatorFunc(t *testing.T) {
 		missingIdentityProviderInstance = "missing-identity-provider-instance"
 		missingRootProviderInstance     = "missing-root-provider-instance"
 	)
-	managementServer, nodeID, bootstrapContents, resolver, cleanup1 := e2e.SetupManagementServer(t, nil)
+	managementServer, nodeID, bootstrapContents, resolver, cleanup1 := e2e.SetupManagementServer(t, e2e.ManagementServerOptions{})
 	defer cleanup1()
 
 	lis, cleanup2 := setupGRPCServer(t, bootstrapContents)
@@ -324,7 +324,7 @@ func (s) TestUnmarshalCluster_WithUpdateValidatorFunc(t *testing.T) {
 			// SetupManagementServer() sets up a bootstrap file with certificate
 			// provider instance names: `e2e.ServerSideCertProviderInstance` and
 			// `e2e.ClientSideCertProviderInstance`.
-			managementServer, nodeID, _, resolver, cleanup1 := e2e.SetupManagementServer(t, nil)
+			managementServer, nodeID, _, resolver, cleanup1 := e2e.SetupManagementServer(t, e2e.ManagementServerOptions{})
 			defer cleanup1()
 
 			port, cleanup2 := startTestService(t, nil)

test/xds/xds_server_integration_test.go

@@ -125,7 +125,7 @@ func hostPortFromListener(lis net.Listener) (string, uint32, error) {
 //     the client and the server. This results in both of them using the
 //     configured fallback credentials (which is insecure creds in this case).
 func (s) TestServerSideXDS_Fallback(t *testing.T) {
-	managementServer, nodeID, bootstrapContents, resolver, cleanup1 := e2e.SetupManagementServer(t, nil)
+	managementServer, nodeID, bootstrapContents, resolver, cleanup1 := e2e.SetupManagementServer(t, e2e.ManagementServerOptions{})
 	defer cleanup1()
 
 	lis, cleanup2 := setupGRPCServer(t, bootstrapContents)
@@ -207,7 +207,7 @@ func (s) TestServerSideXDS_FileWatcherCerts(t *testing.T) {
 	}
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			managementServer, nodeID, bootstrapContents, resolver, cleanup1 := e2e.SetupManagementServer(t, nil)
+			managementServer, nodeID, bootstrapContents, resolver, cleanup1 := e2e.SetupManagementServer(t, e2e.ManagementServerOptions{})
 			defer cleanup1()
 
 			lis, cleanup2 := setupGRPCServer(t, bootstrapContents)
@@ -277,7 +277,7 @@ func (s) TestServerSideXDS_FileWatcherCerts(t *testing.T) {
 // configuration pointing to the use of the file_watcher plugin and we verify
 // that the same client is now able to successfully make an RPC.
 func (s) TestServerSideXDS_SecurityConfigChange(t *testing.T) {
-	managementServer, nodeID, bootstrapContents, resolver, cleanup1 := e2e.SetupManagementServer(t, nil)
+	managementServer, nodeID, bootstrapContents, resolver, cleanup1 := e2e.SetupManagementServer(t, e2e.ManagementServerOptions{})
 	defer cleanup1()
 
 	lis, cleanup2 := setupGRPCServer(t, bootstrapContents)

test/xds/xds_server_rbac_test.go

@@ -60,7 +60,7 @@ func (s) TestServerSideXDS_RouteConfiguration(t *testing.T) {
 	defer func() {
 		envconfig.XDSRBAC = oldRBAC
 	}()
-	managementServer, nodeID, bootstrapContents, resolver, cleanup1 := e2e.SetupManagementServer(t, nil)
+	managementServer, nodeID, bootstrapContents, resolver, cleanup1 := e2e.SetupManagementServer(t, e2e.ManagementServerOptions{})
 	defer cleanup1()
 
 	lis, cleanup2 := setupGRPCServer(t, bootstrapContents)
@@ -605,7 +605,7 @@ func (s) TestRBACHTTPFilter(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			func() {
-				managementServer, nodeID, bootstrapContents, resolver, cleanup1 := e2e.SetupManagementServer(t, nil)
+				managementServer, nodeID, bootstrapContents, resolver, cleanup1 := e2e.SetupManagementServer(t, e2e.ManagementServerOptions{})
 				defer cleanup1()
 
 				lis, cleanup2 := setupGRPCServer(t, bootstrapContents)
@@ -790,7 +790,7 @@ func (s) TestRBACToggledOn_WithBadRouteConfiguration(t *testing.T) {
 		envconfig.XDSRBAC = oldRBAC
 	}()
 
-	managementServer, nodeID, bootstrapContents, resolver, cleanup1 := e2e.SetupManagementServer(t, nil)
+	managementServer, nodeID, bootstrapContents, resolver, cleanup1 := e2e.SetupManagementServer(t, e2e.ManagementServerOptions{})
 	defer cleanup1()
 
 	lis, cleanup2 := setupGRPCServer(t, bootstrapContents)
@@ -847,7 +847,7 @@ func (s) TestRBACToggledOff_WithBadRouteConfiguration(t *testing.T) {
 		envconfig.XDSRBAC = oldRBAC
 	}()
 
-	managementServer, nodeID, bootstrapContents, resolver, cleanup1 := e2e.SetupManagementServer(t, nil)
+	managementServer, nodeID, bootstrapContents, resolver, cleanup1 := e2e.SetupManagementServer(t, e2e.ManagementServerOptions{})
 	defer cleanup1()
 
 	lis, cleanup2 := setupGRPCServer(t, bootstrapContents)

test/xds/xds_server_serving_mode_test.go

@@ -43,7 +43,7 @@ import (
 // change callback is not invoked and client connections to the server are not
 // recycled.
 func (s) TestServerSideXDS_RedundantUpdateSuppression(t *testing.T) {
-	managementServer, nodeID, bootstrapContents, _, cleanup := e2e.SetupManagementServer(t, nil)
+	managementServer, nodeID, bootstrapContents, _, cleanup := e2e.SetupManagementServer(t, e2e.ManagementServerOptions{})
 	defer cleanup()
 
 	creds, err := xdscreds.NewServerCredentials(xdscreds.ServerOptions{FallbackCreds: insecure.NewCredentials()})
@@ -163,7 +163,7 @@ func (s) TestServerSideXDS_RedundantUpdateSuppression(t *testing.T) {
 // xDS enabled gRPC servers. It verifies that appropriate mode changes happen in
 // the server, and also verifies behavior of clientConns under these modes.
 func (s) TestServerSideXDS_ServingModeChanges(t *testing.T) {
-	managementServer, nodeID, bootstrapContents, _, cleanup := e2e.SetupManagementServer(t, nil)
+	managementServer, nodeID, bootstrapContents, _, cleanup := e2e.SetupManagementServer(t, e2e.ManagementServerOptions{})
 	defer cleanup()
 
 	// Configure xDS credentials to be used on the server-side.

xds/csds/csds_test.go

@@ -236,7 +236,7 @@ func commonSetup(ctx context.Context, t *testing.T) (xdsclient.XDSClient, *e2e.M
 
 	// Spin up a xDS management server on a local port.
 	nodeID := uuid.New().String()
-	fs, err := e2e.StartManagementServer(nil)
+	fs, err := e2e.StartManagementServer(e2e.ManagementServerOptions{})
 	if err != nil {
 		t.Fatal(err)
 	}

xds/internal/balancer/clusterresolver/e2e_test/eds_impl_test.go

@@ -175,7 +175,7 @@ func clientEndpointsResource(nodeID, edsServiceName string, localities []localit
 //  4. Replace the backend. Test verifies that all RPCs reach the new backend.
 func (s) TestEDS_OneLocality(t *testing.T) {
 	// Spin up a management server to receive xDS resources from.
-	managementServer, nodeID, bootstrapContents, _, cleanup1 := e2e.SetupManagementServer(t, nil)
+	managementServer, nodeID, bootstrapContents, _, cleanup1 := e2e.SetupManagementServer(t, e2e.ManagementServerOptions{})
 	defer cleanup1()
 
 	// Start backend servers which provide an implementation of the TestService.
@@ -280,7 +280,7 @@ func (s) TestEDS_OneLocality(t *testing.T) {
 // weighted roundrobined across them.
 func (s) TestEDS_MultipleLocalities(t *testing.T) {
 	// Spin up a management server to receive xDS resources from.
-	managementServer, nodeID, bootstrapContents, _, cleanup1 := e2e.SetupManagementServer(t, nil)
+	managementServer, nodeID, bootstrapContents, _, cleanup1 := e2e.SetupManagementServer(t, e2e.ManagementServerOptions{})
 	defer cleanup1()
 
 	// Start backend servers which provide an implementation of the TestService.
@@ -403,7 +403,7 @@ func (s) TestEDS_MultipleLocalities(t *testing.T) {
 // traffic is routed only to backends deemed capable of receiving traffic.
 func (s) TestEDS_EndpointsHealth(t *testing.T) {
 	// Spin up a management server to receive xDS resources from.
-	managementServer, nodeID, bootstrapContents, _, cleanup1 := e2e.SetupManagementServer(t, nil)
+	managementServer, nodeID, bootstrapContents, _, cleanup1 := e2e.SetupManagementServer(t, e2e.ManagementServerOptions{})
 	defer cleanup1()
 
 	// Start backend servers which provide an implementation of the TestService.
@@ -483,7 +483,7 @@ func (s) TestEDS_EndpointsHealth(t *testing.T) {
 // removed" error.
 func (s) TestEDS_EmptyUpdate(t *testing.T) {
 	// Spin up a management server to receive xDS resources from.
-	managementServer, nodeID, bootstrapContents, _, cleanup1 := e2e.SetupManagementServer(t, nil)
+	managementServer, nodeID, bootstrapContents, _, cleanup1 := e2e.SetupManagementServer(t, e2e.ManagementServerOptions{})
 	defer cleanup1()
 
 	// Start backend servers which provide an implementation of the TestService.

xds/internal/httpfilter/fault/fault_test.go

@@ -100,7 +100,7 @@ func (*testService) FullDuplexCall(stream testpb.TestService_FullDuplexCallServe
 func clientSetup(t *testing.T) (*e2e.ManagementServer, string, uint32, func()) {
 	// Spin up a xDS management server on a local port.
 	nodeID := uuid.New().String()
-	fs, err := e2e.StartManagementServer(nil)
+	fs, err := e2e.StartManagementServer(e2e.ManagementServerOptions{})
 	if err != nil {
 		t.Fatal(err)
 	}

xds/internal/test/e2e/controlplane.go

@@ -33,7 +33,7 @@ type controlPlane struct {
 
 func newControlPlane() (*controlPlane, error) {
 	// Spin up an xDS management server on a local port.
-	server, err := e2e.StartManagementServer(nil)
+	server, err := e2e.StartManagementServer(e2e.ManagementServerOptions{})
 	if err != nil {
 		return nil, fmt.Errorf("failed to spin up the xDS management server: %v", err)
 	}

xds/internal/xdsclient/e2e_test/authority_test.go

@@ -77,14 +77,14 @@ func setupForAuthorityTests(ctx context.Context, t *testing.T, idleTimeout time.
 	lisNonDefault := testutils.NewListenerWrapper(t, nil)
 
 	// Start a management server to act as the default authority.
-	defaultAuthorityServer, err := e2e.StartManagementServer(&e2e.ManagementServerOptions{Listener: lisDefault})
+	defaultAuthorityServer, err := e2e.StartManagementServer(e2e.ManagementServerOptions{Listener: lisDefault})
 	if err != nil {
 		t.Fatalf("Failed to spin up the xDS management server: %v", err)
 	}
 	t.Cleanup(func() { defaultAuthorityServer.Stop() })
 
 	// Start a management server to act as the non-default authority.
-	nonDefaultAuthorityServer, err := e2e.StartManagementServer(&e2e.ManagementServerOptions{Listener: lisNonDefault})
+	nonDefaultAuthorityServer, err := e2e.StartManagementServer(e2e.ManagementServerOptions{Listener: lisNonDefault})
 	if err != nil {
 		t.Fatalf("Failed to spin up the xDS management server: %v", err)
 	}