Merge pull request #5125 from SzekeresB/dev/limit-connection-tc

Added tc for limit-connection annotation

test/e2e/annotations/canary.go

@@ -277,7 +277,7 @@ var _ = framework.DescribeAnnotation("canary-*", func() {
 			modIng := framework.NewSingleIngress(host, "/", host, f.Namespace,
 				framework.EchoService, 80, modAnnotations)
 
-			f.EnsureIngress(modIng)
+			f.UpdateIngress(modIng)
 
 			f.WaitForNginxServer(host,
 				func(server string) bool {

test/e2e/annotations/limitconnections.go

@@ -0,0 +1,115 @@
+/*
+Copyright 2020 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 annotations
+
+import (
+	"fmt"
+	"strconv"
+	"strings"
+
+	"github.com/onsi/ginkgo"
+	"github.com/stretchr/testify/assert"
+
+	"k8s.io/ingress-nginx/test/e2e/framework"
+)
+
+var _ = framework.DescribeAnnotation("Annotation - limit-connections", func() {
+	f := framework.NewDefaultFramework("limit-connections")
+
+	ginkgo.BeforeEach(func() {
+		f.NewSlowEchoDeployment()
+	})
+
+	ginkgo.It("should limit-connections", func() {
+		host := "limit-connections"
+
+		ing := framework.NewSingleIngress(host, "/", host, f.Namespace, framework.SlowEchoService, 80, nil)
+		f.EnsureIngress(ing)
+		f.WaitForNginxServer(host, func(server string) bool {
+			return strings.Contains(server, fmt.Sprintf("server_name %s ;", host))
+		})
+
+		// prerequisite
+		configKey := "variables-hash-bucket-size"
+		configValue := "256"
+		f.UpdateNginxConfigMapData(configKey, configValue)
+		f.WaitForNginxConfiguration(func(server string) bool {
+			return strings.Contains(server, fmt.Sprintf("variables_hash_bucket_size %s;", configValue))
+		})
+
+		// limit connections
+		annotations := make(map[string]string)
+		connectionLimit := 8
+		annotations["nginx.ingress.kubernetes.io/limit-connections"] = strconv.Itoa(connectionLimit)
+		ing.SetAnnotations(annotations)
+		f.UpdateIngress(ing)
+		f.WaitForNginxConfiguration(func(server string) bool {
+			return strings.Contains(server, "limit_conn") && strings.Contains(server, fmt.Sprintf("conn %v;", connectionLimit))
+		})
+
+		type asyncResult struct {
+			index  int
+			status int
+		}
+
+		response := make(chan *asyncResult)
+		maxRequestNumber := 20
+		for currentRequestNumber := 0; currentRequestNumber < maxRequestNumber; currentRequestNumber++ {
+			go func(requestNumber int, responeChannel chan *asyncResult) {
+				resp := f.HTTPTestClient().
+					GET("/sleep/10").
+					WithHeader("Host", host).
+					Expect().
+					Raw()
+
+				code := 0
+				if resp != nil {
+					code = resp.StatusCode
+				}
+				responeChannel <- &asyncResult{requestNumber, code}
+			}(currentRequestNumber, response)
+
+		}
+		var results []asyncResult
+		for {
+			res := <-response
+			results = append(results, *res)
+			if len(results) >= maxRequestNumber {
+				break
+			}
+		}
+
+		close(response)
+
+		okRequest := 0
+		failedRequest := 0
+		requestError := 0
+		expectedFail := maxRequestNumber - connectionLimit
+		for _, result := range results {
+			if result.status == 200 {
+				okRequest++
+			} else if result.status == 503 {
+				failedRequest++
+			} else {
+				requestError++
+			}
+		}
+		assert.Equal(ginkgo.GinkgoT(), okRequest, connectionLimit, "expecting the ok (200) requests number should equal the connection limit")
+		assert.Equal(ginkgo.GinkgoT(), failedRequest, expectedFail, "expecting the failed (503) requests number should equal the expected to fail request number")
+		assert.Equal(ginkgo.GinkgoT(), requestError, 0, "expecting the failed (other than 503) requests number should equal zero")
+	})
+})

test/e2e/framework/k8s.go

@@ -61,7 +61,7 @@ func (f *Framework) EnsureConfigMap(configMap *api.ConfigMap) (*api.ConfigMap, e
 	return cm, nil
 }
 
-// EnsureIngress creates an Ingress object or returns it if it already exists.
+// EnsureIngress creates an Ingress object and retunrs it, throws error if it already exists.
 func (f *Framework) EnsureIngress(ingress *networking.Ingress) *networking.Ingress {
 	err := createIngressWithRetries(f.KubeClientSet, f.Namespace, ingress)
 	assert.Nil(ginkgo.GinkgoT(), err, "creating ingress")
@@ -80,7 +80,26 @@ func (f *Framework) EnsureIngress(ingress *networking.Ingress) *networking.Ingre
 	return ing
 }
 
-// EnsureService creates a Service object or returns it if it already exists.
+// UpdateIngress updates an Ingress object and returns the updated object.
+func (f *Framework) UpdateIngress(ingress *networking.Ingress) *networking.Ingress {
+	err := updateIngressWithRetries(f.KubeClientSet, f.Namespace, ingress)
+	assert.Nil(ginkgo.GinkgoT(), err, "updating ingress")
+
+	ing, err := f.KubeClientSet.NetworkingV1beta1().Ingresses(f.Namespace).Get(ingress.Name, metav1.GetOptions{})
+	assert.Nil(ginkgo.GinkgoT(), err, "getting ingress")
+	assert.NotNil(ginkgo.GinkgoT(), ing, "expected an ingress but none returned")
+
+	if ing.Annotations == nil {
+		ing.Annotations = make(map[string]string)
+	}
+
+	// updating an ingress requires a reload.
+	time.Sleep(5 * time.Second)
+
+	return ing
+}
+
+// EnsureService creates a Service object and retunrs it, throws error if it already exists.
 func (f *Framework) EnsureService(service *core.Service) *core.Service {
 	err := createServiceWithRetries(f.KubeClientSet, f.Namespace, service)
 	assert.Nil(ginkgo.GinkgoT(), err, "creating service")
@@ -92,7 +111,7 @@ func (f *Framework) EnsureService(service *core.Service) *core.Service {
 	return s
 }
 
-// EnsureDeployment creates a Deployment object or returns it if it already exists.
+// EnsureDeployment creates a Deployment object and retunrs it, throws error if it already exists.
 func (f *Framework) EnsureDeployment(deployment *appsv1.Deployment) *appsv1.Deployment {
 	err := createDeploymentWithRetries(f.KubeClientSet, f.Namespace, deployment)
 	assert.Nil(ginkgo.GinkgoT(), err, "creating deployment")
@@ -225,9 +244,12 @@ func createDeploymentWithRetries(c kubernetes.Interface, namespace string, obj *
 	}
 	createFunc := func() (bool, error) {
 		_, err := c.AppsV1().Deployments(namespace).Create(obj)
-		if err == nil || k8sErrors.IsAlreadyExists(err) {
+		if err == nil {
 			return true, nil
 		}
+		if k8sErrors.IsAlreadyExists(err) {
+			return false, err
+		}
 		if isRetryableAPIError(err) {
 			return false, nil
 		}
@@ -243,9 +265,12 @@ func createSecretWithRetries(c kubernetes.Interface, namespace string, obj *v1.S
 	}
 	createFunc := func() (bool, error) {
 		_, err := c.CoreV1().Secrets(namespace).Create(obj)
-		if err == nil || k8sErrors.IsAlreadyExists(err) {
+		if err == nil {
 			return true, nil
 		}
+		if k8sErrors.IsAlreadyExists(err) {
+			return false, err
+		}
 		if isRetryableAPIError(err) {
 			return false, nil
 		}
@@ -260,9 +285,12 @@ func createServiceWithRetries(c kubernetes.Interface, namespace string, obj *v1.
 	}
 	createFunc := func() (bool, error) {
 		_, err := c.CoreV1().Services(namespace).Create(obj)
-		if err == nil || k8sErrors.IsAlreadyExists(err) {
+		if err == nil {
 			return true, nil
 		}
+		if k8sErrors.IsAlreadyExists(err) {
+			return false, err
+		}
 		if isRetryableAPIError(err) {
 			return false, nil
 		}
@@ -278,9 +306,12 @@ func createIngressWithRetries(c kubernetes.Interface, namespace string, obj *net
 	}
 	createFunc := func() (bool, error) {
 		_, err := c.NetworkingV1beta1().Ingresses(namespace).Create(obj)
-		if err == nil || k8sErrors.IsAlreadyExists(err) {
+		if err == nil {
 			return true, nil
 		}
+		if k8sErrors.IsAlreadyExists(err) {
+			return false, err
+		}
 		if isRetryableAPIError(err) {
 			return false, nil
 		}
@@ -290,6 +321,24 @@ func createIngressWithRetries(c kubernetes.Interface, namespace string, obj *net
 	return retryWithExponentialBackOff(createFunc)
 }
 
+func updateIngressWithRetries(c kubernetes.Interface, namespace string, obj *networking.Ingress) error {
+	if obj == nil {
+		return fmt.Errorf("Object provided to create is empty")
+	}
+	updateFunc := func() (bool, error) {
+		_, err := c.NetworkingV1beta1().Ingresses(namespace).Update(obj)
+		if err == nil {
+			return true, nil
+		}
+		if isRetryableAPIError(err) {
+			return false, nil
+		}
+		return false, fmt.Errorf("Failed to update object with non-retriable error: %v", err)
+	}
+
+	return retryWithExponentialBackOff(updateFunc)
+}
+
 const (
 	// Parameters for retrying with exponential backoff.
 	retryBackoffInitialDuration = 100 * time.Millisecond