From fec7ddb382dcf0cb9d06f688b5e169741503c17b Mon Sep 17 00:00:00 2001 From: Mustafa Abdelrahman Date: Thu, 22 Oct 2026 09:35:07 +0200 Subject: [PATCH 1/8] use a zone aware clouser to check if ingress controller (skipper) was distributed across zone in this current poll Signed-off-by: Mustafa Abdelrahman --- dataclients/kubernetes/clusterclient.go | 7 ++ dataclients/kubernetes/clusterstate.go | 19 +++++ dataclients/kubernetes/kube.go | 1 + routesrv/eskipbytes.go | 104 +++++++++++++----------- routesrv/polling.go | 100 +++++++++++++---------- routesrv/routesrv.go | 1 + 6 files changed, 143 insertions(+), 89 deletions(-) diff --git a/dataclients/kubernetes/clusterclient.go b/dataclients/kubernetes/clusterclient.go index e62642bc85..e8742cedb2 100644 --- a/dataclients/kubernetes/clusterclient.go +++ b/dataclients/kubernetes/clusterclient.go @@ -84,6 +84,7 @@ type clusterClient struct { loggedMissingRouteGroups bool routeGroupValidator *definitions.RouteGroupValidator ingressValidator *definitions.IngressV1Validator + distributedAcrossZones bool } var ( @@ -750,9 +751,15 @@ func (c *clusterClient) fetchClusterState() (*clusterState, error) { log.Errorf("failed to update ingress status: %v", err) } + c.distributedAcrossZones = state.ingressControllerDistributedAcrossZones("kube-system", "skipper-ingress") + return state, nil } +func (c *clusterClient) IngressControllerDistributedAcrossZones() bool { + return c.distributedAcrossZones +} + func parseNamespaceName(resource string) (string, string, error) { ns, name, ok := strings.Cut(resource, "/") if !ok || ns == "" || name == "" { diff --git a/dataclients/kubernetes/clusterstate.go b/dataclients/kubernetes/clusterstate.go index dca1732da0..8f959aa564 100644 --- a/dataclients/kubernetes/clusterstate.go +++ b/dataclients/kubernetes/clusterstate.go @@ -23,6 +23,25 @@ type clusterState struct { enableEndpointSlices bool } +func (state *clusterState) ingressControllerDistributedAcrossZones(namespace, name string) bool { + if !state.enableEndpointSlices { + return false + } + + eps, ok := state.endpointSlices[newResourceID(namespace, name)] + if !ok { + return false + } + + zones := map[string]struct{}{} + for _, ep := range eps.Endpoints { + if ep.Zone != "" { + zones[ep.Zone] = struct{}{} + } + } + return len(zones) >= minZones +} + func (state *clusterState) getService(namespace, name string) (*service, error) { state.mu.Lock() defer state.mu.Unlock() diff --git a/dataclients/kubernetes/kube.go b/dataclients/kubernetes/kube.go index 8a1b64bf2d..d428579210 100644 --- a/dataclients/kubernetes/kube.go +++ b/dataclients/kubernetes/kube.go @@ -28,6 +28,7 @@ const ( httpRedirectRouteID = "kube__redirect" defaultEastWestDomain = "skipper.cluster.local" minEndpointsByZone = 3 + minZones = 3 ) // PathMode values are used to control the ingress path interpretation. The path mode can diff --git a/routesrv/eskipbytes.go b/routesrv/eskipbytes.go index cf0b481ecb..df96759ad0 100644 --- a/routesrv/eskipbytes.go +++ b/routesrv/eskipbytes.go @@ -74,11 +74,15 @@ type eskipBytes struct { // formatAndSet takes a slice of routes and stores them eskip-formatted // in a synchronized way. It returns the length of the stored data, and // flags signaling whether the data was initialized and updated. -// Each zone's routes are formatted and hashed exactly once per call; only -// the zones whose hash actually changed are recompressed and get a fresh -// Last-Modified. Unchanged zones keep their existing Etag/Last-Modified/ -// compressed bytes untouched, and zones that disappeared entirely are -// removed so requests for them fall back to the global routing table. +// An update is triggered when either the global routes hash or any +// per-zone route hash differs from the previously stored value, so a +// change in zone topology alone (without route changes) is detected +// independently. Each zone's routes are formatted and hashed exactly +// once per call; only the zones whose hash actually changed are +// recompressed and get a fresh Last-Modified. Unchanged zones keep +// their existing Etag/Last-Modified/compressed bytes untouched, and +// zones that disappeared entirely are removed so requests for them +// fall back to the global routing table. func (e *eskipBytes) formatAndSet(routes []*eskip.Route, zoneAwareRoutes map[string][]*eskip.Route) (_ int, _ string, initialized bool, updated bool) { buf := &bytes.Buffer{} @@ -89,55 +93,62 @@ func (e *eskipBytes) formatAndSet(routes []*eskip.Route, zoneAwareRoutes map[str defer e.mu.Unlock() hash := fmt.Sprintf("%x", sha256.Sum256(data)) - updated = hash != e.hash - - if updated { - now := e.now() - if e.zoneHash == nil { - e.zoneHash = make(map[string]string) - e.zoneDataCompressed = make(map[string][]byte) - e.zoneCount = make(map[string]int) - e.zoneLastModified = make(map[string]time.Time) - } + globalUpdated := hash != e.hash - // Format and hash every zone exactly once, so unchanged zones - // never need to be recompressed below. - newZoneHash := make(map[string]string, len(zoneAwareRoutes)) - for zone, zoneRoutes := range zoneAwareRoutes { - zoneBuf := &bytes.Buffer{} - eskip.Fprint(zoneBuf, eskip.PrettyPrintInfo{Pretty: false, IndentStr: ""}, zoneRoutes...) - zoneData := zoneBuf.Bytes() - zoneHash := fmt.Sprintf("%x", sha256.Sum256(zoneData)) - newZoneHash[zone] = zoneHash - - if e.zoneHash[zone] == zoneHash { - continue // this zone's content is unchanged; leave it alone - } - e.zoneHash[zone] = zoneHash - e.zoneDataCompressed[zone] = e.compressLocked(zoneData) - e.zoneCount[zone] = len(zoneRoutes) - e.zoneLastModified[zone] = now - } - for zone := range e.zoneHash { - if _, ok := newZoneHash[zone]; !ok { - // zone no longer present at all - remove it so ServeHTTP - // falls back to the global routing table for it instead - // of serving stale data. - delete(e.zoneHash, zone) - delete(e.zoneDataCompressed, zone) - delete(e.zoneCount, zone) - delete(e.zoneLastModified, zone) - } - } + now := e.now() + + if globalUpdated { e.lastModified = now - e.zdata = e.compressLocked(data) e.hash = hash e.count = len(routes) + e.zdata = e.compressLocked(data) } + + if e.zoneHash == nil { + e.zoneHash = make(map[string]string) + e.zoneDataCompressed = make(map[string][]byte) + e.zoneCount = make(map[string]int) + e.zoneLastModified = make(map[string]time.Time) + } + + // Format and hash every zone exactly once, so unchanged zones + // never need to be recompressed below. + zoneUpdated := false + newZoneHash := make(map[string]string, len(zoneAwareRoutes)) + for zone, zoneRoutes := range zoneAwareRoutes { + zoneBuf := &bytes.Buffer{} + eskip.Fprint(zoneBuf, eskip.PrettyPrintInfo{Pretty: false, IndentStr: ""}, zoneRoutes...) + zoneData := zoneBuf.Bytes() + zoneHash := fmt.Sprintf("%x", sha256.Sum256(zoneData)) + newZoneHash[zone] = zoneHash + + if e.zoneHash[zone] == zoneHash { + continue // this zone's content is unchanged; leave it alone + } + zoneUpdated = true + e.zoneHash[zone] = zoneHash + e.zoneDataCompressed[zone] = e.compressLocked(zoneData) + e.zoneCount[zone] = len(zoneRoutes) + e.zoneLastModified[zone] = now + } + for zone := range e.zoneHash { + if _, ok := newZoneHash[zone]; !ok { + // zone no longer present at all (e.g. the gate closed, or the + // zone's last endpoint disappeared) - remove it so ServeHTTP + // falls back to the global routing table for it instead of + // serving stale data. + zoneUpdated = true + delete(e.zoneHash, zone) + delete(e.zoneDataCompressed, zone) + delete(e.zoneCount, zone) + delete(e.zoneLastModified, zone) + } + } + initialized = !e.initialized e.initialized = true - return len(data), e.hash, initialized, updated + return len(data), e.hash, initialized, globalUpdated || zoneUpdated } // compressLocked compresses the data with gzip and returns @@ -208,6 +219,7 @@ func (e *eskipBytes) ServeHTTP(rw http.ResponseWriter, r *http.Request) { count = e.zoneCount[zone] hash = e.zoneHash[zone] lastModified = e.zoneLastModified[zone] + w.Header().Set("X-Zone-Aware", "true") } } diff --git a/routesrv/polling.go b/routesrv/polling.go index 390aecd1ac..ca8d65a0c0 100644 --- a/routesrv/polling.go +++ b/routesrv/polling.go @@ -42,8 +42,9 @@ type poller struct { cloneRoute []*eskip.Clone // visibility - tracer ot.Tracer - metrics metrics.Metrics + tracer ot.Tracer + metrics metrics.Metrics + isIngressControllerDistributedAcrossZones func() bool } func (p *poller) poll(wg *sync.WaitGroup) { @@ -55,8 +56,9 @@ func (p *poller) poll(wg *sync.WaitGroup) { p.setGaugeToCurrentTime("polling_started_timestamp") var ( - lastRoutesByID map[string]string - lastRoutes []*eskip.Route + lastRoutesByID map[string]string + lastRoutes []*eskip.Route + lastDistributed bool ) for { @@ -73,48 +75,60 @@ func (p *poller) poll(wg *sync.WaitGroup) { "message", fmt.Sprintf("%s: %s", LogRoutesFetchingFailed, err), ) - } else if hasChanged(lastRoutes, routes) { - lastRoutes = routes - - routes = p.process(routes) - routesCount := len(routes) - span.SetTag("routes.count", routesCount) - - switch { - case routesCount == 0: - p.handleEmptyRoutes() - case routesCount > 0: - zoneAwareRoutes := filterRoutesByZone(routes) - routesBytes, routesHash, initialized, updated := p.b.formatAndSet(routes, zoneAwareRoutes) - logger := log.WithFields(log.Fields{"count": routesCount, "bytes": routesBytes, "hash": routesHash}) - if initialized { - logger.Info(LogRoutesInitialized) - span.SetTag("routes.initialized", true) - p.setGaugeToCurrentTime("routes.initialized_timestamp") - } - if updated { - logger.Info(LogRoutesUpdated) - span.SetTag("routes.updated", true) - p.setGaugeToCurrentTime("routes.updated_timestamp") - p.metrics.UpdateGauge("routes.total", float64(routesCount)) - p.metrics.UpdateGauge("routes.byte", float64(routesBytes)) + } else { + distributed := p.isIngressControllerDistributedAcrossZones() + routesChanged := hasChanged(lastRoutes, routes) + distributedChanged := distributed != lastDistributed + + if routesChanged || distributedChanged { + lastRoutes = routes + lastDistributed = distributed + + routes = p.process(routes) + routesCount := len(routes) + span.SetTag("routes.count", routesCount) + + switch { + case routesCount == 0: + p.handleEmptyRoutes() + case routesCount > 0: + var zoneAwareRoutes map[string][]*eskip.Route + if distributed { + zoneAwareRoutes = filterRoutesByZone(routes) + } else { + log.Debug("skipper not covering all zones, falling back to global routing table") + } + routesBytes, routesHash, initialized, updated := p.b.formatAndSet(routes, zoneAwareRoutes) + logger := log.WithFields(log.Fields{"count": routesCount, "bytes": routesBytes, "hash": routesHash}) + if initialized { + logger.Info(LogRoutesInitialized) + span.SetTag("routes.initialized", true) + p.setGaugeToCurrentTime("routes.initialized_timestamp") + } + if updated { + logger.Info(LogRoutesUpdated) + span.SetTag("routes.updated", true) + p.setGaugeToCurrentTime("routes.updated_timestamp") + p.metrics.UpdateGauge("routes.total", float64(routesCount)) + p.metrics.UpdateGauge("routes.byte", float64(routesBytes)) + } + span.SetTag("routes.bytes", routesBytes) + span.SetTag("routes.hash", routesHash) + + if updated && log.IsLevelEnabled(log.DebugLevel) { + routesByID := mapRoutes(routes) + logChanges(routesByID, lastRoutesByID) + lastRoutesByID = routesByID + } } - span.SetTag("routes.bytes", routesBytes) - span.SetTag("routes.hash", routesHash) - - if updated && log.IsLevelEnabled(log.DebugLevel) { - routesByID := mapRoutes(routes) - logChanges(routesByID, lastRoutesByID) - lastRoutesByID = routesByID + } else { + log.Info(LogRoutesNoChange) + span.SetTag("routes.updated", false) + span.SetTag("routes.count", len(lastRoutes)) + if len(routes) == 0 { + p.handleEmptyRoutes() } } - } else { - log.Info(LogRoutesNoChange) - span.SetTag("routes.updated", false) - span.SetTag("routes.count", len(lastRoutes)) - if len(routes) == 0 { - p.handleEmptyRoutes() - } } span.Finish() diff --git a/routesrv/routesrv.go b/routesrv/routesrv.go index fdc882abd0..f3c8d3d66e 100644 --- a/routesrv/routesrv.go +++ b/routesrv/routesrv.go @@ -190,6 +190,7 @@ func New(opts skipper.Options) (*RouteServer, error) { oauth2Config: oauthConfig, tracer: tracer, metrics: m, + isIngressControllerDistributedAcrossZones: dataclient.ClusterClient.IngressControllerDistributedAcrossZones, } rs.wg = &sync.WaitGroup{} From 2020a2335d4e4140689dad4fc5eb235b9e01f523 Mon Sep 17 00:00:00 2001 From: Mustafa Abdelrahman Date: Thu, 22 Oct 2026 12:12:15 +0200 Subject: [PATCH 2/8] make it configurable and work on the skew not only missing zones Signed-off-by: Mustafa Abdelrahman --- config/config.go | 12 ++ config/config_test.go | 8 + dataclients/kubernetes/clusterclient.go | 31 +++- dataclients/kubernetes/clusterstate.go | 26 +++- dataclients/kubernetes/clusterstate_test.go | 163 ++++++++++++++++++++ dataclients/kubernetes/kube.go | 27 ++++ routesrv/export_test.go | 4 + routesrv/routesrv_test.go | 59 ++++++- skipper.go | 19 +++ 9 files changed, 337 insertions(+), 12 deletions(-) diff --git a/config/config.go b/config/config.go index 1ee986769c..597c60a34c 100644 --- a/config/config.go +++ b/config/config.go @@ -192,6 +192,9 @@ type Config struct { KubernetesListChunkSize int `yaml:"kubernetes-list-chunk-size"` KubernetesEnableEndpointSlices bool `yaml:"enable-kubernetes-endpointslices"` KubernetesTopologyZone string `yaml:"kubernetes-topology-zone"` + KubernetesIngressControllerNamespace string `yaml:"kubernetes-ingress-controller-namespace"` + KubernetesIngressControllerName string `yaml:"kubernetes-ingress-controller-name"` + 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 +610,9 @@ 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.StringVar(&cfg.KubernetesIngressControllerNamespace, "kubernetes-ingress-controller-namespace", kubernetes.DefaultIngressControllerNamespace, "namespace of the ingress controller service used for zone distribution checks") + flag.StringVar(&cfg.KubernetesIngressControllerName, "kubernetes-ingress-controller-name", kubernetes.DefaultIngressControllerName, "service name of the ingress controller used for zone distribution checks") + flag.IntVar(&cfg.KubernetesZoneMaxSkew, "kubernetes-zone-max-skew", kubernetes.DefaultMaxZoneSkew, "maximum permitted endpoint count difference between the most and least populated zone for the ingress controller, 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 ..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 +872,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 +1140,9 @@ func (c *Config) ToOptions() skipper.Options { KubernetesListChunkSize: c.KubernetesListChunkSize, KubernetesEnableEndpointslices: c.KubernetesEnableEndpointSlices, KubernetesTopologyZone: c.KubernetesTopologyZone, + KubernetesIngressControllerNamespace: c.KubernetesIngressControllerNamespace, + KubernetesIngressControllerName: c.KubernetesIngressControllerName, + KubernetesZoneMaxSkew: c.KubernetesZoneMaxSkew, KubernetesEnableEastWest: c.KubernetesEnableEastWest, KubernetesEastWestDomain: c.KubernetesEastWestDomain, KubernetesEastWestRangeDomains: c.KubernetesEastWestRangeDomains.values, diff --git a/config/config_test.go b/config/config_test.go index 37c40ffeb8..9ed23d3572 100644 --- a/config/config_test.go +++ b/config/config_test.go @@ -203,6 +203,9 @@ func defaultConfig(with func(*Config)) *Config { MtlsAuthnAppendCA: false, MtlsAuthnCaFile: "", MtlsAuthnCA: x509.NewCertPool(), + KubernetesIngressControllerNamespace: kubernetes.DefaultIngressControllerNamespace, + KubernetesIngressControllerName: kubernetes.DefaultIngressControllerName, + KubernetesZoneMaxSkew: kubernetes.DefaultMaxZoneSkew, } with(cfg) return cfg @@ -437,6 +440,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"}, diff --git a/dataclients/kubernetes/clusterclient.go b/dataclients/kubernetes/clusterclient.go index e8742cedb2..8ee43fe0f9 100644 --- a/dataclients/kubernetes/clusterclient.go +++ b/dataclients/kubernetes/clusterclient.go @@ -81,10 +81,13 @@ type clusterClient struct { enableEndpointSlices bool listChunkSize int - loggedMissingRouteGroups bool - routeGroupValidator *definitions.RouteGroupValidator - ingressValidator *definitions.IngressV1Validator - distributedAcrossZones bool + loggedMissingRouteGroups bool + routeGroupValidator *definitions.RouteGroupValidator + ingressValidator *definitions.IngressV1Validator + distributedAcrossZones bool + ingressControllerNamespace string + ingressControllerName string + maxZoneSkew int } var ( @@ -185,6 +188,24 @@ func newClusterClient(o Options, apiURL, ingCls, rgCls string, quit <-chan struc ingressStatusFromService: o.IngressStatusFromService, } + ns := o.IngressControllerNamespace + if ns == "" { + ns = DefaultIngressControllerNamespace + } + c.ingressControllerNamespace = ns + + name := o.IngressControllerName + if name == "" { + name = DefaultIngressControllerName + } + c.ingressControllerName = name + + if o.MaxZoneSkew != nil { + c.maxZoneSkew = *o.MaxZoneSkew + } else { + c.maxZoneSkew = DefaultMaxZoneSkew + } + if o.KubernetesInCluster { c.tokenProvider = secrets.NewSecretPaths(time.Minute) c.tokenFile = serviceAccountDir + serviceAccountTokenKey @@ -751,7 +772,7 @@ func (c *clusterClient) fetchClusterState() (*clusterState, error) { log.Errorf("failed to update ingress status: %v", err) } - c.distributedAcrossZones = state.ingressControllerDistributedAcrossZones("kube-system", "skipper-ingress") + c.distributedAcrossZones = state.ingressControllerDistributedAcrossZones(c.ingressControllerNamespace, c.ingressControllerName, c.maxZoneSkew) return state, nil } diff --git a/dataclients/kubernetes/clusterstate.go b/dataclients/kubernetes/clusterstate.go index 8f959aa564..e22aa2f044 100644 --- a/dataclients/kubernetes/clusterstate.go +++ b/dataclients/kubernetes/clusterstate.go @@ -2,6 +2,7 @@ package kubernetes import ( "fmt" + "math" "sort" "sync" @@ -23,7 +24,7 @@ type clusterState struct { enableEndpointSlices bool } -func (state *clusterState) ingressControllerDistributedAcrossZones(namespace, name string) bool { +func (state *clusterState) ingressControllerDistributedAcrossZones(namespace, name string, maxZoneSkew int) bool { if !state.enableEndpointSlices { return false } @@ -33,13 +34,30 @@ func (state *clusterState) ingressControllerDistributedAcrossZones(namespace, na return false } - zones := map[string]struct{}{} + zoneCounts := map[string]int{} for _, ep := range eps.Endpoints { if ep.Zone != "" { - zones[ep.Zone] = struct{}{} + zoneCounts[ep.Zone]++ } } - return len(zones) >= minZones + + 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) { diff --git a/dataclients/kubernetes/clusterstate_test.go b/dataclients/kubernetes/clusterstate_test.go index f7c55e96a8..be20c137c4 100644 --- a/dataclients/kubernetes/clusterstate_test.go +++ b/dataclients/kubernetes/clusterstate_test.go @@ -84,3 +84,166 @@ func BenchmarkCachedEndpoint(b *testing.B) { }) } } + +// makeSkipperEndpointSlice builds a skipperEndpointSlice from a map of zone→count. +func makeSkipperEndpointSlice(zoneCounts map[string]int) *skipperEndpointSlice { + eps := &skipperEndpointSlice{} + i := 0 + for zone, count := range zoneCounts { + for range count { + eps.Endpoints = append(eps.Endpoints, &skipperEndpoint{ + Address: "10.0." + strconv.Itoa(i) + ".1", + Zone: zone, + }) + i++ + } + } + return eps +} + +func TestIngressControllerDistributedAcrossZones(t *testing.T) { + const ( + ns = DefaultIngressControllerNamespace + name = DefaultIngressControllerName + ) + + for _, tt := range []struct { + name string + endpointSlicesEnable bool + endpointSlicePresent bool + zoneCounts map[string]int + maxZoneSkew int + customNS string + customName string + want bool + }{ + { + name: "endpoint slices disabled", + endpointSlicesEnable: false, + endpointSlicePresent: true, + zoneCounts: map[string]int{"a": 3, "b": 3, "c": 3}, + maxZoneSkew: 1, + want: false, + }, + { + name: "no endpointslice for ingress controller", + endpointSlicesEnable: true, + endpointSlicePresent: false, + zoneCounts: map[string]int{"a": 3, "b": 3, "c": 3}, + maxZoneSkew: 1, + want: false, + }, + { + name: "fewer than minZones zones", + endpointSlicesEnable: true, + endpointSlicePresent: true, + zoneCounts: map[string]int{"a": 3, "b": 3}, + maxZoneSkew: 1, + want: false, + }, + { + name: "balanced zones, maxSkew=0", + endpointSlicesEnable: true, + endpointSlicePresent: true, + zoneCounts: map[string]int{"a": 3, "b": 3, "c": 3}, + maxZoneSkew: 0, + want: true, + }, + { + name: "balanced zones, maxSkew=1", + endpointSlicesEnable: true, + endpointSlicePresent: true, + zoneCounts: map[string]int{"a": 3, "b": 3, "c": 3}, + maxZoneSkew: 1, + want: true, + }, + { + name: "skew=1 within maxSkew=1", + endpointSlicesEnable: true, + endpointSlicePresent: true, + zoneCounts: map[string]int{"a": 4, "b": 3, "c": 3}, + maxZoneSkew: 1, + want: true, + }, + { + name: "skew=1 exceeds maxSkew=0", + endpointSlicesEnable: true, + endpointSlicePresent: true, + zoneCounts: map[string]int{"a": 4, "b": 3, "c": 3}, + maxZoneSkew: 0, + want: false, + }, + { + name: "skew=2 exceeds maxSkew=1", + endpointSlicesEnable: true, + endpointSlicePresent: true, + zoneCounts: map[string]int{"a": 5, "b": 3, "c": 3}, + maxZoneSkew: 1, + want: false, + }, + { + name: "skew=2 within maxSkew=2", + endpointSlicesEnable: true, + endpointSlicePresent: true, + zoneCounts: map[string]int{"a": 5, "b": 3, "c": 3}, + maxZoneSkew: 2, + want: true, + }, + { + name: "custom namespace and name match", + endpointSlicesEnable: true, + endpointSlicePresent: true, + zoneCounts: map[string]int{"a": 3, "b": 3, "c": 3}, + maxZoneSkew: 1, + customNS: "custom-ns", + customName: "custom-ingress", + want: true, + }, + { + name: "custom namespace and name mismatch", + endpointSlicesEnable: true, + endpointSlicePresent: true, + zoneCounts: map[string]int{"a": 3, "b": 3, "c": 3}, + maxZoneSkew: 1, + customNS: "wrong-ns", + customName: "wrong-ingress", + want: false, + }, + } { + t.Run(tt.name, func(t *testing.T) { + lookupNS := ns + lookupName := name + if tt.customNS != "" { + lookupNS = tt.customNS + } + if tt.customName != "" { + lookupName = tt.customName + } + + endpointSlices := map[definitions.ResourceID]*skipperEndpointSlice{} + if tt.endpointSlicePresent { + // always store under the canonical ns/name; custom test cases + // store under the custom key so that the lookup matches. + storeNS, storeName := ns, name + if tt.customNS != "" && tt.customNS != "wrong-ns" { + storeNS = tt.customNS + } + if tt.customName != "" && tt.customName != "wrong-ingress" { + storeName = tt.customName + } + endpointSlices[newResourceID(storeNS, storeName)] = makeSkipperEndpointSlice(tt.zoneCounts) + } + + state := &clusterState{ + enableEndpointSlices: tt.endpointSlicesEnable, + endpointSlices: endpointSlices, + } + + got := state.ingressControllerDistributedAcrossZones(lookupNS, lookupName, tt.maxZoneSkew) + if got != tt.want { + t.Errorf("ingressControllerDistributedAcrossZones(%q, %q, %d) = %v, want %v", + lookupNS, lookupName, tt.maxZoneSkew, got, tt.want) + } + }) + } +} diff --git a/dataclients/kubernetes/kube.go b/dataclients/kubernetes/kube.go index d428579210..57b2e98a4b 100644 --- a/dataclients/kubernetes/kube.go +++ b/dataclients/kubernetes/kube.go @@ -29,6 +29,17 @@ const ( defaultEastWestDomain = "skipper.cluster.local" minEndpointsByZone = 3 minZones = 3 + + // DefaultIngressControllerNamespace is the default namespace used to look up + // the ingress controller endpointslice when checking zone distribution. + DefaultIngressControllerNamespace = "kube-system" + // DefaultIngressControllerName is the default service name used to look up + // the ingress controller endpointslice when checking zone distribution. + DefaultIngressControllerName = "skipper-ingress" + // DefaultMaxZoneSkew is the default maximum permitted difference between + // the endpoint count in any zone and the global minimum across all zones, + // following Kubernetes TopologySpreadConstraints maxSkew semantics. + DefaultMaxZoneSkew = 1 ) // PathMode values are used to control the ingress path interpretation. The path mode can @@ -284,6 +295,22 @@ type Options struct { // IngressStatusFromService, when set to /, makes skipper update ingress status.loadBalancer.ingress // addresses from the referenced Service object. IngressStatusFromService string + + // IngressControllerNamespace is the namespace of the ingress controller service + // whose endpoint zone distribution is checked to decide whether zone-aware routing + // is enabled. Defaults to "kube-system". + IngressControllerNamespace string + + // IngressControllerName is the service name of the ingress controller whose endpoint + // zone distribution is checked to decide whether zone-aware routing is enabled. + // Defaults to "skipper-ingress". + IngressControllerName string + + // MaxZoneSkew is the maximum permitted difference between the endpoint count in any + // zone and the global minimum across all zones for the ingress controller, following + // Kubernetes TopologySpreadConstraints maxSkew semantics. + // nil means use DefaultMaxZoneSkew. A pointer to 0 means no skew is tolerated. + MaxZoneSkew *int } // Client is a Skipper DataClient implementation used to create routes based on Kubernetes Ingress settings. diff --git a/routesrv/export_test.go b/routesrv/export_test.go index f44440458a..e9be8e0c38 100644 --- a/routesrv/export_test.go +++ b/routesrv/export_test.go @@ -10,6 +10,10 @@ func SetNow(rs *RouteServer, now func() time.Time) { rs.poller.b.now = now } +func SetDistributed(rs *RouteServer, v bool) { + rs.poller.isIngressControllerDistributedAcrossZones = func() bool { return v } +} + func (rs *RouteServer) ListenAndServe() (err error) { if tlsConfig := rs.server.TLSConfig; tlsConfig != nil { if err = rs.server.ListenAndServeTLS("", ""); err != http.ErrServerClosed { diff --git a/routesrv/routesrv_test.go b/routesrv/routesrv_test.go index 2c68882a68..5132ef22fb 100644 --- a/routesrv/routesrv_test.go +++ b/routesrv/routesrv_test.go @@ -1140,7 +1140,7 @@ func TestRoutesWithZone(t *testing.T) { KubernetesURL: ks.URL, KubernetesEnableEndpointslices: true, }) - + routesrv.SetDistributed(rs, true) rs.StartUpdates() defer rs.StopUpdates() @@ -1175,7 +1175,7 @@ func TestEtagWithZoneAwareRoutingFallback(t *testing.T) { KubernetesURL: ks.URL, KubernetesEnableEndpointslices: true, }) - + routesrv.SetDistributed(rs, true) rs.StartUpdates() defer rs.StopUpdates() @@ -1230,7 +1230,7 @@ func TestZoneAwareRoutesGzip(t *testing.T) { KubernetesURL: ks.URL, KubernetesEnableEndpointslices: true, }) - + routesrv.SetDistributed(rs, true) rs.StartUpdates() defer rs.StopUpdates() @@ -1279,6 +1279,7 @@ func TestZoneAwareXCount(t *testing.T) { KubernetesURL: ks.URL, KubernetesEnableEndpointslices: true, }) + routesrv.SetDistributed(rs, true) rs.StartUpdates() defer rs.StopUpdates() require.NoError(t, tl.WaitFor(routesrv.LogRoutesInitialized, waitTimeout)) @@ -1311,6 +1312,7 @@ func TestZoneAwareEtag(t *testing.T) { KubernetesURL: ks.URL, KubernetesEnableEndpointslices: true, }) + routesrv.SetDistributed(rs, true) rs.StartUpdates() defer rs.StopUpdates() require.NoError(t, tl.WaitFor(routesrv.LogRoutesInitialized, waitTimeout)) @@ -1354,6 +1356,7 @@ func TestZoneAwareConditionalGetEtag(t *testing.T) { KubernetesURL: ks.URL, KubernetesEnableEndpointslices: true, }) + routesrv.SetDistributed(rs, true) rs.StartUpdates() defer rs.StopUpdates() require.NoError(t, tl.WaitFor(routesrv.LogRoutesInitialized, waitTimeout)) @@ -1401,6 +1404,7 @@ func TestZoneAwareConditionalGetLastModified(t *testing.T) { KubernetesURL: ks.URL, KubernetesEnableEndpointslices: true, }) + routesrv.SetDistributed(rs, true) rs.StartUpdates() defer rs.StopUpdates() require.NoError(t, tl.WaitFor(routesrv.LogRoutesInitialized, waitTimeout)) @@ -1438,6 +1442,7 @@ func TestZoneAwareFallbackToAllRoutesWhenBelowThreshold(t *testing.T) { KubernetesURL: ks.URL, KubernetesEnableEndpointslices: true, }) + routesrv.SetDistributed(rs, true) rs.StartUpdates() defer rs.StopUpdates() require.NoError(t, tl.WaitFor(routesrv.LogRoutesInitialized, waitTimeout)) @@ -1464,6 +1469,53 @@ func TestZoneAwareFallbackToAllRoutesWhenBelowThreshold(t *testing.T) { assert.Equal(t, fullCount, gzipResp.Header().Get(routing.RoutesCountName)) } +// TestZoneAwareRoutesNotServedWhenNotDistributed verifies that when the ingress +// controller is not distributed across enough zones, /routes/{zone} falls back +// to the global routing table — identical to /routes. +func TestZoneAwareRoutesNotServedWhenNotDistributed(t *testing.T) { + defer tl.Reset() + + ks, _ := newKubeServer(t, loadKubeYAML(t, "testdata/zone-aware-traffic/all-zones-3-addr.yaml")) + ks.Start() + defer ks.Close() + rs := newRouteServerWithOptions(t, skipper.Options{ + SourcePollTimeout: pollInterval, + Kubernetes: true, + KubernetesURL: ks.URL, + KubernetesEnableEndpointslices: true, + }) + routesrv.SetDistributed(rs, false) + rs.StartUpdates() + defer rs.StopUpdates() + require.NoError(t, tl.WaitFor(routesrv.LogRoutesInitialized, waitTimeout)) + + // /routes returns the full unfiltered table + fullResp := getRoutes(rs) + require.Equal(t, http.StatusOK, fullResp.Code) + + // /routes/{zone} must return the same body as /routes — no zone filtering + zoneResp := getZoneAwareRoutes(rs, "eu-central-1a") + require.Equal(t, http.StatusOK, zoneResp.Code) + + fullRoutes := eskip.MustParse(fullResp.Body.String()) + zoneRoutes := eskip.MustParse(zoneResp.Body.String()) + assert.True(t, eskip.EqLists(fullRoutes, zoneRoutes), "zone response must equal global response when not distributed") + + // X-Count must also match + assert.Equal(t, + fullResp.Header().Get(routing.RoutesCountName), + zoneResp.Header().Get(routing.RoutesCountName), + "X-Count must match global count when not distributed", + ) + + // same for gzip + gzipZoneResp := getZoneAwareRoutesGzip(rs, "eu-central-1a") + require.Equal(t, http.StatusOK, gzipZoneResp.Code) + gotGzip, err := eskip.Parse(string(decompressGzip(t, gzipZoneResp.Body))) + require.NoError(t, err) + assert.True(t, eskip.EqLists(fullRoutes, gotGzip), "gzip zone response must also equal global response when not distributed") +} + func TestXCountShouldBeSameForZoneAwareAndZoneUnawareRoutes(t *testing.T) { defer tl.Reset() @@ -1477,6 +1529,7 @@ func TestXCountShouldBeSameForZoneAwareAndZoneUnawareRoutes(t *testing.T) { KubernetesURL: ks.URL, KubernetesEnableEndpointslices: true, }) + routesrv.SetDistributed(rs, true) rs.StartUpdates() defer rs.StopUpdates() require.NoError(t, tl.WaitFor(routesrv.LogRoutesInitialized, waitTimeout)) diff --git a/skipper.go b/skipper.go index d2b8a03fbd..2f3c65ca54 100644 --- a/skipper.go +++ b/skipper.go @@ -378,6 +378,22 @@ type Options struct { // KubernetesTopologyZone if set to non empty string it is used to filter endpointslice endpoints by this value. KubernetesTopologyZone string + // KubernetesIngressControllerNamespace is the namespace of the ingress controller service + // whose endpoint zone distribution is checked to decide whether zone-aware routing is enabled. + // Defaults to "kube-system". + KubernetesIngressControllerNamespace string + + // KubernetesIngressControllerName is the service name of the ingress controller whose endpoint + // zone distribution is checked to decide whether zone-aware routing is enabled. + // Defaults to "skipper-ingress". + KubernetesIngressControllerName string + + // KubernetesZoneMaxSkew is the maximum permitted difference between the endpoint count in any + // zone and the global minimum across all zones for the ingress controller, following Kubernetes + // TopologySpreadConstraints maxSkew semantics. + // 0 means no skew is tolerated — all zones must have an equal endpoint count. + KubernetesZoneMaxSkew int + // KubernetesForceService overrides the default Skipper functionality to route traffic using Kubernetes Endpoints, // instead using Kubernetes Services. KubernetesForceService bool @@ -1279,6 +1295,9 @@ func (o *Options) KubernetesDataClientOptions() kubernetes.Options { DefaultLoadBalancerAlgorithm: o.KubernetesDefaultLoadBalancerAlgorithm, ForwardBackendURL: o.ForwardBackendURL, TopologyZone: o.KubernetesTopologyZone, + IngressControllerNamespace: o.KubernetesIngressControllerNamespace, + IngressControllerName: o.KubernetesIngressControllerName, + MaxZoneSkew: &o.KubernetesZoneMaxSkew, IngressStatusFromService: o.KubernetesStatusFromService, KubernetesApplicationAnnotationLabelKey: o.KubernetesApplicationAnnotationLabelKey, } From a493a7900b7b1ca5ebe2e74dc1b93ee32e717317 Mon Sep 17 00:00:00 2001 From: Mustafa Abdelrahman Date: Fri, 23 Oct 2026 20:48:34 +0200 Subject: [PATCH 3/8] fix(kubernetes): close race, silent-misconfig, and validation gaps in zone-aware gate Signed-off-by: Mustafa Abdelrahman --- dataclients/kubernetes/clusterclient.go | 10 +++++++--- dataclients/kubernetes/clusterstate.go | 1 + dataclients/kubernetes/kube.go | 1 + 3 files changed, 9 insertions(+), 3 deletions(-) diff --git a/dataclients/kubernetes/clusterclient.go b/dataclients/kubernetes/clusterclient.go index 8ee43fe0f9..2641b48252 100644 --- a/dataclients/kubernetes/clusterclient.go +++ b/dataclients/kubernetes/clusterclient.go @@ -16,6 +16,7 @@ import ( "sort" "strconv" "strings" + "sync/atomic" "time" log "github.com/sirupsen/logrus" @@ -84,7 +85,7 @@ type clusterClient struct { loggedMissingRouteGroups bool routeGroupValidator *definitions.RouteGroupValidator ingressValidator *definitions.IngressV1Validator - distributedAcrossZones bool + distributedAcrossZones atomic.Bool ingressControllerNamespace string ingressControllerName string maxZoneSkew int @@ -201,6 +202,9 @@ func newClusterClient(o Options, apiURL, ingCls, rgCls string, quit <-chan struc c.ingressControllerName = name 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 @@ -772,13 +776,13 @@ func (c *clusterClient) fetchClusterState() (*clusterState, error) { log.Errorf("failed to update ingress status: %v", err) } - c.distributedAcrossZones = state.ingressControllerDistributedAcrossZones(c.ingressControllerNamespace, c.ingressControllerName, c.maxZoneSkew) + c.distributedAcrossZones.Store(state.ingressControllerDistributedAcrossZones(c.ingressControllerNamespace, c.ingressControllerName, c.maxZoneSkew)) return state, nil } func (c *clusterClient) IngressControllerDistributedAcrossZones() bool { - return c.distributedAcrossZones + return c.distributedAcrossZones.Load() } func parseNamespaceName(resource string) (string, string, error) { diff --git a/dataclients/kubernetes/clusterstate.go b/dataclients/kubernetes/clusterstate.go index e22aa2f044..6c06df1c66 100644 --- a/dataclients/kubernetes/clusterstate.go +++ b/dataclients/kubernetes/clusterstate.go @@ -31,6 +31,7 @@ func (state *clusterState) ingressControllerDistributedAcrossZones(namespace, na eps, ok := state.endpointSlices[newResourceID(namespace, name)] if !ok { + log.Errorf("ingress controller endpointslice %s/%s not found; zone-aware routing is disabled until -kubernetes-ingress-controller-namespace/-kubernetes-ingress-controller-name match a real service", namespace, name) return false } diff --git a/dataclients/kubernetes/kube.go b/dataclients/kubernetes/kube.go index 57b2e98a4b..8284ff4d74 100644 --- a/dataclients/kubernetes/kube.go +++ b/dataclients/kubernetes/kube.go @@ -310,6 +310,7 @@ type Options struct { // zone and the global minimum across all zones for the ingress controller, following // Kubernetes TopologySpreadConstraints maxSkew semantics. // nil means use DefaultMaxZoneSkew. A pointer to 0 means no skew is tolerated. + // Negative values are rejected: New returns an error if *MaxZoneSkew < 0. MaxZoneSkew *int } From ec341e7da1df45eb86992d5c85362357f39863e6 Mon Sep 17 00:00:00 2001 From: Mustafa Abdelrahman Date: Fri, 23 Oct 2026 21:24:21 +0200 Subject: [PATCH 4/8] naming is hard :) Signed-off-by: Mustafa Abdelrahman --- dataclients/kubernetes/clusterclient.go | 8 ++++---- dataclients/kubernetes/clusterstate.go | 4 ++-- dataclients/kubernetes/clusterstate_test.go | 6 +++--- routesrv/export_test.go | 4 ++-- routesrv/polling.go | 22 ++++++++++----------- routesrv/routesrv.go | 22 ++++++++++----------- routesrv/routesrv_test.go | 20 +++++++++---------- 7 files changed, 43 insertions(+), 43 deletions(-) diff --git a/dataclients/kubernetes/clusterclient.go b/dataclients/kubernetes/clusterclient.go index 2641b48252..2aab8bb7d2 100644 --- a/dataclients/kubernetes/clusterclient.go +++ b/dataclients/kubernetes/clusterclient.go @@ -85,7 +85,7 @@ type clusterClient struct { loggedMissingRouteGroups bool routeGroupValidator *definitions.RouteGroupValidator ingressValidator *definitions.IngressV1Validator - distributedAcrossZones atomic.Bool + balancedAcrossZones atomic.Bool ingressControllerNamespace string ingressControllerName string maxZoneSkew int @@ -776,13 +776,13 @@ func (c *clusterClient) fetchClusterState() (*clusterState, error) { log.Errorf("failed to update ingress status: %v", err) } - c.distributedAcrossZones.Store(state.ingressControllerDistributedAcrossZones(c.ingressControllerNamespace, c.ingressControllerName, c.maxZoneSkew)) + c.balancedAcrossZones.Store(state.isServiceBalancedAcrossZones(c.ingressControllerNamespace, c.ingressControllerName, c.maxZoneSkew)) return state, nil } -func (c *clusterClient) IngressControllerDistributedAcrossZones() bool { - return c.distributedAcrossZones.Load() +func (c *clusterClient) IsIngressControllerBalancedAcrossZones() bool { + return c.balancedAcrossZones.Load() } func parseNamespaceName(resource string) (string, string, error) { diff --git a/dataclients/kubernetes/clusterstate.go b/dataclients/kubernetes/clusterstate.go index 6c06df1c66..313b6de3b2 100644 --- a/dataclients/kubernetes/clusterstate.go +++ b/dataclients/kubernetes/clusterstate.go @@ -24,14 +24,14 @@ type clusterState struct { enableEndpointSlices bool } -func (state *clusterState) ingressControllerDistributedAcrossZones(namespace, name string, maxZoneSkew int) bool { +func (state *clusterState) isServiceBalancedAcrossZones(namespace, name string, maxZoneSkew int) bool { if !state.enableEndpointSlices { return false } eps, ok := state.endpointSlices[newResourceID(namespace, name)] if !ok { - log.Errorf("ingress controller endpointslice %s/%s not found; zone-aware routing is disabled until -kubernetes-ingress-controller-namespace/-kubernetes-ingress-controller-name match a real service", namespace, name) + log.Errorf("endpointslice %s/%s not found; cannot determine zone balance", namespace, name) return false } diff --git a/dataclients/kubernetes/clusterstate_test.go b/dataclients/kubernetes/clusterstate_test.go index be20c137c4..d47b4f9e1a 100644 --- a/dataclients/kubernetes/clusterstate_test.go +++ b/dataclients/kubernetes/clusterstate_test.go @@ -101,7 +101,7 @@ func makeSkipperEndpointSlice(zoneCounts map[string]int) *skipperEndpointSlice { return eps } -func TestIngressControllerDistributedAcrossZones(t *testing.T) { +func TestIsServiceBalancedAcrossZones(t *testing.T) { const ( ns = DefaultIngressControllerNamespace name = DefaultIngressControllerName @@ -239,9 +239,9 @@ func TestIngressControllerDistributedAcrossZones(t *testing.T) { endpointSlices: endpointSlices, } - got := state.ingressControllerDistributedAcrossZones(lookupNS, lookupName, tt.maxZoneSkew) + got := state.isServiceBalancedAcrossZones(lookupNS, lookupName, tt.maxZoneSkew) if got != tt.want { - t.Errorf("ingressControllerDistributedAcrossZones(%q, %q, %d) = %v, want %v", + t.Errorf("isServiceBalancedAcrossZones(%q, %q, %d) = %v, want %v", lookupNS, lookupName, tt.maxZoneSkew, got, tt.want) } }) diff --git a/routesrv/export_test.go b/routesrv/export_test.go index e9be8e0c38..c461bf36df 100644 --- a/routesrv/export_test.go +++ b/routesrv/export_test.go @@ -10,8 +10,8 @@ func SetNow(rs *RouteServer, now func() time.Time) { rs.poller.b.now = now } -func SetDistributed(rs *RouteServer, v bool) { - rs.poller.isIngressControllerDistributedAcrossZones = func() bool { return v } +func SetBalancedAcrossZones(rs *RouteServer, v bool) { + rs.poller.isIngressControllerBalancedAcrossZones = func() bool { return v } } func (rs *RouteServer) ListenAndServe() (err error) { diff --git a/routesrv/polling.go b/routesrv/polling.go index ca8d65a0c0..17d27d1628 100644 --- a/routesrv/polling.go +++ b/routesrv/polling.go @@ -42,9 +42,9 @@ type poller struct { cloneRoute []*eskip.Clone // visibility - tracer ot.Tracer - metrics metrics.Metrics - isIngressControllerDistributedAcrossZones func() bool + tracer ot.Tracer + metrics metrics.Metrics + isIngressControllerBalancedAcrossZones func() bool } func (p *poller) poll(wg *sync.WaitGroup) { @@ -56,9 +56,9 @@ func (p *poller) poll(wg *sync.WaitGroup) { p.setGaugeToCurrentTime("polling_started_timestamp") var ( - lastRoutesByID map[string]string - lastRoutes []*eskip.Route - lastDistributed bool + lastRoutesByID map[string]string + lastRoutes []*eskip.Route + lastBalanced bool ) for { @@ -76,13 +76,13 @@ func (p *poller) poll(wg *sync.WaitGroup) { ) } else { - distributed := p.isIngressControllerDistributedAcrossZones() + balanced := p.isIngressControllerBalancedAcrossZones() routesChanged := hasChanged(lastRoutes, routes) - distributedChanged := distributed != lastDistributed + balancedChanged := balanced != lastBalanced - if routesChanged || distributedChanged { + if routesChanged || balancedChanged { lastRoutes = routes - lastDistributed = distributed + lastBalanced = balanced routes = p.process(routes) routesCount := len(routes) @@ -93,7 +93,7 @@ func (p *poller) poll(wg *sync.WaitGroup) { p.handleEmptyRoutes() case routesCount > 0: var zoneAwareRoutes map[string][]*eskip.Route - if distributed { + if balanced { zoneAwareRoutes = filterRoutesByZone(routes) } else { log.Debug("skipper not covering all zones, falling back to global routing table") diff --git a/routesrv/routesrv.go b/routesrv/routesrv.go index f3c8d3d66e..95794bfa4d 100644 --- a/routesrv/routesrv.go +++ b/routesrv/routesrv.go @@ -180,17 +180,17 @@ func New(opts skipper.Options) (*RouteServer, error) { } rs.poller = &poller{ - client: dataclient, - timeout: opts.SourcePollTimeout, - b: b, - quit: make(chan struct{}), - defaultFilters: opts.DefaultFilters, - editRoute: opts.EditRoute, - cloneRoute: opts.CloneRoute, - oauth2Config: oauthConfig, - tracer: tracer, - metrics: m, - isIngressControllerDistributedAcrossZones: dataclient.ClusterClient.IngressControllerDistributedAcrossZones, + client: dataclient, + timeout: opts.SourcePollTimeout, + b: b, + quit: make(chan struct{}), + defaultFilters: opts.DefaultFilters, + editRoute: opts.EditRoute, + cloneRoute: opts.CloneRoute, + oauth2Config: oauthConfig, + tracer: tracer, + metrics: m, + isIngressControllerBalancedAcrossZones: dataclient.ClusterClient.IsIngressControllerBalancedAcrossZones, } rs.wg = &sync.WaitGroup{} diff --git a/routesrv/routesrv_test.go b/routesrv/routesrv_test.go index 5132ef22fb..afbdae1e8c 100644 --- a/routesrv/routesrv_test.go +++ b/routesrv/routesrv_test.go @@ -1140,7 +1140,7 @@ func TestRoutesWithZone(t *testing.T) { KubernetesURL: ks.URL, KubernetesEnableEndpointslices: true, }) - routesrv.SetDistributed(rs, true) + routesrv.SetBalancedAcrossZones(rs, true) rs.StartUpdates() defer rs.StopUpdates() @@ -1175,7 +1175,7 @@ func TestEtagWithZoneAwareRoutingFallback(t *testing.T) { KubernetesURL: ks.URL, KubernetesEnableEndpointslices: true, }) - routesrv.SetDistributed(rs, true) + routesrv.SetBalancedAcrossZones(rs, true) rs.StartUpdates() defer rs.StopUpdates() @@ -1230,7 +1230,7 @@ func TestZoneAwareRoutesGzip(t *testing.T) { KubernetesURL: ks.URL, KubernetesEnableEndpointslices: true, }) - routesrv.SetDistributed(rs, true) + routesrv.SetBalancedAcrossZones(rs, true) rs.StartUpdates() defer rs.StopUpdates() @@ -1279,7 +1279,7 @@ func TestZoneAwareXCount(t *testing.T) { KubernetesURL: ks.URL, KubernetesEnableEndpointslices: true, }) - routesrv.SetDistributed(rs, true) + routesrv.SetBalancedAcrossZones(rs, true) rs.StartUpdates() defer rs.StopUpdates() require.NoError(t, tl.WaitFor(routesrv.LogRoutesInitialized, waitTimeout)) @@ -1312,7 +1312,7 @@ func TestZoneAwareEtag(t *testing.T) { KubernetesURL: ks.URL, KubernetesEnableEndpointslices: true, }) - routesrv.SetDistributed(rs, true) + routesrv.SetBalancedAcrossZones(rs, true) rs.StartUpdates() defer rs.StopUpdates() require.NoError(t, tl.WaitFor(routesrv.LogRoutesInitialized, waitTimeout)) @@ -1356,7 +1356,7 @@ func TestZoneAwareConditionalGetEtag(t *testing.T) { KubernetesURL: ks.URL, KubernetesEnableEndpointslices: true, }) - routesrv.SetDistributed(rs, true) + routesrv.SetBalancedAcrossZones(rs, true) rs.StartUpdates() defer rs.StopUpdates() require.NoError(t, tl.WaitFor(routesrv.LogRoutesInitialized, waitTimeout)) @@ -1404,7 +1404,7 @@ func TestZoneAwareConditionalGetLastModified(t *testing.T) { KubernetesURL: ks.URL, KubernetesEnableEndpointslices: true, }) - routesrv.SetDistributed(rs, true) + routesrv.SetBalancedAcrossZones(rs, true) rs.StartUpdates() defer rs.StopUpdates() require.NoError(t, tl.WaitFor(routesrv.LogRoutesInitialized, waitTimeout)) @@ -1442,7 +1442,7 @@ func TestZoneAwareFallbackToAllRoutesWhenBelowThreshold(t *testing.T) { KubernetesURL: ks.URL, KubernetesEnableEndpointslices: true, }) - routesrv.SetDistributed(rs, true) + routesrv.SetBalancedAcrossZones(rs, true) rs.StartUpdates() defer rs.StopUpdates() require.NoError(t, tl.WaitFor(routesrv.LogRoutesInitialized, waitTimeout)) @@ -1484,7 +1484,7 @@ func TestZoneAwareRoutesNotServedWhenNotDistributed(t *testing.T) { KubernetesURL: ks.URL, KubernetesEnableEndpointslices: true, }) - routesrv.SetDistributed(rs, false) + routesrv.SetBalancedAcrossZones(rs, false) rs.StartUpdates() defer rs.StopUpdates() require.NoError(t, tl.WaitFor(routesrv.LogRoutesInitialized, waitTimeout)) @@ -1529,7 +1529,7 @@ func TestXCountShouldBeSameForZoneAwareAndZoneUnawareRoutes(t *testing.T) { KubernetesURL: ks.URL, KubernetesEnableEndpointslices: true, }) - routesrv.SetDistributed(rs, true) + routesrv.SetBalancedAcrossZones(rs, true) rs.StartUpdates() defer rs.StopUpdates() require.NoError(t, tl.WaitFor(routesrv.LogRoutesInitialized, waitTimeout)) From 78123d42ef6f87a084a3229d6113245d871f1fa3 Mon Sep 17 00:00:00 2001 From: Mustafa Abdelrahman Date: Fri, 23 Oct 2026 23:26:45 +0200 Subject: [PATCH 5/8] cover as much cases as we can with and without mock Signed-off-by: Mustafa Abdelrahman --- dataclients/kubernetes/clusterclient_test.go | 36 +++ routesrv/eskipbytes_test.go | 74 +++-- routesrv/routesrv_test.go | 304 ++++++++++++++++++ .../custom-ingress-controller-balanced.yaml | 39 +++ .../skipper-ingress-balanced.yaml | 39 +++ .../skipper-ingress-missing-zone.yaml | 48 +++ .../skipper-ingress-skewed.yaml | 63 ++++ 7 files changed, 578 insertions(+), 25 deletions(-) create mode 100644 routesrv/testdata/zone-aware-traffic/custom-ingress-controller-balanced.yaml create mode 100644 routesrv/testdata/zone-aware-traffic/skipper-ingress-balanced.yaml create mode 100644 routesrv/testdata/zone-aware-traffic/skipper-ingress-missing-zone.yaml create mode 100644 routesrv/testdata/zone-aware-traffic/skipper-ingress-skewed.yaml diff --git a/dataclients/kubernetes/clusterclient_test.go b/dataclients/kubernetes/clusterclient_test.go index 4aeb6b4304..9dc19ef408 100644 --- a/dataclients/kubernetes/clusterclient_test.go +++ b/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 TestMissingIngressControllerEndpointSliceLoggedEveryPoll(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 = "endpointslice kube-system/skipper-ingress not found; cannot determine 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 { diff --git a/routesrv/eskipbytes_test.go b/routesrv/eskipbytes_test.go index 9464a25dc0..d76d27bf7b 100644 --- a/routesrv/eskipbytes_test.go +++ b/routesrv/eskipbytes_test.go @@ -107,13 +107,11 @@ func BenchmarkFormatAndSetNoChange(b *testing.B) { // BenchmarkFormatAndSetOneZoneChanged measures the realistic steady-state // poll tick where exactly one zone's route set changes while a sibling zone -// stays byte-identical (e.g. one AZ gains or loses an endpoint). The global -// routes slice is also perturbed each call (appending one marker route), -// since formatAndSet only re-evaluates zones when the global hash changes; -// this mirrors how a real endpoint change always shows up in the global -// table too. Only eu-central-1a's zone routes differ between iterations; -// eu-central-1b is never mutated and is used to assert it was left alone. -// This isolates the cost of handling a single changed zone from the +// stays byte-identical (e.g. one AZ gains or loses an endpoint), and the +// global table is unchanged. Only eu-central-1a's zone routes differ between +// iterations; eu-central-1b is never mutated and is used by +// TestFormatAndSetOnlyTouchesChangedZone to assert it was left alone. This +// isolates the cost of handling a single changed zone from the // all-zones-changed case already covered by BenchmarkFormatAndSet, and is // the scenario the all-or-nothing vs. per-zone update strategy differ on // the most. @@ -134,19 +132,13 @@ func BenchmarkFormatAndSetOneZoneChanged(b *testing.B) { altZoneRoutes[i] = &rc } - baseRoutes := routes - markerA := &eskip.Route{Id: "benchMarkerA"} - markerB := &eskip.Route{Id: "benchMarkerB"} - toggle := false b.ReportAllocs() for b.Loop() { if toggle { zoneAware[changedZone] = altZoneRoutes - routes = append(baseRoutes, markerA) } else { zoneAware[changedZone] = originalZoneRoutes - routes = append(baseRoutes, markerB) } toggle = !toggle e.formatAndSet(routes, zoneAware) @@ -156,9 +148,7 @@ func BenchmarkFormatAndSetOneZoneChanged(b *testing.B) { // TestFormatAndSetOnlyTouchesChangedZone verifies that formatAndSet updates // only the zone whose content actually changed: an untouched zone keeps its // exact Etag/Last-Modified/compressed bytes across a call where a sibling -// zone changes. In this pre-zone-balance-gate contract, formatAndSet only -// re-evaluates zones when the global routes hash changes, so the global -// routes must change too for the zone rebuild to run at all. +// zone changes. func TestFormatAndSetOnlyTouchesChangedZone(t *testing.T) { e := &eskipBytes{now: time.Now} routes := []*eskip.Route{{Id: "r1"}} @@ -173,15 +163,13 @@ func TestFormatAndSetOnlyTouchesChangedZone(t *testing.T) { z1DataBefore := e.zoneDataCompressed["z1"] z1ModBefore := e.zoneLastModified["z1"] - // global routes change (required to trigger a rebuild at all), but // z1's own content stays identical; only z2 changes - routes2 := []*eskip.Route{{Id: "r1"}, {Id: "rGlobal"}} zoneMap2 := map[string][]*eskip.Route{ "z1": {{Id: "r1"}}, "z2": {{Id: "r2"}}, } - _, _, _, updated = e.formatAndSet(routes2, zoneMap2) - require.True(t, updated, "updated should be true since global routes changed") + _, _, _, updated = e.formatAndSet(routes, zoneMap2) + require.True(t, updated, "updated should be true since z2 changed") assert.Equal(t, z1HashBefore, e.zoneHash["z1"], "z1 hash must be untouched when only z2 changes") assert.Equal(t, z1DataBefore, e.zoneDataCompressed["z1"], "z1 compressed bytes must be untouched when only z2 changes") @@ -205,14 +193,13 @@ func TestFormatAndSetRemovesDisappearedZone(t *testing.T) { _, ok := e.zoneDataCompressed["z2"] require.True(t, ok, "z2 must be present after the first call") - // global routes change (required to trigger a rebuild at all), and z2 - // disappears entirely (e.g. its last endpoint went away) - routes2 := []*eskip.Route{{Id: "r1"}, {Id: "rGlobal"}} + // z2 disappears entirely (e.g. the balance gate closed, or its last + // endpoint went away) zoneMap2 := map[string][]*eskip.Route{ "z1": {{Id: "r1"}}, } - _, _, _, updated = e.formatAndSet(routes2, zoneMap2) - assert.True(t, updated, "updated should be true since global routes changed") + _, _, _, updated = e.formatAndSet(routes, zoneMap2) + assert.True(t, updated, "updated should be true since z2 disappeared") _, ok = e.zoneDataCompressed["z2"] assert.False(t, ok, "z2 must be removed from zoneDataCompressed so ServeHTTP falls back to global") @@ -223,3 +210,40 @@ func TestFormatAndSetRemovesDisappearedZone(t *testing.T) { _, ok = e.zoneLastModified["z2"] assert.False(t, ok, "z2 must be removed from zoneLastModified") } + +// TestFormatAndSetDetectsZoneOnlyChange verifies that formatAndSet reports +// updated == true when only the zone map changes (global routes identical), +// and also when only the global routes change (zone map identical) +func TestFormatAndSetDetectsZoneOnlyChange(t *testing.T) { + e := &eskipBytes{now: time.Now} + routes := []*eskip.Route{{Id: "r1"}} + zoneA := map[string][]*eskip.Route{"z1": {{Id: "r1"}}} + + _, _, initialized, updated := e.formatAndSet(routes, zoneA) + require.True(t, initialized, "initialized should be true after fresh start") + require.True(t, updated, "updated should be true after fresh start") + + // same global routes, same zone map -> no update + _, _, initialized, updated = e.formatAndSet(routes, zoneA) + require.False(t, initialized, "initialized should be false with no change") + require.False(t, updated, "updated should be false with no change") + + // same global routes, DIFFERENT zone map -> updated must be true + zoneB := map[string][]*eskip.Route{"z1": {{Id: "r1"}}, "z2": {{Id: "r1"}}} + _, _, initialized, updated = e.formatAndSet(routes, zoneB) + assert.False(t, initialized, "should always be false after first call") + assert.True(t, updated, "should be true when zone only routes change") + + // DIFFERENT global routes, same zone map -> updated must be true + routes2 := []*eskip.Route{{Id: "r2"}} + _, _, initialized, updated = e.formatAndSet(routes2, zoneB) + assert.False(t, initialized, "should always be false after first call") + assert.True(t, updated, "should be true when global only routes change") + + // also true when both changes + routes3 := []*eskip.Route{{Id: "r3"}} + zoneC := map[string][]*eskip.Route{"z1": {{Id: "r1"}}, "z3": {{Id: "r1"}}} + _, _, initialized, updated = e.formatAndSet(routes3, zoneC) + assert.False(t, initialized, "should always be false after first call") + assert.True(t, updated, "should be true when global only routes change") +} diff --git a/routesrv/routesrv_test.go b/routesrv/routesrv_test.go index afbdae1e8c..1950f08614 100644 --- a/routesrv/routesrv_test.go +++ b/routesrv/routesrv_test.go @@ -1516,6 +1516,310 @@ func TestZoneAwareRoutesNotServedWhenNotDistributed(t *testing.T) { assert.True(t, eskip.EqLists(fullRoutes, gotGzip), "gzip zone response must also equal global response when not distributed") } +// TestZoneAwareRoutesUpdateOnBalanceTransitionAlone verifies that flipping the +// ingress-controller balanced-across-zones state alone (no underlying route +// change) still triggers a republish: the global /routes table must stay +// byte-identical, but /routes/{zone} must switch from the global fallback to +// the zone-filtered view. +func TestZoneAwareRoutesUpdateOnBalanceTransitionAlone(t *testing.T) { + defer tl.Reset() + + ks, _ := newKubeServer(t, loadKubeYAML(t, "testdata/zone-aware-traffic/all-zones-3-addr.yaml")) + ks.Start() + defer ks.Close() + rs := newRouteServerWithOptions(t, skipper.Options{ + SourcePollTimeout: pollInterval, + Kubernetes: true, + KubernetesURL: ks.URL, + KubernetesEnableEndpointslices: true, + }) + routesrv.SetBalancedAcrossZones(rs, false) // start NOT balanced + rs.StartUpdates() + defer rs.StopUpdates() + require.NoError(t, tl.WaitFor(routesrv.LogRoutesInitialized, waitTimeout)) + + globalBefore := getRoutes(rs) + require.Equal(t, http.StatusOK, globalBefore.Code) + zoneBefore := getZoneAwareRoutes(rs, "eu-central-1a") + require.Equal(t, http.StatusOK, zoneBefore.Code) + + // not balanced -> /routes/{zone} must equal /routes (global fallback) + assert.Equal(t, globalBefore.Body.String(), zoneBefore.Body.String(), + "zone response must equal global response while not balanced") + + // flip balance state WITHOUT changing any kubernetes fixture / routes + routesrv.SetBalancedAcrossZones(rs, true) + + // a second LogRoutesUpdated must fire even though the underlying + // kubernetes routes never changed + require.NoError(t, tl.WaitForN(routesrv.LogRoutesUpdated, 2, waitTimeout)) + + globalAfter := getRoutes(rs) + require.Equal(t, http.StatusOK, globalAfter.Code) + zoneAfter := getZoneAwareRoutes(rs, "eu-central-1a") + require.Equal(t, http.StatusOK, zoneAfter.Code) + + // global table is unchanged + assert.Equal(t, globalBefore.Body.String(), globalAfter.Body.String(), + "global routes must not change when only the balance state changes") + + // but the zone-specific view must now differ from global (zone filtering kicked in) + assert.NotEqual(t, globalAfter.Body.String(), zoneAfter.Body.String(), + "zone response must diverge from global once balanced across zones") + + assert.True(t, zoneAfter.Header().Get("X-Zone-Aware") == "true", "header should be set when we serve zone aware traffic") +} + +// TestZoneAwareRoutingEndToEndWithRealEndpointSlice exercises the real +// fetchClusterState -> isServiceBalancedAcrossZones -> +// IsIngressControllerBalancedAcrossZones -> poller pipeline, with no +// SetBalancedAcrossZones mock, using real EndpointSlice fixtures for the +// ingress controller (skipper-ingress) itself. It also proves the backend +// service's own zone distribution (per-route, minEndpointsByZone threshold) +// and the ingress-controller's cluster-wide balance gate are independent +// mechanisms that don't interfere with each other. +func TestZoneAwareRoutingEndToEndWithRealEndpointSlice(t *testing.T) { + for _, tc := range []struct { + name string + backendFixture string + ingressFixture string + enableEPSlices bool + wantZoneFiltered bool + }{ + { + name: "balanced", + backendFixture: "all-zones-3-addr.yaml", + ingressFixture: "skipper-ingress-balanced.yaml", + enableEPSlices: true, + wantZoneFiltered: true, + }, + { + name: "ingressSkewed", + backendFixture: "all-zones-3-addr.yaml", + ingressFixture: "skipper-ingress-skewed.yaml", + enableEPSlices: true, + wantZoneFiltered: false, + }, + { + name: "ingressMissingZone", + backendFixture: "all-zones-3-addr.yaml", + ingressFixture: "skipper-ingress-missing-zone.yaml", + enableEPSlices: true, + wantZoneFiltered: false, + }, + { + name: "backend imbalanced", + backendFixture: "all-zones-2-addr.yaml", + ingressFixture: "skipper-ingress-balanced.yaml", + enableEPSlices: true, + wantZoneFiltered: false, + }, + { + name: "both unbalanced", + backendFixture: "all-zones-2-addr.yaml", + ingressFixture: "skipper-ingress-skewed.yaml", + enableEPSlices: true, + wantZoneFiltered: false, + }, + { + name: "endpointslices disabled", + backendFixture: "all-zones-3-addr.yaml", + ingressFixture: "skipper-ingress-balanced.yaml", + enableEPSlices: false, + wantZoneFiltered: false, + }, + } { + t.Run(tc.name, func(t *testing.T) { + defer tl.Reset() + ks, _ := newKubeServer(t, + loadKubeYAML(t, "testdata/zone-aware-traffic/"+tc.backendFixture), + loadKubeYAML(t, "testdata/zone-aware-traffic/"+tc.ingressFixture), + ) + ks.Start() + defer ks.Close() + + rs := newRouteServerWithOptions(t, skipper.Options{ + SourcePollTimeout: pollInterval, + Kubernetes: true, + KubernetesURL: ks.URL, + KubernetesEnableEndpointslices: tc.enableEPSlices, + }) + rs.StartUpdates() + defer rs.StopUpdates() + require.NoError(t, tl.WaitFor(routesrv.LogRoutesInitialized, waitTimeout)) + + global := getRoutes(rs) + require.Equal(t, http.StatusOK, global.Code) + zone := getZoneAwareRoutes(rs, "eu-central-1a") + require.Equal(t, http.StatusOK, zone.Code) + + if tc.wantZoneFiltered { + assert.NotEqual(t, global.Body.String(), zone.Body.String(), + "expected zone-filtered response to diverge from global") + } else { + assert.Equal(t, global.Body.String(), zone.Body.String(), + "expected fallback to global routes") + } + }) + } +} + +// TestZoneAwareRoutingBackendImbalanceDoesNotCloseGate verifies that a +// backend service with fewer than minEndpointsByZone endpoints per zone +// still falls back to all its own endpoints for that route (the existing, +// independent per-route mechanism), while the ingress-controller balance +// gate itself stays open: /routes/{zone} is still served as the zone-aware +// endpoint, it just happens to equal the all-endpoints fallback for this +// particular route. +func TestZoneAwareRoutingBackendImbalanceDoesNotCloseGate(t *testing.T) { + defer tl.Reset() + + ks, _ := newKubeServer(t, + loadKubeYAML(t, "testdata/zone-aware-traffic/all-zones-2-addr.yaml"), + loadKubeYAML(t, "testdata/zone-aware-traffic/skipper-ingress-balanced.yaml"), + ) + ks.Start() + defer ks.Close() + + rs := newRouteServerWithOptions(t, skipper.Options{ + SourcePollTimeout: pollInterval, + Kubernetes: true, + KubernetesURL: ks.URL, + KubernetesEnableEndpointslices: true, + }) + rs.StartUpdates() + defer rs.StopUpdates() + require.NoError(t, tl.WaitFor(routesrv.LogRoutesInitialized, waitTimeout)) + + zoneResp := getZoneAwareRoutes(rs, "eu-central-1a") + require.Equal(t, http.StatusOK, zoneResp.Code) + + got, err := eskip.Parse(zoneResp.Body.String()) + require.NoError(t, err) + want := parseEskipFixture(t, "testdata/zone-aware-traffic/all-zones-2-addr.eskip") + assert.True(t, eskip.EqLists(got, want), + "route with <3 endpoints/zone must still carry all its own endpoints even though ingress controller is balanced") +} + +// TestZoneAwareRoutingUsesConfiguredIngressControllerNamespaceAndName verifies +// that KubernetesIngressControllerNamespace/KubernetesIngressControllerName +// flow through skipper.Options -> kubernetes.Options -> newClusterClient and +// are actually used to look up the ingress-controller EndpointSlice, instead +// of silently falling back to the kube-system/skipper-ingress defaults. +func TestZoneAwareRoutingUsesConfiguredIngressControllerNamespaceAndName(t *testing.T) { + defer tl.Reset() + + ks, _ := newKubeServer(t, + loadKubeYAML(t, "testdata/zone-aware-traffic/all-zones-3-addr.yaml"), + loadKubeYAML(t, "testdata/zone-aware-traffic/custom-ingress-controller-balanced.yaml"), + ) + ks.Start() + defer ks.Close() + + rs := newRouteServerWithOptions(t, skipper.Options{ + SourcePollTimeout: pollInterval, + Kubernetes: true, + KubernetesURL: ks.URL, + KubernetesEnableEndpointslices: true, + KubernetesIngressControllerNamespace: "custom-ns", + KubernetesIngressControllerName: "my-custom-ingress", + }) + rs.StartUpdates() + defer rs.StopUpdates() + require.NoError(t, tl.WaitFor(routesrv.LogRoutesInitialized, waitTimeout)) + + global := getRoutes(rs) + require.Equal(t, http.StatusOK, global.Code) + zone := getZoneAwareRoutes(rs, "eu-central-1a") + require.Equal(t, http.StatusOK, zone.Code) + assert.NotEqual(t, global.Body.String(), zone.Body.String(), + "configured namespace/name must be used to find the ingress controller endpointslice") +} + +// TestZoneAwareRoutingUsesConfiguredMaxZoneSkew verifies that +// KubernetesZoneMaxSkew flows through to the real balance check: a skew that +// the default (1) would reject must be tolerated once configured explicitly. +func TestZoneAwareRoutingUsesConfiguredMaxZoneSkew(t *testing.T) { + defer tl.Reset() + + ks, _ := newKubeServer(t, + loadKubeYAML(t, "testdata/zone-aware-traffic/all-zones-3-addr.yaml"), + loadKubeYAML(t, "testdata/zone-aware-traffic/skipper-ingress-skewed.yaml"), // skew=2 + ) + ks.Start() + defer ks.Close() + + rs := newRouteServerWithOptions(t, skipper.Options{ + SourcePollTimeout: pollInterval, + Kubernetes: true, + KubernetesURL: ks.URL, + KubernetesEnableEndpointslices: true, + KubernetesZoneMaxSkew: 2, // default of 1 would reject this fixture's skew of 2 + }) + rs.StartUpdates() + defer rs.StopUpdates() + require.NoError(t, tl.WaitFor(routesrv.LogRoutesInitialized, waitTimeout)) + + global := getRoutes(rs) + require.Equal(t, http.StatusOK, global.Code) + zone := getZoneAwareRoutes(rs, "eu-central-1a") + require.Equal(t, http.StatusOK, zone.Code) + assert.NotEqual(t, global.Body.String(), zone.Body.String(), + "configured MaxZoneSkew=2 must tolerate a skew of 2 that the default of 1 would reject") +} + +// TestZoneAwareRoutingReactsToRealEndpointSliceChange proves that +// fetchClusterState recomputes the ingress-controller balance gate fresh on +// every poll against live data, not just that a mocked flip propagates +// (see TestZoneAwareRoutesUpdateOnBalanceTransitionAlone). The ingress +// controller's own EndpointSlice is mutated mid-run via the same +// handler.set(...) pattern used for route changes elsewhere in this file. +func TestZoneAwareRoutingReactsToRealEndpointSliceChange(t *testing.T) { + defer tl.Reset() + + ks, handler := newKubeServer(t, + loadKubeYAML(t, "testdata/zone-aware-traffic/all-zones-3-addr.yaml"), + loadKubeYAML(t, "testdata/zone-aware-traffic/skipper-ingress-skewed.yaml"), // start skewed + ) + ks.Start() + defer ks.Close() + + rs := newRouteServerWithOptions(t, skipper.Options{ + SourcePollTimeout: pollInterval, + Kubernetes: true, + KubernetesURL: ks.URL, + KubernetesEnableEndpointslices: true, + }) + rs.StartUpdates() + defer rs.StopUpdates() + require.NoError(t, tl.WaitFor(routesrv.LogRoutesInitialized, waitTimeout)) + + globalBefore := getRoutes(rs) + require.Equal(t, http.StatusOK, globalBefore.Code) + zoneBefore := getZoneAwareRoutes(rs, "eu-central-1a") + require.Equal(t, http.StatusOK, zoneBefore.Code) + assert.Equal(t, globalBefore.Body.String(), zoneBefore.Body.String(), + "skewed skipper-ingress must start in global fallback") + + // mutate the REAL kubernetes API data: skewed -> balanced, no mock involved + handler.set(newKubeAPI(t, + loadKubeYAML(t, "testdata/zone-aware-traffic/all-zones-3-addr.yaml"), + loadKubeYAML(t, "testdata/zone-aware-traffic/skipper-ingress-balanced.yaml"), + )) + + require.NoError(t, tl.WaitForN(routesrv.LogRoutesUpdated, 2, waitTimeout)) + + globalAfter := getRoutes(rs) + require.Equal(t, http.StatusOK, globalAfter.Code) + zoneAfter := getZoneAwareRoutes(rs, "eu-central-1a") + require.Equal(t, http.StatusOK, zoneAfter.Code) + + assert.Equal(t, globalBefore.Body.String(), globalAfter.Body.String(), + "global routes must be unaffected by the ingress-controller endpointslice becoming balanced") + assert.NotEqual(t, globalAfter.Body.String(), zoneAfter.Body.String(), + "zone response must diverge from global once the real endpointslice becomes balanced") +} + func TestXCountShouldBeSameForZoneAwareAndZoneUnawareRoutes(t *testing.T) { defer tl.Reset() diff --git a/routesrv/testdata/zone-aware-traffic/custom-ingress-controller-balanced.yaml b/routesrv/testdata/zone-aware-traffic/custom-ingress-controller-balanced.yaml new file mode 100644 index 0000000000..79b41016d6 --- /dev/null +++ b/routesrv/testdata/zone-aware-traffic/custom-ingress-controller-balanced.yaml @@ -0,0 +1,39 @@ +apiVersion: v1 +kind: Service +metadata: + labels: + app: my-custom-ingress + name: my-custom-ingress + namespace: custom-ns +spec: + clusterIP: 10.3.190.98 + ports: + - name: web + port: 80 + protocol: TCP + targetPort: 9090 + selector: + app: my-custom-ingress + type: ClusterIP +--- +apiVersion: v1 +kind: EndpointSlice +metadata: + labels: + kubernetes.io/service-name: my-custom-ingress + name: my-custom-ingress-foo + namespace: custom-ns +endpoints: + - addresses: + - 10.9.5.1 + zone: eu-central-1a + - addresses: + - 10.9.5.2 + zone: eu-central-1b + - addresses: + - 10.9.5.3 + zone: eu-central-1c +ports: + - name: web + port: 9090 + protocol: TCP diff --git a/routesrv/testdata/zone-aware-traffic/skipper-ingress-balanced.yaml b/routesrv/testdata/zone-aware-traffic/skipper-ingress-balanced.yaml new file mode 100644 index 0000000000..a5b420254a --- /dev/null +++ b/routesrv/testdata/zone-aware-traffic/skipper-ingress-balanced.yaml @@ -0,0 +1,39 @@ +apiVersion: v1 +kind: Service +metadata: + labels: + app: skipper-ingress + name: skipper-ingress + namespace: kube-system +spec: + clusterIP: 10.3.190.99 + ports: + - name: web + port: 80 + protocol: TCP + targetPort: 9090 + selector: + app: skipper-ingress + type: ClusterIP +--- +apiVersion: v1 +kind: EndpointSlice +metadata: + labels: + kubernetes.io/service-name: skipper-ingress + name: skipper-ingress-foo + namespace: kube-system +endpoints: + - addresses: + - 10.9.0.1 + zone: eu-central-1a + - addresses: + - 10.9.0.2 + zone: eu-central-1b + - addresses: + - 10.9.0.3 + zone: eu-central-1c +ports: + - name: web + port: 9090 + protocol: TCP diff --git a/routesrv/testdata/zone-aware-traffic/skipper-ingress-missing-zone.yaml b/routesrv/testdata/zone-aware-traffic/skipper-ingress-missing-zone.yaml new file mode 100644 index 0000000000..c071cbdaaf --- /dev/null +++ b/routesrv/testdata/zone-aware-traffic/skipper-ingress-missing-zone.yaml @@ -0,0 +1,48 @@ +apiVersion: v1 +kind: Service +metadata: + labels: + app: skipper-ingress + name: skipper-ingress + namespace: kube-system +spec: + clusterIP: 10.3.190.99 + ports: + - name: web + port: 80 + protocol: TCP + targetPort: 9090 + selector: + app: skipper-ingress + type: ClusterIP +--- +apiVersion: v1 +kind: EndpointSlice +metadata: + labels: + kubernetes.io/service-name: skipper-ingress + name: skipper-ingress-foo + namespace: kube-system +endpoints: + - addresses: + - 10.9.0.1 + zone: eu-central-1a + - addresses: + - 10.9.0.2 + zone: eu-central-1a + - addresses: + - 10.9.0.3 + zone: eu-central-1a + - addresses: + - 10.9.1.1 + zone: eu-central-1b + - addresses: + - 10.9.1.2 + zone: eu-central-1b + - addresses: + - 10.9.1.3 + zone: eu-central-1b +ports: + - name: web + port: 9090 + protocol: TCP diff --git a/routesrv/testdata/zone-aware-traffic/skipper-ingress-skewed.yaml b/routesrv/testdata/zone-aware-traffic/skipper-ingress-skewed.yaml new file mode 100644 index 0000000000..5148d16e3b --- /dev/null +++ b/routesrv/testdata/zone-aware-traffic/skipper-ingress-skewed.yaml @@ -0,0 +1,63 @@ +apiVersion: v1 +kind: Service +metadata: + labels: + app: skipper-ingress + name: skipper-ingress + namespace: kube-system +spec: + clusterIP: 10.3.190.99 + ports: + - name: web + port: 80 + protocol: TCP + targetPort: 9090 + selector: + app: skipper-ingress + type: ClusterIP +--- +apiVersion: v1 +kind: EndpointSlice +metadata: + labels: + kubernetes.io/service-name: skipper-ingress + name: skipper-ingress-foo + namespace: kube-system +endpoints: + - addresses: + - 10.9.0.1 + zone: eu-central-1a + - addresses: + - 10.9.0.2 + zone: eu-central-1a + - addresses: + - 10.9.0.3 + zone: eu-central-1a + - addresses: + - 10.9.0.4 + zone: eu-central-1a + - addresses: + - 10.9.0.5 + zone: eu-central-1a + - addresses: + - 10.9.1.1 + zone: eu-central-1b + - addresses: + - 10.9.1.2 + zone: eu-central-1b + - addresses: + - 10.9.1.3 + zone: eu-central-1b + - addresses: + - 10.9.2.1 + zone: eu-central-1c + - addresses: + - 10.9.2.2 + zone: eu-central-1c + - addresses: + - 10.9.2.3 + zone: eu-central-1c +ports: + - name: web + port: 9090 + protocol: TCP From 91c67c2cf11ec0ac1204f65a05013d410d210bbc Mon Sep 17 00:00:00 2001 From: Mustafa Abdelrahman Date: Mon, 26 Oct 2026 20:33:56 +0100 Subject: [PATCH 6/8] Rebase and add another test and benchmark MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Benchmark results compared against origin/master ``` goos: linux goarch: amd64 pkg: github.com/zalando/skipper/routesrv cpu: Intel(R) Core(TM) Ultra 7 265U │ before │ after │ │ sec/op │ sec/op vs base │ FormatAndSet-14 268.7m ± ∞ ¹ 277.8m ± ∞ ¹ ~ (p=0.548 n=5) FormatAndSetNoChange-14 60.45m ± ∞ ¹ 63.66m ± ∞ ¹ ~ (p=0.151 n=5) FormatAndSetOneZoneChanged-14 232.6m ± ∞ ¹ 227.9m ± ∞ ¹ ~ (p=0.222 n=5) geomean 155.8m 159.1m +2.16% ¹ need >= 6 samples for confidence interval at level 0.95 │ before │ after │ │ retained_MB │ retained_MB vs base │ FormatAndSet-14 1.986 ± ∞ ¹ 1.986 ± ∞ ¹ ~ (p=1.000 n=5) ² ¹ need >= 6 samples for confidence interval at level 0.95 ² all samples are equal │ before │ after │ │ B/op │ B/op vs base │ FormatAndSet-14 227.8Mi ± ∞ ¹ 227.8Mi ± ∞ ¹ +0.00% (p=0.008 n=5) FormatAndSetNoChange-14 86.34Mi ± ∞ ¹ 86.35Mi ± ∞ ¹ ~ (p=0.421 n=5) FormatAndSetOneZoneChanged-14 229.7Mi ± ∞ ¹ 217.8Mi ± ∞ ¹ -5.15% (p=0.008 n=5) geomean 165.3Mi 162.4Mi -1.75% ¹ need >= 6 samples for confidence interval at level 0.95 │ before │ after │ │ allocs/op │ allocs/op vs base │ FormatAndSet-14 2.380M ± ∞ ¹ 2.380M ± ∞ ¹ +0.00% (p=0.008 n=5) FormatAndSetNoChange-14 700.0k ± ∞ ¹ 700.0k ± ∞ ¹ ~ (p=1.000 n=5) FormatAndSetOneZoneChanged-14 2.483M ± ∞ ¹ 2.268M ± ∞ ¹ -8.65% (p=0.008 n=5) geomean 1.605M 1.558M -2.97% ¹ need >= 6 samples for confidence interval at level 0.95 ``` Signed-off-by: Mustafa Abdelrahman --- routesrv/eskipbytes.go | 82 +++++++------ routesrv/eskipbytes_test.go | 233 +++++++++++++++++++++++++++--------- 2 files changed, 222 insertions(+), 93 deletions(-) diff --git a/routesrv/eskipbytes.go b/routesrv/eskipbytes.go index df96759ad0..d88535cb91 100644 --- a/routesrv/eskipbytes.go +++ b/routesrv/eskipbytes.go @@ -77,12 +77,15 @@ type eskipBytes struct { // An update is triggered when either the global routes hash or any // per-zone route hash differs from the previously stored value, so a // change in zone topology alone (without route changes) is detected -// independently. Each zone's routes are formatted and hashed exactly -// once per call; only the zones whose hash actually changed are -// recompressed and get a fresh Last-Modified. Unchanged zones keep -// their existing Etag/Last-Modified/compressed bytes untouched, and -// zones that disappeared entirely are removed so requests for them -// fall back to the global routing table. +// independently. The per-zone format+hash work only runs when the global +// routes changed, or when zones appeared/disappeared entirely (the +// balance gate flipping); zoneAwareRoutes cannot change content while +// routes and zone presence both stay the same, since it is a pure filter +// of routes. When it does run, only the zones whose hash actually +// changed are recompressed and get a fresh Last-Modified. Unchanged +// zones keep their existing Etag/Last-Modified/compressed bytes +// untouched, and zones that disappeared entirely are removed so +// requests for them fall back to the global routing table. func (e *eskipBytes) formatAndSet(routes []*eskip.Route, zoneAwareRoutes map[string][]*eskip.Route) (_ int, _ string, initialized bool, updated bool) { buf := &bytes.Buffer{} @@ -111,37 +114,46 @@ func (e *eskipBytes) formatAndSet(routes []*eskip.Route, zoneAwareRoutes map[str e.zoneLastModified = make(map[string]time.Time) } - // Format and hash every zone exactly once, so unchanged zones - // never need to be recompressed below. + // zoneAwareRoutes is a pure filter of routes: it can only differ from + // last time if routes changed (globalUpdated), or if zones just + // appeared/disappeared entirely (the balance gate flipping, detected + // cheaply by comparing presence, not content). If neither is true, + // zoneAwareRoutes is guaranteed identical to last call, and the + // per-zone format+hash work below is skipped entirely. + presenceChanged := (len(e.zoneHash) > 0) != (len(zoneAwareRoutes) > 0) zoneUpdated := false - newZoneHash := make(map[string]string, len(zoneAwareRoutes)) - for zone, zoneRoutes := range zoneAwareRoutes { - zoneBuf := &bytes.Buffer{} - eskip.Fprint(zoneBuf, eskip.PrettyPrintInfo{Pretty: false, IndentStr: ""}, zoneRoutes...) - zoneData := zoneBuf.Bytes() - zoneHash := fmt.Sprintf("%x", sha256.Sum256(zoneData)) - newZoneHash[zone] = zoneHash - - if e.zoneHash[zone] == zoneHash { - continue // this zone's content is unchanged; leave it alone - } - zoneUpdated = true - e.zoneHash[zone] = zoneHash - e.zoneDataCompressed[zone] = e.compressLocked(zoneData) - e.zoneCount[zone] = len(zoneRoutes) - e.zoneLastModified[zone] = now - } - for zone := range e.zoneHash { - if _, ok := newZoneHash[zone]; !ok { - // zone no longer present at all (e.g. the gate closed, or the - // zone's last endpoint disappeared) - remove it so ServeHTTP - // falls back to the global routing table for it instead of - // serving stale data. + if globalUpdated || presenceChanged { + // Format and hash every zone exactly once, so unchanged zones + // never need to be recompressed below. + newZoneHash := make(map[string]string, len(zoneAwareRoutes)) + for zone, zoneRoutes := range zoneAwareRoutes { + zoneBuf := &bytes.Buffer{} + eskip.Fprint(zoneBuf, eskip.PrettyPrintInfo{Pretty: false, IndentStr: ""}, zoneRoutes...) + zoneData := zoneBuf.Bytes() + zoneHash := fmt.Sprintf("%x", sha256.Sum256(zoneData)) + newZoneHash[zone] = zoneHash + + if e.zoneHash[zone] == zoneHash { + continue // this zone's content is unchanged; leave it alone + } zoneUpdated = true - delete(e.zoneHash, zone) - delete(e.zoneDataCompressed, zone) - delete(e.zoneCount, zone) - delete(e.zoneLastModified, zone) + e.zoneHash[zone] = zoneHash + e.zoneDataCompressed[zone] = e.compressLocked(zoneData) + e.zoneCount[zone] = len(zoneRoutes) + e.zoneLastModified[zone] = now + } + for zone := range e.zoneHash { + if _, ok := newZoneHash[zone]; !ok { + // zone no longer present at all (e.g. the gate closed, or + // the zone's last endpoint disappeared) - remove it so + // ServeHTTP falls back to the global routing table for it + // instead of serving stale data. + zoneUpdated = true + delete(e.zoneHash, zone) + delete(e.zoneDataCompressed, zone) + delete(e.zoneCount, zone) + delete(e.zoneLastModified, zone) + } } } diff --git a/routesrv/eskipbytes_test.go b/routesrv/eskipbytes_test.go index d76d27bf7b..2287f47cfa 100644 --- a/routesrv/eskipbytes_test.go +++ b/routesrv/eskipbytes_test.go @@ -106,15 +106,7 @@ func BenchmarkFormatAndSetNoChange(b *testing.B) { } // BenchmarkFormatAndSetOneZoneChanged measures the realistic steady-state -// poll tick where exactly one zone's route set changes while a sibling zone -// stays byte-identical (e.g. one AZ gains or loses an endpoint), and the -// global table is unchanged. Only eu-central-1a's zone routes differ between -// iterations; eu-central-1b is never mutated and is used by -// TestFormatAndSetOnlyTouchesChangedZone to assert it was left alone. This -// isolates the cost of handling a single changed zone from the -// all-zones-changed case already covered by BenchmarkFormatAndSet, and is -// the scenario the all-or-nothing vs. per-zone update strategy differ on -// the most. +// poll tick where exactly one zone's endpoints change. func BenchmarkFormatAndSetOneZoneChanged(b *testing.B) { zones := []string{"eu-central-1a", "eu-central-1b", "eu-central-1c"} routes := benchRoutes(20000, 3, zones) @@ -123,25 +115,55 @@ func BenchmarkFormatAndSetOneZoneChanged(b *testing.B) { e := &eskipBytes{now: time.Now} e.formatAndSet(routes, zoneAware) // prime, so the first measured call already has a baseline to diff against - const changedZone = "eu-central-1a" - originalZoneRoutes := zoneAware[changedZone] - altZoneRoutes := make([]*eskip.Route, len(originalZoneRoutes)) - for i, r := range originalZoneRoutes { + // altRoutes differs from routes only in eu-central-1a's endpoint + // addresses; eu-central-1b/c endpoints are untouched + altRoutes := make([]*eskip.Route, len(routes)) + for i, r := range routes { rc := *r - rc.Filters = append([]*eskip.Filter{{Name: "setRequestHeader", Args: []any{"X-Bench", "alt"}}}, r.Filters...) - altZoneRoutes[i] = &rc + eps := make([]*eskip.LBEndpoint, len(r.LBEndpoints)) + for j, ep := range r.LBEndpoints { + epCopy := *ep + if epCopy.Zone == "eu-central-1a" { + epCopy.Address += "-alt" + } + eps[j] = &epCopy + } + rc.LBEndpoints = eps + altRoutes[i] = &rc } + altZoneAware := filterRoutesByZone(altRoutes) toggle := false b.ReportAllocs() for b.Loop() { if toggle { - zoneAware[changedZone] = altZoneRoutes + e.formatAndSet(altRoutes, altZoneAware) } else { - zoneAware[changedZone] = originalZoneRoutes + e.formatAndSet(routes, zoneAware) + } + toggle = !toggle + } +} + +// BenchmarkFormatAndSetBalanceFlip measures the cost of the balance gate +// flipping while routes stays completely unchanged +func BenchmarkFormatAndSetBalanceFlip(b *testing.B) { + zones := []string{"eu-central-1a", "eu-central-1b", "eu-central-1c"} + routes := benchRoutes(20000, 3, zones) + zoneAware := filterRoutesByZone(routes) + + e := &eskipBytes{now: time.Now} + e.formatAndSet(routes, zoneAware) // prime with the gate open + + toggle := true + b.ReportAllocs() + for b.Loop() { + if toggle { + e.formatAndSet(routes, nil) // gate closes + } else { + e.formatAndSet(routes, zoneAware) // gate reopens } toggle = !toggle - e.formatAndSet(routes, zoneAware) } } @@ -151,11 +173,19 @@ func BenchmarkFormatAndSetOneZoneChanged(b *testing.B) { // zone changes. func TestFormatAndSetOnlyTouchesChangedZone(t *testing.T) { e := &eskipBytes{now: time.Now} - routes := []*eskip.Route{{Id: "r1"}} - zoneMap := map[string][]*eskip.Route{ - "z1": {{Id: "r1"}}, - "z2": {{Id: "r1"}}, + z1eps := []*eskip.LBEndpoint{ + {Address: "http://z1-ep1", Zone: "z1"}, + {Address: "http://z1-ep2", Zone: "z1"}, + {Address: "http://z1-ep3", Zone: "z1"}, } + z2v1eps := []*eskip.LBEndpoint{ + {Address: "http://z2-ep1-v1", Zone: "z2"}, + {Address: "http://z2-ep2-v1", Zone: "z2"}, + {Address: "http://z2-ep3-v1", Zone: "z2"}, + } + routes := []*eskip.Route{{Id: "r1", BackendType: eskip.LBBackend, LBEndpoints: append(append([]*eskip.LBEndpoint{}, z1eps...), z2v1eps...)}} + zoneMap := filterRoutesByZone(routes) + _, _, _, updated := e.formatAndSet(routes, zoneMap) require.True(t, updated) @@ -163,15 +193,19 @@ func TestFormatAndSetOnlyTouchesChangedZone(t *testing.T) { z1DataBefore := e.zoneDataCompressed["z1"] z1ModBefore := e.zoneLastModified["z1"] - // z1's own content stays identical; only z2 changes - zoneMap2 := map[string][]*eskip.Route{ - "z1": {{Id: "r1"}}, - "z2": {{Id: "r2"}}, + // change z2 routes + z2v2eps := []*eskip.LBEndpoint{ + {Address: "http://z2-ep1-v2", Zone: "z2"}, + {Address: "http://z2-ep2-v2", Zone: "z2"}, + {Address: "http://z2-ep3-v2", Zone: "z2"}, } - _, _, _, updated = e.formatAndSet(routes, zoneMap2) - require.True(t, updated, "updated should be true since z2 changed") + routes2 := []*eskip.Route{{Id: "r1", BackendType: eskip.LBBackend, LBEndpoints: append(append([]*eskip.LBEndpoint{}, z1eps...), z2v2eps...)}} + zoneMap2 := filterRoutesByZone(routes2) - assert.Equal(t, z1HashBefore, e.zoneHash["z1"], "z1 hash must be untouched when only z2 changes") + _, _, _, updated = e.formatAndSet(routes2, zoneMap2) + require.True(t, updated, "updated should be true since the route (and therefore z2) changed") + + assert.Equal(t, z1HashBefore, e.zoneHash["z1"], "z1 hash must be untouched when only z2's endpoints change") assert.Equal(t, z1DataBefore, e.zoneDataCompressed["z1"], "z1 compressed bytes must be untouched when only z2 changes") assert.Equal(t, z1ModBefore, e.zoneLastModified["z1"], "z1 Last-Modified must be untouched when only z2 changes") assert.NotEqual(t, z1HashBefore, e.zoneHash["z2"], "z2 hash must differ from z1's since z2's content changed") @@ -183,23 +217,20 @@ func TestFormatAndSetOnlyTouchesChangedZone(t *testing.T) { // of stale per-zone data. func TestFormatAndSetRemovesDisappearedZone(t *testing.T) { e := &eskipBytes{now: time.Now} - routes := []*eskip.Route{{Id: "r1"}} - zoneMap := map[string][]*eskip.Route{ - "z1": {{Id: "r1"}}, - "z2": {{Id: "r1"}}, - } + routes := []*eskip.Route{{Id: "r1", BackendType: eskip.LBBackend, LBEndpoints: []*eskip.LBEndpoint{ + {Address: "http://z1-ep", Zone: "z1"}, + {Address: "http://z2-ep", Zone: "z2"}, + }}} + zoneMap := filterRoutesByZone(routes) + _, _, _, updated := e.formatAndSet(routes, zoneMap) require.True(t, updated) _, ok := e.zoneDataCompressed["z2"] require.True(t, ok, "z2 must be present after the first call") - // z2 disappears entirely (e.g. the balance gate closed, or its last - // endpoint went away) - zoneMap2 := map[string][]*eskip.Route{ - "z1": {{Id: "r1"}}, - } - _, _, _, updated = e.formatAndSet(routes, zoneMap2) - assert.True(t, updated, "updated should be true since z2 disappeared") + // the balance gate closes; routes is unchanged, zoneAwareRoutes becomes nil + _, _, _, updated = e.formatAndSet(routes, nil) + assert.True(t, updated, "updated should be true since the balance gate closed") _, ok = e.zoneDataCompressed["z2"] assert.False(t, ok, "z2 must be removed from zoneDataCompressed so ServeHTTP falls back to global") @@ -212,38 +243,124 @@ func TestFormatAndSetRemovesDisappearedZone(t *testing.T) { } // TestFormatAndSetDetectsZoneOnlyChange verifies that formatAndSet reports -// updated == true when only the zone map changes (global routes identical), -// and also when only the global routes change (zone map identical) +// updated == true when only the balance gate flips (global routes +// identical), and also when only the global routes change (gate state +// unchanged). func TestFormatAndSetDetectsZoneOnlyChange(t *testing.T) { e := &eskipBytes{now: time.Now} routes := []*eskip.Route{{Id: "r1"}} - zoneA := map[string][]*eskip.Route{"z1": {{Id: "r1"}}} - _, _, initialized, updated := e.formatAndSet(routes, zoneA) + _, _, initialized, updated := e.formatAndSet(routes, nil) require.True(t, initialized, "initialized should be true after fresh start") require.True(t, updated, "updated should be true after fresh start") - // same global routes, same zone map -> no update - _, _, initialized, updated = e.formatAndSet(routes, zoneA) + // same routes, still not balanced -> no update + _, _, initialized, updated = e.formatAndSet(routes, nil) require.False(t, initialized, "initialized should be false with no change") require.False(t, updated, "updated should be false with no change") - // same global routes, DIFFERENT zone map -> updated must be true - zoneB := map[string][]*eskip.Route{"z1": {{Id: "r1"}}, "z2": {{Id: "r1"}}} - _, _, initialized, updated = e.formatAndSet(routes, zoneB) + // same routes, balance gate opens -> updated must be true + zoneA := map[string][]*eskip.Route{"z1": {{Id: "r1"}}} + _, _, initialized, updated = e.formatAndSet(routes, zoneA) assert.False(t, initialized, "should always be false after first call") - assert.True(t, updated, "should be true when zone only routes change") + assert.True(t, updated, "should be true when the balance gate opens") - // DIFFERENT global routes, same zone map -> updated must be true + // same routes, gate stays open -> no update + _, _, initialized, updated = e.formatAndSet(routes, zoneA) + assert.False(t, updated, "should be false when neither routes nor the gate changed") + + // DIFFERENT global routes, gate stays open -> updated must be true routes2 := []*eskip.Route{{Id: "r2"}} - _, _, initialized, updated = e.formatAndSet(routes2, zoneB) + _, _, initialized, updated = e.formatAndSet(routes2, zoneA) assert.False(t, initialized, "should always be false after first call") assert.True(t, updated, "should be true when global only routes change") - // also true when both changes - routes3 := []*eskip.Route{{Id: "r3"}} - zoneC := map[string][]*eskip.Route{"z1": {{Id: "r1"}}, "z3": {{Id: "r1"}}} - _, _, initialized, updated = e.formatAndSet(routes3, zoneC) + // gate closes again, routes unchanged -> updated must be true + _, _, initialized, updated = e.formatAndSet(routes2, nil) assert.False(t, initialized, "should always be false after first call") - assert.True(t, updated, "should be true when global only routes change") + assert.True(t, updated, "should be true when the balance gate closes") +} + +// TestFormatAndSetHandlesSimultaneousRouteChangeAndBalanceFlip verifies +// that formatAndSet is correct when routes change and the balance gate +// flips on the very same call - a timing coincidence the existing tests +// never exercise together, since each varies exactly one of the two +// signals per call. Both globalUpdated and presenceChanged are true at +// once, in both flip directions. +func TestFormatAndSetHandlesSimultaneousRouteChangeAndBalanceFlip(t *testing.T) { + e := &eskipBytes{now: time.Now} + routes := []*eskip.Route{{Id: "r1"}} + + _, _, _, updated := e.formatAndSet(routes, nil) + require.True(t, updated, "initial call must report updated") + + // gate opens AND routes change on the same call + routes2 := []*eskip.Route{{Id: "r2"}} + zoneA := map[string][]*eskip.Route{"z1": {{Id: "r2"}}} + _, _, _, updated = e.formatAndSet(routes2, zoneA) + require.True(t, updated, "updated must be true when routes and the gate change together") + _, ok := e.zoneDataCompressed["z1"] + assert.True(t, ok, "z1 must be populated when the gate opens, even though routes changed in the same call") + + // gate closes AND routes change again on the same call + routes3 := []*eskip.Route{{Id: "r3"}} + _, _, _, updated = e.formatAndSet(routes3, nil) + require.True(t, updated, "updated must be true when routes and the gate change together") + _, ok = e.zoneDataCompressed["z1"] + assert.False(t, ok, "z1 must be removed when the gate closes, even though routes changed in the same call") +} + +// TestFormatAndSetTreatsEmptyZoneMapLikeNil verifies that a non-nil but +// empty zoneAwareRoutes map (e.g. filterRoutesByZone found no zone labels +// at all, even though balanced is true) is treated identically to nil by +// the presence check: it must not be mistaken for "zones present" and +// must not trigger spurious zone work. +func TestFormatAndSetTreatsEmptyZoneMapLikeNil(t *testing.T) { + e := &eskipBytes{now: time.Now} + routes := []*eskip.Route{{Id: "r1"}} + + _, _, _, updated := e.formatAndSet(routes, map[string][]*eskip.Route{}) + require.True(t, updated, "initial call must report updated") + assert.Empty(t, e.zoneHash, "no zone should be recorded when the zone map is empty") + + // same routes, still an empty (non-nil) zone map -> no update + _, _, _, updated = e.formatAndSet(routes, map[string][]*eskip.Route{}) + assert.False(t, updated, "updated must be false: neither routes nor zone presence changed") + + // same routes, nil this time instead of empty map -> must still be a no-op, + // since nil and empty are equivalent "no zones" states + _, _, _, updated = e.formatAndSet(routes, nil) + assert.False(t, updated, "nil and an empty map must be treated as the same presence state") +} + +// TestFormatAndSetHandlesRapidBalanceOscillation verifies that the gate +// flipping true->false->true across consecutive polls leaves no stale +// data behind: each reopen must repopulate zones from scratch, not reuse +// leftovers from before the gate closed. +func TestFormatAndSetHandlesRapidBalanceOscillation(t *testing.T) { + e := &eskipBytes{now: time.Now} + routes := []*eskip.Route{{Id: "r1"}} + zoneA := map[string][]*eskip.Route{"z1": {{Id: "r1"}}} + + // poll 1: gate opens + _, _, _, updated := e.formatAndSet(routes, zoneA) + require.True(t, updated) + firstHash, ok := e.zoneHash["z1"] + require.True(t, ok, "z1 must be populated after the gate opens") + + // poll 2: gate closes + _, _, _, updated = e.formatAndSet(routes, nil) + require.True(t, updated, "updated must be true when the gate closes") + _, ok = e.zoneHash["z1"] + require.False(t, ok, "z1 must be removed when the gate closes") + + // poll 3: gate reopens, same routes/zone content as poll 1 + _, _, _, updated = e.formatAndSet(routes, zoneA) + require.True(t, updated, "updated must be true when the gate reopens") + secondHash, ok := e.zoneHash["z1"] + require.True(t, ok, "z1 must be repopulated after the gate reopens") + assert.Equal(t, firstHash, secondHash, "z1's hash must match poll 1's, since its content is identical") + + zdata1 := e.zoneDataCompressed["z1"] + assert.NotEmpty(t, zdata1, "z1 must have compressed data after reopening, not a stale/empty entry") } From 5d8f4c937c408106579470772154973481211225 Mon Sep 17 00:00:00 2001 From: Mustafa Abdelrahman Date: Mon, 26 Oct 2026 21:09:21 +0100 Subject: [PATCH 7/8] update current test to check for global routing table hash Signed-off-by: Mustafa Abdelrahman --- routesrv/eskipbytes_test.go | 34 ++++++++++++++++++++++++++++++---- 1 file changed, 30 insertions(+), 4 deletions(-) diff --git a/routesrv/eskipbytes_test.go b/routesrv/eskipbytes_test.go index 2287f47cfa..635cc03d00 100644 --- a/routesrv/eskipbytes_test.go +++ b/routesrv/eskipbytes_test.go @@ -245,7 +245,9 @@ func TestFormatAndSetRemovesDisappearedZone(t *testing.T) { // TestFormatAndSetDetectsZoneOnlyChange verifies that formatAndSet reports // updated == true when only the balance gate flips (global routes // identical), and also when only the global routes change (gate state -// unchanged). +// unchanged). It also verifies that a pure balance-gate flip leaves the +// GLOBAL hash/Last-Modified/count/compressed bytes completely untouched, +// since only zoneAwareRoutes differs in that case, not routes itself. func TestFormatAndSetDetectsZoneOnlyChange(t *testing.T) { e := &eskipBytes{now: time.Now} routes := []*eskip.Route{{Id: "r1"}} @@ -254,19 +256,32 @@ func TestFormatAndSetDetectsZoneOnlyChange(t *testing.T) { require.True(t, initialized, "initialized should be true after fresh start") require.True(t, updated, "updated should be true after fresh start") + globalHashBefore := e.hash + globalModBefore := e.lastModified + globalCountBefore := e.count + globalDataBefore := e.zdata + require.NotEmpty(t, globalHashBefore) + require.NotEmpty(t, globalDataBefore) + // same routes, still not balanced -> no update _, _, initialized, updated = e.formatAndSet(routes, nil) require.False(t, initialized, "initialized should be false with no change") require.False(t, updated, "updated should be false with no change") - // same routes, balance gate opens -> updated must be true + // same routes, balance gate opens -> updated must be true, but the + // global hash/Last-Modified/count/compressed bytes must stay untouched, + // since routes itself never changed - only zoneAwareRoutes did. zoneA := map[string][]*eskip.Route{"z1": {{Id: "r1"}}} _, _, initialized, updated = e.formatAndSet(routes, zoneA) assert.False(t, initialized, "should always be false after first call") assert.True(t, updated, "should be true when the balance gate opens") + assert.Equal(t, globalHashBefore, e.hash, "global hash must be untouched by a pure gate flip") + assert.Equal(t, globalModBefore, e.lastModified, "global Last-Modified must be untouched by a pure gate flip") + assert.Equal(t, globalCountBefore, e.count, "global count must be untouched by a pure gate flip") + assert.Equal(t, globalDataBefore, e.zdata, "global compressed bytes must be untouched by a pure gate flip") // same routes, gate stays open -> no update - _, _, initialized, updated = e.formatAndSet(routes, zoneA) + _, _, _, updated = e.formatAndSet(routes, zoneA) assert.False(t, updated, "should be false when neither routes nor the gate changed") // DIFFERENT global routes, gate stays open -> updated must be true @@ -275,10 +290,21 @@ func TestFormatAndSetDetectsZoneOnlyChange(t *testing.T) { assert.False(t, initialized, "should always be false after first call") assert.True(t, updated, "should be true when global only routes change") - // gate closes again, routes unchanged -> updated must be true + // gate closes again, routes unchanged -> updated must be true, and the + // global hash/Last-Modified/count/compressed bytes (now reflecting + // routes2) must stay untouched by the gate closing. + globalHashAfterRoutes2 := e.hash + globalModAfterRoutes2 := e.lastModified + globalCountAfterRoutes2 := e.count + globalDataAfterRoutes2 := e.zdata + _, _, initialized, updated = e.formatAndSet(routes2, nil) assert.False(t, initialized, "should always be false after first call") assert.True(t, updated, "should be true when the balance gate closes") + assert.Equal(t, globalHashAfterRoutes2, e.hash, "global hash must be untouched when the gate closes") + assert.Equal(t, globalModAfterRoutes2, e.lastModified, "global Last-Modified must be untouched when the gate closes") + assert.Equal(t, globalCountAfterRoutes2, e.count, "global count must be untouched when the gate closes") + assert.Equal(t, globalDataAfterRoutes2, e.zdata, "global compressed bytes must be untouched when the gate closes") } // TestFormatAndSetHandlesSimultaneousRouteChangeAndBalanceFlip verifies From 3de4f887b0d42adce00bb4307645af6fe3f1e772 Mon Sep 17 00:00:00 2001 From: Mustafa Abdelrahman Date: Wed, 28 Oct 2026 21:12:09 +0100 Subject: [PATCH 8/8] Remove atomic bool, use labelSelectors instead of name/namespace combination to also catch canary Signed-off-by: Mustafa Abdelrahman --- config/config.go | 11 +- config/config_test.go | 2 - dataclients/kubernetes/clusterclient.go | 34 ++-- dataclients/kubernetes/clusterclient_test.go | 4 +- dataclients/kubernetes/clusterstate.go | 42 +++- dataclients/kubernetes/clusterstate_test.go | 185 ++++++++++++------ dataclients/kubernetes/kube.go | 36 ++-- routesrv/export_test.go | 2 +- routesrv/polling.go | 8 +- routesrv/routesrv.go | 22 +-- routesrv/routesrv_test.go | 29 ++- .../custom-selector-ingress-balanced.yaml | 40 ++++ .../skipper-ingress-balanced.yaml | 6 +- .../skipper-ingress-missing-zone.yaml | 6 +- .../skipper-ingress-skewed.yaml | 6 +- skipper.go | 18 +- 16 files changed, 284 insertions(+), 167 deletions(-) create mode 100644 routesrv/testdata/zone-aware-traffic/custom-selector-ingress-balanced.yaml diff --git a/config/config.go b/config/config.go index 597c60a34c..e11037d13b 100644 --- a/config/config.go +++ b/config/config.go @@ -192,8 +192,7 @@ type Config struct { KubernetesListChunkSize int `yaml:"kubernetes-list-chunk-size"` KubernetesEnableEndpointSlices bool `yaml:"enable-kubernetes-endpointslices"` KubernetesTopologyZone string `yaml:"kubernetes-topology-zone"` - KubernetesIngressControllerNamespace string `yaml:"kubernetes-ingress-controller-namespace"` - KubernetesIngressControllerName string `yaml:"kubernetes-ingress-controller-name"` + 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"` @@ -610,9 +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.StringVar(&cfg.KubernetesIngressControllerNamespace, "kubernetes-ingress-controller-namespace", kubernetes.DefaultIngressControllerNamespace, "namespace of the ingress controller service used for zone distribution checks") - flag.StringVar(&cfg.KubernetesIngressControllerName, "kubernetes-ingress-controller-name", kubernetes.DefaultIngressControllerName, "service name of the ingress controller used for zone distribution checks") - flag.IntVar(&cfg.KubernetesZoneMaxSkew, "kubernetes-zone-max-skew", kubernetes.DefaultMaxZoneSkew, "maximum permitted endpoint count difference between the most and least populated zone for the ingress controller, following Kubernetes TopologySpreadConstraints maxSkew semantics; 0 means no skew is tolerated") + 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 ..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") @@ -1140,8 +1138,7 @@ func (c *Config) ToOptions() skipper.Options { KubernetesListChunkSize: c.KubernetesListChunkSize, KubernetesEnableEndpointslices: c.KubernetesEnableEndpointSlices, KubernetesTopologyZone: c.KubernetesTopologyZone, - KubernetesIngressControllerNamespace: c.KubernetesIngressControllerNamespace, - KubernetesIngressControllerName: c.KubernetesIngressControllerName, + KubernetesDataPlaneLabelSelector: c.KubernetesDataPlaneLabelSelector.values, KubernetesZoneMaxSkew: c.KubernetesZoneMaxSkew, KubernetesEnableEastWest: c.KubernetesEnableEastWest, KubernetesEastWestDomain: c.KubernetesEastWestDomain, diff --git a/config/config_test.go b/config/config_test.go index 9ed23d3572..05fc49df49 100644 --- a/config/config_test.go +++ b/config/config_test.go @@ -203,8 +203,6 @@ func defaultConfig(with func(*Config)) *Config { MtlsAuthnAppendCA: false, MtlsAuthnCaFile: "", MtlsAuthnCA: x509.NewCertPool(), - KubernetesIngressControllerNamespace: kubernetes.DefaultIngressControllerNamespace, - KubernetesIngressControllerName: kubernetes.DefaultIngressControllerName, KubernetesZoneMaxSkew: kubernetes.DefaultMaxZoneSkew, } with(cfg) diff --git a/dataclients/kubernetes/clusterclient.go b/dataclients/kubernetes/clusterclient.go index 2aab8bb7d2..0612425345 100644 --- a/dataclients/kubernetes/clusterclient.go +++ b/dataclients/kubernetes/clusterclient.go @@ -16,7 +16,6 @@ import ( "sort" "strconv" "strings" - "sync/atomic" "time" log "github.com/sirupsen/logrus" @@ -82,13 +81,12 @@ type clusterClient struct { enableEndpointSlices bool listChunkSize int - loggedMissingRouteGroups bool - routeGroupValidator *definitions.RouteGroupValidator - ingressValidator *definitions.IngressV1Validator - balancedAcrossZones atomic.Bool - ingressControllerNamespace string - ingressControllerName string - maxZoneSkew int + loggedMissingRouteGroups bool + routeGroupValidator *definitions.RouteGroupValidator + ingressValidator *definitions.IngressV1Validator + balancedAcrossZones bool + dataPlaneLabelSelector map[string]string + maxZoneSkew int } var ( @@ -189,17 +187,11 @@ func newClusterClient(o Options, apiURL, ingCls, rgCls string, quit <-chan struc ingressStatusFromService: o.IngressStatusFromService, } - ns := o.IngressControllerNamespace - if ns == "" { - ns = DefaultIngressControllerNamespace - } - c.ingressControllerNamespace = ns - - name := o.IngressControllerName - if name == "" { - name = DefaultIngressControllerName + if o.DataPlaneLabelSelector != nil { + c.dataPlaneLabelSelector = o.DataPlaneLabelSelector + } else { + c.dataPlaneLabelSelector = DefaultDataPlaneLabelSelector } - c.ingressControllerName = name if o.MaxZoneSkew != nil { if *o.MaxZoneSkew < 0 { @@ -776,13 +768,13 @@ func (c *clusterClient) fetchClusterState() (*clusterState, error) { log.Errorf("failed to update ingress status: %v", err) } - c.balancedAcrossZones.Store(state.isServiceBalancedAcrossZones(c.ingressControllerNamespace, c.ingressControllerName, c.maxZoneSkew)) + c.balancedAcrossZones = state.isServiceBalancedAcrossZones(c.dataPlaneLabelSelector, c.maxZoneSkew) return state, nil } -func (c *clusterClient) IsIngressControllerBalancedAcrossZones() bool { - return c.balancedAcrossZones.Load() +func (c *clusterClient) IsBalancedDataPlane() bool { + return c.balancedAcrossZones } func parseNamespaceName(resource string) (string, string, error) { diff --git a/dataclients/kubernetes/clusterclient_test.go b/dataclients/kubernetes/clusterclient_test.go index 9dc19ef408..8356d7f423 100644 --- a/dataclients/kubernetes/clusterclient_test.go +++ b/dataclients/kubernetes/clusterclient_test.go @@ -87,7 +87,7 @@ func TestNewRejectsNegativeMaxZoneSkew(t *testing.T) { assert.ErrorContains(t, err, "MaxZoneSkew") } -func TestMissingIngressControllerEndpointSliceLoggedEveryPoll(t *testing.T) { +func TestMissingDataPlaneServiceLoggedEveryPoll(t *testing.T) { a, err := kubernetestest.NewAPI(kubernetestest.TestAPIOptions{}) require.NoError(t, err) @@ -111,7 +111,7 @@ func TestMissingIngressControllerEndpointSliceLoggedEveryPoll(t *testing.T) { require.NoError(t, err) } - const wantMessage = "endpointslice kube-system/skipper-ingress not found; cannot determine zone balance" + 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()) } diff --git a/dataclients/kubernetes/clusterstate.go b/dataclients/kubernetes/clusterstate.go index 313b6de3b2..67a3d958a1 100644 --- a/dataclients/kubernetes/clusterstate.go +++ b/dataclients/kubernetes/clusterstate.go @@ -24,22 +24,50 @@ type clusterState struct { enableEndpointSlices bool } -func (state *clusterState) isServiceBalancedAcrossZones(namespace, name string, maxZoneSkew int) 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 } - eps, ok := state.endpointSlices[newResourceID(namespace, name)] - if !ok { - log.Errorf("endpointslice %s/%s not found; cannot determine zone balance", namespace, name) + 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 _, ep := range eps.Endpoints { - if ep.Zone != "" { - zoneCounts[ep.Zone]++ + 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 { diff --git a/dataclients/kubernetes/clusterstate_test.go b/dataclients/kubernetes/clusterstate_test.go index d47b4f9e1a..d75932a838 100644 --- a/dataclients/kubernetes/clusterstate_test.go +++ b/dataclients/kubernetes/clusterstate_test.go @@ -101,148 +101,205 @@ func makeSkipperEndpointSlice(zoneCounts map[string]int) *skipperEndpointSlice { return eps } +// makeService builds a *service with the given namespace/name/labels, with +// no spec/status needed for label-matching purposes. +func makeService(namespace, name string, labels map[string]string) *service { + return &service{ + Meta: &definitions.Metadata{ + Namespace: namespace, + Name: name, + Labels: labels, + }, + } +} + func TestIsServiceBalancedAcrossZones(t *testing.T) { - const ( - ns = DefaultIngressControllerNamespace - name = DefaultIngressControllerName - ) + defaultSelector := map[string]string{"application": "skipper-ingress", "component": "ingress"} + defaultLabels := map[string]string{"application": "skipper-ingress", "component": "ingress"} for _, tt := range []struct { name string endpointSlicesEnable bool - endpointSlicePresent bool - zoneCounts map[string]int + services map[definitions.ResourceID]*service + endpointSlices map[definitions.ResourceID]*skipperEndpointSlice + labelSelector map[string]string maxZoneSkew int - customNS string - customName string want bool }{ { name: "endpoint slices disabled", endpointSlicesEnable: false, - endpointSlicePresent: true, - zoneCounts: map[string]int{"a": 3, "b": 3, "c": 3}, + services: map[definitions.ResourceID]*service{newResourceID("kube-system", "skipper-ingress"): makeService("kube-system", "skipper-ingress", defaultLabels)}, + endpointSlices: map[definitions.ResourceID]*skipperEndpointSlice{newResourceID("kube-system", "skipper-ingress"): makeSkipperEndpointSlice(map[string]int{"a": 3, "b": 3, "c": 3})}, + labelSelector: defaultSelector, + maxZoneSkew: 1, + want: false, + }, + { + name: "no service matches the selector", + endpointSlicesEnable: true, + services: map[definitions.ResourceID]*service{newResourceID("kube-system", "other"): makeService("kube-system", "other", map[string]string{"application": "other"})}, + endpointSlices: map[definitions.ResourceID]*skipperEndpointSlice{newResourceID("kube-system", "other"): makeSkipperEndpointSlice(map[string]int{"a": 3, "b": 3, "c": 3})}, + labelSelector: defaultSelector, maxZoneSkew: 1, want: false, }, { - name: "no endpointslice for ingress controller", + name: "service matches but has no endpointslice", endpointSlicesEnable: true, - endpointSlicePresent: false, - zoneCounts: map[string]int{"a": 3, "b": 3, "c": 3}, + services: map[definitions.ResourceID]*service{newResourceID("kube-system", "skipper-ingress"): makeService("kube-system", "skipper-ingress", defaultLabels)}, + endpointSlices: map[definitions.ResourceID]*skipperEndpointSlice{}, + labelSelector: defaultSelector, maxZoneSkew: 1, want: false, }, { name: "fewer than minZones zones", endpointSlicesEnable: true, - endpointSlicePresent: true, - zoneCounts: map[string]int{"a": 3, "b": 3}, + services: map[definitions.ResourceID]*service{newResourceID("kube-system", "skipper-ingress"): makeService("kube-system", "skipper-ingress", defaultLabels)}, + endpointSlices: map[definitions.ResourceID]*skipperEndpointSlice{newResourceID("kube-system", "skipper-ingress"): makeSkipperEndpointSlice(map[string]int{"a": 3, "b": 3})}, + labelSelector: defaultSelector, maxZoneSkew: 1, want: false, }, { name: "balanced zones, maxSkew=0", endpointSlicesEnable: true, - endpointSlicePresent: true, - zoneCounts: map[string]int{"a": 3, "b": 3, "c": 3}, + services: map[definitions.ResourceID]*service{newResourceID("kube-system", "skipper-ingress"): makeService("kube-system", "skipper-ingress", defaultLabels)}, + endpointSlices: map[definitions.ResourceID]*skipperEndpointSlice{newResourceID("kube-system", "skipper-ingress"): makeSkipperEndpointSlice(map[string]int{"a": 3, "b": 3, "c": 3})}, + labelSelector: defaultSelector, maxZoneSkew: 0, want: true, }, { name: "balanced zones, maxSkew=1", endpointSlicesEnable: true, - endpointSlicePresent: true, - zoneCounts: map[string]int{"a": 3, "b": 3, "c": 3}, + services: map[definitions.ResourceID]*service{newResourceID("kube-system", "skipper-ingress"): makeService("kube-system", "skipper-ingress", defaultLabels)}, + endpointSlices: map[definitions.ResourceID]*skipperEndpointSlice{newResourceID("kube-system", "skipper-ingress"): makeSkipperEndpointSlice(map[string]int{"a": 3, "b": 3, "c": 3})}, + labelSelector: defaultSelector, maxZoneSkew: 1, want: true, }, { name: "skew=1 within maxSkew=1", endpointSlicesEnable: true, - endpointSlicePresent: true, - zoneCounts: map[string]int{"a": 4, "b": 3, "c": 3}, + services: map[definitions.ResourceID]*service{newResourceID("kube-system", "skipper-ingress"): makeService("kube-system", "skipper-ingress", defaultLabels)}, + endpointSlices: map[definitions.ResourceID]*skipperEndpointSlice{newResourceID("kube-system", "skipper-ingress"): makeSkipperEndpointSlice(map[string]int{"a": 4, "b": 3, "c": 3})}, + labelSelector: defaultSelector, maxZoneSkew: 1, want: true, }, { name: "skew=1 exceeds maxSkew=0", endpointSlicesEnable: true, - endpointSlicePresent: true, - zoneCounts: map[string]int{"a": 4, "b": 3, "c": 3}, + services: map[definitions.ResourceID]*service{newResourceID("kube-system", "skipper-ingress"): makeService("kube-system", "skipper-ingress", defaultLabels)}, + endpointSlices: map[definitions.ResourceID]*skipperEndpointSlice{newResourceID("kube-system", "skipper-ingress"): makeSkipperEndpointSlice(map[string]int{"a": 4, "b": 3, "c": 3})}, + labelSelector: defaultSelector, maxZoneSkew: 0, want: false, }, { name: "skew=2 exceeds maxSkew=1", endpointSlicesEnable: true, - endpointSlicePresent: true, - zoneCounts: map[string]int{"a": 5, "b": 3, "c": 3}, + services: map[definitions.ResourceID]*service{newResourceID("kube-system", "skipper-ingress"): makeService("kube-system", "skipper-ingress", defaultLabels)}, + endpointSlices: map[definitions.ResourceID]*skipperEndpointSlice{newResourceID("kube-system", "skipper-ingress"): makeSkipperEndpointSlice(map[string]int{"a": 5, "b": 3, "c": 3})}, + labelSelector: defaultSelector, maxZoneSkew: 1, want: false, }, { name: "skew=2 within maxSkew=2", endpointSlicesEnable: true, - endpointSlicePresent: true, - zoneCounts: map[string]int{"a": 5, "b": 3, "c": 3}, + services: map[definitions.ResourceID]*service{newResourceID("kube-system", "skipper-ingress"): makeService("kube-system", "skipper-ingress", defaultLabels)}, + endpointSlices: map[definitions.ResourceID]*skipperEndpointSlice{newResourceID("kube-system", "skipper-ingress"): makeSkipperEndpointSlice(map[string]int{"a": 5, "b": 3, "c": 3})}, + labelSelector: defaultSelector, maxZoneSkew: 2, want: true, }, { - name: "custom namespace and name match", + name: "custom selector matches a differently-named/namespaced service", endpointSlicesEnable: true, - endpointSlicePresent: true, - zoneCounts: map[string]int{"a": 3, "b": 3, "c": 3}, + services: map[definitions.ResourceID]*service{newResourceID("custom-ns", "custom-ingress"): makeService("custom-ns", "custom-ingress", map[string]string{"team": "platform"})}, + endpointSlices: map[definitions.ResourceID]*skipperEndpointSlice{newResourceID("custom-ns", "custom-ingress"): makeSkipperEndpointSlice(map[string]int{"a": 3, "b": 3, "c": 3})}, + labelSelector: map[string]string{"team": "platform"}, maxZoneSkew: 1, - customNS: "custom-ns", - customName: "custom-ingress", want: true, }, { - name: "custom namespace and name mismatch", + name: "selector requires all labels to match (AND semantics)", endpointSlicesEnable: true, - endpointSlicePresent: true, - zoneCounts: map[string]int{"a": 3, "b": 3, "c": 3}, + services: map[definitions.ResourceID]*service{newResourceID("kube-system", "skipper-ingress"): makeService("kube-system", "skipper-ingress", map[string]string{"application": "skipper-ingress"})}, // missing "component" + endpointSlices: map[definitions.ResourceID]*skipperEndpointSlice{newResourceID("kube-system", "skipper-ingress"): makeSkipperEndpointSlice(map[string]int{"a": 3, "b": 3, "c": 3})}, + labelSelector: defaultSelector, maxZoneSkew: 1, - customNS: "wrong-ns", - customName: "wrong-ingress", want: false, }, + { + name: "empty selector matches no service", + endpointSlicesEnable: true, + services: map[definitions.ResourceID]*service{newResourceID("kube-system", "skipper-ingress"): makeService("kube-system", "skipper-ingress", defaultLabels)}, + endpointSlices: map[definitions.ResourceID]*skipperEndpointSlice{newResourceID("kube-system", "skipper-ingress"): makeSkipperEndpointSlice(map[string]int{"a": 3, "b": 3, "c": 3})}, + labelSelector: map[string]string{}, + maxZoneSkew: 1, + want: false, + }, + { + name: "two services matching the selector are aggregated into one balanced pool", + endpointSlicesEnable: true, + services: map[definitions.ResourceID]*service{ + newResourceID("region-a", "skipper-ingress"): makeService("region-a", "skipper-ingress", defaultLabels), + newResourceID("region-b", "skipper-ingress"): makeService("region-b", "skipper-ingress", defaultLabels), + }, + endpointSlices: map[definitions.ResourceID]*skipperEndpointSlice{ + newResourceID("region-a", "skipper-ingress"): makeSkipperEndpointSlice(map[string]int{"a": 2}), + newResourceID("region-b", "skipper-ingress"): makeSkipperEndpointSlice(map[string]int{"b": 2, "c": 2}), + }, + labelSelector: defaultSelector, + maxZoneSkew: 1, + want: true, // combined: a=2, b=2, c=2 -> balanced + }, + { + name: "two matched services combine into an unbalanced pool even though each alone looks fine", + endpointSlicesEnable: true, + services: map[definitions.ResourceID]*service{ + newResourceID("region-a", "skipper-ingress"): makeService("region-a", "skipper-ingress", defaultLabels), + newResourceID("region-b", "skipper-ingress"): makeService("region-b", "skipper-ingress", defaultLabels), + }, + endpointSlices: map[definitions.ResourceID]*skipperEndpointSlice{ + newResourceID("region-a", "skipper-ingress"): makeSkipperEndpointSlice(map[string]int{"a": 10, "b": 1, "c": 1}), + newResourceID("region-b", "skipper-ingress"): makeSkipperEndpointSlice(map[string]int{"a": 1, "b": 1, "c": 1}), + }, + labelSelector: defaultSelector, + maxZoneSkew: 1, + want: false, // combined: a=11, b=2, c=2 -> skew of 9 exceeds maxZoneSkew=1 + }, + { + name: "one matched service has no endpointslice but another does: the one with data still counts", + endpointSlicesEnable: true, + services: map[definitions.ResourceID]*service{ + newResourceID("region-a", "skipper-ingress"): makeService("region-a", "skipper-ingress", defaultLabels), + newResourceID("region-b", "skipper-ingress"): makeService("region-b", "skipper-ingress", defaultLabels), + }, + endpointSlices: map[definitions.ResourceID]*skipperEndpointSlice{ + newResourceID("region-a", "skipper-ingress"): makeSkipperEndpointSlice(map[string]int{"a": 3, "b": 3, "c": 3}), + // region-b has no endpointslice entry at all + }, + labelSelector: defaultSelector, + maxZoneSkew: 1, + want: true, + }, } { t.Run(tt.name, func(t *testing.T) { - lookupNS := ns - lookupName := name - if tt.customNS != "" { - lookupNS = tt.customNS - } - if tt.customName != "" { - lookupName = tt.customName - } - - endpointSlices := map[definitions.ResourceID]*skipperEndpointSlice{} - if tt.endpointSlicePresent { - // always store under the canonical ns/name; custom test cases - // store under the custom key so that the lookup matches. - storeNS, storeName := ns, name - if tt.customNS != "" && tt.customNS != "wrong-ns" { - storeNS = tt.customNS - } - if tt.customName != "" && tt.customName != "wrong-ingress" { - storeName = tt.customName - } - endpointSlices[newResourceID(storeNS, storeName)] = makeSkipperEndpointSlice(tt.zoneCounts) - } - state := &clusterState{ enableEndpointSlices: tt.endpointSlicesEnable, - endpointSlices: endpointSlices, + services: tt.services, + endpointSlices: tt.endpointSlices, } - got := state.isServiceBalancedAcrossZones(lookupNS, lookupName, tt.maxZoneSkew) + got := state.isServiceBalancedAcrossZones(tt.labelSelector, tt.maxZoneSkew) if got != tt.want { - t.Errorf("isServiceBalancedAcrossZones(%q, %q, %d) = %v, want %v", - lookupNS, lookupName, tt.maxZoneSkew, got, tt.want) + t.Errorf("isServiceBalancedAcrossZones(%v, %d) = %v, want %v", + tt.labelSelector, tt.maxZoneSkew, got, tt.want) } }) } diff --git a/dataclients/kubernetes/kube.go b/dataclients/kubernetes/kube.go index 8284ff4d74..104dbc8752 100644 --- a/dataclients/kubernetes/kube.go +++ b/dataclients/kubernetes/kube.go @@ -30,18 +30,21 @@ const ( minEndpointsByZone = 3 minZones = 3 - // DefaultIngressControllerNamespace is the default namespace used to look up - // the ingress controller endpointslice when checking zone distribution. - DefaultIngressControllerNamespace = "kube-system" - // DefaultIngressControllerName is the default service name used to look up - // the ingress controller endpointslice when checking zone distribution. - DefaultIngressControllerName = "skipper-ingress" // DefaultMaxZoneSkew is the default maximum permitted difference between // the endpoint count in any zone and the global minimum across all zones, // following Kubernetes TopologySpreadConstraints maxSkew semantics. DefaultMaxZoneSkew = 1 ) +// DefaultDataPlaneLabelSelector selects the Service(s) considered part of +// the data plane whose combined endpoint zone distribution is checked to +// decide whether zone-aware routing is enabled, when no selector is +// explicitly configured. +var DefaultDataPlaneLabelSelector = map[string]string{ + "application": "skipper-ingress", + "component": "ingress", +} + // PathMode values are used to control the ingress path interpretation. The path mode can // be set globally for all ingress paths, and it can be overruled by the individual ingress // rules using the zalando.org/skipper-ingress-path-mode annotation. When path mode is not @@ -296,18 +299,19 @@ type Options struct { // addresses from the referenced Service object. IngressStatusFromService string - // IngressControllerNamespace is the namespace of the ingress controller service - // whose endpoint zone distribution is checked to decide whether zone-aware routing - // is enabled. Defaults to "kube-system". - IngressControllerNamespace string - - // IngressControllerName is the service name of the ingress controller whose endpoint - // zone distribution is checked to decide whether zone-aware routing is enabled. - // Defaults to "skipper-ingress". - IngressControllerName string + // DataPlaneLabelSelector selects the Service(s) considered part of the data + // plane whose combined endpoint zone distribution is checked to decide + // whether zone-aware routing is enabled. All matching Services across all + // namespaces are aggregated into a single pool before the balance check + // runs. nil means use DefaultDataPlaneLabelSelector + // ({"application": "skipper-ingress", "component": "ingress"}). An + // explicitly empty, non-nil map matches no Service, which permanently + // disables zone-aware routing (safe default-closed behavior, logged on + // every poll). + DataPlaneLabelSelector map[string]string // MaxZoneSkew is the maximum permitted difference between the endpoint count in any - // zone and the global minimum across all zones for the ingress controller, following + // zone and the global minimum across all zones for the data plane, following // Kubernetes TopologySpreadConstraints maxSkew semantics. // nil means use DefaultMaxZoneSkew. A pointer to 0 means no skew is tolerated. // Negative values are rejected: New returns an error if *MaxZoneSkew < 0. diff --git a/routesrv/export_test.go b/routesrv/export_test.go index c461bf36df..65a753cd5b 100644 --- a/routesrv/export_test.go +++ b/routesrv/export_test.go @@ -11,7 +11,7 @@ func SetNow(rs *RouteServer, now func() time.Time) { } func SetBalancedAcrossZones(rs *RouteServer, v bool) { - rs.poller.isIngressControllerBalancedAcrossZones = func() bool { return v } + rs.poller.isBalancedDataPlane = func() bool { return v } } func (rs *RouteServer) ListenAndServe() (err error) { diff --git a/routesrv/polling.go b/routesrv/polling.go index 17d27d1628..c649bd91d9 100644 --- a/routesrv/polling.go +++ b/routesrv/polling.go @@ -42,9 +42,9 @@ type poller struct { cloneRoute []*eskip.Clone // visibility - tracer ot.Tracer - metrics metrics.Metrics - isIngressControllerBalancedAcrossZones func() bool + tracer ot.Tracer + metrics metrics.Metrics + isBalancedDataPlane func() bool } func (p *poller) poll(wg *sync.WaitGroup) { @@ -76,7 +76,7 @@ func (p *poller) poll(wg *sync.WaitGroup) { ) } else { - balanced := p.isIngressControllerBalancedAcrossZones() + balanced := p.isBalancedDataPlane() routesChanged := hasChanged(lastRoutes, routes) balancedChanged := balanced != lastBalanced diff --git a/routesrv/routesrv.go b/routesrv/routesrv.go index 95794bfa4d..3e9a0d4786 100644 --- a/routesrv/routesrv.go +++ b/routesrv/routesrv.go @@ -180,17 +180,17 @@ func New(opts skipper.Options) (*RouteServer, error) { } rs.poller = &poller{ - client: dataclient, - timeout: opts.SourcePollTimeout, - b: b, - quit: make(chan struct{}), - defaultFilters: opts.DefaultFilters, - editRoute: opts.EditRoute, - cloneRoute: opts.CloneRoute, - oauth2Config: oauthConfig, - tracer: tracer, - metrics: m, - isIngressControllerBalancedAcrossZones: dataclient.ClusterClient.IsIngressControllerBalancedAcrossZones, + client: dataclient, + timeout: opts.SourcePollTimeout, + b: b, + quit: make(chan struct{}), + defaultFilters: opts.DefaultFilters, + editRoute: opts.EditRoute, + cloneRoute: opts.CloneRoute, + oauth2Config: oauthConfig, + tracer: tracer, + metrics: m, + isBalancedDataPlane: dataclient.ClusterClient.IsBalancedDataPlane, } rs.wg = &sync.WaitGroup{} diff --git a/routesrv/routesrv_test.go b/routesrv/routesrv_test.go index 1950f08614..80776910d9 100644 --- a/routesrv/routesrv_test.go +++ b/routesrv/routesrv_test.go @@ -1572,7 +1572,7 @@ func TestZoneAwareRoutesUpdateOnBalanceTransitionAlone(t *testing.T) { // TestZoneAwareRoutingEndToEndWithRealEndpointSlice exercises the real // fetchClusterState -> isServiceBalancedAcrossZones -> -// IsIngressControllerBalancedAcrossZones -> poller pipeline, with no +// IsBalancedDataPlane -> poller pipeline, with no // SetBalancedAcrossZones mock, using real EndpointSlice fixtures for the // ingress controller (skipper-ingress) itself. It also proves the backend // service's own zone distribution (per-route, minEndpointsByZone threshold) @@ -1701,28 +1701,27 @@ func TestZoneAwareRoutingBackendImbalanceDoesNotCloseGate(t *testing.T) { "route with <3 endpoints/zone must still carry all its own endpoints even though ingress controller is balanced") } -// TestZoneAwareRoutingUsesConfiguredIngressControllerNamespaceAndName verifies -// that KubernetesIngressControllerNamespace/KubernetesIngressControllerName -// flow through skipper.Options -> kubernetes.Options -> newClusterClient and -// are actually used to look up the ingress-controller EndpointSlice, instead -// of silently falling back to the kube-system/skipper-ingress defaults. -func TestZoneAwareRoutingUsesConfiguredIngressControllerNamespaceAndName(t *testing.T) { +// TestZoneAwareRoutingUsesConfiguredDataPlaneLabelSelector verifies that +// KubernetesDataPlaneLabelSelector flows through skipper.Options -> +// kubernetes.Options -> newClusterClient and is actually used to find the +// data-plane Service's EndpointSlice, instead of silently falling back to +// the application=skipper-ingress,component=ingress default. +func TestZoneAwareRoutingUsesConfiguredDataPlaneLabelSelector(t *testing.T) { defer tl.Reset() ks, _ := newKubeServer(t, loadKubeYAML(t, "testdata/zone-aware-traffic/all-zones-3-addr.yaml"), - loadKubeYAML(t, "testdata/zone-aware-traffic/custom-ingress-controller-balanced.yaml"), + loadKubeYAML(t, "testdata/zone-aware-traffic/custom-selector-ingress-balanced.yaml"), ) ks.Start() defer ks.Close() rs := newRouteServerWithOptions(t, skipper.Options{ - SourcePollTimeout: pollInterval, - Kubernetes: true, - KubernetesURL: ks.URL, - KubernetesEnableEndpointslices: true, - KubernetesIngressControllerNamespace: "custom-ns", - KubernetesIngressControllerName: "my-custom-ingress", + SourcePollTimeout: pollInterval, + Kubernetes: true, + KubernetesURL: ks.URL, + KubernetesEnableEndpointslices: true, + KubernetesDataPlaneLabelSelector: map[string]string{"team": "platform", "role": "data-plane"}, }) rs.StartUpdates() defer rs.StopUpdates() @@ -1733,7 +1732,7 @@ func TestZoneAwareRoutingUsesConfiguredIngressControllerNamespaceAndName(t *test zone := getZoneAwareRoutes(rs, "eu-central-1a") require.Equal(t, http.StatusOK, zone.Code) assert.NotEqual(t, global.Body.String(), zone.Body.String(), - "configured namespace/name must be used to find the ingress controller endpointslice") + "configured label selector must be used to find the data-plane endpointslice") } // TestZoneAwareRoutingUsesConfiguredMaxZoneSkew verifies that diff --git a/routesrv/testdata/zone-aware-traffic/custom-selector-ingress-balanced.yaml b/routesrv/testdata/zone-aware-traffic/custom-selector-ingress-balanced.yaml new file mode 100644 index 0000000000..fd2aad73a8 --- /dev/null +++ b/routesrv/testdata/zone-aware-traffic/custom-selector-ingress-balanced.yaml @@ -0,0 +1,40 @@ +apiVersion: v1 +kind: Service +metadata: + labels: + team: platform + role: data-plane + name: custom-selector-ingress + namespace: custom-ns +spec: + clusterIP: 10.3.190.97 + ports: + - name: web + port: 80 + protocol: TCP + targetPort: 9090 + selector: + team: platform + type: ClusterIP +--- +apiVersion: v1 +kind: EndpointSlice +metadata: + labels: + kubernetes.io/service-name: custom-selector-ingress + name: custom-selector-ingress-foo + namespace: custom-ns +endpoints: + - addresses: + - 10.9.6.1 + zone: eu-central-1a + - addresses: + - 10.9.6.2 + zone: eu-central-1b + - addresses: + - 10.9.6.3 + zone: eu-central-1c +ports: + - name: web + port: 9090 + protocol: TCP diff --git a/routesrv/testdata/zone-aware-traffic/skipper-ingress-balanced.yaml b/routesrv/testdata/zone-aware-traffic/skipper-ingress-balanced.yaml index a5b420254a..462c0927da 100644 --- a/routesrv/testdata/zone-aware-traffic/skipper-ingress-balanced.yaml +++ b/routesrv/testdata/zone-aware-traffic/skipper-ingress-balanced.yaml @@ -2,7 +2,8 @@ apiVersion: v1 kind: Service metadata: labels: - app: skipper-ingress + application: skipper-ingress + component: ingress name: skipper-ingress namespace: kube-system spec: @@ -13,7 +14,8 @@ spec: protocol: TCP targetPort: 9090 selector: - app: skipper-ingress + application: skipper-ingress + component: ingress type: ClusterIP --- apiVersion: v1 diff --git a/routesrv/testdata/zone-aware-traffic/skipper-ingress-missing-zone.yaml b/routesrv/testdata/zone-aware-traffic/skipper-ingress-missing-zone.yaml index c071cbdaaf..f6c33602b8 100644 --- a/routesrv/testdata/zone-aware-traffic/skipper-ingress-missing-zone.yaml +++ b/routesrv/testdata/zone-aware-traffic/skipper-ingress-missing-zone.yaml @@ -2,7 +2,8 @@ apiVersion: v1 kind: Service metadata: labels: - app: skipper-ingress + application: skipper-ingress + component: ingress name: skipper-ingress namespace: kube-system spec: @@ -13,7 +14,8 @@ spec: protocol: TCP targetPort: 9090 selector: - app: skipper-ingress + application: skipper-ingress + component: ingress type: ClusterIP --- apiVersion: v1 diff --git a/routesrv/testdata/zone-aware-traffic/skipper-ingress-skewed.yaml b/routesrv/testdata/zone-aware-traffic/skipper-ingress-skewed.yaml index 5148d16e3b..d93e269f70 100644 --- a/routesrv/testdata/zone-aware-traffic/skipper-ingress-skewed.yaml +++ b/routesrv/testdata/zone-aware-traffic/skipper-ingress-skewed.yaml @@ -2,7 +2,8 @@ apiVersion: v1 kind: Service metadata: labels: - app: skipper-ingress + application: skipper-ingress + component: ingress name: skipper-ingress namespace: kube-system spec: @@ -13,7 +14,8 @@ spec: protocol: TCP targetPort: 9090 selector: - app: skipper-ingress + application: skipper-ingress + component: ingress type: ClusterIP --- apiVersion: v1 diff --git a/skipper.go b/skipper.go index 2f3c65ca54..b2af5cde2b 100644 --- a/skipper.go +++ b/skipper.go @@ -378,15 +378,12 @@ type Options struct { // KubernetesTopologyZone if set to non empty string it is used to filter endpointslice endpoints by this value. KubernetesTopologyZone string - // KubernetesIngressControllerNamespace is the namespace of the ingress controller service - // whose endpoint zone distribution is checked to decide whether zone-aware routing is enabled. - // Defaults to "kube-system". - KubernetesIngressControllerNamespace string - - // KubernetesIngressControllerName is the service name of the ingress controller whose endpoint - // zone distribution is checked to decide whether zone-aware routing is enabled. - // Defaults to "skipper-ingress". - KubernetesIngressControllerName string + // KubernetesDataPlaneLabelSelector selects the Service(s) considered part + // of the data plane whose combined endpoint zone distribution is checked + // to decide whether zone-aware routing is enabled. Defaults to + // kubernetes.DefaultDataPlaneLabelSelector + // ({"application": "skipper-ingress", "component": "ingress"}) when nil. + KubernetesDataPlaneLabelSelector map[string]string // KubernetesZoneMaxSkew is the maximum permitted difference between the endpoint count in any // zone and the global minimum across all zones for the ingress controller, following Kubernetes @@ -1295,8 +1292,7 @@ func (o *Options) KubernetesDataClientOptions() kubernetes.Options { DefaultLoadBalancerAlgorithm: o.KubernetesDefaultLoadBalancerAlgorithm, ForwardBackendURL: o.ForwardBackendURL, TopologyZone: o.KubernetesTopologyZone, - IngressControllerNamespace: o.KubernetesIngressControllerNamespace, - IngressControllerName: o.KubernetesIngressControllerName, + DataPlaneLabelSelector: o.KubernetesDataPlaneLabelSelector, MaxZoneSkew: &o.KubernetesZoneMaxSkew, IngressStatusFromService: o.KubernetesStatusFromService, KubernetesApplicationAnnotationLabelKey: o.KubernetesApplicationAnnotationLabelKey,