xds: Added env var for RLS in xDS (#5050)

* xds: Added env var for RLS in xDS

internal/envconfig/xds.go

@@ -42,6 +42,7 @@ const (
 	aggregateAndDNSSupportEnv    = "GRPC_XDS_EXPERIMENTAL_ENABLE_AGGREGATE_AND_LOGICAL_DNS_CLUSTER"
 	rbacSupportEnv               = "GRPC_XDS_EXPERIMENTAL_RBAC"
 	federationEnv                = "GRPC_EXPERIMENTAL_XDS_FEDERATION"
+	rlsInXDSEnv                  = "GRPC_EXPERIMENTAL_XDS_RLS_LB"
 
 	c2pResolverTestOnlyTrafficDirectorURIEnv = "GRPC_TEST_ONLY_GOOGLE_C2P_RESOLVER_TRAFFIC_DIRECTOR_URI"
 )
@@ -85,6 +86,12 @@ var (
 	// XDSFederation indicates whether federation support is enabled.
 	XDSFederation = strings.EqualFold(os.Getenv(federationEnv), "true")
 
+	// XDSRLS indicates whether processing of Cluster Specifier plugins and
+	// support for the RLS CLuster Specifier is enabled, which can be enabled by
+	// setting the environment variable "GRPC_EXPERIMENTAL_XDS_RLS_LB" to
+	// "true".
+	XDSRLS = strings.EqualFold(os.Getenv(rlsInXDSEnv), "true")
+
 	// C2PResolverTestOnlyTrafficDirectorURI is the TD URI for testing.
 	C2PResolverTestOnlyTrafficDirectorURI = os.Getenv(c2pResolverTestOnlyTrafficDirectorURIEnv)
 )

xds/internal/xdsclient/xdsresource/unmarshal_rds.go

@@ -83,9 +83,13 @@ func unmarshalRouteConfigResource(r *anypb.Any, logger *grpclog.PrefixLogger) (s
 // we are looking for.
 func generateRDSUpdateFromRouteConfiguration(rc *v3routepb.RouteConfiguration, logger *grpclog.PrefixLogger, v2 bool) (RouteConfigUpdate, error) {
 	vhs := make([]*VirtualHost, 0, len(rc.GetVirtualHosts()))
-	csps, err := processClusterSpecifierPlugins(rc.ClusterSpecifierPlugins)
-	if err != nil {
-		return RouteConfigUpdate{}, fmt.Errorf("received route is invalid %v", err)
+	csps := make(map[string]clusterspecifier.BalancerConfig)
+	if envconfig.XDSRLS {
+		var err error
+		csps, err = processClusterSpecifierPlugins(rc.ClusterSpecifierPlugins)
+		if err != nil {
+			return RouteConfigUpdate{}, fmt.Errorf("received route is invalid %v", err)
+		}
 	}
 	// cspNames represents all the cluster specifiers referenced by Route
 	// Actions - any cluster specifiers not referenced by a Route Action can be
@@ -348,6 +352,9 @@ func routesProtoToSlice(routes []*v3routepb.Route, csps map[string]clusterspecif
 			case *v3routepb.RouteAction_ClusterHeader:
 				continue
 			case *v3routepb.RouteAction_ClusterSpecifierPlugin:
+				if !envconfig.XDSRLS {
+					return nil, nil, fmt.Errorf("route %+v, has an unknown ClusterSpecifier: %+v", r, a)
+				}
 				if _, ok := csps[a.ClusterSpecifierPlugin]; !ok {
 					// "When processing RouteActions, if any action includes a
 					// cluster_specifier_plugin value that is not in

xds/internal/xdsclient/xdsresource/unmarshal_rds_test.go

@@ -110,6 +110,12 @@ func (s) TestRDSGenerateRDSUpdateFromRouteConfiguration(t *testing.T) {
 				}},
 			}
 		}
+		goodUpdate = RouteConfigUpdate{
+			VirtualHosts: []*VirtualHost{{
+				Domains: []string{ldsTarget},
+				Routes:  nil,
+			}},
+		}
 		goodUpdateWithClusterSpecifierPluginA = RouteConfigUpdate{
 			VirtualHosts: []*VirtualHost{{
 				Domains: []string{ldsTarget},
@@ -166,11 +172,17 @@ func (s) TestRDSGenerateRDSUpdateFromRouteConfiguration(t *testing.T) {
 		defaultRetryBackoff = RetryBackoff{BaseInterval: 25 * time.Millisecond, MaxInterval: 250 * time.Millisecond}
 	)
 
+	oldRLS := envconfig.XDSRLS
+	defer func() {
+		envconfig.XDSRLS = oldRLS
+	}()
+
 	tests := []struct {
 		name       string
 		rc         *v3routepb.RouteConfiguration
 		wantUpdate RouteConfigUpdate
 		wantError  bool
+		rlsEnabled bool
 	}{
 		{
 			name: "default-route-match-field-is-nil",
@@ -640,21 +652,24 @@ func (s) TestRDSGenerateRDSUpdateFromRouteConfiguration(t *testing.T) {
 			rc: goodRouteConfigWithClusterSpecifierPlugins([]*v3routepb.ClusterSpecifierPlugin{
 				clusterSpecifierPlugin("cspA", configOfClusterSpecifierDoesntExist),
 			}, []string{"cspA"}),
-			wantError: true,
+			wantError:  true,
+			rlsEnabled: true,
 		},
 		{
 			name: "error-in-cluster-specifier-plugin-conversion-method",
 			rc: goodRouteConfigWithClusterSpecifierPlugins([]*v3routepb.ClusterSpecifierPlugin{
 				clusterSpecifierPlugin("cspA", errorClusterSpecifierConfig),
 			}, []string{"cspA"}),
-			wantError: true,
+			wantError:  true,
+			rlsEnabled: true,
 		},
 		{
 			name: "route-action-that-references-undeclared-cluster-specifier-plugin",
 			rc: goodRouteConfigWithClusterSpecifierPlugins([]*v3routepb.ClusterSpecifierPlugin{
 				clusterSpecifierPlugin("cspA", mockClusterSpecifierConfig),
 			}, []string{"cspA", "cspB"}),
-			wantError: true,
+			wantError:  true,
+			rlsEnabled: true,
 		},
 		{
 			name: "emitted-cluster-specifier-plugins",
@@ -662,6 +677,7 @@ func (s) TestRDSGenerateRDSUpdateFromRouteConfiguration(t *testing.T) {
 				clusterSpecifierPlugin("cspA", mockClusterSpecifierConfig),
 			}, []string{"cspA"}),
 			wantUpdate: goodUpdateWithClusterSpecifierPluginA,
+			rlsEnabled: true,
 		},
 		{
 			name: "deleted-cluster-specifier-plugins-not-referenced",
@@ -670,10 +686,26 @@ func (s) TestRDSGenerateRDSUpdateFromRouteConfiguration(t *testing.T) {
 				clusterSpecifierPlugin("cspB", mockClusterSpecifierConfig),
 			}, []string{"cspA"}),
 			wantUpdate: goodUpdateWithClusterSpecifierPluginA,
+			rlsEnabled: true,
+		},
+		{
+			name: "ignore-error-in-cluster-specifier-plugin",
+			rc: goodRouteConfigWithClusterSpecifierPlugins([]*v3routepb.ClusterSpecifierPlugin{
+				clusterSpecifierPlugin("cspA", configOfClusterSpecifierDoesntExist),
+			}, []string{}),
+			wantUpdate: goodUpdate,
+		},
+		{
+			name: "cluster-specifier-plugin-referenced-env-var-off",
+			rc: goodRouteConfigWithClusterSpecifierPlugins([]*v3routepb.ClusterSpecifierPlugin{
+				clusterSpecifierPlugin("cspA", mockClusterSpecifierConfig),
+			}, []string{"cspA"}),
+			wantError: true,
 		},
 	}
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
+			envconfig.XDSRLS = test.rlsEnabled
 			gotUpdate, gotError := generateRDSUpdateFromRouteConfiguration(test.rc, nil, false)
 			if (gotError != nil) != test.wantError ||
 				!cmp.Equal(gotUpdate, test.wantUpdate, cmpopts.EquateEmpty(),