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

@@ -471,10 +471,6 @@ func toZoneNetworkEndpointMapDegradedMode(eds []negtypes.EndpointsData, zoneGett
 				klog.Errorf("Endpoint %q in Endpoints %s/%s receives error when getting pod, err: %v, skipping", endpointAddress.Addresses, ed.Meta.Namespace, ed.Meta.Name, getPodErr)
 				continue
 			}
-			if err := validatePod(pod, nodeLister); err != nil {
-				klog.Errorf("Endpoint %q in Endpoints %s/%s correponds to an invalid pod: %v, skipping", endpointAddress.Addresses, ed.Meta.Namespace, ed.Meta.Name, err)
-				continue
-			}
 			nodeName := pod.Spec.NodeName
 			zone, err := zoneGetter.GetZoneForNode(nodeName)
 			if err != nil {
@@ -486,12 +482,25 @@ func toZoneNetworkEndpointMapDegradedMode(eds []negtypes.EndpointsData, zoneGett
 			}
 
 			podIPs := ipsForPod[types.NamespacedName{Namespace: endpointAddress.TargetRef.Namespace, Name: endpointAddress.TargetRef.Name}]
+			if parseIPAddress(podIPs.IP) == "" {
+				klog.Errorf("For endpoint %q in pod %q, it has an invalid IP address, err: %v, skipping", endpointAddress.Addresses, pod.ObjectMeta.Name, negtypes.ErrEPIPNotFromPod)
+				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
 				// accidental diffs resulting from different formats.
 				networkEndpoint.IPv6 = parseIPAddress(podIPs.IPv6)
 			}
+			// endpoint address should match to the IP of its pod
+			if err = podContainsEndpointAddress(networkEndpoint, pod); err != nil {
+				klog.Errorf("Endpoint %q in Endpoints %s/%s has IP(s) not match to its pod %s: %w, skipping", endpointAddress.Addresses, ed.Meta.Namespace, ed.Meta.Name, pod.Name, err)
+				continue
+			}
+			if err := validatePod(pod, nodeLister, networkEndpoint); err != nil {
+				klog.Errorf("Endpoint %q in Endpoints %s/%s correponds to an invalid pod: %v, skipping", endpointAddress.Addresses, ed.Meta.Namespace, ed.Meta.Name, err)
+				continue
+			}
 			if networkEndpointType == negtypes.NonGCPPrivateEndpointType {
 				// Non-GCP network endpoints don't have associated nodes.
 				networkEndpoint.Node = ""
@@ -519,7 +528,8 @@ func toZoneNetworkEndpointMapDegradedMode(eds []negtypes.EndpointsData, zoneGett
 // it returns error if the pod:
 // 1. is in terminal state
 // 2. corresponds to a non-existent node
-func validatePod(pod *apiv1.Pod, nodeLister cache.Indexer) error {
+// 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, networkEndpoint negtypes.NetworkEndpoint) error {
 	// Terminal Pod means a pod is in PodFailed or PodSucceeded phase
 	phase := pod.Status.Phase
 	if phase == apiv1.PodFailed || phase == apiv1.PodSucceeded {
@@ -529,10 +539,13 @@ func validatePod(pod *apiv1.Pod, nodeLister cache.Indexer) error {
 	if err != nil || !exists {
 		return negtypes.ErrEPNodeNotFound
 	}
-	_, isNode := obj.(*apiv1.Node)
-	if !isNode {
+	node, ok := obj.(*apiv1.Node)
+	if !ok {
 		return negtypes.ErrEPNodeTypeAssertionFailed
 	}
+	if err = nodeContainsPodIP(node, networkEndpoint); err != nil {
+		return err
+	}
 	return nil
 }
 
@@ -562,6 +575,65 @@ 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's IP address does not any of them.
+// If this is a dual stack endpoint, we would validate both IPs
+func podContainsEndpointAddress(networkEndpoint negtypes.NetworkEndpoint, pod *apiv1.Pod) error {
+	endpointIPs := []string{networkEndpoint.IP}
+	if networkEndpoint.IPv6 != "" {
+		endpointIPs = append(endpointIPs, networkEndpoint.IPv6)
+	}
+
+	matching := 0
+	for _, endpointIP := range endpointIPs {
+		// a pod can have at most two PodIPs, one for ipv4 and one for ipv6
+		for _, podIP := range pod.Status.PodIPs {
+			if endpointIP == podIP.IP {
+				matching += 1
+			}
+		}
+	}
+	if matching != len(endpointIPs) {
+		return fmt.Errorf("%w: endpoint %v has IP(s) not match to its pod's IP(s)", negtypes.ErrEPIPNotFromPod, endpointIPs)
+	}
+	return nil
+}
+
+// nodeContainsPodIP checks the node's existing PodCIDR(s),
+// and return error if the pod IP used by the endpoint is not within one of the podCIDR ranges.
+// If this is a dual stack endpoint, we would validate both pod IPs
+func nodeContainsPodIP(node *apiv1.Node, networkEndpoint negtypes.NetworkEndpoint) 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)
+	}
+	podIPs := []net.IP{net.ParseIP(networkEndpoint.IP)}
+	if networkEndpoint.IPv6 != "" {
+		podIPs = append(podIPs, net.ParseIP(networkEndpoint.IPv6))
+	}
+
+	matching := 0
+	for _, podIP := range podIPs {
+		for _, net := range ipnets {
+			if net.Contains(podIP) {
+				matching += 1
+				break
+			}
+		}
+	}
+	if matching != len(podIPs) {
+		return fmt.Errorf("%w: podIP(s) used by endpoint %v not match to the node's PodCIDR range(s)", negtypes.ErrEPIPOutOfPodCIDR, podIPs)
+	}
+	return nil
+}
+
 // 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.