Merge pull request #99287 from anfernee/clientip

Add HNS Load Balancer Healthchecks for ExternalTrafficPolicy: Local

cluster/gce/windows/k8s-node-setup.psm1

@@ -892,8 +892,12 @@ function Configure-HostNetworkingService {
         -Verbose
     $created_hns_network = $true
   }
-
+  # This name of endpoint is referred in pkg/proxy/winkernel/proxier.go as part of
+  # kube-proxy as well. A health check port for every service that is specified as
+  # "externalTrafficPolicy: local" will be added on the endpoint.
+  # PLEASE KEEP THEM CONSISTENT!!!
   $endpoint_name = "cbr0"
+
   $vnic_name = "vEthernet (${endpoint_name})"
 
   $hns_endpoint = Get-HnsEndpoint | Where-Object Name -eq $endpoint_name

cmd/kube-proxy/app/init_windows.go

@@ -41,4 +41,6 @@ func (o *Options) addOSFlags(fs *pflag.FlagSet) {
 	fs.StringVar(&o.config.Winkernel.SourceVip, "source-vip", o.config.Winkernel.SourceVip, "The IP address of the source VIP for non-DSR.")
 	fs.StringVar(&o.config.Winkernel.NetworkName, "network-name", o.config.Winkernel.NetworkName, "The name of the cluster network.")
 	fs.BoolVar(&o.config.Winkernel.EnableDSR, "enable-dsr", o.config.Winkernel.EnableDSR, "If true make kube-proxy apply DSR policies for service VIP")
+	fs.StringVar(&o.config.Winkernel.RootHnsEndpointName, "root-hnsendpoint-name", "cbr0", "The name of the hns endpoint name for root namespace attached to l2bridge")
+	fs.BoolVar(&o.config.Winkernel.ForwardHealthCheckVip, "forward-healthcheck-vip", o.config.Winkernel.ForwardHealthCheckVip, "If true forward service VIP for health check port")
 }

cmd/kube-proxy/app/server_windows.go

@@ -24,7 +24,9 @@ package app
 import (
 	"errors"
 	"fmt"
+	"net"
 	goruntime "runtime"
+	"strconv"
 
 	// Enable pprof HTTP handlers.
 	_ "net/http/pprof"
@@ -97,8 +99,11 @@ func newProxyServer(config *proxyconfigapi.KubeProxyConfiguration, cleanupAndExi
 	}
 
 	var healthzServer healthcheck.ProxierHealthUpdater
+	var healthzPort int
 	if len(config.HealthzBindAddress) > 0 {
 		healthzServer = healthcheck.NewProxierHealthServer(config.HealthzBindAddress, 2*config.IPTables.SyncPeriod.Duration, recorder, nodeRef)
+		_, port, _ := net.SplitHostPort(config.HealthzBindAddress)
+		healthzPort, _ = strconv.Atoi(port)
 	}
 
 	var proxier proxy.Provider
@@ -120,6 +125,7 @@ func newProxyServer(config *proxyconfigapi.KubeProxyConfiguration, cleanupAndExi
 				recorder,
 				healthzServer,
 				config.Winkernel,
+				healthzPort,
 			)
 		} else {
 
@@ -134,6 +140,7 @@ func newProxyServer(config *proxyconfigapi.KubeProxyConfiguration, cleanupAndExi
 				recorder,
 				healthzServer,
 				config.Winkernel,
+				healthzPort,
 			)
 
 		}

pkg/generated/openapi/zz_generated.openapi.go

@@ -49808,8 +49808,24 @@ func schema_k8sio_kube_proxy_config_v1alpha1_KubeProxyWinkernelConfiguration(ref
 							Format:      "",
 						},
 					},
+					"rootHnsEndpointName": {
+						SchemaProps: spec.SchemaProps{
+							Description: "RootHnsEndpointName is the name of hnsendpoint that is attached to l2bridge for root network namespace",
+							Default:     "",
+							Type:        []string{"string"},
+							Format:      "",
+						},
+					},
+					"forwardHealthCheckVip": {
+						SchemaProps: spec.SchemaProps{
+							Description: "ForwardHealthCheckVip forwards service VIP for health check port on Windows",
+							Default:     false,
+							Type:        []string{"boolean"},
+							Format:      "",
+						},
+					},
 				},
