kubernetes: use AWS physical zone IDs for locality

Signed-off-by: Martin Baillie <martin@baillie.id>

Author: martinbaillie

internal/provider/kubernetes/controller.go

@@ -626,6 +626,8 @@ func (r *gatewayAPIReconciler) Reconcile(ctx context.Context, _ reconcile.Reques
 		}
 	}
 
+	r.applyAWSZoneIDs(gwcResources)
+
 	// Sort before storing to:
 	// 1. ensure identical resources are not retranslated
 	//    and updates are avoided by the watchable layer
@@ -2709,10 +2711,10 @@ func (r *gatewayAPIReconciler) watchResources(ctx context.Context, mgr manager.M
 		}
 	}
 
-	// Watch Node CRUDs to update Gateway Address exposed by Service of type NodePort.
+	// Watch Nodes for locality changes and Gateway addresses exposed by NodePort Services.
 	// Node creation/deletion and ExternalIP updates would require update in the Gateway
 	nPredicates := []predicate.TypedPredicate[*corev1.Node]{
-		predicate.TypedGenerationChangedPredicate[*corev1.Node]{},
+		r.nodePredicate(),
 		predicate.NewTypedPredicateFuncs(func(node *corev1.Node) bool {
 			return r.handleNode(node)
 		}),

internal/provider/kubernetes/locality.go

@@ -0,0 +1,69 @@
+// Copyright Envoy Gateway Authors
+// SPDX-License-Identifier: Apache-2.0
+// The full text of the Apache license is available in the LICENSE file at
+// the root of the repo.
+
+package kubernetes
+
+import (
+	corev1 "k8s.io/api/core/v1"
+	discoveryv1 "k8s.io/api/discovery/v1"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
+
+	"github.com/envoyproxy/gateway/internal/gatewayapi/resource"
+)
+
+const (
+	awsZoneIDLabel                       = "topology.k8s.aws/zone-id"
+	endpointSliceControllerName          = "endpointslice-controller.k8s.io"
+	endpointSliceMirroringControllerName = "endpointslicemirroring-controller.k8s.io"
+)
+
+// applyAWSZoneIDs resolves local endpoints without changing cached EndpointSlices.
+func (r *gatewayAPIReconciler) applyAWSZoneIDs(resources resource.ControllerResources) {
+	zoneIDs := make(map[string]string)
+	for _, res := range resources {
+		for i, slice := range res.EndpointSlices {
+			// Slices from other controllers can refer to nodes in another cluster.
+			managedBy := slice.Labels[discoveryv1.LabelManagedBy]
+			if managedBy != endpointSliceControllerName &&
+				managedBy != endpointSliceMirroringControllerName {
+				continue
+			}
+			var updated *discoveryv1.EndpointSlice
+			for j, endpoint := range slice.Endpoints {
+				if endpoint.NodeName == nil || *endpoint.NodeName == "" {
+					continue
+				}
+				zoneID, found := zoneIDs[*endpoint.NodeName]
+				if !found {
+					zoneID = r.store.nodeZoneID(*endpoint.NodeName)
+					zoneIDs[*endpoint.NodeName] = zoneID
+				}
+				if zoneID == "" || (endpoint.Zone != nil && *endpoint.Zone == zoneID) {
+					continue
+				}
+				if updated == nil {
+					updated = slice.DeepCopy()
+				}
+				updated.Endpoints[j].Zone = new(zoneID)
+			}
+			if updated != nil {
+				res.EndpointSlices[i] = updated
+			}
+		}
+	}
+}
+
+func (r *gatewayAPIReconciler) nodePredicate() predicate.TypedPredicate[*corev1.Node] {
+	return predicate.Or(
+		predicate.TypedGenerationChangedPredicate[*corev1.Node]{},
+		predicate.TypedFuncs[*corev1.Node]{
+			UpdateFunc: func(e event.TypedUpdateEvent[*corev1.Node]) bool {
+				return e.ObjectOld != nil && e.ObjectNew != nil &&
+					e.ObjectOld.Labels[awsZoneIDLabel] != e.ObjectNew.Labels[awsZoneIDLabel]
+			},
+		},
+	)
+}

internal/provider/kubernetes/locality_test.go

@@ -0,0 +1,167 @@
+// Copyright Envoy Gateway Authors
+// SPDX-License-Identifier: Apache-2.0
+// The full text of the Apache license is available in the LICENSE file at
+// the root of the repo.
+
+package kubernetes
+
+import (
+	"fmt"
+	"os"
+	"testing"
+
+	endpointv3 "github.com/envoyproxy/go-control-plane/envoy/config/endpoint/v3"
+	resourcev3 "github.com/envoyproxy/go-control-plane/pkg/resource/v3"
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	discoveryv1 "k8s.io/api/discovery/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+	mcsapiv1a1 "sigs.k8s.io/mcs-api/pkg/apis/v1alpha1"
+
+	egv1a1 "github.com/envoyproxy/gateway/api/v1alpha1"
+	"github.com/envoyproxy/gateway/internal/envoygateway/config"
+	"github.com/envoyproxy/gateway/internal/gatewayapi"
+	"github.com/envoyproxy/gateway/internal/gatewayapi/resource"
+	"github.com/envoyproxy/gateway/internal/infrastructure/kubernetes/proxy"
+	"github.com/envoyproxy/gateway/internal/message"
+	"github.com/envoyproxy/gateway/internal/provider/kubernetes/test"
+	xdstranslator "github.com/envoyproxy/gateway/internal/xds/translator"
+)
+
+func TestApplyAWSZoneIDs(t *testing.T) {
+	local := &discoveryv1.EndpointSlice{
+		ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{discoveryv1.LabelManagedBy: endpointSliceControllerName}},
+		Endpoints: []discoveryv1.Endpoint{
+			{NodeName: new("node"), Zone: new("us-east-1d")},
+			{NodeName: new("node")},
+			{NodeName: new("no-id"), Zone: new("us-east-1b")},
+			{NodeName: new("empty-id"), Zone: new("us-east-1b")},
+			{NodeName: new("missing"), Zone: new("us-east-1c")},
+			{Zone: new("explicit")},
+		},
+	}
+	local.Labels[mcsapiv1a1.LabelServiceName] = "also-exported"
+	imported := local.DeepCopy()
+	imported.Labels[mcsapiv1a1.LabelServiceName] = "remote"
+	imported.Labels[discoveryv1.LabelManagedBy] = "mcs-controller"
+	custom := local.DeepCopy()
+	custom.Labels[discoveryv1.LabelManagedBy] = "custom-controller"
+	mirrored := local.DeepCopy()
+	mirrored.Labels[discoveryv1.LabelManagedBy] = endpointSliceMirroringControllerName
+	original := local.DeepCopy()
+	resources := resource.ControllerResources{
+		{EndpointSlices: []*discoveryv1.EndpointSlice{local, imported, custom, mirrored}},
+		{EndpointSlices: []*discoveryv1.EndpointSlice{local}},
+	}
+	store := newProviderStore()
+	for _, node := range []*corev1.Node{
+		{ObjectMeta: metav1.ObjectMeta{Name: "node", Labels: map[string]string{awsZoneIDLabel: "use1-az6"}}},
+		{ObjectMeta: metav1.ObjectMeta{Name: "no-id"}},
+		{ObjectMeta: metav1.ObjectMeta{Name: "empty-id", Labels: map[string]string{awsZoneIDLabel: ""}}},
+	} {
+		store.addNode(node)
+	}
+	r := &gatewayAPIReconciler{store: store}
+	r.applyAWSZoneIDs(resources)
+	for _, res := range resources {
+		require.Equal(t, "use1-az6", *res.EndpointSlices[0].Endpoints[0].Zone)
+		require.Equal(t, "use1-az6", *res.EndpointSlices[0].Endpoints[1].Zone)
+		require.Equal(t, original.Endpoints[2:], res.EndpointSlices[0].Endpoints[2:])
+	}
+	require.Equal(t, original, local)
+	require.Equal(t, "use1-az6", *resources[0].EndpointSlices[3].Endpoints[0].Zone)
+	require.Same(t, imported, resources[0].EndpointSlices[1])
+	require.Same(t, custom, resources[0].EndpointSlices[2])
+	require.Equal(t, original.Endpoints, imported.Endpoints)
+	require.Equal(t, original.Endpoints, custom.Endpoints)
+}
+
+func TestAWSZoneIDNodeUpdates(t *testing.T) {
+	r := &gatewayAPIReconciler{}
+	old := &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: "node"}}
+	updated := old.DeepCopy()
+	updated.Labels = map[string]string{awsZoneIDLabel: "use1-az6"}
+	for _, nodes := range [][2]*corev1.Node{{old, updated}, {updated, old}} {
+		require.True(t, r.nodePredicate().Update(event.TypedUpdateEvent[*corev1.Node]{ObjectOld: nodes[0], ObjectNew: nodes[1]}))
+	}
+	require.False(t, r.nodePredicate().Update(event.TypedUpdateEvent[*corev1.Node]{ObjectOld: updated, ObjectNew: updated}))
+	updated.Generation++
+	require.True(t, r.nodePredicate().Update(event.TypedUpdateEvent[*corev1.Node]{ObjectOld: old, ObjectNew: updated}))
+}
+
+func TestAWSZoneIDLocalityXDS(t *testing.T) {
+	cfg, err := config.New(os.Stdout, os.Stderr)
+	require.NoError(t, err)
+	cfg.EnvoyGateway.Provider = &egv1a1.EnvoyGatewayProvider{Type: egv1a1.ProviderTypeCustom}
+	updates := new(message.ProviderResources)
+	r, err := NewOfflineGatewayAPIController(t.Context(), cfg, nil, updates)
+	require.NoError(t, err)
+	defer updates.Close()
+	gateway := test.GetGateway(types.NamespacedName{Namespace: "default", Name: "gateway"}, "eg", 80)
+	fleet := test.GetService(types.NamespacedName{
+		Namespace: cfg.ControllerNamespace, Name: proxy.ExpectedResourceHashedName("default/gateway"),
+	}, gatewayapi.OwnerLabels(gateway, false), map[string]int32{"dummy": 8080})
+	backend := test.GetService(types.NamespacedName{Namespace: "default", Name: "backend"}, nil, map[string]int32{"dummy": 8080})
+	backend.Spec.TrafficDistribution = new("PreferClose")
+	fleetSlice := test.GetEndpointSlice(types.NamespacedName{Namespace: fleet.Namespace, Name: "fleet"}, fleet.Name, false)
+	backendSlice := test.GetEndpointSlice(types.NamespacedName{Namespace: backend.Namespace, Name: "backend"}, backend.Name, false)
+	for i, slice := range []*discoveryv1.EndpointSlice{fleetSlice, backendSlice} {
+		slice.AddressType = discoveryv1.AddressTypeIPv4
+		slice.Labels[discoveryv1.LabelManagedBy] = endpointSliceControllerName
+		slice.Endpoints[0].NodeName = new("node")
+		slice.Endpoints[0].Zone = new("us-east-1d")
+		slice.Endpoints[0].Addresses = []string{fmt.Sprintf("192.0.2.%d", i+1)}
+	}
+	remoteSlice := backendSlice.DeepCopy()
+	remoteSlice.Name = "remote"
+	remoteSlice.Labels[discoveryv1.LabelManagedBy] = "remote-controller"
+	remoteSlice.Endpoints[0].Zone = new("use1-az2")
+	remoteSlice.Endpoints[0].Addresses = []string{"192.0.2.3"}
+	node := &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: "node", Labels: map[string]string{
+		corev1.LabelTopologyZone: "us-east-1d", awsZoneIDLabel: "use1-az6",
+	}}}
+	for _, obj := range []client.Object{
+		&corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "default"}},
+		&corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: cfg.ControllerNamespace}},
+		test.GetGatewayClass("eg", egv1a1.GatewayControllerName, nil), gateway, fleet, backend,
+		fleetSlice, backendSlice, remoteSlice, node,
+		test.GetHTTPRoute(types.NamespacedName{Namespace: "default", Name: "route"}, gateway.Name,
+			test.GetServiceBackendRef(types.NamespacedName{Name: backend.Name}, 8080), ""),
+	} {
+		require.NoError(t, r.Client.Create(t.Context(), obj))
+	}
+	r.store.addNode(node)
+	require.NoError(t, r.Reconcile(t.Context()))
+	published, ok := updates.GatewayAPIResources.Load(cfg.EnvoyGateway.Gateway.ControllerName)
+	require.True(t, ok)
+	require.Len(t, *published.Resources, 1)
+	gt := &gatewayapi.Translator{
+		GatewayControllerName: egv1a1.GatewayControllerName, GatewayClassName: "eg",
+		ControllerNamespace: cfg.ControllerNamespace, Logger: cfg.Logger,
+	}
+	translated, err := gt.Translate(t.Context(), (*published.Resources)[0])
+	require.NoError(t, err)
+	require.Len(t, translated.XdsIR, 1)
+	zones := map[string]string{}
+	for _, xdsIR := range translated.XdsIR {
+		xt := &xdstranslator.Translator{Logger: cfg.Logger}
+		table, err := xt.Translate(t.Context(), xdsIR)
+		require.NoError(t, err)
+		for _, entry := range table.XdsResources[resourcev3.EndpointType] {
+			for _, locality := range entry.(*endpointv3.ClusterLoadAssignment).Endpoints {
+				for _, endpoint := range locality.LbEndpoints {
+					zones[endpoint.GetEndpoint().Address.GetSocketAddress().Address] = locality.Locality.GetZone()
+				}
+			}
+		}
+	}
+	require.Equal(t, map[string]string{
+		"192.0.2.1": "use1-az6", "192.0.2.2": "use1-az6", "192.0.2.3": "use1-az2",
+	}, zones)
+	var stored discoveryv1.EndpointSlice
+	require.NoError(t, r.Client.Get(t.Context(), client.ObjectKeyFromObject(backendSlice), &stored))
+	require.Equal(t, "us-east-1d", *stored.Endpoints[0].Zone)
+}

