feat(kuma-cp): add locality aware implementation for egress

app/kumactl/cmd/install/testdata/install-control-plane.defaults.golden.yaml

@@ -2086,7 +2086,6 @@ spec:
                                             type: array
                                         required:
                                         - type
-                                        - zones
                                         type: object
                                     required:
                                     - to

app/kumactl/cmd/install/testdata/install-control-plane.gateway-api-present-not-enabled.yaml

@@ -2086,7 +2086,6 @@ spec:
                                             type: array
                                         required:
                                         - type
-                                        - zones
                                         type: object
                                     required:
                                     - to

app/kumactl/cmd/install/testdata/install-control-plane.gateway-api-present.yaml

@@ -2290,7 +2290,6 @@ spec:
                                             type: array
                                         required:
                                         - type
-                                        - zones
                                         type: object
                                     required:
                                     - to

app/kumactl/cmd/install/testdata/install-control-plane.with-helm-set.yaml

@@ -2106,7 +2106,6 @@ spec:
                                             type: array
                                         required:
                                         - type
-                                        - zones
                                         type: object
                                     required:
                                     - to

app/kumactl/cmd/install/testdata/install-crds.all.golden.yaml

@@ -3391,7 +3391,6 @@ spec:
                                             type: array
                                         required:
                                         - type
-                                        - zones
                                         type: object
                                     required:
                                     - to

app/kumactl/cmd/install/testdata/install-crds.experimental-gatewayapi.golden.yaml

@@ -3595,7 +3595,6 @@ spec:
                                             type: array
                                         required:
                                         - type
-                                        - zones
                                         type: object
                                     required:
                                     - to

deployments/charts/kuma/crds/kuma.io_meshloadbalancingstrategies.yaml

@@ -414,7 +414,6 @@ spec:
                                             type: array
                                         required:
                                         - type
-                                        - zones
                                         type: object
                                     required:
                                     - to

docs/generated/openapi.yaml

@@ -4052,7 +4052,6 @@ components:
                                           type: array
                                       required:
                                         - type
-                                        - zones
                                       type: object
                                   required:
                                     - to

docs/generated/raw/crds/kuma.io_meshloadbalancingstrategies.yaml

@@ -414,7 +414,6 @@ spec:
                                             type: array
                                         required:
                                         - type
-                                        - zones
                                         type: object
                                     required:
                                     - to

pkg/core/plugins/interfaces.go

@@ -102,7 +102,7 @@ type EgressPolicyPlugin interface {
 	PolicyPlugin
 	// EgressMatchedPolicies returns all the policies of the plugins' type matching the external service that
 	// should be applied on the zone egress.
-	EgressMatchedPolicies(*core_mesh.ExternalServiceResource, xds_context.Resources) (core_xds.TypedMatchingPolicies, error)
+	EgressMatchedPolicies(tags map[string]string, resources xds_context.Resources) (core_xds.TypedMatchingPolicies, error)
 }
 
 // ProxyPlugin a plugin to modify the proxy. This happens before any `PolicyPlugin` or any envoy generation. and it is applied both for Dataplanes and ZoneProxies

pkg/plugins/policies/core/matchers/egress.go

@@ -7,14 +7,13 @@ import (
 
 	common_api "github.com/kumahq/kuma/api/common/v1alpha1"
 	mesh_proto "github.com/kumahq/kuma/api/mesh/v1alpha1"
-	core_mesh "github.com/kumahq/kuma/pkg/core/resources/apis/mesh"
 	core_model "github.com/kumahq/kuma/pkg/core/resources/model"
 	core_xds "github.com/kumahq/kuma/pkg/core/xds"
 	core_rules "github.com/kumahq/kuma/pkg/plugins/policies/core/rules"
 	xds_context "github.com/kumahq/kuma/pkg/xds/context"
 )
 
