Merge pull request #36557 from brendandburns/azure

Automatic merge from submit-queue

Add support for SourceIP preservation in Azure LBs

@thockin @colemickens now that we are setting DSR for Azure LBs this should "just work"

pkg/cloudprovider/providers/azure/BUILD

@@ -28,6 +28,7 @@ go_library(
     tags = ["automanaged"],
     deps = [
         "//pkg/api:go_default_library",
+        "//pkg/api/service:go_default_library",
         "//pkg/cloudprovider:go_default_library",
         "//pkg/types:go_default_library",
         "//pkg/util/errors:go_default_library",
@@ -51,6 +52,7 @@ go_test(
     tags = ["automanaged"],
     deps = [
         "//pkg/api:go_default_library",
+        "//pkg/api/service:go_default_library",
         "//pkg/types:go_default_library",
         "//vendor:github.com/Azure/azure-sdk-for-go/arm/compute",
         "//vendor:github.com/Azure/azure-sdk-for-go/arm/network",

pkg/cloudprovider/providers/azure/azure_loadbalancer.go

@@ -22,6 +22,7 @@ import (
 	"strings"
 
 	"k8s.io/kubernetes/pkg/api"
+	serviceapi "k8s.io/kubernetes/pkg/api/service"
 	utilerrors "k8s.io/kubernetes/pkg/util/errors"
 
 	"github.com/Azure/azure-sdk-for-go/arm/network"
@@ -339,6 +340,20 @@ func (az *Cloud) reconcileLoadBalancer(lb network.LoadBalancer, pip *network.Pub
 			return lb, false, err
 		}
 
+		if serviceapi.NeedsHealthCheck(service) {
+			podPresencePath, podPresencePort := serviceapi.GetServiceHealthCheckPathPort(service)
+
+			expectedProbes[i] = network.Probe{
+				Name: &lbRuleName,
+				Properties: &network.ProbePropertiesFormat{
+					RequestPath:       to.StringPtr(podPresencePath),
+					Protocol:          network.ProbeProtocolHTTP,
+					Port:              to.Int32Ptr(podPresencePort),
+					IntervalInSeconds: to.Int32Ptr(5),
+					NumberOfProbes:    to.Int32Ptr(2),
+				},
+			}
+		} else {
 			expectedProbes[i] = network.Probe{
 				Name: &lbRuleName,
 				Properties: &network.ProbePropertiesFormat{
@@ -348,6 +363,7 @@ func (az *Cloud) reconcileLoadBalancer(lb network.LoadBalancer, pip *network.Pub
 					NumberOfProbes:    to.Int32Ptr(2),
 				},
 			}
+		}
 
 		expectedRules[i] = network.LoadBalancingRule{
 			Name: &lbRuleName,
@@ -369,7 +385,7 @@ func (az *Cloud) reconcileLoadBalancer(lb network.LoadBalancer, pip *network.Pub
 		}
 	}
 
-	// remove unwated probes
+	// remove unwanted probes
 	dirtyProbes := false
 	var updatedProbes []network.Probe
 	if lb.Properties.Probes != nil {

pkg/cloudprovider/providers/azure/azure_test.go

@@ -22,6 +22,7 @@ import (
 	"testing"
 
 	"k8s.io/kubernetes/pkg/api"
+	serviceapi "k8s.io/kubernetes/pkg/api/service"
 	"k8s.io/kubernetes/pkg/types"
 
 	"github.com/Azure/azure-sdk-for-go/arm/compute"
@@ -56,6 +57,35 @@ func TestReconcileLoadBalancerAddPort(t *testing.T) {
 	validateLoadBalancer(t, lb, svc)
 }
 
+func TestReconcileLoadBalancerNodeHealth(t *testing.T) {
+	az := getTestCloud()
+	svc := getTestService("servicea", 80)
+	svc.Annotations = map[string]string{
+		serviceapi.BetaAnnotationExternalTraffic:     serviceapi.AnnotationValueExternalTrafficLocal,
+		serviceapi.BetaAnnotationHealthCheckNodePort: "32456",
+	}
+	pip := getTestPublicIP()
+	lb := getTestLoadBalancer()
+
+	hosts := []string{}
+
+	lb, updated, err := az.reconcileLoadBalancer(lb, &pip, testClusterName, &svc, hosts)
+	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.Properties.FrontendIPConfigurations) != 1 {
+		t.Error("Expected the loadbalancer to have a frontend ip configuration")
+	}
+
+	validateLoadBalancer(t, lb, svc)
+}
+
 // Test removing all services results in removing the frontend ip configuration
 func TestReconcileLoadBalancerRemoveAllPortsRemovesFrontendConfig(t *testing.T) {
 	az := getTestCloud()
@@ -283,6 +313,18 @@ func validateLoadBalancer(t *testing.T, loadBalancer network.LoadBalancer, servi
 			}
 
 			foundProbe := false
+			if serviceapi.NeedsHealthCheck(&svc) {
+				path, port := serviceapi.GetServiceHealthCheckPathPort(&svc)
+				for _, actualProbe := range *loadBalancer.Properties.Probes {
+					if strings.EqualFold(*actualProbe.Name, wantedRuleName) &&
+						*actualProbe.Properties.Port == port &&
+						*actualProbe.Properties.RequestPath == path &&
+						actualProbe.Properties.Protocol == network.ProbeProtocolHTTP {
+						foundProbe = true
+						break
+					}
+				}
+			} else {
 				for _, actualProbe := range *loadBalancer.Properties.Probes {
 					if strings.EqualFold(*actualProbe.Name, wantedRuleName) &&
 						*actualProbe.Properties.Port == wantedRule.NodePort {
@@ -290,7 +332,11 @@ func validateLoadBalancer(t *testing.T, loadBalancer network.LoadBalancer, servi
 						break
 					}
 				}
+			}
 			if !foundProbe {
+				for _, actualProbe := range *loadBalancer.Properties.Probes {
+					t.Logf("Probe: %s %d", *actualProbe.Name, *actualProbe.Properties.Port)
+				}
 				t.Errorf("Expected loadbalancer probe but didn't find it: %q", wantedRuleName)
 			}
 		}