xds: Add route to filterchain (grpc#4610)

* Added RDS Information from LDS in filter chain

xds/internal/server/listener_wrapper_test.go

@@ -30,6 +30,7 @@ import (
 
 	v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	v3listenerpb "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3"
+	v3routepb "github.com/envoyproxy/go-control-plane/envoy/config/route/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"
 	wrapperspb "github.com/golang/protobuf/ptypes/wrappers"
@@ -87,7 +88,20 @@ var listenerWithFilterChains = &v3listenerpb.Listener{
 				{
 					Name: "filter-1",
 					ConfigType: &v3listenerpb.Filter_TypedConfig{
-						TypedConfig: testutils.MarshalAny(&v3httppb.HttpConnectionManager{}),
+						TypedConfig: testutils.MarshalAny(&v3httppb.HttpConnectionManager{
+							RouteSpecifier: &v3httppb.HttpConnectionManager_RouteConfig{
+								RouteConfig: &v3routepb.RouteConfiguration{
+									Name: "routeName",
+									VirtualHosts: []*v3routepb.VirtualHost{{
+										Domains: []string{"lds.target.good:3333"},
+										Routes: []*v3routepb.Route{{
+											Match: &v3routepb.RouteMatch{
+												PathSpecifier: &v3routepb.RouteMatch_Prefix{Prefix: "/"},
+											},
+											Action: &v3routepb.Route_NonForwardingAction{},
+										}}}}},
+							},
+						}),
 					},
 				},
 			},

xds/internal/testutils/e2e/clientresources.go

@@ -199,7 +199,20 @@ func DefaultServerListener(host string, port uint32, secLevel SecurityLevel) *v3
 					{
 						Name: "filter-1",
 						ConfigType: &v3listenerpb.Filter_TypedConfig{
-							TypedConfig: testutils.MarshalAny(&v3httppb.HttpConnectionManager{}),
+							TypedConfig: testutils.MarshalAny(&v3httppb.HttpConnectionManager{
+								RouteSpecifier: &v3httppb.HttpConnectionManager_RouteConfig{
+									RouteConfig: &v3routepb.RouteConfiguration{
+										Name: "routeName",
+										VirtualHosts: []*v3routepb.VirtualHost{{
+											Domains: []string{"lds.target.good:3333"},
+											Routes: []*v3routepb.Route{{
+												Match: &v3routepb.RouteMatch{
+													PathSpecifier: &v3routepb.RouteMatch_Prefix{Prefix: "/"},
+												},
+												Action: &v3routepb.Route_NonForwardingAction{},
+											}}}}},
+								},
+							}),
 						},
 					},
 				},
@@ -230,7 +243,20 @@ func DefaultServerListener(host string, port uint32, secLevel SecurityLevel) *v3
 					{
 						Name: "filter-1",
 						ConfigType: &v3listenerpb.Filter_TypedConfig{
-							TypedConfig: testutils.MarshalAny(&v3httppb.HttpConnectionManager{}),
+							TypedConfig: testutils.MarshalAny(&v3httppb.HttpConnectionManager{
+								RouteSpecifier: &v3httppb.HttpConnectionManager_RouteConfig{
+									RouteConfig: &v3routepb.RouteConfiguration{
+										Name: "routeName",
+										VirtualHosts: []*v3routepb.VirtualHost{{
+											Domains: []string{"lds.target.good:3333"},
+											Routes: []*v3routepb.Route{{
+												Match: &v3routepb.RouteMatch{
+													PathSpecifier: &v3routepb.RouteMatch_Prefix{Prefix: "/"},
+												},
+												Action: &v3routepb.Route_NonForwardingAction{},
+											}}}}},
+								},
+							}),
 						},
 					},
 				},

xds/internal/xdsclient/client.go

@@ -294,6 +294,25 @@ type HashPolicy struct {
 	RegexSubstitution string
 }
 
