diff --git a/internal/ingress/controller/controller.go b/internal/ingress/controller/controller.go index 895fc85706..ae8424dc13 100644 --- a/internal/ingress/controller/controller.go +++ b/internal/ingress/controller/controller.go @@ -19,8 +19,6 @@ package controller import ( "fmt" "math/rand" - "net" - "reflect" "sort" "strconv" "strings" @@ -291,7 +289,7 @@ func (n *NGINXController) getStreamServices(configmapName string, proto apiv1.Pr for _, sp := range svc.Spec.Ports { if sp.Name == svcPort { if sp.Protocol == proto { - endps = n.getEndpoints(svc, &sp, proto, &healthcheck.Config{}) + endps = getEndpoints(svc, &sp, proto, &healthcheck.Config{}, n.store.GetServiceEndpoints) break } } @@ -302,7 +300,7 @@ func (n *NGINXController) getStreamServices(configmapName string, proto apiv1.Pr for _, sp := range svc.Spec.Ports { if sp.Port == int32(targetPort) { if sp.Protocol == proto { - endps = n.getEndpoints(svc, &sp, proto, &healthcheck.Config{}) + endps = getEndpoints(svc, &sp, proto, &healthcheck.Config{}, n.store.GetServiceEndpoints) break } } @@ -347,7 +345,7 @@ func (n *NGINXController) getDefaultUpstream() *ingress.Backend { return upstream } - endps := n.getEndpoints(svc, &svc.Spec.Ports[0], apiv1.ProtocolTCP, &healthcheck.Config{}) + endps := getEndpoints(svc, &svc.Spec.Ports[0], apiv1.ProtocolTCP, &healthcheck.Config{}, n.store.GetServiceEndpoints) if len(endps) == 0 { glog.Warningf("service %v does not have any active endpoints", svcKey) endps = []ingress.Endpoint{n.DefaultEndpoint()} @@ -528,7 +526,7 @@ func (n *NGINXController) getBackendServers(ingresses []*extensions.Ingress) ([] // check if the location contains endpoints and a custom default backend if location.DefaultBackend != nil { sp := location.DefaultBackend.Spec.Ports[0] - endps := n.getEndpoints(location.DefaultBackend, &sp, apiv1.ProtocolTCP, &healthcheck.Config{}) + endps := getEndpoints(location.DefaultBackend, &sp, apiv1.ProtocolTCP, &healthcheck.Config{}, n.store.GetServiceEndpoints) if len(endps) > 0 { glog.V(3).Infof("using custom default backend in server %v location %v (service %v/%v)", server.Hostname, location.Path, location.DefaultBackend.Namespace, location.DefaultBackend.Name) @@ -770,7 +768,7 @@ func (n *NGINXController) serviceEndpoints(svcKey, backendPort string, servicePort.TargetPort.String() == backendPort || servicePort.Name == backendPort { - endps := n.getEndpoints(svc, &servicePort, apiv1.ProtocolTCP, hz) + endps := getEndpoints(svc, &servicePort, apiv1.ProtocolTCP, hz, n.store.GetServiceEndpoints) if len(endps) == 0 { glog.Warningf("service %v does not have any active endpoints", svcKey) } @@ -804,7 +802,7 @@ func (n *NGINXController) serviceEndpoints(svcKey, backendPort string, Port: int32(externalPort), TargetPort: intstr.FromString(backendPort), } - endps := n.getEndpoints(svc, &servicePort, apiv1.ProtocolTCP, hz) + endps := getEndpoints(svc, &servicePort, apiv1.ProtocolTCP, hz, n.store.GetServiceEndpoints) if len(endps) == 0 { glog.Warningf("service %v does not have any active endpoints", svcKey) return upstreams, nil @@ -1057,97 +1055,6 @@ func (n *NGINXController) createServers(data []*extensions.Ingress, return servers } -// getEndpoints returns a list of : for a given service/target port combination. -func (n *NGINXController) getEndpoints( - s *apiv1.Service, - servicePort *apiv1.ServicePort, - proto apiv1.Protocol, - hz *healthcheck.Config) []ingress.Endpoint { - - upsServers := []ingress.Endpoint{} - - // avoid duplicated upstream servers when the service - // contains multiple port definitions sharing the same - // targetport. - adus := make(map[string]bool) - - // ExternalName services - if s.Spec.Type == apiv1.ServiceTypeExternalName { - glog.V(3).Infof("Ingress using a service %v of type=ExternalName : %v", s.Name) - - targetPort := servicePort.TargetPort.IntValue() - // check for invalid port value - if targetPort <= 0 { - glog.Errorf("ExternalName service with an invalid port: %v", targetPort) - return upsServers - } - - if net.ParseIP(s.Spec.ExternalName) == nil { - _, err := net.LookupHost(s.Spec.ExternalName) - if err != nil { - glog.Errorf("unexpected error resolving host %v: %v", s.Spec.ExternalName, err) - return upsServers - } - } - - return append(upsServers, ingress.Endpoint{ - Address: s.Spec.ExternalName, - Port: fmt.Sprintf("%v", targetPort), - MaxFails: hz.MaxFails, - FailTimeout: hz.FailTimeout, - }) - } - - glog.V(3).Infof("getting endpoints for service %v/%v and port %v", s.Namespace, s.Name, servicePort.String()) - ep, err := n.store.GetServiceEndpoints(s) - if err != nil { - glog.Warningf("unexpected error obtaining service endpoints: %v", err) - return upsServers - } - - for _, ss := range ep.Subsets { - for _, epPort := range ss.Ports { - - if !reflect.DeepEqual(epPort.Protocol, proto) { - continue - } - - var targetPort int32 - - if servicePort.Name == "" { - // ServicePort.Name is optional if there is only one port - targetPort = epPort.Port - } else if servicePort.Name == epPort.Name { - targetPort = epPort.Port - } - - // check for invalid port value - if targetPort <= 0 { - continue - } - - for _, epAddress := range ss.Addresses { - ep := fmt.Sprintf("%v:%v", epAddress.IP, targetPort) - if _, exists := adus[ep]; exists { - continue - } - ups := ingress.Endpoint{ - Address: epAddress.IP, - Port: fmt.Sprintf("%v", targetPort), - MaxFails: hz.MaxFails, - FailTimeout: hz.FailTimeout, - Target: epAddress.TargetRef, - } - upsServers = append(upsServers, ups) - adus[ep] = true - } - } - } - - glog.V(3).Infof("endpoints found: %v", upsServers) - return upsServers -} - func (n *NGINXController) isForceReload() bool { return atomic.LoadInt32(&n.forceReload) != 0 } diff --git a/internal/ingress/controller/endpoints.go b/internal/ingress/controller/endpoints.go new file mode 100644 index 0000000000..62f7474f43 --- /dev/null +++ b/internal/ingress/controller/endpoints.go @@ -0,0 +1,128 @@ +/* +Copyright 2018 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 controller + +import ( + "fmt" + "net" + "reflect" + + "github.com/golang/glog" + + corev1 "k8s.io/api/core/v1" + + "k8s.io/ingress-nginx/internal/ingress" + "k8s.io/ingress-nginx/internal/ingress/annotations/healthcheck" +) + +// getEndpoints returns a list of : for a given service/target port combination. +func getEndpoints( + s *corev1.Service, + port *corev1.ServicePort, + proto corev1.Protocol, + hz *healthcheck.Config, + getServiceEndpoints func(*corev1.Service) (*corev1.Endpoints, error), +) []ingress.Endpoint { + + upsServers := []ingress.Endpoint{} + + if s == nil || port == nil { + return upsServers + } + + // avoid duplicated upstream servers when the service + // contains multiple port definitions sharing the same + // targetport. + adus := make(map[string]bool) + + // ExternalName services + if s.Spec.Type == corev1.ServiceTypeExternalName { + glog.V(3).Infof("Ingress using a service %v of type=ExternalName : %v", s.Name) + + targetPort := port.TargetPort.IntValue() + // check for invalid port value + if targetPort <= 0 { + glog.Errorf("ExternalName service with an invalid port: %v", targetPort) + return upsServers + } + + if net.ParseIP(s.Spec.ExternalName) == nil { + _, err := net.LookupHost(s.Spec.ExternalName) + if err != nil { + glog.Errorf("unexpected error resolving host %v: %v", s.Spec.ExternalName, err) + return upsServers + } + } + + return append(upsServers, ingress.Endpoint{ + Address: s.Spec.ExternalName, + Port: fmt.Sprintf("%v", targetPort), + MaxFails: hz.MaxFails, + FailTimeout: hz.FailTimeout, + }) + } + + glog.V(3).Infof("getting endpoints for service %v/%v and port %v", s.Namespace, s.Name, port.String()) + ep, err := getServiceEndpoints(s) + if err != nil { + glog.Warningf("unexpected error obtaining service endpoints: %v", err) + return upsServers + } + + for _, ss := range ep.Subsets { + for _, epPort := range ss.Ports { + + if !reflect.DeepEqual(epPort.Protocol, proto) { + continue + } + + var targetPort int32 + + if port.Name == "" { + // port.Name is optional if there is only one port + targetPort = epPort.Port + } else if port.Name == epPort.Name { + targetPort = epPort.Port + } + + glog.Infof("TP: %v", targetPort) + // check for invalid port value + if targetPort <= 0 { + continue + } + + for _, epAddress := range ss.Addresses { + ep := fmt.Sprintf("%v:%v", epAddress.IP, targetPort) + if _, exists := adus[ep]; exists { + continue + } + ups := ingress.Endpoint{ + Address: epAddress.IP, + Port: fmt.Sprintf("%v", targetPort), + MaxFails: hz.MaxFails, + FailTimeout: hz.FailTimeout, + Target: epAddress.TargetRef, + } + upsServers = append(upsServers, ups) + adus[ep] = true + } + } + } + + glog.V(3).Infof("endpoints found: %v", upsServers) + return upsServers +} diff --git a/internal/ingress/controller/endpoints_test.go b/internal/ingress/controller/endpoints_test.go new file mode 100644 index 0000000000..08e4d6c9fe --- /dev/null +++ b/internal/ingress/controller/endpoints_test.go @@ -0,0 +1,438 @@ +/* +Copyright 2018 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 controller + +import ( + "fmt" + "testing" + + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/util/intstr" + "k8s.io/ingress-nginx/internal/ingress" + "k8s.io/ingress-nginx/internal/ingress/annotations/healthcheck" +) + +func TestGetEndpoints(t *testing.T) { + tests := []struct { + name string + svc *corev1.Service + port *corev1.ServicePort + proto corev1.Protocol + hz *healthcheck.Config + fn func(*corev1.Service) (*corev1.Endpoints, error) + result []ingress.Endpoint + }{ + { + "no service should return 0 endpoints", + nil, + nil, + corev1.ProtocolTCP, + nil, + func(*corev1.Service) (*corev1.Endpoints, error) { + return nil, nil + }, + []ingress.Endpoint{}, + }, + { + "no service port should return 0 endpoints", + &corev1.Service{}, + nil, + corev1.ProtocolTCP, + nil, + func(*corev1.Service) (*corev1.Endpoints, error) { + return nil, nil + }, + []ingress.Endpoint{}, + }, + { + "a service without endpoints should return 0 endpoints", + &corev1.Service{}, + &corev1.ServicePort{Name: "default"}, + corev1.ProtocolTCP, + nil, + func(*corev1.Service) (*corev1.Endpoints, error) { + return &corev1.Endpoints{}, nil + }, + []ingress.Endpoint{}, + }, + { + "a service type ServiceTypeExternalName service with an invalid port should return 0 endpoints", + &corev1.Service{ + Spec: corev1.ServiceSpec{ + Type: corev1.ServiceTypeExternalName, + }, + }, + &corev1.ServicePort{Name: "default"}, + corev1.ProtocolTCP, + nil, + func(*corev1.Service) (*corev1.Endpoints, error) { + return &corev1.Endpoints{}, nil + }, + []ingress.Endpoint{}, + }, + { + "a service type ServiceTypeExternalName with a valid port should return one endpoint", + &corev1.Service{ + Spec: corev1.ServiceSpec{ + Type: corev1.ServiceTypeExternalName, + ExternalName: "10.0.0.1.xip.io", + Ports: []corev1.ServicePort{ + { + Name: "default", + TargetPort: intstr.FromInt(80), + }, + }, + }, + }, + &corev1.ServicePort{ + Name: "default", + TargetPort: intstr.FromInt(80), + }, + corev1.ProtocolTCP, + &healthcheck.Config{ + MaxFails: 0, + FailTimeout: 0, + }, + func(*corev1.Service) (*corev1.Endpoints, error) { + return &corev1.Endpoints{}, nil + }, + []ingress.Endpoint{ + { + Address: "10.0.0.1.xip.io", + Port: "80", + MaxFails: 0, + FailTimeout: 0, + }, + }, + }, + { + "a service type ServiceTypeExternalName with an invalid ExternalName value should return one endpoint", + &corev1.Service{ + Spec: corev1.ServiceSpec{ + Type: corev1.ServiceTypeExternalName, + ExternalName: "foo.bar", + Ports: []corev1.ServicePort{ + { + Name: "default", + TargetPort: intstr.FromInt(80), + }, + }, + }, + }, + &corev1.ServicePort{ + Name: "default", + TargetPort: intstr.FromInt(80), + }, + corev1.ProtocolTCP, + &healthcheck.Config{ + MaxFails: 0, + FailTimeout: 0, + }, + func(*corev1.Service) (*corev1.Endpoints, error) { + return &corev1.Endpoints{}, nil + }, + []ingress.Endpoint{}, + }, + { + "should return no endpoints when there is an error searching for endpoints", + &corev1.Service{ + Spec: corev1.ServiceSpec{ + Type: corev1.ServiceTypeClusterIP, + ClusterIP: "1.1.1.1", + Ports: []corev1.ServicePort{ + { + Name: "default", + TargetPort: intstr.FromInt(80), + }, + }, + }, + }, + &corev1.ServicePort{ + Name: "default", + TargetPort: intstr.FromInt(80), + }, + corev1.ProtocolTCP, + &healthcheck.Config{ + MaxFails: 0, + FailTimeout: 0, + }, + func(*corev1.Service) (*corev1.Endpoints, error) { + return nil, fmt.Errorf("unexpected error") + }, + []ingress.Endpoint{}, + }, + { + "should return no endpoints when the protocol does not match", + &corev1.Service{ + Spec: corev1.ServiceSpec{ + Type: corev1.ServiceTypeClusterIP, + ClusterIP: "1.1.1.1", + Ports: []corev1.ServicePort{ + { + Name: "default", + TargetPort: intstr.FromInt(80), + }, + }, + }, + }, + &corev1.ServicePort{ + Name: "default", + TargetPort: intstr.FromInt(80), + }, + corev1.ProtocolTCP, + &healthcheck.Config{ + MaxFails: 0, + FailTimeout: 0, + }, + func(*corev1.Service) (*corev1.Endpoints, error) { + nodeName := "dummy" + return &corev1.Endpoints{ + Subsets: []corev1.EndpointSubset{ + { + Addresses: []corev1.EndpointAddress{ + { + IP: "1.1.1.1", + NodeName: &nodeName, + }, + }, + Ports: []corev1.EndpointPort{ + { + Protocol: corev1.ProtocolUDP, + }, + }, + }, + }, + }, nil + }, + []ingress.Endpoint{}, + }, + { + "should return no endpoints when there is no ready Addresses", + &corev1.Service{ + Spec: corev1.ServiceSpec{ + Type: corev1.ServiceTypeClusterIP, + ClusterIP: "1.1.1.1", + Ports: []corev1.ServicePort{ + { + Name: "default", + TargetPort: intstr.FromInt(80), + }, + }, + }, + }, + &corev1.ServicePort{ + Name: "default", + TargetPort: intstr.FromInt(80), + }, + corev1.ProtocolTCP, + &healthcheck.Config{ + MaxFails: 0, + FailTimeout: 0, + }, + func(*corev1.Service) (*corev1.Endpoints, error) { + nodeName := "dummy" + return &corev1.Endpoints{ + Subsets: []corev1.EndpointSubset{ + { + NotReadyAddresses: []corev1.EndpointAddress{ + { + IP: "1.1.1.1", + NodeName: &nodeName, + }, + }, + Ports: []corev1.EndpointPort{ + { + Protocol: corev1.ProtocolUDP, + }, + }, + }, + }, + }, nil + }, + []ingress.Endpoint{}, + }, + { + "should return no endpoints when the name of the port name do not match any port in the endpoint Subsets", + &corev1.Service{ + Spec: corev1.ServiceSpec{ + Type: corev1.ServiceTypeClusterIP, + ClusterIP: "1.1.1.1", + Ports: []corev1.ServicePort{ + { + Name: "default", + TargetPort: intstr.FromInt(80), + }, + }, + }, + }, + &corev1.ServicePort{ + Name: "default", + TargetPort: intstr.FromInt(80), + }, + corev1.ProtocolTCP, + &healthcheck.Config{ + MaxFails: 0, + FailTimeout: 0, + }, + func(*corev1.Service) (*corev1.Endpoints, error) { + nodeName := "dummy" + return &corev1.Endpoints{ + Subsets: []corev1.EndpointSubset{ + { + Addresses: []corev1.EndpointAddress{ + { + IP: "1.1.1.1", + NodeName: &nodeName, + }, + }, + Ports: []corev1.EndpointPort{ + { + Protocol: corev1.ProtocolTCP, + Port: int32(80), + Name: "another-name", + }, + }, + }, + }, + }, nil + }, + []ingress.Endpoint{}, + }, + { + "should return one endpoint when the name of the port name match a port in the endpoint Subsets", + &corev1.Service{ + Spec: corev1.ServiceSpec{ + Type: corev1.ServiceTypeClusterIP, + ClusterIP: "1.1.1.1", + Ports: []corev1.ServicePort{ + { + Name: "default", + TargetPort: intstr.FromInt(80), + }, + }, + }, + }, + &corev1.ServicePort{ + Name: "default", + TargetPort: intstr.FromInt(80), + }, + corev1.ProtocolTCP, + &healthcheck.Config{ + MaxFails: 0, + FailTimeout: 0, + }, + func(*corev1.Service) (*corev1.Endpoints, error) { + nodeName := "dummy" + return &corev1.Endpoints{ + Subsets: []corev1.EndpointSubset{ + { + Addresses: []corev1.EndpointAddress{ + { + IP: "1.1.1.1", + NodeName: &nodeName, + }, + }, + Ports: []corev1.EndpointPort{ + { + Protocol: corev1.ProtocolTCP, + Port: int32(80), + Name: "default", + }, + }, + }, + }, + }, nil + }, + []ingress.Endpoint{ + { + Address: "1.1.1.1", + Port: "80", + MaxFails: 0, + FailTimeout: 0, + }, + }, + }, + { + "should return one endpoint when the name of the port name match more than one port in the endpoint Subsets", + &corev1.Service{ + Spec: corev1.ServiceSpec{ + Type: corev1.ServiceTypeClusterIP, + ClusterIP: "1.1.1.1", + Ports: []corev1.ServicePort{ + { + Name: "default", + TargetPort: intstr.FromString("port-1"), + }, + }, + }, + }, + &corev1.ServicePort{ + Name: "port-1", + TargetPort: intstr.FromString("port-1"), + }, + corev1.ProtocolTCP, + &healthcheck.Config{ + MaxFails: 0, + FailTimeout: 0, + }, + func(*corev1.Service) (*corev1.Endpoints, error) { + nodeName := "dummy" + return &corev1.Endpoints{ + Subsets: []corev1.EndpointSubset{ + { + Addresses: []corev1.EndpointAddress{ + { + IP: "1.1.1.1", + NodeName: &nodeName, + }, + }, + Ports: []corev1.EndpointPort{ + { + Name: "port-1", + Protocol: corev1.ProtocolTCP, + Port: 80, + }, + { + Name: "port-1", + Protocol: corev1.ProtocolTCP, + Port: 80, + }, + }, + }, + }, + }, nil + }, + []ingress.Endpoint{ + { + Address: "1.1.1.1", + Port: "80", + MaxFails: 0, + FailTimeout: 0, + }, + }, + }, + } + + for _, testCase := range tests { + t.Run(testCase.name, func(t *testing.T) { + result := getEndpoints(testCase.svc, testCase.port, testCase.proto, testCase.hz, testCase.fn) + if len(testCase.result) != len(result) { + t.Errorf("expected %v Endpoints but got %v", testCase.result, len(result)) + } + }) + } +}