Fix service port lookup

Use the backend port to find the correct ServicePort object from the service. The ServicePort name is used to match the Endpoint port name. Finally, the endpoint port number is the right one. ServicePort name is optional only if there is only one port, so if ServicePort.Name == "", the Endpoint port match, otherwise the svcPort and epPort names should match.

pkg/converters/configmap/tcpservices.go

@@ -73,7 +73,7 @@ func (c *tcpSvcConverter) Sync(tcpservices map[string]string) {
 			continue
 		}
 		svcport := convutils.FindServicePort(service, svc.port)
-		if svcport.IntValue() == 0 {
+		if svcport == nil {
 			c.logger.Warn("skipping TCP service on public port %d: port not found: %s:%s", publicport, svc.name, svc.port)
 			continue
 		}

pkg/converters/helper_test/k8sobjects.go

@@ -26,8 +26,8 @@ import (
 
 // CreateService ...
 func CreateService(name, port, endpoints string) (*api.Service, *api.Endpoints) {
-	sname := strings.Split(name, "/")
-	sport := strings.Split(port, ":")
+	sname := strings.Split(name, "/") // namespace/name of the service
+	sport := strings.Split(port, ":") // numeric-port -or- name:numeric-port -or- name:numeric-port:named-port
 	if len(sport) < 2 {
 		sport = []string{"", port, port}
 	} else if len(sport) < 3 {
@@ -56,7 +56,7 @@ subsets:
 - addresses: []
   ports:
   - name: ` + sport[0] + `
-    port: ` + sport[2] + `
+    port: ` + sport[1] + `
     protocol: TCP`).(*api.Endpoints)
 
 	addr := []api.EndpointAddress{}

pkg/converters/ingress/ingress.go

@@ -23,7 +23,6 @@ import (
 
 	api "k8s.io/api/core/v1"
 	extensions "k8s.io/api/extensions/v1beta1"
-	"k8s.io/apimachinery/pkg/util/intstr"
 
 	"github.com/jcmoraisjr/haproxy-ingress/pkg/converters/ingress/annotations"
 	ingtypes "github.com/jcmoraisjr/haproxy-ingress/pkg/converters/ingress/types"
@@ -218,11 +217,11 @@ func (c *converter) addBackend(source *annotations.Source, hostpath, fullSvcName
 		// from the api.Service object
 		svcPort = svc.Spec.Ports[0].TargetPort.String()
 	}
-	epport := convutils.FindServicePort(svc, svcPort)
-	if epport.String() == "" {
+	port := convutils.FindServicePort(svc, svcPort)
+	if port == nil {
 		return nil, fmt.Errorf("port not found: '%s'", svcPort)
 	}
-	backend := c.haproxy.AcquireBackend(namespace, svcName, epport.String())
+	backend := c.haproxy.AcquireBackend(namespace, svcName, port.TargetPort.String())
 	mapper, found := c.backendAnnotations[backend]
 	if !found {
 		// New backend, initialize with service annotations, giving precedence
@@ -244,13 +243,13 @@ func (c *converter) addBackend(source *annotations.Source, hostpath, fullSvcName
 	// Configure endpoints
 	if !found {
 		if mapper.Get(ingtypes.BackServiceUpstream).Bool() {
-			if addr, err := convutils.CreateSvcEndpoint(svc, epport); err == nil {
+			if addr, err := convutils.CreateSvcEndpoint(svc, port); err == nil {
 				backend.AcquireEndpoint(addr.IP, addr.Port, addr.TargetRef)
 			} else {
 				c.logger.Error("error adding IP of service '%s': %v", fullSvcName, err)
 			}
 		} else {
-			if err := c.addEndpoints(svc, epport, backend); err != nil {
+			if err := c.addEndpoints(svc, port, backend); err != nil {
 				c.logger.Error("error adding endpoints of service '%s': %v", fullSvcName, err)
 			}
 		}
@@ -270,7 +269,7 @@ func (c *converter) addTLS(namespace, secretName string) convtypes.File {
 	return c.options.DefaultSSLFile
 }
 
-func (c *converter) addEndpoints(svc *api.Service, svcPort intstr.IntOrString, backend *hatypes.Backend) error {
+func (c *converter) addEndpoints(svc *api.Service, svcPort *api.ServicePort, backend *hatypes.Backend) error {
 	ready, notReady, err := convutils.CreateEndpoints(c.cache, svc, svcPort)
 	if err != nil {
 		return err
@@ -288,7 +287,8 @@ func (c *converter) addEndpoints(svc *api.Service, svcPort intstr.IntOrString, b
 			return err
 		}
 		for _, pod := range pods {
-			ep := backend.AcquireEndpoint(pod.Status.PodIP, svcPort.IntValue(), pod.Namespace+"/"+pod.Name)
+			// 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
 		}
 	}

pkg/converters/ingress/ingress_test.go

@@ -113,10 +113,10 @@ func TestSyncSvcNamedPort(t *testing.T) {
 	c := setup(t)
 	defer c.teardown()
 
-	c.createSvc1("default/echo", "httpsvc:1001:8080", "172.17.1.101")
+	c.createSvc1("default/echo", "svcport:8080:http", "172.17.1.101")
 	c.Sync(
-		c.createIng1("default/echo1", "echo1.example.com", "/", "echo:httpsvc"),
-		c.createIng1("default/echo2", "echo2.example.com", "/", "echo:1001"),
+		c.createIng1("default/echo1", "echo1.example.com", "/", "echo:http"),
+		c.createIng1("default/echo2", "echo2.example.com", "/", "echo:svcport"),
 		c.createIng1("default/echo3", "echo3.example.com", "/", "echo:8080"),
 		c.createIng1("default/echo4", "echo4.example.com", "/", "echo:9000"),
 	)
@@ -125,21 +125,21 @@ func TestSyncSvcNamedPort(t *testing.T) {
 - hostname: echo1.example.com
   paths:
   - path: /
-    backend: default_echo_8080
+    backend: default_echo_http
 - hostname: echo2.example.com
   paths:
   - path: /
-    backend: default_echo_8080
+    backend: default_echo_http
 - hostname: echo3.example.com
   paths:
   - path: /
-    backend: default_echo_8080
+    backend: default_echo_http
 - hostname: echo4.example.com
   paths: []
 `)
 
 	c.compareConfigBack(`
-- id: default_echo_8080
+- id: default_echo_http
   endpoints:
   - ip: 172.17.1.101
     port: 8080

pkg/converters/utils/services.go

@@ -22,33 +22,28 @@ import (
 	"strconv"
 
 	api "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/util/intstr"
 
 	"github.com/jcmoraisjr/haproxy-ingress/pkg/converters/types"
 )
 
 // FindServicePort ...
-func FindServicePort(svc *api.Service, servicePort string) intstr.IntOrString {
+func FindServicePort(svc *api.Service, servicePort string) *api.ServicePort {
 	for _, port := range svc.Spec.Ports {
-		if port.Name == servicePort {
-			return port.TargetPort
-		}
-	}
-	for _, port := range svc.Spec.Ports {
-		if port.TargetPort.String() == servicePort {
-			return port.TargetPort
+		if port.Name == servicePort || port.TargetPort.String() == servicePort {
+			return &port
 		}
 	}
 	svcPortNumber, err := strconv.ParseInt(servicePort, 10, 0)
 	if err != nil {
-		return intstr.FromString("")
+		return nil
 	}
+	svcPort := int32(svcPortNumber)
 	for _, port := range svc.Spec.Ports {
-		if port.Port == int32(svcPortNumber) {
-			return port.TargetPort
+		if port.Port == svcPort {
+			return &port
 		}
 	}
-	return intstr.FromString("")
+	return nil
 }
 
 // Endpoint ...
@@ -59,7 +54,7 @@ type Endpoint struct {
 }
 
 // CreateEndpoints ...
-func CreateEndpoints(cache types.Cache, svc *api.Service, svcPort intstr.IntOrString) (ready, notReady []*Endpoint, err error) {
+func CreateEndpoints(cache types.Cache, svc *api.Service, svcPort *api.ServicePort) (ready, notReady []*Endpoint, err error) {
 	if svc.Spec.Type == api.ServiceTypeExternalName {
 		ready, err := createEndpointsExternalName(svc, svcPort)
 		return ready, nil, err
@@ -73,38 +68,45 @@ func CreateEndpoints(cache types.Cache, svc *api.Service, svcPort intstr.IntOrSt
 }
 
 // CreateSvcEndpoint ...
-func CreateSvcEndpoint(svc *api.Service, svcPort intstr.IntOrString) (endpoint *Endpoint, err error) {
-	port := svcPort.IntValue()
+func CreateSvcEndpoint(svc *api.Service, svcPort *api.ServicePort) (endpoint *Endpoint, err error) {
+	port := svcPort.Port
 	if port <= 0 {
-		return nil, fmt.Errorf("invalid port number: %s", svcPort.String())
+		return nil, fmt.Errorf("invalid port number: %d", port)
 	}
-	return newEndpointIP(svc.Spec.ClusterIP, port), nil
+	return newEndpointIP(svc.Spec.ClusterIP, int(port)), nil
 }
 
-func createEndpointsService(endpoints *api.Endpoints, svcPort intstr.IntOrString) (ready, notReady []*Endpoint) {
-	// TODO svcPort.IntValue() doesn't work if svc.targetPort is a pod's named port
+func createEndpointsService(endpoints *api.Endpoints, svcPort *api.ServicePort) (ready, notReady []*Endpoint) {
 	for _, subset := range endpoints.Subsets {
-		for _, port := range subset.Ports {
-			ssport := int(port.Port)
-			if ssport == svcPort.IntValue() && port.Protocol == api.ProtocolTCP {
+		for _, epPort := range subset.Ports {
+			if matchPort(svcPort, &epPort) {
+				port := int(epPort.Port)
 				for _, addr := range subset.Addresses {
-					ready = append(ready, newEndpointAddr(&addr, ssport))
+					ready = append(ready, newEndpointAddr(&addr, port))
 				}
 				for _, addr := range subset.NotReadyAddresses {
-					notReady = append(notReady, newEndpointAddr(&addr, ssport))
+					notReady = append(notReady, newEndpointAddr(&addr, port))
 				}
 			}
 		}
 	}
 	return ready, notReady
 }
 
+func matchPort(svcPort *api.ServicePort, epPort *api.EndpointPort) bool {
+	if epPort.Protocol != api.ProtocolTCP {
+		return false
+	}
+	return svcPort.Name == "" || svcPort.Name == epPort.Name
+}
+
 var lookup = net.LookupIP
 
-func createEndpointsExternalName(svc *api.Service, svcPort intstr.IntOrString) (endpoints []*Endpoint, err error) {
-	port := svcPort.IntValue()
+func createEndpointsExternalName(svc *api.Service, svcPort *api.ServicePort) (endpoints []*Endpoint, err error) {
+	// TODO add support to undeclared ServicePort
+	port := int(svcPort.Port)
 	if port <= 0 {
-		return nil, fmt.Errorf("invalid port number: %s", svcPort.String())
+		return nil, fmt.Errorf("invalid port number: %d", port)
 	}
 	addr, err := lookup(svc.Spec.ExternalName)
 	if err != nil {

pkg/converters/utils/services_test.go

@@ -64,6 +64,64 @@ func TestCreateEndpointsExternalName(t *testing.T) {
 	}
 }
 
+func TestCreateEndpoints(t *testing.T) {
+	testCases := []struct {
+		endpoints   string
+		declarePort string
+		findPort    string
+		expected    []*Endpoint
+	}{
+		// 0
+		{
+			endpoints:   "172.17.0.11,172.17.0.12",
+			declarePort: "svcport:8080:http",
+			findPort:    "8080",
+			expected: []*Endpoint{
+				{IP: "172.17.0.11", Port: 8080},
+				{IP: "172.17.0.12", Port: 8080},
+			},
+		},
+		// 1
+		{
+			endpoints:   "172.17.0.11",
+			declarePort: "svcport:8080:http",
+			findPort:    "svcport",
+			expected: []*Endpoint{
+				{IP: "172.17.0.11", Port: 8080},
+			},
+		},
+		// 2
+		{
+			endpoints:   "172.17.0.12",
+			declarePort: "svcport:8000:http",
+			findPort:    "http",
+			expected: []*Endpoint{
+				{IP: "172.17.0.12", Port: 8000},
+			},
+		},
+	}
+	for _, test := range testCases {
+		c := setup(t)
+		svc, ep := helper_test.CreateService("default/echo", test.declarePort, test.endpoints)
+		cache := &helper_test.CacheMock{
+			SvcList: []*api.Service{svc},
+			EpList:  map[string]*api.Endpoints{"default/echo": ep},
+		}
+		port := FindServicePort(svc, test.findPort)
+		var endpoints []*Endpoint
+		if port != nil {
+			endpoints, _, _ = CreateEndpoints(cache, svc, port)
+		}
+		for _, ep := range endpoints {
+			ep.TargetRef = ""
+		}
+		if !reflect.DeepEqual(endpoints, test.expected) {
+			t.Errorf("endpoints differ: expected=%+v actual=%+v", test.expected, endpoints)
+		}
+		c.teardown()
+	}
+}
+
 type config struct {
 	t *testing.T
 }