Merge pull request #981 from chrismoos/service_upstream

Add annotation to allow use of service ClusterIP for NGINX upstream.

Author: aledbf

controllers/nginx/configuration.md

@@ -57,6 +57,7 @@ The following annotations are supported:
 |[ingress.kubernetes.io/proxy-body-size](#custom-max-body-size)|string|
 |[ingress.kubernetes.io/rewrite-target](#rewrite)|URI|
 |[ingress.kubernetes.io/secure-backends](#secure-backends)|true or false|
+|[ingress.kubernetes.io/service-upstream](#service-upstream)|true or false|
 |[ingress.kubernetes.io/session-cookie-name](#cookie-affinity)|string|
 |[ingress.kubernetes.io/session-cookie-hash](#cookie-affinity)|string|
 |[ingress.kubernetes.io/ssl-redirect](#server-side-https-enforcement-through-redirect)|true or false|
@@ -213,6 +214,17 @@ This is possible thanks to the [ngx_stream_ssl_preread_module](https://nginx.org
 
 By default NGINX uses `http` to reach the services. Adding the annotation `ingress.kubernetes.io/secure-backends: "true"` in the Ingress rule changes the protocol to `https`.
 
+### Service Upstream
+
+By default the NGINX ingress controller uses a list of all endpoints (Pod IP/port) in the NGINX upstream configuration. This annotation disables that behavior and instead uses a single upstream in NGINX, the service's Cluster IP and port. This can be desirable for things like zero-downtime deployments as it reduces the need to reload NGINX configuration when Pods come up and down. See issue [#257](https://github.com/kubernetes/ingress/issues/257).
+
+
+#### Known Issues
+
+If the `service-upstream` annotation is specified the following things should be taken into consideration:
+
+* Sticky Sessions will not work as only round-robin load balancing is supported. 
+* The `proxy_next_upstream` directive will not have any effect meaning on error the request will not be dispatched to another upstream.
 
 ### Server-side HTTPS enforcement through redirect
 

core/pkg/ingress/annotations/serviceupstream/main.go

@@ -0,0 +1,38 @@
+/*
+Copyright 2017 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package serviceupstream
+
+import (
+	extensions "k8s.io/client-go/pkg/apis/extensions/v1beta1"
+	"k8s.io/ingress/core/pkg/ingress/annotations/parser"
+)
+
+const (
+	annotationServiceUpstream = "ingress.kubernetes.io/service-upstream"
+)
+
+type serviceUpstream struct {
+}
+
+// NewParser creates a new serviceUpstream annotation parser
+func NewParser() parser.IngressAnnotation {
+	return serviceUpstream{}
+}
+
+func (s serviceUpstream) Parse(ing *extensions.Ingress) (interface{}, error) {
+	return parser.GetBoolAnnotation(annotationServiceUpstream, ing)
+}

core/pkg/ingress/annotations/serviceupstream/main_test.go

@@ -0,0 +1,112 @@
+/*
+Copyright 2017 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package serviceupstream
+
+import (
+	"testing"
+
+	meta_v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
+	api "k8s.io/client-go/pkg/api/v1"
+	extensions "k8s.io/client-go/pkg/apis/extensions/v1beta1"
+)
+
+func buildIngress() *extensions.Ingress {
+	defaultBackend := extensions.IngressBackend{
+		ServiceName: "default-backend",
+		ServicePort: intstr.FromInt(80),
+	}
+
+	return &extensions.Ingress{
+		ObjectMeta: meta_v1.ObjectMeta{
+			Name:      "foo",
+			Namespace: api.NamespaceDefault,
+		},
+		Spec: extensions.IngressSpec{
+			Backend: &extensions.IngressBackend{
+				ServiceName: "default-backend",
+				ServicePort: intstr.FromInt(80),
+			},
+			Rules: []extensions.IngressRule{
+				{
+					Host: "foo.bar.com",
+					IngressRuleValue: extensions.IngressRuleValue{
+						HTTP: &extensions.HTTPIngressRuleValue{
+							Paths: []extensions.HTTPIngressPath{
+								{
+									Path:    "/foo",
+									Backend: defaultBackend,
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+}
+
+func TestIngressAnnotationServiceUpstreamEnabled(t *testing.T) {
+	ing := buildIngress()
+
+	data := map[string]string{}
+	data[annotationServiceUpstream] = "true"
+	ing.SetAnnotations(data)
+
+	val, _ := NewParser().Parse(ing)
+	enabled, ok := val.(bool)
+	if !ok {
+		t.Errorf("expected a bool type")
+	}
+
+	if !enabled {
+		t.Errorf("expected annotation value to be true, got false")
+	}
+}
+
+func TestIngressAnnotationServiceUpstreamSetFalse(t *testing.T) {
+	ing := buildIngress()
+
+	// Test with explicitly set to false
+	data := map[string]string{}
+	data[annotationServiceUpstream] = "false"
+	ing.SetAnnotations(data)
+
+	val, _ := NewParser().Parse(ing)
+	enabled, ok := val.(bool)
+	if !ok {
+		t.Errorf("expected a bool type")
+	}
+
+	if enabled {
+		t.Errorf("expected annotation value to be false, got true")
+	}
+
+	// Test with no annotation specified, should default to false
+	data = map[string]string{}
+	ing.SetAnnotations(data)
+
+	val, _ = NewParser().Parse(ing)
+	enabled, ok = val.(bool)
+	if !ok {
+		t.Errorf("expected a bool type")
+	}
+
+	if enabled {
+		t.Errorf("expected annotation value to be false, got true")
+	}
+}

core/pkg/ingress/controller/annotations.go

@@ -31,6 +31,7 @@ import (
 	"k8s.io/ingress/core/pkg/ingress/annotations/ratelimit"
 	"k8s.io/ingress/core/pkg/ingress/annotations/rewrite"
 	"k8s.io/ingress/core/pkg/ingress/annotations/secureupstream"
+	"k8s.io/ingress/core/pkg/ingress/annotations/serviceupstream"
 	"k8s.io/ingress/core/pkg/ingress/annotations/sessionaffinity"
 	"k8s.io/ingress/core/pkg/ingress/annotations/snippet"
 	"k8s.io/ingress/core/pkg/ingress/annotations/sslpassthrough"
@@ -64,6 +65,7 @@ func newAnnotationExtractor(cfg extractorConfig) annotationExtractor {
 			"RateLimit":            ratelimit.NewParser(),
 			"Redirect":             rewrite.NewParser(cfg),
 			"SecureUpstream":       secureupstream.NewParser(cfg),
+			"ServiceUpstream":      serviceupstream.NewParser(),
 			"SessionAffinity":      sessionaffinity.NewParser(),
 			"SSLPassthrough":       sslpassthrough.NewParser(),
 			"ConfigurationSnippet": snippet.NewParser(),
@@ -104,8 +106,14 @@ const (
 	healthCheck     = "HealthCheck"
 	sslPassthrough  = "SSLPassthrough"
 	sessionAffinity = "SessionAffinity"
+	serviceUpstream = "ServiceUpstream"
 )
 
+func (e *annotationExtractor) ServiceUpstream(ing *extensions.Ingress) bool {
+	val, _ := e.annotations[serviceUpstream].Parse(ing)
+	return val.(bool)
+}
+
 func (e *annotationExtractor) SecureUpstream(ing *extensions.Ingress) *secureupstream.Secure {
 	val, err := e.annotations[secureUpstream].Parse(ing)
 	if err != nil {

core/pkg/ingress/controller/controller.go

@@ -782,6 +782,7 @@ func (ic *GenericController) createUpstreams(data []interface{}) map[string]*ing
 
 		secUpstream := ic.annotations.SecureUpstream(ing)
 		hz := ic.annotations.HealthCheck(ing)
+		serviceUpstream := ic.annotations.ServiceUpstream(ing)
 
 		var defBackend string
 		if ing.Spec.Backend != nil {
@@ -792,13 +793,27 @@ func (ic *GenericController) createUpstreams(data []interface{}) map[string]*ing
 
 			glog.V(3).Infof("creating upstream %v", defBackend)
 			upstreams[defBackend] = newUpstream(defBackend)
-
 			svcKey := fmt.Sprintf("%v/%v", ing.GetNamespace(), ing.Spec.Backend.ServiceName)
-			endps, err := ic.serviceEndpoints(svcKey, ing.Spec.Backend.ServicePort.String(), hz)
-			upstreams[defBackend].Endpoints = append(upstreams[defBackend].Endpoints, endps...)
-			if err != nil {
-				glog.Warningf("error creating upstream %v: %v", defBackend, err)
+
+			// Add the service cluster endpoint as the upstream instead of individual endpoints
+			// if the serviceUpstream annotation is enabled
+			if serviceUpstream {
+				endpoint, err := ic.getServiceClusterEndpoint(svcKey, ing.Spec.Backend)
+				if err != nil {
+					glog.Errorf("Failed to get service cluster endpoint for service %s: %v", svcKey, err)
+				} else {
+					upstreams[defBackend].Endpoints = []ingress.Endpoint{endpoint}
+				}
 			}
+
+			if len(upstreams[defBackend].Endpoints) == 0 {
+				endps, err := ic.serviceEndpoints(svcKey, ing.Spec.Backend.ServicePort.String(), hz)
+				upstreams[defBackend].Endpoints = append(upstreams[defBackend].Endpoints, endps...)
+				if err != nil {
+					glog.Warningf("error creating upstream %v: %v", defBackend, err)
+				}
+			}
+
 		}
 
 		for _, rule := range ing.Spec.Rules {
@@ -827,12 +842,26 @@ func (ic *GenericController) createUpstreams(data []interface{}) map[string]*ing
 				}
 
 				svcKey := fmt.Sprintf("%v/%v", ing.GetNamespace(), path.Backend.ServiceName)
-				endp, err := ic.serviceEndpoints(svcKey, path.Backend.ServicePort.String(), hz)
-				if err != nil {
-					glog.Warningf("error obtaining service endpoints: %v", err)
-					continue
+
+				// Add the service cluster endpoint as the upstream instead of individual endpoints
+				// if the serviceUpstream annotation is enabled
+				if serviceUpstream {
+					endpoint, err := ic.getServiceClusterEndpoint(svcKey, &path.Backend)
+					if err != nil {
+						glog.Errorf("Failed to get service cluster endpoint for service %s: %v", svcKey, err)
+					} else {
+						upstreams[name].Endpoints = []ingress.Endpoint{endpoint}
+					}
+				}
+
+				if len(upstreams[name].Endpoints) == 0 {
+					endp, err := ic.serviceEndpoints(svcKey, path.Backend.ServicePort.String(), hz)
+					if err != nil {
+						glog.Warningf("error obtaining service endpoints: %v", err)
+						continue
+					}
+					upstreams[name].Endpoints = endp
 				}
-				upstreams[name].Endpoints = endp
 
 				s, exists, err := ic.svcLister.Store.GetByKey(svcKey)
 				if err != nil {
@@ -853,6 +882,24 @@ func (ic *GenericController) createUpstreams(data []interface{}) map[string]*ing
 	return upstreams
 }
 
+func (ic *GenericController) getServiceClusterEndpoint(svcKey string, backend *extensions.IngressBackend) (endpoint ingress.Endpoint, err error) {
+	svcObj, svcExists, err := ic.svcLister.Store.GetByKey(svcKey)
+
+	if !svcExists {
+		return endpoint, fmt.Errorf("service %v does not exist", svcKey)
+	}
+
+	svc := svcObj.(*api.Service)
+	if svc.Spec.ClusterIP == "" {
+		return endpoint, fmt.Errorf("No ClusterIP found for service %s", svcKey)
+	}
+
+	endpoint.Address = svc.Spec.ClusterIP
+	endpoint.Port = backend.ServicePort.String()
+
+	return endpoint, err
+}
+
 // serviceEndpoints returns the upstream servers (endpoints) associated
 // to a service.
 func (ic *GenericController) serviceEndpoints(svcKey, backendPort string,