github
/
k3s
mirror of https://github.com/k3s-io/k3s
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

use topologyPairsMaps for inter pod affinity/anti-affinity maps

pull/8/head
Ahmad Diaa 2018-08-22 00:58:40 +02:00
parent e41d9f1553
commit ac4a082b33
3 changed files with 141 additions and 152 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

158
pkg/scheduler/algorithm/predicates/metadata.go
View File

@ -71,15 +71,15 @@ type predicateMetadata struct {
podPorts []*v1.ContainerPort podPorts []*v1.ContainerPort
topologyPairsAntiAffinityPodsMap *topologyPairsMaps topologyPairsAntiAffinityPodsMap *topologyPairsMaps
// A map of node name to a list of Pods on the node that can potentially match // A map of topology pairs to a list of Pods that can potentially match
// the affinity rules of the "pod". // the affinity rules of the "pod" and its inverse.
nodeNameToMatchingAffinityPods map[string][]*v1.Pod topologyPairsPotentialAffinityPods *topologyPairsMaps
// A map of node name to a list of Pods on the node that can potentially match // A map of topology pairs to a list of Pods that can potentially match
// the anti-affinity rules of the "pod". // the anti-affinity rules of the "pod" and its inverse.
nodeNameToMatchingAntiAffinityPods map[string][]*v1.Pod topologyPairsPotentialAntiAffinityPods *topologyPairsMaps
serviceAffinityInUse bool serviceAffinityInUse bool
serviceAffinityMatchingPodList []*v1.Pod serviceAffinityMatchingPodList []*v1.Pod
serviceAffinityMatchingPodServices []*v1.Service serviceAffinityMatchingPodServices []*v1.Service
// ignoredExtendedResources is a set of extended resource names that will // ignoredExtendedResources is a set of extended resource names that will
// be ignored in the PodFitsResources predicate. // be ignored in the PodFitsResources predicate.
// //
@ -134,19 +134,19 @@ func (pfactory *PredicateMetadataFactory) GetMetadata(pod *v1.Pod, nodeNameToInf
if err != nil { if err != nil {
return nil return nil
} }
affinityPods, antiAffinityPods, err := getPodsMatchingAffinity(pod, nodeNameToInfoMap) topologyPairsAffinityPodsMaps, topologyPairsAntiAffinityPodsMaps, err := getPodsMatchingAffinity(pod, nodeNameToInfoMap)
if err != nil { if err != nil {
glog.Errorf("[predicate meta data generation] error finding pods that match affinity terms: %v", err) glog.Errorf("[predicate meta data generation] error finding pods that match affinity terms: %v", err)
return nil return nil
} }
predicateMetadata := &predicateMetadata{ predicateMetadata := &predicateMetadata{
pod: pod, pod: pod,
podBestEffort: isPodBestEffort(pod), podBestEffort: isPodBestEffort(pod),
podRequest: GetResourceRequest(pod), podRequest: GetResourceRequest(pod),
podPorts: schedutil.GetContainerPorts(pod), podPorts: schedutil.GetContainerPorts(pod),
nodeNameToMatchingAffinityPods: affinityPods, topologyPairsPotentialAffinityPods: topologyPairsAffinityPodsMaps,
nodeNameToMatchingAntiAffinityPods: antiAffinityPods, topologyPairsPotentialAntiAffinityPods: topologyPairsAntiAffinityPodsMaps,
topologyPairsAntiAffinityPodsMap: topologyPairsMaps, topologyPairsAntiAffinityPodsMap: topologyPairsMaps,
} }
for predicateName, precomputeFunc := range predicateMetadataProducers { for predicateName, precomputeFunc := range predicateMetadataProducers {
glog.V(10).Infof("Precompute: %v", predicateName) glog.V(10).Infof("Precompute: %v", predicateName)
@ -200,33 +200,9 @@ func (meta *predicateMetadata) RemovePod(deletedPod *v1.Pod) error {
return fmt.Errorf("deletedPod and meta.pod must not be the same") return fmt.Errorf("deletedPod and meta.pod must not be the same")
} }
meta.topologyPairsAntiAffinityPodsMap.removePod(deletedPod) meta.topologyPairsAntiAffinityPodsMap.removePod(deletedPod)
// Delete pod from the matching affinity or anti-affinity pods if exists. // Delete pod from the matching affinity or anti-affinity topology pairs maps.
affinity := meta.pod.Spec.Affinity meta.topologyPairsPotentialAffinityPods.removePod(deletedPod)
podNodeName := deletedPod.Spec.NodeName meta.topologyPairsPotentialAntiAffinityPods.removePod(deletedPod)
if affinity != nil && len(podNodeName) > 0 {
if affinity.PodAffinity != nil {
for i, p := range meta.nodeNameToMatchingAffinityPods[podNodeName] {
if p == deletedPod {
s := meta.nodeNameToMatchingAffinityPods[podNodeName]
s[i] = s[len(s)-1]
s = s[:len(s)-1]
meta.nodeNameToMatchingAffinityPods[podNodeName] = s
break
}
}
}
if affinity.PodAntiAffinity != nil {
for i, p := range meta.nodeNameToMatchingAntiAffinityPods[podNodeName] {
if p == deletedPod {
s := meta.nodeNameToMatchingAntiAffinityPods[podNodeName]
s[i] = s[len(s)-1]
s = s[:len(s)-1]
meta.nodeNameToMatchingAntiAffinityPods[podNodeName] = s
break
}
}
}
}
// All pods in the serviceAffinityMatchingPodList are in the same namespace. // All pods in the serviceAffinityMatchingPodList are in the same namespace.
// So, if the namespace of the first one is not the same as the namespace of the // So, if the namespace of the first one is not the same as the namespace of the
// deletedPod, we don't need to check the list, as deletedPod isn't in the list. // deletedPod, we don't need to check the list, as deletedPod isn't in the list.
@ -267,29 +243,24 @@ func (meta *predicateMetadata) AddPod(addedPod *v1.Pod, nodeInfo *schedulercache
affinity := meta.pod.Spec.Affinity affinity := meta.pod.Spec.Affinity
podNodeName := addedPod.Spec.NodeName podNodeName := addedPod.Spec.NodeName
if affinity != nil && len(podNodeName) > 0 { if affinity != nil && len(podNodeName) > 0 {
podNode := nodeInfo.Node()
affinityTerms := GetPodAffinityTerms(affinity.PodAffinity)
antiAffinityTerms := GetPodAntiAffinityTerms(affinity.PodAntiAffinity)
if targetPodMatchesAffinityOfPod(meta.pod, addedPod) { if targetPodMatchesAffinityOfPod(meta.pod, addedPod) {
found := false for _, term := range affinityTerms {
for _, p := range meta.nodeNameToMatchingAffinityPods[podNodeName] { if topologyValue, ok := podNode.Labels[term.TopologyKey]; ok {
if p == addedPod { pair := topologyPair{key: term.TopologyKey, value: topologyValue}
found = true meta.topologyPairsPotentialAffinityPods.addTopologyPair(pair, addedPod)
break
} }
} }
if !found {
meta.nodeNameToMatchingAffinityPods[podNodeName] = append(meta.nodeNameToMatchingAffinityPods[podNodeName], addedPod)
}
} }
if targetPodMatchesAntiAffinityOfPod(meta.pod, addedPod) { if targetPodMatchesAntiAffinityOfPod(meta.pod, addedPod) {
found := false for _, term := range antiAffinityTerms {
for _, p := range meta.nodeNameToMatchingAntiAffinityPods[podNodeName] { if topologyValue, ok := podNode.Labels[term.TopologyKey]; ok {
if p == addedPod { pair := topologyPair{key: term.TopologyKey, value: topologyValue}
found = true meta.topologyPairsPotentialAntiAffinityPods.addTopologyPair(pair, addedPod)
break
} }
} }
if !found {
meta.nodeNameToMatchingAntiAffinityPods[podNodeName] = append(meta.nodeNameToMatchingAntiAffinityPods[podNodeName], addedPod)
}
} }
} }
// If addedPod is in the same namespace as the meta.pod, update the list // If addedPod is in the same namespace as the meta.pod, update the list
@ -308,22 +279,17 @@ func (meta *predicateMetadata) AddPod(addedPod *v1.Pod, nodeInfo *schedulercache
// its maps and slices, but it does not copy the contents of pointer values. // its maps and slices, but it does not copy the contents of pointer values.
func (meta *predicateMetadata) ShallowCopy() algorithm.PredicateMetadata { func (meta *predicateMetadata) ShallowCopy() algorithm.PredicateMetadata {
newPredMeta := &predicateMetadata{ newPredMeta := &predicateMetadata{
pod: meta.pod, pod: meta.pod,
podBestEffort: meta.podBestEffort, podBestEffort: meta.podBestEffort,
podRequest: meta.podRequest, podRequest: meta.podRequest,
serviceAffinityInUse: meta.serviceAffinityInUse, serviceAffinityInUse: meta.serviceAffinityInUse,
ignoredExtendedResources: meta.ignoredExtendedResources, ignoredExtendedResources: meta.ignoredExtendedResources,
topologyPairsAntiAffinityPodsMap: meta.topologyPairsAntiAffinityPodsMap,
} }
newPredMeta.podPorts = append([]*v1.ContainerPort(nil), meta.podPorts...) newPredMeta.podPorts = append([]*v1.ContainerPort(nil), meta.podPorts...)
newPredMeta.nodeNameToMatchingAffinityPods = make(map[string][]*v1.Pod) newPredMeta.topologyPairsPotentialAffinityPods = newTopologyPairsMaps()
for k, v := range meta.nodeNameToMatchingAffinityPods { newPredMeta.topologyPairsPotentialAffinityPods.appendMaps(meta.topologyPairsPotentialAffinityPods)
newPredMeta.nodeNameToMatchingAffinityPods[k] = append([]*v1.Pod(nil), v...) newPredMeta.topologyPairsPotentialAntiAffinityPods = newTopologyPairsMaps()
} newPredMeta.topologyPairsPotentialAntiAffinityPods.appendMaps(meta.topologyPairsPotentialAntiAffinityPods)
newPredMeta.nodeNameToMatchingAntiAffinityPods = make(map[string][]*v1.Pod)
for k, v := range meta.nodeNameToMatchingAntiAffinityPods {
newPredMeta.nodeNameToMatchingAntiAffinityPods[k] = append([]*v1.Pod(nil), v...)
}
newPredMeta.topologyPairsAntiAffinityPodsMap = newTopologyPairsMaps() newPredMeta.topologyPairsAntiAffinityPodsMap = newTopologyPairsMaps()
newPredMeta.topologyPairsAntiAffinityPodsMap.appendMaps(meta.topologyPairsAntiAffinityPodsMap) newPredMeta.topologyPairsAntiAffinityPodsMap.appendMaps(meta.topologyPairsAntiAffinityPodsMap)
newPredMeta.serviceAffinityMatchingPodServices = append([]*v1.Service(nil), newPredMeta.serviceAffinityMatchingPodServices = append([]*v1.Service(nil),
@ -373,12 +339,12 @@ func podMatchesAffinityTermProperties(pod *v1.Pod, properties []*affinityTermPro
// It ignores topology. It returns a set of Pods that are checked later by the affinity // 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 // predicate. With this set of pods available, the affinity predicate does not
// need to check all the pods in the cluster. // need to check all the pods in the cluster.
func getPodsMatchingAffinity(pod *v1.Pod, nodeInfoMap map[string]*schedulercache.NodeInfo) (affinityPods map[string][]*v1.Pod, antiAffinityPods map[string][]*v1.Pod, err error) { func getPodsMatchingAffinity(pod *v1.Pod, nodeInfoMap map[string]*schedulercache.NodeInfo) (topologyPairsAffinityPodsMaps *topologyPairsMaps, topologyPairsAntiAffinityPodsMaps *topologyPairsMaps, err error) {
allNodeNames := make([]string, 0, len(nodeInfoMap)) allNodeNames := make([]string, 0, len(nodeInfoMap))
affinity := pod.Spec.Affinity affinity := pod.Spec.Affinity
if affinity == nil || (affinity.PodAffinity == nil && affinity.PodAntiAffinity == nil) { if affinity == nil || (affinity.PodAffinity == nil && affinity.PodAntiAffinity == nil) {
return nil, nil, nil return newTopologyPairsMaps(), newTopologyPairsMaps(), nil
} }
for name := range nodeInfoMap { for name := range nodeInfoMap {
@ -387,16 +353,16 @@ func getPodsMatchingAffinity(pod *v1.Pod, nodeInfoMap map[string]*schedulercache
var lock sync.Mutex var lock sync.Mutex
var firstError error var firstError error
affinityPods = make(map[string][]*v1.Pod) topologyPairsAffinityPodsMaps = newTopologyPairsMaps()
antiAffinityPods = make(map[string][]*v1.Pod) topologyPairsAntiAffinityPodsMaps = newTopologyPairsMaps()
appendResult := func(nodeName string, affPods, antiAffPods []*v1.Pod) { appendResult := func(nodeName string, nodeTopologyPairsAffinityPodsMaps, nodeTopologyPairsAntiAffinityPodsMaps *topologyPairsMaps) {
lock.Lock() lock.Lock()
defer lock.Unlock() defer lock.Unlock()
if len(affPods) > 0 { if len(nodeTopologyPairsAffinityPodsMaps.topologyPairToPods) > 0 {
affinityPods[nodeName] = affPods topologyPairsAffinityPodsMaps.appendMaps(nodeTopologyPairsAffinityPodsMaps)
} }
if len(antiAffPods) > 0 { if len(nodeTopologyPairsAntiAffinityPodsMaps.topologyPairToPods) > 0 {
antiAffinityPods[nodeName] = antiAffPods topologyPairsAntiAffinityPodsMaps.appendMaps(nodeTopologyPairsAntiAffinityPodsMaps)
} }
} }
@ -417,6 +383,8 @@ func getPodsMatchingAffinity(pod *v1.Pod, nodeInfoMap map[string]*schedulercache
return nil, nil, err return nil, nil, err
} }
affinityTerms := GetPodAffinityTerms(affinity.PodAffinity)
antiAffinityTerms := GetPodAntiAffinityTerms(affinity.PodAntiAffinity)
processNode := func(i int) { processNode := func(i int) {
nodeInfo := nodeInfoMap[allNodeNames[i]] nodeInfo := nodeInfoMap[allNodeNames[i]]
node := nodeInfo.Node() node := nodeInfo.Node()
@ -424,24 +392,34 @@ func getPodsMatchingAffinity(pod *v1.Pod, nodeInfoMap map[string]*schedulercache
catchError(fmt.Errorf("nodeInfo.Node is nil")) catchError(fmt.Errorf("nodeInfo.Node is nil"))
return return
} }
affPods := make([]*v1.Pod, 0, len(nodeInfo.Pods())) nodeTopologyPairsAffinityPodsMaps := newTopologyPairsMaps()
antiAffPods := make([]*v1.Pod, 0, len(nodeInfo.Pods())) nodeTopologyPairsAntiAffinityPodsMaps := newTopologyPairsMaps()
for _, existingPod := range nodeInfo.Pods() { for _, existingPod := range nodeInfo.Pods() {
// Check affinity properties. // Check affinity properties.
if podMatchesAffinityTermProperties(existingPod, affinityProperties) { if podMatchesAffinityTermProperties(existingPod, affinityProperties) {
affPods = append(affPods, existingPod) for _, term := range affinityTerms {
if topologyValue, ok := node.Labels[term.TopologyKey]; ok {
pair := topologyPair{key: term.TopologyKey, value: topologyValue}
nodeTopologyPairsAffinityPodsMaps.addTopologyPair(pair, existingPod)
}
}
} }
// Check anti-affinity properties. // Check anti-affinity properties.
if podMatchesAffinityTermProperties(existingPod, antiAffinityProperties) { if podMatchesAffinityTermProperties(existingPod, antiAffinityProperties) {
antiAffPods = append(antiAffPods, existingPod) for _, term := range antiAffinityTerms {
if topologyValue, ok := node.Labels[term.TopologyKey]; ok {
pair := topologyPair{key: term.TopologyKey, value: topologyValue}
nodeTopologyPairsAntiAffinityPodsMaps.addTopologyPair(pair, existingPod)
}
}
} }
} }
if len(antiAffPods) > 0 || len(affPods) > 0 { if len(nodeTopologyPairsAffinityPodsMaps.topologyPairToPods) > 0 || len(nodeTopologyPairsAntiAffinityPodsMaps.topologyPairToPods) > 0 {
appendResult(node.Name, affPods, antiAffPods) appendResult(node.Name, nodeTopologyPairsAffinityPodsMaps, nodeTopologyPairsAntiAffinityPodsMaps)
} }
} }
workqueue.Parallelize(16, len(allNodeNames), processNode) workqueue.Parallelize(16, len(allNodeNames), processNode)
return affinityPods, antiAffinityPods, firstError return topologyPairsAffinityPodsMaps, topologyPairsAntiAffinityPodsMaps, firstError
} }
// podMatchesAffinity returns true if "targetPod" matches any affinity rule of // podMatchesAffinity returns true if "targetPod" matches any affinity rule of

102
pkg/scheduler/algorithm/predicates/metadata_test.go
View File

@ -52,13 +52,6 @@ func (s sortableServices) Swap(i, j int) { s[i], s[j] = s[j], s[i] }
var _ = sort.Interface(&sortableServices{}) var _ = sort.Interface(&sortableServices{})
func sortNodePodMap(np map[string][]*v1.Pod) {
for _, pl := range np {
sortablePods := sortablePods(pl)
sort.Sort(sortablePods)
}
}
// predicateMetadataEquivalent returns true if the two metadata are equivalent. // predicateMetadataEquivalent returns true if the two metadata are equivalent.
// Note: this function does not compare podRequest. // Note: this function does not compare podRequest.
func predicateMetadataEquivalent(meta1, meta2 *predicateMetadata) error { func predicateMetadataEquivalent(meta1, meta2 *predicateMetadata) error {
@ -77,15 +70,11 @@ func predicateMetadataEquivalent(meta1, meta2 *predicateMetadata) error {
for !reflect.DeepEqual(meta1.podPorts, meta2.podPorts) { for !reflect.DeepEqual(meta1.podPorts, meta2.podPorts) {
return fmt.Errorf("podPorts are not equal") return fmt.Errorf("podPorts are not equal")
} }
sortNodePodMap(meta1.nodeNameToMatchingAffinityPods) if !reflect.DeepEqual(meta1.topologyPairsPotentialAffinityPods, meta2.topologyPairsPotentialAffinityPods) {
sortNodePodMap(meta2.nodeNameToMatchingAffinityPods) return fmt.Errorf("topologyPairsPotentialAffinityPods are not equal")
if !reflect.DeepEqual(meta1.nodeNameToMatchingAffinityPods, meta2.nodeNameToMatchingAffinityPods) {
return fmt.Errorf("nodeNameToMatchingAffinityPods are not euqal")
} }
sortNodePodMap(meta1.nodeNameToMatchingAntiAffinityPods) if !reflect.DeepEqual(meta1.topologyPairsPotentialAntiAffinityPods, meta2.topologyPairsPotentialAntiAffinityPods) {
sortNodePodMap(meta2.nodeNameToMatchingAntiAffinityPods) return fmt.Errorf("topologyPairsPotentialAntiAffinityPods are not equal")
if !reflect.DeepEqual(meta1.nodeNameToMatchingAntiAffinityPods, meta2.nodeNameToMatchingAntiAffinityPods) {
return fmt.Errorf("nodeNameToMatchingAntiAffinityPods are not euqal")
} }
if !reflect.DeepEqual(meta1.topologyPairsAntiAffinityPodsMap.podToTopologyPairs, if !reflect.DeepEqual(meta1.topologyPairsAntiAffinityPodsMap.podToTopologyPairs,
meta2.topologyPairsAntiAffinityPodsMap.podToTopologyPairs) { meta2.topologyPairsAntiAffinityPodsMap.podToTopologyPairs) {
@ -454,42 +443,71 @@ func TestPredicateMetadata_ShallowCopy(t *testing.T) {
}, },
}, },
}, },
nodeNameToMatchingAffinityPods: map[string][]*v1.Pod{ topologyPairsPotentialAffinityPods: &topologyPairsMaps{
"nodeA": { topologyPairToPods: map[topologyPair]podSet{
&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "p1", Labels: selector1}, {key: "name", value: "nodeA"}: {
Spec: v1.PodSpec{NodeName: "nodeA"}, &v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "p1", Labels: selector1},
Spec: v1.PodSpec{NodeName: "nodeA"},
}: struct{}{},
},
{key: "name", value: "nodeC"}: {
&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "p2"},
Spec: v1.PodSpec{
NodeName: "nodeC",
},
}: struct{}{},
&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "p6", Labels: selector1},
Spec: v1.PodSpec{NodeName: "nodeC"},
}: struct{}{},
}, },
}, },
"nodeC": { podToTopologyPairs: map[string]topologyPairSet{
&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "p2"}, "p1_": {
Spec: v1.PodSpec{ topologyPair{key: "name", value: "nodeA"}: struct{}{},
NodeName: "nodeC",
},
}, },
&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "p6", Labels: selector1}, "p2_": {
Spec: v1.PodSpec{NodeName: "nodeC"}, topologyPair{key: "name", value: "nodeC"}: struct{}{},
},
"p6_": {
topologyPair{key: "name", value: "nodeC"}: struct{}{},
}, },
}, },
}, },
nodeNameToMatchingAntiAffinityPods: map[string][]*v1.Pod{ topologyPairsPotentialAntiAffinityPods: &topologyPairsMaps{
"nodeN": { topologyPairToPods: map[topologyPair]podSet{
&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "p1", Labels: selector1}, {key: "name", value: "nodeN"}: {
Spec: v1.PodSpec{NodeName: "nodeN"}, &v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "p1", Labels: selector1},
Spec: v1.PodSpec{NodeName: "nodeN"},
}: struct{}{},
&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "p2"},
Spec: v1.PodSpec{
NodeName: "nodeM",
},
}: struct{}{},
&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "p3"},
Spec: v1.PodSpec{
NodeName: "nodeM",
},
}: struct{}{},
}, },
&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "p2"}, {key: "name", value: "nodeM"}: {
Spec: v1.PodSpec{ &v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "p6", Labels: selector1},
NodeName: "nodeM", Spec: v1.PodSpec{NodeName: "nodeM"},
}, }: struct{}{},
},
&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "p3"},
Spec: v1.PodSpec{
NodeName: "nodeM",
},
}, },
}, },
"nodeM": { podToTopologyPairs: map[string]topologyPairSet{
&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "p6", Labels: selector1}, "p1_": {
Spec: v1.PodSpec{NodeName: "nodeM"}, topologyPair{key: "name", value: "nodeN"}: struct{}{},
},
"p2_": {
topologyPair{key: "name", value: "nodeN"}: struct{}{},
},
"p3_": {
topologyPair{key: "name", value: "nodeN"}: struct{}{},
},
"p6_": {
topologyPair{key: "name", value: "nodeM"}: struct{}{},
}, },
}, },
}, },

