Merge pull request #330 from volcano-sh/job-log

omit pods not controlled by volcano
volcano-sh · Jul 15, 2019 · a0a4246 · a0a4246
2 parents afb0007 + 7f7eb55
commit a0a4246
Show file tree

Hide file tree

Showing 3 changed files with 57 additions and 9 deletions.
diff --git a/pkg/controllers/job/job_controller_handler.go b/pkg/controllers/job/job_controller_handler.go
@@ -32,6 +32,7 @@ import (
 	vkbusv1 "volcano.sh/volcano/pkg/apis/bus/v1alpha1"
 	kbtype "volcano.sh/volcano/pkg/apis/scheduling/v1alpha1"
 
+	"volcano.sh/volcano/pkg/apis/helpers"
 	"volcano.sh/volcano/pkg/controllers/apis"
 	vkcache "volcano.sh/volcano/pkg/controllers/cache"
 	vkjobhelpers "volcano.sh/volcano/pkg/controllers/job/helpers"
@@ -136,14 +137,22 @@ func (cc *Controller) addPod(obj interface{}) {
 		glog.Errorf("Failed to convert %v to v1.Pod", obj)
 		return
 	}
+	// Filter out pods that are not created from volcano job
+	if !isControlledBy(pod, helpers.JobKind) {
+		return
+	}
 
 	jobName, found := pod.Annotations[vkbatchv1.JobNameKey]
 	if !found {
+		glog.Infof("Failed to find jobName of Pod <%s/%s>, skipping",
+			pod.Namespace, pod.Name)
 		return
 	}
 
 	version, found := pod.Annotations[vkbatchv1.JobVersion]
 	if !found {
+		glog.Infof("Failed to find jobVersion of Pod <%s/%s>, skipping",
+			pod.Namespace, pod.Name)
 		return
 	}
 
@@ -189,8 +198,24 @@ func (cc *Controller) updatePod(oldObj, newObj interface{}) {
 		return
 	}
 
+	// Filter out pods that are not created from volcano job
+	if !isControlledBy(newPod, helpers.JobKind) {
+		return
+	}
+
+	if newPod.ResourceVersion == oldPod.ResourceVersion {
+		return
+	}
+
+	if newPod.DeletionTimestamp != nil {
+		cc.deletePod(newObj)
+		return
+	}
+
 	taskName, found := newPod.Annotations[vkbatchv1.TaskSpecKey]
 	if !found {
+		glog.Infof("Failed to find taskName of Pod <%s/%s>, skipping",
+			newPod.Namespace, newPod.Name)
 		return
 	}
 
@@ -203,6 +228,8 @@ func (cc *Controller) updatePod(oldObj, newObj interface{}) {
 
 	version, found := newPod.Annotations[vkbatchv1.JobVersion]
 	if !found {
+		glog.Infof("Failed to find jobVersion of Pod <%s/%s>, skipping",
+			newPod.Namespace, newPod.Name)
 		return
 	}
 
@@ -213,15 +240,6 @@ func (cc *Controller) updatePod(oldObj, newObj interface{}) {
 		return
 	}
 
-	if newPod.ResourceVersion == oldPod.ResourceVersion {
-		return
-	}
-
-	if newPod.DeletionTimestamp != nil {
-		cc.deletePod(newObj)
-		return
-	}
-
 	if err := cc.cache.UpdatePod(newPod); err != nil {
 		glog.Errorf("Failed to update Pod <%s/%s>: %v in cache",
 			newPod.Namespace, newPod.Name, err)
@@ -277,6 +295,11 @@ func (cc *Controller) deletePod(obj interface{}) {
 		}
 	}
 
+	// Filter out pods that are not created from volcano job
+	if !isControlledBy(pod, helpers.JobKind) {
+		return
+	}
+
 	taskName, found := pod.Annotations[vkbatchv1.TaskSpecKey]
 	if !found {
 		glog.Infof("Failed to find taskName of Pod <%s/%s>, skipping",
@@ -286,11 +309,15 @@ func (cc *Controller) deletePod(obj interface{}) {
 
 	jobName, found := pod.Annotations[vkbatchv1.JobNameKey]
 	if !found {
+		glog.Infof("Failed to find jobName of Pod <%s/%s>, skipping",
+			pod.Namespace, pod.Name)
 		return
 	}
 
 	version, found := pod.Annotations[vkbatchv1.JobVersion]
 	if !found {
+		glog.Infof("Failed to find jobVersion of Pod <%s/%s>, skipping",
+			pod.Namespace, pod.Name)
 		return
 	}
 

diff --git a/pkg/controllers/job/job_controller_handler_test.go b/pkg/controllers/job/job_controller_handler_test.go
@@ -21,6 +21,7 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/rest"
 	"testing"
+	"volcano.sh/volcano/pkg/apis/helpers"
 
 	"k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -66,13 +67,21 @@ func newController() *Controller {
 }
 
 func buildPod(namespace, name string, p v1.PodPhase, labels map[string]string) *v1.Pod {
+	boolValue := true
 	return &v1.Pod{
 		ObjectMeta: metav1.ObjectMeta{
 			UID:             types.UID(fmt.Sprintf("%v-%v", namespace, name)),
 			Name:            name,
 			Namespace:       namespace,
 			Labels:          labels,
 			ResourceVersion: string(uuid.NewUUID()),
+			OwnerReferences: []metav1.OwnerReference{
+				{
+					APIVersion: helpers.JobKind.GroupVersion().String(),
+					Kind:       helpers.JobKind.Kind,
+					Controller: &boolValue,
+				},
+			},
 		},
 		Status: v1.PodStatus{
 			Phase: p,

diff --git a/pkg/controllers/job/job_controller_util.go b/pkg/controllers/job/job_controller_util.go
@@ -23,6 +23,7 @@ import (
 
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	kbapi "volcano.sh/volcano/pkg/apis/scheduling/v1alpha1"
 
 	"volcano.sh/volcano/pkg/apis/batch/v1alpha1"
@@ -236,3 +237,14 @@ func (p TasksPriority) Less(i, j int) bool {
 }
 
 func (p TasksPriority) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
+
+func isControlledBy(obj metav1.Object, gvk schema.GroupVersionKind) bool {
+	controlerRef := metav1.GetControllerOf(obj)
+	if controlerRef == nil {
+		return false
+	}
+	if controlerRef.APIVersion == gvk.GroupVersion().String() && controlerRef.Kind == gvk.Kind {
+		return true
+	}
+	return false
+}