Merge branch 'opt/opt_task_related' into 'master'

Fix infinite loop when JobInfo incomplete

CloseBug volcano-sh#39 by accepting updating existing JobInfo object in cache when Job attribute is empty. Also add some useful log output.

Issues info:
Issue ID: 39
Title: Infinite loop in vk-controller when pod event comes ahead of job event.
Issue url: CBU-PaaS/Community/volcano/volcano#39


See merge request CBU-PaaS/Community/volcano/volcano!26

Author: mada 00483107

pkg/controllers/job/apis/types.go

@@ -17,6 +17,7 @@ limitations under the License.
 package apis
 
 import (
+	"fmt"
 	"k8s.io/api/core/v1"
 
 	vkbatchv1 "hpw.cloud/volcano/pkg/apis/batch/v1alpha1"
@@ -35,3 +36,9 @@ type Request struct {
 	Event  vkbatchv1.Event
 	Action vkbatchv1.Action
 }
+
+func (r Request) String() string {
+	return fmt.Sprintf(
+		"Job: %s/%s, Task:%s, Event:%s, Action:%s",
+		r.Namespace, r.JobName, r.TaskName, r.Event, r.Action)
+}

pkg/controllers/job/cache/cache.go

@@ -116,8 +116,14 @@ func (jc *jobCache) Add(obj *v1alpha1.Job) error {
 	defer jc.Unlock()
 
 	key := JobKey(obj)
-	if _, found := jc.jobs[key]; found {
-		return fmt.Errorf("duplicated job <%v>", key)
+	if job, found := jc.jobs[key]; found {
+		// JobInfo was created by pod event previously,
+		// therefore we will have an empty Job attribute.
+		if job.Job == nil {
+			job.Job = obj
+		} else {
+			return fmt.Errorf("duplicated job <%v>", key)
+		}
 	}
 
 	jc.jobs[key] = &apis.JobInfo{

pkg/controllers/job/job_controller.go

@@ -218,7 +218,9 @@ func (cc *Controller) worker() {
 	}
 
 	if jobInfo.Job == nil {
-		glog.V(3).Infof("Cache is out of sync for <%v>, retry it.", req)
+		glog.V(3).Infof(
+			"Cache is incomplete for request <%v>, will retry it later.",
+			req)
 		cc.queue.AddRateLimited(req)
 		return
 	}

pkg/controllers/job/job_controller_handler.go

@@ -74,7 +74,7 @@ func (cc *Controller) addJob(obj interface{}) {
 
 	// TODO(k82cn): if failed to add job, the cache should be refresh
 	if err := cc.cache.Add(job); err != nil {
-		glog.Errorf("Failed to add job <%s/%s>: %v",
+		glog.Errorf("Failed to add job <%s/%s>: %v in cache",
 			job.Namespace, job.Name, err)
 	}
 	cc.queue.Add(req)
@@ -95,7 +95,7 @@ func (cc *Controller) updateJob(oldObj, newObj interface{}) {
 	}
 
 	if err := cc.cache.Update(newJob); err != nil {
-		glog.Errorf("Failed to update job <%s/%s>: %v",
+		glog.Errorf("Failed to update job <%s/%s>: %v in cache",
 			newJob.Namespace, newJob.Name, err)
 	}
 
@@ -110,7 +110,7 @@ func (cc *Controller) deleteJob(obj interface{}) {
 	}
 
 	if err := cc.cache.Delete(job); err != nil {
-		glog.Errorf("Failed to delete job <%s/%s>: %v",
+		glog.Errorf("Failed to delete job <%s/%s>: %v in cache",
 			job.Namespace, job.Name, err)
 	}
 }
@@ -124,7 +124,7 @@ func (cc *Controller) addPod(obj interface{}) {
 
 	jobName, found := pod.Annotations[vkbatchv1.JobNameKey]
 	if !found {
-		glog.Errorf("Failed to find jobName of Pod <%s/%s>",
+		glog.Infof("Failed to find jobName of Pod <%s/%s>, skipping",
 			pod.Namespace, pod.Name)
 		return
 	}
@@ -137,7 +137,7 @@ func (cc *Controller) addPod(obj interface{}) {
 	}
 
 	if err := cc.cache.AddPod(pod); err != nil {
-		glog.Errorf("Failed to add Pod <%s/%s>: %v",
+		glog.Errorf("Failed to add Pod <%s/%s>: %v to cache",
 			pod.Namespace, pod.Name, err)
 	}
 	cc.queue.Add(req)
@@ -158,14 +158,14 @@ func (cc *Controller) updatePod(oldObj, newObj interface{}) {
 
 	taskName, found := newPod.Annotations[vkbatchv1.TaskSpecKey]
 	if !found {
-		glog.Errorf("Failed to find taskName of Pod <%s/%s>",
+		glog.Infof("Failed to find taskName of Pod <%s/%s>, skipping",
 			newPod.Namespace, newPod.Name)
 		return
 	}
 
 	jobName, found := newPod.Annotations[vkbatchv1.JobNameKey]
 	if !found {
-		glog.Errorf("Failed to find jobName of Pod <%s/%s>",
+		glog.Infof("Failed to find jobName of Pod <%s/%s>, skipping",
 			newPod.Namespace, newPod.Name)
 		return
 	}
@@ -185,7 +185,7 @@ func (cc *Controller) updatePod(oldObj, newObj interface{}) {
 	}
 
 	if err := cc.cache.UpdatePod(newPod); err != nil {
-		glog.Errorf("Failed to update Pod <%s/%s>: %v",
+		glog.Errorf("Failed to update Pod <%s/%s>: %v in cache",
 			newPod.Namespace, newPod.Name, err)
 	}
 
@@ -211,14 +211,14 @@ func (cc *Controller) deletePod(obj interface{}) {
 
 	taskName, found := pod.Annotations[vkbatchv1.TaskSpecKey]
 	if !found {
-		glog.Errorf("Failed to find taskName of Pod <%s/%s>",
+		glog.Infof("Failed to find taskName of Pod <%s/%s>, skipping",
 			pod.Namespace, pod.Name)
 		return
 	}
 
 	jobName, found := pod.Annotations[vkbatchv1.JobNameKey]
 	if !found {
-		glog.Errorf("Failed to find jobName of Pod <%s/%s>",
+		glog.Infof("Failed to find jobName of Pod <%s/%s>, skipping",
 			pod.Namespace, pod.Name)
 		return
 	}
@@ -232,7 +232,7 @@ func (cc *Controller) deletePod(obj interface{}) {
 	}
 
 	if err := cc.cache.DeletePod(pod); err != nil {
-		glog.Errorf("Failed to update Pod <%s/%s>: %v",
+		glog.Errorf("Failed to update Pod <%s/%s>: %v in cache",
 			pod.Namespace, pod.Name, err)
 	}