-				Required: []string{"networkName", "sourceVip", "enableDSR"},
+				Required: []string{"networkName", "sourceVip", "enableDSR", "rootHnsEndpointName", "forwardHealthCheckVip"},
 			},
 		},
 	}

pkg/proxy/apis/config/scheme/testdata/KubeProxyConfiguration/after/v1alpha1.yaml

@@ -42,5 +42,7 @@ showHiddenMetricsForVersion: ""
 udpIdleTimeout: 250ms
 winkernel:
   enableDSR: false
+  forwardHealthCheckVip: false
   networkName: ""
+  rootHnsEndpointName: ""
   sourceVip: ""

pkg/proxy/apis/config/scheme/testdata/KubeProxyConfiguration/roundtrip/default/v1alpha1.yaml

@@ -42,5 +42,7 @@ showHiddenMetricsForVersion: ""
 udpIdleTimeout: 250ms
 winkernel:
   enableDSR: false
+  forwardHealthCheckVip: false
   networkName: ""
+  rootHnsEndpointName: ""
   sourceVip: ""

pkg/proxy/apis/config/types.go

@@ -99,6 +99,12 @@ type KubeProxyWinkernelConfiguration struct {
 	// enableDSR tells kube-proxy whether HNS policies should be created
 	// with DSR
 	EnableDSR bool
+	// RootHnsEndpointName is the name of hnsendpoint that is attached to
+	// l2bridge for root network namespace
+	RootHnsEndpointName string
+	// ForwardHealthCheckVip forwards service VIP for health check port on
+	// Windows
+	ForwardHealthCheckVip bool
 }
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object

pkg/proxy/apis/config/v1alpha1/zz_generated.conversion.go

@@ -262,6 +262,8 @@ func autoConvert_v1alpha1_KubeProxyWinkernelConfiguration_To_config_KubeProxyWin
 	out.NetworkName = in.NetworkName
 	out.SourceVip = in.SourceVip
 	out.EnableDSR = in.EnableDSR
+	out.RootHnsEndpointName = in.RootHnsEndpointName
+	out.ForwardHealthCheckVip = in.ForwardHealthCheckVip
 	return nil
 }
 
@@ -274,6 +276,8 @@ func autoConvert_config_KubeProxyWinkernelConfiguration_To_v1alpha1_KubeProxyWin
 	out.NetworkName = in.NetworkName
 	out.SourceVip = in.SourceVip
 	out.EnableDSR = in.EnableDSR
+	out.RootHnsEndpointName = in.RootHnsEndpointName
+	out.ForwardHealthCheckVip = in.ForwardHealthCheckVip
 	return nil
 }
 

pkg/proxy/winkernel/hnsV1.go

