Fix service port lookup

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
@@ -68,43 +63,45 @@ func CreateEndpoints(cache types.Cache, svc *api.Service, svcPort intstr.IntOrSt
 	if err != nil {
 		return nil, nil, err
 	}
-	ready, notReady = createEndpointsService(endpoints, svcPort)
-	return ready, notReady, nil
-}
-
-// CreateSvcEndpoint ...
-func CreateSvcEndpoint(svc *api.Service, svcPort intstr.IntOrString) (endpoint *Endpoint, err error) {
-	port := svcPort.IntValue()
-	if port <= 0 {
-		return nil, fmt.Errorf("invalid port number: %s", svcPort.String())
-	}
-	return newEndpointIP(svc.Spec.ClusterIP, 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
 	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
+	return ready, notReady, nil
+}
+
+func matchPort(svcPort *api.ServicePort, epPort *api.EndpointPort) bool {
+	if epPort.Protocol != api.ProtocolTCP {
+		return false
+	}
+	return svcPort.Name == "" || svcPort.Name == epPort.Name
+}
+
+// CreateSvcEndpoint ...
+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: %d", port)
+	}
+	return newEndpointIP(svc.Spec.ClusterIP, int(port)), nil
 }
 
 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
 }