xdsclient: fix TestConcurrentReportLoad to not run for 10s (#8598)

While working on the fix for the xDS client unsubscribe/resubscribe
race, I noticed that the tests in the `internal/xds/xdsclient/tests/`
directory were taking about a minute to run. Upon inspection I found
that `TestConcurrentReportLoad` was running for the configured test
timeout duration of `10s`, but was not failing.

This PR fixes the test to run in a short duration. It also makes a
couple of other cleanups that I noticed when fixing this test.

RELEASE NOTES: none

---------

Co-authored-by: eshitachandwani <[email protected]>

Author: easwars

internal/xds/clients/grpctransport/grpc_transport.go

@@ -111,7 +111,7 @@ func (b *Builder) Build(si clients.ServerIdentifier) (clients.Transport, error)
 
 	if cc, ok := b.connections[si]; ok {
 		if logger.V(2) {
-			logger.Info("Reusing existing connection to the server for ServerIdentifier: %v", si)
+			logger.Infof("Reusing existing connection to the server for ServerIdentifier: %v", si)
 		}
 		b.refs[si]++
 		tr := &grpcTransport{cc: cc}
@@ -148,7 +148,7 @@ func (b *Builder) Build(si clients.ServerIdentifier) (clients.Transport, error)
 	b.refs[si] = 1
 
 	if logger.V(2) {
-		logger.Info("Created a new transport to the server for ServerIdentifier: %v", si)
+		logger.Infof("Created a new transport to the server for ServerIdentifier: %v", si)
 	}
 	return tr, nil
 }

internal/xds/clients/lrsclient/lrsclient.go

@@ -28,7 +28,6 @@ import (
 	"sync"
 	"time"
 
-	"google.golang.org/grpc/grpclog"
 	igrpclog "google.golang.org/grpc/internal/grpclog"
 	"google.golang.org/grpc/internal/xds/clients"
 	clientsinternal "google.golang.org/grpc/internal/xds/clients/internal"
@@ -114,10 +113,6 @@ func (c *LRSClient) getOrCreateLRSStream(serverIdentifier clients.ServerIdentifi
 		c.logger.Infof("Creating a new lrs stream for server identifier %q", serverIdentifier)
 	}
 
-	l := grpclog.Component("xds")
-	logPrefix := clientPrefix(c)
-	c.logger = igrpclog.NewPrefixLogger(l, logPrefix)
-
 	// Create a new transport and create a new lrs stream, and add it to the
 	// map of lrs streams.
 	tr, err := c.transportBuilder.Build(serverIdentifier)
@@ -131,7 +126,7 @@ func (c *LRSClient) getOrCreateLRSStream(serverIdentifier clients.ServerIdentifi
 		transport: tr,
 		backoff:   c.backoff,
 		nodeProto: nodeProto,
-		logPrefix: logPrefix,
+		logPrefix: clientPrefix(c),
 	})
 
 	// Register a stop function that decrements the reference count, stops

internal/xds/xdsclient/tests/loadreport_test.go

@@ -448,11 +448,10 @@ func (s) TestReportLoad_StreamCreation(t *testing.T) {
 
 // TestConcurrentReportLoad verifies that the client can safely handle concurrent
 // requests to initiate load reporting streams. It launches multiple goroutines
-// that all call client.ReportLoad simultaneously.
+// that all call client.ReportLoad simultaneously. We don't verify anything on
+// the server here, since that is done in other tests. This is just to ensure
+// that concurrent ReportLoad() calls work.
 func (s) TestConcurrentReportLoad(t *testing.T) {
-	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
-	defer cancel()
-
 	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{SupportLoadReportingService: true})
 	nodeID := uuid.New().String()
 	bc := e2e.DefaultBootstrapContents(t, nodeID, mgmtServer.Address)
@@ -464,17 +463,29 @@ func (s) TestConcurrentReportLoad(t *testing.T) {
 	}
 
 	// Call ReportLoad() concurrently from multiple go routines.
-	var wg sync.WaitGroup
 	const numGoroutines = 10
+	cancelStores := make([]func(context.Context), numGoroutines)
+	var wg sync.WaitGroup
 	wg.Add(numGoroutines)
-	for range numGoroutines {
-		go func() {
+	for i := range numGoroutines {
+		go func(idx int) {
 			defer wg.Done()
-			_, cancelStore := client.ReportLoad(serverConfig)
-			defer cancelStore(ctx)
-		}()
+			_, cancelStores[idx] = client.ReportLoad(serverConfig)
+		}(i)
 	}
 	wg.Wait()
+
+	// Cancel all the load reporting streams. The last call to cancel is
+	// expected to block until a final load report with pending loads is sent.
+	// But the stream is currently blocked on a recv() call waiting for the LRS
+	// server to send the initial laod reporting response, which it never does.
+	// Hence we use a context with short timeout here.
+	for i, cancel := range cancelStores {
+		sCtx, sCancel := context.WithTimeout(context.Background(), defaultTestShortTimeout)
+		defer sCancel()
+		cancel(sCtx)
+		t.Logf("Cancelled LRS stream %d", i)
+	}
 }
 
 // TestConcurrentChannels verifies that we can create multiple gRPC channels