Merge pull request kubernetes#92312 from Sh4d1/kep_1860

Make Kubernetes aware of the LoadBalancer behaviour

pkg/apis/core/types.go

@@ -3508,13 +3508,34 @@ type LoadBalancerIngress struct {
 	// (typically AWS load-balancers)
 	// +optional
 	Hostname string
+
+	// IPMode specifies how the load-balancer's IP behaves.
+	// Setting this to "VIP" indicates that the traffic passing through
+	// this load-balancer is delivered with the destination IP and port set to the load-balancer's IP and port.
+	// Setting this to "Proxy" indicates that the load-balancer acts like a proxy,
+	// delivering traffic with the destination IP and port set to the node's IP and nodePort or to the pod's IP and targetPort.
+	// This field can only be set when the ip field is also set, and defaults to "VIP" if not specified.
+	// +optional
+	IPMode *LoadBalancerIPMode
 }
 
 const (
 	// MaxServiceTopologyKeys is the largest number of topology keys allowed on a service
 	MaxServiceTopologyKeys = 16
 )
 
+// LoadBalancerIPMode represents the mode of the LoadBalancer ingress IP
+type LoadBalancerIPMode string
+
+const (
+	// LoadBalancerIPModeVIP indicates that the traffic passing through this LoadBalancer
+	// is delivered with the destination IP set to the specified LoadBalancer IP
+	LoadBalancerIPModeVIP LoadBalancerIPMode = "VIP"
+	// LoadBalancerIPModeProxy indicates that the specified LoadBalancer acts like a proxy,
+	// changing the destination IP to the node IP and the source IP to the LoadBalancer (mostly private) IP
+	LoadBalancerIPModeProxy LoadBalancerIPMode = "Proxy"
+)
+
 // IPFamily represents the IP Family (IPv4 or IPv6). This type is used
 // to express the family of an IP expressed by a type (e.g. service.spec.ipFamilies).
 type IPFamily string

pkg/apis/core/v1/defaults.go

