Check for ready endpoints (#3983)

* Refactor check for ready endpoints

Author: jjngx

internal/k8s/controller.go

@@ -26,6 +26,7 @@ import (
 	"time"
 
 	"github.com/nginxinc/kubernetes-ingress/pkg/apis/dos/v1beta1"
+	"golang.org/x/exp/maps"
 
 	"github.com/nginxinc/kubernetes-ingress/internal/k8s/appprotect"
 	"github.com/nginxinc/kubernetes-ingress/internal/k8s/appprotectcommon"
@@ -3596,46 +3597,66 @@ func (lbc *LoadBalancerController) getEndpointsForServiceWithSubselector(targetP
 	return endps, nil
 }
 
+// selectEndpointSlicesForPort returns EndpointSlices that match given targetPort.
+func selectEndpointSlicesForPort(targetPort int32, esx []discovery_v1.EndpointSlice) []discovery_v1.EndpointSlice {
+	eps := make([]discovery_v1.EndpointSlice, 0, len(esx))
+	for _, es := range esx {
+		for _, p := range es.Ports {
+			if p.Port == nil {
+				continue
+			}
+			if *p.Port == targetPort {
+				eps = append(eps, es)
+			}
+		}
+	}
+	return eps
+}
+
+// filterReadyEndpoinsFrom returns ready Endpoints from given EndpointSlices.
+func filterReadyEndpointsFrom(esx []discovery_v1.EndpointSlice) []discovery_v1.Endpoint {
+	epx := make([]discovery_v1.Endpoint, 0, len(esx))
+	for _, es := range esx {
+		for _, e := range es.Endpoints {
+			if e.Conditions.Ready == nil {
+				continue
+			}
+			if *e.Conditions.Ready {
+				epx = append(epx, e)
+			}
+		}
+	}
+	return epx
+}
+
 func getEndpointsFromEndpointSlicesForSubselectedPods(targetPort int32, pods []*api_v1.Pod, svcEndpointSlices []discovery_v1.EndpointSlice) (podEndpoints []podEndpoint) {
-	endpointSet := make(map[podEndpoint]struct{})
-	for _, pod := range pods {
-		for _, endpointSlice := range svcEndpointSlices {
-			for _, port := range endpointSlice.Ports {
-				if *port.Port != targetPort {
-					continue
-				}
-				for _, endpoint := range endpointSlice.Endpoints {
-					if !*endpoint.Conditions.Ready {
-						continue
-					}
-					for _, address := range endpoint.Addresses {
-						if pod.Status.PodIP == address {
-							addr := ipv6SafeAddrPort(pod.Status.PodIP, targetPort)
-							ownerType, ownerName := getPodOwnerTypeAndName(pod)
-							podEndpoint := podEndpoint{
-								Address: addr,
-								PodName: getPodName(endpoint.TargetRef),
-								MeshPodOwner: configs.MeshPodOwner{
-									OwnerType: ownerType,
-									OwnerName: ownerName,
-								},
-							}
-							endpointSet[podEndpoint] = struct{}{}
-							podEndpoints = append(podEndpoints, podEndpoint)
+	// Match ready endpoints IP ddresses with Pod's IP. If they match create a new podEnpoint.
+	makePodEndpoints := func(pods []*api_v1.Pod, endpoints []discovery_v1.Endpoint) []podEndpoint {
+		endpointSet := make(map[podEndpoint]struct{})
+
+		for _, pod := range pods {
+			for _, endpoint := range endpoints {
+				for _, address := range endpoint.Addresses {
+					if pod.Status.PodIP == address {
+						addr := ipv6SafeAddrPort(pod.Status.PodIP, targetPort)
+						ownerType, ownerName := getPodOwnerTypeAndName(pod)
+						podEndpoint := podEndpoint{
+							Address: addr,
+							PodName: getPodName(endpoint.TargetRef),
+							MeshPodOwner: configs.MeshPodOwner{
+								OwnerType: ownerType,
+								OwnerName: ownerName,
+							},
 						}
+						endpointSet[podEndpoint] = struct{}{}
 					}
 				}
 			}
 		}
+		return maps.Keys(endpointSet)
 	}
-	if len(endpointSet) == 0 {
-		return nil
-	}
-	endpoints := make([]podEndpoint, 0, len(endpointSet))
-	for ep := range endpointSet {
-		endpoints = append(endpoints, ep)
-	}
-	return endpoints
+
+	return makePodEndpoints(pods, filterReadyEndpointsFrom(selectEndpointSlicesForPort(targetPort, svcEndpointSlices)))
 }
 
 func ipv6SafeAddrPort(addr string, port int32) string {
@@ -3754,38 +3775,31 @@ func (lbc *LoadBalancerController) getEndpointsForPortFromEndpointSlices(endpoin
 		return nil, fmt.Errorf("no port %v in service %s", backendPort, svc.Name)
 	}
 
-	endpointSet := make(map[podEndpoint]struct{})
-	for _, endpointSlice := range endpointSlices {
-		for _, endpointSlicePort := range endpointSlice.Ports {
-			if *endpointSlicePort.Port == targetPort {
-				for _, endpoint := range endpointSlice.Endpoints {
-					if !*endpoint.Conditions.Ready {
-						continue
-					}
-					for _, endpointAddress := range endpoint.Addresses {
-						address := ipv6SafeAddrPort(endpointAddress, *endpointSlicePort.Port)
-						podEndpoint := podEndpoint{
-							Address: address,
-						}
-						if endpoint.TargetRef != nil {
-							parentType, parentName := lbc.getPodOwnerTypeAndNameFromAddress(endpoint.TargetRef.Namespace, endpoint.TargetRef.Name)
-							podEndpoint.OwnerType = parentType
-							podEndpoint.OwnerName = parentName
-							podEndpoint.PodName = endpoint.TargetRef.Name
-						}
-						endpointSet[podEndpoint] = struct{}{}
-					}
+	makePodEndpoints := func(port int32, epx []discovery_v1.Endpoint) []podEndpoint {
+		endpointSet := make(map[podEndpoint]struct{})
+
+		for _, ep := range epx {
+			for _, addr := range ep.Addresses {
+				address := ipv6SafeAddrPort(addr, port)
+				podEndpoint := podEndpoint{
+					Address: address,
+				}
+				if ep.TargetRef != nil {
+					parentType, parentName := lbc.getPodOwnerTypeAndNameFromAddress(ep.TargetRef.Namespace, ep.TargetRef.Name)
+					podEndpoint.OwnerType = parentType
+					podEndpoint.OwnerName = parentName
+					podEndpoint.PodName = ep.TargetRef.Name
 				}
+				endpointSet[podEndpoint] = struct{}{}
 			}
 		}
+		return maps.Keys(endpointSet)
 	}
-	if len(endpointSet) == 0 {
+
+	endpoints := makePodEndpoints(targetPort, filterReadyEndpointsFrom(selectEndpointSlicesForPort(targetPort, endpointSlices)))
+	if len(endpoints) == 0 {
 		return nil, fmt.Errorf("no endpointslices for target port %v in service %s", targetPort, svc.Name)
 	}
-	endpoints := make([]podEndpoint, 0, len(endpointSet))
-	for ep := range endpointSet {
-		endpoints = append(endpoints, ep)
-	}
 	return endpoints, nil
 }
 

internal/k8s/controller_test.go

@@ -1161,6 +1161,115 @@ func TestGetEndpointSlicesBySubselectedPods_FindOnePodInOneEndpointSlice(t *test
 	}
 }
 
+func TestGetEndpointSlicesBySubselectedPods_GetsEndpointsOnNilValues(t *testing.T) {
+	t.Parallel()
+	endpointPort := int32(8080)
+	endpointReady := true
+	boolPointer := func(b bool) *bool { return &b }
+
+	tests := []struct {
+		desc              string
+		targetPort        int32
+		svcEndpointSlices []discovery_v1.EndpointSlice
+		pods              []*api_v1.Pod
+		want              []podEndpoint
+	}{
+		{
+			desc:       "no endpoints selected on nil endpoint port",
+			targetPort: 8080,
+			want:       []podEndpoint{},
+			pods: []*api_v1.Pod{
+				{
+					ObjectMeta: meta_v1.ObjectMeta{
+						OwnerReferences: []meta_v1.OwnerReference{
+							{
+								Kind:       "Deployment",
+								Name:       "deploy-1",
+								Controller: boolPointer(true),
+							},
+						},
+					},
+					Status: api_v1.PodStatus{
+						PodIP: "1.2.3.4",
+					},
+				},
+			},
+			svcEndpointSlices: []discovery_v1.EndpointSlice{
+				{
+					Ports: []discovery_v1.EndpointPort{
+						{
+							Port: nil,
+						},
+					},
+					Endpoints: []discovery_v1.Endpoint{
+						{
+							Addresses: []string{
+								"1.2.3.4",
+							},
+							Conditions: discovery_v1.EndpointConditions{
+								Ready: &endpointReady,
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			desc:       "no endpoints selected on nil endpoint condition",
+			targetPort: 8080,
+			want:       []podEndpoint{},
+			pods: []*api_v1.Pod{
+				{
+					ObjectMeta: meta_v1.ObjectMeta{
+						OwnerReferences: []meta_v1.OwnerReference{
+							{
+								Kind:       "Deployment",
+								Name:       "deploy-1",
+								Controller: boolPointer(true),
+							},
+						},
+					},
+					Status: api_v1.PodStatus{
+						PodIP: "1.2.3.4",
+					},
+				},
+			},
+			svcEndpointSlices: []discovery_v1.EndpointSlice{
+				{
+					Ports: []discovery_v1.EndpointPort{
+						{
+							Port: &endpointPort,
+						},
+					},
+					Endpoints: []discovery_v1.Endpoint{
+						{
+							Addresses: []string{
+								"1.2.3.4",
+							},
+							Conditions: discovery_v1.EndpointConditions{
+								Ready: nil,
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.desc, func(t *testing.T) {
+			got := getEndpointsFromEndpointSlicesForSubselectedPods(test.targetPort, test.pods, test.svcEndpointSlices)
+			if !cmp.Equal(got, test.want) {
+				t.Error(cmp.Diff(got, test.want))
+			}
+		})
+	}
+}
+
+func TestFilterReadyEndpoints(t *testing.T) {
+	t.Parallel()
+}
+
 func TestGetEndpointSlicesBySubselectedPods_FindOnePodInTwoEndpointSlicesWithDuplicateEndpoints(t *testing.T) {
 	t.Parallel()
 	endpointPort := int32(8080)