resolver/manual: allow calling UpdateState with an un-Built resolver (#8150)

Author: dfawley

balancer/endpointsharding/endpointsharding_test.go

@@ -187,7 +187,7 @@ func (s) TestEndpointShardingBasic(t *testing.T) {
 
 	// When the resolver reports an error, the picker should get updated to
 	// return the resolver error.
-	mr.ReportError(errors.New("test error"))
+	mr.CC().ReportError(errors.New("test error"))
 	testutils.AwaitState(ctx, t, cc, connectivity.TransientFailure)
 	for ; ctx.Err() == nil; <-time.After(time.Millisecond) {
 		_, err := client.EmptyCall(ctx, &testpb.Empty{})

balancer/grpclb/grpclb_test.go

@@ -859,7 +859,7 @@ func (s) TestGRPCLB_Fallback(t *testing.T) {
 	// Push another update to the resolver, this time with a valid balancer
 	// address in the attributes field.
 	rs = resolver.State{
-		ServiceConfig: r.CC.ParseServiceConfig(grpclbConfig),
+		ServiceConfig: r.CC().ParseServiceConfig(grpclbConfig),
 		Addresses:     []resolver.Address{{Addr: beLis.Addr().String()}},
 	}
 	rs = grpclbstate.Set(rs, &grpclbstate.State{BalancerAddresses: []resolver.Address{{Addr: tss.lbAddr, ServerName: lbServerName}}})
@@ -1023,7 +1023,7 @@ func (s) TestGRPCLB_FallBackWithNoServerAddress(t *testing.T) {
 		// fallback and use the fallback backend.
 		r.UpdateState(resolver.State{
 			Addresses:     []resolver.Address{{Addr: beLis.Addr().String()}},
-			ServiceConfig: r.CC.ParseServiceConfig(grpclbConfig),
+			ServiceConfig: r.CC().ParseServiceConfig(grpclbConfig),
 		})
 
 		sCtx, sCancel := context.WithTimeout(context.Background(), defaultTestShortTimeout)
@@ -1051,7 +1051,7 @@ func (s) TestGRPCLB_FallBackWithNoServerAddress(t *testing.T) {
 		// be used.
 		rs := resolver.State{
 			Addresses:     []resolver.Address{{Addr: beLis.Addr().String()}},
-			ServiceConfig: r.CC.ParseServiceConfig(grpclbConfig),
+			ServiceConfig: r.CC().ParseServiceConfig(grpclbConfig),
 		}
 		rs = grpclbstate.Set(rs, &grpclbstate.State{BalancerAddresses: []resolver.Address{{Addr: tss.lbAddr, ServerName: lbServerName}}})
 		r.UpdateState(rs)
@@ -1112,7 +1112,7 @@ func (s) TestGRPCLB_PickFirst(t *testing.T) {
 
 	// Push a service config with grpclb as the load balancing policy and
 	// configure pick_first as its child policy.
-	rs := resolver.State{ServiceConfig: r.CC.ParseServiceConfig(`{"loadBalancingConfig":[{"grpclb":{"childPolicy":[{"pick_first":{}}]}}]}`)}
+	rs := resolver.State{ServiceConfig: r.CC().ParseServiceConfig(`{"loadBalancingConfig":[{"grpclb":{"childPolicy":[{"pick_first":{}}]}}]}`)}
 
 	// Push a resolver update with the remote balancer address specified via
 	// attributes.
@@ -1152,7 +1152,7 @@ func (s) TestGRPCLB_PickFirst(t *testing.T) {
 			},
 		},
 	}
-	rs = grpclbstate.Set(resolver.State{ServiceConfig: r.CC.ParseServiceConfig(grpclbConfig)}, s)
+	rs = grpclbstate.Set(resolver.State{ServiceConfig: r.CC().ParseServiceConfig(grpclbConfig)}, s)
 	r.UpdateState(rs)
 	testC := testgrpc.NewTestServiceClient(cc)
 	if err := roundrobin.CheckRoundRobinRPCs(ctx, testC, beServerAddrs[1:]); err != nil {
@@ -1261,7 +1261,7 @@ func testGRPCLBEmptyServerList(t *testing.T, svcfg string) {
 			},
 		},
 	}
-	rs := grpclbstate.Set(resolver.State{ServiceConfig: r.CC.ParseServiceConfig(svcfg)}, s)
+	rs := grpclbstate.Set(resolver.State{ServiceConfig: r.CC().ParseServiceConfig(svcfg)}, s)
 	r.UpdateState(rs)
 	t.Log("Perform an initial RPC and expect it to succeed...")
 	if _, err := testC.EmptyCall(ctx, &testpb.Empty{}, grpc.WaitForReady(true)); err != nil {
@@ -1329,7 +1329,7 @@ func (s) TestGRPCLBWithTargetNameFieldInConfig(t *testing.T) {
 	// Push a resolver update with grpclb configuration which does not contain the
 	// target_name field. Our fake remote balancer is configured to always
 	// expect `beServerName` as the server name in the initial request.
-	rs := grpclbstate.Set(resolver.State{ServiceConfig: r.CC.ParseServiceConfig(grpclbConfig)},
+	rs := grpclbstate.Set(resolver.State{ServiceConfig: r.CC().ParseServiceConfig(grpclbConfig)},
 		&grpclbstate.State{BalancerAddresses: []resolver.Address{{
 			Addr:       tss.lbAddr,
 			ServerName: lbServerName,
@@ -1366,7 +1366,7 @@ func (s) TestGRPCLBWithTargetNameFieldInConfig(t *testing.T) {
 			},
 		},
 	}
-	rs = grpclbstate.Set(resolver.State{ServiceConfig: r.CC.ParseServiceConfig(lbCfg)}, s)
+	rs = grpclbstate.Set(resolver.State{ServiceConfig: r.CC().ParseServiceConfig(lbCfg)}, s)
 	r.UpdateState(rs)
 	select {
 	case <-ctx.Done():
@@ -1432,7 +1432,7 @@ func runAndCheckStats(t *testing.T, drop bool, statsChan chan *lbpb.ClientStats,
 	cc.Connect()
 	defer cc.Close()
 
-	rstate := resolver.State{ServiceConfig: r.CC.ParseServiceConfig(grpclbConfig)}
+	rstate := resolver.State{ServiceConfig: r.CC().ParseServiceConfig(grpclbConfig)}
 	r.UpdateState(grpclbstate.Set(rstate, &grpclbstate.State{BalancerAddresses: []resolver.Address{{
 		Addr:       tss.lbAddr,
 		ServerName: lbServerName,

balancer/lazy/lazy_ext_test.go

@@ -271,7 +271,7 @@ func (s) TestGoodUpdateThenResolverError(t *testing.T) {
 	defer cc.Close()
 	cc.Connect()
 
-	mr.ReportError(errors.New("test error"))
+	mr.CC().ReportError(errors.New("test error"))
 	// The channel should remain in IDLE as the ExitIdle calls are not
 	// propagated to the lazy balancer from the stub balancer.
 	shortCtx, shortCancel := context.WithTimeout(ctx, defaultTestShortTimeout)
@@ -367,7 +367,7 @@ func (s) TestResolverErrorThenGoodUpdate(t *testing.T) {
 	cc.Connect()
 
 	// Send an error followed by a good update.
-	mr.ReportError(errors.New("test error"))
+	mr.CC().ReportError(errors.New("test error"))
 	mr.UpdateState(resolver.State{
 		Endpoints: []resolver.Endpoint{
 			{Addresses: []resolver.Address{{Addr: backend.Address}}},

balancer/pickfirst/pickfirst_ext_test.go

@@ -73,7 +73,7 @@ func Test(t *testing.T) {
 func parseServiceConfig(t *testing.T, r *manual.Resolver, sc string) *serviceconfig.ParseResult {
 	t.Helper()
 
-	scpr := r.CC.ParseServiceConfig(sc)
+	scpr := r.CC().ParseServiceConfig(sc)
 	if scpr.Err != nil {
 		t.Fatalf("Failed to parse service config %q: %v", sc, scpr.Err)
 	}
@@ -756,7 +756,7 @@ func (s) TestPickFirst_ResolverError_NoPreviousUpdate(t *testing.T) {
 	cc, r, _ := setupPickFirst(t, 0)
 
 	nrErr := errors.New("error from name resolver")
-	r.ReportError(nrErr)
+	r.CC().ReportError(nrErr)
 
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
@@ -789,7 +789,7 @@ func (s) TestPickFirst_ResolverError_WithPreviousUpdate_Ready(t *testing.T) {
 	}
 
 	nrErr := errors.New("error from name resolver")
-	r.ReportError(nrErr)
+	r.CC().ReportError(nrErr)
 
 	// Ensure that RPCs continue to succeed for the next second.
 	client := testgrpc.NewTestServiceClient(cc)
@@ -848,7 +848,7 @@ func (s) TestPickFirst_ResolverError_WithPreviousUpdate_Connecting(t *testing.T)
 	testutils.AwaitState(ctx, t, cc, connectivity.Connecting)
 
 	nrErr := errors.New("error from name resolver")
-	r.ReportError(nrErr)
+	r.CC().ReportError(nrErr)
 
 	// RPCs should fail with deadline exceed error as long as they are in
 	// CONNECTING and not the error returned by the name resolver.
@@ -909,7 +909,7 @@ func (s) TestPickFirst_ResolverError_WithPreviousUpdate_TransientFailure(t *test
 	// error instead of the old error that caused the channel to move to
 	// TRANSIENT_FAILURE in the first place.
 	nrErr := errors.New("error from name resolver")
-	r.ReportError(nrErr)
+	r.CC().ReportError(nrErr)
 	client := testgrpc.NewTestServiceClient(cc)
 	for ; ctx.Err() == nil; <-time.After(defaultTestShortTimeout) {
 		if _, err := client.EmptyCall(ctx, &testpb.Empty{}); strings.Contains(err.Error(), nrErr.Error()) {

balancer/weightedroundrobin/balancer_test.go

@@ -449,7 +449,7 @@ func (s) TestBalancer_TwoAddresses_OOBThenPerCall(t *testing.T) {
 
 	// Update to per-call weights.
 	c := svcConfig(t, perCallConfig)
-	parsedCfg := srv1.R.CC.ParseServiceConfig(c)
+	parsedCfg := srv1.R.CC().ParseServiceConfig(c)
 	if parsedCfg.Err != nil {
 		panic(fmt.Sprintf("Error parsing config %q: %v", c, parsedCfg.Err))
 	}
@@ -563,7 +563,7 @@ func (s) TestBalancer_TwoAddresses_ErrorPenalty(t *testing.T) {
 	newCfg := oobConfig
 	newCfg.ErrorUtilizationPenalty = float64p(0.9)
 	c := svcConfig(t, newCfg)
-	parsedCfg := srv1.R.CC.ParseServiceConfig(c)
+	parsedCfg := srv1.R.CC().ParseServiceConfig(c)
 	if parsedCfg.Err != nil {
 		panic(fmt.Sprintf("Error parsing config %q: %v", c, parsedCfg.Err))
 	}

clientconn_test.go

@@ -62,7 +62,7 @@ func init() {
 }
 
 func parseCfg(r *manual.Resolver, s string) *serviceconfig.ParseResult {
-	scpr := r.CC.ParseServiceConfig(s)
+	scpr := r.CC().ParseServiceConfig(s)
 	if scpr.Err != nil {
 		panic(fmt.Sprintf("Error parsing config %q: %v", s, scpr.Err))
 	}
@@ -666,7 +666,7 @@ func (s) TestResolverServiceConfigBeforeAddressNotPanic(t *testing.T) {
 	cc.Connect()
 	// SwitchBalancer before NewAddress. There was no balancer created, this
 	// makes sure we don't call close on nil balancerWrapper.
-	r.UpdateState(resolver.State{ServiceConfig: r.CC.ParseServiceConfig(grpclbServiceConfig)}) // This should not panic.
+	r.UpdateState(resolver.State{ServiceConfig: r.CC().ParseServiceConfig(grpclbServiceConfig)}) // This should not panic.
 
 	time.Sleep(time.Second) // Sleep to make sure the service config is handled by ClientConn.
 }
@@ -681,7 +681,7 @@ func (s) TestResolverServiceConfigWhileClosingNotPanic(t *testing.T) {
 		cc.Connect()
 		// Send a new service config while closing the ClientConn.
 		go cc.Close()
-		go r.UpdateState(resolver.State{ServiceConfig: r.CC.ParseServiceConfig(rrServiceConfig)}) // This should not panic.
+		go r.UpdateState(resolver.State{ServiceConfig: r.CC().ParseServiceConfig(rrServiceConfig)}) // This should not panic.
 	}
 }
 
@@ -775,7 +775,7 @@ func (s) TestDisableServiceConfigOption(t *testing.T) {
 	}
 	defer cc.Close()
 	cc.Connect()
-	r.UpdateState(resolver.State{ServiceConfig: r.CC.ParseServiceConfig(`{
+	r.UpdateState(resolver.State{ServiceConfig: r.CC().ParseServiceConfig(`{
     "methodConfig": [
         {
             "name": [

internal/stubserver/stubserver.go

@@ -276,7 +276,7 @@ func (ss *StubServer) Stop() {
 }
 
 func parseCfg(r *manual.Resolver, s string) *serviceconfig.ParseResult {
-	g := r.CC.ParseServiceConfig(s)
+	g := r.CC().ParseServiceConfig(s)
 	if g.Err != nil {
 		panic(fmt.Sprintf("Error parsing config %q: %v", s, g.Err))
 	}

resolver/manual/manual.go

@@ -62,7 +62,7 @@ type Resolver struct {
 	// Fields actually belong to the resolver.
 	// Guards access to below fields.
 	mu sync.Mutex
-	CC resolver.ClientConn
+	cc resolver.ClientConn
 	// Storing the most recent state update makes this resolver resilient to
 	// restarts, which is possible with channel idleness.
 	lastSeenState *resolver.State
@@ -78,12 +78,12 @@ func (r *Resolver) InitialState(s resolver.State) {
 func (r *Resolver) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
-	// Call BuildCallback after locking to avoid a race when UpdateState
-	// or ReportError is called before Build returns.
+	// Call BuildCallback after locking to avoid a race when UpdateState or CC
+	// is called before Build returns.
 	r.BuildCallback(target, cc, opts)
-	r.CC = cc
+	r.cc = cc
 	if r.lastSeenState != nil {
-		err := r.CC.UpdateState(*r.lastSeenState)
+		err := r.cc.UpdateState(*r.lastSeenState)
 		go r.UpdateStateCallback(err)
 	}
 	return r, nil
@@ -104,25 +104,27 @@ func (r *Resolver) Close() {
 	r.CloseCallback()
 }
 
-// UpdateState calls CC.UpdateState.
+// UpdateState calls UpdateState(s) on the channel.  If the resolver has not
+// been Built before, this instead sets the initial state of the resolver, like
+// InitialState.
 func (r *Resolver) UpdateState(s resolver.State) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
-	var err error
-	if r.CC == nil {
-		panic("cannot update state as channel has not exited IDLE state")
-	}
-	err = r.CC.UpdateState(s)
 	r.lastSeenState = &s
+	if r.cc == nil {
+		return
+	}
+	err := r.cc.UpdateState(s)
 	r.UpdateStateCallback(err)
 }
 
-// ReportError calls CC.ReportError.
-func (r *Resolver) ReportError(err error) {
+// CC returns r's ClientConn when r was last Built.  Panics if the resolver has
+// not been Built before.
+func (r *Resolver) CC() resolver.ClientConn {
 	r.mu.Lock()
 	defer r.mu.Unlock()
-	if r.CC == nil {
-		panic("cannot report error as channel has not exited IDLE state")
+	if r.cc == nil {
+		panic("Manual resolver instance has not yet been built.")
 	}
-	r.CC.ReportError(err)
+	return r.cc
 }

resolver/manual/manual_test.go

@@ -36,26 +36,14 @@ func TestResolver(t *testing.T) {
 		},
 	})
 
-	t.Run("update_state_panics", func(t *testing.T) {
+	t.Run("cc_panics", func(t *testing.T) {
 		defer func() {
-			want := "cannot update state as channel has not exited IDLE state"
+			want := "Manual resolver instance has not yet been built."
 			if r := recover(); r != want {
 				t.Errorf("expected panic %q, got %q", want, r)
 			}
 		}()
-		r.UpdateState(resolver.State{Addresses: []resolver.Address{
-			{Addr: "address"},
-			{Addr: "anotheraddress"},
-		}})
-	})
-	t.Run("report_error_panics", func(t *testing.T) {
-		defer func() {
-			want := "cannot report error as channel has not exited IDLE state"
-			if r := recover(); r != want {
-				t.Errorf("expected panic %q, got %q", want, r)
-			}
-		}()
-		r.ReportError(errors.New("example"))
+		r.CC()
 	})
 
 	t.Run("happy_path", func(t *testing.T) {
@@ -70,6 +58,6 @@ func TestResolver(t *testing.T) {
 		r.UpdateState(resolver.State{Addresses: []resolver.Address{
 			{Addr: "ok"},
 		}})
-		r.ReportError(errors.New("example"))
+		r.CC().ReportError(errors.New("example"))
 	})
 }

resolver_test.go

@@ -180,7 +180,7 @@ func (s) TestResolverAddressesToEndpointsUsingNewAddresses(t *testing.T) {
 	}
 	cc.Connect()
 	defer cc.Close()
-	r.CC.NewAddress(addrs)
+	r.CC().NewAddress(addrs)
 
 	select {
 	case got := <-stateCh:

test/balancer_switching_test.go

@@ -510,7 +510,7 @@ func (s) TestBalancerSwitch_Graceful(t *testing.T) {
 	// report a ready picker until we ask it to do so.
 	r.UpdateState(resolver.State{
 		Addresses:     addrs[:1],
-		ServiceConfig: r.CC.ParseServiceConfig(fmt.Sprintf(`{"loadBalancingConfig": [{"%v": {}}]}`, t.Name())),
+		ServiceConfig: r.CC().ParseServiceConfig(fmt.Sprintf(`{"loadBalancingConfig": [{"%v": {}}]}`, t.Name())),
 	})
 	select {
 	case <-ctx.Done():

test/balancer_test.go

@@ -631,7 +631,7 @@ func (s) TestWaitForReady(t *testing.T) {
 	client := testgrpc.NewTestServiceClient(cc)
 
 	// Report an error so non-WFR RPCs will give up early.
-	r.CC.ReportError(errors.New("fake resolver error"))
+	r.CC().ReportError(errors.New("fake resolver error"))
 
 	// Ensure the client is not connected to anything and fails non-WFR RPCs.
 	if res, err := client.UnaryCall(ctx, &testpb.SimpleRequest{}); status.Code(err) != codes.Unavailable {

test/parse_config.go

@@ -30,7 +30,7 @@ import (
 func parseServiceConfig(t *testing.T, r *manual.Resolver, sc string) *serviceconfig.ParseResult {
 	t.Helper()
 
-	scpr := r.CC.ParseServiceConfig(sc)
+	scpr := r.CC().ParseServiceConfig(sc)
 	if scpr.Err != nil {
 		t.Fatalf("Failed to parse service config %q: %v", sc, scpr.Err)
 	}

test/resolver_update_test.go

@@ -119,7 +119,7 @@ func (s) TestResolverUpdate_InvalidServiceConfigAsFirstUpdate(t *testing.T) {
 	cc.Connect()
 	defer cc.Close()
 
-	scpr := r.CC.ParseServiceConfig("bad json service config")
+	scpr := r.CC().ParseServiceConfig("bad json service config")
 	r.UpdateState(resolver.State{ServiceConfig: scpr})
 
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
@@ -205,7 +205,7 @@ func (s) TestResolverUpdate_InvalidServiceConfigAfterGoodUpdate(t *testing.T) {
 	// Push a resolver update and verify that our balancer receives the update.
 	addrs := []resolver.Address{{Addr: backend.Address}}
 	const lbCfg = "wrapping balancer LB policy config"
-	goodSC := r.CC.ParseServiceConfig(fmt.Sprintf(`
+	goodSC := r.CC().ParseServiceConfig(fmt.Sprintf(`
 {
   "loadBalancingConfig": [
     {
@@ -244,7 +244,7 @@ func (s) TestResolverUpdate_InvalidServiceConfigAfterGoodUpdate(t *testing.T) {
 	// Push a bad resolver update and ensure that the update is propagated to our
 	// stub balancer. But since the pushed update contains an invalid service
 	// config, our balancer should continue to see the old loadBalancingConfig.
-	badSC := r.CC.ParseServiceConfig("bad json service config")
+	badSC := r.CC().ParseServiceConfig("bad json service config")
 	wantCCS.ResolverState.ServiceConfig = badSC
 	r.UpdateState(resolver.State{Addresses: addrs, ServiceConfig: badSC})
 	ccs, err = ccUpdateCh.Receive(ctx)