xds/clusterimpl: Convert existing unit tests to e2e style (1/N) (#8549)

Author: Pranjali-2501

internal/xds/balancer/clusterimpl/balancer_test.go

@@ -23,7 +23,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"sort"
 	"strings"
 	"sync"
 	"testing"
@@ -64,7 +63,6 @@ const (
 
 var (
 	testBackendEndpoints = []resolver.Endpoint{{Addresses: []resolver.Address{{Addr: "1.1.1.1:1"}}}}
-	cmpOpts              = cmp.Options{cmpopts.EquateEmpty(), cmp.AllowUnexported(loadData{}, localityData{}, requestData{}, serverLoadData{}), sortDataSlice}
 	toleranceCmpOpt      = cmp.Options{cmpopts.EquateApprox(0, 1e-5), cmp.AllowUnexported(loadData{}, localityData{}, requestData{}, serverLoadData{})}
 )
 
@@ -279,345 +277,6 @@ func init() {
 	NewRandomWRR = testutils.NewTestWRR
 }
 
-var sortDataSlice = cmp.Transformer("SortDataSlice", func(in []*loadData) []*loadData {
-	out := append([]*loadData(nil), in...) // Copy input to avoid mutating it
-	sort.Slice(out,
-		func(i, j int) bool {
-			if out[i].cluster < out[j].cluster {
-				return true
-			}
-			if out[i].cluster == out[j].cluster {
-				return out[i].service < out[j].service
-			}
-			return false
-		},
-	)
-	return out
-})
-
-func verifyLoadStoreData(wantStoreData, gotStoreData *loadData) error {
-	if diff := cmp.Diff(wantStoreData, gotStoreData, cmpOpts); diff != "" {
-		return fmt.Errorf("store.stats() returned unexpected diff (-want +got):\n%s", diff)
-	}
-	return nil
-}
-
-// TestDropByCategory verifies that the balancer correctly drops the picks, and
-// that the drops are reported.
-func (s) TestDropByCategory(t *testing.T) {
-	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
-	defer cancel()
-
-	defer xdsclient.ClearCounterForTesting(testClusterName, testServiceName)
-	xdsC := fakeclient.NewClient()
-
-	builder := balancer.Get(Name)
-	cc := testutils.NewBalancerClientConn(t)
-	b := builder.Build(cc, balancer.BuildOptions{})
-	defer b.Close()
-
-	const (
-		dropReason      = "test-dropping-category"
-		dropNumerator   = 1
-		dropDenominator = 2
-	)
-	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: testBackendEndpoints}, xdsC),
-		BalancerConfig: &LBConfig{
-			Cluster:             testClusterName,
-			EDSServiceName:      testServiceName,
-			LoadReportingServer: testLRSServerConfig,
-			DropCategories: []DropConfig{{
-				Category:           dropReason,
-				RequestsPerMillion: million * dropNumerator / dropDenominator,
-			}},
-			ChildPolicy: &internalserviceconfig.BalancerConfig{
-				Name: roundrobin.Name,
-			},
-		},
-	}); err != nil {
-		t.Fatalf("unexpected error from UpdateClientConnState: %v", err)
-	}
-
-	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)
-	}
-
-	sc1 := <-cc.NewSubConnCh
-	sc1.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Connecting})
-	// This should get the connecting picker.
-	if err := cc.WaitForPickerWithErr(ctx, balancer.ErrNoSubConnAvailable); err != nil {
-		t.Fatal(err.Error())
-	}
-
-	sc1.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Ready})
-	// Test pick with one backend.
-
-	testClusterLoadReporter := &testLoadReporter{cluster: testClusterName, service: testServiceName, drops: make(map[string]uint64), localityRPCCount: make(map[clients.Locality]*rpcCountData)}
-
-	const rpcCount = 24
-	if err := cc.WaitForPicker(ctx, func(p balancer.Picker) error {
-		// Override the loadStore in the picker with testClusterLoadReporter.
-		picker := p.(*picker)
-		originalLoadStore := picker.loadStore
-		picker.loadStore = testClusterLoadReporter
-		defer func() { picker.loadStore = originalLoadStore }()
-
-		for i := 0; i < rpcCount; i++ {
-			gotSCSt, err := p.Pick(balancer.PickInfo{})
-			// Even RPCs are dropped.
-			if i%2 == 0 {
-				if err == nil || !strings.Contains(err.Error(), "dropped") {
-					return fmt.Errorf("pick.Pick, got %v, %v, want error RPC dropped", gotSCSt, err)
-				}
-				continue
-			}
-			if err != nil || gotSCSt.SubConn != sc1 {
-				return fmt.Errorf("picker.Pick, got %v, %v, want SubConn=%v", gotSCSt, err, sc1)
-			}
-			if gotSCSt.Done == nil {
-				continue
-			}
-			// Fail 1/4th of the requests that are not dropped.
-			if i%8 == 1 {
-				gotSCSt.Done(balancer.DoneInfo{Err: fmt.Errorf("test error")})
-			} else {
-				gotSCSt.Done(balancer.DoneInfo{})
-			}
-		}
-		return nil
-	}); err != nil {
-		t.Fatal(err.Error())
-	}
-
-	// Dump load data from the store and compare with expected counts.
-	const dropCount = rpcCount * dropNumerator / dropDenominator
-	wantStatsData0 := &loadData{
-		cluster:    testClusterName,
-		service:    testServiceName,
-		totalDrops: dropCount,
-		drops:      map[string]uint64{dropReason: dropCount},
-		localityStats: map[clients.Locality]localityData{
-			{}: {requestStats: requestData{
-				succeeded: (rpcCount - dropCount) * 3 / 4,
-				errored:   (rpcCount - dropCount) / 4,
-				issued:    rpcCount - dropCount,
-			}},
-		},
-	}
-
-	gotStatsData0 := testClusterLoadReporter.stats()
-	if err := verifyLoadStoreData(wantStatsData0, gotStatsData0); err != nil {
-		t.Fatal(err)
-	}
-
-	// Send an update with new drop configs.
-	const (
-		dropReason2      = "test-dropping-category-2"
-		dropNumerator2   = 1
-		dropDenominator2 = 4
-	)
-	if err := b.UpdateClientConnState(balancer.ClientConnState{
-		ResolverState: xdsclient.SetClient(resolver.State{Endpoints: testBackendEndpoints}, xdsC),
-		BalancerConfig: &LBConfig{
-			Cluster:             testClusterName,
-			EDSServiceName:      testServiceName,
-			LoadReportingServer: testLRSServerConfig,
-			DropCategories: []DropConfig{{
-				Category:           dropReason2,
-				RequestsPerMillion: million * dropNumerator2 / dropDenominator2,
-			}},
-			ChildPolicy: &internalserviceconfig.BalancerConfig{
-				Name: roundrobin.Name,
-			},
-		},
-	}); err != nil {
-		t.Fatalf("unexpected error from UpdateClientConnState: %v", err)
-	}
-
-	if err := cc.WaitForPicker(ctx, func(p balancer.Picker) error {
-		// Override the loadStore in the picker with testClusterLoadReporter.
-		picker := p.(*picker)
-		originalLoadStore := picker.loadStore
-		picker.loadStore = testClusterLoadReporter
-		defer func() { picker.loadStore = originalLoadStore }()
-		for i := 0; i < rpcCount; i++ {
-			gotSCSt, err := p.Pick(balancer.PickInfo{})
-			// Even RPCs are dropped.
-			if i%4 == 0 {
-				if err == nil || !strings.Contains(err.Error(), "dropped") {
-					return fmt.Errorf("pick.Pick, got %v, %v, want error RPC dropped", gotSCSt, err)
-				}
-				continue
-			}
-			if err != nil || gotSCSt.SubConn != sc1 {
-				return fmt.Errorf("picker.Pick, got %v, %v, want SubConn=%v", gotSCSt, err, sc1)
-			}
-			if gotSCSt.Done != nil {
-				gotSCSt.Done(balancer.DoneInfo{})
-			}
-		}
-		return nil
-	}); err != nil {
-		t.Fatal(err.Error())
-	}
-
-	const dropCount2 = rpcCount * dropNumerator2 / dropDenominator2
-	wantStatsData1 := &loadData{
-		cluster:    testClusterName,
-		service:    testServiceName,
-		totalDrops: dropCount2,
-		drops:      map[string]uint64{dropReason2: dropCount2},
-		localityStats: map[clients.Locality]localityData{
-			{}: {requestStats: requestData{
-				succeeded: rpcCount - dropCount2,
-				issued:    rpcCount - dropCount2,
-			}},
-		},
-	}
-
-	gotStatsData1 := testClusterLoadReporter.stats()
-	if err := verifyLoadStoreData(wantStatsData1, gotStatsData1); err != nil {
-		t.Fatal(err)
-	}
-}
-
-// TestDropCircuitBreaking verifies that the balancer correctly drops the picks
-// due to circuit breaking, and that the drops are reported.
-func (s) TestDropCircuitBreaking(t *testing.T) {
-	defer xdsclient.ClearCounterForTesting(testClusterName, testServiceName)
-	xdsC := fakeclient.NewClient()
-
-	builder := balancer.Get(Name)
-	cc := testutils.NewBalancerClientConn(t)
-	b := builder.Build(cc, balancer.BuildOptions{})
-	defer b.Close()
-
-	var maxRequest uint32 = 50
-	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: testBackendEndpoints}, xdsC),
-		BalancerConfig: &LBConfig{
-			Cluster:               testClusterName,
-			EDSServiceName:        testServiceName,
-			LoadReportingServer:   testLRSServerConfig,
-			MaxConcurrentRequests: &maxRequest,
-			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)
-	}
-
-	sc1 := <-cc.NewSubConnCh
-	sc1.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Connecting})
-	// This should get the connecting picker.
-	if err := cc.WaitForPickerWithErr(ctx, balancer.ErrNoSubConnAvailable); err != nil {
-		t.Fatal(err.Error())
-	}
-
-	sc1.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Ready})
-	// Test pick with one backend.
-	testClusterLoadReporter := &testLoadReporter{cluster: testClusterName, service: testServiceName, drops: make(map[string]uint64), localityRPCCount: make(map[clients.Locality]*rpcCountData)}
-	const rpcCount = 100
-	if err := cc.WaitForPicker(ctx, func(p balancer.Picker) error {
-		dones := []func(){}
-		// Override the loadStore in the picker with testClusterLoadReporter.
-		picker := p.(*picker)
-		originalLoadStore := picker.loadStore
-		picker.loadStore = testClusterLoadReporter
-		defer func() { picker.loadStore = originalLoadStore }()
-
-		for i := 0; i < rpcCount; i++ {
-			gotSCSt, err := p.Pick(balancer.PickInfo{})
-			if i < 50 && err != nil {
-				return fmt.Errorf("The first 50%% picks should be non-drops, got error %v", err)
-			} else if i > 50 && err == nil {
-				return fmt.Errorf("The second 50%% picks should be drops, got error <nil>")
-			}
-			dones = append(dones, func() {
-				if gotSCSt.Done != nil {
-					gotSCSt.Done(balancer.DoneInfo{})
-				}
-			})
-		}
-		for _, done := range dones {
-			done()
-		}
-
-		dones = []func(){}
-		// Pick without drops.
-		for i := 0; i < 50; i++ {
-			gotSCSt, err := p.Pick(balancer.PickInfo{})
-			if err != nil {
-				t.Errorf("The third 50%% picks should be non-drops, got error %v", err)
-			}
-			dones = append(dones, func() {
-				if gotSCSt.Done != nil {
-					// Fail these requests to test error counts in the load
-					// report.
-					gotSCSt.Done(balancer.DoneInfo{Err: fmt.Errorf("test error")})
-				}
-			})
-		}
-		for _, done := range dones {
-			done()
-		}
-
-		return nil
-	}); err != nil {
-		t.Fatal(err.Error())
-	}
-
-	// Dump load data from the store and compare with expected counts.
-	wantStatsData0 := &loadData{
-		cluster:    testClusterName,
-		service:    testServiceName,
-		totalDrops: uint64(maxRequest),
-		localityStats: map[clients.Locality]localityData{
-			{}: {requestStats: requestData{
-				succeeded: uint64(rpcCount - maxRequest),
-				errored:   50,
-				issued:    uint64(rpcCount - maxRequest + 50),
-			}},
-		},
-	}
-
-	gotStatsData0 := testClusterLoadReporter.stats()
-	if err := verifyLoadStoreData(wantStatsData0, gotStatsData0); err != nil {
-		t.Fatal(err)
-	}
-}
-
 // TestPickerUpdateAfterClose covers the case where a child policy sends a
 // picker update after the cluster_impl policy is closed. Because picker updates
 // are handled in the run() goroutine, which exits before Close() returns, we