Function StartManagementServer

internal/xds/clients/internal/testutils/e2e/server.go:133–190

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 per by the test logic. Regi

(t *testing.T, opts ManagementServerOptions)

//
// Registers a cleanup function on t to stop the management server.
func StartManagementServer(t *testing.T, opts ManagementServerOptions) *ManagementServer {
	t.Helper()

	// 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.AllowResourceSubset
	cache := v3cache.NewSnapshotCache(wait, v3cache.IDHash{}, serverLogger{t})
	t.Logf("Created new snapshot cache...")

	lis := opts.Listener
	if lis == nil {
		var err error
		lis, err = net.Listen("tcp", "localhost:0")
		if err != nil {
			t.Fatalf("Failed to listen on localhost:0: %v", err)
		}
	}

	// 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{
		StreamOpenFunc:     opts.OnStreamOpen,
		StreamClosedFunc:   opts.OnStreamClosed,
		StreamRequestFunc:  opts.OnStreamRequest,
		StreamResponseFunc: opts.OnStreamResponse,
	}

	// Create an xDS management server and register the ADS implementation
	// provided by it on a gRPC server.
	xs := v3server.NewServer(ctx, cache, callbacks)
	gs := grpc.NewServer()
	v3discoverygrpc.RegisterAggregatedDiscoveryServiceServer(gs, xs)
	t.Logf("Registered Aggregated Discovery Service (ADS)...")

	mgmtServer := &ManagementServer{
		Address: lis.Addr().String(),
		cancel:  cancel,
		version: 0,
		gs:      gs,
		xs:      xs,
		cache:   cache,
		logger:  t,
	}
	if opts.SupportLoadReportingService {
		lrs := fakeserver.NewServer(lis.Addr().String())
		v3lrsgrpc.RegisterLoadReportingServiceServer(gs, lrs)
		mgmtServer.LRSServer = lrs
		t.Logf("Registered Load Reporting Service (LRS)...")
	}

	// Start serving.
	go gs.Serve(lis)
	t.Logf("xDS management server serving at: %v...", lis.Addr().String())
	t.Cleanup(mgmtServer.Stop)
	return mgmtServer
}