-func EgressMatchedPolicies(rType core_model.ResourceType, es *core_mesh.ExternalServiceResource, resources xds_context.Resources) (core_xds.TypedMatchingPolicies, error) {
+func EgressMatchedPolicies(rType core_model.ResourceType, tags map[string]string, resources xds_context.Resources) (core_xds.TypedMatchingPolicies, error) {
 	policies := resources.ListOrEmpty(rType)
 
 	if len(policies.GetItems()) == 0 {
@@ -41,9 +40,9 @@ func EgressMatchedPolicies(rType core_model.ResourceType, es *core_mesh.External
 	var fr core_rules.FromRules
 	var err error
 	if isFrom {
-		fr, err = processFromRules(es, policies)
+		fr, err = processFromRules(tags, policies)
 	} else {
-		fr, err = processToRules(es, policies)
+		fr, err = processToRules(tags, policies)
 	}
 	if err != nil {
 		return core_xds.TypedMatchingPolicies{}, err
@@ -56,14 +55,14 @@ func EgressMatchedPolicies(rType core_model.ResourceType, es *core_mesh.External
 }
 
 func processFromRules(
-	es *core_mesh.ExternalServiceResource,
+	tags map[string]string,
 	rl core_model.ResourceList,
 ) (core_rules.FromRules, error) {
 	matchedPolicies := []core_model.Resource{}
 
 	for _, policy := range rl.GetItems() {
 		spec := policy.GetSpec().(core_model.Policy)
-		if !externalServiceSelectedByTargetRef(spec.GetTargetRef(), es) {
+		if !serviceSelectedByTargetRef(spec.GetTargetRef(), tags) {
 			continue
 		}
 		matchedPolicies = append(matchedPolicies, policy)
@@ -113,11 +112,8 @@ func processFromRules(
 //	        disabled: true
 //
 // that's why processToRules() method produces FromRules for the Egress.
-func processToRules(
-	es *core_mesh.ExternalServiceResource,
-	rl core_model.ResourceList,
-) (core_rules.FromRules, error) {
-	matchedPolicies := []core_model.Resource{}
+func processToRules(tags map[string]string, rl core_model.ResourceList) (core_rules.FromRules, error) {
+	var matchedPolicies []core_model.Resource
 
 	for _, policy := range rl.GetItems() {
 		spec := policy.GetSpec().(core_model.Policy)
@@ -128,18 +124,18 @@ func processToRules(
 		}
 
 		for _, item := range to.GetToList() {
-			if externalServiceSelectedByTargetRef(item.GetTargetRef(), es) {
+			if serviceSelectedByTargetRef(item.GetTargetRef(), tags) {
 				matchedPolicies = append(matchedPolicies, policy)
 			}
 		}
 	}
 
 	sort.Sort(ByTargetRef(matchedPolicies))
 
-	toList := []core_rules.PolicyItemWithMeta{}
+	var toList []core_rules.PolicyItemWithMeta
 	for _, policy := range matchedPolicies {
 		for _, item := range policy.GetSpec().(core_model.PolicyWithToList).GetToList() {
-			if !externalServiceSelectedByTargetRef(item.GetTargetRef(), es) {
+			if !serviceSelectedByTargetRef(item.GetTargetRef(), tags) {
 				continue
 			}
 			// convert 'to' policyItem to 'from' policyItem
@@ -175,17 +171,16 @@ func (a *artificialPolicyItem) GetDefault() interface{} {
 	return a.conf
 }
 
-func externalServiceSelectedByTargetRef(tr common_api.TargetRef, es *core_mesh.ExternalServiceResource) bool {
+func serviceSelectedByTargetRef(tr common_api.TargetRef, tags map[string]string) bool {
 	switch tr.Kind {
 	case common_api.Mesh:
 		return true
 	case common_api.MeshSubset:
-		return mesh_proto.TagSelector(tr.Tags).Matches(es.Spec.GetTags())
+		return mesh_proto.TagSelector(tr.Tags).Matches(tags)
 	case common_api.MeshService:
-		return tr.Name == es.Spec.GetTags()[mesh_proto.ServiceTag]
+		return tr.Name == tags[mesh_proto.ServiceTag]
 	case common_api.MeshServiceSubset:
-		return tr.Name == es.Spec.GetTags()[mesh_proto.ServiceTag] &&
-			mesh_proto.TagSelector(tr.Tags).Matches(es.Spec.GetTags())
+		return tr.Name == tags[mesh_proto.ServiceTag] && mesh_proto.TagSelector(tr.Tags).Matches(tags)
 	}
 	return false
 }

pkg/plugins/policies/core/matchers/egress_test.go

@@ -59,7 +59,7 @@ var _ = Describe("EgressMatchedPolicies", func() {
 			resources, _ := readPolicies(given.policiesFile)
 
 			// when
-			policies, err := matchers.EgressMatchedPolicies(policies_api.MeshTrafficPermissionType, es, resources)
+			policies, err := matchers.EgressMatchedPolicies(policies_api.MeshTrafficPermissionType, es.Spec.Tags, resources)
 			Expect(err).ToNot(HaveOccurred())
 
 			// then
@@ -76,7 +76,7 @@ var _ = Describe("EgressMatchedPolicies", func() {
 			resources, _ := readPolicies(given.policiesFile)
 
 			// when
-			policies, err := matchers.EgressMatchedPolicies(v1alpha1.MeshLoadBalancingStrategyType, es, resources)
+			policies, err := matchers.EgressMatchedPolicies(v1alpha1.MeshLoadBalancingStrategyType, es.Spec.Tags, resources)
 			Expect(err).ToNot(HaveOccurred())
 
 			// then

pkg/plugins/policies/core/xds/clusters.go

@@ -10,13 +10,15 @@ import (
 	"github.com/kumahq/kuma/pkg/plugins/runtime/gateway/metadata"
 	"github.com/kumahq/kuma/pkg/xds/envoy/tags"
 	"github.com/kumahq/kuma/pkg/xds/generator"
+	"github.com/kumahq/kuma/pkg/xds/generator/egress"
 )
 
 type Clusters struct {
 	Inbound       map[string]*envoy_cluster.Cluster
 	Outbound      map[string]*envoy_cluster.Cluster
 	OutboundSplit map[string][]*envoy_cluster.Cluster
 	Gateway       map[string]*envoy_cluster.Cluster
+	Egress        map[string]*envoy_cluster.Cluster
 }
 
 func GatherClusters(rs *core_xds.ResourceSet) Clusters {
@@ -25,6 +27,7 @@ func GatherClusters(rs *core_xds.ResourceSet) Clusters {
 		Outbound:      map[string]*envoy_cluster.Cluster{},
 		OutboundSplit: map[string][]*envoy_cluster.Cluster{},
 		Gateway:       map[string]*envoy_cluster.Cluster{},
+		Egress:        map[string]*envoy_cluster.Cluster{},
 	}
 	for _, res := range rs.Resources(envoy_resource.ClusterType) {
 		cluster := res.Resource.(*envoy_cluster.Cluster)
@@ -42,6 +45,8 @@ func GatherClusters(rs *core_xds.ResourceSet) Clusters {
 			clusters.Inbound[cluster.Name] = cluster
 		case metadata.OriginGateway:
 			clusters.Gateway[cluster.Name] = cluster
+		case egress.OriginEgress:
+			clusters.Egress[cluster.Name] = cluster
 		default:
 			continue
 		}

pkg/plugins/policies/core/xds/endpoints.go

@@ -38,16 +38,26 @@ func GatherEndpoints(rs *xds.ResourceSet) EndpointMap {
 	return em
 }
 
-func GatherEgressEndpoints(rs *xds.ResourceSet) map[string]*endpointv3.ClusterLoadAssignment {
-	em := map[string]*endpointv3.ClusterLoadAssignment{}
+func GatherEgressEndpoints(rs *xds.ResourceSet) EndpointMap {
+	em := EndpointMap{}
+	for _, res := range rs.Resources(envoy_resource.EndpointType) {
+		if res.Origin != egress.OriginEgress {
+			continue
+		}
+
+		cla := res.Resource.(*endpointv3.ClusterLoadAssignment)
+		serviceName := tags.ServiceFromClusterName(cla.ClusterName)
+		em[serviceName] = append(em[serviceName], cla)
+	}
+
 	for _, res := range rs.Resources(envoy_resource.ClusterType) {
 		if res.Origin != egress.OriginEgress {
 			continue
 		}
 
 		cluster := res.Resource.(*clusterv3.Cluster)
 		if cluster.LoadAssignment != nil {
-			em[cluster.Name] = cluster.LoadAssignment
+			em[cluster.Name] = append(em[cluster.Name], cluster.LoadAssignment)
 		}
 	}
 	return em

pkg/plugins/policies/meshloadbalancingstrategy/api/v1alpha1/meshloadbalancingstrategy.go

@@ -88,7 +88,7 @@ type FromZone struct {
 type ToZone struct {
 	// Type defines how target zones will be picked from available zones
 	Type  ToZoneType `json:"type"`
-	Zones []string   `json:"zones"`
+	Zones *[]string  `json:"zones,omitempty"`
 }
 
 type FailoverThreshold struct {

pkg/plugins/policies/meshloadbalancingstrategy/api/v1alpha1/schema.yaml

@@ -279,7 +279,6 @@ properties:
                                     type: array
                                 required:
                                   - type
-                                  - zones
                                 type: object
                             required:
                               - to

pkg/plugins/policies/meshloadbalancingstrategy/api/v1alpha1/validator.go

@@ -111,11 +111,11 @@ func validateCrossZone(crossZone *CrossZone) validators.ValidationError {
 		toZonesPath := path.Field("to").Field("zones")
 		switch failover.To.Type {
 		case Any, None:
-			if len(failover.To.Zones) > 0 {
+			if failover.To.Zones != nil && len(*failover.To.Zones) > 0 {
 				verr.AddViolationAt(toZonesPath, fmt.Sprintf("must be empty when type is %s", failover.To.Type))
 			}
 		case AnyExcept, Only:
-			if len(failover.To.Zones) == 0 {
+			if failover.To.Zones == nil || len(*failover.To.Zones) == 0 {
 				verr.AddViolationAt(toZonesPath, fmt.Sprintf("must not be empty when type is %s", failover.To.Type))
 			}
 		default:

pkg/plugins/policies/meshloadbalancingstrategy/api/v1alpha1/validator_test.go

@@ -324,6 +324,14 @@ to:
 			},
 			{
 				Field:   "spec.to[0].default.localityAwareness.crossZone.failover[4].to.zones",
+				Message: "must not be empty when type is Only",
+			},
+			{
+				Field:   "spec.to[0].default.localityAwareness.crossZone.failover[5].to.zones",
+				Message: "must not be empty when type is AnyExcept",
+			},
+			{
+				Field:   "spec.to[0].default.localityAwareness.crossZone.failover[6].to.zones",
 				Message: "must not be empty when type is AnyExcept",
 			},
 		}, `
@@ -355,9 +363,14 @@ to:
             - to:
                 type: Only
                 zones: []
+            - to:
+                type: Only
             - to:
                 type: AnyExcept
                 zones: []
+            - to:
+                type: AnyExcept
+
 `),
 	)
 

pkg/plugins/policies/meshloadbalancingstrategy/k8s/crd/kuma.io_meshloadbalancingstrategies.yaml

@@ -414,7 +414,6 @@ spec:
                                             type: array
                                         required:
                                         - type
-                                        - zones
                                         type: object
                                     required:
                                     - to