Merge pull request #55740 from rjtsdl/jiren-fixazurelb

pkg/cloudprovider/providers/azure/BUILD

@@ -13,6 +13,7 @@ go_library(
         "azure_backoff.go",
         "azure_blobDiskController.go",
         "azure_controllerCommon.go",
+        "azure_fakes.go",
         "azure_file.go",
         "azure_instance_metadata.go",
         "azure_instances.go",
@@ -48,6 +49,7 @@ go_library(
         "//vendor/k8s.io/api/core/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/errors:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/wait:go_default_library",
         "//vendor/k8s.io/client-go/util/flowcontrol:go_default_library",
     ],
@@ -60,10 +62,14 @@ go_test(
     library = ":go_default_library",
     deps = [
         "//pkg/api/v1/service:go_default_library",
+        "//pkg/kubelet/apis:go_default_library",
+        "//vendor/github.com/Azure/azure-sdk-for-go/arm/compute:go_default_library",
         "//vendor/github.com/Azure/azure-sdk-for-go/arm/network:go_default_library",
         "//vendor/github.com/Azure/go-autorest/autorest/to:go_default_library",
         "//vendor/k8s.io/api/core/v1:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
+        "//vendor/k8s.io/client-go/util/flowcontrol:go_default_library",
     ],
 )
 

pkg/cloudprovider/providers/azure/azure_instances.go

@@ -48,19 +48,10 @@ func (az *Cloud) NodeAddresses(name types.NodeName) ([]v1.NodeAddress, error) {
 		}
 		return addresses, nil
 	}
