Add PreferSameTrafficDistribution feature gate and associated API.

Author: danwinship

pkg/apis/core/types.go

@@ -4562,12 +4562,27 @@ const (
 
 // These are valid values for the TrafficDistribution field of a Service.
 const (
-	// Indicates a preference for routing traffic to endpoints that are in the
-	// same zone as the client. Setting this value gives implementations
-	// permission to make different tradeoffs, e.g. optimizing for proximity
-	// rather than equal distribution of load. Users should not set this value
-	// if such tradeoffs are not acceptable.
+	// Indicates a preference for routing traffic to endpoints that are in the same
+	// zone as the client. Users should not set this value unless they have ensured
+	// that clients and endpoints are distributed in such a way that the "same zone"
+	// preference will not result in endpoints getting overloaded.
 	ServiceTrafficDistributionPreferClose = "PreferClose"
+
+	// Indicates a preference for routing traffic to endpoints that are in the same
+	// zone as the client. Users should not set this value unless they have ensured
+	// that clients and endpoints are distributed in such a way that the "same zone"
+	// preference will not result in endpoints getting overloaded.
+	// This is an alias for "PreferClose", but it is an Alpha feature and is only
+	// recognized if the PreferSameTrafficDistribution feature gate is enabled.
+	ServiceTrafficDistributionPreferSameZone = "PreferSameZone"
+
+	// Indicates a preference for routing traffic to endpoints that are on the same
+	// node as the client. Users should not set this value unless they have ensured
+	// that clients and endpoints are distributed in such a way that the "same node"
+	// preference will not result in endpoints getting overloaded.
+	// This is an Alpha feature and is only recognized if the
+	// PreferSameTrafficDistribution feature gate is enabled.
+	ServiceTrafficDistributionPreferSameNode = "PreferSameNode"
 )
 
 // These are the valid conditions of a service.

pkg/apis/core/validation/validation.go

@@ -25,6 +25,7 @@ import (
 	"path/filepath"
 	"reflect"
 	"regexp"
+	"slices"
 	"strings"
 	"sync"
 	"unicode"
@@ -6194,8 +6195,21 @@ func validateServiceTrafficDistribution(service *core.Service) field.ErrorList {
 		return allErrs
 	}
 
-	if *service.Spec.TrafficDistribution != v1.ServiceTrafficDistributionPreferClose {
-		allErrs = append(allErrs, field.NotSupported(field.NewPath("spec").Child("trafficDistribution"), *service.Spec.TrafficDistribution, []string{v1.ServiceTrafficDistributionPreferClose}))
+	var supportedTrafficDistribution []string
+	if !utilfeature.DefaultFeatureGate.Enabled(features.PreferSameTrafficDistribution) {
+		supportedTrafficDistribution = []string{
+			v1.ServiceTrafficDistributionPreferClose,
+		}
+	} else {
+		supportedTrafficDistribution = []string{
+			v1.ServiceTrafficDistributionPreferClose,
+			v1.ServiceTrafficDistributionPreferSameZone,
+			v1.ServiceTrafficDistributionPreferSameNode,
+		}
+	}
+
+	if !slices.Contains(supportedTrafficDistribution, *service.Spec.TrafficDistribution) {
+		allErrs = append(allErrs, field.NotSupported(field.NewPath("spec").Child("trafficDistribution"), *service.Spec.TrafficDistribution, supportedTrafficDistribution))
 	}
 
 	return allErrs

pkg/apis/core/validation/validation_test.go

@@ -16413,12 +16413,38 @@ func TestValidateServiceCreate(t *testing.T) {
 				s.Spec.TrafficDistribution = ptr.To("PreferClose")
 			},
 			numErrs: 0,
+		}, {
+			name: "valid: trafficDistribution field set to PreferSameZone with feature gate",
+			tweakSvc: func(s *core.Service) {
+				s.Spec.TrafficDistribution = ptr.To("PreferSameZone")
+			},
+			featureGates: []featuregate.Feature{features.PreferSameTrafficDistribution},
+			numErrs:      0,
+		}, {
+			name: "valid: trafficDistribution field set to PreferSameNode with feature gate",
+			tweakSvc: func(s *core.Service) {
+				s.Spec.TrafficDistribution = ptr.To("PreferSameNode")
+			},
+			featureGates: []featuregate.Feature{features.PreferSameTrafficDistribution},
+			numErrs:      0,
 		}, {
 			name: "invalid: trafficDistribution field set to Random",
 			tweakSvc: func(s *core.Service) {
 				s.Spec.TrafficDistribution = ptr.To("Random")
 			},
 			numErrs: 1,
+		}, {
+			name: "invalid: trafficDistribution field set to PreferSameZone without feature gate",
+			tweakSvc: func(s *core.Service) {
+				s.Spec.TrafficDistribution = ptr.To("PreferSameZone")
+			},
+			numErrs: 1,
+		}, {
+			name: "invalid: trafficDistribution field set to PreferSameNode without feature gate",
+			tweakSvc: func(s *core.Service) {
+				s.Spec.TrafficDistribution = ptr.To("PreferSameNode")
+			},
+			numErrs: 1,
 		},
 	}
 

pkg/apis/discovery/types.go

@@ -133,9 +133,16 @@ type EndpointConditions struct {
 
 // EndpointHints provides hints describing how an endpoint should be consumed.
 type EndpointHints struct {
-	// forZones indicates the zone(s) this endpoint should be consumed by to
-	// enable topology aware routing. May contain a maximum of 8 entries.
+	// forZones indicates the zone(s) this endpoint should be consumed by when
+	// using topology aware routing. May contain a maximum of 8 entries.
 	ForZones []ForZone
+
+	// forNodes indicates the node(s) this endpoint should be consumed by when
+	// using topology aware routing.
+	// This is an Alpha feature and is only used when the PreferSameTrafficDistribution
+	// feature gate is enabled. May contain a maximum of 8 entries.
+	// +featureGate=PreferSameTrafficDistribution
+	ForNodes []ForNode
 }
 
 // ForZone provides information about which zones should consume this endpoint.
@@ -144,6 +151,12 @@ type ForZone struct {
 	Name string
 }
 
+// ForNode provides information about which nodes should consume this endpoint.
+type ForNode struct {
+	// name represents the name of the node.
+	Name string
+}
+
 // EndpointPort represents a Port used by an EndpointSlice.
 type EndpointPort struct {
 	// The name of this port. All ports in an EndpointSlice must have a unique

pkg/apis/discovery/validation/validation.go

@@ -49,6 +49,7 @@ var (
 	maxPorts          = 20000
 	maxEndpoints      = 1000
 	maxZoneHints      = 8
+	maxNodeHints      = 8
 )
 
 // ValidateEndpointSliceName can be used to check whether the given endpoint
@@ -240,5 +241,25 @@ func validateHints(endpointHints *discovery.EndpointHints, fldPath *field.Path)
 		}
 	}
 
+	fnPath := fldPath.Child("forNodes")
+	if len(endpointHints.ForNodes) > maxNodeHints {
+		allErrs = append(allErrs, field.TooMany(fnPath, len(endpointHints.ForNodes), maxNodeHints))
+		return allErrs
+	}
+
+	nodeNames := make([]string, 0, len(endpointHints.ForNodes))
+	for i, forNode := range endpointHints.ForNodes {
+		nodePath := fnPath.Index(i).Child("name")
+		if slices.Contains(nodeNames, forNode.Name) {
+			allErrs = append(allErrs, field.Duplicate(nodePath, forNode.Name))
+		} else {
+			nodeNames = append(nodeNames, forNode.Name)
+		}
+
+		for _, msg := range apivalidation.ValidateNodeName(forNode.Name, false) {
+			allErrs = append(allErrs, field.Invalid(nodePath, forNode.Name, msg))
+		}
+	}
+
 	return allErrs
 }

