Repository: grpc/grpc-go
File: internal/xds/balancer/clusterimpl/tests/balancer_test.go (Go)

Code before:
		serviceName   = "test-child-policy"
		routeName     = "route-" + serviceName
		clusterName   = "cluster-" + serviceName
		endpointsName = "endpoints-" + serviceName
	)

	resources := e2e.UpdateOptions{
		NodeID:    nodeID,
		Listeners: []*v3listenerpb.Listener{e2e.DefaultClientListener(serviceName, routeName)},
		Routes:    []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig(routeName, serviceName, clusterName)},
		Clusters:  []*v3clusterpb.Cluster{e2e.DefaultCluster(clusterName, endpointsName, e2e.SecurityLevelNone)},
		Endpoints: []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(endpointsName, "localhost", []uint32{testutils.ParsePort(t, server.Address)})},
	}

	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
	defer cancel()
	if err := mgmtServer.Update(ctx, resources); err != nil {
		t.Fatalf("Failed to update xDS resources: %v", err)
	}

	// Register stub pickfirst LB policy so that we can catch config changes.
	pfBuilder := balancer.Get(pickfirst.Name)
	internal.BalancerUnregister(pfBuilder.Name())
	lbCfgCh := make(chan serviceconfig.LoadBalancingConfig, 1)
	updatedChildPolicy := ""
	var mu sync.Mutex
	stub.Register(pfBuilder.Name(), stub.BalancerFuncs{
		ParseConfig: func(lbCfg json.RawMessage) (serviceconfig.LoadBalancingConfig, error) {
			return pfBuilder.(balancer.ConfigParser).ParseConfig(lbCfg)
		},
		Init: func(bd *stub.BalancerData) {
			bd.ChildBalancer = pfBuilder.Build(bd.ClientConn, bd.BuildOptions)
		},
		UpdateClientConnState: func(bd *stub.BalancerData, ccs balancer.ClientConnState) error {
			select {
			case lbCfgCh <- ccs.BalancerConfig:
			default:
			}
			mu.Lock()
			updatedChildPolicy = pfBuilder.Name()
			mu.Unlock()
			return bd.ChildBalancer.UpdateClientConnState(ccs)
		},
		Close: func(bd *stub.BalancerData) {
			bd.ChildBalancer.Close()
		},
	})
	defer balancer.Register(pfBuilder)

	cc, err := grpc.NewClient(fmt.Sprintf("xds:///%s", serviceName), grpc.WithResolvers(resolverBuilder), grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {

Review comment (suggestion):
Please consider using `atomic.Pointer[string]` instead. You can see lot of examples of `atomic.Pointer` used in our codebase.

Code after:

	wantRPCDropRate = float64(dropNumerator2) / float64(dropDenominator2)
	rpcCount = computeIdealNumRpcs(wantRPCDropRate, errorTolerance)
	t.Logf("About to send %d RPCs to test drop rate of %v", rpcCount, wantRPCDropRate)

	if err := verifyDropRateByCategory(ctx, client, int(rpcCount), mgmtServer.LRSServer, resources.Clusters[0].Name, dropReason2, wantRPCDropRate, errorTolerance); err != nil {
		t.Fatal(err)
	}
}

// Tests that circuit breaking limits RPCs in LOGICAL_DNS clusters E2E.
func (s) TestCircuitBreakingLogicalDNS(t *testing.T) {
	// Create an xDS management server that serves ADS requests.
	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{})

	// Create bootstrap configuration pointing to the above management server.
	nodeID := uuid.New().String()
	bc := e2e.DefaultBootstrapContents(t, nodeID, mgmtServer.Address)
	testutils.CreateBootstrapFileForTesting(t, bc)

	// Create an xDS resolver with the above bootstrap configuration.
	if internal.NewXDSResolverWithConfigForTesting == nil {
		t.Fatalf("internal.NewXDSResolverWithConfigForTesting is nil")
	}
	resolverBuilder, err := internal.NewXDSResolverWithConfigForTesting.(func([]byte) (resolver.Builder, error))(bc)
	if err != nil {
		t.Fatalf("Failed to create xDS resolver for testing: %v", err)
	}

	// Start a server backend exposing the test service.
	f := &stubserver.StubServer{
		EmptyCallF: func(context.Context, *testpb.Empty) (*testpb.Empty, error) {
			return &testpb.Empty{}, nil
		},
		FullDuplexCallF: func(stream testgrpc.TestService_FullDuplexCallServer) error {
			for {
				if _, err := stream.Recv(); err != nil {
					return err
				}
			}
		},
	}
	server := stubserver.StartTestService(t, f)
	defer server.Stop()
	host, port := hostAndPortFromAddress(t, server.Address)

	// Configure the xDS management server with default resources. Override the
	// default cluster to include a circuit breaking config.
	const serviceName = "my-test-xds-service"
	const maxRequests = 3
	resources := e2e.DefaultClientResources(e2e.ResourceParams{