Routes for API Gateway (#17158)

* XDS primitive generation for endpoints and clusters

Co-authored-by: Nathan Coleman <nathan.coleman@hashicorp.com>

* server_test

* deleted extra file

* add missing parents to test

* checkpoint

* delete extra file

* httproute flattening code

* linting issue

* so close on this, calling for tonight

* unit test passing

* add in header manip to virtual host

* upstream rebuild commented out

* Use consistent upstream name whether or not we're rebuilding

* Start working through route naming logic

* Fix typos in test descriptions

* Simplify route naming logic

* Simplify RebuildHTTPRouteUpstream

* Merge additional compiled discovery chains instead of overwriting

* Use correct chain for flattened route, clean up + add TODOs

* Remove empty conditional branch

* Restore previous variable declaration

Limit the scope of this PR

* Clean up, improve TODO

* add logging, clean up todos

* clean up function

---------

Co-authored-by: Nathan Coleman <nathan.coleman@hashicorp.com>

agent/consul/discoverychain/gateway.go

@@ -59,9 +59,14 @@ func (l *GatewayChainSynthesizer) SetHostname(hostname string) {
 // single hostname can be specified in multiple routes. Routing for a given
 // hostname must behave based on the aggregate of all rules that apply to it.
 func (l *GatewayChainSynthesizer) AddHTTPRoute(route structs.HTTPRouteConfigEntry) {
-	hostnames := route.FilteredHostnames(l.hostname)
+	//TODO maps are pointers in golang, might not need to set it like this, test later
+	l.matchesByHostname = getHostMatches(l.hostname, &route, l.matchesByHostname)
+}
+
+func getHostMatches(hostname string, route *structs.HTTPRouteConfigEntry, currentMatches map[string][]hostnameMatch) map[string][]hostnameMatch {
+	hostnames := route.FilteredHostnames(hostname)
 	for _, host := range hostnames {
-		matches, ok := l.matchesByHostname[host]
+		matches, ok := currentMatches[host]
 		if !ok {
 			matches = []hostnameMatch{}
 		}
@@ -90,8 +95,10 @@ func (l *GatewayChainSynthesizer) AddHTTPRoute(route structs.HTTPRouteConfigEntr
 			}
 		}
 
-		l.matchesByHostname[host] = matches
+		currentMatches[host] = matches
 	}
+	//TODO def don't think this is needed just testing for now, remove if not needed
+	return currentMatches
 }
 
 // Synthesize assembles a synthetic discovery chain from multiple other discovery chains
@@ -116,6 +123,7 @@ func (l *GatewayChainSynthesizer) Synthesize(chains ...*structs.CompiledDiscover
 
 	compiledChains := make([]*structs.CompiledDiscoveryChain, 0, len(set))
 	for i, service := range services {
+
 		entries := set[i]
 
 		compiled, err := Compile(CompileRequest{
@@ -126,7 +134,6 @@ func (l *GatewayChainSynthesizer) Synthesize(chains ...*structs.CompiledDiscover
 			EvaluateInTrustDomain: l.trustDomain,
 			Entries:               entries,
 		})
-
 		if err != nil {
 			return nil, nil, err
 		}
@@ -188,17 +195,44 @@ func (l *GatewayChainSynthesizer) Synthesize(chains ...*structs.CompiledDiscover
 // consolidateHTTPRoutes combines all rules into the shortest possible list of routes
 // with one route per hostname containing all rules for that hostname.
 func (l *GatewayChainSynthesizer) consolidateHTTPRoutes() []structs.HTTPRouteConfigEntry {
+	return consolidateHTTPRoutes(l.matchesByHostname, l.suffix, l.gateway)
+}
+
+// FlattenHTTPRoute takes in a route and its parent config entries and returns a list of flattened routes
+func FlattenHTTPRoute(route *structs.HTTPRouteConfigEntry, listener *structs.APIGatewayListener, gateway *structs.APIGatewayConfigEntry) []structs.HTTPRouteConfigEntry {
+	//build map[string][]hostnameMatch for route
+	matches := map[string][]hostnameMatch{}
+	matches = getHostMatches(listener.GetHostname(), route, matches)
+	return consolidateHTTPRoutes(matches, listener.Name, gateway)
+}
+
+func RebuildHTTPRouteUpstream(route structs.HTTPRouteConfigEntry, listener structs.APIGatewayListener) structs.Upstream {
+	return structs.Upstream{
+		DestinationName:      route.GetName(),
+		DestinationNamespace: route.NamespaceOrDefault(),
+		DestinationPartition: route.PartitionOrDefault(),
+		IngressHosts:         route.Hostnames,
+		LocalBindPort:        listener.Port,
+		Config: map[string]interface{}{
+			"protocol": string(listener.Protocol),
+		},
+	}
+}
+
+// ConsolidateHTTPRoutes combines all rules into the shortest possible list of routes
+// with one route per hostname containing all rules for that hostname.
+func consolidateHTTPRoutes(matchesByHostname map[string][]hostnameMatch, suffix string, gateway *structs.APIGatewayConfigEntry) []structs.HTTPRouteConfigEntry {
 	var routes []structs.HTTPRouteConfigEntry
 
-	for hostname, rules := range l.matchesByHostname {
+	for hostname, rules := range matchesByHostname {
 		// Create route for this hostname
 		route := structs.HTTPRouteConfigEntry{
 			Kind:           structs.HTTPRoute,
-			Name:           fmt.Sprintf("%s-%s-%s", l.gateway.Name, l.suffix, hostsKey(hostname)),
+			Name:           fmt.Sprintf("%s-%s-%s", gateway.Name, suffix, hostsKey(hostname)),
 			Hostnames:      []string{hostname},
 			Rules:          make([]structs.HTTPRouteRule, 0, len(rules)),
-			Meta:           l.gateway.Meta,
-			EnterpriseMeta: l.gateway.EnterpriseMeta,
+			Meta:           gateway.Meta,
+			EnterpriseMeta: gateway.EnterpriseMeta,
 		}
 
 		// Sort rules for this hostname in order of precedence
@@ -258,12 +292,14 @@ func (l *GatewayChainSynthesizer) synthesizeEntries() ([]structs.IngressService,
 	entries := []*configentry.DiscoveryChainSet{}
 
 	for _, route := range l.consolidateHTTPRoutes() {
-		entrySet := configentry.NewDiscoveryChainSet()
 		ingress, router, splitters, defaults := synthesizeHTTPRouteDiscoveryChain(route)
+
+		services = append(services, ingress)
+
+		entrySet := configentry.NewDiscoveryChainSet()
 		entrySet.AddRouters(router)
 		entrySet.AddSplitters(splitters...)
 		entrySet.AddServices(defaults...)
-		services = append(services, ingress)
 		entries = append(entries, entrySet)
 	}
 

agent/proxycfg/api_gateway.go

@@ -6,7 +6,6 @@ package proxycfg
 import (
 	"context"
 	"fmt"
-
 	"github.com/hashicorp/consul/acl"
 	cachetype "github.com/hashicorp/consul/agent/cache-types"
 	"github.com/hashicorp/consul/agent/proxycfg/internal/watch"
@@ -125,7 +124,9 @@ func (h *handlerAPIGateway) handleUpdate(ctx context.Context, u UpdateEvent, sna
 			return err
 		}
 	default:
-		return (*handlerUpstreams)(h).handleUpdateUpstreams(ctx, u, snap)
+		if err := (*handlerUpstreams)(h).handleUpdateUpstreams(ctx, u, snap); err != nil {
+			return err
+		}
 	}
 
 	return h.recompileDiscoveryChains(snap)
@@ -308,7 +309,7 @@ func (h *handlerAPIGateway) handleRouteConfigUpdate(ctx context.Context, u Updat
 						DestinationNamespace: service.NamespaceOrDefault(),
 						DestinationPartition: service.PartitionOrDefault(),
 						LocalBindPort:        listener.Port,
-						// TODO IngressHosts:         g.Hosts,
+						//IngressHosts:         g.Hosts,
 						// Pass the protocol that was configured on the listener in order
 						// to force that protocol on the Envoy listener.
 						Config: map[string]interface{}{
@@ -452,7 +453,11 @@ func (h *handlerAPIGateway) recompileDiscoveryChains(snap *ConfigSnapshot) error
 		}
 	}
 
-	snap.APIGateway.DiscoveryChain = synthesizedChains
+	// Merge in additional discovery chains
+	for id, chain := range synthesizedChains {
+		snap.APIGateway.DiscoveryChain[id] = chain
+	}
+
 	return nil
 }
 

agent/proxycfg/snapshot.go

@@ -876,6 +876,7 @@ DOMAIN_LOOP:
 			}
 			synthesizer.AddTCPRoute(*route)
 			for _, service := range route.GetServices() {
+
 				id := NewUpstreamIDFromServiceName(structs.NewServiceName(service.Name, &service.EnterpriseMeta))
 				if chain := c.DiscoveryChain[id]; chain != nil {
 					chains = append(chains, chain)

agent/xds/clusters.go

@@ -813,10 +813,10 @@ func (s *ResourceGenerator) clustersFromSnapshotIngressGateway(cfgSnap *proxycfg
 func (s *ResourceGenerator) clustersFromSnapshotAPIGateway(cfgSnap *proxycfg.ConfigSnapshot) ([]proto.Message, error) {
 	var clusters []proto.Message
 	createdClusters := make(map[proxycfg.UpstreamID]bool)
-	readyUpstreams := getReadyUpstreams(cfgSnap)
+	readyUpstreamsList := getReadyUpstreams(cfgSnap)
 
-	for listenerKey, upstreams := range readyUpstreams {
-		for _, upstream := range upstreams {
+	for _, readyUpstreams := range readyUpstreamsList {
+		for _, upstream := range readyUpstreams.upstreams {
 			uid := proxycfg.NewUpstreamID(&upstream)
 
 			// If we've already created a cluster for this upstream, skip it. Multiple listeners may
@@ -839,23 +839,15 @@ func (s *ResourceGenerator) clustersFromSnapshotAPIGateway(cfgSnap *proxycfg.Con
 			}
 
 			for _, cluster := range upstreamClusters {
-				// TODO Something analogous to s.configIngressUpstreamCluster(c, cfgSnap, listenerKey, &u)
-				//   but not sure what that func does yet
-				s.configAPIUpstreamCluster(cluster, cfgSnap, listenerKey, &upstream)
 				clusters = append(clusters, cluster)
 			}
-			createdClusters[uid] = true
 
+			createdClusters[uid] = true
 		}
 	}
 	return clusters, nil
 }
 
-func (s *ResourceGenerator) configAPIUpstreamCluster(c *envoy_cluster_v3.Cluster, cfgSnap *proxycfg.ConfigSnapshot, listenerKey proxycfg.APIGatewayListenerKey, u *structs.Upstream) {
-	//TODO I don't think this is currently needed with what api gateway supports, but will be needed in the future
-
-}
-
 func (s *ResourceGenerator) configIngressUpstreamCluster(c *envoy_cluster_v3.Cluster, cfgSnap *proxycfg.ConfigSnapshot, listenerKey proxycfg.IngressListenerKey, u *structs.Upstream) {
 	var threshold *envoy_cluster_v3.CircuitBreakers_Thresholds
 	setThresholdLimit := func(limitType string, limit int) {

agent/xds/endpoints.go

@@ -521,9 +521,18 @@ func (s *ResourceGenerator) endpointsFromSnapshotIngressGateway(cfgSnap *proxycf
 	return resources, nil
 }
 
-func getReadyUpstreams(cfgSnap *proxycfg.ConfigSnapshot) map[proxycfg.APIGatewayListenerKey][]structs.Upstream {
+// helper struct to persist upstream parent information when ready upstream list is built out
+type readyUpstreams struct {
+	listenerKey      proxycfg.APIGatewayListenerKey
+	listenerCfg      structs.APIGatewayListener
+	boundListenerCfg structs.BoundAPIGatewayListener
+	routeReference   structs.ResourceReference
+	upstreams        []structs.Upstream
+}
+
+func getReadyUpstreams(cfgSnap *proxycfg.ConfigSnapshot) map[string]readyUpstreams {
 
-	readyUpstreams := map[proxycfg.APIGatewayListenerKey][]structs.Upstream{}
+	ready := map[string]readyUpstreams{}
 	for _, l := range cfgSnap.APIGateway.Listeners {
 		//need to account for the state of the Listener when building the upstreams list
 		if !cfgSnap.APIGateway.GatewayConfig.ListenerIsReady(l.Name) {
@@ -534,24 +543,37 @@ func getReadyUpstreams(cfgSnap *proxycfg.ConfigSnapshot) map[proxycfg.APIGateway
 		for _, routeRef := range boundListener.Routes {
 			//get upstreams
 			upstreamMap := cfgSnap.APIGateway.Upstreams[routeRef]
-			for listenerKey, upstreams := range upstreamMap {
+			for _, upstreams := range upstreamMap {
 				for _, u := range upstreams {
-					readyUpstreams[listenerKey] = append(readyUpstreams[listenerKey], u)
+					r, ok := ready[l.Name]
+					if !ok {
+						r = readyUpstreams{
+							listenerKey: proxycfg.APIGatewayListenerKey{
+								Protocol: string(l.Protocol),
+								Port:     l.Port,
+							},
+							listenerCfg:      l,
+							boundListenerCfg: boundListener,
+							routeReference:   routeRef,
+						}
+					}
+					r.upstreams = append(r.upstreams, u)
+					ready[l.Name] = r
 				}
 			}
 		}
 	}
-	return readyUpstreams
+	return ready
 }
 
 func (s *ResourceGenerator) endpointsFromSnapshotAPIGateway(cfgSnap *proxycfg.ConfigSnapshot) ([]proto.Message, error) {
 	var resources []proto.Message
 	createdClusters := make(map[proxycfg.UpstreamID]bool)
 
-	readyUpstreams := getReadyUpstreams(cfgSnap)
+	readyUpstreamsList := getReadyUpstreams(cfgSnap)
 
-	for _, upstreams := range readyUpstreams {
-		for _, u := range upstreams {
+	for _, readyUpstreams := range readyUpstreamsList {
+		for _, u := range readyUpstreams.upstreams {
 			uid := proxycfg.NewUpstreamID(&u)
 
 			// If we've already created endpoints for this upstream, skip it. Multiple listeners may

agent/xds/routes.go

@@ -6,6 +6,7 @@ package xds
 import (
 	"errors"
 	"fmt"
+	"github.com/hashicorp/consul/agent/consul/discoverychain"
 	"net"
 	"sort"
 	"strings"
@@ -36,13 +37,7 @@ func (s *ResourceGenerator) routesFromSnapshot(cfgSnap *proxycfg.ConfigSnapshot)
 	case structs.ServiceKindIngressGateway:
 		return s.routesForIngressGateway(cfgSnap)
 	case structs.ServiceKindAPIGateway:
-		// TODO Find a cleaner solution, can't currently pass unexported property types
-		var err error
-		cfgSnap.IngressGateway, err = cfgSnap.APIGateway.ToIngress(cfgSnap.Datacenter)
-		if err != nil {
-			return nil, err
-		}
-		return s.routesForIngressGateway(cfgSnap)
+		return s.routesForAPIGateway(cfgSnap)
 	case structs.ServiceKindTerminatingGateway:
 		return s.routesForTerminatingGateway(cfgSnap)
 	case structs.ServiceKindMeshGateway:
@@ -430,6 +425,75 @@ func (s *ResourceGenerator) routesForIngressGateway(cfgSnap *proxycfg.ConfigSnap
 	return result, nil
 }
 
+// routesForAPIGateway returns the xDS API representation of the
+// "routes" in the snapshot.
+func (s *ResourceGenerator) routesForAPIGateway(cfgSnap *proxycfg.ConfigSnapshot) ([]proto.Message, error) {
+	var result []proto.Message
+
+	readyUpstreamsList := getReadyUpstreams(cfgSnap)
+
+	for _, readyUpstreams := range readyUpstreamsList {
+		listenerCfg := readyUpstreams.listenerCfg
+		// Do not create any route configuration for TCP listeners
+		if listenerCfg.Protocol == "tcp" {
+			continue
+		}
+
+		routeRef := readyUpstreams.routeReference
+		listenerKey := readyUpstreams.listenerKey
+
+		// Depending on their TLS config, upstreams are either attached to the
+		// default route or have their own routes. We'll add any upstreams that
+		// don't have custom filter chains and routes to this.
+		defaultRoute := &envoy_route_v3.RouteConfiguration{
+			Name: listenerKey.RouteName(),
+			// ValidateClusters defaults to true when defined statically and false
+			// when done via RDS. Re-set the reasonable value of true to prevent
+			// null-routing traffic.
+			ValidateClusters: makeBoolValue(true),
+		}
+
+		route, ok := cfgSnap.APIGateway.HTTPRoutes.Get(routeRef)
+		if !ok {
+			return nil, fmt.Errorf("missing route for route reference %s:%s", routeRef.Name, routeRef.Kind)
+		}
+
+		flattenedRoutes := discoverychain.FlattenHTTPRoute(route, &listenerCfg, cfgSnap.APIGateway.GatewayConfig)
+
+		for _, flattenedRoute := range flattenedRoutes {
+			flattenedRoute := flattenedRoute
+
+			upstream := discoverychain.RebuildHTTPRouteUpstream(flattenedRoute, listenerCfg)
+			uid := proxycfg.NewUpstreamID(&upstream)
+			chain := cfgSnap.APIGateway.DiscoveryChain[uid]
+			if chain == nil {
+				s.Logger.Debug("Discovery chain not found for flattened route", "discovery chain ID", uid)
+				continue
+			}
+
+			domains := generateUpstreamAPIsDomains(listenerKey, upstream, flattenedRoute.Hostnames)
+
+			virtualHost, err := s.makeUpstreamRouteForDiscoveryChain(cfgSnap, uid, chain, domains, false)
+			if err != nil {
+				return nil, err
+			}
+
+			injectHeaderManipToVirtualHostAPIGateway(&flattenedRoute, virtualHost)
+
+			// TODO Handle TLS config and add new route if appropriate
+			//   We need something analogous to routeNameForUpstream used below
+			//   But currently ToIngress is not handeling this usecase
+			defaultRoute.VirtualHosts = append(defaultRoute.VirtualHosts, virtualHost)
+		}
+
+		if len(defaultRoute.VirtualHosts) > 0 {
+			result = append(result, defaultRoute)
+		}
+	}
+
+	return result, nil
+}
+
 func makeHeadersValueOptions(vals map[string]string, add bool) []*envoy_core_v3.HeaderValueOption {
 	opts := make([]*envoy_core_v3.HeaderValueOption, 0, len(vals))
 	for k, v := range vals {
@@ -516,6 +580,11 @@ func generateUpstreamIngressDomains(listenerKey proxycfg.IngressListenerKey, u s
 	return domains
 }
 
+func generateUpstreamAPIsDomains(listenerKey proxycfg.APIGatewayListenerKey, u structs.Upstream, hosts []string) []string {
+	u.IngressHosts = hosts
+	return generateUpstreamIngressDomains(listenerKey, u)
+}
+
 func (s *ResourceGenerator) makeUpstreamRouteForDiscoveryChain(
 	cfgSnap *proxycfg.ConfigSnapshot,
 	uid proxycfg.UpstreamID,
@@ -1019,6 +1088,16 @@ func injectHeaderManipToRoute(dest *structs.ServiceRouteDestination, r *envoy_ro
 	return nil
 }
 
+func injectHeaderManipToVirtualHostAPIGateway(dest *structs.HTTPRouteConfigEntry, vh *envoy_route_v3.VirtualHost) {
+	for _, rule := range dest.Rules {
+		for _, header := range rule.Filters.Headers {
+			vh.RequestHeadersToAdd = append(vh.RequestHeadersToAdd, makeHeadersValueOptions(header.Add, true)...)
+			vh.RequestHeadersToAdd = append(vh.RequestHeadersToAdd, makeHeadersValueOptions(header.Set, false)...)
+			vh.RequestHeadersToRemove = append(vh.RequestHeadersToRemove, header.Remove...)
+		}
+	}
+}
+
 func injectHeaderManipToVirtualHost(dest *structs.IngressService, vh *envoy_route_v3.VirtualHost) error {
 	if !dest.RequestHeaders.IsZero() {
 		vh.RequestHeadersToAdd = append(

test/integration/connect/envoy/case-api-gateway-http-hostnames/verify.bats

@@ -6,7 +6,7 @@ load helpers
   retry_default curl -f -s localhost:20000/stats -o /dev/null
 }
 
-@test "api gateway should have be accepted and not conflicted" {
+@test "api gateway should have been accepted and not conflicted" {
   assert_config_entry_status Accepted True Accepted primary api-gateway api-gateway
   assert_config_entry_status Conflicted False NoConflict primary api-gateway api-gateway
 }
@@ -63,4 +63,4 @@ load helpers
   run retry_long curl -H "Host: foo.bar.baz" -s -f -d hello localhost:9995
   [ "$status" -eq 0 ]
   [[ "$output" == *"hello"* ]]
-}
+}

test/integration/connect/envoy/case-api-gateway-http-simple/verify.bats

@@ -6,7 +6,7 @@ load helpers
   retry_default curl -f -s localhost:20000/stats -o /dev/null
 }
 
-@test "api gateway should have be accepted and not conflicted" {
+@test "api gateway should have been accepted and not conflicted" {
   assert_config_entry_status Accepted True Accepted primary api-gateway api-gateway
   assert_config_entry_status Conflicted False NoConflict primary api-gateway api-gateway
 }
@@ -31,4 +31,4 @@ load helpers
   run retry_default sh -c "curl -s localhost:9998 | grep RBAC"
   [ "$status" -eq 0 ]
   [[ "$output" == "RBAC: access denied" ]]
-}
+}

test/integration/connect/envoy/case-api-gateway-http-tls-overlapping-hosts/verify.bats

@@ -6,7 +6,7 @@ load helpers
   retry_default curl -f -s localhost:20000/stats -o /dev/null
 }
 
-@test "api gateway should have be accepted and not conflicted" {
+@test "api gateway should have been accepted and not conflicted" {
   assert_config_entry_status Accepted True Accepted primary api-gateway api-gateway
   assert_config_entry_status Conflicted False NoConflict primary api-gateway api-gateway
 }
@@ -45,4 +45,4 @@ load helpers
 
 @test "api gateway should fall back to a connect certificate on conflicted SNI on listener 2" {
   assert_cert_has_cn localhost:9998 pri host.consul.example
-}
+}

test/integration/connect/envoy/case-api-gateway-tcp-simple/verify.bats

@@ -6,7 +6,7 @@ load helpers
   retry_default curl -f -s localhost:20000/stats -o /dev/null
 }
 
-@test "api gateway should have be accepted and not conflicted" {
+@test "api gateway should have been accepted and not conflicted" {
   assert_config_entry_status Accepted True Accepted primary api-gateway api-gateway
   assert_config_entry_status Conflicted False NoConflict primary api-gateway api-gateway
 }

test/integration/connect/envoy/case-api-gateway-tcp-tls-overlapping-hosts/verify.bats

@@ -6,7 +6,7 @@ load helpers
   retry_default curl -f -s localhost:20000/stats -o /dev/null
 }
 
-@test "api gateway should have be accepted and not conflicted" {
+@test "api gateway should have been accepted and not conflicted" {
   assert_config_entry_status Accepted True Accepted primary api-gateway api-gateway
   assert_config_entry_status Conflicted False NoConflict primary api-gateway api-gateway
 }
@@ -40,4 +40,4 @@ load helpers
 
 @test "api gateway should fall back to a connect certificate on conflicted SNI on listener 2" {
   assert_cert_has_cn localhost:9998 pri host.consul.example
-}
+}