pkg/apis/discovery/validation/validation_test.go

@@ -235,6 +235,7 @@ func TestValidateEndpointSlice(t *testing.T) {
 					Addresses: generateIPAddresses(1),
 					Hints: &discovery.EndpointHints{
 						ForZones: []discovery.ForZone{{Name: "zone-a"}},
+						ForNodes: []discovery.ForNode{{Name: "node-1"}},
 					},
 				}},
 			},
@@ -518,7 +519,7 @@ func TestValidateEndpointSlice(t *testing.T) {
 				}},
 			},
 		},
-		"invalid-hints": {
+		"invalid-zone-hint": {
 			expectedErrors: 1,
 			endpointSlice: &discovery.EndpointSlice{
 				ObjectMeta:  standardMeta,
@@ -535,7 +536,7 @@ func TestValidateEndpointSlice(t *testing.T) {
 				}},
 			},
 		},
-		"overlapping-hints": {
+		"overlapping-zone-hints": {
 			expectedErrors: 1,
 			endpointSlice: &discovery.EndpointSlice{
 				ObjectMeta:  standardMeta,
@@ -556,7 +557,7 @@ func TestValidateEndpointSlice(t *testing.T) {
 				}},
 			},
 		},
-		"too-many-hints": {
+		"too-many-zone-hints": {
 			expectedErrors: 1,
 			endpointSlice: &discovery.EndpointSlice{
 				ObjectMeta:  standardMeta,
@@ -583,6 +584,74 @@ func TestValidateEndpointSlice(t *testing.T) {
 				}},
 			},
 		},
