Add Amazon ELB proxy protocol support

pkg/cloudprovider/providers/aws/aws.go

@@ -69,6 +69,11 @@ const TagNameSubnetPublicELB = "kubernetes.io/role/elb"
 // This lets us define more advanced semantics in future.
 const ServiceAnnotationLoadBalancerInternal = "service.beta.kubernetes.io/aws-load-balancer-internal"
 
+// Annotation used on the service to enable the proxy protocol on an ELB. Right now we only
+// accept the value "*" which means enable the proxy protocol on all ELB backends. In the
+// future we could adjust this to allow setting the proxy protocol only on certain backends.
+const ServiceAnnotationLoadBalancerProxyProtocol = "service.beta.kubernetes.io/aws-load-balancer-proxy-protocol"
+
 // Service annotation requesting a secure listener. Value is a valid certificate ARN.
 // For more, see http://docs.aws.amazon.com/ElasticLoadBalancing/latest/DeveloperGuide/elb-listener-config.html
 // CertARN is an IAM or CM certificate ARN, e.g. arn:aws:acm:us-east-1:123456789012:certificate/12345678-1234-1234-1234-123456789012
@@ -159,6 +164,8 @@ type ELB interface {
 	DescribeLoadBalancers(*elb.DescribeLoadBalancersInput) (*elb.DescribeLoadBalancersOutput, error)
 	RegisterInstancesWithLoadBalancer(*elb.RegisterInstancesWithLoadBalancerInput) (*elb.RegisterInstancesWithLoadBalancerOutput, error)
 	DeregisterInstancesFromLoadBalancer(*elb.DeregisterInstancesFromLoadBalancerInput) (*elb.DeregisterInstancesFromLoadBalancerOutput, error)
+	CreateLoadBalancerPolicy(*elb.CreateLoadBalancerPolicyInput) (*elb.CreateLoadBalancerPolicyOutput, error)
+	SetLoadBalancerPoliciesForBackendServer(*elb.SetLoadBalancerPoliciesForBackendServerInput) (*elb.SetLoadBalancerPoliciesForBackendServerOutput, error)
 
 	DetachLoadBalancerFromSubnets(*elb.DetachLoadBalancerFromSubnetsInput) (*elb.DetachLoadBalancerFromSubnetsOutput, error)
 	AttachLoadBalancerToSubnets(*elb.AttachLoadBalancerToSubnetsInput) (*elb.AttachLoadBalancerToSubnetsOutput, error)
@@ -2178,6 +2185,16 @@ func (s *AWSCloud) EnsureLoadBalancer(apiService *api.Service, hosts []string) (
 		internalELB = true
 	}
 
+	// Determine if we need to set the Proxy protocol policy
+	proxyProtocol := false
+	proxyProtocolAnnotation := apiService.Annotations[ServiceAnnotationLoadBalancerProxyProtocol]
+	if proxyProtocolAnnotation != "" {
+		if proxyProtocolAnnotation != "*" {
+			return nil, fmt.Errorf("annotation %q=%q detected, but the only value supported currently is '*'", ServiceAnnotationLoadBalancerProxyProtocol, proxyProtocolAnnotation)
+		}
+		proxyProtocol = true
+	}
+
 	// Find the subnets that the ELB will live in
 	subnetIDs, err := s.findELBSubnets(internalELB)
 	if err != nil {
@@ -2230,7 +2247,15 @@ func (s *AWSCloud) EnsureLoadBalancer(apiService *api.Service, hosts []string) (
 	securityGroupIDs := []string{securityGroupID}
 
 	// Build the load balancer itself
-	loadBalancer, err := s.ensureLoadBalancer(serviceName, loadBalancerName, listeners, subnetIDs, securityGroupIDs, internalELB)
+	loadBalancer, err := s.ensureLoadBalancer(
+		serviceName,
+		loadBalancerName,
+		listeners,
+		subnetIDs,
+		securityGroupIDs,
+		internalELB,
+		proxyProtocol,
+	)
 	if err != nil {
 		return nil, err
 	}

pkg/cloudprovider/providers/aws/aws_loadbalancer.go

@@ -28,7 +28,9 @@ import (
 	"k8s.io/kubernetes/pkg/util/sets"
 )
 
-func (s *AWSCloud) ensureLoadBalancer(namespacedName types.NamespacedName, loadBalancerName string, listeners []*elb.Listener, subnetIDs []string, securityGroupIDs []string, internalELB bool) (*elb.LoadBalancerDescription, error) {
+const ProxyProtocolPolicyName = "k8s-proxyprotocol-enabled"
+
+func (s *AWSCloud) ensureLoadBalancer(namespacedName types.NamespacedName, loadBalancerName string, listeners []*elb.Listener, subnetIDs []string, securityGroupIDs []string, internalELB, proxyProtocol bool) (*elb.LoadBalancerDescription, error) {
 	loadBalancer, err := s.describeLoadBalancer(loadBalancerName)
 	if err != nil {
 		return nil, err
@@ -62,6 +64,22 @@ func (s *AWSCloud) ensureLoadBalancer(namespacedName types.NamespacedName, loadB
 		if err != nil {
 			return nil, err
 		}
+
+		if proxyProtocol {
+			err = s.createProxyProtocolPolicy(loadBalancerName)
+			if err != nil {
+				return nil, err
+			}
+
+			for _, listener := range listeners {
+				glog.V(2).Infof("Adjusting AWS loadbalancer proxy protocol on node port %d. Setting to true", *listener.InstancePort)
+				err := s.setBackendPolicies(loadBalancerName, *listener.InstancePort, []*string{aws.String(ProxyProtocolPolicyName)})
+				if err != nil {
+					return nil, err
+				}
+			}
+		}
+
 		dirty = true
 	} else {
 		// TODO: Sync internal vs non-internal
@@ -189,6 +207,73 @@ func (s *AWSCloud) ensureLoadBalancer(namespacedName types.NamespacedName, loadB
 				dirty = true
 			}
 		}
+
+		{
+			// Sync proxy protocol state for new and existing listeners
+
+			proxyPolicies := make([]*string, 0)
+			if proxyProtocol {
+				// Ensure the backend policy exists
+
+				// NOTE The documentation for the AWS API indicates we could get an HTTP 400
+				// back if a policy of the same name already exists. However, the aws-sdk does not
+				// seem to return an error to us in these cases. Therefore this will issue an API
+				// request everytime.
+				err := s.createProxyProtocolPolicy(loadBalancerName)
+				if err != nil {
+					return nil, err
+				}
+
+				proxyPolicies = append(proxyPolicies, aws.String(ProxyProtocolPolicyName))
+			}
+
+			foundBackends := make(map[int64]bool)
+			proxyProtocolBackends := make(map[int64]bool)
+			for _, backendListener := range loadBalancer.BackendServerDescriptions {
+				foundBackends[*backendListener.InstancePort] = false
+				proxyProtocolBackends[*backendListener.InstancePort] = proxyProtocolEnabled(backendListener)
+			}
+
+			for _, listener := range listeners {
+				setPolicy := false
+				instancePort := *listener.InstancePort
+
+				if currentState, ok := proxyProtocolBackends[instancePort]; !ok {
+					// This is a new ELB backend so we only need to worry about
+					// potentientally adding a policy and not removing an
+					// existing one
+					setPolicy = proxyProtocol
+				} else {
+					foundBackends[instancePort] = true
+					// This is an existing ELB backend so we need to determine
+					// if the state changed
+					setPolicy = (currentState != proxyProtocol)
+				}
+
+				if setPolicy {
+					glog.V(2).Infof("Adjusting AWS loadbalancer proxy protocol on node port %d. Setting to %t", instancePort, proxyProtocol)
+					err := s.setBackendPolicies(loadBalancerName, instancePort, proxyPolicies)
+					if err != nil {
+						return nil, err
+					}
+					dirty = true
+				}
+			}
+
+			// We now need to figure out if any backend policies need removed
+			// because these old policies will stick around even if there is no
+			// corresponding listener anymore
+			for instancePort, found := range foundBackends {
+				if !found {
+					glog.V(2).Infof("Adjusting AWS loadbalancer proxy protocol on node port %d. Setting to false", instancePort)
+					err := s.setBackendPolicies(loadBalancerName, instancePort, []*string{})
+					if err != nil {
+						return nil, err
+					}
+					dirty = true
+				}
+			}
+		}
 	}
 
 	if dirty {
@@ -308,3 +393,53 @@ func (s *AWSCloud) ensureLoadBalancerInstances(loadBalancerName string, lbInstan
 
 	return nil
 }
+
+func (s *AWSCloud) createProxyProtocolPolicy(loadBalancerName string) error {
+	request := &elb.CreateLoadBalancerPolicyInput{
+		LoadBalancerName: aws.String(loadBalancerName),
+		PolicyName:       aws.String(ProxyProtocolPolicyName),
+		PolicyTypeName:   aws.String("ProxyProtocolPolicyType"),
+		PolicyAttributes: []*elb.PolicyAttribute{
+			{
+				AttributeName:  aws.String("ProxyProtocol"),
+				AttributeValue: aws.String("true"),
+			},
+		},
+	}
+	glog.V(2).Info("Creating proxy protocol policy on load balancer")
+	_, err := s.elb.CreateLoadBalancerPolicy(request)
+	if err != nil {
+		return fmt.Errorf("error creating proxy protocol policy on load balancer: %v", err)
+	}
+
+	return nil
+}
+
+func (s *AWSCloud) setBackendPolicies(loadBalancerName string, instancePort int64, policies []*string) error {
+	request := &elb.SetLoadBalancerPoliciesForBackendServerInput{
+		InstancePort:     aws.Int64(instancePort),
+		LoadBalancerName: aws.String(loadBalancerName),
+		PolicyNames:      policies,
+	}
+	if len(policies) > 0 {
+		glog.V(2).Infof("Adding AWS loadbalancer backend policies on node port %d", instancePort)
+	} else {
+		glog.V(2).Infof("Removing AWS loadbalancer backend policies on node port %d", instancePort)
+	}
+	_, err := s.elb.SetLoadBalancerPoliciesForBackendServer(request)
+	if err != nil {
+		return fmt.Errorf("error adjusting AWS loadbalancer backend policies: %v", err)
+	}
+
+	return nil
+}
+
+func proxyProtocolEnabled(backend *elb.BackendServerDescription) bool {
+	for _, policy := range backend.PolicyNames {
+		if aws.StringValue(policy) == ProxyProtocolPolicyName {
+			return true
+		}
+	}
+
+	return false
+}

pkg/cloudprovider/providers/aws/aws_test.go

@@ -30,6 +30,7 @@ import (
 	"github.com/golang/glog"
 	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/api/unversioned"
+	"k8s.io/kubernetes/pkg/util/sets"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
@@ -487,6 +488,14 @@ func (elb *FakeELB) ConfigureHealthCheck(*elb.ConfigureHealthCheckInput) (*elb.C
 	panic("Not implemented")
 }
 
+func (elb *FakeELB) CreateLoadBalancerPolicy(*elb.CreateLoadBalancerPolicyInput) (*elb.CreateLoadBalancerPolicyOutput, error) {
+	panic("Not implemented")
+}
+
+func (elb *FakeELB) SetLoadBalancerPoliciesForBackendServer(*elb.SetLoadBalancerPoliciesForBackendServerInput) (*elb.SetLoadBalancerPoliciesForBackendServerOutput, error) {
+	panic("Not implemented")
+}
+
 type FakeASG struct {
 	aws *FakeAWSServices
 }
@@ -1302,3 +1311,30 @@ func TestBuildListener(t *testing.T) {
 		}
 	}
 }
+
+func TestProxyProtocolEnabled(t *testing.T) {
+	policies := sets.NewString(ProxyProtocolPolicyName, "FooBarFoo")
+	fakeBackend := &elb.BackendServerDescription{
+		InstancePort: aws.Int64(80),
+		PolicyNames:  stringSetToPointers(policies),
+	}
+	result := proxyProtocolEnabled(fakeBackend)
+	assert.True(t, result, "expected to find %s in %s", ProxyProtocolPolicyName, policies)
+
+	policies = sets.NewString("FooBarFoo")
+	fakeBackend = &elb.BackendServerDescription{
+		InstancePort: aws.Int64(80),
+		PolicyNames: []*string{
+			aws.String("FooBarFoo"),
+		},
+	}
+	result = proxyProtocolEnabled(fakeBackend)
+	assert.False(t, result, "did not expect to find %s in %s", ProxyProtocolPolicyName, policies)
+
+	policies = sets.NewString()
+	fakeBackend = &elb.BackendServerDescription{
+		InstancePort: aws.Int64(80),
+	}
+	result = proxyProtocolEnabled(fakeBackend)
+	assert.False(t, result, "did not expect to find %s in %s", ProxyProtocolPolicyName, policies)
+}