Merge pull request #389 from jcmoraisjr/jm-fix-drain-port

Fix port number lookup of terminating pods

pkg/converters/ingress/ingress.go

@@ -287,9 +287,14 @@ func (c *converter) addEndpoints(svc *api.Service, svcPort *api.ServicePort, bac
 			return err
 		}
 		for _, pod := range pods {
-			// TODO need to find the correct pod's port number
-			ep := backend.AcquireEndpoint(pod.Status.PodIP, svcPort.TargetPort.IntValue(), pod.Namespace+"/"+pod.Name)
-			ep.Weight = 0
+			targetPort := convutils.FindContainerPort(pod, svcPort)
+			if targetPort > 0 {
+				ep := backend.AcquireEndpoint(pod.Status.PodIP, targetPort, pod.Namespace+"/"+pod.Name)
+				ep.Weight = 0
+			} else {
+				c.logger.Warn("skipping endpoint %s of service %s/%s: port '%s' was not found",
+					pod.Status.PodIP, svc.Namespace, svc.Name, svcPort.TargetPort.String())
+			}
 		}
 	}
 	return nil

pkg/converters/ingress/ingress_test.go

@@ -276,14 +276,15 @@ func TestSyncDrainSupport(t *testing.T) {
 	c := setup(t)
 	defer c.teardown()
 
-	svc, ep := c.createSvc1("default/echo", "8080", "172.17.1.101,172.17.1.102")
+	svc, ep := c.createSvc1("default/echo", "http:8080:http", "172.17.1.101,172.17.1.102")
 	svcName := svc.Namespace + "/" + svc.Name
 	ss := &ep.Subsets[0]
 	addr := ss.Addresses
 	ss.Addresses = []api.EndpointAddress{addr[0]}
 	ss.NotReadyAddresses = []api.EndpointAddress{addr[1]}
-	pod := c.createPod1("default/echo-xxxxx", "172.17.1.103")
-	c.cache.TermPodList[svcName] = []*api.Pod{pod}
+	pod1 := c.createPod1("default/echo-xxxxx", "172.17.1.103", "http:8080")
+	pod2 := c.createPod1("default/echo-yyyyy", "172.17.1.104", "none:8080")
+	c.cache.TermPodList[svcName] = []*api.Pod{pod1, pod2}
 
 	c.SyncDef(
 		map[string]string{"drain-support": "true"},
@@ -294,10 +295,10 @@ func TestSyncDrainSupport(t *testing.T) {
 - hostname: echo.example.com
   paths:
   - path: /
-    backend: default_echo_8080
+    backend: default_echo_http
 `)
 	c.compareConfigBack(`
-- id: default_echo_8080
+- id: default_echo_http
   endpoints:
   - ip: 172.17.1.101
     port: 8080
@@ -312,6 +313,8 @@ func TestSyncDrainSupport(t *testing.T) {
   - ip: 172.17.0.99
     port: 8080
 `)
+
+	c.logger.CompareLogging("WARN skipping endpoint 172.17.1.104 of service default/echo: port 'http' was not found")
 }
 
 func TestSyncRootPathLast(t *testing.T) {
@@ -1195,15 +1198,21 @@ func (c *testConfig) createSvc1(name, port, endpoints string) (*api.Service, *ap
 	return svc, ep
 }
 
-func (c *testConfig) createPod1(name, ip string) *api.Pod {
+func (c *testConfig) createPod1(name, ip, port string) *api.Pod {
 	pname := strings.Split(name, "/")
+	pport := strings.Split(port, ":")
 
 	pod := c.createObject(`
 apiVersion: v1
 kind: Pod
 metadata:
   name: ` + pname[1] + `
   namespace: ` + pname[0] + `
+spec:
+  containers:
+  - ports:
+    - name: ` + pport[0] + `
+      containerPort: ` + pport[1] + `
 status:
   podIP: ` + ip).(*api.Pod)
 

pkg/converters/utils/services.go

@@ -46,6 +46,26 @@ func FindServicePort(svc *api.Service, servicePort string) *api.ServicePort {
 	return nil
 }
 
+// FindContainerPort Find the container's port number of a known servicePort
+// Search criteria:
+// 1. svcPort.TargetPort is a number: this is the right container's port
+// 2. svcPort.TargetPort is a named port (not a number): find a container's port with that name and use its ContainerPort
+// If targetPort is neither a valid port number nor a declared named port, return zero which means that the port was not found
+func FindContainerPort(pod *api.Pod, svcPort *api.ServicePort) int {
+	if targetPort := svcPort.TargetPort.IntValue(); targetPort > 0 {
+		return targetPort
+	}
+	portName := svcPort.TargetPort.String()
+	for _, c := range pod.Spec.Containers {
+		for _, port := range c.Ports {
+			if port.Protocol == svcPort.Protocol && port.Name == portName {
+				return int(port.ContainerPort)
+			}
+		}
+	}
+	return 0
+}
+
 // Endpoint ...
 type Endpoint struct {
 	IP        string