Skip to content

Commit 0e73ac3

Browse files
wangyuqing4wangyuqing4
committed
remove shadowPodgroup in scheduler
1 parent 1a856a1 commit 0e73ac3

File tree

6 files changed

+44
-241
lines changed

6 files changed

+44
-241
lines changed

‎pkg/scheduler/api/pod_group_info.go

+2-2
Original file line numberDiff line numberDiff line change
@@ -44,11 +44,11 @@ const (
4444

4545
//PodGroupVersionV1Alpha2 represents PodGroupVersion of V1Alpha2
4646
PodGroupVersionV1Alpha2 string = "v1alpha2"
47-
// PodPending means the pod group has been accepted by the system, but scheduler can not allocate
47+
// PodGroupPending means the pod group has been accepted by the system, but scheduler can not allocate
4848
// enough resources to it.
4949
PodGroupPending PodGroupPhase = "Pending"
5050

51-
// PodRunning means `spec.minMember` pods of PodGroups has been in running phase.
51+
// PodGroupRunning means `spec.minMember` pods of PodGroups has been in running phase.
5252
PodGroupRunning PodGroupPhase = "Running"
5353

5454
// PodGroupUnknown means part of `spec.minMember` pods are running but the other part can not

‎pkg/scheduler/cache/cache.go

+30-45
Original file line numberDiff line numberDiff line change
@@ -45,7 +45,6 @@ import (
4545
"k8s.io/kubernetes/pkg/scheduler/volumebinder"
4646

4747
"volcano.sh/volcano/cmd/scheduler/app/options"
48-
"volcano.sh/volcano/pkg/apis/scheduling/v1alpha1"
4948
"volcano.sh/volcano/pkg/apis/scheduling/v1alpha2"
5049
kbver "volcano.sh/volcano/pkg/client/clientset/versioned"
5150
"volcano.sh/volcano/pkg/client/clientset/versioned/scheme"
@@ -507,24 +506,9 @@ func (sc *SchedulerCache) Evict(taskInfo *kbapi.TaskInfo, reason string) error {
507506
}
508507
}()
509508

510-
if !shadowPodGroup(job.PodGroup) {
511-
if job.PodGroup.Version == api.PodGroupVersionV1Alpha1 {
512-
pg, err := api.ConvertPodGroupInfoToV1alpha1(job.PodGroup)
513-
if err != nil {
514-
glog.Errorf("Error While converting api.PodGroup to v1alpha.PodGroup with error: %v", err)
515-
return err
516-
}
517-
sc.Recorder.Eventf(pg, v1.EventTypeNormal, "Evict", reason)
518-
} else if job.PodGroup.Version == api.PodGroupVersionV1Alpha2 {
519-
pg, err := api.ConvertPodGroupInfoToV1alpha2(job.PodGroup)
520-
if err != nil {
521-
glog.Errorf("Error While converting api.PodGroup to v2alpha.PodGroup with error: %v", err)
522-
return err
523-
}
524-
sc.Recorder.Eventf(pg, v1.EventTypeNormal, "Evict", reason)
525-
} else {
526-
return fmt.Errorf("Invalid PodGroup Version: %s", job.PodGroup.Version)
527-
}
509+
if err := sc.convertPodGroupInfo(job); err != nil {
510+
glog.Errorf("Error While converting api.PodGroup %v", err)
511+
return err
528512
}
529513

530514
return nil
@@ -778,32 +762,15 @@ func (sc *SchedulerCache) RecordJobStatusEvent(job *kbapi.JobInfo) {
778762
baseErrorMessage = kbapi.AllNodeUnavailableMsg
779763
}
780764

781-
if !shadowPodGroup(job.PodGroup) {
782-
pgUnschedulable := job.PodGroup != nil &&
783-
(job.PodGroup.Status.Phase == api.PodGroupUnknown ||
784-
job.PodGroup.Status.Phase == api.PodGroupPending)
785-
pdbUnschedulabe := job.PDB != nil && len(job.TaskStatusIndex[api.Pending]) != 0
765+
pgUnschedulable := job.PodGroup != nil &&
766+
(job.PodGroup.Status.Phase == api.PodGroupUnknown ||
767+
job.PodGroup.Status.Phase == api.PodGroupPending)
768+
pdbUnschedulabe := job.PDB != nil && len(job.TaskStatusIndex[api.Pending]) != 0
786769

787-
// If pending or unschedulable, record unschedulable event.
788-
if pgUnschedulable || pdbUnschedulabe {
789-
msg := fmt.Sprintf("%v/%v tasks in gang unschedulable: %v", len(job.TaskStatusIndex[api.Pending]), len(job.Tasks), job.FitError())
790-
if job.PodGroup.Version == api.PodGroupVersionV1Alpha1 {
791-
podGroup, err := api.ConvertPodGroupInfoToV1alpha1(job.PodGroup)
792-
if err != nil {
793-
glog.Errorf("Error while converting PodGroup to v1alpha1.PodGroup with error: %v", err)
794-
}
795-
sc.Recorder.Eventf(podGroup, v1.EventTypeWarning,
796-
string(v1alpha1.PodGroupUnschedulableType), msg)
797-
}
798-
799-
if job.PodGroup.Version == api.PodGroupVersionV1Alpha2 {
800-
podGroup, err := api.ConvertPodGroupInfoToV1alpha2(job.PodGroup)
801-
if err != nil {
802-
glog.Errorf("Error while converting PodGroup to v1alpha2.PodGroup with error: %v", err)
803-
}
804-
sc.Recorder.Eventf(podGroup, v1.EventTypeWarning,
805-
string(v1alpha1.PodGroupUnschedulableType), msg)
806-
}
770+
// If pending or unschedulable, record unschedulable event.
771+
if pgUnschedulable || pdbUnschedulabe {
772+
if err := sc.convertPodGroupInfo(job); err != nil {
773+
glog.Errorf("Error While converting api.PodGroup %v", err)
807774
}
808775
}
809776

@@ -825,7 +792,7 @@ func (sc *SchedulerCache) RecordJobStatusEvent(job *kbapi.JobInfo) {
825792

826793
// UpdateJobStatus update the status of job and its tasks.
827794
func (sc *SchedulerCache) UpdateJobStatus(job *kbapi.JobInfo, updatePG bool) (*kbapi.JobInfo, error) {
828-
if updatePG && !shadowPodGroup(job.PodGroup) {
795+
if updatePG {
829796
pg, err := sc.StatusUpdater.UpdatePodGroup(job.PodGroup)
830797
if err != nil {
831798
return nil, err
@@ -837,3 +804,21 @@ func (sc *SchedulerCache) UpdateJobStatus(job *kbapi.JobInfo, updatePG bool) (*k
837804

838805
return job, nil
839806
}
807+
808+
func (sc *SchedulerCache) convertPodGroupInfo(job *kbapi.JobInfo) error {
809+
if job.PodGroup.Version == api.PodGroupVersionV1Alpha1 {
810+
_, err := api.ConvertPodGroupInfoToV1alpha1(job.PodGroup)
811+
if err != nil {
812+
return fmt.Errorf("to v1alpha.PodGroup with error: %v", err)
813+
}
814+
} else if job.PodGroup.Version == api.PodGroupVersionV1Alpha2 {
815+
_, err := api.ConvertPodGroupInfoToV1alpha2(job.PodGroup)
816+
if err != nil {
817+
return fmt.Errorf("to v2alpha.PodGroup with error: %v", err)
818+
}
819+
} else {
820+
return fmt.Errorf("invalid PodGroup Version: %s", job.PodGroup.Version)
821+
}
822+
823+
return nil
824+
}

‎pkg/scheduler/cache/cache_test.go

+1-134
Original file line numberDiff line numberDiff line change
@@ -125,139 +125,6 @@ func buildOwnerReference(owner string) metav1.OwnerReference {
125125
}
126126
}
127127

128-
func TestAddPod(t *testing.T) {
129-
130-
owner := buildOwnerReference("j1")
131-
132-
// case 1:
133-
pod1 := buildPod("c1", "p1", "", v1.PodPending, buildResourceList("1000m", "1G"),
134-
[]metav1.OwnerReference{owner}, make(map[string]string))
135-
pi1 := api.NewTaskInfo(pod1)
136-
pi1.Job = "j1" // The job name is set by cache.
137-
pod2 := buildPod("c1", "p2", "n1", v1.PodRunning, buildResourceList("1000m", "1G"),
138-
[]metav1.OwnerReference{owner}, make(map[string]string))
139-
pi2 := api.NewTaskInfo(pod2)
140-
pi2.Job = "j1" // The job name is set by cache.
141-
142-
j1 := api.NewJobInfo(api.JobID("j1"), pi1, pi2)
143-
pg := createShadowPodGroup(pod1)
144-
j1.SetPodGroup(pg)
145-
146-
node1 := buildNode("n1", buildResourceList("2000m", "10G"))
147-
ni1 := api.NewNodeInfo(node1)
148-
ni1.AddTask(pi2)
149-
150-
tests := []struct {
151-
pods []*v1.Pod
152-
nodes []*v1.Node
153-
expected *SchedulerCache
154-
}{
155-
{
156-
pods: []*v1.Pod{pod1, pod2},
157-
nodes: []*v1.Node{node1},
158-
expected: &SchedulerCache{
159-
Nodes: map[string]*api.NodeInfo{
160-
"n1": ni1,
161-
},
162-
Jobs: map[api.JobID]*api.JobInfo{
163-
"j1": j1,
164-
},
165-
},
166-
},
167-
}
168-
169-
for i, test := range tests {
170-
cache := &SchedulerCache{
171-
Jobs: make(map[api.JobID]*api.JobInfo),
172-
Nodes: make(map[string]*api.NodeInfo),
173-
}
174-
175-
for _, n := range test.nodes {
176-
cache.AddNode(n)
177-
}
178-
179-
for _, p := range test.pods {
180-
cache.AddPod(p)
181-
}
182-
183-
if !cacheEqual(cache, test.expected) {
184-
t.Errorf("case %d: \n expected %v, \n got %v \n",
185-
i, test.expected, cache)
186-
}
187-
}
188-
}
189-
190-
func TestAddNode(t *testing.T) {
191-
owner1 := buildOwnerReference("j1")
192-
owner2 := buildOwnerReference("j2")
193-
194-
// case 1
195-
node1 := buildNode("n1", buildResourceList("2000m", "10G"))
196-
pod1 := buildPod("c1", "p1", "", v1.PodPending, buildResourceList("1000m", "1G"),
197-
[]metav1.OwnerReference{owner1}, make(map[string]string))
198-
pi1 := api.NewTaskInfo(pod1)
199-
pi1.Job = "j1" // The job name is set by cache.
200-
201-
pod2 := buildPod("c1", "p2", "n1", v1.PodRunning, buildResourceList("1000m", "1G"),
202-
[]metav1.OwnerReference{owner2}, make(map[string]string))
203-
pi2 := api.NewTaskInfo(pod2)
204-
pi2.Job = "j2" // The job name is set by cache.
205-
206-
ni1 := api.NewNodeInfo(node1)
207-
ni1.AddTask(pi2)
208-
209-
j1 := api.NewJobInfo("j1")
210-
pg1 := createShadowPodGroup(pod1)
211-
j1.SetPodGroup(pg1)
212-
213-
j2 := api.NewJobInfo("j2")
214-
pg2 := createShadowPodGroup(pod2)
215-
j2.SetPodGroup(pg2)
216-
217-
j1.AddTaskInfo(pi1)
218-
j2.AddTaskInfo(pi2)
219-
220-
tests := []struct {
221-
pods []*v1.Pod
222-
nodes []*v1.Node
223-
expected *SchedulerCache
224-
}{
225-
{
226-
pods: []*v1.Pod{pod1, pod2},
227-
nodes: []*v1.Node{node1},
228-
expected: &SchedulerCache{
229-
Nodes: map[string]*api.NodeInfo{
230-
"n1": ni1,
231-
},
232-
Jobs: map[api.JobID]*api.JobInfo{
233-
"j1": j1,
234-
"j2": j2,
235-
},
236-
},
237-
},
238-
}
239-
240-
for i, test := range tests {
241-
cache := &SchedulerCache{
242-
Nodes: make(map[string]*api.NodeInfo),
243-
Jobs: make(map[api.JobID]*api.JobInfo),
244-
}
245-
246-
for _, p := range test.pods {
247-
cache.AddPod(p)
248-
}
249-
250-
for _, n := range test.nodes {
251-
cache.AddNode(n)
252-
}
253-
254-
if !cacheEqual(cache, test.expected) {
255-
t.Errorf("case %d: \n expected %v, \n got %v \n",
256-
i, test.expected, cache)
257-
}
258-
}
259-
}
260-
261128
func TestGetOrCreateJob(t *testing.T) {
262129
owner1 := buildOwnerReference("j1")
263130
owner2 := buildOwnerReference("j2")
@@ -292,7 +159,7 @@ func TestGetOrCreateJob(t *testing.T) {
292159
},
293160
{
294161
task: pi2,
295-
gotJob: true,
162+
gotJob: false,
296163
},
297164
{
298165
task: pi3,

‎pkg/scheduler/cache/event_handlers.go

+4-15
Original file line numberDiff line numberDiff line change
@@ -46,23 +46,12 @@ func (sc *SchedulerCache) getOrCreateJob(pi *kbapi.TaskInfo) *kbapi.JobInfo {
4646
if pi.Pod.Spec.SchedulerName != sc.schedulerName {
4747
glog.V(4).Infof("Pod %s/%s will not not scheduled by %s, skip creating PodGroup and Job for it",
4848
pi.Pod.Namespace, pi.Pod.Name, sc.schedulerName)
49-
return nil
5049
}
51-
pb := createShadowPodGroup(pi.Pod)
52-
pi.Job = kbapi.JobID(pb.Name)
53-
54-
if _, found := sc.Jobs[pi.Job]; !found {
55-
job := kbapi.NewJobInfo(pi.Job)
56-
job.SetPodGroup(pb)
57-
// Set default queue for shadow podgroup.
58-
job.Queue = kbapi.QueueID(sc.defaultQueue)
50+
return nil
51+
}
5952

60-
sc.Jobs[pi.Job] = job
61-
}
62-
} else {
63-
if _, found := sc.Jobs[pi.Job]; !found {
64-
sc.Jobs[pi.Job] = kbapi.NewJobInfo(pi.Job)
65-
}
53+
if _, found := sc.Jobs[pi.Job]; !found {
54+
sc.Jobs[pi.Job] = kbapi.NewJobInfo(pi.Job)
6655
}
6756

6857
return sc.Jobs[pi.Job]

‎pkg/scheduler/cache/util.go

+1-39
Original file line numberDiff line numberDiff line change
@@ -17,47 +17,9 @@ limitations under the License.
1717
package cache
1818

1919
import (
20-
v1 "k8s.io/api/core/v1"
21-
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
22-
23-
"volcano.sh/volcano/pkg/apis/utils"
24-
"volcano.sh/volcano/pkg/scheduler/api"
25-
)
26-
27-
const (
28-
shadowPodGroupKey = "volcano/shadow-pod-group"
20+
"k8s.io/api/core/v1"
2921
)
3022

31-
func shadowPodGroup(pg *api.PodGroup) bool {
32-
if pg == nil {
33-
return true
34-
}
35-
36-
_, found := pg.Annotations[shadowPodGroupKey]
37-
38-
return found
39-
}
40-
41-
func createShadowPodGroup(pod *v1.Pod) *api.PodGroup {
42-
jobID := api.JobID(utils.GetController(pod))
43-
if len(jobID) == 0 {
44-
jobID = api.JobID(pod.UID)
45-
}
46-
47-
return &api.PodGroup{
48-
ObjectMeta: metav1.ObjectMeta{
49-
Namespace: pod.Namespace,
50-
Name: string(jobID),
51-
Annotations: map[string]string{
52-
shadowPodGroupKey: string(jobID),
53-
},
54-
},
55-
Spec: api.PodGroupSpec{
56-
MinMember: 1,
57-
},
58-
}
59-
}
60-
6123
// responsibleForPod returns true if the pod has asked to be scheduled by the given scheduler.
6224
func responsibleForPod(pod *v1.Pod, schedulerName string) bool {
6325
return schedulerName == pod.Spec.SchedulerName

0 commit comments

Comments
 (0)