feat(kubernetes): gate zone-aware routing on ingress-controller zone balance

config/config.go

@@ -192,6 +192,8 @@ type Config struct {
 	KubernetesListChunkSize                              int                                `yaml:"kubernetes-list-chunk-size"`
 	KubernetesEnableEndpointSlices                       bool                               `yaml:"enable-kubernetes-endpointslices"`
 	KubernetesTopologyZone                               string                             `yaml:"kubernetes-topology-zone"`
+	KubernetesDataPlaneLabelSelector                     mapFlags                           `yaml:"kubernetes-data-plane-label-selector"`
+	KubernetesZoneMaxSkew                                int                                `yaml:"kubernetes-zone-max-skew"`
 	KubernetesEnableEastWest                             bool                               `yaml:"enable-kubernetes-east-west"`
 	KubernetesEastWestDomain                             string                             `yaml:"kubernetes-east-west-domain"`
 	KubernetesEastWestRangeDomains                       *listFlag                          `yaml:"kubernetes-east-west-range-domains"`
@@ -607,6 +609,8 @@ func NewConfig() *Config {
 	flag.IntVar(&cfg.KubernetesListChunkSize, "kubernetes-list-chunk-size", 0, "configures the chunk size (limit parameter) when listing Kubernetes resources. If 0 or negative, pagination is disabled")
 	flag.BoolVar(&cfg.KubernetesEnableEndpointSlices, "enable-kubernetes-endpointslices", false, "Enables that skipper fetches Kubernetes endpointslices instead of endpoints to scale more than 1000 pods within a service")
 	flag.StringVar(&cfg.KubernetesTopologyZone, "kubernetes-topology-zone", "", "sets the topology zone to be used for zone aware routing")
+	flag.Var(&cfg.KubernetesDataPlaneLabelSelector, "kubernetes-data-plane-label-selector", "comma separated key=value label selector used to find the Service(s) that make up the data plane for zone distribution checks, e.g. 'application=skipper-ingress,component=ingress'. Defaults to application=skipper-ingress,component=ingress when unset")
+	flag.IntVar(&cfg.KubernetesZoneMaxSkew, "kubernetes-zone-max-skew", kubernetes.DefaultMaxZoneSkew, "maximum permitted endpoint count difference between the most and least populated zone for the data plane, following Kubernetes TopologySpreadConstraints maxSkew semantics; 0 means no skew is tolerated")
 	flag.BoolVar(&cfg.KubernetesEnableEastWest, "enable-kubernetes-east-west", false, "*Deprecated*: use kubernetes-east-west-range feature. Enables east-west communication, which automatically adds routes for Ingress objects with hostname <name>.<namespace>.skipper.cluster.local")
 	flag.StringVar(&cfg.KubernetesEastWestDomain, "kubernetes-east-west-domain", "", "*Deprecated*: use kubernetes-east-west-range feature. Sets the east-west domain, defaults to .skipper.cluster.local")
 	flag.Var(cfg.KubernetesEastWestRangeDomains, "kubernetes-east-west-range-domains", "set the cluster internal domains for east west traffic. Identified routes to such domains will include the -kubernetes-east-west-range-predicates")
@@ -866,6 +870,9 @@ func validate(c *Config) error {
 	if err != nil {
 		return err
 	}
+	if c.KubernetesZoneMaxSkew < 0 {
+		return fmt.Errorf("kubernetes-zone-max-skew must be >= 0, got %d", c.KubernetesZoneMaxSkew)
+	}
 	return c.parseForwardedHeaders()
 }
 
@@ -1131,6 +1138,8 @@ func (c *Config) ToOptions() skipper.Options {
 		KubernetesListChunkSize:                        c.KubernetesListChunkSize,
 		KubernetesEnableEndpointslices:                 c.KubernetesEnableEndpointSlices,
 		KubernetesTopologyZone:                         c.KubernetesTopologyZone,
+		KubernetesDataPlaneLabelSelector:               c.KubernetesDataPlaneLabelSelector.values,
+		KubernetesZoneMaxSkew:                          c.KubernetesZoneMaxSkew,
 		KubernetesEnableEastWest:                       c.KubernetesEnableEastWest,
 		KubernetesEastWestDomain:                       c.KubernetesEastWestDomain,
 		KubernetesEastWestRangeDomains:                 c.KubernetesEastWestRangeDomains.values,

config/config_test.go

@@ -203,6 +203,7 @@ func defaultConfig(with func(*Config)) *Config {
 		MtlsAuthnAppendCA:                       false,
 		MtlsAuthnCaFile:                         "",
 		MtlsAuthnCA:                             x509.NewCertPool(),
+		KubernetesZoneMaxSkew:                   kubernetes.DefaultMaxZoneSkew,
 	}
 	with(cfg)
 	return cfg
@@ -437,6 +438,11 @@ func Test_NewConfigWithArgs(t *testing.T) {
 			args:    []string{"skipper", "-config-file=nonexistent.yaml"},
 			wantErr: true,
 		},
+		{
+			name:    "negative kubernetes-zone-max-skew is rejected",
+			args:    []string{"skipper", "-kubernetes-zone-max-skew=-1"},
+			wantErr: true,
+		},
 		{
 			name: "test only valid flag overwrite yaml file",
 			args: []string{"skipper", "-config-file=testdata/test.yaml", "-address=localhost:8080", "-refuse-payload=baz"},

dataclients/kubernetes/clusterclient.go

@@ -84,6 +84,9 @@ type clusterClient struct {
 	loggedMissingRouteGroups bool
 	routeGroupValidator      *definitions.RouteGroupValidator
 	ingressValidator         *definitions.IngressV1Validator
+	balancedAcrossZones      bool
+	dataPlaneLabelSelector   map[string]string
+	maxZoneSkew              int
 }
 
 var (
@@ -184,6 +187,21 @@ func newClusterClient(o Options, apiURL, ingCls, rgCls string, quit <-chan struc
 		ingressStatusFromService:     o.IngressStatusFromService,
 	}
 
+	if o.DataPlaneLabelSelector != nil {
+		c.dataPlaneLabelSelector = o.DataPlaneLabelSelector
+	} else {
+		c.dataPlaneLabelSelector = DefaultDataPlaneLabelSelector
+	}
+
+	if o.MaxZoneSkew != nil {
+		if *o.MaxZoneSkew < 0 {
+			return nil, fmt.Errorf("invalid MaxZoneSkew %d: must be >= 0", *o.MaxZoneSkew)
+		}
+		c.maxZoneSkew = *o.MaxZoneSkew
+	} else {
+		c.maxZoneSkew = DefaultMaxZoneSkew
+	}
+
 	if o.KubernetesInCluster {
 		c.tokenProvider = secrets.NewSecretPaths(time.Minute)
 		c.tokenFile = serviceAccountDir + serviceAccountTokenKey
@@ -750,9 +768,15 @@ func (c *clusterClient) fetchClusterState() (*clusterState, error) {
 		log.Errorf("failed to update ingress status: %v", err)
 	}
 
+	c.balancedAcrossZones = state.isServiceBalancedAcrossZones(c.dataPlaneLabelSelector, c.maxZoneSkew)
+
 	return state, nil
 }
 
+func (c *clusterClient) IsBalancedDataPlane() bool {
+	return c.balancedAcrossZones
+}
+
 func parseNamespaceName(resource string) (string, string, error) {
 	ns, name, ok := strings.Cut(resource, "/")
 	if !ok || ns == "" || name == "" {

dataclients/kubernetes/clusterclient_test.go

@@ -80,6 +80,42 @@ func TestMissingRouteGroupsCRDLoggedOnlyOnce(t *testing.T) {
 	}
 }
 
+func TestNewRejectsNegativeMaxZoneSkew(t *testing.T) {
+	skew := -1
+	_, err := kubernetes.New(kubernetes.Options{MaxZoneSkew: &skew})
+	assert.Error(t, err, "expected an error for negative zone skew")
+	assert.ErrorContains(t, err, "MaxZoneSkew")
+}
+
+func TestMissingDataPlaneServiceLoggedEveryPoll(t *testing.T) {
+	a, err := kubernetestest.NewAPI(kubernetestest.TestAPIOptions{})
+	require.NoError(t, err)
+
+	s := httptest.NewServer(a)
+	defer s.Close()
+
+	var logBuf bytes.Buffer
+	log.SetOutput(&logBuf)
+	defer log.SetOutput(os.Stderr)
+
+	c, err := kubernetes.New(kubernetes.Options{
+		KubernetesURL:                  s.URL,
+		KubernetesEnableEndpointslices: true,
+	})
+	require.NoError(t, err)
+	defer c.Close()
+
+	const pollCount = 3
+	for range pollCount {
+		_, err := c.LoadAll()
+		require.NoError(t, err)
+	}
+
+	const wantMessage = "no service found matching map[application:skipper-ingress component:ingress]; cannot determine data-plane zone balance"
+
+	assert.True(t, containsCount(logBuf.String(), wantMessage, pollCount), "expected %q to be logged exactly %d times, log output:\n%s", wantMessage, pollCount, logBuf.String())
+}
+
 func TestLoadRouteGroups(t *testing.T) {
 
 	for _, tt := range []struct {

dataclients/kubernetes/clusterstate.go

@@ -2,6 +2,7 @@ package kubernetes
 
 import (
 	"fmt"
+	"math"
 	"sort"
 	"sync"
 
@@ -23,6 +24,71 @@ type clusterState struct {
 	enableEndpointSlices bool
 }
 
+// matchesAllLabels reports whether m contains every key/value pair in selector.
+func matchesAllLabels(labels, toMatch map[string]string) bool {
+	for k, v := range toMatch {
+		if labels[k] != v {
+			return false
+		}
+	}
+	return true
+}
+
+// isServiceBalancedAcrossZones reports whether the combined endpoints of
+// every Service matching labelSelector are evenly spread across zones.
+func (state *clusterState) isServiceBalancedAcrossZones(labelSelector map[string]string, maxZoneSkew int) bool {
+	if !state.enableEndpointSlices {
+		return false
+	}
+
+	if len(labelSelector) == 0 {
+		log.Errorf("empty data-plane label selector; cannot determine zone balance")
+		return false
+	}
+
+	var matched int
+	zoneCounts := map[string]int{}
+	for resID, svc := range state.services {
+		if svc.Meta == nil || !matchesAllLabels(svc.Meta.Labels, labelSelector) {
+			continue
+		}
+		matched++
+		eps, ok := state.endpointSlices[resID]
+		if !ok {
+			log.Errorf("endpointslice %s not found for service matching %v; excluding it from the data-plane zone balance check", resID, labelSelector)
+			continue
+		}
+		for _, ep := range eps.Endpoints {
+			if ep.Zone != "" {
+				zoneCounts[ep.Zone]++
+			}
+		}
+	}
+
+	if matched == 0 {
+		log.Errorf("no service found matching %v; cannot determine data-plane zone balance", labelSelector)
+		return false
+	}
+
+	if len(zoneCounts) < minZones {
+		return false
+	}
+
+	globalMin := math.MaxInt
+	for _, count := range zoneCounts {
+		if count < globalMin {
+			globalMin = count
+		}
+	}
+
+	for _, count := range zoneCounts {
+		if count-globalMin > maxZoneSkew {
+			return false
+		}
+	}
+	return true
+}
+
 func (state *clusterState) getService(namespace, name string) (*service, error) {
 	state.mu.Lock()
 	defer state.mu.Unlock()