Distinguish different pod-delete scenario

Try to address issue volcano-sh#791
It's a draft solution, need further discussion.

Signed-off-by: pengli <justdoit.pli@gmail.com>

pkg/controllers/cache/cache.go

@@ -59,7 +59,15 @@ func JobKey(job *v1alpha1.Job) string {
 }
 
 func jobTerminated(job *apis.JobInfo) bool {
-	return job.Job == nil && len(job.Pods) == 0
+	for _, pods := range job.Pods {
+		for _, pod := range pods {
+			if pod.Status.Phase != v1.PodSucceeded {
+				return false
+			}
+		}
+	}
+
+	return job.Job == nil
 }
 
 func jobKeyOfPod(pod *v1.Pod) (string, error) {

pkg/controllers/job/job_controller_actions.go

@@ -65,9 +65,13 @@ func (cc *Controller) killJob(jobInfo *apis.JobInfo, podRetainPhase state.PhaseM
 			_, retain := podRetainPhase[pod.Status.Phase]
 
 			if !retain {
-				err := cc.deleteJobPod(job.Name, pod)
+				existing, err := cc.deleteJobPod(job.Name, pod)
 				if err == nil {
-					terminating++
+					if existing {
+						terminating++
+					} else {
+						cc.cache.DeletePod(pod)
+					}
 					continue
 				}
 				// record the err, and then collect the pod info like retained pod
@@ -333,7 +337,7 @@ func (cc *Controller) syncJob(jobInfo *apis.JobInfo, updateStatus state.UpdateSt
 	for _, pod := range podToDelete {
 		go func(pod *v1.Pod) {
 			defer waitDeletionGroup.Done()
-			err := cc.deleteJobPod(job.Name, pod)
+			_, err := cc.deleteJobPod(job.Name, pod)
 			if err != nil {
 				// Failed to delete Pod, waitCreationGroup a moment and then create it again
 				// This is to ensure all podsMap under the same Job created
@@ -530,16 +534,19 @@ func (cc *Controller) createOrUpdatePodGroup(job *batch.Job) error {
 	return nil
 }
 
-func (cc *Controller) deleteJobPod(jobName string, pod *v1.Pod) error {
+func (cc *Controller) deleteJobPod(jobName string, pod *v1.Pod) (bool, error) {
 	err := cc.kubeClient.CoreV1().Pods(pod.Namespace).Delete(pod.Name, nil)
-	if err != nil && !apierrors.IsNotFound(err) {
+	if err != nil {
+		if apierrors.IsNotFound(err) {
+			return false, nil
+		}
 		klog.Errorf("Failed to delete pod %s/%s for Job %s, err %#v",
 			pod.Namespace, pod.Name, jobName, err)
 
-		return fmt.Errorf("failed to delete pod %s, err %#v", pod.Name, err)
+		return true, fmt.Errorf("failed to delete pod %s, err %#v", pod.Name, err)
 	}
 
-	return nil
+	return true, nil
 }
 
 func (cc *Controller) calcPGMinResources(job *batch.Job) *v1.ResourceList {

pkg/controllers/job/job_controller_actions_test.go

@@ -461,7 +461,7 @@ func TestDeleteJobPod(t *testing.T) {
 				}
 			}
 
-			err := fakeController.deleteJobPod(testcase.Job.Name, testcase.DeletePod)
+			_, err := fakeController.deleteJobPod(testcase.Job.Name, testcase.DeletePod)
 			if err != testcase.ExpextVal {
 				t.Errorf("Expected return value to be equal to expected: %s, but got: %s", testcase.ExpextVal, err)
 			}

pkg/controllers/job/job_controller_handler.go

@@ -324,20 +324,23 @@ func (cc *Controller) deletePod(obj interface{}) {
 		return
 	}
 
-	req := apis.Request{
-		Namespace: pod.Namespace,
-		JobName:   jobName,
-		TaskName:  taskName,
-
-		Event:      bus.PodEvictedEvent,
-		JobVersion: int32(dVersion),
+	if pod.Status.Phase == v1.PodSucceeded {
+		return
 	}
 
 	if err := cc.cache.DeletePod(pod); err != nil {
 		klog.Errorf("Failed to delete Pod <%s/%s>: %v in cache",
 			pod.Namespace, pod.Name, err)
 	}
 
+	req := apis.Request{
+		Namespace:  pod.Namespace,
+		JobName:    jobName,
+		TaskName:   taskName,
+		Event:      bus.PodEvictedEvent,
+		JobVersion: int32(dVersion),
+	}
+
 	key := jobhelpers.GetJobKeyByReq(&req)
 	queue := cc.getWorkerQueue(key)
 	queue.Add(req)