internal/provider/kubernetes/store.go

@@ -17,6 +17,7 @@ import (
 type nodeDetails struct {
 	name      string
 	addresses status.NodeAddresses
+	zoneID    string
 }
 
 // kubernetesProviderStore holds cached information for the kubernetes provider.
@@ -35,7 +36,7 @@ func newProviderStore() *kubernetesProviderStore {
 }
 
 func (p *kubernetesProviderStore) addNode(n *corev1.Node) {
-	details := nodeDetails{name: n.Name}
+	details := nodeDetails{name: n.Name, zoneID: n.Labels[awsZoneIDLabel]}
 
 	var internalIPs, externalIPs status.NodeAddresses
 	for _, addr := range n.Status.Addresses {
@@ -74,6 +75,12 @@ func (p *kubernetesProviderStore) removeNode(n *corev1.Node) {
 	delete(p.nodes, n.Name)
 }
 
+func (p *kubernetesProviderStore) nodeZoneID(name string) string {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	return p.nodes[name].zoneID
+}
+
 func (p *kubernetesProviderStore) listNodeAddresses() status.NodeAddresses {
 	p.mu.Lock()
 	defer p.mu.Unlock()

internal/provider/kubernetes/topology_injector.go

@@ -83,7 +83,12 @@ func (m *ProxyTopologyInjector) Handle(ctx context.Context, req admission.Reques
 	}
 	logger = logger.WithValues("node", node)
 
-	if zone, ok := node.Labels[corev1.LabelTopologyZone]; ok {
+	zone, hasZone := node.Labels[corev1.LabelTopologyZone]
+	if node.Labels[awsZoneIDLabel] != "" {
+		zone = node.Labels[awsZoneIDLabel]
+		hasZone = true
+	}
+	if hasZone {
 		if binding.Annotations == nil {
 			binding.Annotations = map[string]string{}
 		}

internal/provider/kubernetes/topology_injector_test.go

@@ -45,6 +45,17 @@ func TestProxyTopologyInjector_Handle(t *testing.T) {
 		},
 	}
 
+	awsNode := defaultNode.DeepCopy()
+	awsNode.Labels = map[string]string{
+		corev1.LabelTopologyZone:   "us-east-1d",
+		"topology.k8s.aws/zone-id": "use1-az6",
+	}
+
+	awsOnlyNode := awsNode.DeepCopy()
+	delete(awsOnlyNode.Labels, corev1.LabelTopologyZone)
+	emptyAWSNode := awsNode.DeepCopy()
+	emptyAWSNode.Labels[awsZoneIDLabel] = ""
+
 	cases := []struct {
 		caseName          string
 		obj               client.Object
@@ -71,6 +82,48 @@ func TestProxyTopologyInjector_Handle(t *testing.T) {
 				},
 			}},
 		},
+		{
+			caseName: "AWS zone ID",
+			obj: &corev1.Binding{
+				ObjectMeta: metav1.ObjectMeta{Name: defaultPod.Name, Namespace: defaultPod.Namespace},
+				Target:     corev1.ObjectReference{Name: awsNode.Name},
+			},
+			node: awsNode,
+			pod:  defaultPod,
+			expectedPatchResp: []jsonpatch.JsonPatchOperation{{
+				Operation: "add",
+				Path:      "/metadata/annotations",
+				Value:     map[string]interface{}{corev1.LabelTopologyZone: "\"use1-az6\""},
+			}},
+		},
+		{
+			caseName: "AWS zone ID without standard label",
+			obj: &corev1.Binding{
+				ObjectMeta: metav1.ObjectMeta{Name: defaultPod.Name, Namespace: defaultPod.Namespace},
+				Target:     corev1.ObjectReference{Name: awsNode.Name},
+			},
+			node: awsOnlyNode,
+			pod:  defaultPod,
+			expectedPatchResp: []jsonpatch.JsonPatchOperation{{
+				Operation: "add",
+				Path:      "/metadata/annotations",
+				Value:     map[string]interface{}{corev1.LabelTopologyZone: "\"use1-az6\""},
+			}},
+		},
+		{
+			caseName: "empty AWS zone ID",
+			obj: &corev1.Binding{
+				ObjectMeta: metav1.ObjectMeta{Name: defaultPod.Name, Namespace: defaultPod.Namespace},
+				Target:     corev1.ObjectReference{Name: awsNode.Name},
+			},
+			node: emptyAWSNode,
+			pod:  defaultPod,
+			expectedPatchResp: []jsonpatch.JsonPatchOperation{{
+				Operation: "add",
+				Path:      "/metadata/annotations",
+				Value:     map[string]interface{}{corev1.LabelTopologyZone: "\"us-east-1d\""},
+			}},
+		},
 		{
 			caseName: "empty target",
 			obj: &corev1.Binding{

release-notes/current/breaking_changes/0-aws-zone-id-locality.md

@@ -0,0 +1,3 @@
+Use physical AWS zone IDs for proxy and local Kubernetes-managed endpoint locality when the node
+label is available. Update Backend zones, `weightedZones`, and imported or custom EndpointSlices to
+use IDs.