xds: Outlier Detection configuration in Cluster Resolver Balancer

xds/internal/balancer/cdsbalancer/cdsbalancer.go

@@ -272,12 +272,12 @@ func buildProviderFunc(configs map[string]*certprovider.BuildableConfig, instanc
 	return provider, nil
 }
 
-func outlierDetectionToConfig(od *xdsresource.OutlierDetection) *outlierdetection.LBConfig { // Already validated - no need to return error
+func outlierDetectionToConfig(od *xdsresource.OutlierDetection) *outlierdetection.ODConfig { // Already validated - no need to return error
 	if od == nil {
 		// "If the outlier_detection field is not set in the Cluster message, a
 		// "no-op" outlier_detection config will be generated, with interval set
 		// to the maximum possible value and all other fields unset." - A50
-		return &outlierdetection.LBConfig{
+		return &outlierdetection.ODConfig{
 			Interval: 1<<63 - 1,
 		}
 	}
@@ -308,7 +308,7 @@ func outlierDetectionToConfig(od *xdsresource.OutlierDetection) *outlierdetectio
 		}
 	}
 
-	return &outlierdetection.LBConfig{
+	return &outlierdetection.ODConfig{
 		Interval:                  od.Interval,
 		BaseEjectionTime:          od.BaseEjectionTime,
 		MaxEjectionTime:           od.MaxEjectionTime,

xds/internal/balancer/cdsbalancer/cdsbalancer_test.go

@@ -56,7 +56,7 @@ var (
 		ServerURI: "self_server",
 		CredsType: "self_creds",
 	}
-	noopODLBCfg = &outlierdetection.LBConfig{
+	noopODLBCfg = &outlierdetection.ODConfig{
 		Interval: 1<<63 - 1,
 	}
 )
@@ -215,7 +215,7 @@ func cdsCCS(cluster string, xdsC xdsclient.XDSClient) balancer.ClientConnState {
 
 // edsCCS is a helper function to construct a good update passed from the
 // cdsBalancer to the edsBalancer.
-func edsCCS(service string, countMax *uint32, enableLRS bool, xdslbpolicy *internalserviceconfig.BalancerConfig, odConfig *outlierdetection.LBConfig) balancer.ClientConnState {
+func edsCCS(service string, countMax *uint32, enableLRS bool, xdslbpolicy *internalserviceconfig.BalancerConfig, odConfig *outlierdetection.ODConfig) balancer.ClientConnState {
 	discoveryMechanism := clusterresolver.DiscoveryMechanism{
 		Type:                  clusterresolver.DiscoveryMechanismTypeEDS,
 		Cluster:               service,
@@ -421,7 +421,7 @@ func (s) TestHandleClusterUpdate(t *testing.T) {
 				FailurePercentageMinimumHosts:  5,
 				FailurePercentageRequestVolume: 50,
 			}},
-			wantCCS: edsCCS(serviceName, nil, false, nil, &outlierdetection.LBConfig{
+			wantCCS: edsCCS(serviceName, nil, false, nil, &outlierdetection.ODConfig{
 				Interval:           10 * time.Second,
 				BaseEjectionTime:   30 * time.Second,
 				MaxEjectionTime:    300 * time.Second,
@@ -846,7 +846,7 @@ func (s) TestOutlierDetectionToConfig(t *testing.T) {
 	tests := []struct {
 		name        string
 		od          *xdsresource.OutlierDetection
-		odLBCfgWant *outlierdetection.LBConfig
+		odLBCfgWant *outlierdetection.ODConfig
 	}{
 		// "if the outlier_detection field is not set in the Cluster resource,
 		// a "no-op" outlier_detection config will be generated in the
@@ -876,7 +876,7 @@ func (s) TestOutlierDetectionToConfig(t *testing.T) {
 				FailurePercentageMinimumHosts:  5,
 				FailurePercentageRequestVolume: 50,
 			},
-			odLBCfgWant: &outlierdetection.LBConfig{
+			odLBCfgWant: &outlierdetection.ODConfig{
 				Interval:            10 * time.Second,
 				BaseEjectionTime:    30 * time.Second,
 				MaxEjectionTime:     300 * time.Second,
@@ -909,7 +909,7 @@ func (s) TestOutlierDetectionToConfig(t *testing.T) {
 				FailurePercentageMinimumHosts:  5,
 				FailurePercentageRequestVolume: 50,
 			},
-			odLBCfgWant: &outlierdetection.LBConfig{
+			odLBCfgWant: &outlierdetection.ODConfig{
 				Interval:           10 * time.Second,
 				BaseEjectionTime:   30 * time.Second,
 				MaxEjectionTime:    300 * time.Second,
@@ -939,7 +939,7 @@ func (s) TestOutlierDetectionToConfig(t *testing.T) {
 				FailurePercentageMinimumHosts:  5,
 				FailurePercentageRequestVolume: 50,
 			},
-			odLBCfgWant: &outlierdetection.LBConfig{
+			odLBCfgWant: &outlierdetection.ODConfig{
 				Interval:           10 * time.Second,
 				BaseEjectionTime:   30 * time.Second,
 				MaxEjectionTime:    300 * time.Second,

xds/internal/balancer/clusterresolver/clusterresolver_test.go

@@ -25,12 +25,20 @@ import (
 	"time"
 
 	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
 	"google.golang.org/grpc/balancer"
+	"google.golang.org/grpc/balancer/roundrobin"
+	"google.golang.org/grpc/balancer/weightedtarget"
 	"google.golang.org/grpc/connectivity"
+	"google.golang.org/grpc/internal/envconfig"
 	"google.golang.org/grpc/internal/grpctest"
+	internalserviceconfig "google.golang.org/grpc/internal/serviceconfig"
 	"google.golang.org/grpc/internal/testutils"
 	"google.golang.org/grpc/resolver"
 	"google.golang.org/grpc/xds/internal"
+	"google.golang.org/grpc/xds/internal/balancer/clusterimpl"
+	"google.golang.org/grpc/xds/internal/balancer/outlierdetection"
+	"google.golang.org/grpc/xds/internal/balancer/priority"
 	"google.golang.org/grpc/xds/internal/testutils/fakeclient"
 	"google.golang.org/grpc/xds/internal/xdsclient"
 	"google.golang.org/grpc/xds/internal/xdsclient/xdsresource"
@@ -130,6 +138,18 @@ func (f *fakeChildBalancer) Close() {}
 
 func (f *fakeChildBalancer) ExitIdle() {}
 
+func (f *fakeChildBalancer) waitForClientConnStateChangeVerifyBalancerConfig(ctx context.Context, wantCCS balancer.ClientConnState) error {
+	ccs, err := f.clientConnState.Receive(ctx)
+	if err != nil {
+		return err
+	}
+	gotCCS := ccs.(balancer.ClientConnState)
+	if diff := cmp.Diff(gotCCS, wantCCS, cmpopts.IgnoreFields(resolver.State{}, "Addresses", "ServiceConfig", "Attributes")); diff != "" {
+		return fmt.Errorf("received unexpected ClientConnState, diff (-got +want): %v", diff)
+	}
+	return nil
+}
+
 func (f *fakeChildBalancer) waitForClientConnStateChange(ctx context.Context) error {
 	_, err := f.clientConnState.Receive(ctx)
 	if err != nil {
@@ -500,3 +520,100 @@ func newLBConfigWithOneEDS(edsServiceName string) *LBConfig {
 		}},
 	}
 }
+
+func newLBConfigWithOneEDSAndOutlierDetection(edsServiceName string, odCfg *outlierdetection.ODConfig) *LBConfig {
+	lbCfg := newLBConfigWithOneEDS(edsServiceName)
+	lbCfg.DiscoveryMechanisms[0].OutlierDetection = odCfg
+	return lbCfg
+}
+
+// TestOutlierDetection tests the Balancer Config sent down to the child
+// priority balancer when Outlier Detection is turned on. The Priority
+// Configuration sent downward should have a top level Outlier Detection Policy
+// for each priority.
+func (s) TestOutlierDetection(t *testing.T) {
+	oldOutlierDetection := envconfig.XDSOutlierDetection
+	envconfig.XDSOutlierDetection = true
+	defer func() {
+		envconfig.XDSOutlierDetection = oldOutlierDetection
+	}()
+
+	edsLBCh := testutils.NewChannel()
+	xdsC, cleanup := setup(edsLBCh)
+	defer cleanup()
+	builder := balancer.Get(Name)
+	edsB := builder.Build(newNoopTestClientConn(), balancer.BuildOptions{})
+	if edsB == nil {
+		t.Fatalf("builder.Build(%s) failed and returned nil", Name)
+	}
+	defer edsB.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	// Update Cluster Resolver with Client Conn State with Outlier Detection
+	// configuration present. This is what will be passed down to this balancer,
+	// as CDS Balancer gets the Cluster Update and converts the Outlier
+	// Detection data to an Outlier Detection configuration and sends it to this
+	// level.
+	if err := edsB.UpdateClientConnState(balancer.ClientConnState{
+		ResolverState:  xdsclient.SetClient(resolver.State{}, xdsC),
+		BalancerConfig: newLBConfigWithOneEDSAndOutlierDetection(testEDSServcie, noopODCfg),
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := xdsC.WaitForWatchEDS(ctx); err != nil {
+		t.Fatalf("xdsClient.WatchEndpoints failed with error: %v", err)
+	}
+
+	// Invoke EDS Callback - causes child balancer to be built and then
+	// UpdateClientConnState called on it with Outlier Detection as a direct
+	// child.
+	xdsC.InvokeWatchEDSCallback("", defaultEndpointsUpdate, nil)
+	edsLB, err := waitForNewChildLB(ctx, edsLBCh)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	localityID := internal.LocalityID{Zone: "zone"}
+	// The priority configuration generated should have Outlier Detection as a
+	// direct child due to Outlier Detection being turned on.
+	pCfgWant := &priority.LBConfig{
+		Children: map[string]*priority.Child{
+			"priority-0-1": {
+				Config: &internalserviceconfig.BalancerConfig{
+					Name: outlierdetection.Name,
+					Config: &outlierdetection.LBConfig{
+						ODConfig: noopODCfg,
+						ChildPolicy: &internalserviceconfig.BalancerConfig{
+							Name: clusterimpl.Name,
+							Config: &clusterimpl.LBConfig{
+								Cluster:        testClusterName,
+								EDSServiceName: "test-eds-service-name",
+								ChildPolicy: &internalserviceconfig.BalancerConfig{
+									Name: weightedtarget.Name,
+									Config: &weightedtarget.LBConfig{
+										Targets: map[string]weightedtarget.Target{
+											assertString(localityID.ToString): {
+												Weight:      100,
+												ChildPolicy: &internalserviceconfig.BalancerConfig{Name: roundrobin.Name},
+											},
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+				IgnoreReresolutionRequests: true,
+			},
+		},
+		Priorities: []string{"priority-0-1"},
+	}
+
+	if err := edsLB.waitForClientConnStateChangeVerifyBalancerConfig(ctx, balancer.ClientConnState{
+		BalancerConfig: pCfgWant,
+	}); err != nil {
+		t.Fatalf("EDS impl got unexpected update: %v", err)
+	}
+}

xds/internal/balancer/clusterresolver/config.go

@@ -103,9 +103,9 @@ type DiscoveryMechanism struct {
 	// DNSHostname is the DNS name to resolve in "host:port" form. For type
 	// LOGICAL_DNS only.
 	DNSHostname string `json:"dnsHostname,omitempty"`
-	// OutlierDetection is the Outlier Detection LB configuration for this
+	// OutlierDetection is the Outlier Detection configuration for this
 	// priority.
-	OutlierDetection *outlierdetection.LBConfig `json:"outlierDetection,omitempty"`
+	OutlierDetection *outlierdetection.ODConfig `json:"outlierDetection,omitempty"`
 }
 
 // Equal returns whether the DiscoveryMechanism is the same with the parameter.

xds/internal/balancer/clusterresolver/configbuilder.go

@@ -26,11 +26,13 @@ import (
 	"google.golang.org/grpc/balancer/roundrobin"
 	"google.golang.org/grpc/balancer/weightedroundrobin"
 	"google.golang.org/grpc/balancer/weightedtarget"
+	"google.golang.org/grpc/internal/envconfig"
 	"google.golang.org/grpc/internal/hierarchy"
 	internalserviceconfig "google.golang.org/grpc/internal/serviceconfig"
 	"google.golang.org/grpc/resolver"
 	"google.golang.org/grpc/xds/internal"
 	"google.golang.org/grpc/xds/internal/balancer/clusterimpl"
+	"google.golang.org/grpc/xds/internal/balancer/outlierdetection"
 	"google.golang.org/grpc/xds/internal/balancer/priority"
 	"google.golang.org/grpc/xds/internal/balancer/ringhash"
 	"google.golang.org/grpc/xds/internal/xdsclient/xdsresource"
@@ -125,30 +127,74 @@ func buildPriorityConfig(priorities []priorityConfig, xdsLBPolicy *internalservi
 			if err != nil {
 				return nil, nil, err
 			}
+			var odCfgs map[string]*outlierdetection.LBConfig
+			if envconfig.XDSOutlierDetection {
+				odCfgs = convertClusterImplMapToOutlierDetection(configs, p.mechanism.OutlierDetection)
+			}
 			retConfig.Priorities = append(retConfig.Priorities, names...)
+			retAddrs = append(retAddrs, addrs...)
+
+			if envconfig.XDSOutlierDetection {
+				for n, c := range odCfgs {
+					retConfig.Children[n] = &priority.Child{
+						Config: &internalserviceconfig.BalancerConfig{Name: outlierdetection.Name, Config: c},
+						// Ignore all re-resolution from EDS children.
+						IgnoreReresolutionRequests: true,
+					}
+				}
+				continue
+			}
 			for n, c := range configs {
 				retConfig.Children[n] = &priority.Child{
 					Config: &internalserviceconfig.BalancerConfig{Name: clusterimpl.Name, Config: c},
 					// Ignore all re-resolution from EDS children.
 					IgnoreReresolutionRequests: true,
 				}
+
 			}
-			retAddrs = append(retAddrs, addrs...)
 		case DiscoveryMechanismTypeLogicalDNS:
 			name, config, addrs := buildClusterImplConfigForDNS(i, p.addresses, p.mechanism)
+			var odCfg *outlierdetection.LBConfig
+			if envconfig.XDSOutlierDetection {
+				odCfg = makeClusterImplOutlierDetectionChild(*config, *p.mechanism.OutlierDetection)
+			}
 			retConfig.Priorities = append(retConfig.Priorities, name)
+			retAddrs = append(retAddrs, addrs...)
+			if envconfig.XDSOutlierDetection {
+				retConfig.Children[name] = &priority.Child{
+					Config: &internalserviceconfig.BalancerConfig{Name: outlierdetection.Name, Config: odCfg},
+					// Not ignore re-resolution from DNS children, they will trigger
+					// DNS to re-resolve.
+					IgnoreReresolutionRequests: false,
+				}
+				continue
+			}
 			retConfig.Children[name] = &priority.Child{
 				Config: &internalserviceconfig.BalancerConfig{Name: clusterimpl.Name, Config: config},
 				// Not ignore re-resolution from DNS children, they will trigger
 				// DNS to re-resolve.
 				IgnoreReresolutionRequests: false,
 			}
-			retAddrs = append(retAddrs, addrs...)
 		}
 	}
 	return retConfig, retAddrs, nil
 }
 
+func convertClusterImplMapToOutlierDetection(ciCfgs map[string]*clusterimpl.LBConfig, odCfg *outlierdetection.ODConfig) map[string]*outlierdetection.LBConfig {
+	odCfgs := make(map[string]*outlierdetection.LBConfig)
+	for n, c := range ciCfgs {
+		odCfgs[n] = makeClusterImplOutlierDetectionChild(*c, *odCfg)
+	}
+	return odCfgs
+}
+
+func makeClusterImplOutlierDetectionChild(ciCfg clusterimpl.LBConfig, odCfg outlierdetection.ODConfig) *outlierdetection.LBConfig {
+	return &outlierdetection.LBConfig{
+		ODConfig:    &odCfg,
+		ChildPolicy: &internalserviceconfig.BalancerConfig{Name: clusterimpl.Name, Config: &ciCfg},
+	}
+}
+
 func buildClusterImplConfigForDNS(parentPriority int, addrStrs []string, mechanism DiscoveryMechanism) (string, *clusterimpl.LBConfig, []resolver.Address) {
 	// Endpoint picking policy for DNS is hardcoded to pick_first.
 	const childPolicy = "pick_first"