Add max-connections annotation

apis/voyager/v1beta1/annotations.go

@@ -205,6 +205,7 @@ const (
 	HSTSIncludeSubDomains = IngressKey + "/hsts-include-subdomains"
 
 	WhitelistSourceRange = IngressKey + "/whitelist-source-range"
+	MaxConnections       = IngressKey + "/max-connections"
 )
 
 const (
@@ -308,6 +309,11 @@ func (r Ingress) WhitelistSourceRange() string {
 	return GetString(r.Annotations, WhitelistSourceRange)
 }
 
+func (r Ingress) MaxConnections() int {
+	v, _ := GetInt(r.Annotations, MaxConnections)
+	return v
+}
+
 func (r Ingress) ProxyBodySize() string {
 	return GetString(r.Annotations, ProxyBodySize)
 }

hack/docker/voyager/templates/default-backend.cfg

@@ -23,6 +23,6 @@ backend {{ .DefaultBackend.Name }}
 	http-request redirect location http://{{$e.ExternalName}}:{{ $e.Port }} code 301 unless https
 	{{ end -}}
 	{{- else }}
-	server {{ $e.Name }} {{ $e.IP }}:{{ $e.Port -}} {{ if $e.Weight }} weight {{ $e.Weight }}{{ end -}} {{ if $.DefaultBackend.Sticky }} cookie {{ $e.Name }}{{ end -}} {{ if $e.TLSOption }} {{ $e.TLSOption }} {{ end -}}
+	server {{ $e.Name }} {{ $e.IP }}:{{ $e.Port -}} {{ if $e.MaxConnections }} maxconn {{ $e.MaxConnections }} {{ end -}} {{ if $e.Weight }} weight {{ $e.Weight }}{{ end -}} {{ if $.DefaultBackend.Sticky }} cookie {{ $e.Name }}{{ end -}} {{ if $e.TLSOption }} {{ $e.TLSOption }} {{ end -}}
 	{{ end -}}
 	{{ end -}}

hack/docker/voyager/templates/global.cfg

@@ -3,7 +3,7 @@ global
 	stats socket /tmp/haproxy
 	server-state-file global
 	server-state-base /var/state/haproxy/
-	maxconn 4000
+	{{ if .MaxConnections }}maxconn {{ .MaxConnections }}{{ end }}
 	# log using a syslog socket
 	log /dev/log local0 info
 	log /dev/log local0 notice

hack/docker/voyager/templates/http-backend.cfg

@@ -22,7 +22,7 @@ backend {{ $path.Backend.Name }}
 	http-request redirect location {{ if $.OffloadSSL }}https://{{ else }}http://{{ end }}{{$e.ExternalName}}:{{ $e.Port }} code 301
 	{{- end }}
 	{{- else }}
-	server {{ $e.Name }} {{ $e.IP }}:{{ $e.Port -}} {{ if $e.Weight }} weight {{ $e.Weight }} {{ end -}} {{ if $path.Backend.Sticky }} cookie {{ backend_hash $e.Name $index $path.Backend.StickyCookieHash }} {{ end -}} {{ if $e.TLSOption }} {{ $e.TLSOption }} {{ end -}}
+	server {{ $e.Name }} {{ $e.IP }}:{{ $e.Port -}} {{ if $e.MaxConnections }} maxconn {{ $e.MaxConnections }} {{ end -}} {{ if $e.Weight }} weight {{ $e.Weight }} {{ end -}} {{ if $path.Backend.Sticky }} cookie {{ backend_hash $e.Name $index $path.Backend.StickyCookieHash }} {{ end -}} {{ if $e.TLSOption }} {{ $e.TLSOption }} {{ end -}}
 	{{ end -}}
 	{{ end }}
 {{ end -}}

hack/docker/voyager/templates/tcp-backend.cfg

@@ -14,6 +14,6 @@ backend {{ .Backend.Name }}
 	{{- if $e.ExternalName }}
 	server {{ $e.Name }} {{ $e.ExternalName }}:{{ $e.Port -}} {{ if $e.DNSResolver }} {{ if $e.CheckHealth }} check{{ end }} resolvers {{ $e.DNSResolver }} resolve-prefer ipv4{{ end -}} {{ if $e.TLSOption }} {{ $e.TLSOption }} {{ end -}}
 	{{- else }}
-	server {{ $e.Name }} {{ $e.IP }}:{{ $e.Port -}} {{ if $e.Weight }} weight {{ $e.Weight }}{{ end -}} {{ if $e.TLSOption }} {{ $e.TLSOption }} {{ end -}}
+	server {{ $e.Name }} {{ $e.IP }}:{{ $e.Port -}} {{ if $e.MaxConnections }} maxconn {{ $e.MaxConnections }} {{ end -}} {{ if $e.Weight }} weight {{ $e.Weight }}{{ end -}} {{ if $e.TLSOption }} {{ $e.TLSOption }} {{ end -}}
 	{{ end -}}
 	{{ end -}}

pkg/haproxy/template_test.go

@@ -57,6 +57,7 @@ func TestTemplate(t *testing.T) {
 				{Name: "first", IP: "10.244.2.2", Port: "2324"},
 			},
 		},