@@ -19,7 +19,7 @@ package v1
 import (
 	"time"
 
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/kubernetes/pkg/util/parsers"
@@ -166,6 +166,17 @@ func SetDefaults_Service(obj *v1.Service) {
 		// further IPFamilies, IPFamilyPolicy defaulting is in ClusterIP alloc/reserve logic
 		// note: conversion logic handles cases where ClusterIPs is used (but not ClusterIP).
 	}
+
+	if utilfeature.DefaultFeatureGate.Enabled(features.LoadBalancerIPMode) &&
+		obj.Spec.Type == v1.ServiceTypeLoadBalancer {
+		ipMode := v1.LoadBalancerIPModeVIP
+
+		for i, ing := range obj.Status.LoadBalancer.Ingress {
+			if ing.IP != "" && ing.IPMode == nil {
+				obj.Status.LoadBalancer.Ingress[i].IPMode = &ipMode
+			}
+		}
+	}
 }
 func SetDefaults_Pod(obj *v1.Pod) {
 	// If limits are specified, but requests are not, default requests to limits

pkg/apis/core/validation/validation.go

@@ -5973,6 +5973,10 @@ func ValidatePodLogOptions(opts *core.PodLogOptions) field.ErrorList {
 	return allErrs
 }
 
+var (
+	supportedLoadBalancerIPMode = sets.NewString(string(core.LoadBalancerIPModeVIP), string(core.LoadBalancerIPModeProxy))
+)
+
 // ValidateLoadBalancerStatus validates required fields on a LoadBalancerStatus
 func ValidateLoadBalancerStatus(status *core.LoadBalancerStatus, fldPath *field.Path) field.ErrorList {
 	allErrs := field.ErrorList{}
@@ -5983,6 +5987,23 @@ func ValidateLoadBalancerStatus(status *core.LoadBalancerStatus, fldPath *field.
 				allErrs = append(allErrs, field.Invalid(idxPath.Child("ip"), ingress.IP, "must be a valid IP address"))
 			}
 		}
+
+		if utilfeature.DefaultFeatureGate.Enabled(features.LoadBalancerIPMode) {
+			if len(ingress.IP) > 0 && ingress.IPMode == nil {
+				allErrs = append(allErrs, field.Required(idxPath.Child("ipMode"), "must be specified when `ip` is set"))
+			}
+		}
+
+		if ingress.IPMode != nil {
+			if len(ingress.IP) == 0 {
+				allErrs = append(allErrs, field.Forbidden(idxPath.Child("ipMode"), "may not be used when `ip` is not set"))
+			}
+
+			if !supportedLoadBalancerIPMode.Has(string(*ingress.IPMode)) {
+				allErrs = append(allErrs, field.NotSupported(idxPath.Child("ipMode"), ingress.IPMode, supportedLoadBalancerIPMode.List()))
+			}
+		}
+
 		if len(ingress.Hostname) > 0 {
 			for _, msg := range validation.IsDNS1123Subdomain(ingress.Hostname) {
 				allErrs = append(allErrs, field.Invalid(idxPath.Child("hostname"), ingress.Hostname, msg))

pkg/apis/core/validation/validation_test.go

@@ -11071,6 +11071,90 @@ func TestValidateServiceCreate(t *testing.T) {
 	}
 }
 
+func TestValidateLoadBalancerStatus(t *testing.T) {
+	defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.LoadBalancerIPMode, true)()
+
+	ipModeVIP := core.LoadBalancerIPModeVIP
+	ipModeProxy := core.LoadBalancerIPModeProxy
+	ipModeDummy := core.LoadBalancerIPMode("dummy")
+
+	testCases := []struct {
+		name          string
+		tweakLBStatus func(s *core.LoadBalancerStatus)
+		numErrs       int
+	}{
+		/* LoadBalancerIPMode*/
+		{
+			name: `valid vip ipMode`,
+			tweakLBStatus: func(s *core.LoadBalancerStatus) {
+				s.Ingress = []core.LoadBalancerIngress{
+					{
+						IP:     "1.2.3.4",
+						IPMode: &ipModeVIP,
+					},
+				}
+			},
+			numErrs: 0,
+		},
+		{
+			name: `valid proxy ipMode`,
+			tweakLBStatus: func(s *core.LoadBalancerStatus) {
+				s.Ingress = []core.LoadBalancerIngress{
+					{
+						IP:     "1.2.3.4",
+						IPMode: &ipModeProxy,
+					},
+				}
+			},
+			numErrs: 0,
+		},
+		{
+			name: `invalid ipMode`,
+			tweakLBStatus: func(s *core.LoadBalancerStatus) {
+				s.Ingress = []core.LoadBalancerIngress{
+					{
+						IP:     "1.2.3.4",
+						IPMode: &ipModeDummy,
+					},
+				}
+			},
+			numErrs: 1,
+		},
+		{
+			name: `missing ipMode`,
+			tweakLBStatus: func(s *core.LoadBalancerStatus) {
+				s.Ingress = []core.LoadBalancerIngress{
+					{
+						IP: "1.2.3.4",
+					},
+				}
+			},
+			numErrs: 1,
+		},
+		{
+			name: `missing ip with ipMode present`,
+			tweakLBStatus: func(s *core.LoadBalancerStatus) {
+				s.Ingress = []core.LoadBalancerIngress{
+					{
+						IPMode: &ipModeProxy,
+					},
+				}
+			},
+			numErrs: 1,
+		},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			s := core.LoadBalancerStatus{}
+			tc.tweakLBStatus(&s)
+			errs := ValidateLoadBalancerStatus(&s, field.NewPath("status"))
+			if len(errs) != tc.numErrs {
+				t.Errorf("Unexpected error list for case %q(expected:%v got %v) - Errors:\n %v", tc.name, tc.numErrs, len(errs), errs.ToAggregate())
+			}
+		})
+	}
+}
+
 func TestValidateServiceExternalTrafficFieldsCombination(t *testing.T) {
 	testCases := []struct {
 		name     string

pkg/features/kube_features.go

@@ -663,6 +663,11 @@ const (
 	//
 	// Enables kubelet support to size memory backed volumes
 	SizeMemoryBackedVolumes featuregate.Feature = "SizeMemoryBackedVolumes"
+
+	// owner: @Sh4d1
+	// alpha: v1.21
+	// LoadBalancerIPMode enables the IPMode field in the LoadBalancerIngress status of a Service
+	LoadBalancerIPMode featuregate.Feature = "LoadBalancerIPMode"
 )
 
 func init() {
@@ -763,6 +768,7 @@ var defaultKubernetesFeatureGates = map[featuregate.Feature]featuregate.FeatureS
 	HPAContainerMetrics:                            {Default: false, PreRelease: featuregate.Alpha},
 	RootCAConfigMap:                                {Default: true, PreRelease: featuregate.Beta},
 	SizeMemoryBackedVolumes:                        {Default: false, PreRelease: featuregate.Alpha},
+	LoadBalancerIPMode:                             {Default: false, PreRelease: featuregate.Alpha},
 
 	// inherited features from generic apiserver, relisted here to get a conflict if it is changed
 	// unintentionally on either side:

pkg/proxy/iptables/BUILD

@@ -39,6 +39,7 @@ go_test(
     srcs = ["proxier_test.go"],
     embed = [":go_default_library"],
     deps = [
+        "//pkg/features:go_default_library",
         "//pkg/proxy:go_default_library",
         "//pkg/proxy/healthcheck:go_default_library",
         "//pkg/proxy/util:go_default_library",
@@ -53,6 +54,8 @@ go_test(
         "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/intstr:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/util/feature:go_default_library",
+        "//staging/src/k8s.io/component-base/featuregate/testing:go_default_library",
         "//vendor/github.com/stretchr/testify/assert:go_default_library",
         "//vendor/k8s.io/klog/v2:go_default_library",
         "//vendor/k8s.io/utils/exec:go_default_library",

pkg/proxy/iptables/proxier.go

@@ -782,10 +782,10 @@ func (proxier *Proxier) deleteEndpointConnections(connectionMap []proxy.ServiceE
 					klog.Errorf("Failed to delete %s endpoint connections for externalIP %s, error: %v", epSvcPair.ServicePortName.String(), extIP, err)
 				}
 			}
-			for _, lbIP := range svcInfo.LoadBalancerIPStrings() {
-				err := conntrack.ClearEntriesForNAT(proxier.exec, lbIP, endpointIP, svcProto)
+			for _, lbIngress := range svcInfo.LoadBalancerIngress() {
+				err := conntrack.ClearEntriesForNAT(proxier.exec, lbIngress.IP, endpointIP, svcProto)
 				if err != nil {
-					klog.Errorf("Failed to delete %s endpoint connections for LoabBalancerIP %s, error: %v", epSvcPair.ServicePortName.String(), lbIP, err)
+					klog.Errorf("Failed to delete %s endpoint connections for LoabBalancerIP %s, error: %v", epSvcPair.ServicePortName.String(), lbIngress.IP, err)
 				}
 			}
 		}
@@ -1161,8 +1161,8 @@ func (proxier *Proxier) syncProxyRules() {
 
 		// Capture load-balancer ingress.
 		fwChain := svcInfo.serviceFirewallChainName
-		for _, ingress := range svcInfo.LoadBalancerIPStrings() {
-			if ingress != "" {
+		for _, ingress := range svcInfo.LoadBalancerIngress() {
+			if ingress.IP != "" {
 				if hasEndpoints {
 					// create service firewall chain
 					if chain, ok := existingNATChains[fwChain]; ok {
@@ -1175,15 +1175,17 @@ func (proxier *Proxier) syncProxyRules() {
 					// This currently works for loadbalancers that preserves source ips.
 					// For loadbalancers which direct traffic to service NodePort, the firewall rules will not apply.
 
-					args = append(args[:0],
-						"-A", string(kubeServicesChain),
-						"-m", "comment", "--comment", fmt.Sprintf(`"%s loadbalancer IP"`, svcNameString),
-						"-m", protocol, "-p", protocol,
-						"-d", utilproxy.ToCIDR(net.ParseIP(ingress)),
-						"--dport", strconv.Itoa(svcInfo.Port()),
-					)
-					// jump to service firewall chain
-					writeLine(proxier.natRules, append(args, "-j", string(fwChain))...)
+					if !utilfeature.DefaultFeatureGate.Enabled(features.LoadBalancerIPMode) || *ingress.IPMode == v1.LoadBalancerIPModeVIP {
+						args = append(args[:0],
+							"-A", string(kubeServicesChain),
+							"-m", "comment", "--comment", fmt.Sprintf(`"%s loadbalancer IP"`, svcNameString),
+							"-m", protocol, "-p", protocol,
+							"-d", utilproxy.ToCIDR(net.ParseIP(ingress.IP)),
+							"--dport", strconv.Itoa(svcInfo.Port()),
+						)
+						// jump to service firewall chain
+						writeLine(proxier.natRules, append(args, "-j", string(fwChain))...)
+					}
 
 					args = append(args[:0],
 						"-A", string(fwChain),
@@ -1218,7 +1220,7 @@ func (proxier *Proxier) syncProxyRules() {
 						// loadbalancer's backend hosts. In this case, request will not hit the loadbalancer but loop back directly.
 						// Need to add the following rule to allow request on host.
 						if allowFromNode {
-							writeLine(proxier.natRules, append(args, "-s", utilproxy.ToCIDR(net.ParseIP(ingress)), "-j", string(chosenChain))...)
+							writeLine(proxier.natRules, append(args, "-s", utilproxy.ToCIDR(net.ParseIP(ingress.IP)), "-j", string(chosenChain))...)
 						}
 					}
 
@@ -1231,7 +1233,7 @@ func (proxier *Proxier) syncProxyRules() {
 						"-A", string(kubeExternalServicesChain),
 						"-m", "comment", "--comment", fmt.Sprintf(`"%s has no endpoints"`, svcNameString),
 						"-m", protocol, "-p", protocol,
-						"-d", utilproxy.ToCIDR(net.ParseIP(ingress)),
+						"-d", utilproxy.ToCIDR(net.ParseIP(ingress.IP)),
 						"--dport", strconv.Itoa(svcInfo.Port()),
 						"-j", "REJECT",
 					)