Skip to content

Commit

Permalink
Added conformance plugin.
Browse files Browse the repository at this point in the history 
Signed-off-by: Da K. Ma <[email protected]>
  • Loading branch information
@k82cn
k82cn committed Mar 1, 2019
1 parent 1de56f8 commit ca72c85
Show file tree
Hide file tree
Showing 8 changed files with 219 additions and 100 deletions.
16 changes: 9 additions & 7 deletions pkg/scheduler/actions/reclaim/reclaim.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -174,15 +174,17 @@ func (alloc *reclaimAction) Execute(ssn *framework.Session) {
glog.V(3).Infof("Reclaimed <%v> for task <%s/%s> requested <%v>.",
reclaimed, task.Namespace, task.Name, task.Resreq)

if err := ssn.Pipeline(task, n.Name); err != nil {
glog.Errorf("Failed to pipline Task <%s/%s> on Node <%s>",
task.Namespace, task.Name, n.Name)
}
if task.Resreq.LessEqual(reclaimed) {
if err := ssn.Pipeline(task, n.Name); err != nil {
glog.Errorf("Failed to pipline Task <%s/%s> on Node <%s>",
task.Namespace, task.Name, n.Name)
}

// Ignore error of pipeline, will be corrected in next scheduling loop.
assigned = true
// Ignore error of pipeline, will be corrected in next scheduling loop.
assigned = true

break
break
}
}

