resolving comments

Author: Pranjali-2501

internal/xds/balancer/clusterimpl/balancer_test.go

@@ -22,9 +22,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
-	"fmt"
 	"strings"
-	"sync"
 	"testing"
 	"time"
 
@@ -37,15 +35,10 @@ import (
 	internalserviceconfig "google.golang.org/grpc/internal/serviceconfig"
 	"google.golang.org/grpc/internal/testutils"
 	xdsinternal "google.golang.org/grpc/internal/xds"
-	"google.golang.org/grpc/internal/xds/bootstrap"
-	"google.golang.org/grpc/internal/xds/clients"
 	"google.golang.org/grpc/internal/xds/testutils/fakeclient"
 	"google.golang.org/grpc/internal/xds/xdsclient"
 	"google.golang.org/grpc/resolver"
 	"google.golang.org/grpc/serviceconfig"
-
-	"github.com/google/go-cmp/cmp"
-	"github.com/google/go-cmp/cmp/cmpopts"
 )
 
 const (
@@ -58,7 +51,6 @@ const (
 
 var (
 	testBackendEndpoints = []resolver.Endpoint{{Addresses: []resolver.Address{{Addr: "1.1.1.1:1"}}}}
-	toleranceCmpOpt      = cmp.Options{cmpopts.EquateApprox(0, 1e-5), cmp.AllowUnexported(loadData{}, localityData{}, requestData{}, serverLoadData{})}
 )
 
 type s struct {
@@ -69,205 +61,6 @@ func Test(t *testing.T) {
 	grpctest.RunSubTests(t, s{})
 }
 
-// testLoadReporter records load data pertaining to a single cluster.
-//
-// It implements loadReporter interface for the picker. Tests can use it to
-// override the loadStore in the picker to verify load reporting.
-type testLoadReporter struct {
-	cluster, service string
-
-	mu               sync.Mutex
-	drops            map[string]uint64
-	localityRPCCount map[clients.Locality]*rpcCountData
-}
-
-// CallStarted records a call started for the clients.Locality.
-func (lr *testLoadReporter) CallStarted(locality clients.Locality) {
-	lr.mu.Lock()
-	defer lr.mu.Unlock()
-	if _, ok := lr.localityRPCCount[locality]; !ok {
-		lr.localityRPCCount[locality] = &rpcCountData{}
-	}
-	lr.localityRPCCount[locality].inProgress++
-	lr.localityRPCCount[locality].issued++
-}
-
-// CallFinished records a call finished for the clients.Locality.
-func (lr *testLoadReporter) CallFinished(locality clients.Locality, err error) {
-	lr.mu.Lock()
-	defer lr.mu.Unlock()
-	if lr.localityRPCCount == nil {
-		return
-	}
-	lrc := lr.localityRPCCount[locality]
-	lrc.inProgress--
-	if err == nil {
-		lrc.succeeded++
-	} else {
-		lrc.errored++
-	}
-}
-
-// CallServerLoad records a server load for the clients.Locality.
-func (lr *testLoadReporter) CallServerLoad(locality clients.Locality, name string, val float64) {
-	lr.mu.Lock()
-	defer lr.mu.Unlock()
-	if lr.localityRPCCount == nil {
-		return
-	}
-	lrc, ok := lr.localityRPCCount[locality]
-	if !ok {
-		return
-	}
-	if lrc.serverLoads == nil {
-		lrc.serverLoads = make(map[string]*rpcLoadData)
-	}
-	if _, ok := lrc.serverLoads[name]; !ok {
-		lrc.serverLoads[name] = &rpcLoadData{}
-	}
-	rld := lrc.serverLoads[name]
-	rld.add(val)
-}
-
-// CallDropped records a call dropped for the category.
-func (lr *testLoadReporter) CallDropped(category string) {
-	lr.mu.Lock()
-	defer lr.mu.Unlock()
-	lr.drops[category]++
-}
-
-// stats returns and resets all loads reported for a cluster and service,
-// except inProgress rpc counts.
-//
-// It returns nil if the store doesn't contain any (new) data.
-func (lr *testLoadReporter) stats() *loadData {
-	lr.mu.Lock()
-	defer lr.mu.Unlock()
-
-	sd := newLoadData(lr.cluster, lr.service)
-	for category, val := range lr.drops {
-		if val == 0 {
-			continue
-		}
-		if category != "" {
-			// Skip drops without category. They are counted in total_drops, but
-			// not in per category. One example is drops by circuit breaking.
-			sd.drops[category] = val
-		}
-		sd.totalDrops += val
-		lr.drops[category] = 0 // clear drops for next report
-	}
-	for locality, countData := range lr.localityRPCCount {
-		if countData.succeeded == 0 && countData.errored == 0 && countData.inProgress == 0 && countData.issued == 0 {
-			continue
-		}
-
-		ld := localityData{
-			requestStats: requestData{
-				succeeded:  countData.succeeded,
-				errored:    countData.errored,
-				inProgress: countData.inProgress,
-				issued:     countData.issued,
-			},
-			loadStats: make(map[string]serverLoadData),
-		}
-		// clear localityRPCCount for next report
-		countData.succeeded = 0
-		countData.errored = 0
-		countData.inProgress = 0
-		countData.issued = 0
-		for key, rld := range countData.serverLoads {
-			s, c := rld.loadAndClear() // get and clear serverLoads for next report
-			if c == 0 {
-				continue
-			}
-			ld.loadStats[key] = serverLoadData{sum: s, count: c}
-		}
-		sd.localityStats[locality] = ld
-	}
-	if sd.totalDrops == 0 && len(sd.drops) == 0 && len(sd.localityStats) == 0 {
-		return nil
-	}
-	return sd
-}
-
-// loadData contains all load data reported to the LoadStore since the most recent
-// call to stats().
-type loadData struct {
-	// cluster is the name of the cluster this data is for.
-	cluster string
-	// service is the name of the EDS service this data is for.
-	service string
-	// totalDrops is the total number of dropped requests.
-	totalDrops uint64
-	// drops is the number of dropped requests per category.
-	drops map[string]uint64
-	// localityStats contains load reports per locality.
-	localityStats map[clients.Locality]localityData
-}
-
-// localityData contains load data for a single locality.
-type localityData struct {
-	// requestStats contains counts of requests made to the locality.
-	requestStats requestData
-	// loadStats contains server load data for requests made to the locality,
-	// indexed by the load type.
-	loadStats map[string]serverLoadData
-}
-
-// requestData contains request counts.
-type requestData struct {
-	// succeeded is the number of succeeded requests.
-	succeeded uint64
-	// errored is the number of requests which ran into errors.
-	errored uint64
-	// inProgress is the number of requests in flight.
-	inProgress uint64
-	// issued is the total number requests that were sent.
-	issued uint64
-}
-
-// serverLoadData contains server load data.
-type serverLoadData struct {
-	// count is the number of load reports.
-	count uint64
-	// sum is the total value of all load reports.
-	sum float64
-}
-
-func newLoadData(cluster, service string) *loadData {
-	return &loadData{
-		cluster:       cluster,
-		service:       service,
-		drops:         make(map[string]uint64),
-		localityStats: make(map[clients.Locality]localityData),
-	}
-}
-
-type rpcCountData struct {
-	succeeded   uint64
-	errored     uint64
-	inProgress  uint64
-	issued      uint64
-	serverLoads map[string]*rpcLoadData
-}
-
-type rpcLoadData struct {
-	sum   float64
-	count uint64
-}
-
-func (rld *rpcLoadData) add(v float64) {
-	rld.sum += v
-	rld.count++
-}
-
-func (rld *rpcLoadData) loadAndClear() (s float64, c uint64) {
-	s, rld.sum = rld.sum, 0
-	c, rld.count = rld.count, 0
-	return s, c
-}
-
 func init() {
 	NewRandomWRR = testutils.NewTestWRR
 }
@@ -419,117 +212,6 @@ func (s) TestClusterNameInAddressAttributes(t *testing.T) {
 	}
 }
 
-// TestUpdateLRSServer covers the cases
-// - the init config specifies "" as the LRS server
-// - config modifies LRS server to a different string
-// - config sets LRS server to nil to stop load reporting
-func (s) TestUpdateLRSServer(t *testing.T) {
-	var testLocality = clients.Locality{
-		Region:  "test-region",
-		Zone:    "test-zone",
-		SubZone: "test-sub-zone",
-	}
-
-	xdsC := fakeclient.NewClient()
-
-	builder := balancer.Get(Name)
-	cc := testutils.NewBalancerClientConn(t)
-	b := builder.Build(cc, balancer.BuildOptions{})
-	defer b.Close()
-
-	endpoints := make([]resolver.Endpoint, len(testBackendEndpoints))
-	for i, e := range testBackendEndpoints {
-		endpoints[i] = xdsinternal.SetLocalityIDInEndpoint(e, testLocality)
-	}
-	testLRSServerConfig, err := bootstrap.ServerConfigForTesting(bootstrap.ServerConfigTestingOptions{
-		URI:          "trafficdirector.googleapis.com:443",
-		ChannelCreds: []bootstrap.ChannelCreds{{Type: "google_default"}},
-	})
-	if err != nil {
-		t.Fatalf("Failed to create LRS server config for testing: %v", err)
-	}
-	if err := b.UpdateClientConnState(balancer.ClientConnState{
-		ResolverState: xdsclient.SetClient(resolver.State{Endpoints: endpoints}, xdsC),
-		BalancerConfig: &LBConfig{
-			Cluster:             testClusterName,
-			EDSServiceName:      testServiceName,
-			LoadReportingServer: testLRSServerConfig,
-			ChildPolicy: &internalserviceconfig.BalancerConfig{
-				Name: roundrobin.Name,
-			},
-		},
-	}); err != nil {
-		t.Fatalf("unexpected error from UpdateClientConnState: %v", err)
-	}
-
-	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
-	defer cancel()
-
-	got, err := xdsC.WaitForReportLoad(ctx)
-	if err != nil {
-		t.Fatalf("xdsClient.ReportLoad failed with error: %v", err)
-	}
-	if got.Server != testLRSServerConfig {
-		t.Fatalf("xdsClient.ReportLoad called with {%q}: want {%q}", got.Server, testLRSServerConfig)
-	}
-
-	testLRSServerConfig2, err := bootstrap.ServerConfigForTesting(bootstrap.ServerConfigTestingOptions{
-		URI:          "trafficdirector-another.googleapis.com:443",
-		ChannelCreds: []bootstrap.ChannelCreds{{Type: "google_default"}},
-	})
-	if err != nil {
-		t.Fatalf("Failed to create LRS server config for testing: %v", err)
-	}
-
-	// Update LRS server to a different name.
-	if err := b.UpdateClientConnState(balancer.ClientConnState{
-		ResolverState: xdsclient.SetClient(resolver.State{Endpoints: endpoints}, xdsC),
-		BalancerConfig: &LBConfig{
-			Cluster:             testClusterName,
-			EDSServiceName:      testServiceName,
-			LoadReportingServer: testLRSServerConfig2,
-			ChildPolicy: &internalserviceconfig.BalancerConfig{
-				Name: roundrobin.Name,
-			},
-		},
-	}); err != nil {
-		t.Fatalf("unexpected error from UpdateClientConnState: %v", err)
-	}
-	if err := xdsC.WaitForCancelReportLoad(ctx); err != nil {
-		t.Fatalf("unexpected error waiting form load report to be canceled: %v", err)
-	}
-	got2, err2 := xdsC.WaitForReportLoad(ctx)
-	if err2 != nil {
-		t.Fatalf("xdsClient.ReportLoad failed with error: %v", err2)
-	}
-	if got2.Server != testLRSServerConfig2 {
-		t.Fatalf("xdsClient.ReportLoad called with {%q}: want {%q}", got2.Server, testLRSServerConfig2)
-	}
-
-	// Update LRS server to nil, to disable LRS.
-	if err := b.UpdateClientConnState(balancer.ClientConnState{
-		ResolverState: xdsclient.SetClient(resolver.State{Endpoints: endpoints}, xdsC),
-		BalancerConfig: &LBConfig{
-			Cluster:        testClusterName,
-			EDSServiceName: testServiceName,
-			ChildPolicy: &internalserviceconfig.BalancerConfig{
-				Name: roundrobin.Name,
-			},
-		},
-	}); err != nil {
-		t.Fatalf("unexpected error from UpdateClientConnState: %v", err)
-	}
-	if err := xdsC.WaitForCancelReportLoad(ctx); err != nil {
-		t.Fatalf("unexpected error waiting form load report to be canceled: %v", err)
-	}
-
-	shortCtx, shortCancel := context.WithTimeout(context.Background(), defaultShortTestTimeout)
-	defer shortCancel()
-	if s, err := xdsC.WaitForReportLoad(shortCtx); err != context.DeadlineExceeded {
-		t.Fatalf("unexpected load report to server: %q", s)
-	}
-}
-
 // Test verifies that child policies was updated on receipt of
 // configuration update.
 func (s) TestChildPolicyUpdatedOnConfigUpdate(t *testing.T) {