+// RouteAction is the action of the route from a received RDS response.
+type RouteAction int
+
+const (
+	// RouteActionUnsupported are routing types currently unsupported by grpc.
+	// According to A36, "A Route with an inappropriate action causes RPCs
+	// matching that route to fail."
+	RouteActionUnsupported RouteAction = iota
+	// RouteActionRoute is the expected route type on the client side. Route
+	// represents routing a request to some upstream cluster. On the client
+	// side, if an RPC matches to a route that is not RouteActionRoute, the RPC
+	// will fail according to A36.
+	RouteActionRoute
+	// RouteActionNonForwardingAction is the expected route type on the server
+	// side. NonForwardingAction represents when a route will generate a
+	// response directly, without forwarding to an upstream host.
+	RouteActionNonForwardingAction
+)
+
 // Route is both a specification of how to match a request as well as an
 // indication of the action to take upon match.
 type Route struct {
@@ -321,6 +340,8 @@ type Route struct {
 	// unused if the matching WeightedCluster contains an override for that
 	// filter.
 	HTTPFilterConfigOverride map[string]httpfilter.FilterConfig
+
+	RouteAction RouteAction
 }
 
 // WeightedCluster contains settings for an xds RouteAction.WeightedCluster.

xds/internal/xdsclient/filter_chain.go

@@ -24,8 +24,10 @@ import (
 	"net"
 
 	v3listenerpb "github.com/envoyproxy/go-control-plane/envoy/config/listener/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"
 	"github.com/golang/protobuf/proto"
+	"github.com/golang/protobuf/ptypes"
 	"google.golang.org/grpc/xds/internal/version"
 )
 
@@ -54,6 +56,15 @@ type FilterChain struct {
 	SecurityCfg *SecurityConfig
 	// HTTPFilters represent the HTTP Filters that comprise this FilterChain.
 	HTTPFilters []HTTPFilter
+	// RouteConfigName is the route configuration name for this FilterChain.
+	//
+	// Only one of RouteConfigName and InlineRouteConfig is set.
+	RouteConfigName string
+	// InlineRouteConfig is the inline route configuration (RDS response)
+	// returned for this filter chain.
+	//
+	// Only one of RouteConfigName and InlineRouteConfig is set.
+	InlineRouteConfig *RouteConfigUpdate
 }
 
 // SourceType specifies the connection source IP match type.
@@ -109,6 +120,11 @@ type FilterChainManager struct {
 	dstPrefixes []*destPrefixEntry
 
 	def *FilterChain // Default filter chain, if specified.
+
+	// RouteConfigNames are the route configuration names which need to be
+	// dynamically queried for RDS Configuration for any FilterChains which
+	// specify to load RDS Configuration dynamically.
+	RouteConfigNames map[string]bool
 }
 
 // destPrefixEntry is the value type of the map indexed on destination prefixes.
@@ -158,7 +174,10 @@ type sourcePrefixEntry struct {
 // create a FilterChainManager.
 func NewFilterChainManager(lis *v3listenerpb.Listener) (*FilterChainManager, error) {
 	// Parse all the filter chains and build the internal data structures.
-	fci := &FilterChainManager{dstPrefixMap: make(map[string]*destPrefixEntry)}
+	fci := &FilterChainManager{
+		dstPrefixMap:     make(map[string]*destPrefixEntry),
+		RouteConfigNames: make(map[string]bool),
+	}
 	if err := fci.addFilterChains(lis.GetFilterChains()); err != nil {
 		return nil, err
 	}
@@ -187,7 +206,7 @@ func NewFilterChainManager(lis *v3listenerpb.Listener) (*FilterChainManager, err
 	var def *FilterChain
 	if dfc := lis.GetDefaultFilterChain(); dfc != nil {
 		var err error
-		if def, err = filterChainFromProto(dfc); err != nil {
+		if def, err = fci.filterChainFromProto(dfc); err != nil {
 			return nil, err
 		}
 	}
@@ -368,7 +387,7 @@ func (fci *FilterChainManager) addFilterChainsForSourcePorts(srcEntry *sourcePre
 		srcPorts = append(srcPorts, int(port))
 	}
 
-	fc, err := filterChainFromProto(fcProto)
+	fc, err := fci.filterChainFromProto(fcProto)
 	if err != nil {
 		return err
 	}
@@ -391,13 +410,19 @@ func (fci *FilterChainManager) addFilterChainsForSourcePorts(srcEntry *sourcePre
 }
 
 // filterChainFromProto extracts the relevant information from the FilterChain
-// proto and stores it in our internal representation.
-func filterChainFromProto(fc *v3listenerpb.FilterChain) (*FilterChain, error) {
-	httpFilters, err := processNetworkFilters(fc.GetFilters())
+// proto and stores it in our internal representation. It also persists any
+// RouteNames which need to be queried dynamically via RDS.
+func (fci *FilterChainManager) filterChainFromProto(fc *v3listenerpb.FilterChain) (*FilterChain, error) {
+	filterChain, err := processNetworkFilters(fc.GetFilters())
 	if err != nil {
 		return nil, err
 	}
-	filterChain := &FilterChain{HTTPFilters: httpFilters}
+	// These route names will be dynamically queried via RDS in the wrapped
+	// listener, which receives the LDS response, if specified for the filter
+	// chain.
+	if filterChain.RouteConfigName != "" {
+		fci.RouteConfigNames[filterChain.RouteConfigName] = true
+	}
 	// If the transport_socket field is not specified, it means that the control
 	// plane has not sent us any security config. This is fine and the server
 	// will use the fallback credentials configured as part of the
@@ -435,6 +460,93 @@ func filterChainFromProto(fc *v3listenerpb.FilterChain) (*FilterChain, error) {
 	return filterChain, nil
 }
 
+func processNetworkFilters(filters []*v3listenerpb.Filter) (*FilterChain, error) {
+	filterChain := &FilterChain{}
+	seenNames := make(map[string]bool, len(filters))
+	seenHCM := false
+	for _, filter := range filters {
+		name := filter.GetName()
+		if name == "" {
+			return nil, fmt.Errorf("network filters {%+v} is missing name field in filter: {%+v}", filters, filter)
+		}
+		if seenNames[name] {
+			return nil, fmt.Errorf("network filters {%+v} has duplicate filter name %q", filters, name)
+		}
+		seenNames[name] = true
+
+		// Network filters have a oneof field named `config_type` where we
+		// only support `TypedConfig` variant.
+		switch typ := filter.GetConfigType().(type) {
+		case *v3listenerpb.Filter_TypedConfig:
+			// The typed_config field has an `anypb.Any` proto which could
+			// directly contain the serialized bytes of the actual filter
+			// configuration, or it could be encoded as a `TypedStruct`.
+			// TODO: Add support for `TypedStruct`.
+			tc := filter.GetTypedConfig()
+
+			// The only network filter that we currently support is the v3
+			// HttpConnectionManager. So, we can directly check the type_url
+			// and unmarshal the config.
+			// TODO: Implement a registry of supported network filters (like
+			// we have for HTTP filters), when we have to support network
+			// filters other than HttpConnectionManager.
+			if tc.GetTypeUrl() != version.V3HTTPConnManagerURL {
+				return nil, fmt.Errorf("network filters {%+v} has unsupported network filter %q in filter {%+v}", filters, tc.GetTypeUrl(), filter)
+			}
+			hcm := &v3httppb.HttpConnectionManager{}
+			if err := ptypes.UnmarshalAny(tc, hcm); err != nil {
+				return nil, fmt.Errorf("network filters {%+v} failed unmarshaling of network filter {%+v}: %v", filters, filter, err)
+			}
+			// "Any filters after HttpConnectionManager should be ignored during
+			// connection processing but still be considered for validity.
+			// HTTPConnectionManager must have valid http_filters." - A36
+			filters, err := processHTTPFilters(hcm.GetHttpFilters(), true)
+			if err != nil {
+				return nil, fmt.Errorf("network filters {%+v} had invalid server side HTTP Filters {%+v}", filters, hcm.GetHttpFilters())
+			}
+			if !seenHCM {
+				// TODO: Implement terminal filter logic, as per A36.
+				filterChain.HTTPFilters = filters
+				seenHCM = true
+				switch hcm.RouteSpecifier.(type) {
+				case *v3httppb.HttpConnectionManager_Rds:
+					if hcm.GetRds().GetConfigSource().GetAds() == nil {
+						return nil, fmt.Errorf("ConfigSource is not ADS: %+v", hcm)
+					}
+					name := hcm.GetRds().GetRouteConfigName()
+					if name == "" {
+						return nil, fmt.Errorf("empty route_config_name: %+v", hcm)
+					}
+					filterChain.RouteConfigName = name
+				case *v3httppb.HttpConnectionManager_RouteConfig:
+					// "RouteConfiguration validation logic inherits all
+					// previous validations made for client-side usage as RDS
+					// does not distinguish between client-side and
+					// server-side." - A36
+					// Can specify v3 here, as will never get to this function
+					// if v2.
+					routeU, err := generateRDSUpdateFromRouteConfiguration(hcm.GetRouteConfig(), nil, false)
+					if err != nil {
+						return nil, fmt.Errorf("failed to parse inline RDS resp: %v", err)
+					}
+					filterChain.InlineRouteConfig = &routeU
+				case nil:
+					// No-op, as no route specifier is a valid configuration on
+					// the server side.
+				default:
+					return nil, fmt.Errorf("unsupported type %T for RouteSpecifier", hcm.RouteSpecifier)
+				}
+			}
+		default:
+			return nil, fmt.Errorf("network filters {%+v} has unsupported config_type %T in filter %s", filters, typ, filter.GetName())
+		}
+	}
+	if !seenHCM {
+		return nil, fmt.Errorf("network filters {%+v} missing HttpConnectionManager filter", filters)
+	}
+	return filterChain, nil
+}
+
 // FilterChainLookupParams wraps parameters to be passed to Lookup.
 type FilterChainLookupParams struct {
 	// IsUnspecified indicates whether the server is listening on a wildcard