@@ -33,6 +33,7 @@ type HostNetworkService interface {
 	getNetworkByName(name string) (*hnsNetworkInfo, error)
 	getEndpointByID(id string) (*endpointsInfo, error)
 	getEndpointByIpAddress(ip string, networkName string) (*endpointsInfo, error)
+	getEndpointByName(id string) (*endpointsInfo, error)
 	createEndpoint(ep *endpointsInfo, networkName string) (*endpointsInfo, error)
 	deleteEndpoint(hnsID string) error
 	getLoadBalancer(endpoints []endpointsInfo, flags loadBalancerFlags, sourceVip string, vip string, protocol uint16, internalPort uint16, externalPort uint16) (*loadBalancerInfo, error)
@@ -62,8 +63,9 @@ func (hns hnsV1) getEndpointByID(id string) (*endpointsInfo, error) {
 		return nil, err
 	}
 	return &endpointsInfo{
-		ip:         hnsendpoint.IPAddress.String(),
+		ip: hnsendpoint.IPAddress.String(),
-		isLocal:    !hnsendpoint.IsRemoteEndpoint, //TODO: Change isLocal to isRemote
+		//TODO: Change isLocal to isRemote
+		isLocal:    !hnsendpoint.IsRemoteEndpoint,
 		macAddress: hnsendpoint.MacAddress,
 		hnsID:      hnsendpoint.Id,
 		hns:        hns,
@@ -108,6 +110,23 @@ func (hns hnsV1) getEndpointByIpAddress(ip string, networkName string) (*endpoin
 
 	return nil, fmt.Errorf("Endpoint %v not found on network %s", ip, networkName)
 }
+
+func (hns hnsV1) getEndpointByName(name string) (*endpointsInfo, error) {
+	hnsendpoint, err := hcsshim.GetHNSEndpointByName(name)
+	if err != nil {
+		klog.ErrorS(err, "failed to get HNS endpoint by name", "name", name)
+		return nil, err
+	}
+	return &endpointsInfo{
+		ip: hnsendpoint.IPAddress.String(),
+		//TODO: Change isLocal to isRemote
+		isLocal:    !hnsendpoint.IsRemoteEndpoint,
+		macAddress: hnsendpoint.MacAddress,
+		hnsID:      hnsendpoint.Id,
+		hns:        hns,
+	}, nil
+}
+
 func (hns hnsV1) createEndpoint(ep *endpointsInfo, networkName string) (*endpointsInfo, error) {
 	hnsNetwork, err := hcsshim.GetHNSNetworkByName(networkName)
 	if err != nil {

pkg/proxy/winkernel/hnsV2.go

@@ -114,6 +114,19 @@ func (hns hnsV2) getEndpointByIpAddress(ip string, networkName string) (*endpoin
 
 	return nil, fmt.Errorf("Endpoint %v not found on network %s", ip, networkName)
 }
+func (hns hnsV2) getEndpointByName(name string) (*endpointsInfo, error) {
+	hnsendpoint, err := hcn.GetEndpointByName(name)
+	if err != nil {
+		return nil, err
+	}
+	return &endpointsInfo{ //TODO: fill out PA
+		ip:         hnsendpoint.IpConfigurations[0].IpAddress,
+		isLocal:    uint32(hnsendpoint.Flags&hcn.EndpointFlagsRemoteEndpoint) == 0, //TODO: Change isLocal to isRemote
+		macAddress: hnsendpoint.MacAddress,
+		hnsID:      hnsendpoint.Id,
+		hns:        hns,
+	}, nil
+}
 func (hns hnsV2) createEndpoint(ep *endpointsInfo, networkName string) (*endpointsInfo, error) {
 	hnsNetwork, err := hcn.GetNetworkByName(networkName)
 	if err != nil {

pkg/proxy/winkernel/hns_test.go

@@ -26,6 +26,8 @@ import (
 
 	"strings"
 	"testing"
+
+	"github.com/google/go-cmp/cmp"
 )
 
 const (
@@ -56,12 +58,12 @@ func TestGetEndpointByID(t *testing.T) {
 	testGetEndpointByID(t, hnsV1)
 	testGetEndpointByID(t, hnsV2)
 }
-func TestGetEndpointByIpAddress(t *testing.T) {
+func TestGetEndpointByIpAddressAndName(t *testing.T) {
 	hnsV1 := hnsV1{}
 	hnsV2 := hnsV2{}
 
-	testGetEndpointByIpAddress(t, hnsV1)
+	testGetEndpointByIpAddressAndName(t, hnsV1)
-	testGetEndpointByIpAddress(t, hnsV2)
+	testGetEndpointByIpAddressAndName(t, hnsV2)
 }
 func TestCreateEndpointLocal(t *testing.T) {
 	hnsV1 := hnsV1{}
@@ -165,7 +167,7 @@ func testGetEndpointByID(t *testing.T, hns HostNetworkService) {
 		t.Error(err)
 	}
 }
-func testGetEndpointByIpAddress(t *testing.T, hns HostNetworkService) {
+func testGetEndpointByIpAddressAndName(t *testing.T, hns HostNetworkService) {
 	Network := mustTestNetwork(t)
 
 	ipConfig := &hcn.IpConfig{
@@ -195,6 +197,15 @@ func testGetEndpointByIpAddress(t *testing.T, hns HostNetworkService) {
 		t.Errorf("%v does not match %v", endpoint.ip, Endpoint.IpConfigurations[0].IpAddress)
 	}
 
+	endpoint, err = hns.getEndpointByName(Endpoint.Name)
+	if err != nil {
+		t.Error(err)
+	}
+	diff := cmp.Diff(endpoint, Endpoint)
+	if diff != "" {
+		t.Errorf("getEndpointByName(%s) returned a different endpoint. Diff: %s ", Endpoint.Name, diff)
+	}
+
 	err = Endpoint.Delete()
 	if err != nil {
 		t.Error(err)

pkg/proxy/winkernel/proxier.go

@@ -87,8 +87,9 @@ type externalIPInfo struct {
 }
 
 type loadBalancerIngressInfo struct {
-	ip    string
+	ip               string
-	hnsID string
+	hnsID            string
+	healthCheckHnsID string
 }
 
 type loadBalancerInfo struct {
@@ -548,6 +549,10 @@ type Proxier struct {
 	hostMac           string
 	isDSR             bool
 	supportedFeatures hcn.SupportedFeatures
+	healthzPort       int
+
+	forwardHealthCheckVip bool
+	rootHnsEndpointName   string
 }
 
 type localPort struct {
@@ -593,6 +598,7 @@ func NewProxier(
 	recorder events.EventRecorder,
 	healthzServer healthcheck.ProxierHealthUpdater,
 	config config.KubeProxyWinkernelConfiguration,
+	healthzPort int,
 ) (*Proxier, error) {
 	masqueradeValue := 1 << uint(masqueradeBit)
 	masqueradeMark := fmt.Sprintf("%#08x/%#08x", masqueradeValue, masqueradeValue)
@@ -684,24 +690,27 @@ func NewProxier(
 
 	isIPv6 := netutils.IsIPv6(nodeIP)
 	proxier := &Proxier{
-		endPointsRefCount:   make(endPointsReferenceCountMap),
+		endPointsRefCount:     make(endPointsReferenceCountMap),
-		serviceMap:          make(proxy.ServiceMap),
+		serviceMap:            make(proxy.ServiceMap),
-		endpointsMap:        make(proxy.EndpointsMap),
+		endpointsMap:          make(proxy.EndpointsMap),
-		masqueradeAll:       masqueradeAll,
+		masqueradeAll:         masqueradeAll,
-		masqueradeMark:      masqueradeMark,
+		masqueradeMark:        masqueradeMark,
-		clusterCIDR:         clusterCIDR,
+		clusterCIDR:           clusterCIDR,
-		hostname:            hostname,
+		hostname:              hostname,
-		nodeIP:              nodeIP,
+		nodeIP:                nodeIP,
-		recorder:            recorder,
+		recorder:              recorder,
-		serviceHealthServer: serviceHealthServer,
+		serviceHealthServer:   serviceHealthServer,
-		healthzServer:       healthzServer,
+		healthzServer:         healthzServer,
-		hns:                 hns,
+		hns:                   hns,
-		network:             *hnsNetworkInfo,
+		network:               *hnsNetworkInfo,
-		sourceVip:           sourceVip,
+		sourceVip:             sourceVip,
-		hostMac:             hostMac,
+		hostMac:               hostMac,
-		isDSR:               isDSR,
+		isDSR:                 isDSR,
-		supportedFeatures:   supportedFeatures,
+		supportedFeatures:     supportedFeatures,
-		isIPv6Mode:          isIPv6,
+		isIPv6Mode:            isIPv6,
+		healthzPort:           healthzPort,
+		rootHnsEndpointName:   config.RootHnsEndpointName,
+		forwardHealthCheckVip: config.ForwardHealthCheckVip,
 	}
 
 	ipFamily := v1.IPv4Protocol
@@ -730,18 +739,19 @@ func NewDualStackProxier(
 	recorder events.EventRecorder,
 	healthzServer healthcheck.ProxierHealthUpdater,
 	config config.KubeProxyWinkernelConfiguration,
+	healthzPort int,
 ) (proxy.Provider, error) {
 
 	// Create an ipv4 instance of the single-stack proxier
 	ipv4Proxier, err := NewProxier(syncPeriod, minSyncPeriod, masqueradeAll, masqueradeBit,
-		clusterCIDR, hostname, nodeIP[0], recorder, healthzServer, config)
+		clusterCIDR, hostname, nodeIP[0], recorder, healthzServer, config, healthzPort)
 
 	if err != nil {
 		return nil, fmt.Errorf("unable to create ipv4 proxier: %v, hostname: %s, clusterCIDR : %s, nodeIP:%v", err, hostname, clusterCIDR, nodeIP[0])
 	}
 
 	ipv6Proxier, err := NewProxier(syncPeriod, minSyncPeriod, masqueradeAll, masqueradeBit,
-		clusterCIDR, hostname, nodeIP[1], recorder, healthzServer, config)
+		clusterCIDR, hostname, nodeIP[1], recorder, healthzServer, config, healthzPort)
 	if err != nil {
 		return nil, fmt.Errorf("unable to create ipv6 proxier: %v, hostname: %s, clusterCIDR : %s, nodeIP:%v", err, hostname, clusterCIDR, nodeIP[1])
 	}
@@ -796,6 +806,10 @@ func (svcInfo *serviceInfo) deleteAllHnsLoadBalancerPolicy() {
 	for _, lbIngressIP := range svcInfo.loadBalancerIngressIPs {
 		hns.deleteLoadBalancer(lbIngressIP.hnsID)
 		lbIngressIP.hnsID = ""
+		if lbIngressIP.healthCheckHnsID != "" {
+			hns.deleteLoadBalancer(lbIngressIP.healthCheckHnsID)
+			lbIngressIP.healthCheckHnsID = ""
+		}
 	}
 }
 
@@ -988,6 +1002,11 @@ func (proxier *Proxier) syncProxyRules() {
 	hnsNetworkName := proxier.network.name
 	hns := proxier.hns
 
+	var gatewayHnsendpoint *endpointsInfo
+	if proxier.forwardHealthCheckVip {
+		gatewayHnsendpoint, _ = hns.getEndpointByName(proxier.rootHnsEndpointName)
+	}
+
 	prevNetworkID := proxier.network.id
 	updatedNetwork, err := hns.getNetworkByName(hnsNetworkName)
 	if updatedNetwork == nil || updatedNetwork.id != prevNetworkID || isNetworkNotFoundError(err) {
@@ -1319,7 +1338,30 @@ func (proxier *Proxier) syncProxyRules() {
 			} else {
 				klog.V(3).InfoS("Skipped creating Hns LoadBalancer for loadBalancer Ingress resources", "lbIngressIP", lbIngressIP)
 			}
+			lbIngressIP.hnsID = hnsLoadBalancer.hnsID
+			klog.V(3).InfoS("Hns LoadBalancer resource created for loadBalancer Ingress resources", "lbIngressIP", lbIngressIP)
 
+			if proxier.forwardHealthCheckVip && gatewayHnsendpoint != nil {
+				nodeport := proxier.healthzPort
+				if svcInfo.HealthCheckNodePort() != 0 {
+					nodeport = svcInfo.HealthCheckNodePort()
+				}
+				hnsHealthCheckLoadBalancer, err := hns.getLoadBalancer(
+					[]endpointsInfo{*gatewayHnsendpoint},
+					loadBalancerFlags{isDSR: false, useMUX: svcInfo.preserveDIP, preserveDIP: svcInfo.preserveDIP},
+					sourceVip,
+					lbIngressIP.ip,
+					Enum(svcInfo.Protocol()),
+					uint16(nodeport),
+					uint16(nodeport),
+				)
+				if err != nil {
+					klog.ErrorS(err, "Policy creation failed")
+					continue
+				}
+				lbIngressIP.healthCheckHnsID = hnsHealthCheckLoadBalancer.hnsID
+				klog.V(3).InfoS("Hns Health Check LoadBalancer resource created for loadBalancer Ingress resources", "ip", lbIngressIP)
+			}
 		}
 		svcInfo.policyApplied = true
 		klog.V(2).InfoS("Policy successfully applied for service", "serviceInfo", svcInfo)

pkg/proxy/winkernel/proxier_test.go

@@ -2,7 +2,7 @@
 // +build windows
 
 /*
-Copyright 2018 The Kubernetes Authors.
+Copyright 2021 The Kubernetes Authors.
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
@@ -73,6 +73,15 @@ func (hns fakeHNS) getEndpointByID(id string) (*endpointsInfo, error) {
 	return nil, nil
 }
 
+func (hns fakeHNS) getEndpointByName(name string) (*endpointsInfo, error) {
+	return &endpointsInfo{
+		isLocal:    true,
+		macAddress: macAddress,
+		hnsID:      guid,
+		hns:        hns,
+	}, nil
+}
+
 func (hns fakeHNS) getEndpointByIpAddress(ip string, networkName string) (*endpointsInfo, error) {
 	_, ipNet, _ := netutils.ParseCIDRSloppy(destinationPrefix)
 
@@ -699,7 +708,6 @@ func TestCreateLoadBalancer(t *testing.T) {
 			t.Errorf("%v does not match %v", svcInfo.hnsID, guid)
 		}
 	}
-
 }
 
 func TestCreateDsrLoadBalancer(t *testing.T) {
@@ -717,6 +725,7 @@ func TestCreateDsrLoadBalancer(t *testing.T) {
 		Port:           "p80",
 		Protocol:       v1.ProtocolTCP,
 	}
+	lbIP := "11.21.31.41"
 
 	makeServiceMap(proxier,
 		makeTestService(svcPortName.Namespace, svcPortName.Name, func(svc *v1.Service) {
@@ -729,6 +738,9 @@ func TestCreateDsrLoadBalancer(t *testing.T) {
 				Protocol: v1.ProtocolTCP,
 				NodePort: int32(svcNodePort),
 			}}
+			svc.Status.LoadBalancer.Ingress = []v1.LoadBalancerIngress{{
+				IP: lbIP,
+			}}
 		}),
 	)
 	tcpProtocol := v1.ProtocolTCP
@@ -761,6 +773,11 @@ func TestCreateDsrLoadBalancer(t *testing.T) {
 		if svcInfo.localTrafficDSR != true {
 			t.Errorf("Failed to create DSR loadbalancer with local traffic policy")
 		}
+		if len(svcInfo.loadBalancerIngressIPs) == 0 {
+			t.Errorf("svcInfo does not have any loadBalancerIngressIPs, %+v", svcInfo)
+		} else if svcInfo.loadBalancerIngressIPs[0].healthCheckHnsID != guid {
+			t.Errorf("The Hns Loadbalancer HealthCheck Id %v does not match %v. ServicePortName %q", svcInfo.loadBalancerIngressIPs[0].healthCheckHnsID, guid, svcPortName.String())
+		}
 	}
 }
 

staging/src/k8s.io/kube-proxy/config/v1alpha1/types.go

@@ -95,6 +95,12 @@ type KubeProxyWinkernelConfiguration struct {
 	// enableDSR tells kube-proxy whether HNS policies should be created
 	// with DSR
 	EnableDSR bool `json:"enableDSR"`
+	// RootHnsEndpointName is the name of hnsendpoint that is attached to
+	// l2bridge for root network namespace
+	RootHnsEndpointName string `json:"rootHnsEndpointName"`
+	// ForwardHealthCheckVip forwards service VIP for health check port on
+	// Windows
+	ForwardHealthCheckVip bool `json:"forwardHealthCheckVip"`
 }
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object