Merge pull request #3628 from 0robustus1/fix-503-error-generation-on-…

…empty-endpoints

Fix 503 error generation on empty endpoints

rootfs/etc/nginx/lua/balancer.lua

@@ -75,7 +75,8 @@ end
 
 local function sync_backend(backend)
   if not backend.endpoints or #backend.endpoints == 0 then
-    ngx.log(ngx.INFO, string.format("there is no endpoint for backend %s. Skipping...", backend.name))
+    ngx.log(ngx.INFO, string.format("there is no endpoint for backend %s. Removing...", backend.name))
+    balancers[backend.name] = nil
     return
   end
 

test/e2e/framework/deployment.go

@@ -33,7 +33,14 @@ func (f *Framework) NewEchoDeployment() {
 // NewEchoDeploymentWithReplicas creates a new deployment of the echoserver image in a particular namespace. Number of
 // replicas is configurable
 func (f *Framework) NewEchoDeploymentWithReplicas(replicas int32) {
-	f.NewDeployment("http-svc", "gcr.io/kubernetes-e2e-test-images/echoserver:2.2", 8080, replicas)
+	f.NewEchoDeploymentWithNameAndReplicas("http-svc", replicas)
+}
+
+// NewEchoDeploymentWithNameAndReplicas creates a new deployment of the echoserver image in a particular namespace. Number of
+// replicas is configurable and
+// name is configurable
+func (f *Framework) NewEchoDeploymentWithNameAndReplicas(name string, replicas int32) {
+	f.NewDeployment(name, "gcr.io/kubernetes-e2e-test-images/echoserver:2.2", 8080, replicas)
 }
 
 // NewSlowEchoDeployment creates a new deployment of the slow echo server image in a particular namespace.

test/e2e/framework/k8s.go

@@ -144,6 +144,9 @@ func WaitForPodsReady(kubeClientSet kubernetes.Interface, timeout time.Duration,
 
 // WaitForEndpoints waits for a given amount of time until an endpoint contains.
 func WaitForEndpoints(kubeClientSet kubernetes.Interface, timeout time.Duration, name, ns string, expectedEndpoints int) error {
+	if expectedEndpoints == 0 {
+		return nil
+	}
 	return wait.Poll(2*time.Second, timeout, func() (bool, error) {
 		endpoint, err := kubeClientSet.CoreV1().Endpoints(ns).Get(name, metav1.GetOptions{})
 		if k8sErrors.IsNotFound(err) {

test/e2e/lua/dynamic_certificates.go

@@ -57,7 +57,7 @@ var _ = framework.IngressNginxDescribe("Dynamic Certificate", func() {
 	})
 
 	It("picks up the certificate when we add TLS spec to existing ingress", func() {
-		ensureIngress(f, host)
+		ensureIngress(f, host, "http-svc")
 
 		ing, err := f.KubeClientSet.ExtensionsV1beta1().Ingresses(f.IngressController.Namespace).Get(host, metav1.GetOptions{})
 		Expect(err).ToNot(HaveOccurred())

test/e2e/lua/dynamic_configuration.go

@@ -48,7 +48,7 @@ var _ = framework.IngressNginxDescribe("Dynamic Configuration", func() {
 
 	BeforeEach(func() {
 		f.NewEchoDeploymentWithReplicas(1)
-		ensureIngress(f, "foo.com")
+		ensureIngress(f, "foo.com", "http-svc")
 	})
 
 	It("configures balancer Lua middleware correctly", func() {
@@ -92,6 +92,60 @@ var _ = framework.IngressNginxDescribe("Dynamic Configuration", func() {
 			Expect(nginxConfig).Should(Equal(newNginxConfig))
 		})
 
+		It("handles endpoints only changes (down scaling of replicas)", func() {
+			var nginxConfig string
+			f.WaitForNginxConfiguration(func(cfg string) bool {
+				nginxConfig = cfg
+				return true
+			})
+
+			replicas := 2
+			err := framework.UpdateDeployment(f.KubeClientSet, f.IngressController.Namespace, "http-svc", replicas, nil)
+			Expect(err).NotTo(HaveOccurred())
+			time.Sleep(waitForLuaSync * 2)
+
+			ensureRequest(f, "foo.com")
+
+			var newNginxConfig string
+			f.WaitForNginxConfiguration(func(cfg string) bool {
+				newNginxConfig = cfg
+				return true
+			})
+			Expect(nginxConfig).Should(Equal(newNginxConfig))
+
+			err = framework.UpdateDeployment(f.KubeClientSet, f.IngressController.Namespace, "http-svc", 0, nil)
+
+			Expect(err).NotTo(HaveOccurred())
+			time.Sleep(waitForLuaSync * 2)
+
+			ensureRequestWithStatus(f, "foo.com", 503)
+		})
+
+		It("handles endpoints only changes consistently (down scaling of replicas vs. empty service)", func() {
+			deploymentName := "scalingecho"
+			f.NewEchoDeploymentWithNameAndReplicas(deploymentName, 0)
+			createIngress(f, "scaling.foo.com", deploymentName)
+			originalResponseCode := runRequest(f, "scaling.foo.com")
+
+			replicas := 2
+			err := framework.UpdateDeployment(f.KubeClientSet, f.IngressController.Namespace, deploymentName, replicas, nil)
+			Expect(err).NotTo(HaveOccurred())
+			time.Sleep(waitForLuaSync * 2)
+
+			expectedSuccessResponseCode := runRequest(f, "scaling.foo.com")
+
+			replicas = 0
+			err = framework.UpdateDeployment(f.KubeClientSet, f.IngressController.Namespace, deploymentName, replicas, nil)
+			Expect(err).NotTo(HaveOccurred())
+			time.Sleep(waitForLuaSync * 2)
+
+			expectedFailureResponseCode := runRequest(f, "scaling.foo.com")
+
+			Expect(originalResponseCode).To(Equal(503), "Expected empty service to return 503 response.")
+			Expect(expectedFailureResponseCode).To(Equal(503), "Expected downscaled replicaset to return 503 response.")
+			Expect(expectedSuccessResponseCode).To(Equal(200), "Expected intermediate scaled replicaset to return a 200 response.")
+		})
+
 		It("handles an annotation change", func() {
 			var nginxConfig string
 			f.WaitForNginxConfiguration(func(cfg string) bool {
@@ -165,17 +219,23 @@ var _ = framework.IngressNginxDescribe("Dynamic Configuration", func() {
 	})
 })
 
-func ensureIngress(f *framework.Framework, host string) *extensions.Ingress {
-	ing := f.EnsureIngress(framework.NewSingleIngress(host, "/", host, f.IngressController.Namespace, "http-svc", 80,
+func ensureIngress(f *framework.Framework, host string, deploymentName string) *extensions.Ingress {
+	ing := createIngress(f, host, deploymentName)
+	time.Sleep(waitForLuaSync)
+	ensureRequest(f, host)
+
+	return ing
+}
+
+func createIngress(f *framework.Framework, host string, deploymentName string) *extensions.Ingress {
+	ing := f.EnsureIngress(framework.NewSingleIngress(host, "/", host, f.IngressController.Namespace, deploymentName, 80,
 		&map[string]string{"nginx.ingress.kubernetes.io/load-balance": "ewma"}))
 
 	f.WaitForNginxServer(host,
 		func(server string) bool {
 			return strings.Contains(server, fmt.Sprintf("server_name %s ;", host)) &&
 				strings.Contains(server, "proxy_pass http://upstream_balancer;")
 		})
-	time.Sleep(waitForLuaSync)
-	ensureRequest(f, host)
 
 	return ing
 }
@@ -189,6 +249,24 @@ func ensureRequest(f *framework.Framework, host string) {
 	Expect(resp.StatusCode).Should(Equal(http.StatusOK))
 }
 
+func ensureRequestWithStatus(f *framework.Framework, host string, statusCode int) {
+	resp, _, errs := gorequest.New().
+		Get(f.IngressController.HTTPURL).
+		Set("Host", host).
+		End()
+	Expect(errs).Should(BeEmpty())
+	Expect(resp.StatusCode).Should(Equal(statusCode))
+}
+
+func runRequest(f *framework.Framework, host string) int {
+	resp, _, errs := gorequest.New().
+		Get(f.IngressController.HTTPURL).
+		Set("Host", host).
+		End()
+	Expect(errs).Should(BeEmpty())
+	return resp.StatusCode
+}
+
 func ensureHTTPSRequest(url string, host string, expectedDNSName string) {
 	resp, _, errs := gorequest.New().
 		Get(url).