xds/resolver: Add support for cluster specifier plugins

xds/internal/resolver/serviceconfig.go

@@ -38,14 +38,17 @@ import (
 	"google.golang.org/grpc/status"
 	"google.golang.org/grpc/xds/internal/balancer/clustermanager"
 	"google.golang.org/grpc/xds/internal/balancer/ringhash"
+	"google.golang.org/grpc/xds/internal/clusterspecifier"
 	"google.golang.org/grpc/xds/internal/httpfilter"
 	"google.golang.org/grpc/xds/internal/httpfilter/router"
 	"google.golang.org/grpc/xds/internal/xdsclient/xdsresource"
 )
 
 const (
-	cdsName               = "cds_experimental"
-	xdsClusterManagerName = "xds_cluster_manager_experimental"
+	cdsName                      = "cds_experimental"
+	xdsClusterManagerName        = "xds_cluster_manager_experimental"
+	clusterPrefix                = "cluster:"
+	clusterSpecifierPluginPrefix = "cluster_specifier_plugin:"
 )
 
 type serviceConfig struct {
@@ -86,9 +89,15 @@ func (r *xdsResolver) pruneActiveClusters() {
 func serviceConfigJSON(activeClusters map[string]*clusterInfo) ([]byte, error) {
 	// Generate children (all entries in activeClusters).
 	children := make(map[string]xdsChildConfig)
-	for cluster := range activeClusters {
-		children[cluster] = xdsChildConfig{
-			ChildPolicy: newBalancerConfig(cdsName, cdsBalancerConfig{Cluster: cluster}),
+	for cluster, ci := range activeClusters {
+		if ci.cspCfg != nil {
+			children[cluster] = xdsChildConfig{
+				ChildPolicy: balancerConfig(ci.cspCfg),
+			}
+		} else {
+			children[cluster] = xdsChildConfig{
+				ChildPolicy: newBalancerConfig(cdsName, cdsBalancerConfig{Cluster: strings.TrimPrefix(cluster, clusterPrefix)}),
+			}
 		}
 	}
 
@@ -158,10 +167,12 @@ func (cs *configSelector) SelectConfig(rpcInfo iresolver.RPCInfo) (*iresolver.RP
 	if rt == nil || rt.clusters == nil {
 		return nil, errNoMatchedRouteFound
 	}
+
 	cluster, ok := rt.clusters.Next().(*routeCluster)
 	if !ok {
 		return nil, status.Errorf(codes.Internal, "error retrieving cluster for match: %v (%T)", cluster, cluster)
 	}
+
 	// Add a ref to the selected cluster, as this RPC needs this cluster until
 	// it is committed.
 	ref := &cs.clusters[cluster.name].refCount
@@ -353,21 +364,22 @@ func (r *xdsResolver) newConfigSelector(su serviceUpdate) (*configSelector, erro
 
 	for i, rt := range su.virtualHost.Routes {
 		clusters := newWRR()
-		for cluster, wc := range rt.WeightedClusters {
+		if rt.ClusterSpecifierPlugin != "" {
+			clusterName := clusterSpecifierPluginPrefix + rt.ClusterSpecifierPlugin
 			clusters.Add(&routeCluster{
-				name:                     cluster,
-				httpFilterConfigOverride: wc.HTTPFilterConfigOverride,
-			}, int64(wc.Weight))
-
-			// Initialize entries in cs.clusters map, creating entries in
-			// r.activeClusters as necessary.  Set to zero as they will be
-			// incremented by incRefs.
-			ci := r.activeClusters[cluster]
-			if ci == nil {
-				ci = &clusterInfo{refCount: 0}
-				r.activeClusters[cluster] = ci
+				name: clusterName,
+			}, 1)
+			r.initializeCluster(clusterName, cs)
+			r.activeClusters[clusterName].cspCfg = su.clusterSpecifierPlugins[rt.ClusterSpecifierPlugin]
+		} else {
+			for cluster, wc := range rt.WeightedClusters {
+				clusterName := clusterPrefix + cluster
+				clusters.Add(&routeCluster{
+					name:                     clusterName,
+					httpFilterConfigOverride: wc.HTTPFilterConfigOverride,
+				}, int64(wc.Weight))
+				r.initializeCluster(clusterName, cs)
 			}
-			cs.clusters[cluster] = ci
 		}
 		cs.routes[i].clusters = clusters
 
@@ -397,9 +409,24 @@ func (r *xdsResolver) newConfigSelector(su serviceUpdate) (*configSelector, erro
 	return cs, nil
 }
 
+// initializeCluster initializes entries in cs.clusters map, creating entries in
+// r.activeClusters as necessary.  Any created entries will be set to zero as
+// they will be incremented by incRefs.
+func (r *xdsResolver) initializeCluster(clusterName string, cs *configSelector) {
+	ci := r.activeClusters[clusterName]
+	if ci == nil {
+		ci = &clusterInfo{refCount: 0}
+		r.activeClusters[clusterName] = ci
+	}
+	cs.clusters[clusterName] = ci
+}
+
 type clusterInfo struct {
 	// number of references to this cluster; accessed atomically
 	refCount int32
+	// cspCfg is the configuration for this cluster if the cluster is a cluster
+	// specifier plugin. This will be nil otherwise.
+	cspCfg clusterspecifier.BalancerConfig
 }
 
 type interceptorList struct {

xds/internal/resolver/watch_service.go

@@ -25,6 +25,7 @@ import (
 
 	"google.golang.org/grpc/internal/grpclog"
 	"google.golang.org/grpc/internal/pretty"
+	"google.golang.org/grpc/xds/internal/clusterspecifier"
 	"google.golang.org/grpc/xds/internal/xdsclient"
 	"google.golang.org/grpc/xds/internal/xdsclient/xdsresource"
 )
@@ -35,6 +36,9 @@ import (
 type serviceUpdate struct {
 	// virtualHost contains routes and other configuration to route RPCs.
 	virtualHost *xdsresource.VirtualHost
+	// clusterSpecifierPlugins contain the configurations for any cluster
+	// specifier plugins emitted by the xdsclient.
+	clusterSpecifierPlugins map[string]clusterspecifier.BalancerConfig
 	// ldsConfig contains configuration that applies to all routes.
 	ldsConfig ldsConfig
 }
@@ -120,7 +124,7 @@ func (w *serviceUpdateWatcher) handleLDSResp(update xdsresource.ListenerUpdate,
 		}
 
 		// Handle the inline RDS update as if it's from an RDS watch.
-		w.updateVirtualHostsFromRDS(*update.InlineRouteConfig)
+		w.applyRouteConfigUpdate(*update.InlineRouteConfig)
 		return
 	}
 
@@ -151,7 +155,7 @@ func (w *serviceUpdateWatcher) handleLDSResp(update xdsresource.ListenerUpdate,
 	w.rdsCancel = w.c.WatchRouteConfig(update.RouteConfigName, w.handleRDSResp)
 }
 
-func (w *serviceUpdateWatcher) updateVirtualHostsFromRDS(update xdsresource.RouteConfigUpdate) {
+func (w *serviceUpdateWatcher) applyRouteConfigUpdate(update xdsresource.RouteConfigUpdate) {
 	matchVh := xdsresource.FindBestMatchingVirtualHost(w.serviceName, update.VirtualHosts)
 	if matchVh == nil {
 		// No matching virtual host found.
@@ -160,6 +164,7 @@ func (w *serviceUpdateWatcher) updateVirtualHostsFromRDS(update xdsresource.Rout
 	}
 
 	w.lastUpdate.virtualHost = matchVh
+	w.lastUpdate.clusterSpecifierPlugins = update.ClusterSpecifierPlugins
 	w.serviceCb(w.lastUpdate, nil)
 }
 
@@ -179,7 +184,7 @@ func (w *serviceUpdateWatcher) handleRDSResp(update xdsresource.RouteConfigUpdat
 		w.serviceCb(serviceUpdate{}, err)
 		return
 	}
-	w.updateVirtualHostsFromRDS(update)
+	w.applyRouteConfigUpdate(update)
 }
 
 func (w *serviceUpdateWatcher) close() {

xds/internal/resolver/xds_resolver.go

@@ -205,7 +205,6 @@ func (r *xdsResolver) sendNewServiceConfig(cs *configSelector) bool {
 		return true
 	}
 
-	// Produce the service config.
 	sc, err := serviceConfigJSON(r.activeClusters)
 	if err != nil {
 		// JSON marshal error; should never happen.