envoy per gateway

* create a xdsIR per gateway

* create a infraIR per gateway

* use the gateway namespace-name as the key for above IRs

* populate the envoy bootstrap config with a node id that matches the IR
key

* populate the xds server snapshot for each xds request based on the
node id

Fixes: #349

Signed-off-by: Arko Dasgupta <[email protected]>

Author: arkodg

internal/cmd/xdstest.go

@@ -187,19 +187,19 @@ func xDSTest() error {
 		for {
 			time.Sleep(10 * time.Second)
 			logger.Info("Updating the cache for first-listener with first-route")
-			err := snapCache.GenerateNewSnapshot(cacheVersion1.GetXdsResources())
+			err := snapCache.GenerateNewSnapshot("", cacheVersion1.GetXdsResources())
 			if err != nil {
 				logger.Error(err, "Something went wrong with generating a snapshot")
 			}
 			time.Sleep(10 * time.Second)
 			logger.Info("Updating the cache for first-listener with second-route")
-			err = snapCache.GenerateNewSnapshot(cacheVersion2.GetXdsResources())
+			err = snapCache.GenerateNewSnapshot("", cacheVersion2.GetXdsResources())
 			if err != nil {
 				logger.Error(err, "Something went wrong with generating a snapshot")
 			}
 			time.Sleep(10 * time.Second)
 			logger.Info("Updating the cache for second-listener with second-route")
-			err = snapCache.GenerateNewSnapshot(cacheVersion3.GetXdsResources())
+			err = snapCache.GenerateNewSnapshot("", cacheVersion3.GetXdsResources())
 			if err != nil {
 				logger.Error(err, "Something went wrong with generating a snapshot")
 			}

internal/gatewayapi/runner/runner.go

@@ -95,22 +95,24 @@ func (r *Runner) subscribeAndTranslate(ctx context.Context) {
 			yamlInfraIR, _ := yaml.Marshal(&result.InfraIR)
 			r.Logger.Info(string(yamlInfraIR))
 
-			// Publish the IRs. Use the service name as the key
-			// to ensure there is always one element in the map.
+			// Publish the IRs.
 			// Also validate the ir before sending it.
-			if err := result.InfraIR.Validate(); err != nil {
-				r.Logger.Error(err, "unable to validate infra ir, skipped sending it")
-			} else {
-				r.InfraIR.Store(r.Name(), result.InfraIR)
+			for key, val := range result.InfraIR {
+				if err := val.Validate(); err != nil {
+					r.Logger.Error(err, "unable to validate infra ir, skipped sending it")
+				} else {
+					r.InfraIR.Store(key, val)
+				}
 			}
-
 			// Wait until all HTTPRoutes have been reconciled , else the translation
 			// result will be incomplete, and might cause churn in the data plane.
 			if xdsIRReady {
-				if err := result.XdsIR.Validate(); err != nil {
-					r.Logger.Error(err, "unable to validate xds ir, skipped sending it")
-				} else {
-					r.XdsIR.Store(r.Name(), result.XdsIR)
+				for key, val := range result.XdsIR {
+					if err := val.Validate(); err != nil {
+						r.Logger.Error(err, "unable to validate xds ir, skipped sending it")
+					} else {
+						r.XdsIR.Store(key, val)
+					}
 				}
 			}
 

internal/gatewayapi/translator.go

@@ -22,9 +22,9 @@ const (
 	KindService   = "Service"
 	KindSecret    = "Secret"
 
-	// OwningGatewayClassLabel is the owner reference label used for managed infra.
-	// The value should be the name of the accepted Envoy GatewayClass.
-	OwningGatewayClassLabel = "gateway.envoyproxy.io/owning-gatewayclass"
+	// OwningGatewayLabel is the owner reference label used for managed infra.
+	// The value should be the name of the accepted Envoy Gateway.
+	OwningGatewayLabel = "gateway.envoyproxy.io/owning-gateway"
 
 	// minEphemeralPort is the first port in the ephemeral port range.
 	minEphemeralPort = 1024
@@ -33,6 +33,9 @@ const (
 	wellKnownPortShift = 10000
 )
 
+type XdsIRMap map[string]*ir.Xds
+type InfraIRMap map[string]*ir.Infra
+
 // Resources holds the Gateway API and related
 // resources that the translators needs as inputs.
 type Resources struct {
@@ -83,11 +86,11 @@ type Translator struct {
 type TranslateResult struct {
 	Gateways   []*v1beta1.Gateway
 	HTTPRoutes []*v1beta1.HTTPRoute
-	XdsIR      *ir.Xds
-	InfraIR    *ir.Infra
+	XdsIR      XdsIRMap
+	InfraIR    InfraIRMap
 }
 
-func newTranslateResult(gateways []*GatewayContext, httpRoutes []*HTTPRouteContext, xdsIR *ir.Xds, infraIR *ir.Infra) *TranslateResult {
+func newTranslateResult(gateways []*GatewayContext, httpRoutes []*HTTPRouteContext, xdsIR XdsIRMap, infraIR InfraIRMap) *TranslateResult {
 	translateResult := &TranslateResult{
 		XdsIR:   xdsIR,
 		InfraIR: infraIR,
@@ -104,11 +107,8 @@ func newTranslateResult(gateways []*GatewayContext, httpRoutes []*HTTPRouteConte
 }
 
 func (t *Translator) Translate(resources *Resources) *TranslateResult {
-	xdsIR := &ir.Xds{}
-
-	infraIR := ir.NewInfra()
-	infraIR.Proxy.Name = string(t.GatewayClassName)
-	infraIR.Proxy.GetProxyMetadata().Labels = GatewayClassOwnerLabel(string(t.GatewayClassName))
+	xdsIR := make(XdsIRMap)
+	infraIR := make(InfraIRMap)
 
 	// Get Gateways belonging to our GatewayClass.
 	gateways := t.GetRelevantGateways(resources.Gateways)
@@ -129,6 +129,7 @@ func (t *Translator) GetRelevantGateways(gateways []*v1beta1.Gateway) []*Gateway
 		if gateway == nil {
 			panic("received nil gateway")
 		}
+
 		if gateway.Spec.GatewayClassName == t.GatewayClassName {
 			gc := &GatewayContext{
 				Gateway: gateway.DeepCopy(),
@@ -151,12 +152,13 @@ type portListeners struct {
 	hostnames map[string]int
 }
 
-func (t *Translator) ProcessListeners(gateways []*GatewayContext, xdsIR *ir.Xds, infraIR *ir.Infra, resources *Resources) {
+func (t *Translator) ProcessListeners(gateways []*GatewayContext, xdsIR XdsIRMap, infraIR InfraIRMap, resources *Resources) {
 	portListenerInfo := map[v1beta1.PortNumber]*portListeners{}
 
 	// Iterate through all listeners and collect info about protocols
 	// and hostnames per port.
 	for _, gateway := range gateways {
+
 		for _, listener := range gateway.listeners {
 			if portListenerInfo[listener.Port] == nil {
 				portListenerInfo[listener.Port] = &portListeners{
@@ -221,6 +223,16 @@ func (t *Translator) ProcessListeners(gateways []*GatewayContext, xdsIR *ir.Xds,
 	// and compute status for each, and add valid ones
 	// to the Xds IR.
 	for _, gateway := range gateways {
+		// init IR per gateway
+		irKey := irStringKey(gateway.Gateway)
+		gwXdsIR := &ir.Xds{}
+		gwInfraIR := ir.NewInfra()
+		gwInfraIR.Proxy.Name = irKey
+		gwInfraIR.Proxy.GetProxyMetadata().Labels = GatewayOwnerLabel(gateway.Name)
+		// save the IR references in the map before the translation starts
+		xdsIR[irKey] = gwXdsIR
+		infraIR[irKey] = gwInfraIR
+
 		for _, listener := range gateway.listeners {
 			// Process protocol & supported kinds
 			switch listener.Protocol {
@@ -453,7 +465,7 @@ func (t *Translator) ProcessListeners(gateways []*GatewayContext, xdsIR *ir.Xds,
 				// see more https://gateway-api.sigs.k8s.io/references/spec/#gateway.networking.k8s.io/v1beta1.Listener.
 				irListener.Hostnames = append(irListener.Hostnames, "*")
 			}
-			xdsIR.HTTP = append(xdsIR.HTTP, irListener)
+			gwXdsIR.HTTP = append(gwXdsIR.HTTP, irListener)
 
 			// Add the listener to the Infra IR. Infra IR ports must have a unique port number.
 			if !slices.Contains(foundPorts, servicePort) {
@@ -469,7 +481,7 @@ func (t *Translator) ProcessListeners(gateways []*GatewayContext, xdsIR *ir.Xds,
 					ContainerPort: containerPort,
 				}
 				// Only 1 listener is supported.
-				infraIR.Proxy.Listeners[0].Ports = append(infraIR.Proxy.Listeners[0].Ports, infraPort)
+				gwInfraIR.Proxy.Listeners[0].Ports = append(gwInfraIR.Proxy.Listeners[0].Ports, infraPort)
 			}
 		}
 	}
@@ -488,7 +500,7 @@ func servicePortToContainerPort(servicePort int32) int32 {
 	return servicePort
 }
 
-func (t *Translator) ProcessHTTPRoutes(httpRoutes []*v1beta1.HTTPRoute, gateways []*GatewayContext, resources *Resources, xdsIR *ir.Xds) []*HTTPRouteContext {
+func (t *Translator) ProcessHTTPRoutes(httpRoutes []*v1beta1.HTTPRoute, gateways []*GatewayContext, resources *Resources, xdsIR XdsIRMap) []*HTTPRouteContext {
 	var relevantHTTPRoutes []*HTTPRouteContext
 
 	for _, h := range httpRoutes {
@@ -1022,8 +1034,8 @@ func (t *Translator) ProcessHTTPRoutes(httpRoutes []*v1beta1.HTTPRoute, gateways
 						})
 					}
 				}
-
-				irListener := xdsIR.GetListener(irListenerName(listener))
+				irKey := irStringKey(listener.gateway)
+				irListener := xdsIR[irKey].GetListener(irListenerName(listener))
 				irListener.Routes = append(irListener.Routes, perHostRoutes...)
 
 				// Theoretically there should only be one parent ref per
@@ -1139,6 +1151,10 @@ func isValidHostname(hostname string) error {
 	return nil
 }
 
+func irStringKey(gateway *v1beta1.Gateway) string {
+	return fmt.Sprintf("%s-%s", gateway.Namespace, gateway.Name)
+}
+
 func irListenerName(listener *ListenerContext) string {
 	return fmt.Sprintf("%s-%s-%s", listener.gateway.Namespace, listener.gateway.Name, listener.Name)
 }
@@ -1162,8 +1178,8 @@ func irTLSConfig(tlsSecret *v1.Secret) *ir.TLSListenerConfig {
 	}
 }
 
-// GatewayClassOwnerLabel returns the GatewayCLass Owner label using
+// GatewayOwnerLabel returns the Gateway Owner label using
 // the provided name as the value.
-func GatewayClassOwnerLabel(name string) map[string]string {
-	return map[string]string{OwningGatewayClassLabel: name}
+func GatewayOwnerLabel(name string) map[string]string {
+	return map[string]string{OwningGatewayLabel: name}
 }

internal/infrastructure/kubernetes/bootstrap.yaml.tpl

@@ -25,7 +25,7 @@ dynamic_resources:
       set_node_on_first_message_only: true
 node:
   cluster: envoy-gateway-system
-  id: envoy-default
+  id: {{ .NodeId }}
 static_resources:
   clusters:
   - connect_timeout: 1s

internal/infrastructure/kubernetes/deployment.go

@@ -64,6 +64,8 @@ type bootstrapParameters struct {
 	XdsServer xdsServerParameters
 	// AdminServer defines the configuration of the Envoy admin interface.
 	AdminServer adminServerParameters
+	// NodeId defined the node ID of the Envoy instance.
+	NodeId string
 }
 
 type xdsServerParameters struct {
@@ -102,7 +104,7 @@ func (i *Infra) expectedDeployment(infra *ir.Infra) (*appsv1.Deployment, error)
 
 	// Set the labels based on the owning gatewayclass name.
 	labels := envoyLabels(infra.GetProxyInfra().GetProxyMetadata().Labels)
-	if _, ok := labels[gatewayapi.OwningGatewayClassLabel]; !ok {
+	if _, ok := labels[gatewayapi.OwningGatewayLabel]; !ok {
 		return nil, fmt.Errorf("missing owning gatewayclass label")
 	}
 
@@ -164,6 +166,7 @@ func expectedContainers(infra *ir.Infra) ([]corev1.Container, error) {
 				Port:          envoyAdminPort,
 				AccessLogPath: envoyAdminAccessLogPath,
 			},
+			NodeId: infra.Proxy.Name,
 		},
 	}
 	if err := cfg.render(); err != nil {

internal/infrastructure/kubernetes/service.go

@@ -34,7 +34,7 @@ func (i *Infra) expectedService(infra *ir.Infra) (*corev1.Service, error) {
 
 	// Set the labels based on the owning gatewayclass name.
 	labels := envoyLabels(infra.GetProxyInfra().GetProxyMetadata().Labels)
-	if _, ok := labels[gatewayapi.OwningGatewayClassLabel]; !ok {
+	if _, ok := labels[gatewayapi.OwningGatewayLabel]; !ok {
 		return nil, fmt.Errorf("missing owning gatewayclass label")
 	}
 

internal/infrastructure/runner/runner.go

@@ -43,23 +43,23 @@ func (r *Runner) subscribeAndTranslate(ctx context.Context) {
 	// Subscribe to resources
 	for range r.InfraIR.Subscribe(ctx) {
 		r.Logger.Info("received a notification")
-		in := r.InfraIR.Get()
-		switch {
-		case in == nil:
-			// The resource map is nil at startup.
-			r.Logger.Info("infra ir is nil, skipping")
-			continue
-		case in.Proxy == nil:
-			if err := r.mgr.DeleteInfra(ctx, in); err != nil {
-				r.Logger.Error(err, "failed to delete infra")
-			}
-		default:
-			// Manage the proxy infra.
-			if err := r.mgr.CreateInfra(ctx, in); err != nil {
-				r.Logger.Error(err, "failed to create new infra")
+		for _, in := range r.InfraIR.LoadAll() {
+			switch {
+			case in == nil:
+				// The resource map is nil at startup.
+				r.Logger.Info("infra ir is nil, skipping")
+				continue
+			case in.Proxy == nil:
+				if err := r.mgr.DeleteInfra(ctx, in); err != nil {
+					r.Logger.Error(err, "failed to delete infra")
+				}
+			default:
+				// Manage the proxy infra.
+				if err := r.mgr.CreateInfra(ctx, in); err != nil {
+					r.Logger.Error(err, "failed to create new infra")
+				}
 			}
 		}
+		r.Logger.Info("subscriber shutting down")
 	}
-
-	r.Logger.Info("subscriber shutting down")
 }

internal/message/types.go

@@ -93,36 +93,12 @@ type XdsIR struct {
 	watchable.Map[string, *ir.Xds]
 }
 
-func (x *XdsIR) Get() *ir.Xds {
-	// Iterate and return the first element
-	for _, v := range x.LoadAll() {
-		return v
-	}
-	return nil
-}
-
 // InfraIR message
 type InfraIR struct {
 	watchable.Map[string, *ir.Infra]
 }
 
-func (i *InfraIR) Get() *ir.Infra {
-	// Iterate and return the first element
-	for _, v := range i.LoadAll() {
-		return v
-	}
-	return nil
-}
-
 // Xds message
 type Xds struct {
 	watchable.Map[string, *xdstypes.ResourceVersionTable]
 }
-
-func (x *Xds) Get() *xdstypes.ResourceVersionTable {
-	// Iterate and return the first element
-	for _, v := range x.LoadAll() {
-		return v
-	}
-	return nil
-}

internal/message/types_test.go

@@ -11,12 +11,6 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	gwapiv1b1 "sigs.k8s.io/gateway-api/apis/v1beta1"
-
-	"github.com/envoyproxy/gateway/internal/ir"
-	xdstypes "github.com/envoyproxy/gateway/internal/xds/types"
-	listenerv3 "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3"
-	xdscachetypes "github.com/envoyproxy/go-control-plane/pkg/cache/types"
-	resourcev3 "github.com/envoyproxy/go-control-plane/pkg/resource/v3"
 )
 
 func TestProviderResources(t *testing.T) {
@@ -166,36 +160,3 @@ func TestProviderResources(t *testing.T) {
 	resources.DeleteGatewayClasses()
 	assert.Nil(t, resources.GetGatewayClasses())
 }
-
-func TestXdsIR(t *testing.T) {
-	msg := new(XdsIR)
-	assert.Nil(t, msg.Get())
-	in := &ir.Xds{
-		HTTP: []*ir.HTTPListener{{Name: "test"}},
-	}
-	msg.Store("xds-ir", in)
-	assert.Equal(t, msg.Get(), in)
-}
-
-func TestInfraIR(t *testing.T) {
-	msg := new(InfraIR)
-	assert.Nil(t, msg.Get())
-	in := &ir.Infra{
-		Proxy: &ir.ProxyInfra{Name: "test"},
-	}
-	msg.Store("infra-ir", in)
-	assert.Equal(t, msg.Get(), in)
-}
-
-func TestXds(t *testing.T) {
-	msg := new(Xds)
-	assert.Nil(t, msg.Get())
-	in := &xdstypes.ResourceVersionTable{
-		XdsResources: xdstypes.XdsResources{
-			resourcev3.ListenerType: []xdscachetypes.Resource{&listenerv3.Listener{Name: "test"}},
-		},
-	}
-	msg.Store("xds", in)
-	diff := cmp.Diff(in, msg.Get(), protocmp.Transform())
-	require.Empty(t, diff)
-}

internal/provider/kubernetes/gateway.go

@@ -113,7 +113,7 @@ func (r *gatewayReconciler) hasMatchingController(obj client.Object) bool {
 func (r *gatewayReconciler) enqueueRequestForOwningGatewayClass() handler.EventHandler {
 	return handler.EnqueueRequestsFromMapFunc(func(a client.Object) []reconcile.Request {
 		labels := a.GetLabels()
-		gcName, found := labels[gatewayapi.OwningGatewayClassLabel]
+		gcName, found := labels[gatewayapi.OwningGatewayLabel]
 		if found {
 			var reqs []reconcile.Request
 			for _, gw := range r.resources.Gateways.LoadAll() {