if assigned {
Expand Down
60 changes: 60 additions & 0 deletions pkg/scheduler/plugins/conformance/conformance.go
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,60 @@
/*
Copyright 2018 The Kubernetes Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package conformance

import (
"github.com/kubernetes-sigs/kube-batch/pkg/scheduler/api"
"github.com/kubernetes-sigs/kube-batch/pkg/scheduler/framework"
"k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/kubernetes/pkg/apis/scheduling"
)

type conformancePlugin struct {
}

func New() framework.Plugin {
return &conformancePlugin{}
}

func (pp *conformancePlugin) Name() string {
return "conformance"
}

func (pp *conformancePlugin) OnSessionOpen(ssn *framework.Session) {
evictableFn := func(evictor *api.TaskInfo, evictees []*api.TaskInfo) []*api.TaskInfo {
var victims []*api.TaskInfo

for _, evictee := range evictees {
className := evictee.Pod.Spec.PriorityClassName
// Skip critical pod.
if className == scheduling.SystemClusterCritical ||
className == scheduling.SystemNodeCritical ||
evictee.Namespace == v1.NamespaceSystem {
continue
}

victims = append(victims, evictee)
}

return victims
}

ssn.AddPreemptableFn(pp.Name(), evictableFn)
ssn.AddReclaimableFn(pp.Name(), evictableFn)
}

func (pp *conformancePlugin) OnSessionClose(ssn *framework.Session) {}
8 changes: 4 additions & 4 deletions pkg/scheduler/plugins/prioritize/prioritize.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -183,28 +183,28 @@ func (pp *prioritizePlugin) OnSessionOpen(ssn *framework.Session) {

host, err := priorities.LeastRequestedPriorityMap(task.Pod, nil, nodeInfo)
if err != nil {
glog.V(3).Infof("Least Requested Priority Failed because of Error: %v", err)
glog.Warningf("Least Requested Priority Failed because of Error: %v", err)
return 0, err
}
score = score + host.Score

host, err = priorities.CalculateNodeAffinityPriorityMap(task.Pod, nil, nodeInfo)
if err != nil {
glog.V(3).Infof("Calculate Node Affinity Priority Failed because of Error: %v", err)
glog.Warningf("Calculate Node Affinity Priority Failed because of Error: %v", err)
return 0, err
}
score = score + host.Score

mapFn := priorities.NewInterPodAffinityPriority(cn, nl, pl, v1.DefaultHardPodAffinitySymmetricWeight)
interPodAffinityScore, err = mapFn(task.Pod, nodeMap, nodeSlice)
if err != nil {
glog.V(3).Infof("Calculate Inter Pod Affinity Priority Failed because of Error: %v", err)
glog.Warningf("Calculate Inter Pod Affinity Priority Failed because of Error: %v", err)
return 0, err
}
hostScore := getInterPodAffinityScore(node.Name, interPodAffinityScore)
score = score + hostScore

glog.V(3).Infof("Total Score for that node is: %f", score)
glog.V(4).Infof("Total Score for that node is: %d", score)
return score, nil
}
ssn.AddPriorityFn(pp.Name(), priorityFn)
Expand Down
60 changes: 30 additions & 30 deletions test/e2e/job.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ var _ = Describe("Job E2E Test", func() {
})

err := waitPodGroupReady(context, pg)
Expect(err).NotTo(HaveOccurred())
checkError(context, err)
})

It("Schedule Multiple Jobs", func() {
Expand Down Expand Up @@ -70,13 +70,13 @@ var _ = Describe("Job E2E Test", func() {
_, pg3 := createJobEx(context, job)

err := waitPodGroupReady(context, pg1)
Expect(err).NotTo(HaveOccurred())
checkError(context, err)

err = waitPodGroupReady(context, pg2)
Expect(err).NotTo(HaveOccurred())
checkError(context, err)

err = waitPodGroupReady(context, pg3)
Expect(err).NotTo(HaveOccurred())
checkError(context, err)
})

It("Gang scheduling", func() {
Expand All @@ -86,7 +86,7 @@ var _ = Describe("Job E2E Test", func() {

replicaset := createReplicaSet(context, "rs-1", rep, "nginx", oneCPU)
err := waitReplicaSetReady(context, replicaset.Name)
Expect(err).NotTo(HaveOccurred())
checkError(context, err)

job := &jobSpec{
name: "gang-qj",
Expand All @@ -103,16 +103,16 @@ var _ = Describe("Job E2E Test", func() {

_, pg := createJobEx(context, job)
err = waitPodGroupPending(context, pg)
Expect(err).NotTo(HaveOccurred())
checkError(context, err)

err = waitPodGroupUnschedulable(context, pg)
Expect(err).NotTo(HaveOccurred())
checkError(context, err)

err = deleteReplicaSet(context, replicaset.Name)
Expect(err).NotTo(HaveOccurred())
checkError(context, err)

err = waitPodGroupReady(context, pg)
Expect(err).NotTo(HaveOccurred())
checkError(context, err)
})

It("Gang scheduling: Full Occupied", func() {
Expand All @@ -135,15 +135,15 @@ var _ = Describe("Job E2E Test", func() {
job.name = "gang-fq-qj1"
_, pg1 := createJobEx(context, job)
err := waitPodGroupReady(context, pg1)
Expect(err).NotTo(HaveOccurred())
checkError(context, err)

job.name = "gang-fq-qj2"
_, pg2 := createJobEx(context, job)
err = waitPodGroupPending(context, pg2)
Expect(err).NotTo(HaveOccurred())
checkError(context, err)

err = waitPodGroupReady(context, pg1)
Expect(err).NotTo(HaveOccurred())
checkError(context, err)
})

It("Preemption", func() {
Expand All @@ -167,15 +167,15 @@ var _ = Describe("Job E2E Test", func() {
job.name = "preemptee-qj"
_, pg1 := createJobEx(context, job)
err := waitTasksReady(context, pg1, int(rep))
Expect(err).NotTo(HaveOccurred())
checkError(context, err)

job.name = "preemptor-qj"
_, pg2 := createJobEx(context, job)
err = waitTasksReady(context, pg1, int(rep)/2)
Expect(err).NotTo(HaveOccurred())
checkError(context, err)

err = waitTasksReady(context, pg2, int(rep)/2)
Expect(err).NotTo(HaveOccurred())
checkError(context, err)
})

It("Multiple Preemption", func() {
Expand All @@ -199,24 +199,24 @@ var _ = Describe("Job E2E Test", func() {
job.name = "preemptee-qj"
_, pg1 := createJobEx(context, job)
err := waitTasksReady(context, pg1, int(rep))
Expect(err).NotTo(HaveOccurred())
checkError(context, err)

job.name = "preemptor-qj1"
_, pg2 := createJobEx(context, job)
Expect(err).NotTo(HaveOccurred())
checkError(context, err)

job.name = "preemptor-qj2"
_, pg3 := createJobEx(context, job)
Expect(err).NotTo(HaveOccurred())
checkError(context, err)

err = waitTasksReady(context, pg1, int(rep)/3)
Expect(err).NotTo(HaveOccurred())
checkError(context, err)

err = waitTasksReady(context, pg2, int(rep)/3)
Expect(err).NotTo(HaveOccurred())
checkError(context, err)

err = waitTasksReady(context, pg3, int(rep)/3)
Expect(err).NotTo(HaveOccurred())
checkError(context, err)
})

It("Schedule BestEffort Job", func() {
Expand Down Expand Up @@ -246,7 +246,7 @@ var _ = Describe("Job E2E Test", func() {
_, pg := createJobEx(context, job)

err := waitPodGroupReady(context, pg)
Expect(err).NotTo(HaveOccurred())
checkError(context, err)
})

It("Statement", func() {
Expand All @@ -271,18 +271,18 @@ var _ = Describe("Job E2E Test", func() {
job.name = "st-qj-1"
_, pg1 := createJobEx(context, job)
err := waitPodGroupReady(context, pg1)
Expect(err).NotTo(HaveOccurred())
checkError(context, err)

now := time.Now()

job.name = "st-qj-2"
_, pg2 := createJobEx(context, job)
err = waitPodGroupUnschedulable(context, pg2)
Expect(err).NotTo(HaveOccurred())
checkError(context, err)

// No preemption event
evicted, err := podGroupEvicted(context, pg1, now)()
Expect(err).NotTo(HaveOccurred())
checkError(context, err)
Expect(evicted).NotTo(BeTrue())
})

Expand All @@ -295,7 +295,7 @@ var _ = Describe("Job E2E Test", func() {

replicaset := createReplicaSet(context, "rs-1", rep/2, "nginx", slot)
err := waitReplicaSetReady(context, replicaset.Name)
Expect(err).NotTo(HaveOccurred())
checkError(context, err)

_, pg := createJobEx(context, &jobSpec{
name: "multi-pod-job",
Expand Down Expand Up @@ -323,7 +323,7 @@ var _ = Describe("Job E2E Test", func() {
}

err = waitTasksReadyEx(context, pg, expteced)
Expect(err).NotTo(HaveOccurred())
checkError(context, err)
})

It("Try to fit unassigned task with different resource requests in one loop", func() {
Expand All @@ -336,7 +336,7 @@ var _ = Describe("Job E2E Test", func() {

replicaset := createReplicaSet(context, "rs-1", rep-1, "nginx", slot)
err := waitReplicaSetReady(context, replicaset.Name)
Expect(err).NotTo(HaveOccurred())
checkError(context, err)

_, pg := createJobEx(context, &jobSpec{
name: "multi-task-diff-resource-job",
Expand All @@ -360,10 +360,10 @@ var _ = Describe("Job E2E Test", func() {
})

err = waitPodGroupPending(context, pg)
Expect(err).NotTo(HaveOccurred())
checkError(context, err)

// task_1 has been scheduled
err = waitTasksReady(context, pg, int(minMemberOverride))
Expect(err).NotTo(HaveOccurred())
checkError(context, err)
})
})
18 changes: 9 additions & 9 deletions test/e2e/predicates.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ import (
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"

v1 "k8s.io/api/core/v1"
"k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
schedulerapi "k8s.io/kubernetes/pkg/scheduler/api"
)
Expand Down Expand Up @@ -67,7 +67,7 @@ var _ = Describe("Predicates E2E Test", func() {

_, pg := createJobEx(context, job)
err := waitPodGroupReady(context, pg)
Expect(err).NotTo(HaveOccurred())
checkError(context, err)

pods := getPodOfPodGroup(context, pg)
for _, pod := range pods {
Expand Down Expand Up @@ -97,10 +97,10 @@ var _ = Describe("Predicates E2E Test", func() {
_, pg := createJobEx(context, job)

err := waitTasksReady(context, pg, nn)
Expect(err).NotTo(HaveOccurred())
checkError(context, err)

err = waitTasksPending(context, pg, nn)
Expect(err).NotTo(HaveOccurred())
checkError(context, err)
})

It("Pod Affinity", func() {
Expand Down Expand Up @@ -142,7 +142,7 @@ var _ = Describe("Predicates E2E Test", func() {

_, pg := createJobEx(context, job)
err := waitPodGroupReady(context, pg)
Expect(err).NotTo(HaveOccurred())
checkError(context, err)

pods := getPodOfPodGroup(context, pg)
// All pods should be scheduled to the same node.
Expand All @@ -165,7 +165,7 @@ var _ = Describe("Predicates E2E Test", func() {
}

err := taintAllNodes(context, taints)
Expect(err).NotTo(HaveOccurred())
checkError(context, err)

job := &jobSpec{
name: "tt-job",
Expand All @@ -181,13 +181,13 @@ var _ = Describe("Predicates E2E Test", func() {

_, pg := createJobEx(context, job)
err = waitPodGroupPending(context, pg)
Expect(err).NotTo(HaveOccurred())
checkError(context, err)

err = removeTaintsFromAllNodes(context, taints)
Expect(err).NotTo(HaveOccurred())
checkError(context, err)

err = waitPodGroupReady(context, pg)
Expect(err).NotTo(HaveOccurred())
checkError(context, err)
})

})
Loading

0 comments on commit ca72c85

Please sign in to comment.