Introduce an upstream-hash-by annotation to support consistent hashing by nginx variable or text

pkg/ingress/controller/annotations.go

@@ -41,6 +41,7 @@ import (
 	"k8s.io/ingress-nginx/pkg/ingress/annotations/sessionaffinity"
 	"k8s.io/ingress-nginx/pkg/ingress/annotations/snippet"
 	"k8s.io/ingress-nginx/pkg/ingress/annotations/sslpassthrough"
+	"k8s.io/ingress-nginx/pkg/ingress/annotations/upstreamhashby"
 	"k8s.io/ingress-nginx/pkg/ingress/annotations/upstreamvhost"
 	"k8s.io/ingress-nginx/pkg/ingress/annotations/vtsfilterkey"
 	"k8s.io/ingress-nginx/pkg/ingress/errors"
@@ -82,6 +83,7 @@ func newAnnotationExtractor(cfg extractorConfig) annotationExtractor {
 			"Alias":                alias.NewParser(),
 			"ClientBodyBufferSize": clientbodybuffersize.NewParser(),
 			"DefaultBackend":       defaultbackend.NewParser(cfg),
+			"UpstreamHashBy":       upstreamhashby.NewParser(),
 			"UpstreamVhost":        upstreamvhost.NewParser(),
 			"VtsFilterKey":         vtsfilterkey.NewParser(),
 			"ServerSnippet":        serversnippet.NewParser(),
@@ -131,6 +133,7 @@ const (
 	clientBodyBufferSize = "ClientBodyBufferSize"
 	certificateAuth      = "CertificateAuth"
 	serverSnippet        = "ServerSnippet"
+	upstreamHashBy       = "UpstreamHashBy"
 )
 
 func (e *annotationExtractor) ServiceUpstream(ing *extensions.Ingress) bool {
@@ -189,3 +192,8 @@ func (e *annotationExtractor) ServerSnippet(ing *extensions.Ingress) string {
 	val, _ := e.annotations[serverSnippet].Parse(ing)
 	return val.(string)
 }
+
+func (e *annotationExtractor) UpstreamHashBy(ing *extensions.Ingress) string {
+	val, _ := e.annotations[upstreamHashBy].Parse(ing)
+	return val.(string)
+}

pkg/ingress/controller/annotations_test.go

@@ -37,6 +37,7 @@ const (
 	annotationAffinityType       = "ingress.kubernetes.io/affinity"
 	annotationAffinityCookieName = "ingress.kubernetes.io/session-cookie-name"
 	annotationAffinityCookieHash = "ingress.kubernetes.io/session-cookie-hash"
+	annotationUpstreamHashBy     = "ingress.kubernetes.io/upstream-hash-by"
 )
 
 type mockCfg struct {
@@ -233,6 +234,30 @@ func TestSSLPassthrough(t *testing.T) {
 	}
 }
 
+func TestUpstreamHashBy(t *testing.T) {
+	ec := newAnnotationExtractor(mockCfg{})
+	ing := buildIngress()
+
+	fooAnns := []struct {
+		annotations map[string]string
+		er          string
+	}{
+		{map[string]string{annotationUpstreamHashBy: "$request_uri"}, "$request_uri"},
+		{map[string]string{annotationUpstreamHashBy: "false"}, "false"},
+		{map[string]string{annotationUpstreamHashBy + "_no": "true"}, ""},
+		{map[string]string{}, ""},
+		{nil, ""},
+	}
+
+	for _, foo := range fooAnns {
+		ing.SetAnnotations(foo.annotations)
+		r := ec.UpstreamHashBy(ing)
+		if r != foo.er {
+			t.Errorf("Returned %v but expected %v", r, foo.er)
+		}
+	}
+}
+
 func TestAffinitySession(t *testing.T) {
 	ec := newAnnotationExtractor(mockCfg{})
 	ing := buildIngress()

pkg/ingress/controller/controller.go

@@ -707,6 +707,7 @@ func (ic *GenericController) createUpstreams(data []*extensions.Ingress, du *ing
 		secUpstream := ic.annotations.SecureUpstream(ing)
 		hz := ic.annotations.HealthCheck(ing)
 		serviceUpstream := ic.annotations.ServiceUpstream(ing)
+		upstreamHashBy := ic.annotations.UpstreamHashBy(ing)
 
 		var defBackend string
 		if ing.Spec.Backend != nil {
@@ -767,6 +768,10 @@ func (ic *GenericController) createUpstreams(data []*extensions.Ingress, du *ing
 					upstreams[name].SecureCACert = secUpstream.CACert
 				}
 
+				if upstreams[name].UpstreamHashBy == "" {
+					upstreams[name].UpstreamHashBy = upstreamHashBy
+				}
+
 				svcKey := fmt.Sprintf("%v/%v", ing.GetNamespace(), path.Backend.ServiceName)
 
 				// Add the service cluster endpoint as the upstream instead of individual endpoints