Add support for SourceIP preservation in Azure LBs

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,14 +340,29 @@ func (az *Cloud) reconcileLoadBalancer(lb network.LoadBalancer, pip *network.Pub
 			return lb, false, err
 		}
 
-		expectedProbes[i] = network.Probe{
-			Name: &lbRuleName,
-			Properties: &network.ProbePropertiesFormat{
-				Protocol:          probeProto,
-				Port:              to.Int32Ptr(port.NodePort),
-				IntervalInSeconds: to.Int32Ptr(5),
-				NumberOfProbes:    to.Int32Ptr(2),
-			},
+		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{
+					Protocol:          probeProto,
+					Port:              to.Int32Ptr(port.NodePort),
+					IntervalInSeconds: to.Int32Ptr(5),
+					NumberOfProbes:    to.Int32Ptr(2),
+				},
+			}
 		}
 
 		expectedRules[i] = network.LoadBalancingRule{
@@ -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,14 +313,30 @@ func validateLoadBalancer(t *testing.T, loadBalancer network.LoadBalancer, servi
 			}
 
 			foundProbe := false
-			for _, actualProbe := range *loadBalancer.Properties.Probes {
-				if strings.EqualFold(*actualProbe.Name, wantedRuleName) &&
-					*actualProbe.Properties.Port == wantedRule.NodePort {
-					foundProbe = true
-					break
+			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 {
+						foundProbe = true
+						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)
 			}
 		}