Repository: grpc/grpc-go
File: xds/internal/balancer/clusterresolver/e2e_test/eds_impl_test.go (Go)

Code before:
		t.Run(tc.name, func(t *testing.T) {
			origDualstackEndpointsEnabled := envconfig.XDSDualstackEndpointsEnabled
			defer func() {
				envconfig.XDSDualstackEndpointsEnabled = origDualstackEndpointsEnabled
			}()
			envconfig.XDSDualstackEndpointsEnabled = tc.dualstackEndpointsEnabled

			// Wrap the round robin balancer to intercept resolver updates.
			originalRRBuilder := balancer.Get(roundrobin.Name)
			defer func() {
				balancer.Register(originalRRBuilder)
			}()
			resolverUpdateCh := make(chan resolver.State, 1)
			stub.Register(roundrobin.Name, stub.BalancerFuncs{
				Init: func(bd *stub.BalancerData) {
					bd.Data = originalRRBuilder.Build(bd.ClientConn, bd.BuildOptions)
				},
				Close: func(bd *stub.BalancerData) {
					bd.Data.(balancer.Balancer).Close()
				},
				UpdateClientConnState: func(bd *stub.BalancerData, ccs balancer.ClientConnState) error {
					select {
					case resolverUpdateCh <- ccs.ResolverState:
					default:
						// Don't block forever in case of multiple updates.
					}
					return bd.Data.(balancer.Balancer).UpdateClientConnState(ccs)
				},
			})

			// Spin up a management server to receive xDS resources from.
			mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{})

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

			// Create xDS resources for consumption by the test. We start off with a
			// single backend in a single EDS locality.
			resources := clientEndpointsResource(nodeID, edsServiceName, []e2e.LocalityOptions{{
				Name:   localityName1,
				Weight: 1,
				Backends: []e2e.BackendOptions{{
					Ports: ports,
				}},
			}})
			if err := mgmtServer.Update(ctx, resources); err != nil {
				t.Fatal(err)
			}

			// Create an xDS client talking to the above management server, configured

Review comment (bug):
I received some comments against performing non-deterministic writes to channels like this in tests because it can lead to flakiness as well.

In the PR description, you mentioned that when the test failed, there was no logs. Were you able to repro this on g3 with logs enabled to figure out the exact cause of the problem? 

Code after:
		t.Run(tc.name, func(t *testing.T) {
			origDualstackEndpointsEnabled := envconfig.XDSDualstackEndpointsEnabled
			defer func() {
				envconfig.XDSDualstackEndpointsEnabled = origDualstackEndpointsEnabled
			}()
			envconfig.XDSDualstackEndpointsEnabled = tc.dualstackEndpointsEnabled

			// Wrap the round robin balancer to intercept resolver updates.
			originalRRBuilder := balancer.Get(roundrobin.Name)
			defer func() {
				balancer.Register(originalRRBuilder)
			}()
			resolverState := atomic.Pointer[resolver.State]{}
			resolverState.Store(&resolver.State{})
			stub.Register(roundrobin.Name, stub.BalancerFuncs{
				Init: func(bd *stub.BalancerData) {
					bd.Data = originalRRBuilder.Build(bd.ClientConn, bd.BuildOptions)
				},
				Close: func(bd *stub.BalancerData) {
					bd.Data.(balancer.Balancer).Close()
				},
				UpdateClientConnState: func(bd *stub.BalancerData, ccs balancer.ClientConnState) error {
					resolverState.Store(&ccs.ResolverState)
					return bd.Data.(balancer.Balancer).UpdateClientConnState(ccs)
				},
			})

			// Spin up a management server to receive xDS resources from.
			mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{})

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

			// Create xDS resources for consumption by the test. We start off with a
			// single backend in a single EDS locality.
			resources := clientEndpointsResource(nodeID, edsServiceName, []e2e.LocalityOptions{{
				Name:   localityName1,
				Weight: 1,
				Backends: []e2e.BackendOptions{{
					Ports: ports,
				}},
			}})
			if err := mgmtServer.Update(ctx, resources); err != nil {
				t.Fatal(err)
			}

			// Create an xDS client talking to the above management server, configured
			// with a short watch expiry timeout.
			xdsClient, close, err := xdsclient.NewForTesting(xdsclient.OptionsForTesting{
				Name:     t.Name(),