opensource/grpc-go
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

eds: move edsBalancer to the package edsbalancer (#3294)

Browse Source 
edsBalancer (the old xds balancer) was in `package balancer`, one level above the eds implementation. It's a thin wrapper of the eds impl (and fallback in the future).

This change moves the thin wrapper to `package edsbalancer`, and also renames some structs.
This commit is contained in:
Menghan Li
2020-01-03 11:08:28 -08:00
committed by GitHub
parent 1f28521e57
commit 287dc3e58a
15 changed files with 234 additions and 213 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
xds/experimental/xds_experimental.go
View File

@ -23,8 +23,7 @@
package experimental package experimental
import ( import (
_ "google.golang.org/grpc/xds/internal/balancer" // Register the xds_balancer _ "google.golang.org/grpc/xds/internal/balancer" // Register the balancers.
_ "google.golang.org/grpc/xds/internal/balancer/cdsbalancer" // Register the cds balancer _ "google.golang.org/grpc/xds/internal/resolver" // Register the xds_resolver
_ "google.golang.org/grpc/xds/internal/resolver" // Register the xds_resolver _ "google.golang.org/grpc/xds/internal/resolver/old" // Register the old xds_resolver
_ "google.golang.org/grpc/xds/internal/resolver/old" // Register the old xds_resolver
) )

24
xds/internal/balancer/balancer.go Normal file
View File

@ -0,0 +1,24 @@
/*
*
* Copyright 2020 gRPC authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package balancer
import (
_ "google.golang.org/grpc/xds/internal/balancer/cdsbalancer" // Register the CDS balancer
_ "google.golang.org/grpc/xds/internal/balancer/edsbalancer" // Register the EDS balancer
)

4
xds/internal/balancer/cdsbalancer/cdsbalancer.go
View File

@ -30,9 +30,9 @@ import (
"google.golang.org/grpc/internal/buffer" "google.golang.org/grpc/internal/buffer"
"google.golang.org/grpc/resolver" "google.golang.org/grpc/resolver"
"google.golang.org/grpc/serviceconfig" "google.golang.org/grpc/serviceconfig"
"google.golang.org/grpc/xds/internal/balancer/edsbalancer"
xdsinternal "google.golang.org/grpc/xds/internal" xdsinternal "google.golang.org/grpc/xds/internal"
xdsbalancer "google.golang.org/grpc/xds/internal/balancer"
xdsclient "google.golang.org/grpc/xds/internal/client" xdsclient "google.golang.org/grpc/xds/internal/client"
) )
@ -229,7 +229,7 @@ func (b *cdsBalancer) run() {
break break
} }
} }
lbCfg := &xdsbalancer.XDSConfig{EDSServiceName: update.cds.ServiceName} lbCfg := &edsbalancer.EDSConfig{EDSServiceName: update.cds.ServiceName}
if update.cds.EnableLRS { if update.cds.EnableLRS {
// An empty string here indicates that the edsBalancer // An empty string here indicates that the edsBalancer
// should use the same xDS server for load reporting as // should use the same xDS server for load reporting as

4
xds/internal/balancer/cdsbalancer/cdsbalancer_test.go
View File

@ -31,7 +31,7 @@ import (
"google.golang.org/grpc/resolver" "google.golang.org/grpc/resolver"
"google.golang.org/grpc/serviceconfig" "google.golang.org/grpc/serviceconfig"
xdsinternal "google.golang.org/grpc/xds/internal" xdsinternal "google.golang.org/grpc/xds/internal"
xdsbalancer "google.golang.org/grpc/xds/internal/balancer" "google.golang.org/grpc/xds/internal/balancer/edsbalancer"
xdsclient "google.golang.org/grpc/xds/internal/client" xdsclient "google.golang.org/grpc/xds/internal/client"
"google.golang.org/grpc/xds/internal/testutils" "google.golang.org/grpc/xds/internal/testutils"
"google.golang.org/grpc/xds/internal/testutils/fakeclient" "google.golang.org/grpc/xds/internal/testutils/fakeclient"
@ -194,7 +194,7 @@ func cdsCCS(cluster string, xdsClient interface{}) balancer.ClientConnState {
// edsCCS is a helper function to construct a good update passed from the // edsCCS is a helper function to construct a good update passed from the
// cdsBalancer to the edsBalancer. // cdsBalancer to the edsBalancer.
func edsCCS(service string, enableLRS bool, xdsClient interface{}) balancer.ClientConnState { func edsCCS(service string, enableLRS bool, xdsClient interface{}) balancer.ClientConnState {
lbCfg := &xdsbalancer.XDSConfig{EDSServiceName: service} lbCfg := &edsbalancer.EDSConfig{EDSServiceName: service}
if enableLRS { if enableLRS {
lbCfg.LrsLoadReportingServerName = new(string) lbCfg.LrsLoadReportingServerName = new(string)
} }

20
xds/internal/balancer/config.go → xds/internal/balancer/edsbalancer/config.go
View File

@ -15,7 +15,7 @@
* limitations under the License. * limitations under the License.
*/ */
package balancer package edsbalancer
import ( import (
"encoding/json" "encoding/json"
@ -25,9 +25,9 @@ import (
"google.golang.org/grpc/serviceconfig" "google.golang.org/grpc/serviceconfig"
) )
// XDSConfig represents the loadBalancingConfig section of the service config // EDSConfig represents the loadBalancingConfig section of the service config
// for xDS balancers. // for EDS balancers.
type XDSConfig struct { type EDSConfig struct {
serviceconfig.LoadBalancingConfig serviceconfig.LoadBalancingConfig
// BalancerName represents the load balancer to use. // BalancerName represents the load balancer to use.
BalancerName string BalancerName string
@ -46,10 +46,10 @@ type XDSConfig struct {
LrsLoadReportingServerName *string LrsLoadReportingServerName *string
} }
// xdsConfigJSON is the intermediate unmarshal result of XDSConfig. ChildPolicy // edsConfigJSON is the intermediate unmarshal result of EDSConfig. ChildPolicy
// and Fallbackspolicy are post-processed, and for each, the first installed // and Fallbackspolicy are post-processed, and for each, the first installed
// policy is kept. // policy is kept.
type xdsConfigJSON struct { type edsConfigJSON struct {
BalancerName string BalancerName string
ChildPolicy []*loadBalancingConfig ChildPolicy []*loadBalancingConfig
FallbackPolicy []*loadBalancingConfig FallbackPolicy []*loadBalancingConfig
@ -60,8 +60,8 @@ type xdsConfigJSON struct {
// UnmarshalJSON parses the JSON-encoded byte slice in data and stores it in l. // UnmarshalJSON parses the JSON-encoded byte slice in data and stores it in l.
// When unmarshalling, we iterate through the childPolicy/fallbackPolicy lists // When unmarshalling, we iterate through the childPolicy/fallbackPolicy lists
// and select the first LB policy which has been registered. // and select the first LB policy which has been registered.
func (l *XDSConfig) UnmarshalJSON(data []byte) error { func (l *EDSConfig) UnmarshalJSON(data []byte) error {
var configJSON xdsConfigJSON var configJSON edsConfigJSON
if err := json.Unmarshal(data, &configJSON); err != nil { if err := json.Unmarshal(data, &configJSON); err != nil {
return err return err
} }
@ -87,8 +87,8 @@ func (l *XDSConfig) UnmarshalJSON(data []byte) error {
} }
// MarshalJSON returns a JSON encoding of l. // MarshalJSON returns a JSON encoding of l.
func (l *XDSConfig) MarshalJSON() ([]byte, error) { func (l *EDSConfig) MarshalJSON() ([]byte, error) {
return nil, fmt.Errorf("XDSConfig.MarshalJSON() is unimplemented") return nil, fmt.Errorf("EDSConfig.MarshalJSON() is unimplemented")
} }
// loadBalancingConfig represents a single load balancing config, // loadBalancingConfig represents a single load balancing config,

51
xds/internal/balancer/xds.go → xds/internal/balancer/edsbalancer/eds.go
View File

@ -16,8 +16,8 @@
* *
*/ */
// Package balancer contains xds balancer implementation. // Package edsbalancer contains EDS balancer implementation.
package balancer package edsbalancer
import ( import (
"context" "context"
@ -32,7 +32,6 @@ import (
"google.golang.org/grpc/grpclog" "google.golang.org/grpc/grpclog"
"google.golang.org/grpc/resolver" "google.golang.org/grpc/resolver"
"google.golang.org/grpc/serviceconfig" "google.golang.org/grpc/serviceconfig"
"google.golang.org/grpc/xds/internal/balancer/edsbalancer"
"google.golang.org/grpc/xds/internal/balancer/lrs" "google.golang.org/grpc/xds/internal/balancer/lrs"
xdsclient "google.golang.org/grpc/xds/internal/client" xdsclient "google.golang.org/grpc/xds/internal/client"
) )
@ -43,8 +42,8 @@ const (
) )
var ( var (
newEDSBalancer = func(cc balancer.ClientConn, loadStore lrs.Store) edsBalancerInterface { newEDSBalancer = func(cc balancer.ClientConn, loadStore lrs.Store) edsBalancerImplInterface {
return edsbalancer.NewXDSBalancer(cc, loadStore) return newEDSBalancerImpl(cc, loadStore)
} }
) )
@ -66,7 +65,7 @@ func (b *edsBalancerBuilder) Build(cc balancer.ClientConn, opts balancer.BuildOp
xdsClientUpdate: make(chan interface{}), xdsClientUpdate: make(chan interface{}),
} }
loadStore := lrs.NewStore() loadStore := lrs.NewStore()
x.xdsLB = newEDSBalancer(x.cc, loadStore) x.edsImpl = newEDSBalancer(x.cc, loadStore)
x.client = newXDSClientWrapper(x.handleEDSUpdate, x.loseContact, x.buildOpts, loadStore) x.client = newXDSClientWrapper(x.handleEDSUpdate, x.loseContact, x.buildOpts, loadStore)
go x.run() go x.run()
return x return x
@ -77,18 +76,18 @@ func (b *edsBalancerBuilder) Name() string {
} }
func (b *edsBalancerBuilder) ParseConfig(c json.RawMessage) (serviceconfig.LoadBalancingConfig, error) { func (b *edsBalancerBuilder) ParseConfig(c json.RawMessage) (serviceconfig.LoadBalancingConfig, error) {
var cfg XDSConfig var cfg EDSConfig
if err := json.Unmarshal(c, &cfg); err != nil { if err := json.Unmarshal(c, &cfg); err != nil {
return nil, fmt.Errorf("unable to unmarshal balancer config %s into xds config, error: %v", string(c), err) return nil, fmt.Errorf("unable to unmarshal balancer config %s into EDSConfig, error: %v", string(c), err)
} }
return &cfg, nil return &cfg, nil
} }
// edsBalancerInterface defines the interface that edsBalancer must implement to // edsBalancerImplInterface defines the interface that edsBalancerImpl must
// communicate with edsBalancer. // implement to communicate with edsBalancer.
// //
// It's implemented by the real eds balancer and a fake testing eds balancer. // It's implemented by the real eds balancer and a fake testing eds balancer.
type edsBalancerInterface interface { type edsBalancerImplInterface interface {
// HandleEDSResponse passes the received EDS message from traffic director to eds balancer. // HandleEDSResponse passes the received EDS message from traffic director to eds balancer.
HandleEDSResponse(edsResp *xdsclient.EDSUpdate) HandleEDSResponse(edsResp *xdsclient.EDSUpdate)
// HandleChildPolicy updates the eds balancer the intra-cluster load balancing policy to use. // HandleChildPolicy updates the eds balancer the intra-cluster load balancing policy to use.
@ -101,8 +100,8 @@ type edsBalancerInterface interface {
var _ balancer.V2Balancer = (*edsBalancer)(nil) // Assert that we implement V2Balancer var _ balancer.V2Balancer = (*edsBalancer)(nil) // Assert that we implement V2Balancer
// edsBalancer manages xdsClient and the actual balancer that does load // edsBalancer manages xdsClient and the actual EDS balancer implementation that
// balancing. // does load balancing.
// //
// It currently has only an edsBalancer. Later, we may add fallback. // It currently has only an edsBalancer. Later, we may add fallback.
type edsBalancer struct { type edsBalancer struct {
@ -115,9 +114,9 @@ type edsBalancer struct {
grpcUpdate chan interface{} grpcUpdate chan interface{}
xdsClientUpdate chan interface{} xdsClientUpdate chan interface{}
client *xdsclientWrapper // may change when passed a different service config client *xdsclientWrapper // may change when passed a different service config
config *XDSConfig // may change when passed a different service config config *EDSConfig // may change when passed a different service config
xdsLB edsBalancerInterface edsImpl edsBalancerImplInterface
} }
// run gets executed in a goroutine once edsBalancer is created. It monitors updates from grpc, // run gets executed in a goroutine once edsBalancer is created. It monitors updates from grpc,
@ -134,8 +133,8 @@ func (x *edsBalancer) run() {
if x.client != nil { if x.client != nil {
x.client.close() x.client.close()
} }
if x.xdsLB != nil { if x.edsImpl != nil {
x.xdsLB.Close() x.edsImpl.Close()
} }
return return
} }
@ -145,11 +144,11 @@ func (x *edsBalancer) run() {
func (x *edsBalancer) handleGRPCUpdate(update interface{}) { func (x *edsBalancer) handleGRPCUpdate(update interface{}) {
switch u := update.(type) { switch u := update.(type) {
case *subConnStateUpdate: case *subConnStateUpdate:
if x.xdsLB != nil { if x.edsImpl != nil {
x.xdsLB.HandleSubConnStateChange(u.sc, u.state.ConnectivityState) x.edsImpl.HandleSubConnStateChange(u.sc, u.state.ConnectivityState)
} }
case *balancer.ClientConnState: case *balancer.ClientConnState:
cfg, _ := u.BalancerConfig.(*XDSConfig) cfg, _ := u.BalancerConfig.(*EDSConfig)
if cfg == nil { if cfg == nil {
// service config parsing failed. should never happen. // service config parsing failed. should never happen.
return return
@ -162,13 +161,13 @@ func (x *edsBalancer) handleGRPCUpdate(update interface{}) {
return return
} }
// We will update the xdsLB with the new child policy, if we got a // We will update the edsImpl with the new child policy, if we got a
// different one. // different one.
if x.xdsLB != nil && !reflect.DeepEqual(cfg.ChildPolicy, x.config.ChildPolicy) { if x.edsImpl != nil && !reflect.DeepEqual(cfg.ChildPolicy, x.config.ChildPolicy) {
if cfg.ChildPolicy != nil { if cfg.ChildPolicy != nil {
x.xdsLB.HandleChildPolicy(cfg.ChildPolicy.Name, cfg.ChildPolicy.Config) x.edsImpl.HandleChildPolicy(cfg.ChildPolicy.Name, cfg.ChildPolicy.Config)
} else { } else {
x.xdsLB.HandleChildPolicy(roundrobin.Name, nil) x.edsImpl.HandleChildPolicy(roundrobin.Name, nil)
} }
} }
@ -184,7 +183,7 @@ func (x *edsBalancer) handleXDSClientUpdate(update interface{}) {
// TODO: this func should accept (*xdsclient.EDSUpdate, error), and process // TODO: this func should accept (*xdsclient.EDSUpdate, error), and process
// the error, instead of having a separate loseContact signal. // the error, instead of having a separate loseContact signal.
case *xdsclient.EDSUpdate: case *xdsclient.EDSUpdate:
x.xdsLB.HandleEDSResponse(u) x.edsImpl.HandleEDSResponse(u)
case *loseContact: case *loseContact:
// loseContact can be useful for going into fallback. // loseContact can be useful for going into fallback.
default: default:

125
xds/internal/balancer/edsbalancer/edsbalancer.go → xds/internal/balancer/edsbalancer/eds_impl.go
View File

@ -14,7 +14,6 @@
* limitations under the License. * limitations under the License.
*/ */
// Package edsbalancer implements a balancer to handle EDS responses.
package edsbalancer package edsbalancer
import ( import (
@ -51,13 +50,13 @@ type balancerGroupWithConfig struct {
configs map[internal.Locality]*localityConfig configs map[internal.Locality]*localityConfig
} }
// EDSBalancer does load balancing based on the EDS responses. Note that it // edsBalancerImpl does load balancing based on the EDS responses. Note that it
// doesn't implement the balancer interface. It's intended to be used by a high // doesn't implement the balancer interface. It's intended to be used by a high
// level balancer implementation. // level balancer implementation.
// //
// The localities are picked as weighted round robin. A configurable child // The localities are picked as weighted round robin. A configurable child
// policy is used to manage endpoints in each locality. // policy is used to manage endpoints in each locality.
type EDSBalancer struct { type edsBalancerImpl struct {
cc balancer.ClientConn cc balancer.ClientConn
subBalancerBuilder balancer.Builder subBalancerBuilder balancer.Builder
@ -89,9 +88,9 @@ type EDSBalancer struct {
innerState balancer.State // The state of the picker without drop support. innerState balancer.State // The state of the picker without drop support.
} }
// NewXDSBalancer create a new EDSBalancer. // newEDSBalancerImpl create a new edsBalancerImpl.
func NewXDSBalancer(cc balancer.ClientConn, loadStore lrs.Store) *EDSBalancer { func newEDSBalancerImpl(cc balancer.ClientConn, loadStore lrs.Store) *edsBalancerImpl {
xdsB := &EDSBalancer{ edsImpl := &edsBalancerImpl{
cc: cc, cc: cc,
subBalancerBuilder: balancer.Get(roundrobin.Name), subBalancerBuilder: balancer.Get(roundrobin.Name),
@ -104,7 +103,7 @@ func NewXDSBalancer(cc balancer.ClientConn, loadStore lrs.Store) *EDSBalancer {
// response. Otherwise the balancer group will be started with round-robin, // response. Otherwise the balancer group will be started with round-robin,
// and if users specify a different sub-balancer, all balancers in balancer // and if users specify a different sub-balancer, all balancers in balancer
// group will be closed and recreated when sub-balancer update happens. // group will be closed and recreated when sub-balancer update happens.
return xdsB return edsImpl
} }
// HandleChildPolicy updates the child balancers handling endpoints. Child // HandleChildPolicy updates the child balancers handling endpoints. Child
@ -112,17 +111,17 @@ func NewXDSBalancer(cc balancer.ClientConn, loadStore lrs.Store) *EDSBalancer {
// the old child balancer will be used. // the old child balancer will be used.
// //
// HandleChildPolicy and HandleEDSResponse must be called by the same goroutine. // HandleChildPolicy and HandleEDSResponse must be called by the same goroutine.
func (xdsB *EDSBalancer) HandleChildPolicy(name string, config json.RawMessage) { func (edsImpl *edsBalancerImpl) HandleChildPolicy(name string, config json.RawMessage) {
if xdsB.subBalancerBuilder.Name() == name { if edsImpl.subBalancerBuilder.Name() == name {
return return
} }
newSubBalancerBuilder := balancer.Get(name) newSubBalancerBuilder := balancer.Get(name)
if newSubBalancerBuilder == nil { if newSubBalancerBuilder == nil {
grpclog.Infof("EDSBalancer: failed to find balancer with name %q, keep using %q", name, xdsB.subBalancerBuilder.Name()) grpclog.Infof("edsBalancerImpl: failed to find balancer with name %q, keep using %q", name, edsImpl.subBalancerBuilder.Name())
return return
} }
xdsB.subBalancerBuilder = newSubBalancerBuilder edsImpl.subBalancerBuilder = newSubBalancerBuilder
for _, bgwc := range xdsB.priorityToLocalities { for _, bgwc := range edsImpl.priorityToLocalities {
if bgwc == nil { if bgwc == nil {
continue continue
} }
@ -131,7 +130,7 @@ func (xdsB *EDSBalancer) HandleChildPolicy(name string, config json.RawMessage)
// switching sub-balancers (keep old balancer around until new // switching sub-balancers (keep old balancer around until new
// balancer becomes ready). // balancer becomes ready).
bgwc.bg.remove(id) bgwc.bg.remove(id)
bgwc.bg.add(id, config.weight, xdsB.subBalancerBuilder) bgwc.bg.add(id, config.weight, edsImpl.subBalancerBuilder)
bgwc.bg.handleResolvedAddrs(id, config.addrs) bgwc.bg.handleResolvedAddrs(id, config.addrs)
} }
} }
@ -139,7 +138,7 @@ func (xdsB *EDSBalancer) HandleChildPolicy(name string, config json.RawMessage)
// updateDrops compares new drop policies with the old. If they are different, // updateDrops compares new drop policies with the old. If they are different,
// it updates the drop policies and send ClientConn an updated picker. // it updates the drop policies and send ClientConn an updated picker.
func (xdsB *EDSBalancer) updateDrops(dropPolicies []xdsclient.OverloadDropConfig) { func (edsImpl *edsBalancerImpl) updateDrops(dropPolicies []xdsclient.OverloadDropConfig) {
var ( var (
newDrops []*dropper newDrops []*dropper
dropsChanged bool dropsChanged bool
@ -151,30 +150,30 @@ func (xdsB *EDSBalancer) updateDrops(dropPolicies []xdsclient.OverloadDropConfig
) )
newDrops = append(newDrops, newDropper(numerator, denominator, dropPolicy.Category)) newDrops = append(newDrops, newDropper(numerator, denominator, dropPolicy.Category))
// The following reading xdsB.drops doesn't need mutex because it can only // The following reading edsImpl.drops doesn't need mutex because it can only
// be updated by the code following. // be updated by the code following.
if dropsChanged { if dropsChanged {
continue continue
} }
if i >= len(xdsB.drops) { if i >= len(edsImpl.drops) {
dropsChanged = true dropsChanged = true
continue continue
} }
if oldDrop := xdsB.drops[i]; numerator != oldDrop.numerator || denominator != oldDrop.denominator { if oldDrop := edsImpl.drops[i]; numerator != oldDrop.numerator || denominator != oldDrop.denominator {
dropsChanged = true dropsChanged = true
} }
} }
if dropsChanged { if dropsChanged {
xdsB.pickerMu.Lock() edsImpl.pickerMu.Lock()
xdsB.drops = newDrops edsImpl.drops = newDrops
if xdsB.innerState.Picker != nil { if edsImpl.innerState.Picker != nil {
// Update picker with old inner picker, new drops. // Update picker with old inner picker, new drops.
xdsB.cc.UpdateState(balancer.State{ edsImpl.cc.UpdateState(balancer.State{
ConnectivityState: xdsB.innerState.ConnectivityState, ConnectivityState: edsImpl.innerState.ConnectivityState,
Picker: newDropPicker(xdsB.innerState.Picker, newDrops, xdsB.loadStore)}, Picker: newDropPicker(edsImpl.innerState.Picker, newDrops, edsImpl.loadStore)},
) )
} }
xdsB.pickerMu.Unlock() edsImpl.pickerMu.Unlock()
} }
} }
@ -182,7 +181,7 @@ func (xdsB *EDSBalancer) updateDrops(dropPolicies []xdsclient.OverloadDropConfig
// SubConns. It also handles drops. // SubConns. It also handles drops.
// //
// HandleChildPolicy and HandleEDSResponse must be called by the same goroutine. // HandleChildPolicy and HandleEDSResponse must be called by the same goroutine.
func (xdsB *EDSBalancer) HandleEDSResponse(edsResp *xdsclient.EDSUpdate) { func (edsImpl *edsBalancerImpl) HandleEDSResponse(edsResp *xdsclient.EDSUpdate) {
// TODO: Unhandled fields from EDS response: // TODO: Unhandled fields from EDS response:
// - edsResp.GetPolicy().GetOverprovisioningFactor() // - edsResp.GetPolicy().GetOverprovisioningFactor()
// - locality.GetPriority() // - locality.GetPriority()
@ -192,7 +191,7 @@ func (xdsB *EDSBalancer) HandleEDSResponse(edsResp *xdsclient.EDSUpdate) {
// - socketAddress.GetNamedPort(), socketAddress.GetResolverName() // - socketAddress.GetNamedPort(), socketAddress.GetResolverName()
// - resolve endpoint's name with another resolver // - resolve endpoint's name with another resolver
xdsB.updateDrops(edsResp.Drops) edsImpl.updateDrops(edsResp.Drops)
// Filter out all localities with weight 0. // Filter out all localities with weight 0.
// //
@ -222,7 +221,7 @@ func (xdsB *EDSBalancer) HandleEDSResponse(edsResp *xdsclient.EDSUpdate) {
priorityLowest = priority priorityLowest = priority
} }
bgwc, ok := xdsB.priorityToLocalities[priority] bgwc, ok := edsImpl.priorityToLocalities[priority]
if !ok { if !ok {
// Create balancer group if it's never created (this is the first // Create balancer group if it's never created (this is the first
// time this priority is received). We don't start it here. It may // time this priority is received). We don't start it here. It may
@ -230,24 +229,24 @@ func (xdsB *EDSBalancer) HandleEDSResponse(edsResp *xdsclient.EDSUpdate) {
// new lowest priority). // new lowest priority).
bgwc = &balancerGroupWithConfig{ bgwc = &balancerGroupWithConfig{
bg: newBalancerGroup( bg: newBalancerGroup(
xdsB.ccWrapperWithPriority(priority), xdsB.loadStore, edsImpl.ccWrapperWithPriority(priority), edsImpl.loadStore,
), ),
configs: make(map[internal.Locality]*localityConfig), configs: make(map[internal.Locality]*localityConfig),
} }
xdsB.priorityToLocalities[priority] = bgwc edsImpl.priorityToLocalities[priority] = bgwc
priorityChanged = true priorityChanged = true
} }
xdsB.handleEDSResponsePerPriority(bgwc, newLocalities) edsImpl.handleEDSResponsePerPriority(bgwc, newLocalities)
} }
xdsB.priorityLowest = priorityLowest edsImpl.priorityLowest = priorityLowest
// Delete priorities that are removed in the latest response, and also close // Delete priorities that are removed in the latest response, and also close
// the balancer group. // the balancer group.
for p, bgwc := range xdsB.priorityToLocalities { for p, bgwc := range edsImpl.priorityToLocalities {
if _, ok := newLocalitiesWithPriority[p]; !ok { if _, ok := newLocalitiesWithPriority[p]; !ok {
delete(xdsB.priorityToLocalities, p) delete(edsImpl.priorityToLocalities, p)
bgwc.bg.close() bgwc.bg.close()
delete(xdsB.priorityToState, p) delete(edsImpl.priorityToState, p)
priorityChanged = true priorityChanged = true
} }
} }
@ -256,11 +255,11 @@ func (xdsB *EDSBalancer) HandleEDSResponse(edsResp *xdsclient.EDSUpdate) {
// E.g. priorityInUse was removed, or all priorities are down, and a new // E.g. priorityInUse was removed, or all priorities are down, and a new
// lower priority was added. // lower priority was added.
if priorityChanged { if priorityChanged {
xdsB.handlePriorityChange() edsImpl.handlePriorityChange()
} }
} }
func (xdsB *EDSBalancer) handleEDSResponsePerPriority(bgwc *balancerGroupWithConfig, newLocalities []xdsclient.Locality) { func (edsImpl *edsBalancerImpl) handleEDSResponsePerPriority(bgwc *balancerGroupWithConfig, newLocalities []xdsclient.Locality) {
// newLocalitiesSet contains all names of localities in the new EDS response // newLocalitiesSet contains all names of localities in the new EDS response
// for the same priority. It's used to delete localities that are removed in // for the same priority. It's used to delete localities that are removed in
// the new EDS response. // the new EDS response.
@ -285,7 +284,7 @@ func (xdsB *EDSBalancer) handleEDSResponsePerPriority(bgwc *balancerGroupWithCon
address := resolver.Address{ address := resolver.Address{
Addr: lbEndpoint.Address, Addr: lbEndpoint.Address,
} }
if xdsB.subBalancerBuilder.Name() == weightedroundrobin.Name && lbEndpoint.Weight != 0 { if edsImpl.subBalancerBuilder.Name() == weightedroundrobin.Name && lbEndpoint.Weight != 0 {
address.Metadata = &weightedroundrobin.AddrInfo{ address.Metadata = &weightedroundrobin.AddrInfo{
Weight: lbEndpoint.Weight, Weight: lbEndpoint.Weight,
} }
@ -296,7 +295,7 @@ func (xdsB *EDSBalancer) handleEDSResponsePerPriority(bgwc *balancerGroupWithCon
config, ok := bgwc.configs[lid] config, ok := bgwc.configs[lid]
if !ok { if !ok {
// A new balancer, add it to balancer group and balancer map. // A new balancer, add it to balancer group and balancer map.
bgwc.bg.add(lid, newWeight, xdsB.subBalancerBuilder) bgwc.bg.add(lid, newWeight, edsImpl.subBalancerBuilder)
config = &localityConfig{ config = &localityConfig{
weight: newWeight, weight: newWeight,
} }
@ -336,19 +335,19 @@ func (xdsB *EDSBalancer) handleEDSResponsePerPriority(bgwc *balancerGroupWithCon
} }
// HandleSubConnStateChange handles the state change and update pickers accordingly. // HandleSubConnStateChange handles the state change and update pickers accordingly.
func (xdsB *EDSBalancer) HandleSubConnStateChange(sc balancer.SubConn, s connectivity.State) { func (edsImpl *edsBalancerImpl) HandleSubConnStateChange(sc balancer.SubConn, s connectivity.State) {
xdsB.subConnMu.Lock() edsImpl.subConnMu.Lock()
var bgwc *balancerGroupWithConfig var bgwc *balancerGroupWithConfig
if p, ok := xdsB.subConnToPriority[sc]; ok { if p, ok := edsImpl.subConnToPriority[sc]; ok {
if s == connectivity.Shutdown { if s == connectivity.Shutdown {
// Only delete sc from the map when state changed to Shutdown. // Only delete sc from the map when state changed to Shutdown.
delete(xdsB.subConnToPriority, sc) delete(edsImpl.subConnToPriority, sc)
} }
bgwc = xdsB.priorityToLocalities[p] bgwc = edsImpl.priorityToLocalities[p]
} }
xdsB.subConnMu.Unlock() edsImpl.subConnMu.Unlock()
if bgwc == nil { if bgwc == nil {
grpclog.Infof("EDSBalancer: priority not found for sc state change") grpclog.Infof("edsBalancerImpl: priority not found for sc state change")
return return
} }
if bg := bgwc.bg; bg != nil { if bg := bgwc.bg; bg != nil {
@ -358,27 +357,27 @@ func (xdsB *EDSBalancer) HandleSubConnStateChange(sc balancer.SubConn, s connect
// updateState first handles priority, and then wraps picker in a drop picker // updateState first handles priority, and then wraps picker in a drop picker
// before forwarding the update. // before forwarding the update.
func (xdsB *EDSBalancer) updateState(priority priorityType, s balancer.State) { func (edsImpl *edsBalancerImpl) updateState(priority priorityType, s balancer.State) {
_, ok := xdsB.priorityToLocalities[priority] _, ok := edsImpl.priorityToLocalities[priority]
if !ok { if !ok {
grpclog.Infof("eds: received picker update from unknown priority") grpclog.Infof("eds: received picker update from unknown priority")
return return
} }
if xdsB.handlePriorityWithNewState(priority, s) { if edsImpl.handlePriorityWithNewState(priority, s) {
xdsB.pickerMu.Lock() edsImpl.pickerMu.Lock()
defer xdsB.pickerMu.Unlock() defer edsImpl.pickerMu.Unlock()
xdsB.innerState = s edsImpl.innerState = s
// Don't reset drops when it's a state change. // Don't reset drops when it's a state change.
xdsB.cc.UpdateState(balancer.State{ConnectivityState: s.ConnectivityState, Picker: newDropPicker(s.Picker, xdsB.drops, xdsB.loadStore)}) edsImpl.cc.UpdateState(balancer.State{ConnectivityState: s.ConnectivityState, Picker: newDropPicker(s.Picker, edsImpl.drops, edsImpl.loadStore)})
} }
} }
func (xdsB *EDSBalancer) ccWrapperWithPriority(priority priorityType) *edsBalancerWrapperCC { func (edsImpl *edsBalancerImpl) ccWrapperWithPriority(priority priorityType) *edsBalancerWrapperCC {
return &edsBalancerWrapperCC{ return &edsBalancerWrapperCC{
ClientConn: xdsB.cc, ClientConn: edsImpl.cc,
priority: priority, priority: priority,
parent: xdsB, parent: edsImpl,
} }
} }
@ -387,7 +386,7 @@ func (xdsB *EDSBalancer) ccWrapperWithPriority(priority priorityType) *edsBalanc
type edsBalancerWrapperCC struct { type edsBalancerWrapperCC struct {
balancer.ClientConn balancer.ClientConn
priority priorityType priority priorityType
parent *EDSBalancer parent *edsBalancerImpl
} }
func (ebwcc *edsBalancerWrapperCC) NewSubConn(addrs []resolver.Address, opts balancer.NewSubConnOptions) (balancer.SubConn, error) { func (ebwcc *edsBalancerWrapperCC) NewSubConn(addrs []resolver.Address, opts balancer.NewSubConnOptions) (balancer.SubConn, error) {
@ -400,20 +399,20 @@ func (ebwcc *edsBalancerWrapperCC) UpdateState(state balancer.State) {
ebwcc.parent.updateState(ebwcc.priority, state) ebwcc.parent.updateState(ebwcc.priority, state)
} }
func (xdsB *EDSBalancer) newSubConn(priority priorityType, addrs []resolver.Address, opts balancer.NewSubConnOptions) (balancer.SubConn, error) { func (edsImpl *edsBalancerImpl) newSubConn(priority priorityType, addrs []resolver.Address, opts balancer.NewSubConnOptions) (balancer.SubConn, error) {
sc, err := xdsB.cc.NewSubConn(addrs, opts) sc, err := edsImpl.cc.NewSubConn(addrs, opts)
if err != nil { if err != nil {
return nil, err return nil, err
} }
xdsB.subConnMu.Lock() edsImpl.subConnMu.Lock()
xdsB.subConnToPriority[sc] = priority edsImpl.subConnToPriority[sc] = priority
xdsB.subConnMu.Unlock() edsImpl.subConnMu.Unlock()
return sc, nil return sc, nil
} }
// Close closes the balancer. // Close closes the balancer.
func (xdsB *EDSBalancer) Close() { func (edsImpl *edsBalancerImpl) Close() {
for _, bgwc := range xdsB.priorityToLocalities { for _, bgwc := range edsImpl.priorityToLocalities {
if bg := bgwc.bg; bg != nil { if bg := bgwc.bg; bg != nil {
bg.close() bg.close()
} }

112
xds/internal/balancer/edsbalancer/priority.go → xds/internal/balancer/edsbalancer/eds_impl_priority.go
View File

@ -38,28 +38,28 @@ import (
// - If priorityInUse has a non-Ready state, and also there's a priority lower // - If priorityInUse has a non-Ready state, and also there's a priority lower
// than priorityInUse (which means a lower priority was added), set the next // than priorityInUse (which means a lower priority was added), set the next
// priority as new priorityInUse, and start the bg. // priority as new priorityInUse, and start the bg.
func (xdsB *EDSBalancer) handlePriorityChange() { func (edsImpl *edsBalancerImpl) handlePriorityChange() {
xdsB.priorityMu.Lock() edsImpl.priorityMu.Lock()
defer xdsB.priorityMu.Unlock() defer edsImpl.priorityMu.Unlock()
// Everything was removed by EDS. // Everything was removed by EDS.
if !xdsB.priorityLowest.isSet() { if !edsImpl.priorityLowest.isSet() {
xdsB.priorityInUse = newPriorityTypeUnset() edsImpl.priorityInUse = newPriorityTypeUnset()
xdsB.cc.UpdateState(balancer.State{ConnectivityState: connectivity.TransientFailure, Picker: base.NewErrPickerV2(balancer.ErrTransientFailure)}) edsImpl.cc.UpdateState(balancer.State{ConnectivityState: connectivity.TransientFailure, Picker: base.NewErrPickerV2(balancer.ErrTransientFailure)})
return return
} }
// priorityInUse wasn't set, use 0. // priorityInUse wasn't set, use 0.
if !xdsB.priorityInUse.isSet() { if !edsImpl.priorityInUse.isSet() {
xdsB.startPriority(newPriorityType(0)) edsImpl.startPriority(newPriorityType(0))
return return
} }
// priorityInUse was deleted, use the new lowest. // priorityInUse was deleted, use the new lowest.
if _, ok := xdsB.priorityToLocalities[xdsB.priorityInUse]; !ok { if _, ok := edsImpl.priorityToLocalities[edsImpl.priorityInUse]; !ok {
xdsB.priorityInUse = xdsB.priorityLowest edsImpl.priorityInUse = edsImpl.priorityLowest
if s, ok := xdsB.priorityToState[xdsB.priorityLowest]; ok { if s, ok := edsImpl.priorityToState[edsImpl.priorityLowest]; ok {
xdsB.cc.UpdateState(*s) edsImpl.cc.UpdateState(*s)
} else { } else {
// If state for priorityLowest is not found, this means priorityLowest was // If state for priorityLowest is not found, this means priorityLowest was
// started, but never sent any update. The init timer fired and // started, but never sent any update. The init timer fired and
@ -69,16 +69,16 @@ func (xdsB *EDSBalancer) handlePriorityChange() {
// We don't have an old state to send to parent, but we also don't // We don't have an old state to send to parent, but we also don't
// want parent to keep using picker from old_priorityInUse. Send an // want parent to keep using picker from old_priorityInUse. Send an
// update to trigger block picks until a new picker is ready. // update to trigger block picks until a new picker is ready.
xdsB.cc.UpdateState(balancer.State{ConnectivityState: connectivity.Connecting, Picker: base.NewErrPickerV2(balancer.ErrNoSubConnAvailable)}) edsImpl.cc.UpdateState(balancer.State{ConnectivityState: connectivity.Connecting, Picker: base.NewErrPickerV2(balancer.ErrNoSubConnAvailable)})
} }
return return
} }
// priorityInUse is not ready, look for next priority, and use if found. // priorityInUse is not ready, look for next priority, and use if found.
if s, ok := xdsB.priorityToState[xdsB.priorityInUse]; ok && s.ConnectivityState != connectivity.Ready { if s, ok := edsImpl.priorityToState[edsImpl.priorityInUse]; ok && s.ConnectivityState != connectivity.Ready {
pNext := xdsB.priorityInUse.nextLower() pNext := edsImpl.priorityInUse.nextLower()
if _, ok := xdsB.priorityToLocalities[pNext]; ok { if _, ok := edsImpl.priorityToLocalities[pNext]; ok {
xdsB.startPriority(pNext) edsImpl.startPriority(pNext)
} }
} }
} }
@ -86,11 +86,11 @@ func (xdsB *EDSBalancer) handlePriorityChange() {
// startPriority sets priorityInUse to p, and starts the balancer group for p. // startPriority sets priorityInUse to p, and starts the balancer group for p.
// It also starts a timer to fall to next priority after timeout. // It also starts a timer to fall to next priority after timeout.
// //
// Caller must hold priorityMu, priority must exist, and xdsB.priorityInUse must // Caller must hold priorityMu, priority must exist, and edsImpl.priorityInUse
// be non-nil. // must be non-nil.
func (xdsB *EDSBalancer) startPriority(priority priorityType) { func (edsImpl *edsBalancerImpl) startPriority(priority priorityType) {
xdsB.priorityInUse = priority edsImpl.priorityInUse = priority
p := xdsB.priorityToLocalities[priority] p := edsImpl.priorityToLocalities[priority]
// NOTE: this will eventually send addresses to sub-balancers. If the // NOTE: this will eventually send addresses to sub-balancers. If the
// sub-balancer tries to update picker, it will result in a deadlock on // sub-balancer tries to update picker, it will result in a deadlock on
// priorityMu. But it's not an expected behavior for the balancer to // priorityMu. But it's not an expected behavior for the balancer to
@ -103,52 +103,52 @@ func (xdsB *EDSBalancer) startPriority(priority priorityType) {
// //
// In all the cases, the existing init timer is either closed, also already // In all the cases, the existing init timer is either closed, also already
// expired. There's no need to close the old timer. // expired. There's no need to close the old timer.
xdsB.priorityInitTimer = time.AfterFunc(defaultPriorityInitTimeout, func() { edsImpl.priorityInitTimer = time.AfterFunc(defaultPriorityInitTimeout, func() {
xdsB.priorityMu.Lock() edsImpl.priorityMu.Lock()
defer xdsB.priorityMu.Unlock() defer edsImpl.priorityMu.Unlock()
if !xdsB.priorityInUse.equal(priority) { if !edsImpl.priorityInUse.equal(priority) {
return return
} }
xdsB.priorityInitTimer = nil edsImpl.priorityInitTimer = nil
pNext := priority.nextLower() pNext := priority.nextLower()
if _, ok := xdsB.priorityToLocalities[pNext]; ok { if _, ok := edsImpl.priorityToLocalities[pNext]; ok {
xdsB.startPriority(pNext) edsImpl.startPriority(pNext)
} }
}) })
} }
// handlePriorityWithNewState start/close priorities based on the connectivity // handlePriorityWithNewState start/close priorities based on the connectivity
// state. It returns whether the state should be forwarded to parent ClientConn. // state. It returns whether the state should be forwarded to parent ClientConn.
func (xdsB *EDSBalancer) handlePriorityWithNewState(priority priorityType, s balancer.State) bool { func (edsImpl *edsBalancerImpl) handlePriorityWithNewState(priority priorityType, s balancer.State) bool {
xdsB.priorityMu.Lock() edsImpl.priorityMu.Lock()
defer xdsB.priorityMu.Unlock() defer edsImpl.priorityMu.Unlock()
if !xdsB.priorityInUse.isSet() { if !edsImpl.priorityInUse.isSet() {
grpclog.Infof("eds: received picker update when no priority is in use (EDS returned an empty list)") grpclog.Infof("eds: received picker update when no priority is in use (EDS returned an empty list)")
return false return false
} }
if xdsB.priorityInUse.higherThan(priority) { if edsImpl.priorityInUse.higherThan(priority) {
// Lower priorities should all be closed, this is an unexpected update. // Lower priorities should all be closed, this is an unexpected update.
grpclog.Infof("eds: received picker update from priority lower then priorityInUse") grpclog.Infof("eds: received picker update from priority lower then priorityInUse")
return false return false
} }
bState, ok := xdsB.priorityToState[priority] bState, ok := edsImpl.priorityToState[priority]
if !ok { if !ok {
bState = &balancer.State{} bState = &balancer.State{}
xdsB.priorityToState[priority] = bState edsImpl.priorityToState[priority] = bState
} }
oldState := bState.ConnectivityState oldState := bState.ConnectivityState
*bState = s *bState = s
switch s.ConnectivityState { switch s.ConnectivityState {
case connectivity.Ready: case connectivity.Ready:
return xdsB.handlePriorityWithNewStateReady(priority) return edsImpl.handlePriorityWithNewStateReady(priority)
case connectivity.TransientFailure: case connectivity.TransientFailure:
return xdsB.handlePriorityWithNewStateTransientFailure(priority) return edsImpl.handlePriorityWithNewStateTransientFailure(priority)
case connectivity.Connecting: case connectivity.Connecting:
return xdsB.handlePriorityWithNewStateConnecting(priority, oldState) return edsImpl.handlePriorityWithNewStateConnecting(priority, oldState)
default: default:
// New state is Idle, should never happen. Don't forward. // New state is Idle, should never happen. Don't forward.
return false return false
@ -169,19 +169,19 @@ func (xdsB *EDSBalancer) handlePriorityWithNewState(priority priorityType, s bal
// Caller must make sure priorityInUse is not higher than priority. // Caller must make sure priorityInUse is not higher than priority.
// //
// Caller must hold priorityMu. // Caller must hold priorityMu.
func (xdsB *EDSBalancer) handlePriorityWithNewStateReady(priority priorityType) bool { func (edsImpl *edsBalancerImpl) handlePriorityWithNewStateReady(priority priorityType) bool {
// If one priority higher or equal to priorityInUse goes Ready, stop the // If one priority higher or equal to priorityInUse goes Ready, stop the
// init timer. If update is from higher than priorityInUse, // init timer. If update is from higher than priorityInUse,
// priorityInUse will be closed, and the init timer will become useless. // priorityInUse will be closed, and the init timer will become useless.
if timer := xdsB.priorityInitTimer; timer != nil { if timer := edsImpl.priorityInitTimer; timer != nil {
timer.Stop() timer.Stop()
xdsB.priorityInitTimer = nil edsImpl.priorityInitTimer = nil
} }
if xdsB.priorityInUse.lowerThan(priority) { if edsImpl.priorityInUse.lowerThan(priority) {
xdsB.priorityInUse = priority edsImpl.priorityInUse = priority
for i := priority.nextLower(); !i.lowerThan(xdsB.priorityLowest); i = i.nextLower() { for i := priority.nextLower(); !i.lowerThan(edsImpl.priorityLowest); i = i.nextLower() {
xdsB.priorityToLocalities[i].bg.close() edsImpl.priorityToLocalities[i].bg.close()
} }
return true return true
} }
@ -205,20 +205,20 @@ func (xdsB *EDSBalancer) handlePriorityWithNewStateReady(priority priorityType)
// Caller must make sure priorityInUse is not higher than priority. // Caller must make sure priorityInUse is not higher than priority.
// //
// Caller must hold priorityMu. // Caller must hold priorityMu.
func (xdsB *EDSBalancer) handlePriorityWithNewStateTransientFailure(priority priorityType) bool { func (edsImpl *edsBalancerImpl) handlePriorityWithNewStateTransientFailure(priority priorityType) bool {
if xdsB.priorityInUse.lowerThan(priority) { if edsImpl.priorityInUse.lowerThan(priority) {
return false return false
} }
// priorityInUse sends a failure. Stop its init timer. // priorityInUse sends a failure. Stop its init timer.
if timer := xdsB.priorityInitTimer; timer != nil { if timer := edsImpl.priorityInitTimer; timer != nil {
timer.Stop() timer.Stop()
xdsB.priorityInitTimer = nil edsImpl.priorityInitTimer = nil
} }
pNext := priority.nextLower() pNext := priority.nextLower()
if _, okNext := xdsB.priorityToLocalities[pNext]; !okNext { if _, okNext := edsImpl.priorityToLocalities[pNext]; !okNext {
return true return true
} }
xdsB.startPriority(pNext) edsImpl.startPriority(pNext)
return true return true
} }
@ -247,18 +247,18 @@ func (xdsB *EDSBalancer) handlePriorityWithNewStateTransientFailure(priority pri
// Caller must make sure priorityInUse is not higher than priority. // Caller must make sure priorityInUse is not higher than priority.
// //
// Caller must hold priorityMu. // Caller must hold priorityMu.
func (xdsB *EDSBalancer) handlePriorityWithNewStateConnecting(priority priorityType, oldState connectivity.State) bool { func (edsImpl *edsBalancerImpl) handlePriorityWithNewStateConnecting(priority priorityType, oldState connectivity.State) bool {
if xdsB.priorityInUse.lowerThan(priority) { if edsImpl.priorityInUse.lowerThan(priority) {
return false return false
} }
switch oldState { switch oldState {
case connectivity.Ready: case connectivity.Ready:
pNext := priority.nextLower() pNext := priority.nextLower()
if _, okNext := xdsB.priorityToLocalities[pNext]; !okNext { if _, okNext := edsImpl.priorityToLocalities[pNext]; !okNext {
return true return true
} }
xdsB.startPriority(pNext) edsImpl.startPriority(pNext)
return true return true
case connectivity.Idle: case connectivity.Idle:
return true return true

14
xds/internal/balancer/edsbalancer/priority_test.go → xds/internal/balancer/edsbalancer/eds_impl_priority_test.go
View File

@ -33,7 +33,7 @@ import (
// Init 0 and 1; 0 is up, use 0; add 2, use 0; remove 2, use 0. // Init 0 and 1; 0 is up, use 0; add 2, use 0; remove 2, use 0.
func TestEDSPriority_HighPriorityReady(t *testing.T) { func TestEDSPriority_HighPriorityReady(t *testing.T) {
cc := newTestClientConn(t) cc := newTestClientConn(t)
edsb := NewXDSBalancer(cc, nil) edsb := newEDSBalancerImpl(cc, nil)
// Two localities, with priorities [0, 1], each with one backend. // Two localities, with priorities [0, 1], each with one backend.
clab1 := xdsclient.NewClusterLoadAssignmentBuilder(testClusterNames[0], nil) clab1 := xdsclient.NewClusterLoadAssignmentBuilder(testClusterNames[0], nil)
@ -99,7 +99,7 @@ func TestEDSPriority_HighPriorityReady(t *testing.T) {
// down, use 2; remove 2, use 1. // down, use 2; remove 2, use 1.
func TestEDSPriority_SwitchPriority(t *testing.T) { func TestEDSPriority_SwitchPriority(t *testing.T) {
cc := newTestClientConn(t) cc := newTestClientConn(t)
edsb := NewXDSBalancer(cc, nil) edsb := newEDSBalancerImpl(cc, nil)
// Two localities, with priorities [0, 1], each with one backend. // Two localities, with priorities [0, 1], each with one backend.
clab1 := xdsclient.NewClusterLoadAssignmentBuilder(testClusterNames[0], nil) clab1 := xdsclient.NewClusterLoadAssignmentBuilder(testClusterNames[0], nil)
@ -206,7 +206,7 @@ func TestEDSPriority_SwitchPriority(t *testing.T) {
// Init 0 and 1; 0 and 1 both down; add 2, use 2. // Init 0 and 1; 0 and 1 both down; add 2, use 2.
func TestEDSPriority_HigherDownWhileAddingLower(t *testing.T) { func TestEDSPriority_HigherDownWhileAddingLower(t *testing.T) {
cc := newTestClientConn(t) cc := newTestClientConn(t)
edsb := NewXDSBalancer(cc, nil) edsb := newEDSBalancerImpl(cc, nil)
// Two localities, with different priorities, each with one backend. // Two localities, with different priorities, each with one backend.
clab1 := xdsclient.NewClusterLoadAssignmentBuilder(testClusterNames[0], nil) clab1 := xdsclient.NewClusterLoadAssignmentBuilder(testClusterNames[0], nil)
@ -270,7 +270,7 @@ func TestEDSPriority_HigherReadyCloseAllLower(t *testing.T) {
defer time.Sleep(10 * time.Millisecond) defer time.Sleep(10 * time.Millisecond)
cc := newTestClientConn(t) cc := newTestClientConn(t)
edsb := NewXDSBalancer(cc, nil) edsb := newEDSBalancerImpl(cc, nil)
// Two localities, with priorities [0,1,2], each with one backend. // Two localities, with priorities [0,1,2], each with one backend.
clab1 := xdsclient.NewClusterLoadAssignmentBuilder(testClusterNames[0], nil) clab1 := xdsclient.NewClusterLoadAssignmentBuilder(testClusterNames[0], nil)
@ -349,7 +349,7 @@ func TestEDSPriority_InitTimeout(t *testing.T) {
}()() }()()
cc := newTestClientConn(t) cc := newTestClientConn(t)
edsb := NewXDSBalancer(cc, nil) edsb := newEDSBalancerImpl(cc, nil)
// Two localities, with different priorities, each with one backend. // Two localities, with different priorities, each with one backend.
clab1 := xdsclient.NewClusterLoadAssignmentBuilder(testClusterNames[0], nil) clab1 := xdsclient.NewClusterLoadAssignmentBuilder(testClusterNames[0], nil)
@ -398,7 +398,7 @@ func TestEDSPriority_InitTimeout(t *testing.T) {
// - add localities to existing p0 and p1 // - add localities to existing p0 and p1
func TestEDSPriority_MultipleLocalities(t *testing.T) { func TestEDSPriority_MultipleLocalities(t *testing.T) {
cc := newTestClientConn(t) cc := newTestClientConn(t)
edsb := NewXDSBalancer(cc, nil) edsb := newEDSBalancerImpl(cc, nil)
// Two localities, with different priorities, each with one backend. // Two localities, with different priorities, each with one backend.
clab0 := xdsclient.NewClusterLoadAssignmentBuilder(testClusterNames[0], nil) clab0 := xdsclient.NewClusterLoadAssignmentBuilder(testClusterNames[0], nil)
@ -510,7 +510,7 @@ func TestEDSPriority_RemovesAllLocalities(t *testing.T) {
}()() }()()
cc := newTestClientConn(t) cc := newTestClientConn(t)
edsb := NewXDSBalancer(cc, nil) edsb := newEDSBalancerImpl(cc, nil)
// Two localities, with different priorities, each with one backend. // Two localities, with different priorities, each with one backend.
clab0 := xdsclient.NewClusterLoadAssignmentBuilder(testClusterNames[0], nil) clab0 := xdsclient.NewClusterLoadAssignmentBuilder(testClusterNames[0], nil)

12
xds/internal/balancer/edsbalancer/edsbalancer_test.go → xds/internal/balancer/edsbalancer/eds_impl_test.go
View File

@ -52,7 +52,7 @@ func init() {
// - change drop rate // - change drop rate
func TestEDS_OneLocality(t *testing.T) { func TestEDS_OneLocality(t *testing.T) {
cc := newTestClientConn(t) cc := newTestClientConn(t)
edsb := NewXDSBalancer(cc, nil) edsb := newEDSBalancerImpl(cc, nil)
// One locality with one backend. // One locality with one backend.
clab1 := xdsclient.NewClusterLoadAssignmentBuilder(testClusterNames[0], nil) clab1 := xdsclient.NewClusterLoadAssignmentBuilder(testClusterNames[0], nil)
@ -158,7 +158,7 @@ func TestEDS_OneLocality(t *testing.T) {
// - update locality weight // - update locality weight
func TestEDS_TwoLocalities(t *testing.T) { func TestEDS_TwoLocalities(t *testing.T) {
cc := newTestClientConn(t) cc := newTestClientConn(t)
edsb := NewXDSBalancer(cc, nil) edsb := newEDSBalancerImpl(cc, nil)
// Two localities, each with one backend. // Two localities, each with one backend.
clab1 := xdsclient.NewClusterLoadAssignmentBuilder(testClusterNames[0], nil) clab1 := xdsclient.NewClusterLoadAssignmentBuilder(testClusterNames[0], nil)
@ -288,7 +288,7 @@ func TestEDS_TwoLocalities(t *testing.T) {
// healthy ones are used. // healthy ones are used.
func TestEDS_EndpointsHealth(t *testing.T) { func TestEDS_EndpointsHealth(t *testing.T) {
cc := newTestClientConn(t) cc := newTestClientConn(t)
edsb := NewXDSBalancer(cc, nil) edsb := newEDSBalancerImpl(cc, nil)
// Two localities, each 3 backend, one Healthy, one Unhealthy, one Unknown. // Two localities, each 3 backend, one Healthy, one Unhealthy, one Unknown.
clab1 := xdsclient.NewClusterLoadAssignmentBuilder(testClusterNames[0], nil) clab1 := xdsclient.NewClusterLoadAssignmentBuilder(testClusterNames[0], nil)
@ -359,7 +359,7 @@ func TestEDS_EndpointsHealth(t *testing.T) {
} }
func TestClose(t *testing.T) { func TestClose(t *testing.T) {
edsb := NewXDSBalancer(nil, nil) edsb := newEDSBalancerImpl(nil, nil)
// This is what could happen when switching between fallback and eds. This // This is what could happen when switching between fallback and eds. This
// make sure it doesn't panic. // make sure it doesn't panic.
edsb.Close() edsb.Close()
@ -416,7 +416,7 @@ func (tcp *testConstPicker) Pick(info balancer.PickInfo) (balancer.PickResult, e
// eds response. // eds response.
func TestEDS_UpdateSubBalancerName(t *testing.T) { func TestEDS_UpdateSubBalancerName(t *testing.T) {
cc := newTestClientConn(t) cc := newTestClientConn(t)
edsb := NewXDSBalancer(cc, nil) edsb := newEDSBalancerImpl(cc, nil)
t.Logf("update sub-balancer to test-const-balancer") t.Logf("update sub-balancer to test-const-balancer")
edsb.HandleChildPolicy("test-const-balancer", nil) edsb.HandleChildPolicy("test-const-balancer", nil)
@ -575,7 +575,7 @@ func TestEDS_LoadReport(t *testing.T) {
testLoadStore := newTestLoadStore() testLoadStore := newTestLoadStore()
cc := newTestClientConn(t) cc := newTestClientConn(t)
edsb := NewXDSBalancer(cc, testLoadStore) edsb := newEDSBalancerImpl(cc, testLoadStore)
backendToBalancerID := make(map[balancer.SubConn]internal.Locality) backendToBalancerID := make(map[balancer.SubConn]internal.Locality)

50
xds/internal/balancer/xds_test.go → xds/internal/balancer/edsbalancer/eds_test.go
View File

@ -16,7 +16,7 @@
* *
*/ */
package balancer package edsbalancer
import ( import (
"bytes" "bytes"
@ -68,21 +68,21 @@ func Test(t *testing.T) {
const testBalancerNameFooBar = "foo.bar" const testBalancerNameFooBar = "foo.bar"
func newTestClientConn() *testClientConn { func newNoopTestClientConn() *noopTestClientConn {
return &testClientConn{newSubConns: testutils.NewChannelWithSize(10)} return &noopTestClientConn{}
} }
type testClientConn struct { // noopTestClientConn is used in EDS balancer config update tests that only
// cover the config update handling, but not SubConn/load-balancing.
type noopTestClientConn struct {
balancer.ClientConn balancer.ClientConn
newSubConns *testutils.Channel
} }
func (t *testClientConn) NewSubConn(addrs []resolver.Address, opts balancer.NewSubConnOptions) (balancer.SubConn, error) { func (t *noopTestClientConn) NewSubConn(addrs []resolver.Address, opts balancer.NewSubConnOptions) (balancer.SubConn, error) {
t.newSubConns.Send(addrs)
return nil, nil return nil, nil
} }
func (testClientConn) Target() string { return testServiceName } func (noopTestClientConn) Target() string { return testServiceName }
type scStateChange struct { type scStateChange struct {
sc balancer.SubConn sc balancer.SubConn
@ -131,7 +131,7 @@ func (f *fakeEDSBalancer) waitForSubConnStateChange(wantState *scStateChange) er
return nil return nil
} }
func newFakeEDSBalancer(cc balancer.ClientConn, loadStore lrs.Store) edsBalancerInterface { func newFakeEDSBalancer(cc balancer.ClientConn, loadStore lrs.Store) edsBalancerImplInterface {
return &fakeEDSBalancer{ return &fakeEDSBalancer{
cc: cc, cc: cc,
childPolicy: testutils.NewChannelWithSize(10), childPolicy: testutils.NewChannelWithSize(10),
@ -188,7 +188,7 @@ func waitForNewEDSLB(t *testing.T, ch *testutils.Channel) *fakeEDSBalancer {
// cleanup. // cleanup.
func setup(edsLBCh *testutils.Channel, xdsClientCh *testutils.Channel) func() { func setup(edsLBCh *testutils.Channel, xdsClientCh *testutils.Channel) func() {
origNewEDSBalancer := newEDSBalancer origNewEDSBalancer := newEDSBalancer
newEDSBalancer = func(cc balancer.ClientConn, loadStore lrs.Store) edsBalancerInterface { newEDSBalancer = func(cc balancer.ClientConn, loadStore lrs.Store) edsBalancerImplInterface {
edsLB := newFakeEDSBalancer(cc, loadStore) edsLB := newFakeEDSBalancer(cc, loadStore)
defer func() { edsLBCh.Send(edsLB) }() defer func() { edsLBCh.Send(edsLB) }()
return edsLB return edsLB
@ -221,7 +221,7 @@ func (s) TestXDSConfigBalancerNameUpdate(t *testing.T) {
defer cancel() defer cancel()
builder := balancer.Get(edsName) builder := balancer.Get(edsName)
cc := newTestClientConn() cc := newNoopTestClientConn()
edsB, ok := builder.Build(cc, balancer.BuildOptions{Target: resolver.Target{Endpoint: testEDSClusterName}}).(*edsBalancer) edsB, ok := builder.Build(cc, balancer.BuildOptions{Target: resolver.Target{Endpoint: testEDSClusterName}}).(*edsBalancer)
if !ok { if !ok {
t.Fatalf("builder.Build(%s) returned type {%T}, want {*edsBalancer}", edsName, edsB) t.Fatalf("builder.Build(%s) returned type {%T}, want {*edsBalancer}", edsName, edsB)
@ -233,7 +233,7 @@ func (s) TestXDSConfigBalancerNameUpdate(t *testing.T) {
balancerName := fmt.Sprintf("balancer-%d", i) balancerName := fmt.Sprintf("balancer-%d", i)
edsB.UpdateClientConnState(balancer.ClientConnState{ edsB.UpdateClientConnState(balancer.ClientConnState{
ResolverState: resolver.State{Addresses: addrs}, ResolverState: resolver.State{Addresses: addrs},
BalancerConfig: &XDSConfig{ BalancerConfig: &EDSConfig{
BalancerName: balancerName, BalancerName: balancerName,
EDSServiceName: testEDSClusterName, EDSServiceName: testEDSClusterName,
}, },
@ -303,7 +303,7 @@ func (s) TestXDSConnfigChildPolicyUpdate(t *testing.T) {
defer cancel() defer cancel()
builder := balancer.Get(edsName) builder := balancer.Get(edsName)
cc := newTestClientConn() cc := newNoopTestClientConn()
edsB, ok := builder.Build(cc, balancer.BuildOptions{Target: resolver.Target{Endpoint: testServiceName}}).(*edsBalancer) edsB, ok := builder.Build(cc, balancer.BuildOptions{Target: resolver.Target{Endpoint: testServiceName}}).(*edsBalancer)
if !ok { if !ok {
t.Fatalf("builder.Build(%s) returned type {%T}, want {*edsBalancer}", edsName, edsB) t.Fatalf("builder.Build(%s) returned type {%T}, want {*edsBalancer}", edsName, edsB)
@ -311,7 +311,7 @@ func (s) TestXDSConnfigChildPolicyUpdate(t *testing.T) {
defer edsB.Close() defer edsB.Close()
edsB.UpdateClientConnState(balancer.ClientConnState{ edsB.UpdateClientConnState(balancer.ClientConnState{
BalancerConfig: &XDSConfig{ BalancerConfig: &EDSConfig{
BalancerName: testBalancerNameFooBar, BalancerName: testBalancerNameFooBar,
ChildPolicy: &loadBalancingConfig{ ChildPolicy: &loadBalancingConfig{
Name: fakeBalancerA, Name: fakeBalancerA,
@ -329,7 +329,7 @@ func (s) TestXDSConnfigChildPolicyUpdate(t *testing.T) {
}) })
edsB.UpdateClientConnState(balancer.ClientConnState{ edsB.UpdateClientConnState(balancer.ClientConnState{
BalancerConfig: &XDSConfig{ BalancerConfig: &EDSConfig{
BalancerName: testBalancerNameFooBar, BalancerName: testBalancerNameFooBar,
ChildPolicy: &loadBalancingConfig{ ChildPolicy: &loadBalancingConfig{
Name: fakeBalancerB, Name: fakeBalancerB,
@ -353,7 +353,7 @@ func (s) TestXDSSubConnStateChange(t *testing.T) {
defer cancel() defer cancel()
builder := balancer.Get(edsName) builder := balancer.Get(edsName)
cc := newTestClientConn() cc := newNoopTestClientConn()
edsB, ok := builder.Build(cc, balancer.BuildOptions{Target: resolver.Target{Endpoint: testEDSClusterName}}).(*edsBalancer) edsB, ok := builder.Build(cc, balancer.BuildOptions{Target: resolver.Target{Endpoint: testEDSClusterName}}).(*edsBalancer)
if !ok { if !ok {
t.Fatalf("builder.Build(%s) returned type {%T}, want {*edsBalancer}", edsName, edsB) t.Fatalf("builder.Build(%s) returned type {%T}, want {*edsBalancer}", edsName, edsB)
@ -363,7 +363,7 @@ func (s) TestXDSSubConnStateChange(t *testing.T) {
addrs := []resolver.Address{{Addr: "1.1.1.1:10001"}, {Addr: "2.2.2.2:10002"}, {Addr: "3.3.3.3:10003"}} addrs := []resolver.Address{{Addr: "1.1.1.1:10001"}, {Addr: "2.2.2.2:10002"}, {Addr: "3.3.3.3:10003"}}
edsB.UpdateClientConnState(balancer.ClientConnState{ edsB.UpdateClientConnState(balancer.ClientConnState{
ResolverState: resolver.State{Addresses: addrs}, ResolverState: resolver.State{Addresses: addrs},
BalancerConfig: &XDSConfig{ BalancerConfig: &EDSConfig{
BalancerName: testBalancerNameFooBar, BalancerName: testBalancerNameFooBar,
EDSServiceName: testEDSClusterName, EDSServiceName: testEDSClusterName,
}, },
@ -411,7 +411,7 @@ func TestXDSBalancerConfigParsing(t *testing.T) {
{ {
name: "jsonpb-generated", name: "jsonpb-generated",
js: b.Bytes(), js: b.Bytes(),
want: &XDSConfig{ want: &EDSConfig{
ChildPolicy: &loadBalancingConfig{ ChildPolicy: &loadBalancingConfig{
Name: "round_robin", Name: "round_robin",
Config: json.RawMessage("{}"), Config: json.RawMessage("{}"),
@ -444,7 +444,7 @@ func TestXDSBalancerConfigParsing(t *testing.T) {
"edsServiceName": "eds.service", "edsServiceName": "eds.service",
"lrsLoadReportingServerName": "lrs.server" "lrsLoadReportingServerName": "lrs.server"
}`), }`),
want: &XDSConfig{ want: &EDSConfig{
BalancerName: "fake.foo.bar", BalancerName: "fake.foo.bar",
ChildPolicy: &loadBalancingConfig{ ChildPolicy: &loadBalancingConfig{
Name: "fake_balancer_A", Name: "fake_balancer_A",
@ -468,7 +468,7 @@ func TestXDSBalancerConfigParsing(t *testing.T) {
"balancerName": "fake.foo.bar", "balancerName": "fake.foo.bar",
"edsServiceName": "eds.service" "edsServiceName": "eds.service"
}`), }`),
want: &XDSConfig{ want: &EDSConfig{
BalancerName: "fake.foo.bar", BalancerName: "fake.foo.bar",
EDSServiceName: testEDSName, EDSServiceName: testEDSName,
LrsLoadReportingServerName: nil, LrsLoadReportingServerName: nil,
@ -494,17 +494,17 @@ func TestLoadbalancingConfigParsing(t *testing.T) {
tests := []struct { tests := []struct {
name string name string
s string s string
want *XDSConfig want *EDSConfig
}{ }{
{ {
name: "empty", name: "empty",
s: "{}", s: "{}",
want: &XDSConfig{}, want: &EDSConfig{},
}, },
{ {
name: "success1", name: "success1",
s: `{"childPolicy":[{"pick_first":{}}]}`, s: `{"childPolicy":[{"pick_first":{}}]}`,
want: &XDSConfig{ want: &EDSConfig{
ChildPolicy: &loadBalancingConfig{ ChildPolicy: &loadBalancingConfig{
Name: "pick_first", Name: "pick_first",
Config: json.RawMessage(`{}`), Config: json.RawMessage(`{}`),
@ -514,7 +514,7 @@ func TestLoadbalancingConfigParsing(t *testing.T) {
{ {
name: "success2", name: "success2",
s: `{"childPolicy":[{"round_robin":{}},{"pick_first":{}}]}`, s: `{"childPolicy":[{"round_robin":{}},{"pick_first":{}}]}`,
want: &XDSConfig{ want: &EDSConfig{
ChildPolicy: &loadBalancingConfig{ ChildPolicy: &loadBalancingConfig{
Name: "round_robin", Name: "round_robin",
Config: json.RawMessage(`{}`), Config: json.RawMessage(`{}`),
@ -524,7 +524,7 @@ func TestLoadbalancingConfigParsing(t *testing.T) {
} }
for _, tt := range tests { for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) { t.Run(tt.name, func(t *testing.T) {
var cfg XDSConfig var cfg EDSConfig
if err := json.Unmarshal([]byte(tt.s), &cfg); err != nil || !reflect.DeepEqual(&cfg, tt.want) { if err := json.Unmarshal([]byte(tt.s), &cfg); err != nil || !reflect.DeepEqual(&cfg, tt.want) {
t.Errorf("test name: %s, parseFullServiceConfig() = %+v, err: %v, want %+v, <nil>", tt.name, cfg, err, tt.want) t.Errorf("test name: %s, parseFullServiceConfig() = %+v, err: %v, want %+v, <nil>", tt.name, cfg, err, tt.want)
} }

6
xds/internal/balancer/xds_client_wrapper.go → xds/internal/balancer/edsbalancer/xds_client_wrapper.go
View File

@ -16,7 +16,7 @@
* *
*/ */
package balancer package edsbalancer
import ( import (
"google.golang.org/grpc" "google.golang.org/grpc"
@ -122,7 +122,7 @@ func (c *xdsclientWrapper) replaceXDSClient(newClient xdsClientInterface, newBal
// the balancerName (from bootstrap file or from service config) changed. // the balancerName (from bootstrap file or from service config) changed.
// - if balancer names are the same, do nothing, and return false // - if balancer names are the same, do nothing, and return false
// - if balancer names are different, create new one, and return true // - if balancer names are different, create new one, and return true
func (c *xdsclientWrapper) updateXDSClient(config *XDSConfig, attr *attributes.Attributes) bool { func (c *xdsclientWrapper) updateXDSClient(config *EDSConfig, attr *attributes.Attributes) bool {
if attr != nil { if attr != nil {
if clientFromAttr, _ := attr.Value(xdsinternal.XDSClientID).(xdsClientInterface); clientFromAttr != nil { if clientFromAttr, _ := attr.Value(xdsinternal.XDSClientID).(xdsClientInterface); clientFromAttr != nil {
// This will also clear balancerName, to indicate that client is // This will also clear balancerName, to indicate that client is
@ -217,7 +217,7 @@ func (c *xdsclientWrapper) startLoadReport(edsServiceNameBeingWatched string, lo
// handleUpdate applies the service config and attributes updates to the client, // handleUpdate applies the service config and attributes updates to the client,
// including updating the xds_client to use, and updating the EDS name to watch. // including updating the xds_client to use, and updating the EDS name to watch.
func (c *xdsclientWrapper) handleUpdate(config *XDSConfig, attr *attributes.Attributes) { func (c *xdsclientWrapper) handleUpdate(config *EDSConfig, attr *attributes.Attributes) {
clientChanged := c.updateXDSClient(config, attr) clientChanged := c.updateXDSClient(config, attr)
var ( var (

10
xds/internal/balancer/xds_client_test.go → xds/internal/balancer/edsbalancer/xds_client_wrapper_test.go
View File

@ -16,7 +16,7 @@
* *
*/ */
package balancer package edsbalancer
import ( import (
"errors" "errors"
@ -93,7 +93,7 @@ func (s) TestClientWrapperWatchEDS(t *testing.T) {
}, },
} { } {
t.Run(test.name, func(t *testing.T) { t.Run(test.name, func(t *testing.T) {
cw.handleUpdate(&XDSConfig{ cw.handleUpdate(&EDSConfig{
BalancerName: fakeServer.Address, BalancerName: fakeServer.Address,
EDSServiceName: test.edsServiceName, EDSServiceName: test.edsServiceName,
}, nil) }, nil)
@ -140,7 +140,7 @@ func (s) TestClientWrapperHandleUpdateError(t *testing.T) {
defer cw.close() defer cw.close()
xdsC := fakeclient.NewClient() xdsC := fakeclient.NewClient()
cw.handleUpdate(&XDSConfig{EDSServiceName: testEDSClusterName}, attributes.New(xdsinternal.XDSClientID, xdsC)) cw.handleUpdate(&EDSConfig{EDSServiceName: testEDSClusterName}, attributes.New(xdsinternal.XDSClientID, xdsC))
gotCluster, err := xdsC.WaitForWatchEDS() gotCluster, err := xdsC.WaitForWatchEDS()
if err != nil { if err != nil {
t.Fatalf("xdsClient.WatchEDS failed with error: %v", err) t.Fatalf("xdsClient.WatchEDS failed with error: %v", err)
@ -175,7 +175,7 @@ func (s) TestClientWrapperGetsXDSClientInAttributes(t *testing.T) {
// Verify that the eds watch is registered for the expected resource name. // Verify that the eds watch is registered for the expected resource name.
xdsC1 := fakeclient.NewClient() xdsC1 := fakeclient.NewClient()
cw.handleUpdate(&XDSConfig{EDSServiceName: testEDSClusterName}, attributes.New(xdsinternal.XDSClientID, xdsC1)) cw.handleUpdate(&EDSConfig{EDSServiceName: testEDSClusterName}, attributes.New(xdsinternal.XDSClientID, xdsC1))
gotCluster, err := xdsC1.WaitForWatchEDS() gotCluster, err := xdsC1.WaitForWatchEDS()
if err != nil { if err != nil {
t.Fatalf("xdsClient.WatchEDS failed with error: %v", err) t.Fatalf("xdsClient.WatchEDS failed with error: %v", err)
@ -189,7 +189,7 @@ func (s) TestClientWrapperGetsXDSClientInAttributes(t *testing.T) {
// (because clientWrapper only closes clients that it creates, it does not // (because clientWrapper only closes clients that it creates, it does not
// close client that are passed through attributes). // close client that are passed through attributes).
xdsC2 := fakeclient.NewClient() xdsC2 := fakeclient.NewClient()
cw.handleUpdate(&XDSConfig{EDSServiceName: testEDSClusterName}, attributes.New(xdsinternal.XDSClientID, xdsC2)) cw.handleUpdate(&EDSConfig{EDSServiceName: testEDSClusterName}, attributes.New(xdsinternal.XDSClientID, xdsC2))
gotCluster, err = xdsC2.WaitForWatchEDS() gotCluster, err = xdsC2.WaitForWatchEDS()
if err != nil { if err != nil {
t.Fatalf("xdsClient.WatchEDS failed with error: %v", err) t.Fatalf("xdsClient.WatchEDS failed with error: %v", err)

6
xds/internal/balancer/xds_lrs_test.go → xds/internal/balancer/edsbalancer/xds_lrs_test.go
View File

@ -16,7 +16,7 @@
* *
*/ */
package balancer package edsbalancer
import ( import (
"testing" "testing"
@ -33,7 +33,7 @@ import (
// server (empty string). // server (empty string).
func (s) TestXDSLoadReporting(t *testing.T) { func (s) TestXDSLoadReporting(t *testing.T) {
builder := balancer.Get(edsName) builder := balancer.Get(edsName)
cc := newTestClientConn() cc := newNoopTestClientConn()
edsB, ok := builder.Build(cc, balancer.BuildOptions{Target: resolver.Target{Endpoint: testEDSClusterName}}).(*edsBalancer) edsB, ok := builder.Build(cc, balancer.BuildOptions{Target: resolver.Target{Endpoint: testEDSClusterName}}).(*edsBalancer)
if !ok { if !ok {
t.Fatalf("builder.Build(%s) returned type {%T}, want {*edsBalancer}", edsName, edsB) t.Fatalf("builder.Build(%s) returned type {%T}, want {*edsBalancer}", edsName, edsB)
@ -43,7 +43,7 @@ func (s) TestXDSLoadReporting(t *testing.T) {
xdsC := fakeclient.NewClient() xdsC := fakeclient.NewClient()
edsB.UpdateClientConnState(balancer.ClientConnState{ edsB.UpdateClientConnState(balancer.ClientConnState{
ResolverState: resolver.State{Attributes: attributes.New(xdsinternal.XDSClientID, xdsC)}, ResolverState: resolver.State{Attributes: attributes.New(xdsinternal.XDSClientID, xdsC)},
BalancerConfig: &XDSConfig{LrsLoadReportingServerName: new(string)}, BalancerConfig: &EDSConfig{LrsLoadReportingServerName: new(string)},
}) })
gotCluster, err := xdsC.WaitForWatchEDS() gotCluster, err := xdsC.WaitForWatchEDS()

2
xds/internal/balancer/xds_old.go → xds/internal/balancer/edsbalancer/xds_old.go
View File

@ -15,7 +15,7 @@
* limitations under the License. * limitations under the License.
*/ */
package balancer package edsbalancer
import "google.golang.org/grpc/balancer" import "google.golang.org/grpc/balancer"