33
pkg/scheduler/algorithm/predicates/predicates.go
View File

@ -1395,23 +1395,16 @@ func (c *PodAffinityChecker) satisfiesExistingPodsAntiAffinity(pod *v1.Pod, meta
// anyPodsMatchingTopologyTerms checks whether any of the nodes given via // anyPodsMatchingTopologyTerms checks whether any of the nodes given via
// "targetPods" matches topology of all the "terms" for the give "pod" and "nodeInfo". // "targetPods" matches topology of all the "terms" for the give "pod" and "nodeInfo".
func (c *PodAffinityChecker) anyPodsMatchingTopologyTerms(pod *v1.Pod, targetPods map[string][]*v1.Pod, nodeInfo *schedulercache.NodeInfo, terms []v1.PodAffinityTerm) (bool, error) { func (c *PodAffinityChecker) anyPodsMatchingTopologyTerms(pod *v1.Pod, targetPods *topologyPairsMaps, nodeInfo *schedulercache.NodeInfo, terms []v1.PodAffinityTerm) (bool, error) {
for nodeName, targetPods := range targetPods { podNameToMatchingTermsCount := make(map[string]int)
targetPodNodeInfo, err := c.info.GetNodeInfo(nodeName) node := nodeInfo.Node()
if err != nil { podTermsCount := len(terms)
return false, err for _, term := range terms {
} pair := topologyPair{key: term.TopologyKey, value: node.Labels[term.TopologyKey]}
if len(targetPods) > 0 { for existingPod := range targetPods.topologyPairToPods[pair] {
allTermsMatched := true existingPodFullName := schedutil.GetPodFullName(existingPod)
for _, term := range terms { podNameToMatchingTermsCount[existingPodFullName] = podNameToMatchingTermsCount[existingPodFullName] + 1
if !priorityutil.NodesHaveSameTopologyKey(nodeInfo.Node(), targetPodNodeInfo, term.TopologyKey) { if podNameToMatchingTermsCount[existingPodFullName] == podTermsCount {
allTermsMatched = false
break
}
}
if allTermsMatched {
// We have 1 or more pods on the target node that have already matched namespace and selector
// and all of the terms topologies matched the target node. So, there is at least 1 matching pod on the node.
return true, nil return true, nil
} }
} }
@ -1429,7 +1422,7 @@ func (c *PodAffinityChecker) satisfiesPodsAffinityAntiAffinity(pod *v1.Pod,
} }
if predicateMeta, ok := meta.(*predicateMetadata); ok { if predicateMeta, ok := meta.(*predicateMetadata); ok {
// Check all affinity terms. // Check all affinity terms.
matchingPods := predicateMeta.nodeNameToMatchingAffinityPods matchingPods := predicateMeta.topologyPairsPotentialAffinityPods
if affinityTerms := GetPodAffinityTerms(affinity.PodAffinity); len(affinityTerms) > 0 { if affinityTerms := GetPodAffinityTerms(affinity.PodAffinity); len(affinityTerms) > 0 {
matchExists, err := c.anyPodsMatchingTopologyTerms(pod, matchingPods, nodeInfo, affinityTerms) matchExists, err := c.anyPodsMatchingTopologyTerms(pod, matchingPods, nodeInfo, affinityTerms)
if err != nil { if err != nil {
@ -1442,7 +1435,7 @@ func (c *PodAffinityChecker) satisfiesPodsAffinityAntiAffinity(pod *v1.Pod,
// to not leave such pods in pending state forever, we check that if no other pod // to not leave such pods in pending state forever, we check that if no other pod
// in the cluster matches the namespace and selector of this pod and the pod matches // in the cluster matches the namespace and selector of this pod and the pod matches
// its own terms, then we allow the pod to pass the affinity check. // its own terms, then we allow the pod to pass the affinity check.
if !(len(matchingPods) == 0 && targetPodMatchesAffinityOfPod(pod, pod)) { if !(len(matchingPods.topologyPairToPods) == 0 && targetPodMatchesAffinityOfPod(pod, pod)) {
glog.V(10).Infof("Cannot schedule pod %+v onto node %v, because of PodAffinity", glog.V(10).Infof("Cannot schedule pod %+v onto node %v, because of PodAffinity",
podName(pod), node.Name) podName(pod), node.Name)
return ErrPodAffinityRulesNotMatch, nil return ErrPodAffinityRulesNotMatch, nil
@ -1451,7 +1444,7 @@ func (c *PodAffinityChecker) satisfiesPodsAffinityAntiAffinity(pod *v1.Pod,
} }
// Check all anti-affinity terms. // Check all anti-affinity terms.
matchingPods = predicateMeta.nodeNameToMatchingAntiAffinityPods matchingPods = predicateMeta.topologyPairsPotentialAntiAffinityPods
if antiAffinityTerms := GetPodAntiAffinityTerms(affinity.PodAntiAffinity); len(antiAffinityTerms) > 0 { if antiAffinityTerms := GetPodAntiAffinityTerms(affinity.PodAntiAffinity); len(antiAffinityTerms) > 0 {
matchExists, err := c.anyPodsMatchingTopologyTerms(pod, matchingPods, nodeInfo, antiAffinityTerms) matchExists, err := c.anyPodsMatchingTopologyTerms(pod, matchingPods, nodeInfo, antiAffinityTerms)
if err != nil || matchExists { if err != nil || matchExists {