+		MaxConnections: 3000,
 	}
 	testParsedConfig := TemplateData{
 		SharedInfo: si,
@@ -255,6 +256,22 @@ func TestTemplate(t *testing.T) {
 				FrontendName: "with-whitelist-http",
 				OffloadSSL:   true,
 			},
+			{
+				SharedInfo:   si,
+				FrontendName: "http-with-backend-maxconn",
+				Port:         80,
+				Paths: []*HTTPPath{
+					{
+						Backend: Backend{
+							Name: "backend-maxconn",
+							Endpoints: []*Endpoint{
+								{Name: "first", IP: "10.244.2.1", Port: "2323", MaxConnections: 20, Weight: 2},
+								{Name: "second", IP: "10.244.2.2", Port: "2323", Weight: 5},
+							},
+						},
+					},
+				},
+			},
 		},
 		TCPService: []*TCPService{
 			{

pkg/haproxy/types.go

@@ -29,6 +29,7 @@ type SharedInfo struct {
 	HSTSPreload           bool
 	HSTSIncludeSubDomains bool
 	WhitelistSourceRange  string
+	MaxConnections        int
 }
 
 type StatsInfo struct {
@@ -106,6 +107,7 @@ type Endpoint struct {
 	IP             string
 	Port           string
 	Weight         int
+	MaxConnections int
 	ExternalName   string
 	UseDNSResolver bool
 	DNSResolver    string

pkg/ingress/parser.go

@@ -133,6 +133,9 @@ func (c *controller) getEndpoints(s *apiv1.Service, servicePort *apiv1.ServicePo
 								if val, ok := pod.Annotations[api.BackendWeight]; ok {
 									ep.Weight, _ = strconv.Atoi(val)
 								}
+								if val, ok := pod.Annotations[api.MaxConnections]; ok {
+									ep.MaxConnections, _ = strconv.Atoi(val)
+								}
 							}
 						}
 					}
@@ -215,6 +218,7 @@ func (c *controller) generateConfig() error {
 		HSTSPreload:           c.Ingress.HSTSPreload(),
 		HSTSIncludeSubDomains: c.Ingress.HSTSIncludeSubDomains(),
 		WhitelistSourceRange:  c.Ingress.WhitelistSourceRange(),
+		MaxConnections:        c.Ingress.MaxConnections(),
 	}
 	if c.Opt.CloudProvider == "aws" && c.Ingress.LBType() == api.LBTypeLoadBalancer {
 		si.AcceptProxy = c.Ingress.KeepSourceIP()

test/e2e/ingress_ops.go

@@ -645,4 +645,220 @@ var _ = Describe("IngressOperations", func() {
 			// TODO @ dipta: how to test if whitelist is actually working?
 		})
 	})
+
+	Describe("With Global MaxConnections (1) Specified", func() {
+		BeforeEach(func() {
+			ing.Annotations[api.MaxConnections] = "1"
+			ing.Annotations[api.DefaultsTimeOut] = `{"connect": "300s", "server": "300s"}`
+		})
+
+		It("Should Allow 1 Connection Concurrently", func() {
+			By("Getting HTTP endpoints")
+
+			eps, err := f.Ingress.GetHTTPEndpoints(ing)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(len(eps)).Should(BeNumerically(">=", 1))
+
+			errChan := make(chan error)
+			go func() {
+				// request-1: take 30s to response
+				errChan <- f.Ingress.DoHTTPWithTimeout(framework.NoRetry, 300, "", ing, eps, "GET",
+					"/testpath/ok?delay=30",
+					func(r *testserverclient.Response) bool {
+						return Expect(r.Status).Should(Equal(http.StatusOK)) &&
+							Expect(r.Method).Should(Equal("GET")) &&
+							Expect(r.Path).Should(Equal("/testpath/ok"))
+					})
+			}()
+
+			time.Sleep(time.Second * 5) // to ensure request-1 always hits server before request-2
+
+			// request-2: responses instantaneously
+			err = f.Ingress.DoHTTPWithTimeout(framework.NoRetry, 5, "", ing, eps, "GET",
+				"/testpath/ok",
+				func(r *testserverclient.Response) bool {
+					return Expect(r.Status).Should(Equal(http.StatusOK)) &&
+						Expect(r.Method).Should(Equal("GET")) &&
+						Expect(r.Path).Should(Equal("/testpath/ok"))
+				})
+
+			// request-1 should block request-2 since maxconn = 1
+			// request-2 should be timeout (sleep: 5s + client-timeout: 5s < request-1: 30s)
+			Expect(err).To(HaveOccurred())
+			Expect(<-errChan).NotTo(HaveOccurred()) // check request-1
+
+		})
+	})
+
+	Describe("With Global MaxConnections (2) Specified", func() {
+		BeforeEach(func() {
+			ing.Annotations[api.MaxConnections] = "2"
+			ing.Annotations[api.DefaultsTimeOut] = `{"connect": "300s", "server": "300s"}`
+		})
+
+		It("Should Allow 2 Connections Concurrently", func() {
+			By("Getting HTTP endpoints")
+
+			eps, err := f.Ingress.GetHTTPEndpoints(ing)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(len(eps)).Should(BeNumerically(">=", 1))
+
+			errChan := make(chan error)
+			go func() {
+				// request-1: take 30s to response
+				errChan <- f.Ingress.DoHTTPWithTimeout(framework.NoRetry, 300, "", ing, eps, "GET",
+					"/testpath/ok?delay=30",
+					func(r *testserverclient.Response) bool {
+						return Expect(r.Status).Should(Equal(http.StatusOK)) &&
+							Expect(r.Method).Should(Equal("GET")) &&
+							Expect(r.Path).Should(Equal("/testpath/ok"))
+					})
+			}()
+
+			time.Sleep(time.Second * 5) // to ensure request-1 always hits server before request-2
+
+			// request-2: responses instantaneously
+			err = f.Ingress.DoHTTPWithTimeout(framework.NoRetry, 5, "", ing, eps, "GET",
+				"/testpath/ok",
+				func(r *testserverclient.Response) bool {
+					return Expect(r.Status).Should(Equal(http.StatusOK)) &&
+						Expect(r.Method).Should(Equal("GET")) &&
+						Expect(r.Path).Should(Equal("/testpath/ok"))
+				})
+
+			Expect(err).NotTo(HaveOccurred())       // request-1 should not block request-2 since maxconn = 2
+			Expect(<-errChan).NotTo(HaveOccurred()) // check request-1
+
+		})
+	})
+
+	Describe("With Pod MaxConnections (1) Specified", func() {
+		BeforeEach(func() {
+			meta, err := f.Ingress.CreateResourceWithBackendMaxConn(1)
+			Expect(err).NotTo(HaveOccurred())
+
+			ing.Spec.Rules = []api.IngressRule{
+				{
+					IngressRuleValue: api.IngressRuleValue{
+						HTTP: &api.HTTPIngressRuleValue{
+							Paths: []api.HTTPIngressPath{
+								{
+									Path: "/testpath",
+									Backend: api.HTTPIngressBackend{
+										IngressBackend: api.IngressBackend{
+											ServiceName: meta.Name,
+											ServicePort: intstr.FromInt(80),
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			}
+
+			ing.Annotations[api.DefaultsTimeOut] = `{"connect": "300s", "server": "300s"}`
+		})
+
+		It("Should Allow 1 Connection Concurrently", func() {
+			By("Getting HTTP endpoints")
+
+			eps, err := f.Ingress.GetHTTPEndpoints(ing)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(len(eps)).Should(BeNumerically(">=", 1))
+
+			errChan := make(chan error)
+			go func() {
+				// request-1: take 30s to response
+				errChan <- f.Ingress.DoHTTPWithTimeout(framework.NoRetry, 300, "", ing, eps, "GET",
+					"/testpath/ok?delay=30",
+					func(r *testserverclient.Response) bool {
+						return Expect(r.Status).Should(Equal(http.StatusOK)) &&
+							Expect(r.Method).Should(Equal("GET")) &&
+							Expect(r.Path).Should(Equal("/testpath/ok"))
+					})
+			}()
+
+			time.Sleep(time.Second * 5) // to ensure request-1 always hits server before request-2
+
+			// request-2: responses instantaneously
+			err = f.Ingress.DoHTTPWithTimeout(framework.NoRetry, 5, "", ing, eps, "GET",
+				"/testpath/ok",
+				func(r *testserverclient.Response) bool {
+					return Expect(r.Status).Should(Equal(http.StatusOK)) &&
+						Expect(r.Method).Should(Equal("GET")) &&
+						Expect(r.Path).Should(Equal("/testpath/ok"))
+				})
+
+			// request-1 should block request-2 since maxconn = 1
+			// request-2 should be timeout (sleep: 5s + client-timeout: 5s < request-1: 30s)
+			Expect(err).To(HaveOccurred())
+			Expect(<-errChan).NotTo(HaveOccurred()) // check request-1
+
+		})
+	})
+
+	Describe("With Pod MaxConnections (2) Specified", func() {
+		BeforeEach(func() {
+			meta, err := f.Ingress.CreateResourceWithBackendMaxConn(2)
+			Expect(err).NotTo(HaveOccurred())
+
+			ing.Spec.Rules = []api.IngressRule{
+				{
+					IngressRuleValue: api.IngressRuleValue{
+						HTTP: &api.HTTPIngressRuleValue{
+							Paths: []api.HTTPIngressPath{
+								{
+									Path: "/testpath",
+									Backend: api.HTTPIngressBackend{
+										IngressBackend: api.IngressBackend{
+											ServiceName: meta.Name,
+											ServicePort: intstr.FromInt(80),
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			}
+
+			ing.Annotations[api.DefaultsTimeOut] = `{"connect": "300s", "server": "300s"}`
+		})
+
+		It("Should Allow 2 Connections Concurrently", func() {
+			By("Getting HTTP endpoints")
+
+			eps, err := f.Ingress.GetHTTPEndpoints(ing)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(len(eps)).Should(BeNumerically(">=", 1))
+
+			errChan := make(chan error)
+			go func() {
+				// request-1: take 30s to response
+				errChan <- f.Ingress.DoHTTPWithTimeout(framework.NoRetry, 300, "", ing, eps, "GET",
+					"/testpath/ok?delay=30",
+					func(r *testserverclient.Response) bool {
+						return Expect(r.Status).Should(Equal(http.StatusOK)) &&
+							Expect(r.Method).Should(Equal("GET")) &&
+							Expect(r.Path).Should(Equal("/testpath/ok"))
+					})
+			}()
+
+			time.Sleep(time.Second * 5) // to ensure request-1 always hits server before request-2
+
+			// request-2: responses instantaneously
+			err = f.Ingress.DoHTTPWithTimeout(framework.NoRetry, 5, "", ing, eps, "GET",
+				"/testpath/ok",
+				func(r *testserverclient.Response) bool {
+					return Expect(r.Status).Should(Equal(http.StatusOK)) &&
+						Expect(r.Method).Should(Equal("GET")) &&
+						Expect(r.Path).Should(Equal("/testpath/ok"))
+				})
+
+			Expect(err).NotTo(HaveOccurred())       // request-1 should not block request-2 since maxconn = 2
+			Expect(<-errChan).NotTo(HaveOccurred()) // check request-1
+
+		})
+	})
 })

test/framework/ingress_suite.go

@@ -18,7 +18,7 @@ import (
 )
 
 const (
-	testServerImage = "appscode/test-server:2.0"
+	testServerImage = "appscode/test-server:2.1"
 )
 
 var (
@@ -196,6 +196,21 @@ func (i *ingressInvocation) DoHTTP(retryCount int, host string, ing *api_v1beta1
 	return nil
 }
 
+func (i *ingressInvocation) DoHTTPWithTimeout(retryCount int, timeout int, host string, ing *api_v1beta1.Ingress, eps []string, method, path string, matcher func(resp *testserverclient.Response) bool) error {
+	for _, url := range eps {
+		resp, err := testserverclient.NewTestHTTPClientWithTimeout(url, timeout).WithHost(host).Method(method).Path(path).DoWithRetry(retryCount)
+		if err != nil {
+			return err
+		}
+
+		log.Infoln("HTTP Response received from server", *resp)
+		if !matcher(resp) {
+			return errors.New("Failed to match")
+		}
+	}
+	return nil
+}
+
 func (i *ingressInvocation) DoHTTPWithHeader(retryCount int, ing *api_v1beta1.Ingress, eps []string, method, path string, h map[string]string, matcher func(resp *testserverclient.Response) bool) error {
 	for _, url := range eps {
 		resp, err := testserverclient.NewTestHTTPClient(url).Method(method).Header(h).Path(path).DoWithRetry(retryCount)