-	ip, err := az.getIPForMachine(name)
+	ip, err := az.GetIPForMachineWithRetry(name)
 	if err != nil {
-		if az.CloudProviderBackoff {
-			glog.V(2).Infof("NodeAddresses(%s) backing off", name)
-			ip, err = az.GetIPForMachineWithRetry(name)
-			if err != nil {
-				glog.V(2).Infof("NodeAddresses(%s) abort backoff", name)
-				return nil, err
-			}
-		} else {
-			glog.Errorf("error: az.NodeAddresses, az.getIPForMachine(%s), err=%v", name, err)
-			return nil, err
-		}
+		glog.V(2).Infof("NodeAddresses(%s) abort backoff", name)
+		return nil, err
 	}
 
 	return []v1.NodeAddress{
@@ -199,39 +190,6 @@ func (az *Cloud) CurrentNodeName(hostname string) (types.NodeName, error) {
 	return types.NodeName(hostname), nil
 }
 
-func (az *Cloud) listAllNodesInResourceGroup() ([]compute.VirtualMachine, error) {
-	allNodes := []compute.VirtualMachine{}
-
-	az.operationPollRateLimiter.Accept()
-	glog.V(10).Infof("VirtualMachinesClient.List(%s): start", az.ResourceGroup)
-	result, err := az.VirtualMachinesClient.List(az.ResourceGroup)
-	glog.V(10).Infof("VirtualMachinesClient.List(%s): end", az.ResourceGroup)
-	if err != nil {
-		glog.Errorf("error: az.listAllNodesInResourceGroup(), az.VirtualMachinesClient.List(%s), err=%v", az.ResourceGroup, err)
-		return nil, err
-	}
-
-	morePages := (result.Value != nil && len(*result.Value) > 1)
-
-	for morePages {
-		allNodes = append(allNodes, *result.Value...)
-
-		az.operationPollRateLimiter.Accept()
-		glog.V(10).Infof("VirtualMachinesClient.ListAllNextResults(%v): start", az.ResourceGroup)
-		result, err = az.VirtualMachinesClient.ListAllNextResults(result)
-		glog.V(10).Infof("VirtualMachinesClient.ListAllNextResults(%v): end", az.ResourceGroup)
-		if err != nil {
-			glog.Errorf("error: az.listAllNodesInResourceGroup(), az.VirtualMachinesClient.ListAllNextResults(%v), err=%v", result, err)
-			return nil, err
-		}
-
-		morePages = (result.Value != nil && len(*result.Value) > 1)
-	}
-
-	return allNodes, nil
-
-}
-
 // mapNodeNameToVMName maps a k8s NodeName to an Azure VM Name
 // This is a simple string cast.
 func mapNodeNameToVMName(nodeName types.NodeName) string {

pkg/cloudprovider/providers/azure/azure_loadbalancer.md

+# Azure LoadBalancer
+
+The way azure define LoadBalancer is different with GCE or AWS. Azure's LB can have multiple frontend IP refs. The GCE and AWS can only allow one, if you want more, you better to have another LB. Because of the fact, Public IP is not part of the LB in Azure. NSG is not part of LB in Azure either. However, you cannot delete them in parallel, Public IP can only be delete after LB's frontend IP ref is removed. 
+
+For different Azure Resources, such as LB, Public IP, NSG. They are the same tier azure resources. We need to make sure there is no connection in their own ensure loops. In another words, They would be eventually reconciled regardless of other resources' state. They should only depends on service state.
+
+Despite the ideal philosophy above, we have to face the reality. NSG depends on LB's frontend ip to adjust NSG rules. So when we want to reconcile NSG, the LB should contain the corresponding frontend ip config.
+
+And also, For Azure, we cannot afford to have more than 1 worker of service_controller. Because, different services could operate on the same LB, concurrent execution could result in conflict or unexpected result. For AWS and GCE, they apparently doesn't have the problem, they use one LB per service, no such conflict.
+
+There are two load balancers per availability set internal and external. There is a limit on number of services that can be associated with a single load balancer.
+By default primary load balancer is selected. Services can be annotated to allow auto selection of available load balancers. Service annotations can also be used to provide specific availability sets that host the load balancers. Note that in case of auto selection or specific availability set selection, when the availability set is lost incase of downtime or cluster scale down the services are currently not auto assigned to an available load balancer.
+Service Annotation for Auto and specific load balancer mode
+
+- service.beta.kubernetes.io/azure-load-balancer-mode" (__auto__|as1,as2...)
+
+## Introduce Functions
+
+- reconcileLoadBalancer(clusterName string, service *v1.Service, nodes []*v1.Node, wantLb bool) (*network.LoadBalancer, error)
+  - Go through lb's properties, update based on wantLb
+  - If any change on the lb, no matter if the lb exists or not
+    - Call az cloud to CreateOrUpdate on this lb, or Delete if nothing left
+  - return lb, err
+
+- reconcileSecurityGroup(clusterName string, service *v1.Service, lbIP *string, wantLb bool) (*network.SecurityGroup, error)
+  - Go though NSG' properties, update based on wantLb
+    - Use destinationIPAddress as target address if possible
+    - Consolidate NSG rules if possible
+  - If any change on the NSG, (the NSG should always exists)
+    - Call az cloud to CreateOrUpdate on this NSG
+  - return sg, err
+
+- reconcilePublicIP(clusterName string, service *v1.Service, wantLb bool) (*network.PublicIPAddress, error)
+  - List all the public ip in the resource group
+  - Make sure we only touch Public IP resources has tags[service] = "namespace/serviceName"
+    - skip for wantLb && !isInternal && pipName == desiredPipName
+    - delete other public ip resources if any
+  - if !isInternal && wantLb 
+    - ensure Public IP with desiredPipName exists
+
+- getServiceLoadBalancer(service *v1.Service, clusterName string, nodes []*v1.Node, wantLb bool) (lb, status, exists, error)
+  - gets the loadbalancer for the service if it already exists
+  - If wantLb is TRUE then -it selects a new load balancer, the selction helps distribute the services across load balancers
+  - In case the selected load balancer does not exists it returns network.LoadBalancer struct with added metadata (such as name, location) and existsLB set to FALSE 
+  - By default - cluster default LB is returned
+
+## Define interface behaviors
+
+### GetLoadBalancer
+
+- Get LoadBalancer status, return status, error
+  - return the load balancer status for this service
+  - it will not create or update or delete any resource
+
+### EnsureLoadBalancer
+
+- Reconcile LB for the flipped service
+  - Call reconcileLoadBalancer(clusterName, flipedService, nil, false/* wantLb */)
+- Reconcile Public IP
+  - Call reconcilePublicIP(cluster, service, true)
+- Reconcile LB's related and owned resources, such as FrontEndIPConfig, Rules, Probe.
+  - Call reconcileLoadBalancer(clusterName, service, nodes, true /* wantLb */)
+- Reconcile NSG rules, it need to be called after reconcileLB
+  - Call reconcileSecurityGroup(clusterName, service, lbStatus, true /* wantLb */)
+
+### UpdateLoadBalancer
+
+- Has no difference with EnsureLoadBalancer
+
+### EnsureLoadBalancerDeleted
+
+- Reconcile NSG first, before reconcile LB, because SG need LB to be there
+  - Call reconcileSecurityGroup(clusterName, service, nil, false /* wantLb */)
+- Reconcile LB's related and owned resources, such as FrontEndIPConfig, Rules, Probe.
+  - Call reconcileLoadBalancer(clusterName, service, nodes, false)
+- Reconcile Public IP, public IP needs related LB reconciled first
+  - Call reconcilePublicIP(cluster, service, false)
\ No newline at end of file

pkg/cloudprovider/providers/azure/azure_util.go

@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"hash/crc32"
 	"regexp"
+	"sort"
 	"strconv"
 	"strings"
 
@@ -31,6 +32,7 @@ import (
 	"github.com/Azure/azure-sdk-for-go/arm/network"
 	"github.com/golang/glog"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/sets"
 )
 
 const (
@@ -44,6 +46,12 @@ const (
 	loadBalancerRuleIDTemplate  = "/subscriptions/%s/resourceGroups/%s/providers/Microsoft.Network/loadBalancers/%s/loadBalancingRules/%s"
 	loadBalancerProbeIDTemplate = "/subscriptions/%s/resourceGroups/%s/providers/Microsoft.Network/loadBalancers/%s/probes/%s"
 	securityRuleIDTemplate      = "/subscriptions/%s/resourceGroups/%s/providers/Microsoft.Network/networkSecurityGroups/%s/securityRules/%s"
+
+	// InternalLoadBalancerNameSuffix is load balancer posfix
+	InternalLoadBalancerNameSuffix = "-internal"
+
+	// nodeLabelRole specifies the role of a node
+	nodeLabelRole = "kubernetes.io/role"
 )
 
 var providerIDRE = regexp.MustCompile(`^` + CloudProviderName + `://(?:.*)/Microsoft.Compute/virtualMachines/(.+)$`)
@@ -116,6 +124,143 @@ func (az *Cloud) getSecurityRuleID(securityRuleName string) string {
 		securityRuleName)
 }
 
+// returns the full identifier of a publicIPAddress.
+func (az *Cloud) getpublicIPAddressID(pipName string) string {
+	return fmt.Sprintf(
+		publicIPAddressIDTemplate,
+		az.SubscriptionID,
+		az.ResourceGroup,
+		pipName)
+}
+
+// getLoadBalancerAvailabilitySetNames selects all possible availability sets for
+// service load balancer, if the service has no loadbalancer mode annotaion returns the
+// primary availability set if service annotation for loadbalancer availability set
+// exists then return the eligible a availability set
+func (az *Cloud) getLoadBalancerAvailabilitySetNames(service *v1.Service, nodes []*v1.Node) (availabilitySetNames *[]string, err error) {
+	hasMode, isAuto, serviceAvailabilitySetNames := getServiceLoadBalancerMode(service)
+	if !hasMode {
+		// no mode specified in service annotation default to PrimaryAvailabilitySetName
+		availabilitySetNames = &[]string{az.Config.PrimaryAvailabilitySetName}
+		return availabilitySetNames, nil
+	}
+	availabilitySetNames, err = az.getAgentPoolAvailabiliySets(nodes)
+	if err != nil {
+		glog.Errorf("az.getLoadBalancerAvailabilitySetNames - getAgentPoolAvailabiliySets failed err=(%v)", err)
+		return nil, err
+	}
+	if len(*availabilitySetNames) == 0 {
+		glog.Errorf("az.getLoadBalancerAvailabilitySetNames - No availability sets found for nodes in the cluster, node count(%d)", len(nodes))
+		return nil, fmt.Errorf("No availability sets found for nodes, node count(%d)", len(nodes))
+	}
+	// sort the list to have deterministic selection
+	sort.Strings(*availabilitySetNames)
+	if !isAuto {
+		if serviceAvailabilitySetNames == nil || len(serviceAvailabilitySetNames) == 0 {
+			return nil, fmt.Errorf("service annotation for LoadBalancerMode is empty, it should have __auto__ or availability sets value")
+		}
+		// validate availability set exists
+		var found bool
+		for sasx := range serviceAvailabilitySetNames {
+			for asx := range *availabilitySetNames {
+				if strings.EqualFold((*availabilitySetNames)[asx], serviceAvailabilitySetNames[sasx]) {
+					found = true
+					serviceAvailabilitySetNames[sasx] = (*availabilitySetNames)[asx]
+					break
+				}
+			}
+			if !found {
+				glog.Errorf("az.getLoadBalancerAvailabilitySetNames - Availability set (%s) in service annotation not found", serviceAvailabilitySetNames[sasx])
+				return nil, fmt.Errorf("availability set (%s) - not found", serviceAvailabilitySetNames[sasx])
+			}
+		}
+		availabilitySetNames = &serviceAvailabilitySetNames
+	}
+
+	return availabilitySetNames, nil
+}
+
+// lists the virtual machines for for the resource group and then builds
+// a list of availability sets that match the nodes available to k8s
+func (az *Cloud) getAgentPoolAvailabiliySets(nodes []*v1.Node) (agentPoolAvailabilitySets *[]string, err error) {
+	vms, err := az.VirtualMachineClientListWithRetry()
+	if err != nil {
+		glog.Errorf("az.getNodeAvailabilitySet - VirtualMachineClientListWithRetry failed, err=%v", err)
+		return nil, err
+	}
+	vmNameToAvailabilitySetID := make(map[string]string, len(vms))
+	for vmx := range vms {
+		vm := vms[vmx]
+		if vm.AvailabilitySet != nil {
+			vmNameToAvailabilitySetID[*vm.Name] = *vm.AvailabilitySet.ID
+		}
+	}
+	availabilitySetIDs := sets.NewString()
+	agentPoolAvailabilitySets = &[]string{}
+	for nx := range nodes {
+		nodeName := (*nodes[nx]).Name
+		if isMasterNode(nodes[nx]) {
+			continue
+		}
+		asID, ok := vmNameToAvailabilitySetID[nodeName]
+		if !ok {
+			glog.Errorf("az.getNodeAvailabilitySet - Node(%s) has no availability sets", nodeName)
+			return nil, fmt.Errorf("Node (%s) - has no availability sets", nodeName)
+		}
+		if availabilitySetIDs.Has(asID) {
+			// already added in the list
+			continue
+		}
+		asName, err := getLastSegment(asID)
+		if err != nil {
+			glog.Errorf("az.getNodeAvailabilitySet - Node (%s)- getLastSegment(%s), err=%v", nodeName, asID, err)
+			return nil, err
+		}
+		// AvailabilitySet ID is currently upper cased in a indeterministic way
+		// We want to keep it lower case, before the ID get fixed
+		asName = strings.ToLower(asName)
+
+		*agentPoolAvailabilitySets = append(*agentPoolAvailabilitySets, asName)
+	}
+
+	return agentPoolAvailabilitySets, nil
+}
+
+func (az *Cloud) mapLoadBalancerNameToAvailabilitySet(lbName string, clusterName string) (availabilitySetName string) {
+	availabilitySetName = strings.TrimSuffix(lbName, InternalLoadBalancerNameSuffix)
+	if strings.EqualFold(clusterName, lbName) {
+		availabilitySetName = az.Config.PrimaryAvailabilitySetName
+	}
+
+	return availabilitySetName
+}
+
+// For a load balancer, all frontend ip should reference either a subnet or publicIpAddress.
+// Thus Azure do not allow mixed type (public and internal) load balancer.
+// So we'd have a separate name for internal load balancer.
+// This would be the name for Azure LoadBalancer resource.
+func (az *Cloud) getLoadBalancerName(clusterName string, availabilitySetName string, isInternal bool) string {
+	lbNamePrefix := availabilitySetName
+	if strings.EqualFold(availabilitySetName, az.Config.PrimaryAvailabilitySetName) {
+		lbNamePrefix = clusterName
+	}
+	if isInternal {
+		return fmt.Sprintf("%s%s", lbNamePrefix, InternalLoadBalancerNameSuffix)
+	}
+	return lbNamePrefix
+}
+
+// isMasterNode returns returns true is the node has a master role label.
+// The master role is determined by looking for:
+// * a kubernetes.io/role="master" label
+func isMasterNode(node *v1.Node) bool {
+	if val, ok := node.Labels[nodeLabelRole]; ok && val == "master" {
+		return true
+	}
+
+	return false
+}
+
 // returns the deepest child's identifier from a full identifier string.
 func getLastSegment(ID string) (string, error) {
 	parts := strings.Split(ID, "/")
@@ -179,16 +324,8 @@ func getPrimaryIPConfig(nic network.Interface) (*network.InterfaceIPConfiguratio
 	return nil, fmt.Errorf("failed to determine the determine primary ipconfig. nicname=%q", *nic.Name)
 }
 
-// For a load balancer, all frontend ip should reference either a subnet or publicIpAddress.
-// Thus Azure do not allow mixed type (public and internal) load balancer.
-// So we'd have a separate name for internal load balancer.
-// This would be the name for Azure LoadBalancer resource.
-func getLoadBalancerName(clusterName string, isInternal bool) string {
-	if isInternal {
-		return fmt.Sprintf("%s-internal", clusterName)
-	}
-
-	return clusterName
+func isInternalLoadBalancer(lb *network.LoadBalancer) bool {
+	return strings.HasSuffix(*lb.Name, InternalLoadBalancerNameSuffix)
 }
 
 func getBackendPoolName(clusterName string) string {
@@ -203,6 +340,10 @@ func getLoadBalancerRuleName(service *v1.Service, port v1.ServicePort, subnetNam
 }
 
 func getSecurityRuleName(service *v1.Service, port v1.ServicePort, sourceAddrPrefix string) string {
+	if useSharedSecurityRule(service) {
+		safePrefix := strings.Replace(sourceAddrPrefix, "/", "_", -1)
+		return fmt.Sprintf("shared-%s-%d-%s", port.Protocol, port.Port, safePrefix)
+	}
 	safePrefix := strings.Replace(sourceAddrPrefix, "/", "_", -1)
 	return fmt.Sprintf("%s-%s-%d-%s", getRulePrefix(service), port.Protocol, port.Port, safePrefix)
 }

pkg/cloudprovider/providers/azure/azure_wrap.go

@@ -40,6 +40,19 @@ func checkResourceExistsFromError(err error) (bool, error) {
 	return false, v
 }
 
+// If it is StatusNotFound return nil,
+// Otherwise, return what it is
+func ignoreStatusNotFoundFromError(err error) error {
+	if err == nil {
+		return nil
+	}
+	v, ok := err.(autorest.DetailedError)
+	if ok && v.StatusCode == http.StatusNotFound {
+		return nil
+	}
+	return err
+}
+
 func (az *Cloud) getVirtualMachine(nodeName types.NodeName) (vm compute.VirtualMachine, exists bool, err error) {
 	var realErr error
 
@@ -103,7 +116,6 @@ func (az *Cloud) getSecurityGroup() (sg network.SecurityGroup, exists bool, err
 
 func (az *Cloud) getAzureLoadBalancer(name string) (lb network.LoadBalancer, exists bool, err error) {
 	var realErr error
-
 	az.operationPollRateLimiter.Accept()
 	glog.V(10).Infof("LoadBalancerClient.Get(%s): start", name)
 	lb, err = az.LoadBalancerClient.Get(az.ResourceGroup, name, "")
@@ -121,6 +133,25 @@ func (az *Cloud) getAzureLoadBalancer(name string) (lb network.LoadBalancer, exi
 	return lb, exists, err
 }
 
+func (az *Cloud) listLoadBalancers() (lbListResult network.LoadBalancerListResult, exists bool, err error) {
+	var realErr error
+
+	az.operationPollRateLimiter.Accept()
+	glog.V(10).Infof("LoadBalancerClient.List(%s): start", az.ResourceGroup)
+	lbListResult, err = az.LoadBalancerClient.List(az.ResourceGroup)
+	glog.V(10).Infof("LoadBalancerClient.List(%s): end", az.ResourceGroup)
+	exists, realErr = checkResourceExistsFromError(err)
+	if realErr != nil {
+		return lbListResult, false, realErr
+	}
+
+	if !exists {
+		return lbListResult, false, nil
+	}
+
+	return lbListResult, exists, err
+}
+
 func (az *Cloud) getPublicIPAddress(name string) (pip network.PublicIPAddress, exists bool, err error) {
 	var realErr error