diff --git a/internal/xds/balancer/cdsbalancer/aggregate_cluster_test.go b/internal/xds/balancer/cdsbalancer/aggregate_cluster_test.go index 9cb040d5c3ba..f565ed8c56a7 100644 --- a/internal/xds/balancer/cdsbalancer/aggregate_cluster_test.go +++ b/internal/xds/balancer/cdsbalancer/aggregate_cluster_test.go @@ -90,8 +90,8 @@ func verifyDNSResolution(ctx context.Context, t *testing.T, dnsTargetCh chan res // Tests the case where the cluster resource requested is a leaf cluster. The // management server sends two updates for the same leaf cluster resource. The -// test verifies that the load balancing configuration pushed to the priority LB -// policy contains the expected discovery mechanism corresponding to the leaf +// test verifies that the load balancing configuration pushed to the top-level +// LB policy contains the expected configuration corresponding to the leaf // cluster, on both occasions. func (s) TestAggregateClusterSuccess_LeafNode(t *testing.T) { tests := []struct { @@ -105,43 +105,21 @@ func (s) TestAggregateClusterSuccess_LeafNode(t *testing.T) { name: "eds", firstClusterResource: e2e.DefaultCluster(clusterName, serviceName, e2e.SecurityLevelNone), secondClusterResource: e2e.DefaultCluster(clusterName, serviceName+"-new", e2e.SecurityLevelNone), - wantFirstChildCfg: &priority.LBConfig{ - Children: map[string]*priority.Child{ - "priority-0-0": { - Config: createPriorityConfig(clusterName), - IgnoreReresolutionRequests: true, - }, - }, - Priorities: []string{"priority-0-0"}, - }, - wantSecondChildCfg: &priority.LBConfig{ - Children: map[string]*priority.Child{ - "priority-1-0": { - Config: createPriorityConfig(clusterName), - IgnoreReresolutionRequests: true, - }, - }, - Priorities: []string{"priority-1-0"}, - }, + wantFirstChildCfg: createLeafClusterConfig(clusterName, "priority-0-0", true), + wantSecondChildCfg: createLeafClusterConfig(clusterName, "priority-1-0", true), }, { name: "dns", firstClusterResource: makeLogicalDNSClusterResource(clusterName, "dns_host", uint32(port)), secondClusterResource: makeLogicalDNSClusterResource(clusterName, "dns_host_new", uint32(port)), - wantFirstChildCfg: &priority.LBConfig{ - Children: map[string]*priority.Child{"priority-0": {Config: createPriorityConfig(clusterName)}}, - Priorities: []string{"priority-0"}, - }, - wantSecondChildCfg: &priority.LBConfig{ - Children: map[string]*priority.Child{"priority-1": {Config: createPriorityConfig(clusterName)}}, - Priorities: []string{"priority-1"}, - }, + wantFirstChildCfg: createLeafClusterConfig(clusterName, "priority-0", false), + wantSecondChildCfg: createLeafClusterConfig(clusterName, "priority-1", false), }, } for _, test := range tests { t.Run(test.name, func(t *testing.T) { - lbCfgCh, _, _, _ := registerWrappedPriorityPolicy(t) + lbCfgCh := registerWrappedOutlierDetectionPolicy(t) mgmtServer, nodeID, _ := setupWithManagementServer(t, nil, nil) // Push the first cluster resource through the management server and @@ -277,15 +255,16 @@ func (s) TestAggregateClusterSuccess_ThenUpdateChildClusters(t *testing.T) { // Tests the case where the cluster resource requested is an aggregate cluster // root pointing to two child clusters, one of type EDS and the other of type -// LogicalDNS. The test verifies that the load balancing configuration pushed to -// the priority LB policy contains the discovery mechanisms for both child +// LogicalDNS. The test verifies that the load balancing configuration pushed +// to the priority LB policy contains the discovery mechanisms for both child // clusters. The test then updates the root cluster resource requested by the // cds LB policy to a leaf cluster of type EDS and verifies the load balancing -// configuration pushed to the priority LB policy contains a single discovery -// mechanism. +// configuration pushed to the outlier detection LB policy contains the leaf +// cluster config. func (s) TestAggregateClusterSuccess_ThenChangeRootToEDS(t *testing.T) { dnsTargetCh, dnsR := setupDNS(t) lbCfgCh, _, _, _ := registerWrappedPriorityPolicy(t) + odCfgCh := registerWrappedOutlierDetectionPolicy(t) mgmtServer, nodeID, _ := setupWithManagementServer(t, nil, nil) // Configure the management server with the aggregate cluster resource @@ -332,21 +311,17 @@ func (s) TestAggregateClusterSuccess_ThenChangeRootToEDS(t *testing.T) { }, Endpoints: []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, host, []uint32{port})}, } + // Drain odCfgCh before sending non-aggregate update, as aggregate children pushed outlier configs into odCfgCh. + for len(odCfgCh) > 0 { + <-odCfgCh + } if err := mgmtServer.Update(ctx, resources); err != nil { t.Fatal(err) } // Since the service name of the EDS cluster remains same, same priority name // is used. - wantChildCfg = &priority.LBConfig{ - Children: map[string]*priority.Child{ - "priority-0-0": { - Config: createPriorityConfig(clusterName), - IgnoreReresolutionRequests: true, - }, - }, - Priorities: []string{"priority-0-0"}, - } - if err := compareLoadBalancingConfig(ctx, lbCfgCh, wantChildCfg); err != nil { + wantLeafChildCfg := createLeafClusterConfig(clusterName, "priority-0-0", true) + if err := compareLoadBalancingConfig(ctx, odCfgCh, wantLeafChildCfg); err != nil { t.Fatal(err) } } @@ -354,10 +329,11 @@ func (s) TestAggregateClusterSuccess_ThenChangeRootToEDS(t *testing.T) { // Tests the case where a requested cluster resource switches between being a // leaf and an aggregate cluster pointing to an EDS and LogicalDNS child // cluster. In each of these cases, the test verifies that the load balancing -// configuration pushed to the priority LB policy contains the expected config. +// configuration pushed to the top-level child policy contains the expected config. func (s) TestAggregatedClusterSuccess_SwitchBetweenLeafAndAggregate(t *testing.T) { dnsTargetCh, dnsR := setupDNS(t) lbCfgCh, _, _, _ := registerWrappedPriorityPolicy(t) + odCfgCh := registerWrappedOutlierDetectionPolicy(t) mgmtServer, nodeID, _ := setupWithManagementServer(t, nil, nil) // Start off with the requested cluster being a leaf EDS cluster. @@ -373,16 +349,8 @@ func (s) TestAggregatedClusterSuccess_SwitchBetweenLeafAndAggregate(t *testing.T if err := mgmtServer.Update(ctx, resources); err != nil { t.Fatal(err) } - wantChildCfg := &priority.LBConfig{ - Children: map[string]*priority.Child{ - "priority-0-0": { - Config: createPriorityConfig(clusterName), - IgnoreReresolutionRequests: true, - }, - }, - Priorities: []string{"priority-0-0"}, - } - if err := compareLoadBalancingConfig(ctx, lbCfgCh, wantChildCfg); err != nil { + wantLeafChildCfg := createLeafClusterConfig(clusterName, "priority-0-0", true) + if err := compareLoadBalancingConfig(ctx, odCfgCh, wantLeafChildCfg); err != nil { t.Fatal(err) } @@ -404,7 +372,7 @@ func (s) TestAggregatedClusterSuccess_SwitchBetweenLeafAndAggregate(t *testing.T } verifyDNSResolution(ctx, t, dnsTargetCh, dnsR, dnsHostName, dnsPort) - wantChildCfg = &priority.LBConfig{ + wantChildCfg := &priority.LBConfig{ Children: map[string]*priority.Child{ "priority-0-0": { Config: createPriorityConfig(edsClusterName), @@ -426,19 +394,14 @@ func (s) TestAggregatedClusterSuccess_SwitchBetweenLeafAndAggregate(t *testing.T Clusters: []*v3clusterpb.Cluster{e2e.DefaultCluster(clusterName, serviceName, e2e.SecurityLevelNone)}, Endpoints: []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(serviceName, host, []uint32{port})}, } + // Drain odCfgCh before sending non-aggregate update, as aggregate children pushed outlier configs into odCfgCh. + for len(odCfgCh) > 0 { + <-odCfgCh + } if err := mgmtServer.Update(ctx, resources); err != nil { t.Fatal(err) } - wantChildCfg = &priority.LBConfig{ - Children: map[string]*priority.Child{ - "priority-0-0": { - Config: createPriorityConfig(clusterName), - IgnoreReresolutionRequests: true, - }, - }, - Priorities: []string{"priority-0-0"}, - } - if err := compareLoadBalancingConfig(ctx, lbCfgCh, wantChildCfg); err != nil { + if err := compareLoadBalancingConfig(ctx, odCfgCh, wantLeafChildCfg); err != nil { t.Fatal(err) } } diff --git a/internal/xds/balancer/cdsbalancer/cdsbalancer.go b/internal/xds/balancer/cdsbalancer/cdsbalancer.go index 08b6b451510d..9d6496eeee80 100644 --- a/internal/xds/balancer/cdsbalancer/cdsbalancer.go +++ b/internal/xds/balancer/cdsbalancer/cdsbalancer.go @@ -25,7 +25,6 @@ import ( "google.golang.org/grpc/balancer" "google.golang.org/grpc/balancer/base" "google.golang.org/grpc/connectivity" - "google.golang.org/grpc/internal/balancer/nop" "google.golang.org/grpc/internal/grpclog" "google.golang.org/grpc/internal/pretty" internalserviceconfig "google.golang.org/grpc/internal/serviceconfig" @@ -41,16 +40,20 @@ import ( const cdsName = "cds_experimental" var ( - // newChildBalancer is a helper function to build a new priority balancer - // and will be overridden in unittests. - newChildBalancer = func(cc balancer.ClientConn, opts balancer.BuildOptions) (balancer.Balancer, error) { - builder := balancer.Get(priority.Name) + // newChildBalancer is a helper function to build a new child balancer + // and its config parser, and will be overridden in unittests. + newChildBalancer = func(name string, cc balancer.ClientConn, opts balancer.BuildOptions) (balancer.Balancer, balancer.ConfigParser, error) { + builder := balancer.Get(name) if builder == nil { - return nil, fmt.Errorf("xds: no balancer builder with name %v", priority.Name) + return nil, nil, fmt.Errorf("xds: no balancer builder with name %v", name) } - // We directly pass the parent clientConn to the underlying priority + parser, ok := builder.(balancer.ConfigParser) + if !ok { + return nil, nil, fmt.Errorf("xds: balancer builder for %v does not implement ConfigParser", name) + } + // We directly pass the parent clientConn to the underlying child // balancer because the cdsBalancer does not deal with subConns. - return builder.Build(cc, opts), nil + return builder.Build(cc, opts), parser, nil } ) @@ -65,26 +68,11 @@ type bb struct{} // Build creates a new CDS balancer with the ClientConn. func (bb) Build(cc balancer.ClientConn, opts balancer.BuildOptions) balancer.Balancer { - builder := balancer.Get(priority.Name) - if builder == nil { - // Shouldn't happen, registered through imported Priority builder. Still, - // defensive programming. - logger.Errorf("%q LB policy is needed but not registered", priority.Name) - return nop.NewBalancer(cc, fmt.Errorf("%q LB policy is needed but not registered", priority.Name)) - } - parser, ok := builder.(balancer.ConfigParser) - if !ok { - // Shouldn't happen, imported Priority builder has this method. - logger.Errorf("%q LB policy does not implement a config parser", priority.Name) - return nop.NewBalancer(cc, fmt.Errorf("%q LB policy does not implement a config parser", priority.Name)) - } - b := &cdsBalancer{ - bOpts: opts, - childConfigParser: parser, - clusterConfigs: make(map[string]*xdsresource.ClusterResult), - priorityConfigs: make(map[string]*priorityConfig), - cc: cc, + bOpts: opts, + clusterConfigs: make(map[string]*xdsresource.ClusterResult), + priorityConfigs: make(map[string]*priorityConfig), + cc: cc, } b.logger = prefixLogger(b) b.logger.Infof("Created") @@ -134,6 +122,7 @@ type cdsBalancer struct { // protect access to these fields. xdsClient xdsclient.XDSClient childLB balancer.Balancer // Child policy, built upon resolution of the cluster graph. + childLBName string // Name of the child policy. clusterConfigs map[string]*xdsresource.ClusterResult // Cluster name to the last received result for that cluster. priorityConfigs map[string]*priorityConfig // Hostname to priority config for that leaf cluster. lbCfg *lbConfig // Current load balancing configuration. @@ -267,18 +256,45 @@ func (b *cdsBalancer) handleClusterUpdate() error { // A child policy is created if one doesn't already exist. The newly built // configuration is then pushed to the child policy. func (b *cdsBalancer) updateChildConfig() error { + clusterName := b.lbCfg.ClusterName + clusterConfig := b.clusterConfigs[clusterName].Config + isAggregate := clusterConfig.Cluster.ClusterType == xdsresource.ClusterTypeAggregate + + var topLBName string + if isAggregate { + topLBName = priority.Name + } else { + topLBName = outlierdetection.Name + } + + if b.childLB != nil && b.childLBName != topLBName { + b.childLB.Close() + b.childLB = nil + } + if b.childLB == nil { - childLB, err := newChildBalancer(b.cc, b.bOpts) + childLB, parser, err := newChildBalancer(topLBName, b.cc, b.bOpts) if err != nil { - return fmt.Errorf("failed to create child policy of type %s: %v", priority.Name, err) + return fmt.Errorf("failed to create child policy of type %s: %v", topLBName, err) } b.childLB = childLB + b.childLBName = topLBName + b.childConfigParser = parser } - childCfgBytes, endpoints, err := buildPriorityConfigJSON(b.priorities, &b.xdsLBPolicy) + var childCfgBytes []byte + var endpoints []resolver.Endpoint + var err error + + if isAggregate { + childCfgBytes, endpoints, err = buildAggregateClusterConfigJSON(b.priorities, &b.xdsLBPolicy) + } else { + childCfgBytes, endpoints, err = buildLeafClusterConfigJSON(b.priorities, &b.xdsLBPolicy) + } if err != nil { return fmt.Errorf("failed to build child policy config: %v", err) } + childCfg, err := b.childConfigParser.ParseConfig(childCfgBytes) if err != nil { return fmt.Errorf("failed to parse child policy config. This should never happen because the config was generated: %v", err) diff --git a/internal/xds/balancer/cdsbalancer/cdsbalancer_test.go b/internal/xds/balancer/cdsbalancer/cdsbalancer_test.go index 7e1a9b2201fa..38ab28132053 100644 --- a/internal/xds/balancer/cdsbalancer/cdsbalancer_test.go +++ b/internal/xds/balancer/cdsbalancer/cdsbalancer_test.go @@ -152,11 +152,19 @@ func registerWrappedPriorityPolicy(t *testing.T) (chan serviceconfig.LoadBalanci }, ExitIdle: func(bd *stub.BalancerData) { bd.ChildBalancer.ExitIdle() - close(exitIdleCh) + select { + case <-exitIdleCh: + default: + close(exitIdleCh) + } }, Close: func(bd *stub.BalancerData) { bd.ChildBalancer.Close() - close(closeCh) + select { + case <-closeCh: + default: + close(closeCh) + } }, }) t.Cleanup(func() { balancer.Register(priorityBuilder) }) @@ -164,6 +172,35 @@ func registerWrappedPriorityPolicy(t *testing.T) (chan serviceconfig.LoadBalanci return lbCfgCh, resolverErrCh, exitIdleCh, closeCh } +func registerWrappedOutlierDetectionPolicy(t *testing.T) chan serviceconfig.LoadBalancingConfig { + odBuilder := balancer.Get(outlierdetection.Name) + internal.BalancerUnregister(odBuilder.Name()) + + lbCfgCh := make(chan serviceconfig.LoadBalancingConfig, 1) + + stub.Register(outlierdetection.Name, stub.BalancerFuncs{ + Init: func(bd *stub.BalancerData) { + bd.ChildBalancer = odBuilder.Build(bd.ClientConn, bd.BuildOptions) + }, + ParseConfig: func(lbCfg json.RawMessage) (serviceconfig.LoadBalancingConfig, error) { + return odBuilder.(balancer.ConfigParser).ParseConfig(lbCfg) + }, + UpdateClientConnState: func(bd *stub.BalancerData, ccs balancer.ClientConnState) error { + select { + case lbCfgCh <- ccs.BalancerConfig: + default: + } + return bd.ChildBalancer.UpdateClientConnState(ccs) + }, + Close: func(bd *stub.BalancerData) { + bd.ChildBalancer.Close() + }, + }) + t.Cleanup(func() { balancer.Register(odBuilder) }) + + return lbCfgCh +} + // setupDNS unregisters the DNS resolver and registers a manual resolver for the // same scheme. This allows the test to fake the DNS resolution by supplying the // addresses of the test backends. @@ -236,19 +273,29 @@ func compareLoadBalancingConfig(ctx context.Context, lbCfgCh chan serviceconfig. if err != nil { return fmt.Errorf("failed to marshal expected child config to JSON: %v", err) } - select { - case lbCfg := <-lbCfgCh: - gotJSON, err := json.Marshal(lbCfg) - if err != nil { - return fmt.Errorf("failed to marshal received LB config into JSON: %v", err) - } - if diff := cmp.Diff(wantJSON, gotJSON); diff != "" { - return fmt.Errorf("child policy received unexpected diff in config (-want +got):\n%s", diff) + var lastErr error + // Loop over updates received on lbCfgCh to consume intermediate configuration + // updates (e.g. pushed as individual discovery mechanisms resolve) until a + // configuration matching wantChildCfg is received, or the context deadline expires. + for { + select { + case lbCfg := <-lbCfgCh: + gotJSON, err := json.Marshal(lbCfg) + if err != nil { + return fmt.Errorf("failed to marshal received LB config into JSON: %v", err) + } + if diff := cmp.Diff(wantJSON, gotJSON); diff != "" { + lastErr = fmt.Errorf("child policy received unexpected diff in config (-want +got):\n%s", diff) + continue + } + return nil + case <-ctx.Done(): + if lastErr != nil { + return lastErr + } + return fmt.Errorf("timeout when waiting for child policy to receive its configuration") } - case <-ctx.Done(): - return fmt.Errorf("timeout when waiting for child policy to receive its configuration") } - return nil } func verifyRPCError(gotErr error, wantCode codes.Code, wantErr, wantNodeID string) error { @@ -290,6 +337,41 @@ func createPriorityConfig(cluster string) *iserviceconfig.BalancerConfig { } } +// createLeafClusterConfig returns the expected LoadBalancingConfig tree for a +// leaf (non-aggregate) cluster under gRFC A75 topology: +// outlier_detection -> cluster_impl -> priority -> wrr_locality -> round_robin. +func createLeafClusterConfig(cluster string, pName string, ignoreReresolution bool) *outlierdetection.LBConfig { + return &outlierdetection.LBConfig{ + Interval: iserviceconfig.Duration(10 * time.Second), + BaseEjectionTime: iserviceconfig.Duration(30 * time.Second), + MaxEjectionTime: iserviceconfig.Duration(300 * time.Second), + MaxEjectionPercent: 10, + ChildPolicy: &iserviceconfig.BalancerConfig{ + Name: clusterimpl.Name, + Config: &clusterimpl.LBConfig{ + Cluster: cluster, + ChildPolicy: &iserviceconfig.BalancerConfig{ + Name: priority.Name, + Config: &priority.LBConfig{ + Children: map[string]*priority.Child{ + pName: { + Config: &iserviceconfig.BalancerConfig{ + Name: wrrlocality.Name, + Config: &wrrlocality.LBConfig{ + ChildPolicy: &iserviceconfig.BalancerConfig{Name: roundrobin.Name}, + }, + }, + IgnoreReresolutionRequests: ignoreReresolution, + }, + }, + Priorities: []string{pName}, + }, + }, + }, + }, + } +} + // Tests the case where a configuration with an empty cluster name is pushed to // the CDS LB policy. Verifies that ErrBadResolverState is returned. func (s) TestConfigurationUpdate_EmptyCluster(t *testing.T) { @@ -414,32 +496,32 @@ func (s) TestClusterUpdate_Success(t *testing.T) { } return c }(), - wantChildCfg: &priority.LBConfig{ - Children: map[string]*priority.Child{ - "priority-0-0": { - Config: &iserviceconfig.BalancerConfig{ - Name: outlierdetection.Name, - Config: &outlierdetection.LBConfig{ - Interval: iserviceconfig.Duration(10 * time.Second), // default interval - BaseEjectionTime: iserviceconfig.Duration(30 * time.Second), - MaxEjectionTime: iserviceconfig.Duration(300 * time.Second), - MaxEjectionPercent: 10, - ChildPolicy: &iserviceconfig.BalancerConfig{ - Name: clusterimpl.Name, - Config: &clusterimpl.LBConfig{ - Cluster: clusterName, - ChildPolicy: &iserviceconfig.BalancerConfig{ + wantChildCfg: &outlierdetection.LBConfig{ + Interval: iserviceconfig.Duration(10 * time.Second), // default interval + BaseEjectionTime: iserviceconfig.Duration(30 * time.Second), + MaxEjectionTime: iserviceconfig.Duration(300 * time.Second), + MaxEjectionPercent: 10, + ChildPolicy: &iserviceconfig.BalancerConfig{ + Name: clusterimpl.Name, + Config: &clusterimpl.LBConfig{ + Cluster: clusterName, + ChildPolicy: &iserviceconfig.BalancerConfig{ + Name: priority.Name, + Config: &priority.LBConfig{ + Children: map[string]*priority.Child{ + "priority-0-0": { + Config: &iserviceconfig.BalancerConfig{ Name: wrrlocality.Name, Config: &wrrlocality.LBConfig{ChildPolicy: &iserviceconfig.BalancerConfig{Name: roundrobin.Name}}, }, + IgnoreReresolutionRequests: true, }, }, + Priorities: []string{"priority-0-0"}, }, }, - IgnoreReresolutionRequests: true, }, }, - Priorities: []string{"priority-0-0"}, }, }, { @@ -459,35 +541,35 @@ func (s) TestClusterUpdate_Success(t *testing.T) { } return c }(), - wantChildCfg: &priority.LBConfig{ - Children: map[string]*priority.Child{ - "priority-0-0": { - Config: &iserviceconfig.BalancerConfig{ - Name: outlierdetection.Name, - Config: &outlierdetection.LBConfig{ - Interval: iserviceconfig.Duration(10 * time.Second), // default interval - BaseEjectionTime: iserviceconfig.Duration(30 * time.Second), - MaxEjectionTime: iserviceconfig.Duration(300 * time.Second), - MaxEjectionPercent: 10, - ChildPolicy: &iserviceconfig.BalancerConfig{ - Name: clusterimpl.Name, - Config: &clusterimpl.LBConfig{ - Cluster: clusterName, - ChildPolicy: &iserviceconfig.BalancerConfig{ + wantChildCfg: &outlierdetection.LBConfig{ + Interval: iserviceconfig.Duration(10 * time.Second), // default interval + BaseEjectionTime: iserviceconfig.Duration(30 * time.Second), + MaxEjectionTime: iserviceconfig.Duration(300 * time.Second), + MaxEjectionPercent: 10, + ChildPolicy: &iserviceconfig.BalancerConfig{ + Name: clusterimpl.Name, + Config: &clusterimpl.LBConfig{ + Cluster: clusterName, + ChildPolicy: &iserviceconfig.BalancerConfig{ + Name: priority.Name, + Config: &priority.LBConfig{ + Children: map[string]*priority.Child{ + "priority-0-0": { + Config: &iserviceconfig.BalancerConfig{ Name: ringhash.Name, Config: &iringhash.LBConfig{ MinRingSize: 100, MaxRingSize: 1000, }, }, + IgnoreReresolutionRequests: true, }, }, + Priorities: []string{"priority-0-0"}, }, }, - IgnoreReresolutionRequests: true, }, }, - Priorities: []string{"priority-0-0"}, }, }, { @@ -502,41 +584,41 @@ func (s) TestClusterUpdate_Success(t *testing.T) { c.OutlierDetection = &v3clusterpb.OutlierDetection{} return c }(), - wantChildCfg: &priority.LBConfig{ - Children: map[string]*priority.Child{ - "priority-0-0": { - Config: &iserviceconfig.BalancerConfig{ - Name: outlierdetection.Name, - Config: &outlierdetection.LBConfig{ - Interval: iserviceconfig.Duration(10 * time.Second), // default interval - BaseEjectionTime: iserviceconfig.Duration(30 * time.Second), - MaxEjectionTime: iserviceconfig.Duration(300 * time.Second), - MaxEjectionPercent: 10, - SuccessRateEjection: &outlierdetection.SuccessRateEjection{ - StdevFactor: 1900, - EnforcementPercentage: 100, - MinimumHosts: 5, - RequestVolume: 100, - }, - ChildPolicy: &iserviceconfig.BalancerConfig{ - Name: clusterimpl.Name, - Config: &clusterimpl.LBConfig{ - Cluster: clusterName, - ChildPolicy: &iserviceconfig.BalancerConfig{ + wantChildCfg: &outlierdetection.LBConfig{ + Interval: iserviceconfig.Duration(10 * time.Second), // default interval + BaseEjectionTime: iserviceconfig.Duration(30 * time.Second), + MaxEjectionTime: iserviceconfig.Duration(300 * time.Second), + MaxEjectionPercent: 10, + SuccessRateEjection: &outlierdetection.SuccessRateEjection{ + StdevFactor: 1900, + EnforcementPercentage: 100, + MinimumHosts: 5, + RequestVolume: 100, + }, + ChildPolicy: &iserviceconfig.BalancerConfig{ + Name: clusterimpl.Name, + Config: &clusterimpl.LBConfig{ + Cluster: clusterName, + ChildPolicy: &iserviceconfig.BalancerConfig{ + Name: priority.Name, + Config: &priority.LBConfig{ + Children: map[string]*priority.Child{ + "priority-0-0": { + Config: &iserviceconfig.BalancerConfig{ Name: ringhash.Name, Config: &iringhash.LBConfig{ MinRingSize: 1024, // default sizes MaxRingSize: 4096, }, }, + IgnoreReresolutionRequests: true, }, }, + Priorities: []string{"priority-0-0"}, }, }, - IgnoreReresolutionRequests: true, }, }, - Priorities: []string{"priority-0-0"}, }, }, { @@ -564,54 +646,54 @@ func (s) TestClusterUpdate_Success(t *testing.T) { } return c }(), - wantChildCfg: &priority.LBConfig{ - Children: map[string]*priority.Child{ - "priority-0-0": { - Config: &iserviceconfig.BalancerConfig{ - Name: outlierdetection.Name, - Config: &outlierdetection.LBConfig{ - Interval: iserviceconfig.Duration(10 * time.Second), // default interval - BaseEjectionTime: iserviceconfig.Duration(30 * time.Second), - MaxEjectionTime: iserviceconfig.Duration(300 * time.Second), - MaxEjectionPercent: 10, - SuccessRateEjection: &outlierdetection.SuccessRateEjection{ - StdevFactor: 1900, - EnforcementPercentage: 100, - MinimumHosts: 5, - RequestVolume: 100, - }, - FailurePercentageEjection: &outlierdetection.FailurePercentageEjection{ - Threshold: 85, - EnforcementPercentage: 5, - MinimumHosts: 5, - RequestVolume: 50, - }, - ChildPolicy: &iserviceconfig.BalancerConfig{ - Name: clusterimpl.Name, - Config: &clusterimpl.LBConfig{ - Cluster: clusterName, - ChildPolicy: &iserviceconfig.BalancerConfig{ + wantChildCfg: &outlierdetection.LBConfig{ + Interval: iserviceconfig.Duration(10 * time.Second), // default interval + BaseEjectionTime: iserviceconfig.Duration(30 * time.Second), + MaxEjectionTime: iserviceconfig.Duration(300 * time.Second), + MaxEjectionPercent: 10, + SuccessRateEjection: &outlierdetection.SuccessRateEjection{ + StdevFactor: 1900, + EnforcementPercentage: 100, + MinimumHosts: 5, + RequestVolume: 100, + }, + FailurePercentageEjection: &outlierdetection.FailurePercentageEjection{ + Threshold: 85, + EnforcementPercentage: 5, + MinimumHosts: 5, + RequestVolume: 50, + }, + ChildPolicy: &iserviceconfig.BalancerConfig{ + Name: clusterimpl.Name, + Config: &clusterimpl.LBConfig{ + Cluster: clusterName, + ChildPolicy: &iserviceconfig.BalancerConfig{ + Name: priority.Name, + Config: &priority.LBConfig{ + Children: map[string]*priority.Child{ + "priority-0-0": { + Config: &iserviceconfig.BalancerConfig{ Name: ringhash.Name, Config: &iringhash.LBConfig{ MinRingSize: 1024, // default sizes MaxRingSize: 4096, }, }, + IgnoreReresolutionRequests: true, }, }, + Priorities: []string{"priority-0-0"}, }, }, - IgnoreReresolutionRequests: true, }, }, - Priorities: []string{"priority-0-0"}, }, }, } for _, test := range tests { t.Run(test.name, func(t *testing.T) { - lbCfgCh, _, _, _ := registerWrappedPriorityPolicy(t) + lbCfgCh := registerWrappedOutlierDetectionPolicy(t) mgmtServer, nodeID, _ := setupWithManagementServer(t, nil, nil) ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout) diff --git a/internal/xds/balancer/cdsbalancer/configbuilder.go b/internal/xds/balancer/cdsbalancer/configbuilder.go index 6efeeb14b65e..65e610b93766 100644 --- a/internal/xds/balancer/cdsbalancer/configbuilder.go +++ b/internal/xds/balancer/cdsbalancer/configbuilder.go @@ -77,10 +77,103 @@ func hostName(clusterName string, update xdsresource.ClusterUpdate) string { } } -// buildPriorityConfigJSON builds balancer config for the passed in -// priorities. +// buildLeafClusterConfigJSON builds the balancer config for a leaf +// (non-aggregate) cluster according to gRFC A75. // -// The built tree of balancers (see test for the output struct). +// The built tree of balancers: +// +// ┌─────────────────┐ +// │outlier_detection│ +// └────────┬────────┘ +// │ +// ┌────────▼───────┐ +// │ cluster_impl │ +// └────────┬───────┘ +// │ +// ┌───▼────┐ +// │priority│ +// └┬──────┬┘ +// │ │ +// ┌──────────▼─┐ ┌─▼──────────┐ +// │xDSLBPolicy │ │xDSLBPolicy │ (Locality and Endpoint picking layer) +// └────────────┘ └────────────┘ +func buildLeafClusterConfigJSON(priorities []*priorityConfig, xdsLBPolicy *internalserviceconfig.BalancerConfig) ([]byte, []resolver.Endpoint, error) { + odCfg, endpoints, err := buildLeafClusterConfig(priorities[0], xdsLBPolicy) + if err != nil { + return nil, nil, err + } + ret, err := json.Marshal(odCfg) + if err != nil { + return nil, nil, fmt.Errorf("failed to marshal built leaf cluster config: %v", err) + } + return ret, endpoints, nil +} + +func buildLeafClusterConfig(p *priorityConfig, xdsLBPolicy *internalserviceconfig.BalancerConfig) (*outlierdetection.LBConfig, []resolver.Endpoint, error) { + clusterUpdate := p.clusterConfig.Cluster + priorityLBConfig := &priority.LBConfig{ + Children: make(map[string]*priority.Child), + } + var retEndpoints []resolver.Endpoint + + switch clusterUpdate.ClusterType { + case xdsresource.ClusterTypeEDS: + priorities := [][]xdsresource.Locality{{}} + edsUpdate := p.clusterConfig.EndpointConfig.EDSUpdate + if len(edsUpdate.Localities) != 0 { + priorities = groupLocalitiesByPriority(edsUpdate.Localities) + } + priorityNames := p.childNameGen.generate(priorities) + priorityLBConfig.Priorities = priorityNames + + for i, pName := range priorityNames { + priorityLocalities := priorities[i] + endpoints := priorityLocalitiesToEndpoints(priorityLocalities, pName, *p.clusterConfig.Cluster) + retEndpoints = append(retEndpoints, endpoints...) + priorityLBConfig.Children[pName] = &priority.Child{ + Config: xdsLBPolicy, + IgnoreReresolutionRequests: true, + } + } + case xdsresource.ClusterTypeLogicalDNS: + pName := fmt.Sprintf("priority-%v", p.childNameGen.prefix) + priorityLBConfig.Priorities = []string{pName} + endpoints := p.clusterConfig.EndpointConfig.DNSEndpoints.Endpoints + var retEndpoint resolver.Endpoint + for _, e := range endpoints { + retEndpoint.Addresses = append(retEndpoint.Addresses, e.Addresses...) + } + localityStr := xdsinternal.LocalityString(clients.Locality{}) + retEndpoint = hostname.Set(hierarchy.SetInEndpoint(retEndpoint, []string{pName, localityStr}), clusterUpdate.DNSHostName) + retEndpoint = wrrlocality.SetAddrInfo(retEndpoint, wrrlocality.AddrInfo{LocalityWeight: 1}) + retEndpoints = append(retEndpoints, retEndpoint) + priorityLBConfig.Children[pName] = &priority.Child{ + Config: xdsLBPolicy, + IgnoreReresolutionRequests: false, + } + } + + ciCfg := &clusterimpl.LBConfig{ + Cluster: clusterUpdate.ClusterName, + ChildPolicy: &internalserviceconfig.BalancerConfig{ + Name: priority.Name, + Config: priorityLBConfig, + }, + } + + odCfg := p.outlierDetection + odCfg.ChildPolicy = &internalserviceconfig.BalancerConfig{ + Name: clusterimpl.Name, + Config: ciCfg, + } + + return &odCfg, retEndpoints, nil +} + +// buildAggregateClusterConfigJSON builds balancer config for the passed in +// priorities (legacy / aggregate cluster tree). +// +// The built tree of balancers: // // ┌────────┐ // │priority│ @@ -93,8 +186,8 @@ func hostName(clusterName string, update xdsresource.ClusterUpdate) string { // ┌──────▼─────┐ ┌─────▼──────┐ // │xDSLBPolicy │ │xDSLBPolicy │ (Locality and Endpoint picking layer) // └────────────┘ └────────────┘ -func buildPriorityConfigJSON(priorities []*priorityConfig, xdsLBPolicy *internalserviceconfig.BalancerConfig) ([]byte, []resolver.Endpoint, error) { - pc, endpoints, err := buildPriorityConfig(priorities, xdsLBPolicy) +func buildAggregateClusterConfigJSON(priorities []*priorityConfig, xdsLBPolicy *internalserviceconfig.BalancerConfig) ([]byte, []resolver.Endpoint, error) { + pc, endpoints, err := buildAggregateClusterConfig(priorities, xdsLBPolicy) if err != nil { return nil, nil, fmt.Errorf("failed to build priority config: %v", err) } @@ -105,7 +198,7 @@ func buildPriorityConfigJSON(priorities []*priorityConfig, xdsLBPolicy *internal return ret, endpoints, nil } -func buildPriorityConfig(priorities []*priorityConfig, xdsLBPolicy *internalserviceconfig.BalancerConfig) (*priority.LBConfig, []resolver.Endpoint, error) { +func buildAggregateClusterConfig(priorities []*priorityConfig, xdsLBPolicy *internalserviceconfig.BalancerConfig) (*priority.LBConfig, []resolver.Endpoint, error) { var ( retConfig = &priority.LBConfig{Children: make(map[string]*priority.Child)} retEndpoints []resolver.Endpoint @@ -258,6 +351,18 @@ func groupLocalitiesByPriority(localities []xdsresource.Locality) [][]xdsresourc // addresses with their path hierarchy set to [priority-name, locality-name], so // priority and the xDS LB Policy know which child policy each address is for. func priorityLocalitiesToClusterImpl(localities []xdsresource.Locality, priorityName string, clusterUpdate xdsresource.ClusterUpdate, xdsLBPolicy *internalserviceconfig.BalancerConfig) (*clusterimpl.LBConfig, []resolver.Endpoint, error) { + endpoints := priorityLocalitiesToEndpoints(localities, priorityName, clusterUpdate) + return &clusterimpl.LBConfig{ + Cluster: clusterUpdate.ClusterName, + ChildPolicy: xdsLBPolicy, + }, endpoints, nil +} + +// priorityLocalitiesToEndpoints takes a list of localities (with the same +// priority), and generates a list of addresses with their path hierarchy set to +// [priority-name, locality-name], so priority and the xDS LB Policy know which +// child policy each address is for. +func priorityLocalitiesToEndpoints(localities []xdsresource.Locality, priorityName string, clusterUpdate xdsresource.ClusterUpdate) []resolver.Endpoint { var retEndpoints []resolver.Endpoint // Compute the sum of locality weights to normalize locality weights. The @@ -336,10 +441,7 @@ func priorityLocalitiesToClusterImpl(localities []xdsresource.Locality, priority retEndpoints = append(retEndpoints, resolverEndpoint) } } - return &clusterimpl.LBConfig{ - Cluster: clusterUpdate.ClusterName, - ChildPolicy: xdsLBPolicy, - }, retEndpoints, nil + return retEndpoints } // fixedPointFractionalBits is the number of bits used for the fractional part diff --git a/internal/xds/balancer/cdsbalancer/configbuilder_test.go b/internal/xds/balancer/cdsbalancer/configbuilder_test.go index 282e2a49f6ea..e353fbccfbec 100644 --- a/internal/xds/balancer/cdsbalancer/configbuilder_test.go +++ b/internal/xds/balancer/cdsbalancer/configbuilder_test.go @@ -130,9 +130,10 @@ func makeLocality(localityIdx int, localityWeight, priority uint32, endpointCoun } } -// TestBuildPriorityConfigJSON is a sanity check that the built balancer config -// can be parsed. The behavior test is covered by TestBuildPriorityConfig. -func (s) TestBuildPriorityConfigJSON(t *testing.T) { +// TestBuildClusterConfigJSON is a sanity check that built balancer configs +// for aggregate and leaf clusters can be parsed by their respective policy +// parsers. +func (s) TestBuildClusterConfigJSON(t *testing.T) { testLRSServerConfig, err := bootstrap.ServerConfigForTesting(bootstrap.ServerConfigTestingOptions{ URI: "trafficdirector.googleapis.com:443", ChannelCreds: []bootstrap.ChannelCreds{{Type: "google_default"}}, @@ -141,68 +142,121 @@ func (s) TestBuildPriorityConfigJSON(t *testing.T) { t.Fatalf("Failed to create LRS server config for testing: %v", err) } - gotConfig, _, err := buildPriorityConfigJSON([]*priorityConfig{ + tests := []struct { + name string + buildFunc func([]*priorityConfig, *iserviceconfig.BalancerConfig) ([]byte, []resolver.Endpoint, error) + priorities []*priorityConfig + xdsLBPolicy *iserviceconfig.BalancerConfig + policyName string + }{ { - clusterConfig: &xdsresource.ClusterConfig{ - Cluster: &xdsresource.ClusterUpdate{ - ClusterName: testClusterName, - ClusterType: xdsresource.ClusterTypeEDS, - EDSServiceName: testEDSServiceName, - MaxRequests: newUint32(testMaxRequests), - LRSServerConfig: testLRSServerConfig, + name: "aggregate", + buildFunc: buildAggregateClusterConfigJSON, + priorities: []*priorityConfig{ + { + clusterConfig: &xdsresource.ClusterConfig{ + Cluster: &xdsresource.ClusterUpdate{ + ClusterName: testClusterName, + ClusterType: xdsresource.ClusterTypeEDS, + EDSServiceName: testEDSServiceName, + MaxRequests: newUint32(testMaxRequests), + LRSServerConfig: testLRSServerConfig, + }, + EndpointConfig: &xdsresource.EndpointConfig{ + EDSUpdate: &xdsresource.EndpointsUpdate{ + Drops: []xdsresource.OverloadDropConfig{{ + Category: testDropCategory, + Numerator: testDropOverMillion, + Denominator: million, + }}, + Localities: []xdsresource.Locality{ + makeLocality(0, 20, 0, 2), + makeLocality(1, 80, 0, 2), + makeLocality(2, 20, 1, 2), + makeLocality(3, 80, 1, 2), + }, + }, + }, + }, + childNameGen: newNameGenerator(0), }, - EndpointConfig: &xdsresource.EndpointConfig{ - EDSUpdate: &xdsresource.EndpointsUpdate{ - Drops: []xdsresource.OverloadDropConfig{{ - Category: testDropCategory, - Numerator: testDropOverMillion, - Denominator: million, - }}, - Localities: []xdsresource.Locality{ - makeLocality(0, 20, 0, 2), - makeLocality(1, 80, 0, 2), - makeLocality(2, 20, 1, 2), - makeLocality(3, 80, 1, 2), + { + clusterConfig: &xdsresource.ClusterConfig{ + Cluster: &xdsresource.ClusterUpdate{ + ClusterType: xdsresource.ClusterTypeLogicalDNS, + }, + EndpointConfig: &xdsresource.EndpointConfig{ + DNSEndpoints: &xdsresource.DNSUpdate{Endpoints: []resolver.Endpoint{makeResolverEndpoint(4, 0), makeResolverEndpoint(4, 1)}}, }, }, + childNameGen: newNameGenerator(1), }, }, - childNameGen: newNameGenerator(0), + xdsLBPolicy: nil, + policyName: priority.Name, }, { - clusterConfig: &xdsresource.ClusterConfig{ - Cluster: &xdsresource.ClusterUpdate{ - ClusterType: xdsresource.ClusterTypeLogicalDNS, - }, - EndpointConfig: &xdsresource.EndpointConfig{ - DNSEndpoints: &xdsresource.DNSUpdate{Endpoints: []resolver.Endpoint{makeResolverEndpoint(4, 0), makeResolverEndpoint(4, 1)}}, + name: "leaf", + buildFunc: buildLeafClusterConfigJSON, + priorities: []*priorityConfig{{ + clusterConfig: &xdsresource.ClusterConfig{ + Cluster: &xdsresource.ClusterUpdate{ + ClusterName: testClusterName, + ClusterType: xdsresource.ClusterTypeEDS, + EDSServiceName: testEDSServiceName, + MaxRequests: newUint32(testMaxRequests), + LRSServerConfig: testLRSServerConfig, + }, + EndpointConfig: &xdsresource.EndpointConfig{ + EDSUpdate: &xdsresource.EndpointsUpdate{ + Drops: []xdsresource.OverloadDropConfig{{ + Category: testDropCategory, + Numerator: testDropOverMillion, + Denominator: million, + }}, + Localities: []xdsresource.Locality{ + makeLocality(0, 20, 0, 2), + makeLocality(1, 80, 0, 2), + makeLocality(2, 20, 1, 2), + makeLocality(3, 80, 1, 2), + }, + }, + }, }, - }, - childNameGen: newNameGenerator(1), + outlierDetection: noopODCfg, + childNameGen: newNameGenerator(0), + }}, + xdsLBPolicy: &iserviceconfig.BalancerConfig{Name: roundrobin.Name}, + policyName: outlierdetection.Name, }, - }, nil) - if err != nil { - t.Fatalf("buildPriorityConfigJSON(...) failed: %v", err) } - var prettyGot bytes.Buffer - if err := json.Indent(&prettyGot, gotConfig, ">>> ", " "); err != nil { - t.Fatalf("json.Indent() failed: %v", err) - } - // Print the indented json if this test fails. - t.Log(prettyGot.String()) + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + gotConfig, _, err := tt.buildFunc(tt.priorities, tt.xdsLBPolicy) + if err != nil { + t.Fatalf("%s(...) failed: %v", tt.name, err) + } - priorityB := balancer.Get(priority.Name) - if _, err = priorityB.(balancer.ConfigParser).ParseConfig(gotConfig); err != nil { - t.Fatalf("ParseConfig(%+v) failed: %v", gotConfig, err) + var prettyGot bytes.Buffer + if err := json.Indent(&prettyGot, gotConfig, ">>> ", " "); err != nil { + t.Fatalf("json.Indent() failed: %v", err) + } + t.Log(prettyGot.String()) + + b := balancer.Get(tt.policyName) + if _, err = b.(balancer.ConfigParser).ParseConfig(gotConfig); err != nil { + t.Fatalf("ParseConfig(%+v) failed: %v", gotConfig, err) + } + }) } } -// TestBuildPriorityConfig tests the priority config generation. Each top level +// TestBuildAggregateClusterConfig tests the priority config generation. Each top level // balancer per priority should be an Outlier Detection balancer, with a Cluster // Impl Balancer as a child. -func (s) TestBuildPriorityConfig(t *testing.T) { - gotConfig, _, _ := buildPriorityConfig([]*priorityConfig{ +func (s) TestBuildAggregateClusterConfig(t *testing.T) { + gotConfig, _, _ := buildAggregateClusterConfig([]*priorityConfig{ { // EDS - OD config should be the top level for both of the EDS // priorities balancer This EDS priority will have multiple sub @@ -317,6 +371,329 @@ func testEndpointForDNS(endpoints []resolver.Endpoint, localityWeight uint32, pa return retEndpoint } +// TestBuildLeafClusterConfig_DNS tests buildLeafClusterConfig with a +// LOGICAL_DNS cluster configuration, verifying the expected policy hierarchy +// and endpoints. +func (s) TestBuildLeafClusterConfig_DNS(t *testing.T) { + for _, tt := range []struct { + name string + endpoints []resolver.Endpoint + xdsLBPolicy *iserviceconfig.BalancerConfig + }{ + { + name: "one_endpoint_one_address", + endpoints: []resolver.Endpoint{{Addresses: []resolver.Address{{Addr: "addr-0-0"}}}}, + xdsLBPolicy: &iserviceconfig.BalancerConfig{Name: pickfirst.Name}, + }, + { + name: "one_endpoint_multiple_addresses", + endpoints: []resolver.Endpoint{{Addresses: []resolver.Address{ + {Addr: "addr-0-0"}, + {Addr: "addr-0-1"}, + }}}, + xdsLBPolicy: &iserviceconfig.BalancerConfig{Name: wrrlocality.Name}, + }, + { + name: "multiple_endpoints_one_address_each", + endpoints: []resolver.Endpoint{ + {Addresses: []resolver.Address{{Addr: "addr-0-0"}}}, + {Addresses: []resolver.Address{{Addr: "addr-0-1"}}}, + }, + xdsLBPolicy: &iserviceconfig.BalancerConfig{Name: roundrobin.Name}, + }, + { + name: "multiple_endpoints_multiple_addresses", + endpoints: []resolver.Endpoint{ + {Addresses: []resolver.Address{ + {Addr: "addr-0-0"}, + {Addr: "addr-0-1"}, + }}, + {Addresses: []resolver.Address{ + {Addr: "addr-1-0"}, + {Addr: "addr-1-1"}, + }}, + }, + xdsLBPolicy: &iserviceconfig.BalancerConfig{Name: roundrobin.Name}, + }, + } { + t.Run(tt.name, func(t *testing.T) { + gotODConfig, gotEndpoints, err := buildLeafClusterConfig( + &priorityConfig{ + clusterConfig: &xdsresource.ClusterConfig{ + Cluster: &xdsresource.ClusterUpdate{ + ClusterName: testClusterName2, + ClusterType: xdsresource.ClusterTypeLogicalDNS, + }, + EndpointConfig: &xdsresource.EndpointConfig{DNSEndpoints: &xdsresource.DNSUpdate{Endpoints: tt.endpoints}}, + }, + outlierDetection: noopODCfg, + childNameGen: newNameGenerator(3), + }, + tt.xdsLBPolicy) + if err != nil { + t.Fatalf("buildLeafClusterConfig() failed: %v", err) + } + + wantODConfig := &outlierdetection.LBConfig{ + Interval: iserviceconfig.Duration(10 * time.Second), + BaseEjectionTime: iserviceconfig.Duration(30 * time.Second), + MaxEjectionTime: iserviceconfig.Duration(300 * time.Second), + MaxEjectionPercent: 10, + ChildPolicy: &iserviceconfig.BalancerConfig{ + Name: clusterimpl.Name, + Config: &clusterimpl.LBConfig{ + Cluster: testClusterName2, + ChildPolicy: &iserviceconfig.BalancerConfig{ + Name: priority.Name, + Config: &priority.LBConfig{ + Children: map[string]*priority.Child{ + "priority-3": { + Config: tt.xdsLBPolicy, + IgnoreReresolutionRequests: false, + }, + }, + Priorities: []string{"priority-3"}, + }, + }, + }, + }, + } + if diff := cmp.Diff(wantODConfig, gotODConfig); diff != "" { + t.Errorf("buildLeafClusterConfig() config diff (-want +got) %v", diff) + } + + wantEndpoints := []resolver.Endpoint{testEndpointForDNS(tt.endpoints, 1, []string{"priority-3", xdsinternal.LocalityString(clients.Locality{})})} + if diff := cmp.Diff(wantEndpoints, gotEndpoints, endpointCmpOpts); diff != "" { + t.Errorf("buildLeafClusterConfig() endpoints diff (-want +got) %v", diff) + } + }) + } +} + +// TestBuildLeafClusterConfig_EDS_PickFirstWeightedShuffling_Disabled tests +// buildLeafClusterConfig with an EDS cluster configuration when pick_first +// weighted shuffling is disabled. +func (s) TestBuildLeafClusterConfig_EDS_PickFirstWeightedShuffling_Disabled(t *testing.T) { + testutils.SetEnvConfig(t, &envconfig.PickFirstWeightedShuffling, false) + + testLRSServerConfig, err := bootstrap.ServerConfigForTesting(bootstrap.ServerConfigTestingOptions{ + URI: "trafficdirector.googleapis.com:443", + ChannelCreds: []bootstrap.ChannelCreds{{Type: "google_default"}}, + }) + if err != nil { + t.Fatalf("Failed to create LRS server config for testing: %v", err) + } + + // Create test localities with 2 endpoints each. + // Localities are passed in shuffled order to verify sorting by priority. + loc0 := makeLocality(0, 20, 0, 2) + loc1 := makeLocality(1, 80, 0, 2) + loc2 := makeLocality(2, 20, 1, 2) + loc3 := makeLocality(3, 80, 1, 2) + + gotODConfig, gotEndpoints, err := buildLeafClusterConfig( + &priorityConfig{ + clusterConfig: &xdsresource.ClusterConfig{ + Cluster: &xdsresource.ClusterUpdate{ + ClusterName: testClusterName, + ClusterType: xdsresource.ClusterTypeEDS, + EDSServiceName: testEDSServiceName, + LRSServerConfig: testLRSServerConfig, + MaxRequests: newUint32(testMaxRequests), + }, + EndpointConfig: &xdsresource.EndpointConfig{ + EDSUpdate: &xdsresource.EndpointsUpdate{ + Drops: []xdsresource.OverloadDropConfig{{ + Category: testDropCategory, + Numerator: testDropOverMillion, + Denominator: million, + }}, + Localities: []xdsresource.Locality{ + makeLocality(3, 80, 1, 2), + makeLocality(1, 80, 0, 2), + makeLocality(2, 20, 1, 2), + makeLocality(0, 20, 0, 2), + }, + }, + }, + }, + outlierDetection: noopODCfg, + childNameGen: newNameGenerator(2), + }, + nil, + ) + if err != nil { + t.Fatalf("buildLeafClusterConfig() failed: %v", err) + } + + wantODConfig := &outlierdetection.LBConfig{ + Interval: iserviceconfig.Duration(10 * time.Second), + BaseEjectionTime: iserviceconfig.Duration(30 * time.Second), + MaxEjectionTime: iserviceconfig.Duration(300 * time.Second), + MaxEjectionPercent: 10, + ChildPolicy: &iserviceconfig.BalancerConfig{ + Name: clusterimpl.Name, + Config: &clusterimpl.LBConfig{ + Cluster: testClusterName, + ChildPolicy: &iserviceconfig.BalancerConfig{ + Name: priority.Name, + Config: &priority.LBConfig{ + Children: map[string]*priority.Child{ + "priority-2-0": { + IgnoreReresolutionRequests: true, + }, + "priority-2-1": { + IgnoreReresolutionRequests: true, + }, + }, + Priorities: []string{"priority-2-0", "priority-2-1"}, + }, + }, + }, + }, + } + // Endpoint weight is the product of locality weight and endpoint weight. + wantEndpoints := []resolver.Endpoint{ + testEndpointWithAttrs(loc0.Endpoints[0].ResolverEndpoint, 20, 20*1, "priority-2-0", &loc0.ID), + testEndpointWithAttrs(loc0.Endpoints[1].ResolverEndpoint, 20, 20*1, "priority-2-0", &loc0.ID), + testEndpointWithAttrs(loc1.Endpoints[0].ResolverEndpoint, 80, 80*1, "priority-2-0", &loc1.ID), + testEndpointWithAttrs(loc1.Endpoints[1].ResolverEndpoint, 80, 80*1, "priority-2-0", &loc1.ID), + testEndpointWithAttrs(loc2.Endpoints[0].ResolverEndpoint, 20, 20*1, "priority-2-1", &loc2.ID), + testEndpointWithAttrs(loc2.Endpoints[1].ResolverEndpoint, 20, 20*1, "priority-2-1", &loc2.ID), + testEndpointWithAttrs(loc3.Endpoints[0].ResolverEndpoint, 80, 80*1, "priority-2-1", &loc3.ID), + testEndpointWithAttrs(loc3.Endpoints[1].ResolverEndpoint, 80, 80*1, "priority-2-1", &loc3.ID), + } + + if diff := cmp.Diff(wantODConfig, gotODConfig); diff != "" { + t.Errorf("buildLeafClusterConfig() config diff (-want +got) %v", diff) + } + if diff := cmp.Diff(wantEndpoints, gotEndpoints, endpointCmpOpts); diff != "" { + t.Errorf("buildLeafClusterConfig() endpoints diff (-want +got) %v", diff) + } +} + +// TestBuildLeafClusterConfig_EDS_PickFirstWeightedShuffling_Enabled tests +// buildLeafClusterConfig with an EDS cluster configuration when pick_first +// weighted shuffling is enabled. +func (s) TestBuildLeafClusterConfig_EDS_PickFirstWeightedShuffling_Enabled(t *testing.T) { + testutils.SetEnvConfig(t, &envconfig.PickFirstWeightedShuffling, true) + + testLRSServerConfig, err := bootstrap.ServerConfigForTesting(bootstrap.ServerConfigTestingOptions{ + URI: "trafficdirector.googleapis.com:443", + ChannelCreds: []bootstrap.ChannelCreds{{Type: "google_default"}}, + }) + if err != nil { + t.Fatalf("Failed to create LRS server config for testing: %v", err) + } + + // Create test localities with 2 endpoints each. + // Localities are passed in shuffled order to verify sorting by priority. + loc0 := makeLocality(0, 20, 0, 2) + loc1 := makeLocality(1, 80, 0, 2) + loc2 := makeLocality(2, 20, 1, 2) + loc3 := makeLocality(3, 80, 1, 2) + + gotODConfig, gotEndpoints, err := buildLeafClusterConfig( + &priorityConfig{ + clusterConfig: &xdsresource.ClusterConfig{ + Cluster: &xdsresource.ClusterUpdate{ + ClusterName: testClusterName, + ClusterType: xdsresource.ClusterTypeEDS, + EDSServiceName: testEDSServiceName, + LRSServerConfig: testLRSServerConfig, + MaxRequests: newUint32(testMaxRequests), + }, + EndpointConfig: &xdsresource.EndpointConfig{ + EDSUpdate: &xdsresource.EndpointsUpdate{ + Drops: []xdsresource.OverloadDropConfig{{ + Category: testDropCategory, + Numerator: testDropOverMillion, + Denominator: million, + }}, + Localities: []xdsresource.Locality{ + makeLocality(3, 80, 1, 2), + makeLocality(1, 80, 0, 2), + makeLocality(2, 20, 1, 2), + makeLocality(0, 20, 0, 2), + }, + }, + }, + }, + outlierDetection: noopODCfg, + childNameGen: newNameGenerator(2), + }, + nil, + ) + if err != nil { + t.Fatalf("buildLeafClusterConfig() failed: %v", err) + } + + wantODConfig := &outlierdetection.LBConfig{ + Interval: iserviceconfig.Duration(10 * time.Second), + BaseEjectionTime: iserviceconfig.Duration(30 * time.Second), + MaxEjectionTime: iserviceconfig.Duration(300 * time.Second), + MaxEjectionPercent: 10, + ChildPolicy: &iserviceconfig.BalancerConfig{ + Name: clusterimpl.Name, + Config: &clusterimpl.LBConfig{ + Cluster: testClusterName, + ChildPolicy: &iserviceconfig.BalancerConfig{ + Name: priority.Name, + Config: &priority.LBConfig{ + Children: map[string]*priority.Child{ + "priority-2-0": { + IgnoreReresolutionRequests: true, + }, + "priority-2-1": { + IgnoreReresolutionRequests: true, + }, + }, + Priorities: []string{"priority-2-0", "priority-2-1"}, + }, + }, + }, + }, + } + // Endpoints weights are the product of normalized locality weight and + // endpoint weight, represented as a fixed-point number in uQ1.31 format. + // Locality weights are normalized as: + // P1: locality 3: 80 / (100) = 0.8 + // P0: locality 1: 80 / (100) = 0.8 + // P1: locality 2: 20 / (100) = 0.2 + // P0: locality 0: 20 / (100) = 0.2 + // In fixed-point uQ1.31 format, the weights are: + // locality 3: 0.8 * 2^31 = 1717986918 + // locality 1: 0.8 * 2^31 = 1717986918 + // locality 2: 0.2 * 2^31 = 429496729 + // locality 0: 0.2 * 2^31 = 429496729 + // + // There are two endpoints in each locality, each with weight 1. So, their + // normalized weights are 0.5 each. And the final endpoint weights are a + // product of their locality weights and 0.5, which turns out to be either + // 1717986918 * 0.5 = 858993459, or, + // 429496729 * 0.5 = 214748364 + wantEndpoints := []resolver.Endpoint{ + testEndpointWithAttrs(loc0.Endpoints[0].ResolverEndpoint, 20, 214748364, "priority-2-0", &loc0.ID), + testEndpointWithAttrs(loc0.Endpoints[1].ResolverEndpoint, 20, 214748364, "priority-2-0", &loc0.ID), + testEndpointWithAttrs(loc1.Endpoints[0].ResolverEndpoint, 80, 858993459, "priority-2-0", &loc1.ID), + testEndpointWithAttrs(loc1.Endpoints[1].ResolverEndpoint, 80, 858993459, "priority-2-0", &loc1.ID), + testEndpointWithAttrs(loc2.Endpoints[0].ResolverEndpoint, 20, 214748364, "priority-2-1", &loc2.ID), + testEndpointWithAttrs(loc2.Endpoints[1].ResolverEndpoint, 20, 214748364, "priority-2-1", &loc2.ID), + testEndpointWithAttrs(loc3.Endpoints[0].ResolverEndpoint, 80, 858993459, "priority-2-1", &loc3.ID), + testEndpointWithAttrs(loc3.Endpoints[1].ResolverEndpoint, 80, 858993459, "priority-2-1", &loc3.ID), + } + + if diff := cmp.Diff(wantODConfig, gotODConfig); diff != "" { + t.Errorf("buildLeafClusterConfig() config diff (-want +got) %v", diff) + } + if diff := cmp.Diff(wantEndpoints, gotEndpoints, endpointCmpOpts); diff != "" { + t.Errorf("buildLeafClusterConfig() endpoints diff (-want +got) %v", diff) + } +} + +// TestBuildClusterImplConfigForDNS tests buildClusterImplConfigForDNS +// to verify the generated clusterimpl config and endpoint attributes +// for LOGICAL_DNS clusters. func (s) TestBuildClusterImplConfigForDNS(t *testing.T) { for _, tt := range []struct { name string @@ -390,6 +767,8 @@ func (s) TestBuildClusterImplConfigForDNS(t *testing.T) { } } +// TestBuildClusterImplConfigForEDS_PickFirstWeightedShuffling_Disabled tests +// buildClusterImplConfigForEDS when pick_first weighted shuffling is disabled. func (s) TestBuildClusterImplConfigForEDS_PickFirstWeightedShuffling_Disabled(t *testing.T) { testutils.SetEnvConfig(t, &envconfig.PickFirstWeightedShuffling, false) @@ -465,6 +844,8 @@ func (s) TestBuildClusterImplConfigForEDS_PickFirstWeightedShuffling_Disabled(t } } +// TestBuildClusterImplConfigForEDS_PickFirstWeightedShuffling_Enabled tests +// buildClusterImplConfigForEDS when pick_first weighted shuffling is enabled. func (s) TestBuildClusterImplConfigForEDS_PickFirstWeightedShuffling_Enabled(t *testing.T) { testutils.SetEnvConfig(t, &envconfig.PickFirstWeightedShuffling, true) @@ -557,6 +938,8 @@ func (s) TestBuildClusterImplConfigForEDS_PickFirstWeightedShuffling_Enabled(t * } } +// TestGroupLocalitiesByPriority tests groupLocalitiesByPriority to ensure +// localities are properly grouped and ordered by priority. func (s) TestGroupLocalitiesByPriority(t *testing.T) { // Create localities for two priorities (p0 and p1). p0Loc0 := makeLocality(0, 20, 0, 2) @@ -625,6 +1008,9 @@ func (s) TestGroupLocalitiesByPriority(t *testing.T) { } } +// TestPriorityLocalitiesToClusterImpl_PickFirstWeightedShuffling_Disabled +// tests priorityLocalitiesToClusterImpl when pick_first weighted shuffling is +// disabled. func (s) TestPriorityLocalitiesToClusterImpl_PickFirstWeightedShuffling_Disabled(t *testing.T) { testutils.SetEnvConfig(t, &envconfig.PickFirstWeightedShuffling, false) tests := []struct { @@ -763,6 +1149,9 @@ func (s) TestPriorityLocalitiesToClusterImpl_PickFirstWeightedShuffling_Disabled } } +// TestPriorityLocalitiesToClusterImpl_PickFirstWeightedShuffling_Enabled tests +// priorityLocalitiesToClusterImpl when pick_first weighted shuffling is +// enabled. func (s) TestPriorityLocalitiesToClusterImpl_PickFirstWeightedShuffling_Enabled(t *testing.T) { testutils.SetEnvConfig(t, &envconfig.PickFirstWeightedShuffling, true) tests := []struct { @@ -954,6 +1343,9 @@ func testEndpointWithAttrs(endpoint resolver.Endpoint, localityWeight, endpointW return endpoint } +// TestConvertClusterImplMapToOutlierDetection tests +// convertClusterImplMapToOutlierDetection to ensure clusterimpl configs are +// properly wrapped in outlier detection configs. func (s) TestConvertClusterImplMapToOutlierDetection(t *testing.T) { tests := []struct { name string diff --git a/internal/xds/balancer/cdsbalancer/e2e_test/balancer_test.go b/internal/xds/balancer/cdsbalancer/e2e_test/balancer_test.go index 6b7df2f26d07..64bab9171a47 100644 --- a/internal/xds/balancer/cdsbalancer/e2e_test/balancer_test.go +++ b/internal/xds/balancer/cdsbalancer/e2e_test/balancer_test.go @@ -300,21 +300,21 @@ func (s) TestErrorFromParentLB_ResourceNotFound(t *testing.T) { } // Test verifies that when the received Cluster resource contains outlier -// detection configuration, the LB config pushed to the priority policy contains -// the appropriate configuration for the outlier detection LB policy. +// detection configuration, the LB config pushed to the outlier detection policy +// contains the appropriate configuration for the outlier detection LB policy. func (s) TestOutlierDetectionConfigPropagationToChildPolicy(t *testing.T) { - // Unregister the priority balancer builder for the duration of this test, + // Unregister the outlier detection balancer builder for the duration of this test, // and register a policy under the same name that makes the LB config // pushed to it available to the test. - priorityBuilder := balancer.Get(priority.Name) - internal.BalancerUnregister(priorityBuilder.Name()) + odBuilder := balancer.Get(outlierdetection.Name) + internal.BalancerUnregister(odBuilder.Name()) lbCfgCh := make(chan serviceconfig.LoadBalancingConfig, 1) - stub.Register(priority.Name, stub.BalancerFuncs{ + stub.Register(outlierdetection.Name, stub.BalancerFuncs{ Init: func(bd *stub.BalancerData) { - bd.ChildBalancer = priorityBuilder.Build(bd.ClientConn, bd.BuildOptions) + bd.ChildBalancer = odBuilder.Build(bd.ClientConn, bd.BuildOptions) }, ParseConfig: func(lbCfg json.RawMessage) (serviceconfig.LoadBalancingConfig, error) { - return priorityBuilder.(balancer.ConfigParser).ParseConfig(lbCfg) + return odBuilder.(balancer.ConfigParser).ParseConfig(lbCfg) }, UpdateClientConnState: func(bd *stub.BalancerData, ccs balancer.ClientConnState) error { select { @@ -327,7 +327,7 @@ func (s) TestOutlierDetectionConfigPropagationToChildPolicy(t *testing.T) { bd.ChildBalancer.Close() }, }) - defer balancer.Register(priorityBuilder) + defer balancer.Register(odBuilder) managementServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{}) @@ -368,29 +368,27 @@ func (s) TestOutlierDetectionConfigPropagationToChildPolicy(t *testing.T) { _, cleanup := setupAndDial(t, bootstrapContents) defer cleanup() - // The priority configuration generated should have Outlier Detection as a - // direct child due to Outlier Detection being turned on. - wantCfg := &priority.LBConfig{ - Children: map[string]*priority.Child{ - "priority-0-0": { - Config: &iserviceconfig.BalancerConfig{ - Name: outlierdetection.Name, - Config: &outlierdetection.LBConfig{ - Interval: iserviceconfig.Duration(10 * time.Second), // default interval - BaseEjectionTime: iserviceconfig.Duration(30 * time.Second), - MaxEjectionTime: iserviceconfig.Duration(300 * time.Second), - MaxEjectionPercent: 10, - SuccessRateEjection: &outlierdetection.SuccessRateEjection{ - StdevFactor: 2000, - EnforcementPercentage: 50, - MinimumHosts: 10, - RequestVolume: 50, - }, - ChildPolicy: &iserviceconfig.BalancerConfig{ - Name: clusterimpl.Name, - Config: &clusterimpl.LBConfig{ - Cluster: clusterName, - ChildPolicy: &iserviceconfig.BalancerConfig{ + wantCfg := &outlierdetection.LBConfig{ + Interval: iserviceconfig.Duration(10 * time.Second), // default interval + BaseEjectionTime: iserviceconfig.Duration(30 * time.Second), + MaxEjectionTime: iserviceconfig.Duration(300 * time.Second), + MaxEjectionPercent: 10, + SuccessRateEjection: &outlierdetection.SuccessRateEjection{ + StdevFactor: 2000, + EnforcementPercentage: 50, + MinimumHosts: 10, + RequestVolume: 50, + }, + ChildPolicy: &iserviceconfig.BalancerConfig{ + Name: clusterimpl.Name, + Config: &clusterimpl.LBConfig{ + Cluster: clusterName, + ChildPolicy: &iserviceconfig.BalancerConfig{ + Name: priority.Name, + Config: &priority.LBConfig{ + Children: map[string]*priority.Child{ + "priority-0-0": { + Config: &iserviceconfig.BalancerConfig{ Name: wrrlocality.Name, Config: &wrrlocality.LBConfig{ ChildPolicy: &iserviceconfig.BalancerConfig{ @@ -398,19 +396,19 @@ func (s) TestOutlierDetectionConfigPropagationToChildPolicy(t *testing.T) { }, }, }, + IgnoreReresolutionRequests: true, }, }, + Priorities: []string{"priority-0-0"}, }, }, - IgnoreReresolutionRequests: true, }, }, - Priorities: []string{"priority-0-0"}, } select { case lbCfg := <-lbCfgCh: - gotCfg := lbCfg.(*priority.LBConfig) + gotCfg := lbCfg.(*outlierdetection.LBConfig) if diff := cmp.Diff(wantCfg, gotCfg); diff != "" { t.Fatalf("Child policy received unexpected diff in config (-want +got):\n%s", diff) }