cdsbalancer: test cleanup part 3/N (#6564)

internal/testutils/xds/e2e/clientresources.go

@@ -33,6 +33,7 @@ import (
 	v3endpointpb "github.com/envoyproxy/go-control-plane/envoy/config/endpoint/v3"
 	v3listenerpb "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3"
 	v3routepb "github.com/envoyproxy/go-control-plane/envoy/config/route/v3"
+	v3aggregateclusterpb "github.com/envoyproxy/go-control-plane/envoy/extensions/clusters/aggregate/v3"
 	v3routerpb "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/router/v3"
 	v3httppb "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/network/http_connection_manager/v3"
 	v3tlspb "github.com/envoyproxy/go-control-plane/envoy/extensions/transport_sockets/tls/v3"
@@ -450,12 +451,37 @@ const (
 	LoadBalancingPolicyRingHash
 )
 
+// ClusterType specifies the type of the Cluster resource.
+type ClusterType int
+
+const (
+	// ClusterTypeEDS specifies a Cluster that uses EDS to resolve endpoints.
+	ClusterTypeEDS ClusterType = iota
+	// ClusterTypeLogicalDNS specifies a Cluster that uses DNS to resolve
+	// endpoints.
+	ClusterTypeLogicalDNS
+	// ClusterTypeAggregate specifies a Cluster that is made up of child
+	// clusters.
+	ClusterTypeAggregate
+)
+
 // ClusterOptions contains options to configure a Cluster resource.
 type ClusterOptions struct {
+	Type ClusterType
 	// ClusterName is the name of the Cluster resource.
 	ClusterName string
-	// ServiceName is the EDS service name of the Cluster.
+	// ServiceName is the EDS service name of the Cluster. Applicable only when
+	// cluster type is EDS.
 	ServiceName string
+	// ChildNames is the list of child Cluster names. Applicable only when
+	// cluster type is Aggregate.
+	ChildNames []string
+	// DNSHostName is the dns host name of the Cluster. Applicable only when the
+	// cluster type is DNS.
+	DNSHostName string
+	// DNSPort is the port number of the Cluster. Applicable only when the
+	// cluster type is DNS.
+	DNSPort uint32
 	// Policy is the LB policy to be used.
 	Policy LoadBalancingPolicy
 	// SecurityLevel determines the security configuration for the Cluster.
@@ -504,17 +530,51 @@ func ClusterResourceWithOptions(opts ClusterOptions) *v3clusterpb.Cluster {
 		lbPolicy = v3clusterpb.Cluster_RING_HASH
 	}
 	cluster := &v3clusterpb.Cluster{
-		Name:                 opts.ClusterName,
-		ClusterDiscoveryType: &v3clusterpb.Cluster_Type{Type: v3clusterpb.Cluster_EDS},
-		EdsClusterConfig: &v3clusterpb.Cluster_EdsClusterConfig{
+		Name:     opts.ClusterName,
+		LbPolicy: lbPolicy,
+	}
+	switch opts.Type {
+	case ClusterTypeEDS:
+		cluster.ClusterDiscoveryType = &v3clusterpb.Cluster_Type{Type: v3clusterpb.Cluster_EDS}
+		cluster.EdsClusterConfig = &v3clusterpb.Cluster_EdsClusterConfig{
 			EdsConfig: &v3corepb.ConfigSource{
 				ConfigSourceSpecifier: &v3corepb.ConfigSource_Ads{
 					Ads: &v3corepb.AggregatedConfigSource{},
 				},
 			},
 			ServiceName: opts.ServiceName,
-		},
-		LbPolicy: lbPolicy,
+		}
+	case ClusterTypeLogicalDNS:
+		cluster.ClusterDiscoveryType = &v3clusterpb.Cluster_Type{Type: v3clusterpb.Cluster_LOGICAL_DNS}
+		cluster.LoadAssignment = &v3endpointpb.ClusterLoadAssignment{
+			Endpoints: []*v3endpointpb.LocalityLbEndpoints{{
+				LbEndpoints: []*v3endpointpb.LbEndpoint{{
+					HostIdentifier: &v3endpointpb.LbEndpoint_Endpoint{
+						Endpoint: &v3endpointpb.Endpoint{
+							Address: &v3corepb.Address{
+								Address: &v3corepb.Address_SocketAddress{
+									SocketAddress: &v3corepb.SocketAddress{
+										Address: opts.DNSHostName,
+										PortSpecifier: &v3corepb.SocketAddress_PortValue{
+											PortValue: opts.DNSPort,
+										},
+									},
+								},
+							},
+						},
+					},
+				}},
+			}},
+		}
+	case ClusterTypeAggregate:
+		cluster.ClusterDiscoveryType = &v3clusterpb.Cluster_ClusterType{
+			ClusterType: &v3clusterpb.Cluster_CustomClusterType{
+				Name: "envoy.clusters.aggregate",
+				TypedConfig: testutils.MarshalAny(&v3aggregateclusterpb.ClusterConfig{
+					Clusters: opts.ChildNames,
+				}),
+			},
+		}
 	}
 	if tlsContext != nil {
 		cluster.TransportSocket = &v3corepb.TransportSocket{

xds/internal/balancer/cdsbalancer/cdsbalancer_test.go

@@ -61,7 +61,11 @@ import (
 
 const (
 	clusterName             = "cluster1"
+	edsClusterName          = clusterName + "-eds"
+	dnsClusterName          = clusterName + "-dns"
 	serviceName             = "service1"
+	dnsHostName             = "dns_host"
+	dnsPort                 = uint32(8080)
 	defaultTestTimeout      = 5 * time.Second
 	defaultTestShortTimeout = 10 * time.Millisecond // For events expected to *not* happen.
 )
@@ -218,6 +222,9 @@ func setupWithManagementServer(t *testing.T) (*e2e.ManagementServer, string, *gr
 			}
 			return nil
 		},
+		// Required for aggregate clusters as all resources cannot be requested
+		// at once.
+		AllowResourceSubset: true,
 	})
 	t.Cleanup(cleanup)
 

xds/internal/balancer/clusterresolver/e2e_test/aggregate_cluster_test.go

@@ -33,7 +33,6 @@ import (
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/internal"
 	"google.golang.org/grpc/internal/stubserver"
-	"google.golang.org/grpc/internal/testutils"
 	"google.golang.org/grpc/internal/testutils/pickfirst"
 	"google.golang.org/grpc/internal/testutils/xds/e2e"
 	"google.golang.org/grpc/peer"
@@ -50,7 +49,6 @@ import (
 	v3clusterpb "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3"
 	v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	v3endpointpb "github.com/envoyproxy/go-control-plane/envoy/config/endpoint/v3"
-	v3aggregateclusterpb "github.com/envoyproxy/go-control-plane/envoy/extensions/clusters/aggregate/v3"
 	v3discoverypb "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
 	testgrpc "google.golang.org/grpc/interop/grpc_testing"
 	testpb "google.golang.org/grpc/interop/grpc_testing"
@@ -59,48 +57,22 @@ import (
 // makeAggregateClusterResource returns an aggregate cluster resource with the
 // given name and list of child names.
 func makeAggregateClusterResource(name string, childNames []string) *v3clusterpb.Cluster {
-	return &v3clusterpb.Cluster{
-		Name: name,
-		ClusterDiscoveryType: &v3clusterpb.Cluster_ClusterType{
-			ClusterType: &v3clusterpb.Cluster_CustomClusterType{
-				Name: "envoy.clusters.aggregate",
-				TypedConfig: testutils.MarshalAny(&v3aggregateclusterpb.ClusterConfig{
-					Clusters: childNames,
-				}),
-			},
-		},
-		LbPolicy: v3clusterpb.Cluster_ROUND_ROBIN,
-	}
+	return e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
+		ClusterName: name,
+		Type:        e2e.ClusterTypeAggregate,
+		ChildNames:  childNames,
+	})
 }
 
 // makeLogicalDNSClusterResource returns a LOGICAL_DNS cluster resource with the
 // given name and given DNS host and port.
 func makeLogicalDNSClusterResource(name, dnsHost string, dnsPort uint32) *v3clusterpb.Cluster {
-	return &v3clusterpb.Cluster{
-		Name:                 name,
-		ClusterDiscoveryType: &v3clusterpb.Cluster_Type{Type: v3clusterpb.Cluster_LOGICAL_DNS},
-		LbPolicy:             v3clusterpb.Cluster_ROUND_ROBIN,
-		LoadAssignment: &v3endpointpb.ClusterLoadAssignment{
-			Endpoints: []*v3endpointpb.LocalityLbEndpoints{{
-				LbEndpoints: []*v3endpointpb.LbEndpoint{{
-					HostIdentifier: &v3endpointpb.LbEndpoint_Endpoint{
-						Endpoint: &v3endpointpb.Endpoint{
-							Address: &v3corepb.Address{
-								Address: &v3corepb.Address_SocketAddress{
-									SocketAddress: &v3corepb.SocketAddress{
-										Address: dnsHost,
-										PortSpecifier: &v3corepb.SocketAddress_PortValue{
-											PortValue: dnsPort,
-										},
-									},
-								},
-							},
-						},
-					},
-				}},
-			}},
-		},
-	}
+	return e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
+		ClusterName: name,
+		Type:        e2e.ClusterTypeLogicalDNS,
+		DNSHostName: dnsHost,
+		DNSPort:     dnsPort,
+	})
 }
 
 // setupDNS unregisters the DNS resolver and registers a manual resolver for the

xds/internal/balancer/clusterresolver/resource_resolver.go

@@ -200,7 +200,7 @@ func (rr *resourceResolver) updateMechanisms(mechanisms []DiscoveryMechanism) {
 	for dm, r := range rr.childrenMap {
 		if !newDMs[dm] {
 			delete(rr.childrenMap, dm)
-			r.r.stop()
+			go r.r.stop()
 		}
 	}
 	// Regenerate even if there's no change in discovery mechanism, in case