Merge pull request #1068 from bakins/kubernetes-loadbalancer-annotations

Allow setting load balancer method and sticky using service annotations

docs/toml.md

@@ -1061,6 +1061,11 @@ Annotations can be used on containers to override default behaviour for the whol
 
 - `traefik.frontend.rule.type: PathPrefixStrip`: override the default frontend rule type (Default: `PathPrefix`).
 
+Annotations can be used on the Kubernetes service to override default behaviour:
+
+- `traefik.backend.loadbalancer.method=drr`: override the default `wrr` load balancer algorithm
+- `traefik.backend.loadbalancer.sticky=true`: enable backend sticky sessions
+
 You can find here an example [ingress](https://raw.githubusercontent.com/containous/traefik/master/examples/k8s/cheese-ingress.yaml) and [replication controller](https://raw.githubusercontent.com/containous/traefik/master/examples/k8s/traefik.yaml).
 
 ## Consul backend

provider/kubernetes.go

@@ -114,6 +114,10 @@ func (provider *Kubernetes) loadIngresses(k8sClient k8s.Client) (*types.Configur
 				if _, exists := templateObjects.Backends[r.Host+pa.Path]; !exists {
 					templateObjects.Backends[r.Host+pa.Path] = &types.Backend{
 						Servers: make(map[string]types.Server),
+						LoadBalancer: &types.LoadBalancer{
+							Sticky: false,
+							Method: "wrr",
+						},
 					}
 				}
 				if _, exists := templateObjects.Frontends[r.Host+pa.Path]; !exists {
@@ -167,6 +171,12 @@ func (provider *Kubernetes) loadIngresses(k8sClient k8s.Client) (*types.Configur
 					continue
 				}
 
+				if service.Annotations["traefik.backend.loadbalancer.method"] == "drr" {
+					templateObjects.Backends[r.Host+pa.Path].LoadBalancer.Method = "drr"
+				}
+				if service.Annotations["traefik.backend.loadbalancer.sticky"] == "true" {
+					templateObjects.Backends[r.Host+pa.Path].LoadBalancer.Sticky = true
+				}
 				protocol := "http"
 				for _, port := range service.Spec.Ports {
 					if equalPorts(port, pa.Backend.ServicePort) {

provider/kubernetes_test.go

@@ -216,7 +216,10 @@ func TestLoadIngresses(t *testing.T) {
 					},
 				},
 				CircuitBreaker: nil,
-				LoadBalancer:   nil,
+				LoadBalancer: &types.LoadBalancer{
+					Sticky: false,
+					Method: "wrr",
+				},
 			},
 			"bar": {
 				Servers: map[string]types.Server{
@@ -234,7 +237,10 @@ func TestLoadIngresses(t *testing.T) {
 					},
 				},
 				CircuitBreaker: nil,
-				LoadBalancer:   nil,
+				LoadBalancer: &types.LoadBalancer{
+					Sticky: false,
+					Method: "wrr",
+				},
 			},
 		},
 		Frontends: map[string]*types.Frontend{
@@ -564,7 +570,10 @@ func TestGetPassHostHeader(t *testing.T) {
 					},
 				},
 				CircuitBreaker: nil,
-				LoadBalancer:   nil,
+				LoadBalancer: &types.LoadBalancer{
+					Sticky: false,
+					Method: "wrr",
+				},
 			},
 		},
 		Frontends: map[string]*types.Frontend{
@@ -673,7 +682,10 @@ func TestOnlyReferencesServicesFromOwnNamespace(t *testing.T) {
 					},
 				},
 				CircuitBreaker: nil,
-				LoadBalancer:   nil,
+				LoadBalancer: &types.LoadBalancer{
+					Sticky: false,
+					Method: "wrr",
+				},
 			},
 		},
 		Frontends: map[string]*types.Frontend{
@@ -859,7 +871,10 @@ func TestLoadNamespacedIngresses(t *testing.T) {
 					},
 				},
 				CircuitBreaker: nil,