+		"invalid-node-hints": {
+			expectedErrors: 2,
+			endpointSlice: &discovery.EndpointSlice{
+				ObjectMeta:  standardMeta,
+				AddressType: discovery.AddressTypeIPv4,
+				Ports: []discovery.EndpointPort{{
+					Name:     ptr.To("http"),
+					Protocol: ptr.To(api.ProtocolTCP),
+				}},
+				Endpoints: []discovery.Endpoint{{
+					Addresses: generateIPAddresses(1),
+					Hints: &discovery.EndpointHints{
+						ForNodes: []discovery.ForNode{
+							{Name: "!@#$!@"},
+							{Name: ""},
+						},
+					},
+				}},
+			},
+		},
+		"overlapping-node-hints": {
+			expectedErrors: 1,
+			endpointSlice: &discovery.EndpointSlice{
+				ObjectMeta:  standardMeta,
+				AddressType: discovery.AddressTypeIPv4,
+				Ports: []discovery.EndpointPort{{
+					Name:     ptr.To("http"),
+					Protocol: ptr.To(api.ProtocolTCP),
+				}},
+				Endpoints: []discovery.Endpoint{{
+					Addresses: generateIPAddresses(1),
+					Hints: &discovery.EndpointHints{
+						ForNodes: []discovery.ForNode{
+							{Name: "node-1"},
+							{Name: "node-2"},
+							{Name: "node-1"},
+						},
+					},
+				}},
+			},
+		},
+		"too-many-node-hints": {
+			expectedErrors: 1,
+			endpointSlice: &discovery.EndpointSlice{
+				ObjectMeta:  standardMeta,
+				AddressType: discovery.AddressTypeIPv4,
+				Ports: []discovery.EndpointPort{{
+					Name:     ptr.To("http"),
+					Protocol: ptr.To(api.ProtocolTCP),
+				}},
+				Endpoints: []discovery.Endpoint{{
+					Addresses: generateIPAddresses(1),
+					Hints: &discovery.EndpointHints{
+						ForNodes: []discovery.ForNode{
+							{Name: "node-1"},
+							{Name: "node-2"},
+							{Name: "node-3"},
+							{Name: "node-4"},
+							{Name: "node-5"},
+							{Name: "node-6"},
+							{Name: "node-7"},
+							{Name: "node-8"},
+							{Name: "node-9"},
+						},
+					},
+				}},
+			},
+		},
 		"empty-everything": {
 			expectedErrors: 3,
 			endpointSlice:  &discovery.EndpointSlice{},

pkg/features/kube_features.go

@@ -557,6 +557,12 @@ const (
 	// Enables PortForward to be proxied with a websocket client
 	PortForwardWebsockets featuregate.Feature = "PortForwardWebsockets"
 
+	// owner: @danwinship
+	// kep: https://kep.k8s.io/3015
+	//
+	// Enables PreferSameZone and PreferSameNode values for trafficDistribution
+	PreferSameTrafficDistribution featuregate.Feature = "PreferSameTrafficDistribution"
+
 	// owner: @jessfraz
 	//
 	// Enables control over ProcMountType for containers.

pkg/features/versioned_kube_features.go

@@ -621,6 +621,10 @@ var defaultVersionedKubernetesFeatureGates = map[featuregate.Feature]featuregate
 		{Version: version.MustParse("1.31"), Default: true, PreRelease: featuregate.Beta},
 	},
 
+	PreferSameTrafficDistribution: {
+		{Version: version.MustParse("1.33"), Default: false, PreRelease: featuregate.Alpha},
+	},
+
 	ProcMountType: {
 		{Version: version.MustParse("1.12"), Default: false, PreRelease: featuregate.Alpha},
 		{Version: version.MustParse("1.31"), Default: false, PreRelease: featuregate.Beta},

pkg/registry/discovery/endpointslice/strategy.go

@@ -142,12 +142,16 @@ func (endpointSliceStrategy) AllowUnconditionalUpdate() bool {
 // dropDisabledConditionsOnCreate will drop any fields that are disabled.
 func dropDisabledFieldsOnCreate(endpointSlice *discovery.EndpointSlice) {
 	dropHints := !utilfeature.DefaultFeatureGate.Enabled(features.TopologyAwareHints)
+	dropNodeHints := !utilfeature.DefaultFeatureGate.Enabled(features.PreferSameTrafficDistribution)
+	if !dropHints && !dropNodeHints {
+		return
+	}
 
-	if dropHints {
-		for i := range endpointSlice.Endpoints {
-			if dropHints {
-				endpointSlice.Endpoints[i].Hints = nil
-			}
+	for i := range endpointSlice.Endpoints {
+		if dropHints {
+			endpointSlice.Endpoints[i].Hints = nil
+		} else if endpointSlice.Endpoints[i].Hints != nil {
+			endpointSlice.Endpoints[i].Hints.ForNodes = nil
 		}
 	}
 }
@@ -156,20 +160,27 @@ func dropDisabledFieldsOnCreate(endpointSlice *discovery.EndpointSlice) {
 // been set on the EndpointSlice.
 func dropDisabledFieldsOnUpdate(oldEPS, newEPS *discovery.EndpointSlice) {
 	dropHints := !utilfeature.DefaultFeatureGate.Enabled(features.TopologyAwareHints)
-	if dropHints {
+	dropNodeHints := !utilfeature.DefaultFeatureGate.Enabled(features.PreferSameTrafficDistribution)
+	if dropHints || dropNodeHints {
 		for _, ep := range oldEPS.Endpoints {
 			if ep.Hints != nil {
 				dropHints = false
-				break
+				if ep.Hints.ForNodes != nil {
+					dropNodeHints = false
+					break
+				}
 			}
 		}
 	}
+	if !dropHints && !dropNodeHints {
+		return
+	}
 
-	if dropHints {
-		for i := range newEPS.Endpoints {
-			if dropHints {
-				newEPS.Endpoints[i].Hints = nil
-			}
+	for i := range newEPS.Endpoints {
+		if dropHints {
+			newEPS.Endpoints[i].Hints = nil
+		} else if newEPS.Endpoints[i].Hints != nil {
+			newEPS.Endpoints[i].Hints.ForNodes = nil
 		}
 	}
 }