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

Code before:

	// Update load reporting config. This needs to be done before updating the
	// child policy because we need the loadStore from the updated client to be
	// passed to the ccWrapper, so that the next picker from the child policy
	// will pick up the new loadStore.
	if err := b.updateLoadStore(newConfig); err != nil {
		return err
	}

	if b.config == nil || b.config.ChildPolicy.Name != newConfig.ChildPolicy.Name {
		if b.config == nil || b.config.ChildPolicy.Name != newConfig.ChildPolicy.Name {
			strCfg, err := newConfig.ChildPolicy.MarshalJSON()
			if err != nil {
				return fmt.Errorf("error marshaling config: %v", err)
			}
			r := json.RawMessage(strCfg)
			sc, err := gracefulswitch.ParseConfig(r)
			if err != nil {
				return fmt.Errorf("error parsing child config: %v", err)
			}
			if err := b.child.UpdateClientConnState(balancer.ClientConnState{
				ResolverState:  s.ResolverState,
				BalancerConfig: sc,
			}); err != nil {
				return fmt.Errorf("error switching to child of type %q: %v", newConfig.ChildPolicy.Name, err)
			}
		}
	}
	b.config = newConfig

	// Notify run() of this new config, in case drop and request counter need
	// update (which means a new picker needs to be generated).
	b.pickerUpdateCh.Put(newConfig)

	// Addresses and sub-balancer config are sent to sub-balancer.
	return b.child.UpdateClientConnState(balancer.ClientConnState{
		ResolverState:  s.ResolverState,
		BalancerConfig: b.config.ChildPolicy.Config,
	})
}

func (b *clusterImplBalancer) ResolverError(err error) {
	if b.closed.HasFired() {
		b.logger.Warningf("xds: received resolver error {%+v} after clusterImplBalancer was closed", err)
		return
	}
	b.child.ResolverError(err)
}

func (b *clusterImplBalancer) updateSubConnState(sc balancer.SubConn, s balancer.SubConnState, cb func(balancer.SubConnState)) {
	if b.closed.HasFired() {

Review comment (suggestion):
And we probably don't need this here. We can do have the call to `b.child.UpdateClientConnState` in the last line of this method take care of pushing the configuration to the child.

Code after:
	if err != nil {
		return err
	}

	b.config = newConfig

	b.telemetryLabels = newConfig.TelemetryLabels
	dc := b.handleDropAndRequestCount(newConfig)
	if dc != nil && b.childState.Picker != nil {
		b.ClientConn.UpdateState(balancer.State{
			ConnectivityState: b.childState.ConnectivityState,
			Picker:            b.newPicker(dc),
		})
	}

	// Addresses and sub-balancer config are sent to sub-balancer.
	return b.child.UpdateClientConnState(balancer.ClientConnState{
		ResolverState:  s.ResolverState,
		BalancerConfig: parsedCfg,
	})
}

func (b *clusterImplBalancer) UpdateClientConnState(s balancer.ClientConnState) error {
	// Handle the update in a blocking fashion.
	errCh := make(chan error, 1)
	callback := func(context.Context) {
		errCh <- b.updateClientConnState(s)
	}
	onFailure := func() {
		// An attempt to schedule callback fails only when an update is received
		// after Close().
		errCh <- errBalancerClosed
	}
	b.serializer.ScheduleOr(callback, onFailure)
	return <-errCh
}

func (b *clusterImplBalancer) ResolverError(err error) {
	b.serializer.TrySchedule(func(context.Context) {
		b.child.ResolverError(err)
	})
}

func (b *clusterImplBalancer) updateSubConnState(_ balancer.SubConn, s balancer.SubConnState, cb func(balancer.SubConnState)) {
	// Trigger re-resolution when a SubConn turns transient failure. This is
	// necessary for the LogicalDNS in cluster_resolver policy to re-resolve.
	//
	// Note that this happens not only for the addresses from DNS, but also for
	// EDS (cluster_impl doesn't know if it's DNS or EDS, only the parent
	// knows). The parent priority policy is configured to ignore re-resolution
	// signal from the EDS children.