-				LoadBalancer:   nil,
+				LoadBalancer: &types.LoadBalancer{
+					Sticky: false,
+					Method: "wrr",
+				},
 			},
 			"bar": {
 				Servers: map[string]types.Server{
@@ -873,7 +888,10 @@ func TestLoadNamespacedIngresses(t *testing.T) {
 					},
 				},
 				CircuitBreaker: nil,
-				LoadBalancer:   nil,
+				LoadBalancer: &types.LoadBalancer{
+					Sticky: false,
+					Method: "wrr",
+				},
 			},
 		},
 		Frontends: map[string]*types.Frontend{
@@ -1097,7 +1115,10 @@ func TestLoadMultipleNamespacedIngresses(t *testing.T) {
 					},
 				},
 				CircuitBreaker: nil,
-				LoadBalancer:   nil,
+				LoadBalancer: &types.LoadBalancer{
+					Sticky: false,
+					Method: "wrr",
+				},
 			},
 			"bar": {
 				Servers: map[string]types.Server{
@@ -1111,7 +1132,10 @@ func TestLoadMultipleNamespacedIngresses(t *testing.T) {
 					},
 				},
 				CircuitBreaker: nil,
-				LoadBalancer:   nil,
+				LoadBalancer: &types.LoadBalancer{
+					Sticky: false,
+					Method: "wrr",
+				},
 			},
 			"awesome/quix": {
 				Servers: map[string]types.Server{
@@ -1121,7 +1145,10 @@ func TestLoadMultipleNamespacedIngresses(t *testing.T) {
 					},
 				},
 				CircuitBreaker: nil,
-				LoadBalancer:   nil,
+				LoadBalancer: &types.LoadBalancer{
+					Sticky: false,
+					Method: "wrr",
+				},
 			},
 		},
 		Frontends: map[string]*types.Frontend{
@@ -1235,7 +1262,10 @@ func TestHostlessIngress(t *testing.T) {
 					},
 				},
 				CircuitBreaker: nil,
-				LoadBalancer:   nil,
+				LoadBalancer: &types.LoadBalancer{
+					Sticky: false,
+					Method: "wrr",
+				},
 			},
 		},
 		Frontends: map[string]*types.Frontend{
@@ -1258,6 +1288,238 @@ func TestHostlessIngress(t *testing.T) {
 	}
 }
 
