Merge pull request volcano-sh#289 from k82cn/ka_288

Deleted Job when PDB deleted.

pkg/scheduler/api/helpers.go

@@ -98,5 +98,7 @@ func MergeErrors(errs ...error) error {
 
 // JobTerminated checkes whether job was terminated.
 func JobTerminated(job *JobInfo) bool {
-	return job.SchedSpec == nil && len(job.Tasks) == 0
+	return job.SchedSpec == nil &&
+		job.PDB == nil &&
+		len(job.Tasks) == 0
 }

pkg/scheduler/api/job_info.go

@@ -162,6 +162,10 @@ func (ps *JobInfo) SetPDB(pbd *policyv1.PodDisruptionBudget) {
 	ps.PDB = pbd
 }
 
+func (ps *JobInfo) UnsetPDB() {
+	ps.PDB = nil
+}
+
 func (ps *JobInfo) GetTasks(statuses ...TaskStatus) []*TaskInfo {
 	var res []*TaskInfo
 

pkg/scheduler/cache/event_handlers.go

@@ -362,7 +362,6 @@ func (sc *SchedulerCache) deleteSchedulingSpec(ss *arbv1.SchedulingSpec) error {
 	// Unset SchedulingSpec
 	job.UnsetSchedulingSpec()
 
-	// TODO (k82cn): find another way to clean up Job.
 	sc.deleteJob(job)
 
 	return nil
@@ -456,12 +455,24 @@ func (sc *SchedulerCache) setPDB(pdb *policyv1.PodDisruptionBudget) error {
 }
 
 // Assumes that lock is already acquired.
-func (sc *SchedulerCache) updatePDB(oldQueue, newQueue *policyv1.PodDisruptionBudget) error {
-	return sc.setPDB(newQueue)
+func (sc *SchedulerCache) updatePDB(oldPDB, newPDB *policyv1.PodDisruptionBudget) error {
+	return sc.setPDB(newPDB)
 }
 
 // Assumes that lock is already acquired.
-func (sc *SchedulerCache) deletePDB(queue *policyv1.PodDisruptionBudget) error {
+func (sc *SchedulerCache) deletePDB(pdb *policyv1.PodDisruptionBudget) error {
+	jobID := arbapi.JobID(utils.GetController(pdb))
+
+	job, found := sc.Jobs[jobID]
+	if !found {
+		return fmt.Errorf("can not found job %v:%v/%v", jobID, pdb.Namespace, pdb.Name)
+	}
+
+	// Unset SchedulingSpec
+	job.UnsetPDB()
+
+	sc.deleteJob(job)
+
 	return nil
 }