Merge pull request #51757 from itowlson/azure-load-balancer-subnet-redux

Automatic merge from submit-queue (batch tested with PRs 50294, 50422, 51757, 52379, 52014). If you want to cherry-pick this change to another branch, please follow the instructions <a href="https://github.com/kubernetes/community/blob/master/contributors/devel/cherry-picks.md">here</a>..

Azure cloud provider: expose services on non-default subnets

**What this PR does / why we need it**: The Azure cloud provider allows users to specify that a service should be exposed on an internal load balancer instead of the default external load balancer.  However, in a VNet environment, such services are currently always exposed on the master subnet.  Where there are multiple subnets in the VNet, it's desirable to be able to expose an internal service on any subnet.  This PR allows this via a new annotation, `service.beta.kubernetes.io/azure-load-balancer-internal-subnet`.

**Which issue this PR fixes**: fixes https://github.com/Azure/acs-engine/issues/1296 (no corresponding issue has been raised in the k8s core repo)

**Special notes for your reviewer**: None

**Release note**:

```release-note
A new service annotation has been added for services of type LoadBalancer on Azure, 
to specify the subnet on which the service's front end IP should be provisioned. The 
annotation is service.beta.kubernetes.io/azure-load-balancer-internal-subnet and its 
value is the subnet name (not the subnet ARM ID).  If omitted, the default is the 
master subnet.  It is ignored if the service is not on Azure, if the type is not 
LoadBalancer, or if the load balancer is not internal.
```

pkg/cloudprovider/providers/azure/azure_loadbalancer.go

