github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-07-30 15:05:27 +00:00
Code Issues Projects Releases Wiki Activity

Merge pull request #86205 from Huang-Wei/eps-pred-prefitler

Browse Source 
Move EvenPodsSpread metadata computation logic as a PreFilter Plugin
This commit is contained in:
Kubernetes Prow Robot 2019-12-13 21:19:36 -08:00 committed by GitHub
commit b92d20033e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
15 changed files with 411 additions and 741 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -105,11 +105,11 @@ func (paths *criticalPaths) update(tpVal string, num int32) {
}
}
// evenPodsSpreadMetadata combines tpKeyToCriticalPaths and tpPairToMatchNum
// PodTopologySpreadMetadata combines tpKeyToCriticalPaths and tpPairToMatchNum
// to represent:
// (1) critical paths where the least pods are matched on each spread constraint.
// (2) number of pods matched on each spread constraint.
type evenPodsSpreadMetadata struct {
type PodTopologySpreadMetadata struct {
constraints []topologySpreadConstraint
// We record 2 critical paths instead of all critical paths here.
// criticalPaths[0].matchNum always holds the minimum matching number.
@ -309,10 +309,6 @@ func (m *podFitsResourcesMetadata) clone() *podFitsResourcesMetadata {
type predicateMetadata struct {
pod *v1.Pod
// evenPodsSpreadMetadata holds info of the minimum match number on each topology spread constraint,
// and the match number of all valid topology pairs.
evenPodsSpreadMetadata *evenPodsSpreadMetadata
serviceAffinityMetadata *serviceAffinityMetadata
podFitsResourcesMetadata *podFitsResourcesMetadata
}
@ -357,27 +353,8 @@ func (f *MetadataProducerFactory) GetPredicateMetadata(pod *v1.Pod, sharedLister
return nil
}
var allNodes []*schedulernodeinfo.NodeInfo
if sharedLister != nil {
var err error
allNodes, err = sharedLister.NodeInfos().List()
if err != nil {
klog.Errorf("failed to list NodeInfos: %v", err)
return nil
}
}
// evenPodsSpreadMetadata represents how existing pods match "pod"
// on its spread constraints
evenPodsSpreadMetadata, err := getEvenPodsSpreadMetadata(pod, allNodes)
if err != nil {
klog.Errorf("Error calculating spreadConstraintsMap: %v", err)
return nil
}
predicateMetadata := &predicateMetadata{
pod: pod,
evenPodsSpreadMetadata: evenPodsSpreadMetadata,
podFitsResourcesMetadata: getPodFitsResourcesMetedata(pod),
}
for predicateName, precomputeFunc := range predicateMetadataProducers {
@ -414,7 +391,8 @@ func GetPodAffinityMetadata(pod *v1.Pod, allNodes []*schedulernodeinfo.NodeInfo,
}, nil
}
func getEvenPodsSpreadMetadata(pod *v1.Pod, allNodes []*schedulernodeinfo.NodeInfo) (*evenPodsSpreadMetadata, error) {
// GetPodTopologySpreadMetadata computes pod topology spread metadata.
func GetPodTopologySpreadMetadata(pod *v1.Pod, allNodes []*schedulernodeinfo.NodeInfo) (*PodTopologySpreadMetadata, error) {
// We have feature gating in APIServer to strip the spec
// so don't need to re-check feature gate, just check length of constraints.
constraints, err := filterHardTopologySpreadConstraints(pod.Spec.TopologySpreadConstraints)
@ -422,14 +400,14 @@ func getEvenPodsSpreadMetadata(pod *v1.Pod, allNodes []*schedulernodeinfo.NodeIn
return nil, err
}
if len(constraints) == 0 {
return nil, nil
return &PodTopologySpreadMetadata{}, nil
}
var lock sync.Mutex
// TODO(Huang-Wei): It might be possible to use "make(map[topologyPair]*int32)".
// In that case, need to consider how to init each tpPairToCount[pair] in an atomic fashion.
m := evenPodsSpreadMetadata{
m := PodTopologySpreadMetadata{
constraints: constraints,
tpKeyToCriticalPaths: make(map[string]*criticalPaths, len(constraints)),
tpPairToMatchNum: make(map[topologyPair]int32),
@ -526,15 +504,17 @@ func (m topologyToMatchedTermCount) clone() topologyToMatchedTermCount {
return copy
}
func (m *evenPodsSpreadMetadata) addPod(addedPod, preemptorPod *v1.Pod, node *v1.Node) {
m.updatePod(addedPod, preemptorPod, node, 1)
// AddPod updates the metadata with addedPod.
func (m *PodTopologySpreadMetadata) AddPod(addedPod, preemptorPod *v1.Pod, node *v1.Node) {
m.updateWithPod(addedPod, preemptorPod, node, 1)
}
func (m *evenPodsSpreadMetadata) removePod(deletedPod, preemptorPod *v1.Pod, node *v1.Node) {
m.updatePod(deletedPod, preemptorPod, node, -1)
// RemovePod updates the metadata with deletedPod.
func (m *PodTopologySpreadMetadata) RemovePod(deletedPod, preemptorPod *v1.Pod, node *v1.Node) {
m.updateWithPod(deletedPod, preemptorPod, node, -1)
}
func (m *evenPodsSpreadMetadata) updatePod(updatedPod, preemptorPod *v1.Pod, node *v1.Node, delta int32) {
func (m *PodTopologySpreadMetadata) updateWithPod(updatedPod, preemptorPod *v1.Pod, node *v1.Node, delta int32) {
if m == nil || updatedPod.Namespace != preemptorPod.Namespace || node == nil {
return
}
@ -556,12 +536,13 @@ func (m *evenPodsSpreadMetadata) updatePod(updatedPod, preemptorPod *v1.Pod, nod
}
}
func (m *evenPodsSpreadMetadata) clone() *evenPodsSpreadMetadata {
// c could be nil when EvenPodsSpread feature is disabled
// Clone makes a deep copy of PodTopologySpreadMetadata.
func (m *PodTopologySpreadMetadata) Clone() *PodTopologySpreadMetadata {
// m could be nil when EvenPodsSpread feature is disabled
if m == nil {
return nil
}
cp := evenPodsSpreadMetadata{
cp := PodTopologySpreadMetadata{
// constraints are shared because they don't change.
constraints: m.constraints,
tpKeyToCriticalPaths: make(map[string]*criticalPaths, len(m.tpKeyToCriticalPaths)),
@ -584,7 +565,6 @@ func (meta *predicateMetadata) RemovePod(deletedPod *v1.Pod, node *v1.Node) erro
if deletedPodFullName == schedutil.GetPodFullName(meta.pod) {
return fmt.Errorf("deletedPod and meta.pod must not be the same")
}
meta.evenPodsSpreadMetadata.removePod(deletedPod, meta.pod, node)
meta.serviceAffinityMetadata.removePod(deletedPod, node)
return nil
@ -601,10 +581,6 @@ func (meta *predicateMetadata) AddPod(addedPod *v1.Pod, node *v1.Node) error {
return fmt.Errorf("node not found")
}
// Update meta.evenPodsSpreadMetadata if meta.pod has hard spread constraints
// and addedPod matches that
meta.evenPodsSpreadMetadata.addPod(addedPod, meta.pod, node)
meta.serviceAffinityMetadata.addPod(addedPod, meta.pod, node)
return nil
@ -616,7 +592,6 @@ func (meta *predicateMetadata) ShallowCopy() Metadata {
newPredMeta := &predicateMetadata{
pod: meta.pod,
}
newPredMeta.evenPodsSpreadMetadata = meta.evenPodsSpreadMetadata.clone()
newPredMeta.serviceAffinityMetadata = meta.serviceAffinityMetadata.clone()
newPredMeta.podFitsResourcesMetadata = meta.podFitsResourcesMetadata.clone()
return (Metadata)(newPredMeta)

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

@ -235,15 +235,6 @@ func TestPredicateMetadata_ShallowCopy(t *testing.T) {
AllowedPodNumber: 4,
},
},
evenPodsSpreadMetadata: &evenPodsSpreadMetadata{
tpKeyToCriticalPaths: map[string]*criticalPaths{
"name": {{"nodeA", 1}, {"nodeC", 2}},
},
tpPairToMatchNum: map[topologyPair]int32{
{key: "name", value: "nodeA"}: 1,
{key: "name", value: "nodeC"}: 2,
},
},
serviceAffinityMetadata: &serviceAffinityMetadata{
matchingPodList: []*v1.Pod{
{ObjectMeta: metav1.ObjectMeta{Name: "pod1"}},
@ -485,7 +476,7 @@ func TestGetTPMapMatchingSpreadConstraints(t *testing.T) {
pod *v1.Pod
nodes []*v1.Node
existingPods []*v1.Pod
want *evenPodsSpreadMetadata
want *PodTopologySpreadMetadata
}{
{
name: "clean cluster with one spreadConstraint",
@ -498,7 +489,7 @@ func TestGetTPMapMatchingSpreadConstraints(t *testing.T) {
st.MakeNode().Name("node-x").Label("zone", "zone2").Label("node", "node-x").Obj(),
st.MakeNode().Name("node-y").Label("zone", "zone2").Label("node", "node-y").Obj(),
},
want: &evenPodsSpreadMetadata{
want: &PodTopologySpreadMetadata{
constraints: []topologySpreadConstraint{
{
maxSkew: 5,
@ -533,7 +524,7 @@ func TestGetTPMapMatchingSpreadConstraints(t *testing.T) {
st.MakePod().Name("p-y1").Node("node-y").Label("foo", "").Obj(),
st.MakePod().Name("p-y2").Node("node-y").Label("foo", "").Obj(),
},
want: &evenPodsSpreadMetadata{
want: &PodTopologySpreadMetadata{
constraints: []topologySpreadConstraint{
{
maxSkew: 1,
@ -570,7 +561,7 @@ func TestGetTPMapMatchingSpreadConstraints(t *testing.T) {
st.MakePod().Name("p-y1").Node("node-y").Label("foo", "").Obj(),
st.MakePod().Name("p-y2").Node("node-y").Label("foo", "").Obj(),
},
want: &evenPodsSpreadMetadata{
want: &PodTopologySpreadMetadata{
constraints: []topologySpreadConstraint{
{
maxSkew: 1,
@ -606,7 +597,7 @@ func TestGetTPMapMatchingSpreadConstraints(t *testing.T) {
st.MakePod().Name("p-y1").Namespace("ns2").Node("node-y").Label("foo", "").Obj(),
st.MakePod().Name("p-y2").Node("node-y").Label("foo", "").Obj(),
},
want: &evenPodsSpreadMetadata{
want: &PodTopologySpreadMetadata{
constraints: []topologySpreadConstraint{
{
maxSkew: 1,
@ -644,7 +635,7 @@ func TestGetTPMapMatchingSpreadConstraints(t *testing.T) {
st.MakePod().Name("p-y3").Node("node-y").Label("foo", "").Obj(),
st.MakePod().Name("p-y4").Node("node-y").Label("foo", "").Obj(),
},
want: &evenPodsSpreadMetadata{
want: &PodTopologySpreadMetadata{
constraints: []topologySpreadConstraint{
{
maxSkew: 1,
@ -693,7 +684,7 @@ func TestGetTPMapMatchingSpreadConstraints(t *testing.T) {
st.MakePod().Name("p-y3").Node("node-y").Label("foo", "").Obj(),
st.MakePod().Name("p-y4").Node("node-y").Label("foo", "").Obj(),
},
want: &evenPodsSpreadMetadata{
want: &PodTopologySpreadMetadata{
constraints: []topologySpreadConstraint{
{
maxSkew: 1,
@ -734,7 +725,7 @@ func TestGetTPMapMatchingSpreadConstraints(t *testing.T) {
st.MakePod().Name("p-a").Node("node-a").Label("foo", "").Obj(),
st.MakePod().Name("p-b").Node("node-b").Label("bar", "").Obj(),
},
want: &evenPodsSpreadMetadata{
want: &PodTopologySpreadMetadata{
constraints: []topologySpreadConstraint{
{
maxSkew: 1,
@ -780,7 +771,7 @@ func TestGetTPMapMatchingSpreadConstraints(t *testing.T) {
st.MakePod().Name("p-y3").Node("node-y").Label("foo", "").Obj(),
st.MakePod().Name("p-y4").Node("node-y").Label("foo", "").Label("bar", "").Obj(),
},
want: &evenPodsSpreadMetadata{
want: &PodTopologySpreadMetadata{
constraints: []topologySpreadConstraint{
{
maxSkew: 1,
@ -828,7 +819,7 @@ func TestGetTPMapMatchingSpreadConstraints(t *testing.T) {
st.MakePod().Name("p-y3").Node("node-y").Label("foo", "").Obj(),
st.MakePod().Name("p-y4").Node("node-y").Label("foo", "").Obj(),
},
want: &evenPodsSpreadMetadata{
want: &PodTopologySpreadMetadata{
constraints: []topologySpreadConstraint{
{
maxSkew: 1,
@ -859,7 +850,7 @@ func TestGetTPMapMatchingSpreadConstraints(t *testing.T) {
t.Run(tt.name, func(t *testing.T) {
s := nodeinfosnapshot.NewSnapshot(nodeinfosnapshot.CreateNodeInfoMap(tt.existingPods, tt.nodes))
l, _ := s.NodeInfos().List()
got, _ := getEvenPodsSpreadMetadata(tt.pod, l)
got, _ := GetPodTopologySpreadMetadata(tt.pod, l)
got.sortCriticalPaths()
if !reflect.DeepEqual(got, tt.want) {
t.Errorf("getEvenPodsSpreadMetadata() = %#v, want %#v", *got, *tt.want)
@ -883,7 +874,7 @@ func TestPodSpreadCache_addPod(t *testing.T) {
existingPods []*v1.Pod
nodeIdx int // denotes which node 'addedPod' belongs to
nodes []*v1.Node
want *evenPodsSpreadMetadata
want *PodTopologySpreadMetadata
}{
{
name: "node a and b both impact current min match",
@ -897,7 +888,7 @@ func TestPodSpreadCache_addPod(t *testing.T) {
st.MakeNode().Name("node-a").Label("zone", "zone1").Label("node", "node-a").Obj(),
st.MakeNode().Name("node-b").Label("zone", "zone1").Label("node", "node-b").Obj(),
},
want: &evenPodsSpreadMetadata{
want: &PodTopologySpreadMetadata{
constraints: []topologySpreadConstraint{nodeConstraint},
tpKeyToCriticalPaths: map[string]*criticalPaths{
"node": {{"node-b", 0}, {"node-a", 1}},
@ -922,7 +913,7 @@ func TestPodSpreadCache_addPod(t *testing.T) {
st.MakeNode().Name("node-a").Label("zone", "zone1").Label("node", "node-a").Obj(),
st.MakeNode().Name("node-b").Label("zone", "zone1").Label("node", "node-b").Obj(),
},
want: &evenPodsSpreadMetadata{
want: &PodTopologySpreadMetadata{
constraints: []topologySpreadConstraint{nodeConstraint},
tpKeyToCriticalPaths: map[string]*criticalPaths{
"node": {{"node-a", 1}, {"node-b", 1}},
@ -947,7 +938,7 @@ func TestPodSpreadCache_addPod(t *testing.T) {
st.MakeNode().Name("node-a").Label("zone", "zone1").Label("node", "node-a").Obj(),
st.MakeNode().Name("node-b").Label("zone", "zone1").Label("node", "node-b").Obj(),
},
want: &evenPodsSpreadMetadata{
want: &PodTopologySpreadMetadata{
constraints: []topologySpreadConstraint{nodeConstraint},
tpKeyToCriticalPaths: map[string]*criticalPaths{
"node": {{"node-a", 0}, {"node-b", 1}},
@ -972,7 +963,7 @@ func TestPodSpreadCache_addPod(t *testing.T) {
st.MakeNode().Name("node-a").Label("zone", "zone1").Label("node", "node-a").Obj(),
st.MakeNode().Name("node-b").Label("zone", "zone1").Label("node", "node-b").Obj(),
},
want: &evenPodsSpreadMetadata{
want: &PodTopologySpreadMetadata{
constraints: []topologySpreadConstraint{nodeConstraint},
tpKeyToCriticalPaths: map[string]*criticalPaths{
"node": {{"node-a", 0}, {"node-b", 2}},
@ -996,7 +987,7 @@ func TestPodSpreadCache_addPod(t *testing.T) {
st.MakeNode().Name("node-a").Label("zone", "zone1").Label("node", "node-a").Obj(),
st.MakeNode().Name("node-x").Label("zone", "zone2").Label("node", "node-x").Obj(),
},
want: &evenPodsSpreadMetadata{
want: &PodTopologySpreadMetadata{
constraints: []topologySpreadConstraint{zoneConstraint, nodeConstraint},
tpKeyToCriticalPaths: map[string]*criticalPaths{
"zone": {{"zone2", 0}, {"zone1", 1}},
@ -1025,7 +1016,7 @@ func TestPodSpreadCache_addPod(t *testing.T) {
st.MakeNode().Name("node-a").Label("zone", "zone1").Label("node", "node-a").Obj(),
st.MakeNode().Name("node-x").Label("zone", "zone2").Label("node", "node-x").Obj(),
},
want: &evenPodsSpreadMetadata{
want: &PodTopologySpreadMetadata{
constraints: []topologySpreadConstraint{zoneConstraint, nodeConstraint},
tpKeyToCriticalPaths: map[string]*criticalPaths{
"zone": {{"zone1", 1}, {"zone2", 1}},
@ -1057,7 +1048,7 @@ func TestPodSpreadCache_addPod(t *testing.T) {
st.MakeNode().Name("node-b").Label("zone", "zone1").Label("node", "node-b").Obj(),
st.MakeNode().Name("node-x").Label("zone", "zone2").Label("node", "node-x").Obj(),
},
want: &evenPodsSpreadMetadata{
want: &PodTopologySpreadMetadata{
constraints: []topologySpreadConstraint{zoneConstraint, nodeConstraint},
tpKeyToCriticalPaths: map[string]*criticalPaths{
"zone": {{"zone2", 1}, {"zone1", 3}},
@ -1090,7 +1081,7 @@ func TestPodSpreadCache_addPod(t *testing.T) {
st.MakeNode().Name("node-b").Label("zone", "zone1").Label("node", "node-b").Obj(),
st.MakeNode().Name("node-x").Label("zone", "zone2").Label("node", "node-x").Obj(),
},
want: &evenPodsSpreadMetadata{
want: &PodTopologySpreadMetadata{
constraints: []topologySpreadConstraint{
zoneConstraint,
{
@ -1130,7 +1121,7 @@ func TestPodSpreadCache_addPod(t *testing.T) {
st.MakeNode().Name("node-b").Label("zone", "zone1").Label("node", "node-b").Obj(),
st.MakeNode().Name("node-x").Label("zone", "zone2").Label("node", "node-x").Obj(),
},
want: &evenPodsSpreadMetadata{
want: &PodTopologySpreadMetadata{
constraints: []topologySpreadConstraint{
zoneConstraint,
{
@ -1157,11 +1148,11 @@ func TestPodSpreadCache_addPod(t *testing.T) {
t.Run(tt.name, func(t *testing.T) {
s := nodeinfosnapshot.NewSnapshot(nodeinfosnapshot.CreateNodeInfoMap(tt.existingPods, tt.nodes))
l, _ := s.NodeInfos().List()
evenPodsSpreadMetadata, _ := getEvenPodsSpreadMetadata(tt.preemptor, l)
evenPodsSpreadMetadata.addPod(tt.addedPod, tt.preemptor, tt.nodes[tt.nodeIdx])
evenPodsSpreadMetadata.sortCriticalPaths()
if !reflect.DeepEqual(evenPodsSpreadMetadata, tt.want) {
t.Errorf("evenPodsSpreadMetadata#addPod() = %v, want %v", evenPodsSpreadMetadata, tt.want)
podTopologySpreadMeta, _ := GetPodTopologySpreadMetadata(tt.preemptor, l)
podTopologySpreadMeta.AddPod(tt.addedPod, tt.preemptor, tt.nodes[tt.nodeIdx])
podTopologySpreadMeta.sortCriticalPaths()
if !reflect.DeepEqual(podTopologySpreadMeta, tt.want) {
t.Errorf("podTopologySpreadMeta#addPod() = %v, want %v", podTopologySpreadMeta, tt.want)
}
})
}
@ -1183,7 +1174,7 @@ func TestPodSpreadCache_removePod(t *testing.T) {
deletedPodIdx int // need to reuse *Pod of existingPods[i]
deletedPod *v1.Pod // this field is used only when deletedPodIdx is -1
nodeIdx int // denotes which node "deletedPod" belongs to
want *evenPodsSpreadMetadata
want *PodTopologySpreadMetadata
}{
{
// A high priority pod may not be scheduled due to node taints or resource shortage.
@ -1204,7 +1195,7 @@ func TestPodSpreadCache_removePod(t *testing.T) {
},
deletedPodIdx: 0, // remove pod "p-a1"
nodeIdx: 0, // node-a
want: &evenPodsSpreadMetadata{
want: &PodTopologySpreadMetadata{
constraints: []topologySpreadConstraint{zoneConstraint},
tpKeyToCriticalPaths: map[string]*criticalPaths{
"zone": {{"zone1", 1}, {"zone2", 1}},
@ -1234,7 +1225,7 @@ func TestPodSpreadCache_removePod(t *testing.T) {
},
deletedPodIdx: 0, // remove pod "p-a1"
nodeIdx: 0, // node-a
want: &evenPodsSpreadMetadata{
want: &PodTopologySpreadMetadata{
constraints: []topologySpreadConstraint{zoneConstraint},
tpKeyToCriticalPaths: map[string]*criticalPaths{
"zone": {{"zone1", 1}, {"zone2", 2}},
@ -1265,7 +1256,7 @@ func TestPodSpreadCache_removePod(t *testing.T) {
},
deletedPodIdx: 0, // remove pod "p-a0"
nodeIdx: 0, // node-a
want: &evenPodsSpreadMetadata{
want: &PodTopologySpreadMetadata{
constraints: []topologySpreadConstraint{zoneConstraint},
tpKeyToCriticalPaths: map[string]*criticalPaths{
"zone": {{"zone1", 2}, {"zone2", 2}},
@ -1296,7 +1287,7 @@ func TestPodSpreadCache_removePod(t *testing.T) {
deletedPodIdx: -1,
deletedPod: st.MakePod().Name("p-a0").Node("node-a").Label("bar", "").Obj(),
nodeIdx: 0, // node-a
want: &evenPodsSpreadMetadata{
want: &PodTopologySpreadMetadata{
constraints: []topologySpreadConstraint{zoneConstraint},
tpKeyToCriticalPaths: map[string]*criticalPaths{
"zone": {{"zone1", 2}, {"zone2", 2}},
@ -1327,7 +1318,7 @@ func TestPodSpreadCache_removePod(t *testing.T) {
},
deletedPodIdx: 3, // remove pod "p-x1"
nodeIdx: 2, // node-x
want: &evenPodsSpreadMetadata{
want: &PodTopologySpreadMetadata{
constraints: []topologySpreadConstraint{zoneConstraint, nodeConstraint},
tpKeyToCriticalPaths: map[string]*criticalPaths{
"zone": {{"zone2", 1}, {"zone1", 3}},
@ -1347,7 +1338,7 @@ func TestPodSpreadCache_removePod(t *testing.T) {
t.Run(tt.name, func(t *testing.T) {
s := nodeinfosnapshot.NewSnapshot(nodeinfosnapshot.CreateNodeInfoMap(tt.existingPods, tt.nodes))
l, _ := s.NodeInfos().List()
evenPodsSpreadMetadata, _ := getEvenPodsSpreadMetadata(tt.preemptor, l)
podTopologySpreadMeta, _ := GetPodTopologySpreadMetadata(tt.preemptor, l)
var deletedPod *v1.Pod
if tt.deletedPodIdx < len(tt.existingPods) && tt.deletedPodIdx >= 0 {
@ -1355,10 +1346,10 @@ func TestPodSpreadCache_removePod(t *testing.T) {
} else {
deletedPod = tt.deletedPod
}
evenPodsSpreadMetadata.removePod(deletedPod, tt.preemptor, tt.nodes[tt.nodeIdx])
evenPodsSpreadMetadata.sortCriticalPaths()
if !reflect.DeepEqual(evenPodsSpreadMetadata, tt.want) {
t.Errorf("evenPodsSpreadMetadata#removePod() = %v, want %v", evenPodsSpreadMetadata, tt.want)
podTopologySpreadMeta.RemovePod(deletedPod, tt.preemptor, tt.nodes[tt.nodeIdx])
podTopologySpreadMeta.sortCriticalPaths()
if !reflect.DeepEqual(podTopologySpreadMeta, tt.want) {
t.Errorf("podTopologySpreadMeta#removePod() = %v, want %v", podTopologySpreadMeta, tt.want)
}
})
}
@ -1408,7 +1399,7 @@ func BenchmarkTestGetTPMapMatchingSpreadConstraints(b *testing.B) {
l, _ := s.NodeInfos().List()
b.ResetTimer()
for i := 0; i < b.N; i++ {
getEvenPodsSpreadMetadata(tt.pod, l)
GetPodTopologySpreadMetadata(tt.pod, l)
}
})
}
@ -1420,7 +1411,7 @@ var (
)
// sortCriticalPaths is only served for testing purpose.
func (m *evenPodsSpreadMetadata) sortCriticalPaths() {
func (m *PodTopologySpreadMetadata) sortCriticalPaths() {
for _, paths := range m.tpKeyToCriticalPaths {
// If two paths both hold minimum matching number, and topologyValue is unordered.
if paths[0].matchNum == paths[1].matchNum && paths[0].topologyValue > paths[1].topologyValue {

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

@ -1647,28 +1647,33 @@ func (c *VolumeBindingChecker) predicate(pod *v1.Pod, meta Metadata, nodeInfo *s
return true, nil, nil
}
// EvenPodsSpreadPredicate checks if a pod can be scheduled on a node which satisfies
// its topologySpreadConstraints.
// EvenPodsSpreadPredicate is the legacy function using old path of metadata.
// DEPRECATED
func EvenPodsSpreadPredicate(pod *v1.Pod, meta Metadata, nodeInfo *schedulernodeinfo.NodeInfo) (bool, []PredicateFailureReason, error) {
return false, nil, fmt.Errorf("this function should never be called")
}
// PodTopologySpreadPredicate checks if a pod can be scheduled on a node which satisfies
// its topologySpreadConstraints.
func PodTopologySpreadPredicate(pod *v1.Pod, meta *PodTopologySpreadMetadata, nodeInfo *schedulernodeinfo.NodeInfo) (bool, []PredicateFailureReason, error) {
node := nodeInfo.Node()
if node == nil {
return false, nil, fmt.Errorf("node not found")
}
var epsMeta *evenPodsSpreadMetadata
if predicateMeta, ok := meta.(*predicateMetadata); ok {
epsMeta = predicateMeta.evenPodsSpreadMetadata
} else { // We don't have precomputed metadata. We have to follow a slow path to check spread constraints.
// TODO(autoscaler): get it implemented
return false, nil, errors.New("metadata not pre-computed for EvenPodsSpreadPredicate")
// nil meta is illegal.
if meta == nil {
// TODO(autoscaler): get it implemented.
return false, nil, errors.New("metadata not pre-computed for PodTopologySpreadPredicate")
}
if epsMeta == nil || len(epsMeta.tpPairToMatchNum) == 0 || len(epsMeta.constraints) == 0 {
// However, "empty" meta is legit which tolerates every toSchedule Pod.
if len(meta.tpPairToMatchNum) == 0 || len(meta.constraints) == 0 {
return true, nil, nil
}
podLabelSet := labels.Set(pod.Labels)
for _, c := range epsMeta.constraints {
for _, c := range meta.constraints {
tpKey := c.topologyKey
tpVal, ok := node.Labels[c.topologyKey]
if !ok {
@ -1682,16 +1687,16 @@ func EvenPodsSpreadPredicate(pod *v1.Pod, meta Metadata, nodeInfo *schedulernode
}
pair := topologyPair{key: tpKey, value: tpVal}
paths, ok := epsMeta.tpKeyToCriticalPaths[tpKey]
paths, ok := meta.tpKeyToCriticalPaths[tpKey]
if !ok {
// error which should not happen
klog.Errorf("internal error: get paths from key %q of %#v", tpKey, epsMeta.tpKeyToCriticalPaths)
klog.Errorf("internal error: get paths from key %q of %#v", tpKey, meta.tpKeyToCriticalPaths)
continue
}
// judging criteria:
// 'existing matching num' + 'if self-match (1 or 0)' - 'global min matching num' <= 'maxSkew'
minMatchNum := paths[0].matchNum
matchNum := epsMeta.tpPairToMatchNum[pair]
matchNum := meta.tpPairToMatchNum[pair]
skew := matchNum + selfMatchNum - minMatchNum
if skew > c.maxSkew {
klog.V(5).Infof("node '%s' failed spreadConstraint[%s]: matchNum(%d) + selfMatchNum(%d) - minMatchNum(%d) > maxSkew(%d)", node.Name, tpKey, matchNum, selfMatchNum, minMatchNum, c.maxSkew)

445
pkg/scheduler/algorithm/predicates/predicates_test.go
View File

@ -37,7 +37,6 @@ import (
fakelisters "k8s.io/kubernetes/pkg/scheduler/listers/fake"
schedulernodeinfo "k8s.io/kubernetes/pkg/scheduler/nodeinfo"
nodeinfosnapshot "k8s.io/kubernetes/pkg/scheduler/nodeinfo/snapshot"
st "k8s.io/kubernetes/pkg/scheduler/testing"
)
var (
@ -2658,447 +2657,3 @@ func TestCheckNodeUnschedulablePredicate(t *testing.T) {
}
}
}
func TestEvenPodsSpreadPredicate_SingleConstraint(t *testing.T) {
tests := []struct {
name string
pod *v1.Pod
nodes []*v1.Node
existingPods []*v1.Pod
fits map[string]bool
}{
{
name: "no existing pods",
pod: st.MakePod().Name("p").Label("foo", "").SpreadConstraint(
1, "zone", hardSpread, st.MakeLabelSelector().Exists("foo").Obj(),
).Obj(),
nodes: []*v1.Node{
st.MakeNode().Name("node-a").Label("zone", "zone1").Label("node", "node-a").Obj(),
st.MakeNode().Name("node-b").Label("zone", "zone1").Label("node", "node-b").Obj(),
st.MakeNode().Name("node-x").Label("zone", "zone2").Label("node", "node-x").Obj(),
st.MakeNode().Name("node-y").Label("zone", "zone2").Label("node", "node-y").Obj(),
},
fits: map[string]bool{
"node-a": true,
"node-b": true,
"node-x": true,
"node-y": true,
},
},
{
name: "no existing pods, incoming pod doesn't match itself",
pod: st.MakePod().Name("p").Label("foo", "").SpreadConstraint(
1, "zone", hardSpread, st.MakeLabelSelector().Exists("bar").Obj(),
).Obj(),
nodes: []*v1.Node{
st.MakeNode().Name("node-a").Label("zone", "zone1").Label("node", "node-a").Obj(),
st.MakeNode().Name("node-b").Label("zone", "zone1").Label("node", "node-b").Obj(),
st.MakeNode().Name("node-x").Label("zone", "zone2").Label("node", "node-x").Obj(),
st.MakeNode().Name("node-y").Label("zone", "zone2").Label("node", "node-y").Obj(),
},
fits: map[string]bool{
"node-a": true,
"node-b": true,
"node-x": true,
"node-y": true,
},
},
{
name: "existing pods with mis-matched namespace doesn't count",
pod: st.MakePod().Name("p").Label("foo", "").SpreadConstraint(
1, "zone", hardSpread, st.MakeLabelSelector().Exists("foo").Obj(),
).Obj(),
nodes: []*v1.Node{
st.MakeNode().Name("node-a").Label("zone", "zone1").Label("node", "node-a").Obj(),
st.MakeNode().Name("node-b").Label("zone", "zone1").Label("node", "node-b").Obj(),
st.MakeNode().Name("node-x").Label("zone", "zone2").Label("node", "node-x").Obj(),
st.MakeNode().Name("node-y").Label("zone", "zone2").Label("node", "node-y").Obj(),
},
existingPods: []*v1.Pod{
st.MakePod().Name("p-a1").Namespace("ns1").Node("node-a").Label("foo", "").Obj(),
st.MakePod().Name("p-b1").Namespace("ns2").Node("node-a").Label("foo", "").Obj(),
st.MakePod().Name("p-x1").Node("node-x").Label("foo", "").Obj(),
st.MakePod().Name("p-y1").Node("node-y").Label("foo", "").Obj(),
},
fits: map[string]bool{
"node-a": true,
"node-b": true,
"node-x": false,
"node-y": false,
},
},
{
name: "pods spread across zones as 3/3, all nodes fit",
pod: st.MakePod().Name("p").Label("foo", "").SpreadConstraint(
1, "zone", hardSpread, st.MakeLabelSelector().Exists("foo").Obj(),
).Obj(),
nodes: []*v1.Node{
st.MakeNode().Name("node-a").Label("zone", "zone1").Label("node", "node-a").Obj(),
st.MakeNode().Name("node-b").Label("zone", "zone1").Label("node", "node-b").Obj(),
st.MakeNode().Name("node-x").Label("zone", "zone2").Label("node", "node-x").Obj(),
st.MakeNode().Name("node-y").Label("zone", "zone2").Label("node", "node-y").Obj(),
},
existingPods: []*v1.Pod{
st.MakePod().Name("p-a1").Node("node-a").Label("foo", "").Obj(),
st.MakePod().Name("p-a2").Node("node-a").Label("foo", "").Obj(),
st.MakePod().Name("p-b1").Node("node-b").Label("foo", "").Obj(),
st.MakePod().Name("p-y1").Node("node-y").Label("foo", "").Obj(),
st.MakePod().Name("p-y2").Node("node-y").Label("foo", "").Obj(),
st.MakePod().Name("p-y3").Node("node-y").Label("foo", "").Obj(),
},
fits: map[string]bool{
"node-a": true,
"node-b": true,
"node-x": true,
"node-y": true,
},
},
{
// TODO(Huang-Wei): maybe document this to remind users that typos on node labels
// can cause unexpected behavior
name: "pods spread across zones as 1/2 due to absence of label 'zone' on node-b",
pod: st.MakePod().Name("p").Label("foo", "").SpreadConstraint(
1, "zone", hardSpread, st.MakeLabelSelector().Exists("foo").Obj(),
).Obj(),
nodes: []*v1.Node{
st.MakeNode().Name("node-a").Label("zone", "zone1").Label("node", "node-a").Obj(),
st.MakeNode().Name("node-b").Label("zon", "zone1").Label("node", "node-b").Obj(),
st.MakeNode().Name("node-x").Label("zone", "zone2").Label("node", "node-x").Obj(),
st.MakeNode().Name("node-y").Label("zone", "zone2").Label("node", "node-y").Obj(),
},
existingPods: []*v1.Pod{
st.MakePod().Name("p-a1").Node("node-a").Label("foo", "").Obj(),
st.MakePod().Name("p-b1").Node("node-b").Label("foo", "").Obj(),
st.MakePod().Name("p-x1").Node("node-x").Label("foo", "").Obj(),
st.MakePod().Name("p-y1").Node("node-y").Label("foo", "").Obj(),
},
fits: map[string]bool{
"node-a": true,
"node-b": false,
"node-x": false,
"node-y": false,
},
},
{
name: "pods spread across nodes as 2/1/0/3, only node-x fits",
pod: st.MakePod().Name("p").Label("foo", "").SpreadConstraint(
1, "node", hardSpread, st.MakeLabelSelector().Exists("foo").Obj(),
).Obj(),
nodes: []*v1.Node{
st.MakeNode().Name("node-a").Label("zone", "zone1").Label("node", "node-a").Obj(),
st.MakeNode().Name("node-b").Label("zone", "zone1").Label("node", "node-b").Obj(),
st.MakeNode().Name("node-x").Label("zone", "zone2").Label("node", "node-x").Obj(),
st.MakeNode().Name("node-y").Label("zone", "zone2").Label("node", "node-y").Obj(),
},
existingPods: []*v1.Pod{
st.MakePod().Name("p-a1").Node("node-a").Label("foo", "").Obj(),
st.MakePod().Name("p-a2").Node("node-a").Label("foo", "").Obj(),
st.MakePod().Name("p-b1").Node("node-b").Label("foo", "").Obj(),
st.MakePod().Name("p-y1").Node("node-y").Label("foo", "").Obj(),
st.MakePod().Name("p-y2").Node("node-y").Label("foo", "").Obj(),
st.MakePod().Name("p-y3").Node("node-y").Label("foo", "").Obj(),
},
fits: map[string]bool{
"node-a": false,
"node-b": false,
"node-x": true,
"node-y": false,
},
},
{
name: "pods spread across nodes as 2/1/0/3, maxSkew is 2, node-b and node-x fit",
pod: st.MakePod().Name("p").Label("foo", "").SpreadConstraint(
2, "node", hardSpread, st.MakeLabelSelector().Exists("foo").Obj(),
).Obj(),
nodes: []*v1.Node{
st.MakeNode().Name("node-a").Label("zone", "zone1").Label("node", "node-a").Obj(),
st.MakeNode().Name("node-b").Label("zone", "zone1").Label("node", "node-b").Obj(),
st.MakeNode().Name("node-x").Label("zone", "zone2").Label("node", "node-x").Obj(),
st.MakeNode().Name("node-y").Label("zone", "zone2").Label("node", "node-y").Obj(),
},
existingPods: []*v1.Pod{
st.MakePod().Name("p-a1").Node("node-a").Label("foo", "").Obj(),
st.MakePod().Name("p-a2").Node("node-a").Label("foo", "").Obj(),
st.MakePod().Name("p-b1").Node("node-b").Label("foo", "").Obj(),
st.MakePod().Name("p-y1").Node("node-y").Label("foo", "").Obj(),
st.MakePod().Name("p-y2").Node("node-y").Label("foo", "").Obj(),
st.MakePod().Name("p-y3").Node("node-y").Label("foo", "").Obj(),
},
fits: map[string]bool{
"node-a": false,
"node-b": true,
"node-x": true,
"node-y": false,
},
},
{
// not a desired case, but it can happen
// TODO(Huang-Wei): document this "pod-not-match-itself" case
// in this case, placement of the new pod doesn't change pod distribution of the cluster
// as the incoming pod doesn't have label "foo"
name: "pods spread across nodes as 2/1/0/3, but pod doesn't match itself",
pod: st.MakePod().Name("p").Label("bar", "").SpreadConstraint(
1, "node", hardSpread, st.MakeLabelSelector().Exists("foo").Obj(),
).Obj(),
nodes: []*v1.Node{
st.MakeNode().Name("node-a").Label("zone", "zone1").Label("node", "node-a").Obj(),
st.MakeNode().Name("node-b").Label("zone", "zone1").Label("node", "node-b").Obj(),
st.MakeNode().Name("node-x").Label("zone", "zone2").Label("node", "node-x").Obj(),
st.MakeNode().Name("node-y").Label("zone", "zone2").Label("node", "node-y").Obj(),
},
existingPods: []*v1.Pod{
st.MakePod().Name("p-a1").Node("node-a").Label("foo", "").Obj(),
st.MakePod().Name("p-a2").Node("node-a").Label("foo", "").Obj(),
st.MakePod().Name("p-b1").Node("node-b").Label("foo", "").Obj(),
st.MakePod().Name("p-y1").Node("node-y").Label("foo", "").Obj(),
st.MakePod().Name("p-y2").Node("node-y").Label("foo", "").Obj(),
st.MakePod().Name("p-y3").Node("node-y").Label("foo", "").Obj(),
},
fits: map[string]bool{
"node-a": false,
"node-b": true,
"node-x": true,
"node-y": false,
},
},
{
// only node-a and node-y are considered, so pods spread as 2/~1~/~0~/3
// ps: '~num~' is a markdown symbol to denote a crossline through 'num'
// but in this unit test, we don't run NodeAffinityPredicate, so node-b and node-x are
// still expected to be fits;
// the fact that node-a fits can prove the underlying logic works
name: "incoming pod has nodeAffinity, pods spread as 2/~1~/~0~/3, hence node-a fits",
pod: st.MakePod().Name("p").Label("foo", "").
NodeAffinityIn("node", []string{"node-a", "node-y"}).
SpreadConstraint(1, "node", hardSpread, st.MakeLabelSelector().Exists("foo").Obj()).
Obj(),
nodes: []*v1.Node{
st.MakeNode().Name("node-a").Label("zone", "zone1").Label("node", "node-a").Obj(),
st.MakeNode().Name("node-b").Label("zone", "zone1").Label("node", "node-b").Obj(),
st.MakeNode().Name("node-x").Label("zone", "zone2").Label("node", "node-x").Obj(),
st.MakeNode().Name("node-y").Label("zone", "zone2").Label("node", "node-y").Obj(),
},
existingPods: []*v1.Pod{
st.MakePod().Name("p-a1").Node("node-a").Label("foo", "").Obj(),
st.MakePod().Name("p-a2").Node("node-a").Label("foo", "").Obj(),
st.MakePod().Name("p-b1").Node("node-b").Label("foo", "").Obj(),
st.MakePod().Name("p-y1").Node("node-y").Label("foo", "").Obj(),
st.MakePod().Name("p-y2").Node("node-y").Label("foo", "").Obj(),
st.MakePod().Name("p-y3").Node("node-y").Label("foo", "").Obj(),
},
fits: map[string]bool{
"node-a": true,
"node-b": true, // in real case, it's false
"node-x": true, // in real case, it's false
"node-y": false,
},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
s := nodeinfosnapshot.NewSnapshot(nodeinfosnapshot.CreateNodeInfoMap(tt.existingPods, tt.nodes))
factory := &MetadataProducerFactory{}
meta := factory.GetPredicateMetadata(tt.pod, s)
for _, node := range tt.nodes {
fits, _, _ := EvenPodsSpreadPredicate(tt.pod, meta, s.NodeInfoMap[node.Name])
if fits != tt.fits[node.Name] {
t.Errorf("[%s]: expected %v got %v", node.Name, tt.fits[node.Name], fits)
}
}
})
}
}
func TestEvenPodsSpreadPredicate_MultipleConstraints(t *testing.T) {
tests := []struct {
name string
pod *v1.Pod
nodes []*v1.Node
existingPods []*v1.Pod
fits map[string]bool
}{
{
// 1. to fulfil "zone" constraint, incoming pod can be placed on any zone (hence any node)
// 2. to fulfil "node" constraint, incoming pod can be placed on node-x
// intersection of (1) and (2) returns node-x
name: "two constraints on zone and node, spreads = [3/3, 2/1/0/3]",
pod: st.MakePod().Name("p").Label("foo", "").
SpreadConstraint(1, "zone", hardSpread, st.MakeLabelSelector().Exists("foo").Obj()).
SpreadConstraint(1, "node", hardSpread, st.MakeLabelSelector().Exists("foo").Obj()).
Obj(),
nodes: []*v1.Node{
st.MakeNode().Name("node-a").Label("zone", "zone1").Label("node", "node-a").Obj(),
st.MakeNode().Name("node-b").Label("zone", "zone1").Label("node", "node-b").Obj(),
st.MakeNode().Name("node-x").Label("zone", "zone2").Label("node", "node-x").Obj(),
st.MakeNode().Name("node-y").Label("zone", "zone2").Label("node", "node-y").Obj(),
},
existingPods: []*v1.Pod{
st.MakePod().Name("p-a1").Node("node-a").Label("foo", "").Obj(),
st.MakePod().Name("p-a2").Node("node-a").Label("foo", "").Obj(),
st.MakePod().Name("p-b1").Node("node-b").Label("foo", "").Obj(),
st.MakePod().Name("p-y1").Node("node-y").Label("foo", "").Obj(),
st.MakePod().Name("p-y2").Node("node-y").Label("foo", "").Obj(),
st.MakePod().Name("p-y3").Node("node-y").Label("foo", "").Obj(),
},
fits: map[string]bool{
"node-a": false,
"node-b": false,
"node-x": true,
"node-y": false,
},
},
{
// 1. to fulfil "zone" constraint, incoming pod can be placed on zone1 (node-a or node-b)
// 2. to fulfil "node" constraint, incoming pod can be placed on node-x
// intersection of (1) and (2) returns no node
name: "two constraints on zone and node, spreads = [3/4, 2/1/0/4]",
pod: st.MakePod().Name("p").Label("foo", "").
SpreadConstraint(1, "zone", hardSpread, st.MakeLabelSelector().Exists("foo").Obj()).
SpreadConstraint(1, "node", hardSpread, st.MakeLabelSelector().Exists("foo").Obj()).
Obj(),
nodes: []*v1.Node{
st.MakeNode().Name("node-a").Label("zone", "zone1").Label("node", "node-a").Obj(),
st.MakeNode().Name("node-b").Label("zone", "zone1").Label("node", "node-b").Obj(),
st.MakeNode().Name("node-x").Label("zone", "zone2").Label("node", "node-x").Obj(),
st.MakeNode().Name("node-y").Label("zone", "zone2").Label("node", "node-y").Obj(),
},
existingPods: []*v1.Pod{
st.MakePod().Name("p-a1").Node("node-a").Label("foo", "").Obj(),
st.MakePod().Name("p-a2").Node("node-a").Label("foo", "").Obj(),
st.MakePod().Name("p-b1").Node("node-b").Label("foo", "").Obj(),
st.MakePod().Name("p-y1").Node("node-y").Label("foo", "").Obj(),
st.MakePod().Name("p-y2").Node("node-y").Label("foo", "").Obj(),
st.MakePod().Name("p-y3").Node("node-y").Label("foo", "").Obj(),
st.MakePod().Name("p-y4").Node("node-y").Label("foo", "").Obj(),
},
fits: map[string]bool{
"node-a": false,
"node-b": false,
"node-x": false,
"node-y": false,
},
},
{
// 1. to fulfil "zone" constraint, incoming pod can be placed on zone2 (node-x or node-y)
// 2. to fulfil "node" constraint, incoming pod can be placed on node-b or node-x
// intersection of (1) and (2) returns node-x
name: "constraints hold different labelSelectors, spreads = [1/0, 1/0/0/1]",
pod: st.MakePod().Name("p").Label("foo", "").Label("bar", "").
SpreadConstraint(1, "zone", hardSpread, st.MakeLabelSelector().Exists("foo").Obj()).
SpreadConstraint(1, "node", hardSpread, st.MakeLabelSelector().Exists("bar").Obj()).
Obj(),
nodes: []*v1.Node{
st.MakeNode().Name("node-a").Label("zone", "zone1").Label("node", "node-a").Obj(),
st.MakeNode().Name("node-b").Label("zone", "zone1").Label("node", "node-b").Obj(),
st.MakeNode().Name("node-x").Label("zone", "zone2").Label("node", "node-x").Obj(),
st.MakeNode().Name("node-y").Label("zone", "zone2").Label("node", "node-y").Obj(),
},
existingPods: []*v1.Pod{
st.MakePod().Name("p-a1").Node("node-a").Label("foo", "").Obj(),
st.MakePod().Name("p-y1").Node("node-y").Label("bar", "").Obj(),
},
fits: map[string]bool{
"node-a": false,
"node-b": false,
"node-x": true,
"node-y": false,
},
},
{
// 1. to fulfil "zone" constraint, incoming pod can be placed on zone2 (node-x or node-y)
// 2. to fulfil "node" constraint, incoming pod can be placed on node-a or node-b
// intersection of (1) and (2) returns no node
name: "constraints hold different labelSelectors, spreads = [1/0, 0/0/1/1]",
pod: st.MakePod().Name("p").Label("foo", "").Label("bar", "").
SpreadConstraint(1, "zone", hardSpread, st.MakeLabelSelector().Exists("foo").Obj()).
SpreadConstraint(1, "node", hardSpread, st.MakeLabelSelector().Exists("bar").Obj()).
Obj(),
nodes: []*v1.Node{
st.MakeNode().Name("node-a").Label("zone", "zone1").Label("node", "node-a").Obj(),
st.MakeNode().Name("node-b").Label("zone", "zone1").Label("node", "node-b").Obj(),
st.MakeNode().Name("node-x").Label("zone", "zone2").Label("node", "node-x").Obj(),
st.MakeNode().Name("node-y").Label("zone", "zone2").Label("node", "node-y").Obj(),
},
existingPods: []*v1.Pod{
st.MakePod().Name("p-a1").Node("node-a").Label("foo", "").Obj(),
st.MakePod().Name("p-x1").Node("node-x").Label("bar", "").Obj(),
st.MakePod().Name("p-y1").Node("node-y").Label("bar", "").Obj(),
},
fits: map[string]bool{
"node-a": false,
"node-b": false,
"node-x": false,
"node-y": false,
},
},
{
// 1. to fulfil "zone" constraint, incoming pod can be placed on zone1 (node-a or node-b)
// 2. to fulfil "node" constraint, incoming pod can be placed on node-b or node-x
// intersection of (1) and (2) returns node-b
name: "constraints hold different labelSelectors, spreads = [2/3, 1/0/0/1]",
pod: st.MakePod().Name("p").Label("foo", "").Label("bar", "").
SpreadConstraint(1, "zone", hardSpread, st.MakeLabelSelector().Exists("foo").Obj()).
SpreadConstraint(1, "node", hardSpread, st.MakeLabelSelector().Exists("bar").Obj()).
Obj(),
nodes: []*v1.Node{
st.MakeNode().Name("node-a").Label("zone", "zone1").Label("node", "node-a").Obj(),
st.MakeNode().Name("node-b").Label("zone", "zone1").Label("node", "node-b").Obj(),
st.MakeNode().Name("node-x").Label("zone", "zone2").Label("node", "node-x").Obj(),
st.MakeNode().Name("node-y").Label("zone", "zone2").Label("node", "node-y").Obj(),
},
existingPods: []*v1.Pod{
st.MakePod().Name("p-a1").Node("node-a").Label("foo", "").Obj(),
st.MakePod().Name("p-a2").Node("node-a").Label("foo", "").Label("bar", "").Obj(),
st.MakePod().Name("p-y1").Node("node-y").Label("foo", "").Obj(),
st.MakePod().Name("p-y2").Node("node-y").Label("foo", "").Label("bar", "").Obj(),
st.MakePod().Name("p-y3").Node("node-y").Label("foo", "").Obj(),
},
fits: map[string]bool{
"node-a": false,
"node-b": true,
"node-x": false,
"node-y": false,
},
},
{
// 1. pod doesn't match itself on "zone" constraint, so it can be put onto any zone
// 2. to fulfil "node" constraint, incoming pod can be placed on node-a or node-b
// intersection of (1) and (2) returns node-a and node-b
name: "constraints hold different labelSelectors but pod doesn't match itself on 'zone' constraint",
pod: st.MakePod().Name("p").Label("bar", "").
SpreadConstraint(1, "zone", hardSpread, st.MakeLabelSelector().Exists("foo").Obj()).
SpreadConstraint(1, "node", hardSpread, st.MakeLabelSelector().Exists("bar").Obj()).
Obj(),
nodes: []*v1.Node{
st.MakeNode().Name("node-a").Label("zone", "zone1").Label("node", "node-a").Obj(),
st.MakeNode().Name("node-b").Label("zone", "zone1").Label("node", "node-b").Obj(),
st.MakeNode().Name("node-x").Label("zone", "zone2").Label("node", "node-x").Obj(),
st.MakeNode().Name("node-y").Label("zone", "zone2").Label("node", "node-y").Obj(),
},
existingPods: []*v1.Pod{
st.MakePod().Name("p-a1").Node("node-a").Label("foo", "").Obj(),
st.MakePod().Name("p-x1").Node("node-x").Label("bar", "").Obj(),
st.MakePod().Name("p-y1").Node("node-y").Label("bar", "").Obj(),
},
fits: map[string]bool{
"node-a": true,
"node-b": true,
"node-x": false,
"node-y": false,
},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
s := nodeinfosnapshot.NewSnapshot(nodeinfosnapshot.CreateNodeInfoMap(tt.existingPods, tt.nodes))
factory := &MetadataProducerFactory{}
meta := factory.GetPredicateMetadata(tt.pod, s)
for _, node := range tt.nodes {
fits, _, _ := EvenPodsSpreadPredicate(tt.pod, meta, s.NodeInfoMap[node.Name])
if fits != tt.fits[node.Name] {
t.Errorf("[%s]: expected %v got %v", node.Name, tt.fits[node.Name], fits)
}
}
})
}
}

1
pkg/scheduler/algorithmprovider/defaults/defaults.go
View File

@ -58,6 +58,7 @@ func defaultPredicates() sets.String {
// The returned function is used to restore the state of registered predicates/priorities
// when this function is called, and should be called in tests which may modify the value
// of a feature gate temporarily.
// TODO(Huang-Wei): refactor this function to have a clean way to disable/enable plugins.
func ApplyFeatureGates() (restore func()) {
snapshot := scheduler.RegisteredPredicatesAndPrioritiesSnapshot()

3
pkg/scheduler/apis/config/testing/compatibility_test.go
View File

@ -1191,6 +1191,9 @@ func TestCompatibility_v1_Scheduler(t *testing.T) {
features.EvenPodsSpread: true,
},
wantPlugins: map[string][]config.Plugin{
"PreFilterPlugin": {
{Name: "PodTopologySpread"},
},
"FilterPlugin": {
{Name: "NodeUnschedulable"},
{Name: "TaintToleration"},

58
pkg/scheduler/core/extender_test.go
View File

@ -361,16 +361,17 @@ var _ algorithm.SchedulerExtender = &FakeExtender{}
func TestGenericSchedulerWithExtenders(t *testing.T) {
tests := []struct {
name string
registerFilterPlugin st.RegisterFilterPluginFunc
registerScorePlugin st.RegisterScorePluginFunc
extenders []FakeExtender
nodes []string
expectedResult ScheduleResult
expectsErr bool
name string
registerPlugins []st.RegisterPluginFunc
extenders []FakeExtender
nodes []string
expectedResult ScheduleResult
expectsErr bool
}{
{
registerFilterPlugin: st.RegisterFilterPlugin("TrueFilter", NewTrueFilterPlugin),
registerPlugins: []st.RegisterPluginFunc{
st.RegisterFilterPlugin("TrueFilter", NewTrueFilterPlugin),
},
extenders: []FakeExtender{
{
predicates: []fitPredicate{truePredicateExtender},
@ -384,7 +385,9 @@ func TestGenericSchedulerWithExtenders(t *testing.T) {
name: "test 1",
},
{
registerFilterPlugin: st.RegisterFilterPlugin("TrueFilter", NewTrueFilterPlugin),
registerPlugins: []st.RegisterPluginFunc{
st.RegisterFilterPlugin("TrueFilter", NewTrueFilterPlugin),
},
extenders: []FakeExtender{
{
predicates: []fitPredicate{truePredicateExtender},
@ -398,7 +401,9 @@ func TestGenericSchedulerWithExtenders(t *testing.T) {
name: "test 2",
},
{
registerFilterPlugin: st.RegisterFilterPlugin("TrueFilter", NewTrueFilterPlugin),
registerPlugins: []st.RegisterPluginFunc{
st.RegisterFilterPlugin("TrueFilter", NewTrueFilterPlugin),
},
extenders: []FakeExtender{
{
predicates: []fitPredicate{truePredicateExtender},
@ -416,7 +421,9 @@ func TestGenericSchedulerWithExtenders(t *testing.T) {
name: "test 3",
},
{
registerFilterPlugin: st.RegisterFilterPlugin("TrueFilter", NewTrueFilterPlugin),
registerPlugins: []st.RegisterPluginFunc{
st.RegisterFilterPlugin("TrueFilter", NewTrueFilterPlugin),
},
extenders: []FakeExtender{
{
predicates: []fitPredicate{machine2PredicateExtender},
@ -430,7 +437,9 @@ func TestGenericSchedulerWithExtenders(t *testing.T) {
name: "test 4",
},
{
registerFilterPlugin: st.RegisterFilterPlugin("TrueFilter", NewTrueFilterPlugin),
registerPlugins: []st.RegisterPluginFunc{
st.RegisterFilterPlugin("TrueFilter", NewTrueFilterPlugin),
},
extenders: []FakeExtender{
{
predicates: []fitPredicate{truePredicateExtender},
@ -447,7 +456,9 @@ func TestGenericSchedulerWithExtenders(t *testing.T) {
name: "test 5",
},
{
registerFilterPlugin: st.RegisterFilterPlugin("TrueFilter", NewTrueFilterPlugin),
registerPlugins: []st.RegisterPluginFunc{
st.RegisterFilterPlugin("TrueFilter", NewTrueFilterPlugin),
},
extenders: []FakeExtender{
{
predicates: []fitPredicate{truePredicateExtender},
@ -469,8 +480,10 @@ func TestGenericSchedulerWithExtenders(t *testing.T) {
name: "test 6",
},
{
registerFilterPlugin: st.RegisterFilterPlugin("TrueFilter", NewTrueFilterPlugin),
registerScorePlugin: st.RegisterScorePlugin("Machine2Prioritizer", newMachine2PrioritizerPlugin(), 20),
registerPlugins: []st.RegisterPluginFunc{
st.RegisterFilterPlugin("TrueFilter", NewTrueFilterPlugin),
st.RegisterScorePlugin("Machine2Prioritizer", newMachine2PrioritizerPlugin(), 20),
},
extenders: []FakeExtender{
{
predicates: []fitPredicate{truePredicateExtender},
@ -494,8 +507,10 @@ func TestGenericSchedulerWithExtenders(t *testing.T) {
// If scheduler sends the pod by mistake, the test would fail
// because of the errors from errorPredicateExtender and/or
// errorPrioritizerExtender.
registerFilterPlugin: st.RegisterFilterPlugin("TrueFilter", NewTrueFilterPlugin),
registerScorePlugin: st.RegisterScorePlugin("Machine2Prioritizer", newMachine2PrioritizerPlugin(), 1),
registerPlugins: []st.RegisterPluginFunc{
st.RegisterFilterPlugin("TrueFilter", NewTrueFilterPlugin),
st.RegisterScorePlugin("Machine2Prioritizer", newMachine2PrioritizerPlugin(), 1),
},
extenders: []FakeExtender{
{
predicates: []fitPredicate{errorPredicateExtender},
@ -518,7 +533,9 @@ func TestGenericSchedulerWithExtenders(t *testing.T) {
//
// If scheduler did not ignore the extender, the test would fail
// because of the errors from errorPredicateExtender.
registerFilterPlugin: st.RegisterFilterPlugin("TrueFilter", NewTrueFilterPlugin),
registerPlugins: []st.RegisterPluginFunc{
st.RegisterFilterPlugin("TrueFilter", NewTrueFilterPlugin),
},
extenders: []FakeExtender{
{
predicates: []fitPredicate{errorPredicateExtender},
@ -560,9 +577,8 @@ func TestGenericSchedulerWithExtenders(t *testing.T) {
Score: &schedulerapi.PluginSet{},
}
var pluginConfigs []schedulerapi.PluginConfig
test.registerFilterPlugin(&registry, plugins, pluginConfigs)
if test.registerScorePlugin != nil {
test.registerScorePlugin(&registry, plugins, pluginConfigs)
for _, f := range test.registerPlugins {
f(&registry, plugins, pluginConfigs)
}
fwk, _ := framework.NewFramework(registry, plugins, pluginConfigs)

259
pkg/scheduler/core/generic_scheduler_test.go
View File

@ -371,19 +371,17 @@ func TestGenericScheduler(t *testing.T) {
tests := []struct {
name string
registerFilterPlugins []st.RegisterFilterPluginFunc
registerScorePlugins []st.RegisterScorePluginFunc
registerPlugins []st.RegisterPluginFunc
alwaysCheckAllPredicates bool
nodes []string
pvcs []v1.PersistentVolumeClaim
pod *v1.Pod
pods []*v1.Pod
buildPredMeta bool // build predicates metadata or not
expectedHosts sets.String
wErr error
}{
{
registerFilterPlugins: []st.RegisterFilterPluginFunc{
registerPlugins: []st.RegisterPluginFunc{
st.RegisterFilterPlugin("FalseFilter", NewFalseFilterPlugin),
},
nodes: []string{"machine1", "machine2"},
@ -400,7 +398,7 @@ func TestGenericScheduler(t *testing.T) {
},
},
{
registerFilterPlugins: []st.RegisterFilterPluginFunc{
registerPlugins: []st.RegisterPluginFunc{
st.RegisterFilterPlugin("TrueFilter", NewTrueFilterPlugin),
},
nodes: []string{"machine1", "machine2"},
@ -411,7 +409,7 @@ func TestGenericScheduler(t *testing.T) {
},
{
// Fits on a machine where the pod ID matches the machine name
registerFilterPlugins: []st.RegisterFilterPluginFunc{
registerPlugins: []st.RegisterPluginFunc{
st.RegisterFilterPlugin("MatchFilter", NewMatchFilterPlugin),
},
nodes: []string{"machine1", "machine2"},
@ -421,10 +419,8 @@ func TestGenericScheduler(t *testing.T) {
wErr: nil,
},
{
registerFilterPlugins: []st.RegisterFilterPluginFunc{
registerPlugins: []st.RegisterPluginFunc{
st.RegisterFilterPlugin("TrueFilter", NewTrueFilterPlugin),
},
registerScorePlugins: []st.RegisterScorePluginFunc{
st.RegisterScorePlugin("NumericMap", newNumericMapPlugin(), 1),
},
nodes: []string{"3", "2", "1"},
@ -434,10 +430,8 @@ func TestGenericScheduler(t *testing.T) {
wErr: nil,
},
{
registerFilterPlugins: []st.RegisterFilterPluginFunc{
registerPlugins: []st.RegisterPluginFunc{
st.RegisterFilterPlugin("MatchFilter", NewMatchFilterPlugin),
},
registerScorePlugins: []st.RegisterScorePluginFunc{
st.RegisterScorePlugin("NumericMap", newNumericMapPlugin(), 1),
},
nodes: []string{"3", "2", "1"},
@ -447,10 +441,8 @@ func TestGenericScheduler(t *testing.T) {
wErr: nil,
},
{
registerFilterPlugins: []st.RegisterFilterPluginFunc{
registerPlugins: []st.RegisterPluginFunc{
st.RegisterFilterPlugin("TrueFilter", NewTrueFilterPlugin),
},
registerScorePlugins: []st.RegisterScorePluginFunc{
st.RegisterScorePlugin("NumericMap", newNumericMapPlugin(), 1),
st.RegisterScorePlugin("ReverseNumericMap", newReverseNumericMapPlugin(), 2),
},
@ -461,11 +453,9 @@ func TestGenericScheduler(t *testing.T) {
wErr: nil,
},
{
registerFilterPlugins: []st.RegisterFilterPluginFunc{
registerPlugins: []st.RegisterPluginFunc{
st.RegisterFilterPlugin("TrueFilter", NewTrueFilterPlugin),
st.RegisterFilterPlugin("FalseFilter", NewFalseFilterPlugin),
},
registerScorePlugins: []st.RegisterScorePluginFunc{
st.RegisterScorePlugin("NumericMap", newNumericMapPlugin(), 1),
},
nodes: []string{"3", "2", "1"},
@ -483,11 +473,9 @@ func TestGenericScheduler(t *testing.T) {
},
},
{
registerFilterPlugins: []st.RegisterFilterPluginFunc{
registerPlugins: []st.RegisterPluginFunc{
st.RegisterFilterPlugin("NoPodsFilter", NewNoPodsFilterPlugin),
st.RegisterFilterPlugin("MatchFilter", NewMatchFilterPlugin),
},
registerScorePlugins: []st.RegisterScorePluginFunc{
st.RegisterScorePlugin("NumericMap", newNumericMapPlugin(), 1),
},
pods: []*v1.Pod{
@ -516,7 +504,7 @@ func TestGenericScheduler(t *testing.T) {
},
{
// Pod with existing PVC
registerFilterPlugins: []st.RegisterFilterPluginFunc{
registerPlugins: []st.RegisterPluginFunc{
st.RegisterFilterPlugin("TrueFilter", NewTrueFilterPlugin),
},
nodes: []string{"machine1", "machine2"},
@ -541,7 +529,7 @@ func TestGenericScheduler(t *testing.T) {
},
{
// Pod with non existing PVC
registerFilterPlugins: []st.RegisterFilterPluginFunc{
registerPlugins: []st.RegisterPluginFunc{
st.RegisterFilterPlugin("TrueFilter", NewTrueFilterPlugin),
},
nodes: []string{"machine1", "machine2"},
@ -564,7 +552,7 @@ func TestGenericScheduler(t *testing.T) {
},
{
// Pod with deleting PVC
registerFilterPlugins: []st.RegisterFilterPluginFunc{
registerPlugins: []st.RegisterPluginFunc{
st.RegisterFilterPlugin("TrueFilter", NewTrueFilterPlugin),
},
nodes: []string{"machine1", "machine2"},
@ -587,10 +575,8 @@ func TestGenericScheduler(t *testing.T) {
wErr: fmt.Errorf("persistentvolumeclaim \"existingPVC\" is being deleted"),
},
{
registerFilterPlugins: []st.RegisterFilterPluginFunc{
registerPlugins: []st.RegisterPluginFunc{
st.RegisterFilterPlugin("TrueFilter", NewTrueFilterPlugin),
},
registerScorePlugins: []st.RegisterScorePluginFunc{
st.RegisterScorePlugin("FalseMap", newFalseMapPlugin(), 1),
st.RegisterScorePlugin("TrueMap", newTrueMapPlugin(), 2),
},
@ -601,8 +587,14 @@ func TestGenericScheduler(t *testing.T) {
},
{
name: "test even pods spread predicate - 2 nodes with maxskew=1",
registerFilterPlugins: []st.RegisterFilterPluginFunc{
st.RegisterFilterPlugin(podtopologyspread.Name, podtopologyspread.New),
registerPlugins: []st.RegisterPluginFunc{
st.RegisterPluginAsExtensions(
podtopologyspread.Name,
1,
podtopologyspread.New,
"PreFilter",
"Filter",
),
},
nodes: []string{"machine1", "machine2"},
pod: &v1.Pod{
@ -636,14 +628,19 @@ func TestGenericScheduler(t *testing.T) {
},
},
},
buildPredMeta: true,
expectedHosts: sets.NewString("machine2"),
wErr: nil,
},
{
name: "test even pods spread predicate - 3 nodes with maxskew=2",
registerFilterPlugins: []st.RegisterFilterPluginFunc{
st.RegisterFilterPlugin(podtopologyspread.Name, podtopologyspread.New),
registerPlugins: []st.RegisterPluginFunc{
st.RegisterPluginAsExtensions(
podtopologyspread.Name,
1,
podtopologyspread.New,
"PreFilter",
"Filter",
),
},
nodes: []string{"machine1", "machine2", "machine3"},
pod: &v1.Pod{
@ -695,19 +692,16 @@ func TestGenericScheduler(t *testing.T) {
},
},
},
buildPredMeta: true,
expectedHosts: sets.NewString("machine2", "machine3"),
wErr: nil,
},
{
name: "test with filter plugin returning Unschedulable status",
registerFilterPlugins: []st.RegisterFilterPluginFunc{
registerPlugins: []st.RegisterPluginFunc{
st.RegisterFilterPlugin(
"FakeFilter",
NewFakeFilterPlugin(map[string]framework.Code{"3": framework.Unschedulable}),
),
},
registerScorePlugins: []st.RegisterScorePluginFunc{
st.RegisterScorePlugin("NumericMap", newNumericMapPlugin(), 1),
},
nodes: []string{"3"},
@ -724,13 +718,11 @@ func TestGenericScheduler(t *testing.T) {
},
{
name: "test with filter plugin returning UnschedulableAndUnresolvable status",
registerFilterPlugins: []st.RegisterFilterPluginFunc{
registerPlugins: []st.RegisterPluginFunc{
st.RegisterFilterPlugin(
"FakeFilter",
NewFakeFilterPlugin(map[string]framework.Code{"3": framework.UnschedulableAndUnresolvable}),
),
},
registerScorePlugins: []st.RegisterScorePluginFunc{
st.RegisterScorePlugin("NumericMap", newNumericMapPlugin(), 1),
},
nodes: []string{"3"},
@ -747,13 +739,11 @@ func TestGenericScheduler(t *testing.T) {
},
{
name: "test with partial failed filter plugin",
registerFilterPlugins: []st.RegisterFilterPluginFunc{
registerPlugins: []st.RegisterPluginFunc{
st.RegisterFilterPlugin(
"FakeFilter",
NewFakeFilterPlugin(map[string]framework.Code{"1": framework.Unschedulable}),
),
},
registerScorePlugins: []st.RegisterScorePluginFunc{
st.RegisterScorePlugin("NumericMap", newNumericMapPlugin(), 1),
},
nodes: []string{"1", "2"},
@ -767,44 +757,42 @@ func TestGenericScheduler(t *testing.T) {
client := clientsetfake.NewSimpleClientset()
informerFactory := informers.NewSharedInformerFactory(client, 0)
registry := framework.Registry{}
plugins := &schedulerapi.Plugins{
Filter: &schedulerapi.PluginSet{},
Score: &schedulerapi.PluginSet{},
}
var pluginConfigs []schedulerapi.PluginConfig
for _, f := range test.registerFilterPlugins {
f(&registry, plugins, pluginConfigs)
}
for _, f := range test.registerScorePlugins {
f(&registry, plugins, pluginConfigs)
}
fwk, _ := framework.NewFramework(registry, plugins, pluginConfigs)
cache := internalcache.New(time.Duration(0), wait.NeverStop)
for _, pod := range test.pods {
cache.AddPod(pod)
}
var nodes []*v1.Node
for _, name := range test.nodes {
cache.AddNode(&v1.Node{ObjectMeta: metav1.ObjectMeta{Name: name, Labels: map[string]string{"hostname": name}}})
node := &v1.Node{ObjectMeta: metav1.ObjectMeta{Name: name, Labels: map[string]string{"hostname": name}}}
nodes = append(nodes, node)
cache.AddNode(node)
}
registry := framework.Registry{}
plugins := &schedulerapi.Plugins{
PreFilter: &schedulerapi.PluginSet{},
Filter: &schedulerapi.PluginSet{},
Score: &schedulerapi.PluginSet{},
}
var pluginConfigs []schedulerapi.PluginConfig
for _, f := range test.registerPlugins {
f(&registry, plugins, pluginConfigs)
}
snapshot := nodeinfosnapshot.NewSnapshot(nodeinfosnapshot.CreateNodeInfoMap(test.pods, nodes))
fwk, _ := framework.NewFramework(registry, plugins, pluginConfigs, framework.WithSnapshotSharedLister(snapshot))
pvcs := []v1.PersistentVolumeClaim{}
pvcs = append(pvcs, test.pvcs...)
pvcLister := fakelisters.PersistentVolumeClaimLister(pvcs)
predMetaProducer := algorithmpredicates.EmptyMetadataProducer
if test.buildPredMeta {
f := &algorithmpredicates.MetadataProducerFactory{}
predMetaProducer = f.GetPredicateMetadata
}
scheduler := NewGenericScheduler(
cache,
internalqueue.NewSchedulingQueue(nil),
nil,
predMetaProducer,
algorithmpredicates.EmptyMetadataProducer,
// test.prioritizers,
priorities.EmptyMetadataProducer,
emptySnapshot,
snapshot,
fwk,
[]algorithm.SchedulerExtender{},
nil,
@ -829,7 +817,7 @@ func TestGenericScheduler(t *testing.T) {
}
// makeScheduler makes a simple genericScheduler for testing.
func makeScheduler(nodes []*v1.Node, fns ...st.RegisterFilterPluginFunc) *genericScheduler {
func makeScheduler(nodes []*v1.Node, fns ...st.RegisterPluginFunc) *genericScheduler {
cache := internalcache.New(time.Duration(0), wait.NeverStop)
for _, n := range nodes {
cache.AddNode(n)
@ -1151,7 +1139,7 @@ func TestZeroRequest(t *testing.T) {
Score: &schedulerapi.PluginSet{},
}
var pluginConfigs []schedulerapi.PluginConfig
pluginRegistrations := []st.RegisterScorePluginFunc{
pluginRegistrations := []st.RegisterPluginFunc{
st.RegisterScorePlugin(noderesources.LeastAllocatedName, noderesources.NewLeastAllocated, 1),
st.RegisterScorePlugin(noderesources.BalancedAllocationName, noderesources.NewBalancedAllocation, 1),
st.RegisterScorePlugin(defaultpodtopologyspread.Name, defaultpodtopologyspread.New, 1),
@ -1314,7 +1302,7 @@ func TestSelectNodesForPreemption(t *testing.T) {
tests := []struct {
name string
registerFilterPlugins []st.RegisterFilterPluginFunc
registerPlugins []st.RegisterPluginFunc
nodes []string
pod *v1.Pod
pods []*v1.Pod
@ -1325,7 +1313,7 @@ func TestSelectNodesForPreemption(t *testing.T) {
}{
{
name: "a pod that does not fit on any machine",
registerFilterPlugins: []st.RegisterFilterPluginFunc{
registerPlugins: []st.RegisterPluginFunc{
st.RegisterFilterPlugin("FalseFilter", NewFalseFilterPlugin),
},
nodes: []string{"machine1", "machine2"},
@ -1338,7 +1326,7 @@ func TestSelectNodesForPreemption(t *testing.T) {
},
{
name: "a pod that fits with no preemption",
registerFilterPlugins: []st.RegisterFilterPluginFunc{
registerPlugins: []st.RegisterPluginFunc{
st.RegisterFilterPlugin("TrueFilter", NewTrueFilterPlugin),
},
nodes: []string{"machine1", "machine2"},
@ -1351,7 +1339,7 @@ func TestSelectNodesForPreemption(t *testing.T) {
},
{
name: "a pod that fits on one machine with no preemption",
registerFilterPlugins: []st.RegisterFilterPluginFunc{
registerPlugins: []st.RegisterPluginFunc{
st.RegisterFilterPlugin("MatchFilter", NewMatchFilterPlugin),
},
nodes: []string{"machine1", "machine2"},
@ -1364,7 +1352,7 @@ func TestSelectNodesForPreemption(t *testing.T) {
},
{
name: "a pod that fits on both machines when lower priority pods are preempted",
registerFilterPlugins: []st.RegisterFilterPluginFunc{
registerPlugins: []st.RegisterPluginFunc{
st.RegisterFilterPlugin(noderesources.FitName, noderesources.NewFit),
},
nodes: []string{"machine1", "machine2"},
@ -1377,7 +1365,7 @@ func TestSelectNodesForPreemption(t *testing.T) {
},
{
name: "a pod that would fit on the machines, but other pods running are higher priority",
registerFilterPlugins: []st.RegisterFilterPluginFunc{
registerPlugins: []st.RegisterPluginFunc{
st.RegisterFilterPlugin(noderesources.FitName, noderesources.NewFit),
},
nodes: []string{"machine1", "machine2"},
@ -1390,7 +1378,7 @@ func TestSelectNodesForPreemption(t *testing.T) {
},
{
name: "medium priority pod is preempted, but lower priority one stays as it is small",
registerFilterPlugins: []st.RegisterFilterPluginFunc{
registerPlugins: []st.RegisterPluginFunc{
st.RegisterFilterPlugin(noderesources.FitName, noderesources.NewFit),
},
nodes: []string{"machine1", "machine2"},
@ -1404,7 +1392,7 @@ func TestSelectNodesForPreemption(t *testing.T) {
},
{
name: "mixed priority pods are preempted",
registerFilterPlugins: []st.RegisterFilterPluginFunc{
registerPlugins: []st.RegisterPluginFunc{
st.RegisterFilterPlugin(noderesources.FitName, noderesources.NewFit),
},
nodes: []string{"machine1", "machine2"},
@ -1420,7 +1408,7 @@ func TestSelectNodesForPreemption(t *testing.T) {
},
{
name: "mixed priority pods are preempted, pick later StartTime one when priorities are equal",
registerFilterPlugins: []st.RegisterFilterPluginFunc{
registerPlugins: []st.RegisterPluginFunc{
st.RegisterFilterPlugin(noderesources.FitName, noderesources.NewFit),
},
nodes: []string{"machine1", "machine2"},
@ -1436,7 +1424,7 @@ func TestSelectNodesForPreemption(t *testing.T) {
},
{
name: "pod with anti-affinity is preempted",
registerFilterPlugins: []st.RegisterFilterPluginFunc{
registerPlugins: []st.RegisterPluginFunc{
st.RegisterFilterPlugin(noderesources.FitName, noderesources.NewFit),
st.RegisterFilterPlugin(interpodaffinity.Name, interpodaffinity.New),
},
@ -1471,8 +1459,14 @@ func TestSelectNodesForPreemption(t *testing.T) {
},
{
name: "preemption to resolve even pods spread FitError",
registerFilterPlugins: []st.RegisterFilterPluginFunc{
st.RegisterFilterPlugin(podtopologyspread.Name, podtopologyspread.New),
registerPlugins: []st.RegisterPluginFunc{
st.RegisterPluginAsExtensions(
podtopologyspread.Name,
1,
podtopologyspread.New,
"PreFilter",
"Filter",
),
},
nodes: []string{"node-a/zone1", "node-b/zone1", "node-x/zone2"},
pod: &v1.Pod{
@ -1547,7 +1541,7 @@ func TestSelectNodesForPreemption(t *testing.T) {
},
{
name: "get Unschedulable in the preemption phase when the filter plugins filtering the nodes",
registerFilterPlugins: []st.RegisterFilterPluginFunc{
registerPlugins: []st.RegisterPluginFunc{
st.RegisterFilterPlugin(noderesources.FitName, noderesources.NewFit),
},
nodes: []string{"machine1", "machine2"},
@ -1591,7 +1585,8 @@ func TestSelectNodesForPreemption(t *testing.T) {
registry := framework.Registry{}
plugins := &schedulerapi.Plugins{
Filter: &schedulerapi.PluginSet{},
PreFilter: &schedulerapi.PluginSet{},
Filter: &schedulerapi.PluginSet{},
}
var pluginConfigs []schedulerapi.PluginConfig
// For each test, prepend a FakeFilterPlugin.
@ -1605,7 +1600,7 @@ func TestSelectNodesForPreemption(t *testing.T) {
)
registerFakeFilterFunc(&registry, plugins, pluginConfigs)
// Next, register other filter plugins defined in test struct.
for _, f := range test.registerFilterPlugins {
for _, f := range test.registerPlugins {
f(&registry, plugins, pluginConfigs)
}
// Use a real snapshot since it's needed in some Filter Plugin (e.g., PodAffinity)
@ -1639,6 +1634,11 @@ func TestSelectNodesForPreemption(t *testing.T) {
newnode.ObjectMeta.Labels = map[string]string{"hostname": "newnode"}
nodes = append(nodes, newnode)
state := framework.NewCycleState()
// Some tests rely on PreFilter plugin to compute its CycleState.
preFilterStatus := fwk.RunPreFilterPlugins(context.Background(), state, test.pod)
if !preFilterStatus.IsSuccess() {
t.Errorf("Unexpected preFilterStatus: %v", preFilterStatus)
}
nodeToPods, err := g.selectNodesForPreemption(context.Background(), state, test.pod, nodes, nil)
if err != nil {
t.Error(err)
@ -1660,7 +1660,7 @@ func TestPickOneNodeForPreemption(t *testing.T) {
defer algorithmpredicates.SetPredicatesOrderingDuringTest(order)()
tests := []struct {
name string
registerFilterPlugin st.RegisterFilterPluginFunc
registerFilterPlugin st.RegisterPluginFunc
nodes []string
pod *v1.Pod
pods []*v1.Pod
@ -2052,8 +2052,7 @@ func TestPreempt(t *testing.T) {
extenders []*FakeExtender
failedNodeToStatusMap framework.NodeToStatusMap
nodeNames []string
registerFilterPlugin st.RegisterFilterPluginFunc
buildPredMeta bool
registerPlugin st.RegisterPluginFunc
expectedNode string
expectedPods []string // list of preempted pods
}{
@ -2070,9 +2069,9 @@ func TestPreempt(t *testing.T) {
{ObjectMeta: metav1.ObjectMeta{Name: "m2.1", UID: types.UID("m2.1")}, Spec: v1.PodSpec{Containers: largeContainers, Priority: &highPriority, NodeName: "machine2"}, Status: v1.PodStatus{Phase: v1.PodRunning}},
{ObjectMeta: metav1.ObjectMeta{Name: "m3.1", UID: types.UID("m3.1")}, Spec: v1.PodSpec{Containers: mediumContainers, Priority: &midPriority, NodeName: "machine3"}, Status: v1.PodStatus{Phase: v1.PodRunning}},
},
registerFilterPlugin: st.RegisterFilterPlugin(noderesources.FitName, noderesources.NewFit),
expectedNode: "machine1",
expectedPods: []string{"m1.1", "m1.2"},
registerPlugin: st.RegisterFilterPlugin(noderesources.FitName, noderesources.NewFit),
expectedNode: "machine1",
expectedPods: []string{"m1.1", "m1.2"},
},
{
name: "One node doesn't need any preemption",
@ -2087,9 +2086,9 @@ func TestPreempt(t *testing.T) {
{ObjectMeta: metav1.ObjectMeta{Name: "m2.1", UID: types.UID("m2.1")}, Spec: v1.PodSpec{Containers: largeContainers, Priority: &highPriority, NodeName: "machine2"}, Status: v1.PodStatus{Phase: v1.PodRunning}},
},
registerFilterPlugin: st.RegisterFilterPlugin(noderesources.FitName, noderesources.NewFit),
expectedNode: "machine3",
expectedPods: []string{},
registerPlugin: st.RegisterFilterPlugin(noderesources.FitName, noderesources.NewFit),
expectedNode: "machine3",
expectedPods: []string{},
},
{
name: "preemption for topology spread constraints",
@ -2162,11 +2161,16 @@ func TestPreempt(t *testing.T) {
"node-b": framework.NewStatus(framework.Unschedulable, algorithmpredicates.ErrTopologySpreadConstraintsNotMatch.GetReason()),
"node-x": framework.NewStatus(framework.Unschedulable, algorithmpredicates.ErrTopologySpreadConstraintsNotMatch.GetReason()),
},
buildPredMeta: true,
nodeNames: []string{"node-a/zone1", "node-b/zone1", "node-x/zone2"},
registerFilterPlugin: st.RegisterFilterPlugin(podtopologyspread.Name, podtopologyspread.New),
expectedNode: "node-b",
expectedPods: []string{"pod-b1"},
nodeNames: []string{"node-a/zone1", "node-b/zone1", "node-x/zone2"},
registerPlugin: st.RegisterPluginAsExtensions(
podtopologyspread.Name,
1,
podtopologyspread.New,
"PreFilter",
"Filter",
),
expectedNode: "node-b",
expectedPods: []string{"pod-b1"},
},
{
name: "Scheduler extenders allow only machine1, otherwise machine3 would have been chosen",
@ -2189,9 +2193,9 @@ func TestPreempt(t *testing.T) {
predicates: []fitPredicate{machine1PredicateExtender},
},
},
registerFilterPlugin: st.RegisterFilterPlugin(noderesources.FitName, noderesources.NewFit),
expectedNode: "machine1",
expectedPods: []string{"m1.1", "m1.2"},
registerPlugin: st.RegisterFilterPlugin(noderesources.FitName, noderesources.NewFit),
expectedNode: "machine1",
expectedPods: []string{"m1.1", "m1.2"},
},
{
name: "Scheduler extenders do not allow any preemption",
@ -2211,9 +2215,9 @@ func TestPreempt(t *testing.T) {
predicates: []fitPredicate{falsePredicateExtender},
},
},
registerFilterPlugin: st.RegisterFilterPlugin(noderesources.FitName, noderesources.NewFit),
expectedNode: "",
expectedPods: []string{},
registerPlugin: st.RegisterFilterPlugin(noderesources.FitName, noderesources.NewFit),
expectedNode: "",
expectedPods: []string{},
},
{
name: "One scheduler extender allows only machine1, the other returns error but ignorable. Only machine1 would be chosen",
@ -2237,9 +2241,9 @@ func TestPreempt(t *testing.T) {
predicates: []fitPredicate{machine1PredicateExtender},
},
},
registerFilterPlugin: st.RegisterFilterPlugin(noderesources.FitName, noderesources.NewFit),
expectedNode: "machine1",
expectedPods: []string{"m1.1", "m1.2"},
registerPlugin: st.RegisterFilterPlugin(noderesources.FitName, noderesources.NewFit),
expectedNode: "machine1",
expectedPods: []string{"m1.1", "m1.2"},
},
{
name: "One scheduler extender allows only machine1, but it is not interested in given pod, otherwise machine1 would have been chosen",
@ -2263,9 +2267,9 @@ func TestPreempt(t *testing.T) {
predicates: []fitPredicate{truePredicateExtender},
},
},
registerFilterPlugin: st.RegisterFilterPlugin(noderesources.FitName, noderesources.NewFit),
expectedNode: "machine3",
expectedPods: []string{},
registerPlugin: st.RegisterFilterPlugin(noderesources.FitName, noderesources.NewFit),
expectedNode: "machine3",
expectedPods: []string{},
},
{
name: "no preempting in pod",
@ -2280,9 +2284,9 @@ func TestPreempt(t *testing.T) {
{ObjectMeta: metav1.ObjectMeta{Name: "m2.1", UID: types.UID("m2.1")}, Spec: v1.PodSpec{Containers: largeContainers, Priority: &highPriority, NodeName: "machine2"}, Status: v1.PodStatus{Phase: v1.PodRunning}},
{ObjectMeta: metav1.ObjectMeta{Name: "m3.1", UID: types.UID("m3.1")}, Spec: v1.PodSpec{Containers: mediumContainers, Priority: &midPriority, NodeName: "machine3"}, Status: v1.PodStatus{Phase: v1.PodRunning}},
},
registerFilterPlugin: st.RegisterFilterPlugin(noderesources.FitName, noderesources.NewFit),
expectedNode: "",
expectedPods: nil,
registerPlugin: st.RegisterFilterPlugin(noderesources.FitName, noderesources.NewFit),
expectedNode: "",
expectedPods: nil,
},
{
name: "PreemptionPolicy is nil",
@ -2297,9 +2301,9 @@ func TestPreempt(t *testing.T) {
{ObjectMeta: metav1.ObjectMeta{Name: "m2.1", UID: types.UID("m2.1")}, Spec: v1.PodSpec{Containers: largeContainers, Priority: &highPriority, NodeName: "machine2"}, Status: v1.PodStatus{Phase: v1.PodRunning}},
{ObjectMeta: metav1.ObjectMeta{Name: "m3.1", UID: types.UID("m3.1")}, Spec: v1.PodSpec{Containers: mediumContainers, Priority: &midPriority, NodeName: "machine3"}, Status: v1.PodStatus{Phase: v1.PodRunning}},
},
registerFilterPlugin: st.RegisterFilterPlugin(noderesources.FitName, noderesources.NewFit),
expectedNode: "machine1",
expectedPods: []string{"m1.1", "m1.2"},
registerPlugin: st.RegisterFilterPlugin(noderesources.FitName, noderesources.NewFit),
expectedNode: "machine1",
expectedPods: []string{"m1.1", "m1.2"},
},
}
@ -2319,6 +2323,7 @@ func TestPreempt(t *testing.T) {
if len(test.nodeNames) != 0 {
nodeNames = test.nodeNames
}
var nodes []*v1.Node
for i, name := range nodeNames {
node := makeNode(name, 1000*5, priorityutil.DefaultMemoryRequest*5)
// if possible, split node name by '/' to form labels in a format of
@ -2329,6 +2334,7 @@ func TestPreempt(t *testing.T) {
}
node.Name = node.ObjectMeta.Labels["hostname"]
cache.AddNode(node)
nodes = append(nodes, node)
nodeNames[i] = node.Name
// Set nodeInfo to extenders to mock extenders' cache for preemption.
@ -2336,33 +2342,30 @@ func TestPreempt(t *testing.T) {
cachedNodeInfo.SetNode(node)
cachedNodeInfoMap[node.Name] = cachedNodeInfo
}
extenders := []algorithm.SchedulerExtender{}
var extenders []algorithm.SchedulerExtender
for _, extender := range test.extenders {
// Set nodeInfoMap as extenders cached node information.
extender.cachedNodeNameToInfo = cachedNodeInfoMap
extenders = append(extenders, extender)
}
predMetaProducer := algorithmpredicates.EmptyMetadataProducer
if test.buildPredMeta {
f := &algorithmpredicates.MetadataProducerFactory{}
predMetaProducer = f.GetPredicateMetadata
}
registry := framework.Registry{}
plugins := &schedulerapi.Plugins{
Filter: &schedulerapi.PluginSet{},
PreFilter: &schedulerapi.PluginSet{},
Filter: &schedulerapi.PluginSet{},
}
var pluginConfigs []schedulerapi.PluginConfig
test.registerFilterPlugin(&registry, plugins, pluginConfigs)
fwk, _ := framework.NewFramework(registry, plugins, pluginConfigs)
test.registerPlugin(&registry, plugins, pluginConfigs)
snapshot := nodeinfosnapshot.NewSnapshot(nodeinfosnapshot.CreateNodeInfoMap(test.pods, nodes))
fwk, _ := framework.NewFramework(registry, plugins, pluginConfigs, framework.WithSnapshotSharedLister(snapshot))
scheduler := NewGenericScheduler(
cache,
internalqueue.NewSchedulingQueue(nil),
nil,
predMetaProducer,
algorithmpredicates.EmptyMetadataProducer,
priorities.EmptyMetadataProducer,
emptySnapshot,
snapshot,
fwk,
extenders,
nil,
@ -2373,7 +2376,11 @@ func TestPreempt(t *testing.T) {
schedulerapi.DefaultPercentageOfNodesToScore,
true)
state := framework.NewCycleState()
scheduler.Snapshot()
// Some tests rely on PreFilter plugin to compute its CycleState.
preFilterStatus := fwk.RunPreFilterPlugins(context.Background(), state, test.pod)
if !preFilterStatus.IsSuccess() {
t.Errorf("Unexpected preFilterStatus: %v", preFilterStatus)
}
// Call Preempt and check the expected results.
failedNodeToStatusMap := defaultFailedNodeToStatusMap
if test.failedNodeToStatusMap != nil {

1
pkg/scheduler/framework/plugins/default_registry.go
View File

@ -205,6 +205,7 @@ func NewDefaultConfigProducerRegistry() *ConfigProducerRegistry {
})
registry.RegisterPredicate(predicates.EvenPodsSpreadPred,
func(_ ConfigProducerArgs) (plugins config.Plugins, pluginConfig []config.PluginConfig) {
plugins.PreFilter = appendToPluginSet(plugins.PreFilter, podtopologyspread.Name, nil)
plugins.Filter = appendToPluginSet(plugins.Filter, podtopologyspread.Name, nil)
return
})

4
pkg/scheduler/framework/plugins/podtopologyspread/BUILD
View File

@ -10,9 +10,11 @@ go_library(
"//pkg/scheduler/algorithm/priorities:go_default_library",
"//pkg/scheduler/framework/plugins/migration:go_default_library",
"//pkg/scheduler/framework/v1alpha1:go_default_library",
"//pkg/scheduler/listers:go_default_library",
"//pkg/scheduler/nodeinfo:go_default_library",
"//staging/src/k8s.io/api/core/v1:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/runtime:go_default_library",
"//vendor/k8s.io/klog:go_default_library",
],
)
@ -21,8 +23,6 @@ go_test(
srcs = ["pod_topology_spread_test.go"],
embed = [":go_default_library"],
deps = [
"//pkg/scheduler/algorithm/predicates:go_default_library",
"//pkg/scheduler/framework/plugins/migration:go_default_library",
"//pkg/scheduler/framework/v1alpha1:go_default_library",
"//pkg/scheduler/nodeinfo/snapshot:go_default_library",
"//pkg/scheduler/testing:go_default_library",

112
pkg/scheduler/framework/plugins/podtopologyspread/pod_topology_spread.go
View File

@ -22,36 +22,125 @@ import (
"k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/klog"
"k8s.io/kubernetes/pkg/scheduler/algorithm/predicates"
"k8s.io/kubernetes/pkg/scheduler/algorithm/priorities"
"k8s.io/kubernetes/pkg/scheduler/framework/plugins/migration"
framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
schedulerlisters "k8s.io/kubernetes/pkg/scheduler/listers"
"k8s.io/kubernetes/pkg/scheduler/nodeinfo"
)
// PodTopologySpread is a plugin that ensures pod's topologySpreadConstraints is satisfied.
type PodTopologySpread struct {
handle framework.FrameworkHandle
snapshotSharedLister schedulerlisters.SharedLister
}
var _ framework.PreFilterPlugin = &PodTopologySpread{}
var _ framework.FilterPlugin = &PodTopologySpread{}
var _ framework.ScorePlugin = &PodTopologySpread{}
// Name is the name of the plugin used in the plugin registry and configurations.
const Name = "PodTopologySpread"
const (
// Name is the name of the plugin used in the plugin registry and configurations.
Name = "PodTopologySpread"
// preFilterStateKey is the key in CycleState to PodTopologySpread pre-computed data.
// Using the name of the plugin will likely help us avoid collisions with other plugins.
preFilterStateKey = "PreFilter" + Name
)
// Name returns name of the plugin. It is used in logs, etc.
func (pl *PodTopologySpread) Name() string {
return Name
}
// preFilterState computed at PreFilter and used at Filter.
type preFilterState struct {
// `nil` represents the meta is not set at all (in PreFilter phase)
// An empty `PodTopologySpreadMetadata` object denotes it's a legit meta and is set in PreFilter phase.
meta *predicates.PodTopologySpreadMetadata
}
// Clone makes a copy of the given state.
func (s *preFilterState) Clone() framework.StateData {
copy := &preFilterState{
meta: s.meta.Clone(),
}
return copy
}
// PreFilter invoked at the prefilter extension point.
func (pl *PodTopologySpread) PreFilter(ctx context.Context, cycleState *framework.CycleState, pod *v1.Pod) *framework.Status {
var meta *predicates.PodTopologySpreadMetadata
var allNodes []*nodeinfo.NodeInfo
var err error
if allNodes, err = pl.snapshotSharedLister.NodeInfos().List(); err != nil {
return framework.NewStatus(framework.Error, fmt.Sprintf("failed to list NodeInfos: %v", err))
}
if meta, err = predicates.GetPodTopologySpreadMetadata(pod, allNodes); err != nil {
return framework.NewStatus(framework.Error, fmt.Sprintf("Error calculating podTopologySpreadMetadata: %v", err))
}
s := &preFilterState{
meta: meta,
}
cycleState.Write(preFilterStateKey, s)
return nil
}
// PreFilterExtensions returns prefilter extensions, pod add and remove.
func (pl *PodTopologySpread) PreFilterExtensions() framework.PreFilterExtensions {
return pl
}
// AddPod from pre-computed data in cycleState.
func (pl *PodTopologySpread) AddPod(ctx context.Context, cycleState *framework.CycleState, podToSchedule *v1.Pod, podToAdd *v1.Pod, nodeInfo *nodeinfo.NodeInfo) *framework.Status {
meta, err := getPodTopologySpreadMetadata(cycleState)
if err != nil {
return framework.NewStatus(framework.Error, err.Error())
}
meta.AddPod(podToAdd, podToSchedule, nodeInfo.Node())
return nil
}
// RemovePod from pre-computed data in cycleState.
func (pl *PodTopologySpread) RemovePod(ctx context.Context, cycleState *framework.CycleState, podToSchedule *v1.Pod, podToRemove *v1.Pod, nodeInfo *nodeinfo.NodeInfo) *framework.Status {
meta, err := getPodTopologySpreadMetadata(cycleState)
if err != nil {
return framework.NewStatus(framework.Error, err.Error())
}
meta.RemovePod(podToRemove, podToSchedule, nodeInfo.Node())
return nil
}
func getPodTopologySpreadMetadata(cycleState *framework.CycleState) (*predicates.PodTopologySpreadMetadata, error) {
c, err := cycleState.Read(preFilterStateKey)
if err != nil {
// The metadata wasn't pre-computed in prefilter. We ignore the error for now since
// we are able to handle that by computing it again (e.g. in Filter()).
klog.Error(err)
return nil, nil
}
s, ok := c.(*preFilterState)
if !ok {
return nil, fmt.Errorf("%+v convert to podtopologyspread.state error", c)
}
return s.meta, nil
}
// Filter invoked at the filter extension point.
func (pl *PodTopologySpread) Filter(ctx context.Context, cycleState *framework.CycleState, pod *v1.Pod, nodeInfo *nodeinfo.NodeInfo) *framework.Status {
meta, ok := migration.CovertStateRefToPredMeta(migration.PredicateMetadata(cycleState))
if !ok {
return migration.ErrorToFrameworkStatus(fmt.Errorf("%+v convert to predicates.Metadata error", cycleState))
meta, err := getPodTopologySpreadMetadata(cycleState)
if err != nil {
return framework.NewStatus(framework.Error, err.Error())
}
_, reasons, err := predicates.EvenPodsSpreadPredicate(pod, meta, nodeInfo)
_, reasons, err := predicates.PodTopologySpreadPredicate(pod, meta, nodeInfo)
return migration.PredicateResultToFrameworkStatus(reasons, err)
}
@ -59,7 +148,7 @@ func (pl *PodTopologySpread) Filter(ctx context.Context, cycleState *framework.C
// The "score" returned in this function is the matching number of pods on the `nodeName`,
// it is normalized later.
func (pl *PodTopologySpread) Score(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeName string) (int64, *framework.Status) {
nodeInfo, err := pl.handle.SnapshotSharedLister().NodeInfos().Get(nodeName)
nodeInfo, err := pl.snapshotSharedLister.NodeInfos().Get(nodeName)
if err != nil {
return 0, framework.NewStatus(framework.Error, fmt.Sprintf("getting node %q from Snapshot: %v", nodeName, err))
}
@ -72,7 +161,7 @@ func (pl *PodTopologySpread) Score(ctx context.Context, state *framework.CycleSt
// NormalizeScore invoked after scoring all nodes.
func (pl *PodTopologySpread) NormalizeScore(ctx context.Context, state *framework.CycleState, pod *v1.Pod, scores framework.NodeScoreList) *framework.Status {
meta := migration.PriorityMetadata(state)
err := priorities.CalculateEvenPodsSpreadPriorityReduce(pod, meta, pl.handle.SnapshotSharedLister(), scores)
err := priorities.CalculateEvenPodsSpreadPriorityReduce(pod, meta, pl.snapshotSharedLister, scores)
return migration.ErrorToFrameworkStatus(err)
}
@ -83,5 +172,8 @@ func (pl *PodTopologySpread) ScoreExtensions() framework.ScoreExtensions {
// New initializes a new plugin and returns it.
func New(_ *runtime.Unknown, h framework.FrameworkHandle) (framework.Plugin, error) {
return &PodTopologySpread{handle: h}, nil
if h.SnapshotSharedLister() == nil {
return nil, fmt.Errorf("SnapshotSharedlister is nil")
}
return &PodTopologySpread{snapshotSharedLister: h.SnapshotSharedLister()}, nil
}

30
pkg/scheduler/framework/plugins/podtopologyspread/pod_topology_spread_test.go
View File

@ -21,8 +21,6 @@ import (
"testing"
"k8s.io/api/core/v1"
"k8s.io/kubernetes/pkg/scheduler/algorithm/predicates"
"k8s.io/kubernetes/pkg/scheduler/framework/plugins/migration"
framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
nodeinfosnapshot "k8s.io/kubernetes/pkg/scheduler/nodeinfo/snapshot"
st "k8s.io/kubernetes/pkg/scheduler/testing"
@ -32,7 +30,7 @@ var (
hardSpread = v1.DoNotSchedule
)
func TestPodTopologySpread_Filter_SingleConstraint(t *testing.T) {
func TestSingleConstraint(t *testing.T) {
tests := []struct {
name string
pod *v1.Pod
@ -270,14 +268,16 @@ func TestPodTopologySpread_Filter_SingleConstraint(t *testing.T) {
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
snapshot := nodeinfosnapshot.NewSnapshot(nodeinfosnapshot.CreateNodeInfoMap(tt.existingPods, tt.nodes))
factory := &predicates.MetadataProducerFactory{}
meta := factory.GetPredicateMetadata(tt.pod, snapshot)
p := &PodTopologySpread{snapshotSharedLister: snapshot}
state := framework.NewCycleState()
state.Write(migration.PredicatesStateKey, &migration.PredicatesStateData{Reference: meta})
plugin, _ := New(nil, nil)
preFilterStatus := p.PreFilter(context.Background(), state, tt.pod)
if !preFilterStatus.IsSuccess() {
t.Errorf("preFilter failed with status: %v", preFilterStatus)
}
for _, node := range tt.nodes {
nodeInfo, _ := snapshot.NodeInfos().Get(node.Name)
status := plugin.(*PodTopologySpread).Filter(context.Background(), state, tt.pod, nodeInfo)
status := p.Filter(context.Background(), state, tt.pod, nodeInfo)
if status.IsSuccess() != tt.fits[node.Name] {
t.Errorf("[%s]: expected %v got %v", node.Name, tt.fits[node.Name], status.IsSuccess())
}
@ -286,7 +286,7 @@ func TestPodTopologySpread_Filter_SingleConstraint(t *testing.T) {
}
}
func TestPodTopologySpread_Filter_MultipleConstraints(t *testing.T) {
func TestMultipleConstraints(t *testing.T) {
tests := []struct {
name string
pod *v1.Pod
@ -468,14 +468,16 @@ func TestPodTopologySpread_Filter_MultipleConstraints(t *testing.T) {
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
snapshot := nodeinfosnapshot.NewSnapshot(nodeinfosnapshot.CreateNodeInfoMap(tt.existingPods, tt.nodes))
factory := &predicates.MetadataProducerFactory{}
meta := factory.GetPredicateMetadata(tt.pod, snapshot)
p := &PodTopologySpread{snapshotSharedLister: snapshot}
state := framework.NewCycleState()
state.Write(migration.PredicatesStateKey, &migration.PredicatesStateData{Reference: meta})
plugin, _ := New(nil, nil)
preFilterStatus := p.PreFilter(context.Background(), state, tt.pod)
if !preFilterStatus.IsSuccess() {
t.Errorf("preFilter failed with status: %v", preFilterStatus)
}
for _, node := range tt.nodes {
nodeInfo, _ := snapshot.NodeInfos().Get(node.Name)
status := plugin.(*PodTopologySpread).Filter(context.Background(), state, tt.pod, nodeInfo)
status := p.Filter(context.Background(), state, tt.pod, nodeInfo)
if status.IsSuccess() != tt.fits[node.Name] {
t.Errorf("[%s]: expected %v got %v", node.Name, tt.fits[node.Name], status.IsSuccess())
}

2
pkg/scheduler/framework/v1alpha1/framework.go
View File

@ -87,7 +87,7 @@ type extensionPoint struct {
// the set of plugins to be configured at this extension point.
plugins *config.PluginSet
// a pointer to the slice storing plugins implementations that will run at this
// extenstion point.
// extension point.
slicePtr interface{}
}

6
pkg/scheduler/scheduler_test.go
View File

@ -568,7 +568,7 @@ func TestSchedulerErrorWithLongBinding(t *testing.T) {
// queuedPodStore: pods queued before processing.
// cache: scheduler cache that might contain assumed pods.
func setupTestSchedulerWithOnePodOnNode(t *testing.T, queuedPodStore *clientcache.FIFO, scache internalcache.Cache,
informerFactory informers.SharedInformerFactory, stop chan struct{}, f st.RegisterFilterPluginFunc, pod *v1.Pod, node *v1.Node) (*Scheduler, chan *v1.Binding, chan error) {
informerFactory informers.SharedInformerFactory, stop chan struct{}, f st.RegisterPluginFunc, pod *v1.Pod, node *v1.Node) (*Scheduler, chan *v1.Binding, chan error) {
scheduler, bindingChan, errChan := setupTestScheduler(queuedPodStore, scache, informerFactory, f, nil)
@ -679,7 +679,7 @@ func TestSchedulerFailedSchedulingReasons(t *testing.T) {
// queuedPodStore: pods queued before processing.
// scache: scheduler cache that might contain assumed pods.
func setupTestScheduler(queuedPodStore *clientcache.FIFO, scache internalcache.Cache, informerFactory informers.SharedInformerFactory, f st.RegisterFilterPluginFunc, recorder events.EventRecorder) (*Scheduler, chan *v1.Binding, chan error) {
func setupTestScheduler(queuedPodStore *clientcache.FIFO, scache internalcache.Cache, informerFactory informers.SharedInformerFactory, f st.RegisterPluginFunc, recorder events.EventRecorder) (*Scheduler, chan *v1.Binding, chan error) {
registry := framework.Registry{}
plugins := &schedulerapi.Plugins{
Filter: &schedulerapi.PluginSet{},
@ -736,7 +736,7 @@ func setupTestScheduler(queuedPodStore *clientcache.FIFO, scache internalcache.C
return sched, bindingChan, errChan
}
func setupTestSchedulerLongBindingWithRetry(queuedPodStore *clientcache.FIFO, scache internalcache.Cache, informerFactory informers.SharedInformerFactory, f st.RegisterFilterPluginFunc, stop chan struct{}, bindingTime time.Duration) (*Scheduler, chan *v1.Binding) {
func setupTestSchedulerLongBindingWithRetry(queuedPodStore *clientcache.FIFO, scache internalcache.Cache, informerFactory informers.SharedInformerFactory, f st.RegisterPluginFunc, stop chan struct{}, bindingTime time.Duration) (*Scheduler, chan *v1.Binding) {
registry := framework.Registry{}
plugins := &schedulerapi.Plugins{
Filter: &schedulerapi.PluginSet{},

54
pkg/scheduler/testing/framework_helpers.go
View File

@ -21,32 +21,54 @@ import (
framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
)
// RegisterFilterPluginFunc is a function signature used in method RegisterFilterPlugin()
// RegisterPluginFunc is a function signature used in method RegisterFilterPlugin()
// to register a Filter Plugin to a given registry.
type RegisterFilterPluginFunc func(reg *framework.Registry, plugins *schedulerapi.Plugins, pluginConfigs []schedulerapi.PluginConfig)
type RegisterPluginFunc func(reg *framework.Registry, plugins *schedulerapi.Plugins, pluginConfigs []schedulerapi.PluginConfig)
// RegisterFilterPlugin returns a function to register a Filter Plugin to a given registry.
func RegisterFilterPlugin(pluginName string, pluginNewFunc framework.PluginFactory) RegisterFilterPluginFunc {
return func(reg *framework.Registry, plugins *schedulerapi.Plugins, pluginConfigs []schedulerapi.PluginConfig) {
reg.Register(pluginName, pluginNewFunc)
plugins.Filter.Enabled = append(plugins.Filter.Enabled, schedulerapi.Plugin{Name: pluginName})
//lint:ignore SA4006 this value of pluginConfigs is never used.
//lint:ignore SA4010 this result of append is never used.
pluginConfigs = append(pluginConfigs, schedulerapi.PluginConfig{Name: pluginName})
}
func RegisterFilterPlugin(pluginName string, pluginNewFunc framework.PluginFactory) RegisterPluginFunc {
return RegisterPluginAsExtensions(pluginName, 1, pluginNewFunc, "Filter")
}
// RegisterScorePluginFunc is a function signature used in method RegisterScorePlugin()
// to register a Score Plugin to a given registry.
type RegisterScorePluginFunc func(reg *framework.Registry, plugins *schedulerapi.Plugins, pluginConfigs []schedulerapi.PluginConfig)
// RegisterScorePlugin returns a function to register a Score Plugin to a given registry.
func RegisterScorePlugin(pluginName string, pluginNewFunc framework.PluginFactory, weight int32) RegisterScorePluginFunc {
func RegisterScorePlugin(pluginName string, pluginNewFunc framework.PluginFactory, weight int32) RegisterPluginFunc {
return RegisterPluginAsExtensions(pluginName, weight, pluginNewFunc, "Score")
}
// RegisterPluginAsExtensions returns a function to register a Plugin as given extensionPoints to a given registry.
func RegisterPluginAsExtensions(pluginName string, weight int32, pluginNewFunc framework.PluginFactory, extensions ...string) RegisterPluginFunc {
return func(reg *framework.Registry, plugins *schedulerapi.Plugins, pluginConfigs []schedulerapi.PluginConfig) {
reg.Register(pluginName, pluginNewFunc)
plugins.Score.Enabled = append(plugins.Score.Enabled, schedulerapi.Plugin{Name: pluginName, Weight: weight})
for _, extension := range extensions {
pluginSet := getPluginSetByExtension(plugins, extension)
if pluginSet == nil {
continue
}
pluginSet.Enabled = append(pluginSet.Enabled, schedulerapi.Plugin{Name: pluginName, Weight: weight})
}
//lint:ignore SA4006 this value of pluginConfigs is never used.
//lint:ignore SA4010 this result of append is never used.
pluginConfigs = append(pluginConfigs, schedulerapi.PluginConfig{Name: pluginName})
}
}
func getPluginSetByExtension(plugins *schedulerapi.Plugins, extension string) *schedulerapi.PluginSet {
switch extension {
case "Filter":
return plugins.Filter
case "PreFilter":
return plugins.PreFilter
case "PostFilter":
return plugins.PostFilter
case "Score":
return plugins.Score
case "Bind":
return plugins.Bind
case "Reserve":
return plugins.Reserve
case "Permit":
return plugins.Permit
default:
return nil
}
}