Filter pods that have out of range IP

Filter pods have IPs outside of the corresponding nodes' IP ranges.

pkg/neg/syncers/transaction_test.go

@@ -1799,12 +1799,16 @@ func TestEnableDegradedMode(t *testing.T) {
 			_, s := newTestTransactionSyncer(fakeCloud, negtypes.VmIpPortEndpointType, false)
 			s.NegSyncerKey.NegName = tc.negName
 			s.needInit = false
-			addPodsToLister(s.podLister)
+			addPodsToLister(s.podLister, getDefaultEndpointSlices())
 			for i := 1; i <= 4; i++ {
 				s.nodeLister.Add(&corev1.Node{
 					ObjectMeta: metav1.ObjectMeta{
 						Name: fmt.Sprintf("instance%v", i),
 					},
+					Spec: corev1.NodeSpec{
+						PodCIDR:  fmt.Sprintf("10.100.%v.0/24", i),
+						PodCIDRs: []string{fmt.Sprintf("200%v:db8::/48", i), fmt.Sprintf("10.100.%v.0/24", i)},
+					},
 				})
 			}
 			for _, eps := range tc.testEndpointSlices {

pkg/neg/syncers/utils.go

@@ -486,6 +486,11 @@ func toZoneNetworkEndpointMapDegradedMode(eds []negtypes.EndpointsData, zoneGett
 			}
 
 			podIPs := ipsForPod[types.NamespacedName{Namespace: endpointAddress.TargetRef.Namespace, Name: endpointAddress.TargetRef.Name}]
+			// endpoint address should match to the IP of its pod
+			if err = podContainsEndpointAddress(podIPs.IP, pod); err != nil {
+				klog.V(2).Infof("Endpoint %q does not have an address %s that matches to the IP(s) of its pod %v: %w, skipping", endpointAddress.Addresses, podIPs.IP, pod.Status.PodIPs, err)
+				continue
+			}
 			networkEndpoint := negtypes.NetworkEndpoint{IP: podIPs.IP, Port: matchPort, Node: nodeName}
 			if enableDualStackNEG {
 				// Convert all addresses to a standard form as per rfc5952 to prevent
@@ -519,6 +524,7 @@ func toZoneNetworkEndpointMapDegradedMode(eds []negtypes.EndpointsData, zoneGett
 // it returns error if the pod:
 // 1. is in terminal state
 // 2. corresponds to a non-existent node
+// 3. have an IP that matches to a podIP, but is outside of the node's allocated IP range
 func validatePod(pod *apiv1.Pod, nodeLister cache.Indexer) error {
 	// Terminal Pod means a pod is in PodFailed or PodSucceeded phase
 	phase := pod.Status.Phase
@@ -529,10 +535,14 @@ func validatePod(pod *apiv1.Pod, nodeLister cache.Indexer) error {
 	if err != nil || !exists {
 		return negtypes.ErrEPNodeNotFound
 	}
-	_, isNode := obj.(*apiv1.Node)
+	node, isNode := obj.(*apiv1.Node)
 	if !isNode {
 		return negtypes.ErrEPNodeTypeAssertionFailed
 	}
+	if err = nodeContainsPodIP(node, pod); err != nil {
+		klog.V(2).Info("Pod %s/%s has an IP %v that is outside of the node's allocated IP range(s) %v, skipping", pod.ObjectMeta.Namespace, pod.ObjectMeta.Name, pod.Status.PodIP, node.Spec.PodCIDRs)
+		return err
+	}
 	return nil
 }
 
@@ -562,6 +572,41 @@ func ipsForPod(eds []negtypes.EndpointsData) map[types.NamespacedName]negtypes.N
 	return result
 }
 
+// podContainsEndpointAddress checks the pod's existing PodIP(s)
+// and return error if the given endpoint address does not any of them.
+func podContainsEndpointAddress(endpointAddr string, pod *apiv1.Pod) error {
+	// a pod can have at most two PodIPs, one for ipv4 and one for ipv6
+	for _, podIP := range pod.Status.PodIPs {
+		if endpointAddr == podIP.IP {
+			return nil
+		}
+	}
+	return negtypes.ErrEPIPNotFromPod
+}
+
+// nodeContainsPodIP checks the node's existing PodCIDR(s)
+// and return error if the given podIP is not within one of the ranges
+func nodeContainsPodIP(node *apiv1.Node, pod *apiv1.Pod) error {
+	ipnets := []*net.IPNet{}
+	// a node can have at most two PodCIDRs, one for ipv4 and one for ipv6
+	for _, podCIDR := range node.Spec.PodCIDRs {
+		podCIDR = strings.TrimSpace(podCIDR)
+		_, ipnet, err := net.ParseCIDR(podCIDR)
+		if err != nil {
+			// swallow errors for CIDRs that are invalid
+			continue
+		}
+		ipnets = append(ipnets, ipnet)
+	}
+	podIP := net.ParseIP(pod.Status.PodIP)
+	for _, net := range ipnets {
+		if net.Contains(podIP) {
+			return nil
+		}
+	}
+	return negtypes.ErrEPIPOutOfPodCIDR
+}
+
 // retrieveExistingZoneNetworkEndpointMap lists existing network endpoints in the neg and return the zone and endpoints map
 func retrieveExistingZoneNetworkEndpointMap(negName string, zoneGetter negtypes.ZoneGetter, cloud negtypes.NetworkEndpointGroupCloud, version meta.Version, mode negtypes.EndpointsCalculatorMode) (map[string]negtypes.NetworkEndpointSet, labels.EndpointPodLabelMap, error) {
 	// Include zones that have non-candidate nodes currently. It is possible that NEGs were created in those zones previously and the endpoints now became non-candidates.