+func TestLoadBalancerAnnotation(t *testing.T) {
+	ingresses := []*v1beta1.Ingress{{
+		ObjectMeta: v1.ObjectMeta{
+			Namespace: "testing",
+		},
+		Spec: v1beta1.IngressSpec{
+			Rules: []v1beta1.IngressRule{
+				{
+					Host: "foo",
+					IngressRuleValue: v1beta1.IngressRuleValue{
+						HTTP: &v1beta1.HTTPIngressRuleValue{
+							Paths: []v1beta1.HTTPIngressPath{
+								{
+									Path: "/bar",
+									Backend: v1beta1.IngressBackend{
+										ServiceName: "service1",
+										ServicePort: intstr.FromInt(80),
+									},
+								},
+							},
+						},
+					},
+				},
+				{
+					Host: "bar",
+					IngressRuleValue: v1beta1.IngressRuleValue{
+						HTTP: &v1beta1.HTTPIngressRuleValue{
+							Paths: []v1beta1.HTTPIngressPath{
+								{
+									Backend: v1beta1.IngressBackend{
+										ServiceName: "service2",
+										ServicePort: intstr.FromInt(802),
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}}
+	services := []*v1.Service{
+		{
+			ObjectMeta: v1.ObjectMeta{
+				Name:      "service1",
+				UID:       "1",
+				Namespace: "testing",
+				Annotations: map[string]string{
+					"traefik.backend.loadbalancer.method": "drr",
+				},
+			},
+			Spec: v1.ServiceSpec{
+				ClusterIP: "10.0.0.1",
+				Ports: []v1.ServicePort{
+					{
+						Port: 80,
+					},
+				},
+			},
+		},
+		{
+			ObjectMeta: v1.ObjectMeta{
+				Name:      "service2",
+				UID:       "2",
+				Namespace: "testing",
+				Annotations: map[string]string{
+					"traefik.backend.loadbalancer.sticky": "true",
+				},
+			},
+			Spec: v1.ServiceSpec{
+				ClusterIP: "10.0.0.2",
+				Ports: []v1.ServicePort{
+					{
+						Port: 802,
+					},
+				},
+			},
+		},
+	}
+	endpoints := []*v1.Endpoints{
+		{
+			ObjectMeta: v1.ObjectMeta{
+				Name:      "service1",
+				UID:       "1",
+				Namespace: "testing",
+			},
+			Subsets: []v1.EndpointSubset{
+				{
+					Addresses: []v1.EndpointAddress{
+						{
+							IP: "10.10.0.1",
+						},
+					},
+					Ports: []v1.EndpointPort{
+						{
+							Port: 8080,
+						},
+					},
+				},
+				{
+					Addresses: []v1.EndpointAddress{
+						{
+							IP: "10.21.0.1",
+						},
+					},
+					Ports: []v1.EndpointPort{
+						{
+							Port: 8080,
+						},
+					},
+				},
+			},
+		},
+		{
+			ObjectMeta: v1.ObjectMeta{
+				Name:      "service2",
+				UID:       "2",
+				Namespace: "testing",
+			},
+			Subsets: []v1.EndpointSubset{
+				{
+					Addresses: []v1.EndpointAddress{
+						{
+							IP: "10.15.0.1",
+						},
+					},
+					Ports: []v1.EndpointPort{
+						{
+							Name: "http",
+							Port: 8080,
+						},
+					},
+				},
+				{
+					Addresses: []v1.EndpointAddress{
+						{
+							IP: "10.15.0.2",
+						},
+					},
+					Ports: []v1.EndpointPort{
+						{
+							Name: "http",
+							Port: 8080,
+						},
+					},
+				},
+			},
+		},
+	}
+	watchChan := make(chan interface{})
+	client := clientMock{
+		ingresses: ingresses,
+		services:  services,
+		endpoints: endpoints,
+		watchChan: watchChan,
+	}
+	provider := Kubernetes{}
+	actual, err := provider.loadIngresses(client)
+	if err != nil {
+		t.Fatalf("error %+v", err)
+	}
+
+	expected := &types.Configuration{
+		Backends: map[string]*types.Backend{
+			"foo/bar": {
+				Servers: map[string]types.Server{
+					"http://10.10.0.1:8080": {
+						URL:    "http://10.10.0.1:8080",
+						Weight: 1,
+					},
+					"http://10.21.0.1:8080": {
+						URL:    "http://10.21.0.1:8080",
+						Weight: 1,
+					},
+				},
+				CircuitBreaker: nil,
+				LoadBalancer: &types.LoadBalancer{
+					Method: "drr",
+					Sticky: false,
+				},
+			},
+			"bar": {
+				Servers: map[string]types.Server{
+					"http://10.15.0.1:8080": {
+						URL:    "http://10.15.0.1:8080",
+						Weight: 1,
+					},
+					"http://10.15.0.2:8080": {
+						URL:    "http://10.15.0.2:8080",
+						Weight: 1,
+					},
+				},
+				CircuitBreaker: nil,
+				LoadBalancer: &types.LoadBalancer{
+					Method: "wrr",
+					Sticky: true,
+				},
+			},
+		},
+		Frontends: map[string]*types.Frontend{
+			"foo/bar": {
+				Backend:        "foo/bar",
+				PassHostHeader: true,
+				Priority:       len("/bar"),
+				Routes: map[string]types.Route{
+					"/bar": {
+						Rule: "PathPrefix:/bar",
+					},
+					"foo": {
+						Rule: "Host:foo",
+					},
+				},
+			},
+			"bar": {
+				Backend:        "bar",
+				PassHostHeader: true,
+				Routes: map[string]types.Route{
+					"bar": {
+						Rule: "Host:bar",
+					},
+				},
+			},
+		},
+	}
+	actualJSON, _ := json.Marshal(actual)
+	expectedJSON, _ := json.Marshal(expected)
+
+	if !reflect.DeepEqual(actual, expected) {
+		t.Fatalf("expected %+v, got %+v", string(expectedJSON), string(actualJSON))
+	}
+}
+
 type clientMock struct {
 	ingresses []*v1beta1.Ingress
 	services  []*v1.Service

templates/kubernetes.tmpl

@@ -1,4 +1,9 @@
 [backends]{{range $backendName, $backend := .Backends}}
+    [backends."{{$backendName}}".loadbalancer]
+      method = "{{$backend.LoadBalancer.Method}}"
+      {{if $backend.LoadBalancer.Sticky}}
+          sticky = true
+      {{end}}
     {{range $serverName, $server := $backend.Servers}}
     [backends."{{$backendName}}".servers."{{$serverName}}"]
     url = "{{$server.URL}}"