Skip to content

Commit

Permalink
clusterresolver: merge P(p)arseConfig functions (#5462)
Browse files Browse the repository at this point in the history
  • Loading branch information
easwars committed Jun 24, 2022
1 parent d883f3d commit 4b75005
Show file tree
Hide file tree
Showing 5 changed files with 27 additions and 37 deletions.
6 changes: 6 additions & 0 deletions xds/internal/balancer/clusterresolver/clusterresolver.go
Expand Up @@ -23,10 +23,12 @@ import (
"encoding/json"
"errors"
"fmt"
"strings"

"google.golang.org/grpc/attributes"
"google.golang.org/grpc/balancer"
"google.golang.org/grpc/balancer/base"
"google.golang.org/grpc/balancer/roundrobin"
"google.golang.org/grpc/connectivity"
"google.golang.org/grpc/internal/buffer"
"google.golang.org/grpc/internal/grpclog"
Expand All @@ -35,6 +37,7 @@ import (
"google.golang.org/grpc/resolver"
"google.golang.org/grpc/serviceconfig"
"google.golang.org/grpc/xds/internal/balancer/priority"
"google.golang.org/grpc/xds/internal/balancer/ringhash"
"google.golang.org/grpc/xds/internal/xdsclient"
"google.golang.org/grpc/xds/internal/xdsclient/xdsresource"
)
Expand Down Expand Up @@ -99,6 +102,9 @@ func (bb) ParseConfig(c json.RawMessage) (serviceconfig.LoadBalancingConfig, err
if err := json.Unmarshal(c, &cfg); err != nil {
return nil, fmt.Errorf("unable to unmarshal balancer config %s into cluster-resolver config, error: %v", string(c), err)
}
if lbp := cfg.XDSLBPolicy; lbp != nil && !strings.EqualFold(lbp.Name, roundrobin.Name) && !strings.EqualFold(lbp.Name, ringhash.Name) {
return nil, fmt.Errorf("unsupported child policy with name %q, not one of {%q,%q}", lbp.Name, roundrobin.Name, ringhash.Name)
}
return &cfg, nil
}

Expand Down
19 changes: 0 additions & 19 deletions xds/internal/balancer/clusterresolver/config.go
Expand Up @@ -21,13 +21,10 @@ import (
"bytes"
"encoding/json"
"fmt"
"strings"

"google.golang.org/grpc/balancer/roundrobin"
internalserviceconfig "google.golang.org/grpc/internal/serviceconfig"
"google.golang.org/grpc/serviceconfig"
"google.golang.org/grpc/xds/internal/balancer/outlierdetection"
"google.golang.org/grpc/xds/internal/balancer/ringhash"
"google.golang.org/grpc/xds/internal/xdsclient/bootstrap"
)

Expand Down Expand Up @@ -167,19 +164,3 @@ type LBConfig struct {
// is responsible for both locality picking and endpoint picking.
XDSLBPolicy *internalserviceconfig.BalancerConfig `json:"xdsLbPolicy,omitempty"`
}

const (
rrName = roundrobin.Name
rhName = ringhash.Name
)

func parseConfig(c json.RawMessage) (*LBConfig, error) {
var cfg LBConfig
if err := json.Unmarshal(c, &cfg); err != nil {
return nil, err
}
if lbp := cfg.XDSLBPolicy; lbp != nil && !strings.EqualFold(lbp.Name, rrName) && !strings.EqualFold(lbp.Name, rhName) {
return nil, fmt.Errorf("unsupported child policy with name %q, not one of {%q,%q}", lbp.Name, rrName, rhName)
}
return &cfg, nil
}
25 changes: 14 additions & 11 deletions xds/internal/balancer/clusterresolver/config_test.go
Expand Up @@ -23,7 +23,7 @@ import (
"testing"

"github.com/google/go-cmp/cmp"
"google.golang.org/grpc/internal/balancer/stub"
"google.golang.org/grpc/balancer"
internalserviceconfig "google.golang.org/grpc/internal/serviceconfig"
"google.golang.org/grpc/xds/internal/balancer/ringhash"
"google.golang.org/grpc/xds/internal/xdsclient/bootstrap"
Expand Down Expand Up @@ -91,14 +91,6 @@ func TestDiscoveryMechanismTypeUnmarshalJSON(t *testing.T) {
}
}

func init() {
// This is needed now for the config parsing tests to pass. Otherwise they
// will fail with "RING_HASH unsupported".
//
// TODO: delete this once ring-hash policy is implemented and imported.
stub.Register(rhName, stub.BalancerFuncs{})
}

const (
testJSONConfig1 = `{
"discoveryMechanisms": [{
Expand Down Expand Up @@ -257,7 +249,7 @@ func TestParseConfig(t *testing.T) {
},
XDSLBPolicy: &internalserviceconfig.BalancerConfig{
Name: ringhash.Name,
Config: nil,
Config: &ringhash.LBConfig{MinRingSize: 1024, MaxRingSize: 8388608}, // Ringhash LB config with default min and max.
},
},
wantErr: false,
Expand All @@ -269,11 +261,22 @@ func TestParseConfig(t *testing.T) {
},
}
for _, tt := range tests {
b := balancer.Get(Name)
if b == nil {
t.Fatalf("LB policy %q not registered", Name)
}
cfgParser, ok := b.(balancer.ConfigParser)
if !ok {
t.Fatalf("LB policy %q does not support config parsing", Name)
}
t.Run(tt.name, func(t *testing.T) {
got, err := parseConfig([]byte(tt.js))
got, err := cfgParser.ParseConfig([]byte(tt.js))
if (err != nil) != tt.wantErr {
t.Fatalf("parseConfig() error = %v, wantErr %v", err, tt.wantErr)
}
if tt.wantErr {
return
}
if diff := cmp.Diff(got, tt.want); diff != "" {
t.Errorf("parseConfig() got unexpected output, diff (-got +want): %v", diff)
}
Expand Down
10 changes: 5 additions & 5 deletions xds/internal/balancer/clusterresolver/configbuilder.go
Expand Up @@ -304,30 +304,30 @@ func priorityLocalitiesToClusterImpl(localities []xdsresource.Locality, priority
// ChildPolicy is not set. Will be set based on xdsLBPolicy
}

if xdsLBPolicy == nil || xdsLBPolicy.Name == rrName {
if xdsLBPolicy == nil || xdsLBPolicy.Name == roundrobin.Name {
// If lb policy is ROUND_ROBIN:
// - locality-picking policy is weighted_target
// - endpoint-picking policy is round_robin
logger.Infof("xds lb policy is %q, building config with weighted_target + round_robin", rrName)
logger.Infof("xds lb policy is %q, building config with weighted_target + round_robin", roundrobin.Name)
// Child of weighted_target is hardcoded to round_robin.
wtConfig, addrs := localitiesToWeightedTarget(localities, priorityName, rrBalancerConfig)
clusterImplCfg.ChildPolicy = &internalserviceconfig.BalancerConfig{Name: weightedtarget.Name, Config: wtConfig}
return clusterImplCfg, addrs, nil
}

if xdsLBPolicy.Name == rhName {
if xdsLBPolicy.Name == ringhash.Name {
// If lb policy is RIHG_HASH, will build one ring_hash policy as child.
// The endpoints from all localities will be flattened to one addresses
// list, and the ring_hash policy will pick endpoints from it.
logger.Infof("xds lb policy is %q, building config with ring_hash", rhName)
logger.Infof("xds lb policy is %q, building config with ring_hash", ringhash.Name)
addrs := localitiesToRingHash(localities, priorityName)
// Set child to ring_hash, note that the ring_hash config is from
// xdsLBPolicy.
clusterImplCfg.ChildPolicy = &internalserviceconfig.BalancerConfig{Name: ringhash.Name, Config: xdsLBPolicy.Config}
return clusterImplCfg, addrs, nil
}

return nil, nil, fmt.Errorf("unsupported xds LB policy %q, not one of {%q,%q}", xdsLBPolicy.Name, rrName, rhName)
return nil, nil, fmt.Errorf("unsupported xds LB policy %q, not one of {%q,%q}", xdsLBPolicy.Name, roundrobin.Name, ringhash.Name)
}

// localitiesToRingHash takes a list of localities (with the same priority), and
Expand Down
4 changes: 2 additions & 2 deletions xds/internal/balancer/clusterresolver/configbuilder_test.go
Expand Up @@ -731,7 +731,7 @@ func TestPriorityLocalitiesToClusterImpl(t *testing.T) {
},
},
priorityName: "test-priority",
childPolicy: &internalserviceconfig.BalancerConfig{Name: rrName},
childPolicy: &internalserviceconfig.BalancerConfig{Name: roundrobin.Name},
mechanism: DiscoveryMechanism{
Cluster: testClusterName,
Type: DiscoveryMechanismTypeEDS,
Expand Down Expand Up @@ -789,7 +789,7 @@ func TestPriorityLocalitiesToClusterImpl(t *testing.T) {
},
},
priorityName: "test-priority",
childPolicy: &internalserviceconfig.BalancerConfig{Name: rhName, Config: &ringhash.LBConfig{MinRingSize: 1, MaxRingSize: 2}},
childPolicy: &internalserviceconfig.BalancerConfig{Name: ringhash.Name, Config: &ringhash.LBConfig{MinRingSize: 1, MaxRingSize: 2}},
// lrsServer is nil, so LRS policy will not be used.
wantConfig: &clusterimpl.LBConfig{
ChildPolicy: &internalserviceconfig.BalancerConfig{
Expand Down

0 comments on commit 4b75005

Please sign in to comment.