@@ -35,6 +35,10 @@ import (
 // ServiceAnnotationLoadBalancerInternal is the annotation used on the service
 const ServiceAnnotationLoadBalancerInternal = "service.beta.kubernetes.io/azure-load-balancer-internal"
 
+// ServiceAnnotationLoadBalancerInternalSubnet is the annotation used on the service
+// to specify what subnet it is exposed on
+const ServiceAnnotationLoadBalancerInternalSubnet = "service.beta.kubernetes.io/azure-load-balancer-internal-subnet"
+
 // GetLoadBalancer returns whether the specified load balancer exists, and
 // if so, what its status is.
 func (az *Cloud) GetLoadBalancer(clusterName string, service *v1.Service) (status *v1.LoadBalancerStatus, exists bool, err error) {
@@ -54,7 +58,7 @@ func (az *Cloud) GetLoadBalancer(clusterName string, service *v1.Service) (statu
 	var lbIP *string
 
 	if isInternal {
-		lbFrontendIPConfigName := getFrontendIPConfigName(service)
+		lbFrontendIPConfigName := getFrontendIPConfigName(service, subnet(service))
 		for _, ipConfiguration := range *lb.FrontendIPConfigurations {
 			if lbFrontendIPConfigName == *ipConfiguration.Name {
 				lbIP = ipConfiguration.PrivateIPAddress
@@ -182,7 +186,11 @@ func (az *Cloud) EnsureLoadBalancer(clusterName string, service *v1.Service, nod
 	var fipConfigurationProperties *network.FrontendIPConfigurationPropertiesFormat
 
 	if isInternal {
-		subnet, existsSubnet, err := az.getSubnet(az.VnetName, az.SubnetName)
+		subnetName := subnet(service)
+		if subnetName == nil {
+			subnetName = &az.SubnetName
+		}
+		subnet, existsSubnet, err := az.getSubnet(az.VnetName, *subnetName)
 		if err != nil {
 			return nil, err
 		}
@@ -366,7 +374,7 @@ func (az *Cloud) cleanupLoadBalancer(clusterName string, service *v1.Service, is
 
 		if !isInternalLb {
 			// Find public ip resource to clean up from IP configuration
-			lbFrontendIPConfigName := getFrontendIPConfigName(service)
+			lbFrontendIPConfigName := getFrontendIPConfigName(service, nil)
 			for _, config := range *lb.FrontendIPConfigurations {
 				if strings.EqualFold(*config.Name, lbFrontendIPConfigName) {
 					if config.PublicIPAddress != nil {
@@ -523,7 +531,7 @@ func (az *Cloud) reconcileLoadBalancer(lb network.LoadBalancer, fipConfiguration
 	isInternal := requiresInternalLoadBalancer(service)
 	lbName := getLoadBalancerName(clusterName, isInternal)
 	serviceName := getServiceName(service)
-	lbFrontendIPConfigName := getFrontendIPConfigName(service)
+	lbFrontendIPConfigName := getFrontendIPConfigName(service, subnet(service))
 	lbFrontendIPConfigID := az.getFrontendIPConfigID(lbName, lbFrontendIPConfigName)
 	lbBackendPoolName := getBackendPoolName(clusterName)
 	lbBackendPoolID := az.getBackendPoolID(lbName, lbBackendPoolName)
@@ -568,13 +576,23 @@ func (az *Cloud) reconcileLoadBalancer(lb network.LoadBalancer, fipConfiguration
 	if !wantLb {
 		for i := len(newConfigs) - 1; i >= 0; i-- {
 			config := newConfigs[i]
-			if strings.EqualFold(*config.Name, lbFrontendIPConfigName) {
+			if serviceOwnsFrontendIP(config, service) {
 				glog.V(3).Infof("reconcile(%s)(%t): lb frontendconfig(%s) - dropping", serviceName, wantLb, lbFrontendIPConfigName)
 				newConfigs = append(newConfigs[:i], newConfigs[i+1:]...)
 				dirtyConfigs = true
 			}
 		}
 	} else {
+		if isInternal {
+			for i := len(newConfigs) - 1; i >= 0; i-- {
+				config := newConfigs[i]
+				if serviceOwnsFrontendIP(config, service) && !strings.EqualFold(*config.Name, lbFrontendIPConfigName) {
+					glog.V(3).Infof("reconcile(%s)(%t): lb frontendconfig(%s) - dropping", serviceName, wantLb, *config.Name)
+					newConfigs = append(newConfigs[:i], newConfigs[i+1:]...)
+					dirtyConfigs = true
+				}
+			}
+		}
 		foundConfig := false
 		for _, config := range newConfigs {
 			if strings.EqualFold(*config.Name, lbFrontendIPConfigName) {
@@ -608,7 +626,7 @@ func (az *Cloud) reconcileLoadBalancer(lb network.LoadBalancer, fipConfiguration
 	var expectedProbes []network.Probe
 	var expectedRules []network.LoadBalancingRule
 	for _, port := range ports {
-		lbRuleName := getLoadBalancerRuleName(service, port)
+		lbRuleName := getLoadBalancerRuleName(service, port, subnet(service))
 
 		transportProto, _, probeProto, err := getProtocolsFromKubernetesProtocol(port.Protocol)
 		if err != nil {
@@ -652,6 +670,7 @@ func (az *Cloud) reconcileLoadBalancer(lb network.LoadBalancer, fipConfiguration
 		if service.Spec.SessionAffinity == v1.ServiceAffinityClientIP {
 			loadDistribution = network.SourceIP
 		}
+
 		expectedRule := network.LoadBalancingRule{
 			Name: &lbRuleName,
 			LoadBalancingRulePropertiesFormat: &network.LoadBalancingRulePropertiesFormat{
@@ -994,3 +1013,13 @@ func requiresInternalLoadBalancer(service *v1.Service) bool {
 
 	return false
 }
+
+func subnet(service *v1.Service) *string {
+	if requiresInternalLoadBalancer(service) {
+		if l, ok := service.Annotations[ServiceAnnotationLoadBalancerInternalSubnet]; ok {
+			return &l
+		}
+	}
+
+	return nil
+}

pkg/cloudprovider/providers/azure/azure_test.go

@@ -67,6 +67,101 @@ func TestReconcileLoadBalancerAddPort(t *testing.T) {
 	validateLoadBalancer(t, lb, svc)
 }
 
+// Test addition of a new service on an internal LB with a subnet.
+func TestReconcileLoadBalancerAddServiceOnInternalSubnet(t *testing.T) {
+	az := getTestCloud()
+	svc := getInternalTestService("servicea", 80)
+	addTestSubnet(t, &svc)
+	configProperties := getTestInternalFipConfigurationProperties(to.StringPtr("TestSubnet"))
+	lb := getTestLoadBalancer()
+	nodes := []*v1.Node{}
+
+	lb, updated, err := az.reconcileLoadBalancer(lb, &configProperties, testClusterName, &svc, nodes)
+	if err != nil {
+		t.Errorf("Unexpected error: %q", err)
+	}
+
+	if !updated {
+		t.Error("Expected the loadbalancer to need an update")
+	}
+
+	// ensure we got a frontend ip configuration
+	if len(*lb.FrontendIPConfigurations) != 1 {
+		t.Error("Expected the loadbalancer to have a frontend ip configuration")
+	}
+
+	validateLoadBalancer(t, lb, svc)
+}
+
+// Test addition of services on an internal LB using both default and explicit subnets.
+func TestReconcileLoadBalancerAddServicesOnMultipleSubnets(t *testing.T) {
+	az := getTestCloud()
+	svc1 := getTestService("service1", v1.ProtocolTCP, 8081)
+	svc2 := getInternalTestService("service2", 8081)
+	addTestSubnet(t, &svc2)
+	configProperties1 := getTestPublicFipConfigurationProperties()
+	configProperties2 := getTestInternalFipConfigurationProperties(to.StringPtr("TestSubnet"))
+	lb := getTestLoadBalancer()
+	nodes := []*v1.Node{}
+
+	lb, updated, err := az.reconcileLoadBalancer(lb, &configProperties1, testClusterName, &svc1, nodes)
+	if err != nil {
+		t.Errorf("Unexpected error reconciling svc1: %q", err)
+	}
+
+	lb, updated, err = az.reconcileLoadBalancer(lb, &configProperties2, testClusterName, &svc2, nodes)
+	if err != nil {
+		t.Errorf("Unexpected error reconciling svc2: %q", err)
+	}
+
+	if !updated {
+		t.Error("Expected the loadbalancer to need an update")
+	}
+
+	// ensure we got a frontend ip configuration for each service
+	if len(*lb.FrontendIPConfigurations) != 2 {
+		t.Error("Expected the loadbalancer to have 2 frontend ip configurations")
+	}
+
+	validateLoadBalancer(t, lb, svc1, svc2)
+}
+
+// Test moving a service exposure from one subnet to another.
+func TestReconcileLoadBalancerEditServiceSubnet(t *testing.T) {
+	az := getTestCloud()
+	svc := getInternalTestService("service1", 8081)
+	addTestSubnet(t, &svc)
+	configProperties := getTestInternalFipConfigurationProperties(to.StringPtr("TestSubnet"))
+	lb := getTestLoadBalancer()
+	nodes := []*v1.Node{}
+
+	lb, updated, err := az.reconcileLoadBalancer(lb, &configProperties, testClusterName, &svc, nodes)
+	if err != nil {
+		t.Errorf("Unexpected error reconciling initial svc: %q", err)
+	}
+
+	validateLoadBalancer(t, lb, svc)
+
+	svc.Annotations[ServiceAnnotationLoadBalancerInternalSubnet] = "NewSubnet"
+	configProperties = getTestInternalFipConfigurationProperties(to.StringPtr("NewSubnet"))
+
+	lb, updated, err = az.reconcileLoadBalancer(lb, &configProperties, testClusterName, &svc, nodes)
+	if err != nil {
+		t.Errorf("Unexpected error reconciling edits to svc: %q", err)
+	}
+
+	if !updated {
+		t.Error("Expected the loadbalancer to need an update")
+	}
+
+	// ensure we got a frontend ip configuration for the service
+	if len(*lb.FrontendIPConfigurations) != 1 {
+		t.Error("Expected the loadbalancer to have 1 frontend ip configuration")
+	}
+
+	validateLoadBalancer(t, lb, svc)
+}
+
 func TestReconcileLoadBalancerNodeHealth(t *testing.T) {
 	az := getTestCloud()
 	svc := getTestService("servicea", v1.ProtocolTCP, 80)
@@ -299,6 +394,17 @@ func getTestPublicFipConfigurationProperties() network.FrontendIPConfigurationPr
 	}
 }
 
+func getTestInternalFipConfigurationProperties(expectedSubnetName *string) network.FrontendIPConfigurationPropertiesFormat {
+	var expectedSubnet *network.Subnet
+	if expectedSubnetName != nil {
+		expectedSubnet = &network.Subnet{Name: expectedSubnetName}
+	}
+	return network.FrontendIPConfigurationPropertiesFormat{
+		PublicIPAddress: &network.PublicIPAddress{ID: to.StringPtr("/this/is/a/public/ip/address/id")},
+		Subnet:          expectedSubnet,
+	}
+}
+
 func getTestService(identifier string, proto v1.Protocol, requestedPorts ...int32) v1.Service {
 	ports := []v1.ServicePort{}
 	for _, port := range requestedPorts {
@@ -337,7 +443,7 @@ func getTestLoadBalancer(services ...v1.Service) network.LoadBalancer {
 
 	for _, service := range services {
 		for _, port := range service.Spec.Ports {
-			ruleName := getLoadBalancerRuleName(&service, port)
+			ruleName := getLoadBalancerRuleName(&service, port, nil)
 			rules = append(rules, network.LoadBalancingRule{
 				Name: to.StringPtr(ruleName),
 				LoadBalancingRulePropertiesFormat: &network.LoadBalancingRulePropertiesFormat{
@@ -406,13 +512,19 @@ func validateLoadBalancer(t *testing.T, loadBalancer network.LoadBalancer, servi
 	expectedRuleCount := 0
 	expectedFrontendIPCount := 0
 	expectedProbeCount := 0
+	expectedFrontendIPs := []ExpectedFrontendIPInfo{}
 	for _, svc := range services {
 		if len(svc.Spec.Ports) > 0 {
 			expectedFrontendIPCount++
+			expectedFrontendIP := ExpectedFrontendIPInfo{
+				Name:   getFrontendIPConfigName(&svc, subnet(&svc)),
+				Subnet: subnet(&svc),
+			}
+			expectedFrontendIPs = append(expectedFrontendIPs, expectedFrontendIP)
 		}
 		for _, wantedRule := range svc.Spec.Ports {
 			expectedRuleCount++
-			wantedRuleName := getLoadBalancerRuleName(&svc, wantedRule)
+			wantedRuleName := getLoadBalancerRuleName(&svc, wantedRule, subnet(&svc))
 			foundRule := false
 			for _, actualRule := range *loadBalancer.LoadBalancingRules {
 				if strings.EqualFold(*actualRule.Name, wantedRuleName) &&
@@ -467,6 +579,13 @@ func validateLoadBalancer(t *testing.T, loadBalancer network.LoadBalancer, servi
 		t.Errorf("Expected the loadbalancer to have %d frontend IPs. Found %d.\n%v", expectedFrontendIPCount, frontendIPCount, loadBalancer.FrontendIPConfigurations)
 	}
 
+	frontendIPs := *loadBalancer.FrontendIPConfigurations
+	for _, expectedFrontendIP := range expectedFrontendIPs {
+		if !expectedFrontendIP.existsIn(frontendIPs) {
+			t.Errorf("Expected the loadbalancer to have frontend IP %s/%s. Found %s", expectedFrontendIP.Name, to.String(expectedFrontendIP.Subnet), describeFIPs(frontendIPs))
+		}
+	}
+
 	lenRules := len(*loadBalancer.LoadBalancingRules)
 	if lenRules != expectedRuleCount {
 		t.Errorf("Expected the loadbalancer to have %d rules. Found %d.\n%v", expectedRuleCount, lenRules, loadBalancer.LoadBalancingRules)
@@ -478,6 +597,44 @@ func validateLoadBalancer(t *testing.T, loadBalancer network.LoadBalancer, servi
 	}
 }
 
+type ExpectedFrontendIPInfo struct {
+	Name   string
+	Subnet *string
+}
+
+func (expected ExpectedFrontendIPInfo) matches(frontendIP network.FrontendIPConfiguration) bool {
+	return strings.EqualFold(expected.Name, to.String(frontendIP.Name)) && strings.EqualFold(to.String(expected.Subnet), to.String(subnetName(frontendIP)))
+}
+
+func (expected ExpectedFrontendIPInfo) existsIn(frontendIPs []network.FrontendIPConfiguration) bool {
+	for _, fip := range frontendIPs {
+		if expected.matches(fip) {
+			return true
+		}
+	}
+	return false
+}
+
+func subnetName(frontendIP network.FrontendIPConfiguration) *string {
+	if frontendIP.Subnet != nil {
+		return frontendIP.Subnet.Name
+	}
+	return nil
+}
+
+func describeFIPs(frontendIPs []network.FrontendIPConfiguration) string {
+	description := ""
+	for _, actualFIP := range frontendIPs {
+		actualSubnetName := ""
+		if actualFIP.Subnet != nil {
+			actualSubnetName = to.String(actualFIP.Subnet.Name)
+		}
+		actualFIPText := fmt.Sprintf("%s/%s ", to.String(actualFIP.Name), actualSubnetName)
+		description = description + actualFIPText
+	}
+	return description
+}
+
 func validateSecurityGroup(t *testing.T, securityGroup network.SecurityGroup, services ...v1.Service) {
 	expectedRuleCount := 0
 	for _, svc := range services {
@@ -887,3 +1044,10 @@ func TestMetadataParsing(t *testing.T) {
 		t.Errorf("Unexpected inequality:\n%#v\nvs\n%#v", network, networkJSON)
 	}
 }
+
+func addTestSubnet(t *testing.T, svc *v1.Service) {
+	if svc.Annotations[ServiceAnnotationLoadBalancerInternal] != "true" {
+		t.Error("Subnet added to non-internal service")
+	}
+	svc.Annotations[ServiceAnnotationLoadBalancerInternalSubnet] = "TestSubnet"
+}

pkg/cloudprovider/providers/azure/azure_util.go

@@ -195,8 +195,11 @@ func getBackendPoolName(clusterName string) string {
 	return clusterName
 }
 
-func getLoadBalancerRuleName(service *v1.Service, port v1.ServicePort) string {
-	return fmt.Sprintf("%s-%s-%d", getRulePrefix(service), port.Protocol, port.Port)
+func getLoadBalancerRuleName(service *v1.Service, port v1.ServicePort, subnetName *string) string {
+	if subnetName == nil {
+		return fmt.Sprintf("%s-%s-%d", getRulePrefix(service), port.Protocol, port.Port)
+	}
+	return fmt.Sprintf("%s-%s-%s-%d", getRulePrefix(service), *subnetName, port.Protocol, port.Port)
 }
 
 func getSecurityRuleName(service *v1.Service, port v1.ServicePort, sourceAddrPrefix string) string {
@@ -224,8 +227,17 @@ func serviceOwnsRule(service *v1.Service, rule string) bool {
 	return strings.HasPrefix(strings.ToUpper(rule), strings.ToUpper(prefix))
 }
 
-func getFrontendIPConfigName(service *v1.Service) string {
-	return cloudprovider.GetLoadBalancerName(service)
+func serviceOwnsFrontendIP(fip network.FrontendIPConfiguration, service *v1.Service) bool {
+	baseName := cloudprovider.GetLoadBalancerName(service)
+	return strings.HasPrefix(*fip.Name, baseName)
+}
+
+func getFrontendIPConfigName(service *v1.Service, subnetName *string) string {
+	baseName := cloudprovider.GetLoadBalancerName(service)
+	if subnetName != nil {
+		return fmt.Sprintf("%s-%s", baseName, *subnetName)
+	}
+	return baseName
 }
 
 // This returns the next available rule priority level for a given set of security rules.