Equip getEndpointsFromSlices to select Serving Endpoints

This wires up `getEndpointsFromSlices` to have different selection
modes for querying Endpoint objects. This is necessary so that
`persistent-drainable` can eventually select Endpoints that are
`Serving` instead of only `Ready`.

This added a unit test to demonstrate the selection behavior, and
the wiring up in the controller remains to be done.

The Endpoint type was also updated to have an `IsDraining` field
when the Endpoint is not Ready.

Author: gulachek

internal/ingress/controller/controller.go

@@ -536,7 +536,7 @@ func (n *NGINXController) getStreamServices(configmapName string, proto apiv1.Pr
 				sp := svc.Spec.Ports[i]
 				if sp.Name == svcPort {
 					if sp.Protocol == proto {
-						endps = getEndpointsFromSlices(svc, &sp, proto, zone, n.store.GetServiceEndpointsSlices)
+						endps = getEndpointsFromSlices(svc, &sp, proto, zone, ReadyEndpoints, n.store.GetServiceEndpointsSlices)
 						break
 					}
 				}
@@ -548,7 +548,7 @@ func (n *NGINXController) getStreamServices(configmapName string, proto apiv1.Pr
 				//nolint:gosec // Ignore G109 error
 				if sp.Port == int32(targetPort) {
 					if sp.Protocol == proto {
-						endps = getEndpointsFromSlices(svc, &sp, proto, zone, n.store.GetServiceEndpointsSlices)
+						endps = getEndpointsFromSlices(svc, &sp, proto, zone, ReadyEndpoints, n.store.GetServiceEndpointsSlices)
 						break
 					}
 				}
@@ -605,7 +605,7 @@ func (n *NGINXController) getDefaultUpstream() *ingress.Backend {
 	} else {
 		zone = emptyZone
 	}
-	endps := getEndpointsFromSlices(svc, &svc.Spec.Ports[0], apiv1.ProtocolTCP, zone, n.store.GetServiceEndpointsSlices)
+	endps := getEndpointsFromSlices(svc, &svc.Spec.Ports[0], apiv1.ProtocolTCP, zone, ReadyEndpoints, n.store.GetServiceEndpointsSlices)
 	if len(endps) == 0 {
 		klog.Warningf("Service %q does not have any active Endpoint", svcKey)
 		endps = []ingress.Endpoint{n.DefaultEndpoint()}
@@ -940,7 +940,7 @@ func (n *NGINXController) getBackendServers(ingresses []*ingress.Ingress) ([]*in
 				} else {
 					zone = emptyZone
 				}
-				endps := getEndpointsFromSlices(location.DefaultBackend, &sp, apiv1.ProtocolTCP, zone, n.store.GetServiceEndpointsSlices)
+				endps := getEndpointsFromSlices(location.DefaultBackend, &sp, apiv1.ProtocolTCP, zone, ReadyEndpoints, n.store.GetServiceEndpointsSlices)
 				// custom backend is valid only if contains at least one endpoint
 				if len(endps) > 0 {
 					name := fmt.Sprintf("custom-default-backend-%v-%v", location.DefaultBackend.GetNamespace(), location.DefaultBackend.GetName())
@@ -1050,7 +1050,7 @@ func (n *NGINXController) createUpstreams(data []*ingress.Ingress, du *ingress.B
 
 			if len(upstreams[defBackend].Endpoints) == 0 {
 				_, port := upstreamServiceNameAndPort(ing.Spec.DefaultBackend.Service)
-				endps, err := n.serviceEndpoints(svcKey, port.String())
+				endps, err := n.serviceEndpoints(svcKey, port.String(), ReadyEndpoints)
 				upstreams[defBackend].Endpoints = append(upstreams[defBackend].Endpoints, endps...)
 				if err != nil {
 					klog.Warningf("Error creating upstream %q: %v", defBackend, err)
@@ -1115,7 +1115,8 @@ func (n *NGINXController) createUpstreams(data []*ingress.Ingress, du *ingress.B
 
 				if len(upstreams[name].Endpoints) == 0 {
 					_, port := upstreamServiceNameAndPort(path.Backend.Service)
-					endp, err := n.serviceEndpoints(svcKey, port.String())
+					endp, err := n.serviceEndpoints(svcKey, port.String(), ReadyEndpoints)
+
 					if err != nil {
 						klog.Warningf("Error obtaining Endpoints for Service %q: %v", svcKey, err)
 						n.metricCollector.IncOrphanIngress(ing.Namespace, ing.Name, orphanMetricLabelNoService)
@@ -1184,7 +1185,7 @@ func (n *NGINXController) getServiceClusterEndpoint(svcKey string, backend *netw
 }
 
 // serviceEndpoints returns the upstream servers (Endpoints) associated with a Service.
-func (n *NGINXController) serviceEndpoints(svcKey, backendPort string) ([]ingress.Endpoint, error) {
+func (n *NGINXController) serviceEndpoints(svcKey, backendPort string, epSelectionMode EndpointSelectionMode) ([]ingress.Endpoint, error) {
 	var upstreams []ingress.Endpoint
 
 	svc, err := n.store.GetService(svcKey)
@@ -1205,7 +1206,7 @@ func (n *NGINXController) serviceEndpoints(svcKey, backendPort string) ([]ingres
 			return upstreams, nil
 		}
 		servicePort := externalNamePorts(backendPort, svc)
-		endps := getEndpointsFromSlices(svc, servicePort, apiv1.ProtocolTCP, zone, n.store.GetServiceEndpointsSlices)
+		endps := getEndpointsFromSlices(svc, servicePort, apiv1.ProtocolTCP, zone, epSelectionMode, n.store.GetServiceEndpointsSlices)
 		if len(endps) == 0 {
 			klog.Warningf("Service %q does not have any active Endpoint.", svcKey)
 			return upstreams, nil
@@ -1221,7 +1222,7 @@ func (n *NGINXController) serviceEndpoints(svcKey, backendPort string) ([]ingres
 		if strconv.Itoa(int(servicePort.Port)) == backendPort ||
 			servicePort.TargetPort.String() == backendPort ||
 			servicePort.Name == backendPort {
-			endps := getEndpointsFromSlices(svc, &servicePort, apiv1.ProtocolTCP, zone, n.store.GetServiceEndpointsSlices)
+			endps := getEndpointsFromSlices(svc, &servicePort, apiv1.ProtocolTCP, zone, epSelectionMode, n.store.GetServiceEndpointsSlices)
 			if len(endps) == 0 {
 				klog.Warningf("Service %q does not have any active Endpoint.", svcKey)
 			}

internal/ingress/controller/endpointslices.go

@@ -34,9 +34,16 @@ import (
 	"k8s.io/ingress-nginx/pkg/apis/ingress"
 )
 
+type EndpointSelectionMode int
+
+const (
+	ReadyEndpoints EndpointSelectionMode = iota
+	ServingEndpoints
+)
+
 // getEndpointsFromSlices returns a list of Endpoint structs for a given service/target port combination.
 func getEndpointsFromSlices(s *corev1.Service, port *corev1.ServicePort, proto corev1.Protocol, zoneForHints string,
-	getServiceEndpointsSlices func(string) ([]*discoveryv1.EndpointSlice, error),
+	epSelectionMode EndpointSelectionMode, getServiceEndpointsSlices func(string) ([]*discoveryv1.EndpointSlice, error),
 ) []ingress.Endpoint {
 	upsServers := []ingress.Endpoint{}
 
@@ -153,9 +160,19 @@ func getEndpointsFromSlices(s *corev1.Service, port *corev1.ServicePort, proto c
 		}
 
 		for _, ep := range eps.Endpoints {
-			if (ep.Conditions.Ready != nil) && !(*ep.Conditions.Ready) {
-				continue
+			epIsReady := (ep.Conditions.Ready == nil) || *ep.Conditions.Ready
+			if epSelectionMode == ReadyEndpoints {
+				if !epIsReady {
+					continue
+				}
+			} else {
+				// assume epSelectionMode == ServingEndpoints.
+				epIsServing := (ep.Conditions.Serving == nil) || *ep.Conditions.Serving
+				if !epIsServing {
+					continue
+				}
 			}
+
 			epHasZone := false
 			if useTopologyHints {
 				for _, epzone := range ep.Hints.ForZones {
@@ -176,10 +193,12 @@ func getEndpointsFromSlices(s *corev1.Service, port *corev1.ServicePort, proto c
 					if _, exists := processedUpstreamServers[hostPort]; exists {
 						continue
 					}
+
 					ups := ingress.Endpoint{
-						Address: epAddress,
-						Port:    fmt.Sprintf("%v", epPort),
-						Target:  ep.TargetRef,
+						Address:    epAddress,
+						Port:       fmt.Sprintf("%v", epPort),
+						Target:     ep.TargetRef,
+						IsDraining: !epIsReady,
 					}
 					upsServers = append(upsServers, ups)
 					processedUpstreamServers[hostPort] = struct{}{}