fix PodAntiAffinity issues

- update logic of verifying incoming pod's anti-affinity
- rename podMatchesAffinityTermProperties to podMatchesAllAffinityTermProperties
- add podMatchesAnyAffinityTermProperties which is used in some PodAntiAffinity cases
- rename some functions to make it more readable
- add unit tests to verify correctness of PodAffinity and PodAntiAffinity
  - verifying "Existing pod anti-affinity"
  - verifying "incoming pod's anti-affinity"
  - verifying "incoming pod's affinity"

Author: Huang-Wei

pkg/scheduler/algorithm/predicates/metadata.go

@@ -72,10 +72,10 @@ type predicateMetadata struct {
 
 	topologyPairsAntiAffinityPodsMap *topologyPairsMaps
 	// A map of topology pairs to a list of Pods that can potentially match
-	// the affinity rules of the "pod" and its inverse.
+	// the affinity terms of the "pod" and its inverse.
 	topologyPairsPotentialAffinityPods *topologyPairsMaps
 	// A map of topology pairs to a list of Pods that can potentially match
-	// the anti-affinity rules of the "pod" and its inverse.
+	// the anti-affinity terms of the "pod" and its inverse.
 	topologyPairsPotentialAntiAffinityPods *topologyPairsMaps
 	serviceAffinityInUse                   bool
 	serviceAffinityMatchingPodList         []*v1.Pod
@@ -130,11 +130,14 @@ func (pfactory *PredicateMetadataFactory) GetMetadata(pod *v1.Pod, nodeNameToInf
 	if pod == nil {
 		return nil
 	}
-	topologyPairsMaps, err := getMatchingTopologyPairs(pod, nodeNameToInfoMap)
+	// existingPodAntiAffinityMap will be used later for efficient check on existing pods' anti-affinity
+	existingPodAntiAffinityMap, err := getTPMapMatchingExistingAntiAffinity(pod, nodeNameToInfoMap)
 	if err != nil {
 		return nil
 	}
-	topologyPairsAffinityPodsMaps, topologyPairsAntiAffinityPodsMaps, err := getPodsMatchingAffinity(pod, nodeNameToInfoMap)
+	// incomingPodAffinityMap will be used later for efficient check on incoming pod's affinity
+	// incomingPodAntiAffinityMap will be used later for efficient check on incoming pod's anti-affinity
+	incomingPodAffinityMap, incomingPodAntiAffinityMap, err := getTPMapMatchingIncomingAffinityAntiAffinity(pod, nodeNameToInfoMap)
 	if err != nil {
 		glog.Errorf("[predicate meta data generation] error finding pods that match affinity terms: %v", err)
 		return nil
@@ -144,9 +147,9 @@ func (pfactory *PredicateMetadataFactory) GetMetadata(pod *v1.Pod, nodeNameToInf
 		podBestEffort: isPodBestEffort(pod),
 		podRequest:    GetResourceRequest(pod),
 		podPorts:      schedutil.GetContainerPorts(pod),
-		topologyPairsPotentialAffinityPods:     topologyPairsAffinityPodsMaps,
-		topologyPairsPotentialAntiAffinityPods: topologyPairsAntiAffinityPodsMaps,
-		topologyPairsAntiAffinityPodsMap:       topologyPairsMaps,
+		topologyPairsPotentialAffinityPods:     incomingPodAffinityMap,
+		topologyPairsPotentialAntiAffinityPods: incomingPodAntiAffinityMap,
+		topologyPairsAntiAffinityPodsMap:       existingPodAntiAffinityMap,
 	}
 	for predicateName, precomputeFunc := range predicateMetadataProducers {
 		glog.V(10).Infof("Precompute: %v", predicateName)
@@ -185,6 +188,9 @@ func (topologyPairsMaps *topologyPairsMaps) removePod(deletedPod *v1.Pod) {
 }
 
 func (topologyPairsMaps *topologyPairsMaps) appendMaps(toAppend *topologyPairsMaps) {
+	if toAppend == nil {
+		return
+	}
 	for pair := range toAppend.topologyPairToPods {
 		for pod := range toAppend.topologyPairToPods[pair] {
 			topologyPairsMaps.addTopologyPair(pair, pod)
@@ -232,13 +238,11 @@ func (meta *predicateMetadata) AddPod(addedPod *v1.Pod, nodeInfo *schedulercache
 		return fmt.Errorf("invalid node in nodeInfo")
 	}
 	// Add matching anti-affinity terms of the addedPod to the map.
-	topologyPairsMaps, err := getMatchingTopologyPairsOfExistingPod(meta.pod, addedPod, nodeInfo.Node())
+	topologyPairsMaps, err := getMatchingAntiAffinityTopologyPairsOfPod(meta.pod, addedPod, nodeInfo.Node())
 	if err != nil {
 		return err
 	}
-	if len(topologyPairsMaps.podToTopologyPairs) > 0 {
-		meta.topologyPairsAntiAffinityPodsMap.appendMaps(topologyPairsMaps)
-	}
+	meta.topologyPairsAntiAffinityPodsMap.appendMaps(topologyPairsMaps)
 	// Add the pod to nodeNameToMatchingAffinityPods and nodeNameToMatchingAntiAffinityPods if needed.
 	affinity := meta.pod.Spec.Affinity
 	podNodeName := addedPod.Spec.NodeName
@@ -325,8 +329,8 @@ func getAffinityTermProperties(pod *v1.Pod, terms []v1.PodAffinityTerm) (propert
 	return properties, nil
 }
 
-// podMatchesAffinityTermProperties return true IFF the given pod matches all the given properties.
-func podMatchesAffinityTermProperties(pod *v1.Pod, properties []*affinityTermProperties) bool {
+// podMatchesAllAffinityTermProperties returns true IFF the given pod matches all the given properties.
+func podMatchesAllAffinityTermProperties(pod *v1.Pod, properties []*affinityTermProperties) bool {
 	if len(properties) == 0 {
 		return false
 	}
@@ -338,11 +342,71 @@ func podMatchesAffinityTermProperties(pod *v1.Pod, properties []*affinityTermPro
 	return true
 }
 
-// getPodsMatchingAffinity finds existing Pods that match affinity terms of the given "pod".
-// It ignores topology. It returns a set of Pods that are checked later by the affinity
-// predicate. With this set of pods available, the affinity predicate does not
+// podMatchesAnyAffinityTermProperties returns true if the given pod matches any given property.
+func podMatchesAnyAffinityTermProperties(pod *v1.Pod, properties []*affinityTermProperties) bool {
+	if len(properties) == 0 {
+		return false
+	}
+	for _, property := range properties {
+		if priorityutil.PodMatchesTermsNamespaceAndSelector(pod, property.namespaces, property.selector) {
+			return true
+		}
+	}
+	return false
+}
+
+// getTPMapMatchingExistingAntiAffinity calculates the following for each existing pod on each node:
+// (1) Whether it has PodAntiAffinity
+// (2) Whether any AffinityTerm matches the incoming pod
+func getTPMapMatchingExistingAntiAffinity(pod *v1.Pod, nodeInfoMap map[string]*schedulercache.NodeInfo) (*topologyPairsMaps, error) {
+	allNodeNames := make([]string, 0, len(nodeInfoMap))
+	for name := range nodeInfoMap {
+		allNodeNames = append(allNodeNames, name)
+	}
+
+	var lock sync.Mutex
+	var firstError error
+
+	topologyMaps := newTopologyPairsMaps()
+
+	appendTopologyPairsMaps := func(toAppend *topologyPairsMaps) {
+		lock.Lock()
+		defer lock.Unlock()
+		topologyMaps.appendMaps(toAppend)
+	}
+	catchError := func(err error) {
+		lock.Lock()
+		defer lock.Unlock()
+		if firstError == nil {
+			firstError = err
+		}
+	}
+
+	processNode := func(i int) {
+		nodeInfo := nodeInfoMap[allNodeNames[i]]
+		node := nodeInfo.Node()
+		if node == nil {
+			catchError(fmt.Errorf("node not found"))
+			return
+		}
+		for _, existingPod := range nodeInfo.PodsWithAffinity() {
+			existingPodTopologyMaps, err := getMatchingAntiAffinityTopologyPairsOfPod(pod, existingPod, node)
+			if err != nil {
+				catchError(err)
+				return
+			}
+			appendTopologyPairsMaps(existingPodTopologyMaps)
+		}
+	}
+	workqueue.Parallelize(16, len(allNodeNames), processNode)
+	return topologyMaps, firstError
+}
+
+// getTPMapMatchingIncomingAffinityAntiAffinity finds existing Pods that match affinity terms of the given "pod".
+// It returns a topologyPairsMaps that are checked later by the affinity
+// predicate. With this topologyPairsMaps available, the affinity predicate does not
 // need to check all the pods in the cluster.
-func getPodsMatchingAffinity(pod *v1.Pod, nodeInfoMap map[string]*schedulercache.NodeInfo) (topologyPairsAffinityPodsMaps *topologyPairsMaps, topologyPairsAntiAffinityPodsMaps *topologyPairsMaps, err error) {
+func getTPMapMatchingIncomingAffinityAntiAffinity(pod *v1.Pod, nodeInfoMap map[string]*schedulercache.NodeInfo) (topologyPairsAffinityPodsMaps *topologyPairsMaps, topologyPairsAntiAffinityPodsMaps *topologyPairsMaps, err error) {
 	allNodeNames := make([]string, 0, len(nodeInfoMap))
 
 	affinity := pod.Spec.Affinity
@@ -377,17 +441,13 @@ func getPodsMatchingAffinity(pod *v1.Pod, nodeInfoMap map[string]*schedulercache
 		}
 	}
 
-	affinityProperties, err := getAffinityTermProperties(pod, GetPodAffinityTerms(affinity.PodAffinity))
-	if err != nil {
-		return nil, nil, err
-	}
-	antiAffinityProperties, err := getAffinityTermProperties(pod, GetPodAntiAffinityTerms(affinity.PodAntiAffinity))
+	affinityTerms := GetPodAffinityTerms(affinity.PodAffinity)
+	affinityProperties, err := getAffinityTermProperties(pod, affinityTerms)
 	if err != nil {
 		return nil, nil, err
 	}
-
-	affinityTerms := GetPodAffinityTerms(affinity.PodAffinity)
 	antiAffinityTerms := GetPodAntiAffinityTerms(affinity.PodAntiAffinity)
+
 	processNode := func(i int) {
 		nodeInfo := nodeInfoMap[allNodeNames[i]]
 		node := nodeInfo.Node()
@@ -399,7 +459,7 @@ func getPodsMatchingAffinity(pod *v1.Pod, nodeInfoMap map[string]*schedulercache
 		nodeTopologyPairsAntiAffinityPodsMaps := newTopologyPairsMaps()
 		for _, existingPod := range nodeInfo.Pods() {
 			// Check affinity properties.
-			if podMatchesAffinityTermProperties(existingPod, affinityProperties) {
+			if podMatchesAllAffinityTermProperties(existingPod, affinityProperties) {
 				for _, term := range affinityTerms {
 					if topologyValue, ok := node.Labels[term.TopologyKey]; ok {
 						pair := topologyPair{key: term.TopologyKey, value: topologyValue}
@@ -408,8 +468,14 @@ func getPodsMatchingAffinity(pod *v1.Pod, nodeInfoMap map[string]*schedulercache
 				}
 			}
 			// Check anti-affinity properties.
-			if podMatchesAffinityTermProperties(existingPod, antiAffinityProperties) {
-				for _, term := range antiAffinityTerms {
+			for _, term := range antiAffinityTerms {
+				namespaces := priorityutil.GetNamespacesFromPodAffinityTerm(pod, &term)
+				selector, err := metav1.LabelSelectorAsSelector(term.LabelSelector)
+				if err != nil {
+					catchError(err)
+					return
+				}
+				if priorityutil.PodMatchesTermsNamespaceAndSelector(existingPod, namespaces, selector) {
 					if topologyValue, ok := node.Labels[term.TopologyKey]; ok {
 						pair := topologyPair{key: term.TopologyKey, value: topologyValue}
 						nodeTopologyPairsAntiAffinityPodsMaps.addTopologyPair(pair, existingPod)
@@ -425,7 +491,7 @@ func getPodsMatchingAffinity(pod *v1.Pod, nodeInfoMap map[string]*schedulercache
 	return topologyPairsAffinityPodsMaps, topologyPairsAntiAffinityPodsMaps, firstError
 }
 
-// podMatchesAffinity returns true if "targetPod" matches any affinity rule of
+// targetPodMatchesAffinityOfPod returns true if "targetPod" matches ALL affinity terms of
 // "pod". Similar to getPodsMatchingAffinity, this function does not check topology.
 // So, whether the targetPod actually matches or not needs further checks for a specific
 // node.
@@ -439,11 +505,11 @@ func targetPodMatchesAffinityOfPod(pod, targetPod *v1.Pod) bool {
 		glog.Errorf("error in getting affinity properties of Pod %v", pod.Name)
 		return false
 	}
-	return podMatchesAffinityTermProperties(targetPod, affinityProperties)
+	return podMatchesAllAffinityTermProperties(targetPod, affinityProperties)
 }
 
-// targetPodMatchesAntiAffinityOfPod returns true if "targetPod" matches any anti-affinity
-// rule of "pod". Similar to getPodsMatchingAffinity, this function does not check topology.
+// targetPodMatchesAntiAffinityOfPod returns true if "targetPod" matches ANY anti-affinity
+// term of "pod". Similar to getPodsMatchingAffinity, this function does not check topology.
 // So, whether the targetPod actually matches or not needs further checks for a specific
 // node.
 func targetPodMatchesAntiAffinityOfPod(pod, targetPod *v1.Pod) bool {
@@ -456,5 +522,5 @@ func targetPodMatchesAntiAffinityOfPod(pod, targetPod *v1.Pod) bool {
 		glog.Errorf("error in getting anti-affinity properties of Pod %v", pod.Name)
 		return false
 	}
-	return podMatchesAffinityTermProperties(targetPod, properties)
+	return podMatchesAnyAffinityTermProperties(targetPod, properties)
 }