review comments pass #1

internal/testutils/xds/e2e/server.go

@@ -25,6 +25,7 @@ import (
 	"net"
 	"reflect"
 	"strconv"
+	"testing"
 
 	"github.com/envoyproxy/go-control-plane/pkg/cache/types"
 	"google.golang.org/grpc"
@@ -120,10 +121,10 @@ type ManagementServerOptions struct {
 // StartManagementServer initializes a management server which implements the
 // AggregatedDiscoveryService endpoint. The management server is initialized
 // with no resources. Tests should call the Update() method to change the
-// 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) {
+// resource snapshot held by the management server, as per by the test logic.
+//
+// Registers a cleanup function on t to stop the management server.
+func StartManagementServer(t *testing.T, opts ManagementServerOptions) *ManagementServer {
 	// 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.
@@ -136,7 +137,7 @@ func StartManagementServer(opts ManagementServerOptions) (*ManagementServer, err
 		var err error
 		lis, err = net.Listen("tcp", "localhost:0")
 		if err != nil {
-			return nil, fmt.Errorf("listening on local host and port: %v", err)
+			t.Fatalf("Failed to listen on localhost:0: %v", err)
 		}
 	}
 
@@ -175,8 +176,8 @@ func StartManagementServer(opts ManagementServerOptions) (*ManagementServer, err
 	// Start serving.
 	go gs.Serve(lis)
 	logger.Infof("xDS management server serving at: %v...", lis.Addr().String())
-
-	return mgmtServer, nil
+	t.Cleanup(mgmtServer.Stop)
+	return mgmtServer
 }
 
 // UpdateOptions wraps parameters to be passed to the Update() method.

internal/testutils/xds/e2e/setup_management_server.go

@@ -25,72 +25,27 @@ import (
 	"path"
 	"testing"
 
-	"github.com/google/uuid"
-	"google.golang.org/grpc/internal"
 	"google.golang.org/grpc/internal/xds/bootstrap"
-	"google.golang.org/grpc/resolver"
 	"google.golang.org/grpc/testdata"
 )
 
-// SetupManagementServer performs the following:
-// - spin up an xDS management server on a local port
-// - set up certificates for consumption by the file_watcher plugin
-// - creates a bootstrap file in a temporary location
-// - creates an xDS resolver using the above bootstrap contents
-//
-// Returns the following:
-// - management server
-// - nodeID to be used by the client when connecting to the management server
-// - 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()) {
-	t.Helper()
-
-	// Spin up an xDS management server on a local port.
-	server, err := StartManagementServer(opts)
-	if err != nil {
-		t.Fatalf("Failed to spin up the xDS management server: %v", err)
-	}
-	defer func() {
-		if err != nil {
-			server.Stop()
-		}
-	}()
-
-	nodeID := uuid.New().String()
-	bootstrapContents, err := DefaultBootstrapContents(nodeID, fmt.Sprintf("passthrough:///%s", server.Address))
-	if err != nil {
-		server.Stop()
-		t.Fatal(err)
-	}
-	var rb resolver.Builder
-	if newResolver := internal.NewXDSResolverWithConfigForTesting; newResolver != nil {
-		rb, err = newResolver.(func([]byte) (resolver.Builder, error))(bootstrapContents)
-		if err != nil {
-			server.Stop()
-			t.Fatalf("Failed to create xDS resolver for testing: %v", err)
-		}
-	}
-
-	return server, nodeID, bootstrapContents, rb, func() { server.Stop() }
-}
-
 // DefaultBootstrapContents creates a default bootstrap configuration with the
 // given node ID and server URI. It also creates certificate provider
 // configuration and sets the listener resource name template to be used on the
 // server side.
-func DefaultBootstrapContents(nodeID, serverURI string) ([]byte, error) {
+func DefaultBootstrapContents(t *testing.T, nodeID, serverURI string) []byte {
+	t.Helper()
+
 	// Create a directory to hold certs and key files used on the server side.
-	serverDir, err := createTmpDirWithFiles("testServerSideXDS*", "x509/server1_cert.pem", "x509/server1_key.pem", "x509/client_ca_cert.pem")
+	serverDir, err := CreateTmpDirWithFiles("testServerSideXDS*", "x509/server1_cert.pem", "x509/server1_key.pem", "x509/client_ca_cert.pem")
 	if err != nil {
-		return nil, fmt.Errorf("failed to create bootstrap configuration: %v", err)
+		t.Fatalf("Failed to create bootstrap configuration: %v", err)
 	}
 
 	// Create a directory to hold certs and key files used on the client side.
-	clientDir, err := createTmpDirWithFiles("testClientSideXDS*", "x509/client1_cert.pem", "x509/client1_key.pem", "x509/server_ca_cert.pem")
+	clientDir, err := CreateTmpDirWithFiles("testClientSideXDS*", "x509/client1_cert.pem", "x509/client1_key.pem", "x509/server_ca_cert.pem")
 	if err != nil {
-		return nil, fmt.Errorf("failed to create bootstrap configuration: %v", err)
+		t.Fatalf("Failed to create bootstrap configuration: %v", err)
 	}
 
 	// Create certificate providers section of the bootstrap config with entries
@@ -103,27 +58,17 @@ func DefaultBootstrapContents(nodeID, serverURI string) ([]byte, error) {
 	// Create the bootstrap configuration.
 	bs, err := bootstrap.NewContentsForTesting(bootstrap.ConfigOptionsForTesting{
 		Servers: []json.RawMessage{[]byte(fmt.Sprintf(`{
-			"server_uri": %q,
+			"server_uri": "passthrough:///%s",
 			"channel_creds": [{"type": "insecure"}]
 		}`, serverURI))},
 		NodeID:                             nodeID,
 		CertificateProviders:               cpc,
 		ServerListenerResourceNameTemplate: ServerListenerResourceNameTemplate,
-		Authorities: map[string]json.RawMessage{
-			// Most tests that use new style xdstp resource names do not specify
-			// an authority. These end up looking up an entry with the empty key
-			// in the authorities map. Having an entry with an empty key and
-			// empty configuration, results in these resources also using the
-			// top-level configuration, which is what we want mostly for our
-			// tests, unless explicitly specified by tests that use multiple
-			// authorities etc.
-			"": []byte(`{}`),
-		},
 	})
 	if err != nil {
-		return nil, fmt.Errorf("failed to create bootstrap configuration: %v", err)
+		t.Fatalf("Failed to create bootstrap configuration: %v", err)
 	}
-	return bs, nil
+	return bs
 }
 
 const (
@@ -144,11 +89,11 @@ func createTmpFile(src, dst string) error {
 	return nil
 }
 
-// createTempDirWithFiles creates a temporary directory under the system default
+// CreateTmpDirWithFiles creates a temporary directory under the system default
 // tempDir with the given dirSuffix. It also reads from certSrc, keySrc and
 // rootSrc files are creates appropriate files under the newly create tempDir.
 // Returns the name of the created tempDir.
-func createTmpDirWithFiles(dirSuffix, certSrc, keySrc, rootSrc string) (string, error) {
+func CreateTmpDirWithFiles(dirSuffix, certSrc, keySrc, rootSrc string) (string, error) {
 	// Create a temp directory. Passing an empty string for the first argument
 	// uses the system temp directory.
 	dir, err := os.MkdirTemp("", dirSuffix)

test/xds/xds_client_ack_nack_test.go

@@ -24,12 +24,15 @@ import (
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
+	"github.com/google/uuid"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/internal"
 	"google.golang.org/grpc/internal/grpcsync"
 	"google.golang.org/grpc/internal/stubserver"
 	"google.golang.org/grpc/internal/testutils"
 	"google.golang.org/grpc/internal/testutils/xds/e2e"
+	"google.golang.org/grpc/resolver"
 
 	v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	v3discoverypb "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
@@ -83,7 +86,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 := e2e.StartManagementServer(t, e2e.ManagementServerOptions{
 		Listener: lis,
 		OnStreamRequest: func(id int64, req *v3discoverypb.DiscoveryRequest) error {
 			// Return early under the following circumstances:
@@ -122,7 +125,19 @@ func (s) TestClientResourceVersionAfterStreamRestart(t *testing.T) {
 			streamRestarted.Fire()
 		},
 	})
-	defer cleanup1()
+
+	// Create bootstrap configuration pointing to the above management server.
+	nodeID := uuid.New().String()
+	bootstrapContents := e2e.DefaultBootstrapContents(t, nodeID, managementServer.Address)
+
+	// Create an xDS resolver with the above bootstrap configuration.
+	var xdsResolver resolver.Builder
+	if newResolver := internal.NewXDSResolverWithConfigForTesting; newResolver != nil {
+		xdsResolver, err = newResolver.(func([]byte) (resolver.Builder, error))(bootstrapContents)
+		if err != nil {
+			t.Fatalf("Failed to create xDS resolver for testing: %v", err)
+		}
+	}
 
 	server := stubserver.StartTestService(t, nil)
 	defer server.Stop()
@@ -142,7 +157,7 @@ func (s) TestClientResourceVersionAfterStreamRestart(t *testing.T) {
 	}
 
 	// Create a ClientConn and make a successful RPC.
-	cc, err := grpc.NewClient(fmt.Sprintf("xds:///%s", serviceName), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithResolvers(resolver))
+	cc, err := grpc.NewClient(fmt.Sprintf("xds:///%s", serviceName), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithResolvers(xdsResolver))
 	if err != nil {
 		t.Fatalf("failed to dial local test server: %v", err)
 	}

test/xds/xds_client_affinity_test.go

@@ -83,8 +83,7 @@ func ringhashCluster(clusterName, edsServiceName string) *v3clusterpb.Cluster {
 // propagated to pick the ring_hash policy. It doesn't test the affinity
 // behavior in ring_hash policy.
 func (s) TestClientSideAffinitySanityCheck(t *testing.T) {
-	managementServer, nodeID, _, resolver, cleanup1 := e2e.SetupManagementServer(t, e2e.ManagementServerOptions{})
-	defer cleanup1()
+	managementServer, nodeID, _, xdsResolver := setupManagementServerAndResolver(t)
 
 	server := stubserver.StartTestService(t, nil)
 	defer server.Stop()
@@ -115,7 +114,7 @@ func (s) TestClientSideAffinitySanityCheck(t *testing.T) {
 	}
 
 	// Create a ClientConn and make a successful RPC.
-	cc, err := grpc.NewClient(fmt.Sprintf("xds:///%s", serviceName), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithResolvers(resolver))
+	cc, err := grpc.NewClient(fmt.Sprintf("xds:///%s", serviceName), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithResolvers(xdsResolver))
 	if err != nil {
 		t.Fatalf("failed to dial local test server: %v", err)
 	}

test/xds/xds_client_certificate_providers_test.go

@@ -60,34 +60,7 @@ import (
 // used on the client.
 func (s) TestClientSideXDS_WithNoCertificateProvidersInBootstrap_Success(t *testing.T) {
 	// Spin up an xDS management server.
-	mgmtServer, err := e2e.StartManagementServer(e2e.ManagementServerOptions{})
-	if err != nil {
-		t.Fatalf("Failed to start management server: %v", err)
-	}
-	defer mgmtServer.Stop()
-
-	// Create bootstrap configuration with no certificate providers.
-	nodeID := uuid.New().String()
-	bs, err := bootstrap.NewContentsForTesting(bootstrap.ConfigOptionsForTesting{
-		Servers: []json.RawMessage{[]byte(fmt.Sprintf(`{
-			"server_uri": %q,
-			"channel_creds": [{"type": "insecure"}]
-		}`, mgmtServer.Address))},
-		NodeID: nodeID,
-	})
-	if err != nil {
-		t.Fatalf("Failed to create bootstrap configuration: %v", err)
-	}
-
-	// Create an xDS resolver with the above bootstrap configuration.
-	newResolver := internal.NewXDSResolverWithConfigForTesting
-	if newResolver == nil {
-		t.Fatal("internal.NewXDSResolverWithConfigForTesting is unset")
-	}
-	resolverBuilder, err := newResolver.(func([]byte) (resolver.Builder, error))(bs)
-	if err != nil {
-		t.Fatalf("Failed to create xDS resolver for testing: %v", err)
-	}
+	mgmtServer, nodeID, _, resolverBuilder := setupManagementServerAndResolver(t)
 
 	// Spin up a test backend.
 	server := stubserver.StartTestService(t, nil)
@@ -138,34 +111,29 @@ func (s) TestClientSideXDS_WithNoCertificateProvidersInBootstrap_Success(t *test
 // channel creation does not fail, but it moves to TRANSIENT_FAILURE and
 // subsequent rpcs fail.
 func (s) TestClientSideXDS_WithNoCertificateProvidersInBootstrap_Failure(t *testing.T) {
-	// Spin up an xDS management server.
-	mgmtServer, err := e2e.StartManagementServer(e2e.ManagementServerOptions{})
-	if err != nil {
-		t.Fatalf("Failed to start management server: %v", err)
-	}
-	defer mgmtServer.Stop()
+	// Start an xDS management server.
+	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{AllowResourceSubset: true})
 
-	// Create bootstrap configuration with no certificate providers.
+	// Create bootstrap configuration pointing to the above management server.
 	nodeID := uuid.New().String()
-	bs, err := bootstrap.NewContentsForTesting(bootstrap.ConfigOptionsForTesting{
+	bc, err := bootstrap.NewContentsForTesting(bootstrap.ConfigOptionsForTesting{
 		Servers: []json.RawMessage{[]byte(fmt.Sprintf(`{
-			"server_uri": %q,
-			"channel_creds": [{"type": "insecure"}]
-		}`, mgmtServer.Address))},
+					"server_uri": %q,
+					"channel_creds": [{"type": "insecure"}]
+				}`, mgmtServer.Address))},
 		NodeID: nodeID,
 	})
 	if err != nil {
 		t.Fatalf("Failed to create bootstrap configuration: %v", err)
 	}
 
 	// Create an xDS resolver with the above bootstrap configuration.
-	newResolver := internal.NewXDSResolverWithConfigForTesting
-	if newResolver == nil {
-		t.Fatal("internal.NewXDSResolverWithConfigForTesting is unset")
-	}
-	resolverBuilder, err := newResolver.(func([]byte) (resolver.Builder, error))(bs)
-	if err != nil {
-		t.Fatalf("Failed to create xDS resolver for testing: %v", err)
+	var resolverBuilder resolver.Builder
+	if newResolver := internal.NewXDSResolverWithConfigForTesting; newResolver != nil {
+		resolverBuilder, err = newResolver.(func([]byte) (resolver.Builder, error))(bc)
+		if err != nil {
+			t.Fatalf("Failed to create xDS resolver for testing: %v", err)
+		}
 	}
 
 	// Spin up a test backend.
@@ -225,17 +193,29 @@ func (s) TestClientSideXDS_WithNoCertificateProvidersInBootstrap_Failure(t *test
 // The test verifies that RPCs to the first two clusters succeed, while RPCs to
 // the third cluster fails with an appropriate code and error message.
 func (s) TestClientSideXDS_WithValidAndInvalidSecurityConfiguration(t *testing.T) {
-	// Spin up an xDS management server. This uses a bootstrap config with a
-	// certificate provider instance name e2e.ClientSideCertProviderInstance.
-	mgmtServer, nodeID, _, resolver, cleanup := e2e.SetupManagementServer(t, e2e.ManagementServerOptions{AllowResourceSubset: true})
-	defer cleanup()
+	// Spin up an xDS management server.
+	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{AllowResourceSubset: true})
+
+	// Create bootstrap configuration pointing to the above management server.
+	nodeID := uuid.New().String()
+	bc := e2e.DefaultBootstrapContents(t, nodeID, mgmtServer.Address)
+
+	// Create an xDS resolver with the above bootstrap configuration.
+	var xdsResolver resolver.Builder
+	if newResolver := internal.NewXDSResolverWithConfigForTesting; newResolver != nil {
+		var err error
+		xdsResolver, err = newResolver.(func([]byte) (resolver.Builder, error))(bc)
+		if err != nil {
+			t.Fatalf("Failed to create xDS resolver for testing: %v", err)
+		}
+	}
 
 	// Create test backends for all three clusters
 	// backend1 configured with TLS creds, represents cluster1
 	// backend2 configured with insecure creds, represents cluster2
 	// backend3 configured with insecure creds, represents cluster3
-	creds := testutils.CreateServerTLSCredentials(t, tls.RequireAndVerifyClientCert)
-	server1 := stubserver.StartTestService(t, nil, grpc.Creds(creds))
+	serverCreds := testutils.CreateServerTLSCredentials(t, tls.RequireAndVerifyClientCert)
+	server1 := stubserver.StartTestService(t, nil, grpc.Creds(serverCreds))
 	defer server1.Stop()
 	server2 := stubserver.StartTestService(t, nil)
 	defer server2.Stop()
@@ -331,13 +311,13 @@ func (s) TestClientSideXDS_WithValidAndInvalidSecurityConfiguration(t *testing.T
 	}
 
 	// Create client-side xDS credentials with an insecure fallback.
-	creds, err := xdscreds.NewClientCredentials(xdscreds.ClientOptions{FallbackCreds: insecure.NewCredentials()})
+	clientCreds, err := xdscreds.NewClientCredentials(xdscreds.ClientOptions{FallbackCreds: insecure.NewCredentials()})
 	if err != nil {
 		t.Fatal(err)
 	}
 
 	// Create a ClientConn.
-	cc, err := grpc.NewClient(fmt.Sprintf("xds:///%s", serviceName), grpc.WithTransportCredentials(creds), grpc.WithResolvers(resolver))
+	cc, err := grpc.NewClient(fmt.Sprintf("xds:///%s", serviceName), grpc.WithTransportCredentials(clientCreds), grpc.WithResolvers(xdsResolver))
 	if err != nil {
 		t.Fatalf("failed to dial local test server: %v", err)
 	}

test/xds/xds_client_custom_lb_test.go

@@ -221,8 +221,9 @@ func (s) TestWrrLocality(t *testing.T) {
 	}
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			managementServer, nodeID, _, r, cleanup := e2e.SetupManagementServer(t, e2e.ManagementServerOptions{})
-			defer cleanup()
+			// Start an xDS management server.
+			managementServer, nodeID, _, xdsResolver := setupManagementServerAndResolver(t)
+
 			routeConfigName := "route-" + serviceName
 			clusterName := "cluster-" + serviceName
 			endpointsName := "endpoints-" + serviceName
@@ -253,7 +254,7 @@ func (s) TestWrrLocality(t *testing.T) {
 				t.Fatal(err)
 			}
 
-			cc, err := grpc.NewClient(fmt.Sprintf("xds:///%s", serviceName), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithResolvers(r))
+			cc, err := grpc.NewClient(fmt.Sprintf("xds:///%s", serviceName), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithResolvers(xdsResolver))
 			if err != nil {
 				t.Fatalf("Failed to dial local test server: %v", err)
 			}