xds: make e2e tests use a single management server instance (#4399)

xds/internal/test/xds_client_integration_test.go

@@ -23,48 +23,26 @@ package xds_test
 
 import (
 	"context"
+	"fmt"
 	"net"
 	"testing"
 
-	"github.com/google/uuid"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
-	"google.golang.org/grpc/internal/xds"
 	"google.golang.org/grpc/xds/internal/testutils"
 	"google.golang.org/grpc/xds/internal/testutils/e2e"
 
 	testpb "google.golang.org/grpc/test/grpc_testing"
 )
 
 // clientSetup performs a bunch of steps common to all xDS client tests here:
-// - spin up an xDS management server on a local port
 // - spin up a gRPC server and register the test service on it
 // - create a local TCP listener and start serving on it
 //
 // Returns the following:
-// - the management server: tests use this to configure resources
-// - nodeID expected by the management server: this is set in the Node proto
-//   sent by the xdsClient for queries.
 // - the port the server is listening on
 // - cleanup function to be invoked by the tests when done
-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()
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	// Create a bootstrap file in a temporary directory.
-	bootstrapCleanup, err := xds.SetupBootstrapFile(xds.BootstrapOptions{
-		Version:   xds.TransportV3,
-		NodeID:    nodeID,
-		ServerURI: fs.Address,
-	})
-	if err != nil {
-		t.Fatal(err)
-	}
-
+func clientSetup(t *testing.T) (uint32, func()) {
 	// Initialize a gRPC server and register the stubServer on it.
 	server := grpc.NewServer()
 	testpb.RegisterTestServiceServer(server, &testService{})
@@ -81,30 +59,29 @@ func clientSetup(t *testing.T) (*e2e.ManagementServer, string, uint32, func()) {
 		}
 	}()
 
-	return fs, nodeID, uint32(lis.Addr().(*net.TCPAddr).Port), func() {
-		fs.Stop()
-		bootstrapCleanup()
+	return uint32(lis.Addr().(*net.TCPAddr).Port), func() {
 		server.Stop()
 	}
 }
 
 func (s) TestClientSideXDS(t *testing.T) {
-	fs, nodeID, port, cleanup := clientSetup(t)
+	port, cleanup := clientSetup(t)
 	defer cleanup()
 
+	serviceName := xdsServiceName + "-client-side-xds"
 	resources := e2e.DefaultClientResources(e2e.ResourceParams{
-		DialTarget: "myservice",
-		NodeID:     nodeID,
+		DialTarget: serviceName,
+		NodeID:     xdsClientNodeID,
 		Host:       "localhost",
 		Port:       port,
 		SecLevel:   e2e.SecurityLevelNone,
 	})
-	if err := fs.Update(resources); err != nil {
+	if err := managementServer.Update(resources); err != nil {
 		t.Fatal(err)
 	}
 
 	// Create a ClientConn and make a successful RPC.
-	cc, err := grpc.Dial("xds:///myservice", grpc.WithTransportCredentials(insecure.NewCredentials()))
+	cc, err := grpc.Dial(fmt.Sprintf("xds:///%s", serviceName), grpc.WithTransportCredentials(insecure.NewCredentials()))
 	if err != nil {
 		t.Fatalf("failed to dial local test server: %v", err)
 	}

xds/internal/test/xds_integration_test.go

@@ -24,10 +24,26 @@ package xds_test
 
 import (
 	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"log"
+	"os"
+	"path"
 	"testing"
 	"time"
 
+	"github.com/google/uuid"
+	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/internal/grpctest"
+	"google.golang.org/grpc/internal/leakcheck"
+	"google.golang.org/grpc/internal/xds/env"
+	"google.golang.org/grpc/testdata"
+	"google.golang.org/grpc/xds/internal/testutils/e2e"
+
+	xdsinternal "google.golang.org/grpc/internal/xds"
 	testpb "google.golang.org/grpc/test/grpc_testing"
 )
 
@@ -51,3 +67,148 @@ type testService struct {
 func (*testService) EmptyCall(context.Context, *testpb.Empty) (*testpb.Empty, error) {
 	return &testpb.Empty{}, nil
 }
+
+var (
+	// Globals corresponding to the single instance of the xDS management server
+	// which is spawned for all the tests in this package.
+	managementServer *e2e.ManagementServer
+	xdsClientNodeID  string
+)
+
+// TestMain sets up an xDS management server, runs all tests, and stops the
+// management server.
+func TestMain(m *testing.M) {
+	// The management server is started and stopped from here, but the leakcheck
+	// runs after every individual test. So, we need to skip the goroutine which
+	// spawns the management server and is blocked on the call to `Serve()`.
+	leakcheck.RegisterIgnoreGoroutine("e2e.StartManagementServer")
+
+	cancel, err := setupManagementServer()
+	if err != nil {
+		log.Printf("setupManagementServer() failed: %v", err)
+		os.Exit(1)
+	}
+
+	code := m.Run()
+	cancel()
+	os.Exit(code)
+}
+
+func createTmpFile(src, dst string) error {
+	data, err := ioutil.ReadFile(src)
+	if err != nil {
+		return fmt.Errorf("ioutil.ReadFile(%q) failed: %v", src, err)
+	}
+	if err := ioutil.WriteFile(dst, data, os.ModePerm); err != nil {
+		return fmt.Errorf("ioutil.WriteFile(%q) failed: %v", dst, err)
+	}
+	return nil
+}
+
+// createTempDirWithFiles 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) {
+	// Create a temp directory. Passing an empty string for the first argument
+	// uses the system temp directory.
+	dir, err := ioutil.TempDir("", dirSuffix)
+	if err != nil {
+		return "", fmt.Errorf("ioutil.TempDir() failed: %v", err)
+	}
+
+	if err := createTmpFile(testdata.Path(certSrc), path.Join(dir, certFile)); err != nil {
+		return "", err
+	}
+	if err := createTmpFile(testdata.Path(keySrc), path.Join(dir, keyFile)); err != nil {
+		return "", err
+	}
+	if err := createTmpFile(testdata.Path(rootSrc), path.Join(dir, rootFile)); err != nil {
+		return "", err
+	}
+	return dir, nil
+}
+
+// createClientTLSCredentials creates client-side TLS transport credentials.
+func createClientTLSCredentials(t *testing.T) credentials.TransportCredentials {
+	t.Helper()
+
+	cert, err := tls.LoadX509KeyPair(testdata.Path("x509/client1_cert.pem"), testdata.Path("x509/client1_key.pem"))
+	if err != nil {
+		t.Fatalf("tls.LoadX509KeyPair(x509/client1_cert.pem, x509/client1_key.pem) failed: %v", err)
+	}
+	b, err := ioutil.ReadFile(testdata.Path("x509/server_ca_cert.pem"))
+	if err != nil {
+		t.Fatalf("ioutil.ReadFile(x509/server_ca_cert.pem) failed: %v", err)
+	}
+	roots := x509.NewCertPool()
+	if !roots.AppendCertsFromPEM(b) {
+		t.Fatal("failed to append certificates")
+	}
+	return credentials.NewTLS(&tls.Config{
+		Certificates: []tls.Certificate{cert},
+		RootCAs:      roots,
+		ServerName:   "x.test.example.com",
+	})
+}
+
+// setupManagement server performs the following:
+// - spin up an xDS management server on a local port
+// - set up certificates for consumption by the file_watcher plugin
+// - sets up the global variables which refer to this management server and the
+//   nodeID to be used when talking to this management server.
+//
+// Returns a function to be invoked by the caller to stop the management server.
+func setupManagementServer() (func(), error) {
+	// Turn on the env var protection for client-side security.
+	origClientSideSecurityEnvVar := env.ClientSideSecuritySupport
+	env.ClientSideSecuritySupport = true
+
+	// Spin up an xDS management server on a local port.
+	var err error
+	managementServer, err = e2e.StartManagementServer()
+	if err != nil {
+		return nil, err
+	}
+
+	// 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")
+	if err != nil {
+		managementServer.Stop()
+		return nil, 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")
+	if err != nil {
+		managementServer.Stop()
+		return nil, err
+	}
+
+	// Create certificate providers section of the bootstrap config with entries
+	// for both the client and server sides.
+	cpc := map[string]json.RawMessage{
+		e2e.ServerSideCertProviderInstance: e2e.DefaultFileWatcherConfig(path.Join(serverDir, certFile), path.Join(serverDir, keyFile), path.Join(serverDir, rootFile)),
+		e2e.ClientSideCertProviderInstance: e2e.DefaultFileWatcherConfig(path.Join(clientDir, certFile), path.Join(clientDir, keyFile), path.Join(clientDir, rootFile)),
+	}
+
+	// Create a bootstrap file in a temporary directory.
+	xdsClientNodeID = uuid.New().String()
+	bootstrapCleanup, err := xdsinternal.SetupBootstrapFile(xdsinternal.BootstrapOptions{
+		Version:                            xdsinternal.TransportV3,
+		NodeID:                             xdsClientNodeID,
+		ServerURI:                          managementServer.Address,
+		CertificateProviders:               cpc,
+		ServerListenerResourceNameTemplate: e2e.ServerListenerResourceNameTemplate,
+	})
+	if err != nil {
+		managementServer.Stop()
+		return nil, err
+	}
+
+	return func() {
+		managementServer.Stop()
+		bootstrapCleanup()
+		env.ClientSideSecuritySupport = origClientSideSecurityEnvVar
+	}, nil
+}