Repository: kubeflow/training-operator (language: python)

Issue:
panic happened when add podgroup watch
kubeadm version: v1.20.11
k8s version: 1.20.11
I have build training-operator image with this merged pr.
pytorch job can use gang-scheduler, now. However nil pointer problem happened, when  training controller build or watch pytorch jobs .
I have detected problem happened, when log event object in https://github.com/kubeflow/training-operator/blob/master/pkg/common/util/reconciler.go#L113
Some exception may happened.

------
errors as follows:

```bash
time="2022-10-31T11:55:15Z" level=info msg="PyTorchJob=ddp-vol, ReplicaType=Worker expected=1, running=1, succeeded=0, failed=0, Replicas=1"
E1031 11:56:00.164678       1 runtime.go:79] Observed a panic: "invalid memory address or nil pointer dereference" (runtime error: invalid memory address or nil pointer dereference)
goroutine 796 [running]:
k8s.io/apimachinery/pkg/util/runtime.logPanic({0x185a700?, 0x2a0c1a0})
	/go/pkg/mod/k8s.io/apimachinery@v0.24.1/pkg/util/runtime/runtime.go:75 +0x99
k8s.io/apimachinery/pkg/util/runtime.HandleCrash({0x0, 0x0, 0xc000064150?})
	/go/pkg/mod/k8s.io/apimachinery@v0.24.1/pkg/util/runtime/runtime.go:49 +0x75
panic({0x185a700, 0x2a0c1a0})
	/usr/local/go/src/runtime/panic.go:884 +0x212
github.com/sirupsen/logrus.(*Entry).Logf(0xc0003dca00?, 0x1213940?, {0x1ade9f4?, 0xc000f92050?}, {0xc000d29b48?, 0xc00003c000?, 0x47?})
	/go/pkg/mod/github.com/sirupsen/logrus@v1.8.1/entry.go:337 +0x22
github.com/sirupsen/logrus.(*Entry).Debugf(...)
	/go/pkg/mod/github.com/sirupsen/logrus@v1.8.1/entry.go:347
github.com/kubeflow/training-operator/pkg/common/util.OnDependentUpdateFunc.func1({{0x1d5b0d0?, 0xc00045e000?}, {0x1d5b0d0?, 0xc000624d80?}})
	/workspace/pkg/common/util/reconciler.go:113 +0x4fd
sigs.k8s.io/controller-runtime/pkg/predicate.Funcs.Update(...)
	/go/pkg/mod/sigs.k8s.io/controller-runtime@v0.12.1/pkg/predicate/predicate.go:88
sigs.k8s.io/controller-runtime/pkg/source/internal.EventHandler.OnUpdate({{0x1d445a0, 0xc0002c5c20}, {0x1d4d7c8, 0xc00099c1a0}, {0xc0008a0d70, 0x1, 0x1}}, {0x1a6fb40?, 0xc00045e000}, {0x1a6fb40, ...})
	/go/pkg/mod/sigs.k8s.io/controller-runtime@v0.12.1/pkg/source/internal/eventsource.go:88 +0x43d
k8s.io/client-go/tools/cache.(*processorListener).run.func1()
	/go/pkg/mod/k8s.io/client-go@v0.24.1/tools/cache/shared_informer.go:816 +0xf7
k8s.io/apimachinery/pkg/util/wait.BackoffUntil.func1(0xc00046ea20?)
	/go/pkg/mod/k8s.io/apimachinery@v0.24.1/pkg/util/wait/wait.go:155 +0x3e
k8s.io/apimachinery/pkg/util/wait.BackoffUntil(0xc0007bf738?, {0x1d2b040, 0xc000722c00}, 0x1, 0xc00091a540)
	/go/pkg/mod/k8s.io/apimachinery@v0.24.1/pkg/util/wait/wait.go:156 +0xb6
k8s.io/apimachinery/pkg/util/wait.JitterUntil(0xc00046ed50?, 0x3b9aca00, 0x0, 0x90?, 0xc0005a0420?)
	/go/pkg/mod/k8s.io/apimachinery@v0.24.1/pkg/util/wait/wait.go:133 +0x89
k8s.io/apimachinery/pkg/util/wait.Until(...)
	/go/pkg/mod/k8s.io/apimachinery@v0.24.1/pkg/util/wait/wait.go:90
k8s.io/client-go/tools/cache.(*processorListener).run(0xc00036c280?)
	/go/pkg/mod/k8s.io/client-go@v0.24.1/tools/cache/shared_informer.go:812 +0x6b
k8s.io/apimachinery/pkg/util/wait.(*Group).Start.func1()
	/go/pkg/mod/k8s.io/apimachinery@v0.24.1/pkg/util/wait/wait.go:73 +0x5a
created by k8s.io/apimachinery/pkg/util/wait.(*Group).Start
	/go/pkg/mod/k8s.io/apimachinery@v0.24.1/pkg/util/wait/wait.go:71 +0x85
panic: runtime error: invalid memory address or nil pointer dereference [recovered]
	panic: runtime error: invalid memory address or nil pointer dereference
[signal SIGSEGV: segmentation violation code=0x1 addr=0x0 pc=0x14eb0c2]

goroutine 796 [running]:
k8s.io/apimachinery/pkg/util/runtime.HandleCrash({0x0, 0x0, 0xc000064150?})
	/go/pkg/mod/k8s.io/apimachinery@v0.24.1/pkg/util/runtime/runtime.go:56 +0xd7
panic({0x185a700, 0x2a0c1a0})
	/usr/local/go/src/runtime/panic.go:884 +0x212
github.com/sirupsen/logrus.(*Entry).Logf(0xc0003dca00?, 0x1213940?, {0x1ade9f4?, 0xc000f92050?}, {0xc000d29b48?, 0xc00003c000?, 0x47?})
	/go/pkg/mod/github.com/sirupsen/logrus@v1.8.1/entry.go:337 +0x22
github.com/sirupsen/logrus.(*Entry).Debugf(...)
	/go/pkg/mod/github.com/sirupsen/logrus@v1.8.1/entry.go:347
github.com/kubeflow/training-operator/pkg/common/util.OnDependentUpdateFunc.func1({{0x1d5b0d0?, 0xc00045e000?}, {0x1d5b0d0?, 0xc000624d80?}})
	/workspace/pkg/common/util/reconciler.go:113 +0x4fd
sigs.k8s.io/controller-runtime/pkg/predicate.Funcs.Update(...)
	/go/pkg/mod/sigs.k8s.io/controller-runtime@v0.12.1/pkg/predicate/predicate.go:88
sigs.k8s.io/controller-runtime/pkg/source/internal.EventHandler.OnUpdate({{0x1d445a0, 0xc0002c5c20}, {0x1d4d7c8, 0xc00099c1a0}, {0xc0008a0d70, 0x1, 0x1}}, {0x1a6fb40?, 0xc00045e000}, {0x1a6fb40, ...})
	/go/pkg/mod/sigs.k8s.io/controller-runtime@v0.12.1/pkg/source/internal/eventsource.go:88 +0x43d
k8s.io/client-go/tools/cache.(*processorListener).run.func1()
	/go/pkg/mod/k8s.io/client-go@v0.24.1/tools/cache/shared_informer.go:816 +0xf7
k8s.io/apimachinery/pkg/util/wait.BackoffUntil.func1(0xc00046ea20?)
	/go/pkg/mod/k8s.io/apimachinery@v0.24.1/pkg/util/wait/wait.go:155 +0x3e
k8s.io/apimachinery/pkg/util/wait.BackoffUntil(0xc0007bf738?, {0x1d2b040, 0xc000722c00}, 0x1, 0xc00091a540)
	/go/pkg/mod/k8s.io/apimachinery@v0.24.1/pkg/util/wait/wait.go:156 +0xb6
k8s.io/apimachinery/pkg/util/wait.JitterUntil(0xc00046ed50?, 0x3b9aca00, 0x0, 0x90?, 0xc0005a0420?)
	/go/pkg/mod/k8s.io/apimachinery@v0.24.1/pkg/util/wait/wait.go:133 +0x89
k8s.io/apimachinery/pkg/util/wait.Until(...)
	/go/pkg/mod/k8s.io/apimachinery@v0.24.1/pkg/util/wait/wait.go:90
k8s.io/client-go/tools/cache.(*processorListener).run(0xc00036c280?)
	/go/pkg/mod/k8s.io/client-go@v0.24.1/tools/cache/shared_informer.go:812 +0x6b
k8s.io/apimachinery/pkg/util/wait.(*Group).Start.func1()
	/go/pkg/mod/k8s.io/apimachinery@v0.24.1/pkg/util/wait/wait.go:73 +0x5a
created by k8s.io/apimachinery/pkg/util/wait.(*Group).Start
	/go/pkg/mod/k8s.io/apimachinery@v0.24.1/pkg/util/wait/wait.go:71 +0x85

```



_Originally posted by @liuweibin6566396837 in https://github.com/kubeflow/training-operator/issues/1666#issuecomment-1297018043_
      


Discussion / hints:

Fix patch:
diff --git a/pkg/common/util/reconciler.go b/pkg/common/util/reconciler.go
index 880437bfac..28cd653da2 100644
--- a/pkg/common/util/reconciler.go
+++ b/pkg/common/util/reconciler.go
@@ -22,7 +22,6 @@ import (
 	"github.com/kubeflow/common/pkg/controller.v1/common"
 	"github.com/kubeflow/common/pkg/controller.v1/expectation"
 	commonutil "github.com/kubeflow/common/pkg/util"
-	"github.com/sirupsen/logrus"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"sigs.k8s.io/controller-runtime/pkg/event"
@@ -57,12 +56,13 @@ func OnDependentCreateFunc(exp expectation.ControllerExpectationsInterface) func
 		if controllerRef := metav1.GetControllerOf(e.Object); controllerRef != nil {
 			jobKey := fmt.Sprintf("%s/%s", e.Object.GetNamespace(), controllerRef.Name)
 			var expectKey string
-			if _, ok := e.Object.(*corev1.Pod); ok {
+			switch e.Object.(type) {
+			case *corev1.Pod:
 				expectKey = expectation.GenExpectationPodsKey(jobKey, rtype)
-			}
-
-			if _, ok := e.Object.(*corev1.Service); ok {
+			case *corev1.Service:
 				expectKey = expectation.GenExpectationServicesKey(jobKey, rtype)
+			default:
+				return false
 			}
 			exp.CreationObserved(expectKey)
 			return true
@@ -83,13 +83,16 @@ func OnDependentUpdateFunc(jc *common.JobController) func(updateEvent event.Upda
 			return false
 		}
 
-		var logger *logrus.Entry
-		if _, ok := newObj.(*corev1.Pod); ok {
-			logger = commonutil.LoggerForPod(newObj.(*corev1.Pod), jc.Controller.GetAPIGroupVersionKind().Kind)
-		}
+		kind := jc.Controller.GetAPIGroupVersionKind().Kind
+		var logger = LoggerForGenericKind(newObj, kind)
 
-		if _, ok := newObj.(*corev1.Service); ok {
+		switch obj := newObj.(type) {
+		case *corev1.Pod:
+			logger = commonutil.LoggerForPod(obj, jc.Controller.GetAPIGroupVersionKind().Kind)
+		case *corev1.Service:
 			logger = commonutil.LoggerForService(newObj.(*corev1.Service), jc.Controller.GetAPIGroupVersionKind().Kind)
+		default:
+			return false
 		}
 
 		newControllerRef := metav1.GetControllerOf(newObj)
@@ -151,14 +154,14 @@ func OnDependentDeleteFunc(exp expectation.ControllerExpectationsInterface) func
 		if controllerRef := metav1.GetControllerOf(e.Object); controllerRef != nil {
 			jobKey := fmt.Sprintf("%s/%s", e.Object.GetNamespace(), controllerRef.Name)
 			var expectKey string
-			if _, ok := e.Object.(*corev1.Pod); ok {
+			switch e.Object.(type) {
+			case *corev1.Pod:
 				expectKey = expectation.GenExpectationPodsKey(jobKey, rtype)
-			}
-
-			if _, ok := e.Object.(*corev1.Service); ok {
+			case *corev1.Service:
 				expectKey = expectation.GenExpectationServicesKey(jobKey, rtype)
+			default:
+				return false
 			}
-
 			exp.DeletionObserved(expectKey)
 			return true
 		}
diff --git a/pkg/controller.v1/mxnet/mxjob_controller.go b/pkg/controller.v1/mxnet/mxjob_controller.go
index c0d977c488..9eb0584b88 100644
--- a/pkg/controller.v1/mxnet/mxjob_controller.go
+++ b/pkg/controller.v1/mxnet/mxjob_controller.go
@@ -226,9 +226,9 @@ func (r *MXJobReconciler) SetupWithManager(mgr ctrl.Manager) error {
 			IsController: true,
 			OwnerType:    &kubeflowv1.MXJob{},
 		}, predicate.Funcs{
-			CreateFunc: util.OnDependentCreateFunc(r.Expectations),
-			UpdateFunc: util.OnDependentUpdateFunc(&r.JobController),
-			DeleteFunc: util.OnDependentDeleteFunc(r.Expectations),
+			CreateFunc: util.OnDependentCreateFuncGeneric(r.Expectations),
+			UpdateFunc: util.OnDependentUpdateFuncGeneric(&r.JobController),
+			DeleteFunc: util.OnDependentDeleteFuncGeneric(r.Expectations),
 		}); err != nil {
 			return err
 		}
diff --git a/pkg/controller.v1/pytorch/pytorchjob_controller.go b/pkg/controller.v1/pytorch/pytorchjob_controller.go
index 18aa303a1a..211ec906b0 100644
--- a/pkg/controller.v1/pytorch/pytorchjob_controller.go
+++ b/pkg/controller.v1/pytorch/pytorchjob_controller.go
@@ -224,9 +224,9 @@ func (r *PyTorchJobReconciler) SetupWithManager(mgr ctrl.Manager) error {
 			IsController: true,
 			OwnerType:    &kubeflowv1.PyTorchJob{},
 		}, predicate.Funcs{
-			CreateFunc: util.OnDependentCreateFunc(r.Expectations),
-			UpdateFunc: util.OnDependentUpdateFunc(&r.JobController),
-			DeleteFunc: util.OnDependentDeleteFunc(r.Expectations),
+			CreateFunc: util.OnDependentCreateFuncGeneric(r.Expectations),
+			UpdateFunc: util.OnDependentUpdateFuncGeneric(&r.JobController),
+			DeleteFunc: util.OnDependentDeleteFuncGeneric(r.Expectations),
 		}); err != nil {
 			return err
 		}
diff --git a/pkg/controller.v1/tensorflow/tfjob_controller.go b/pkg/controller.v1/tensorflow/tfjob_controller.go
index 5f29bf5141..4b207c9b64 100644
--- a/pkg/controller.v1/tensorflow/tfjob_controller.go
+++ b/pkg/controller.v1/tensorflow/tfjob_controller.go
@@ -245,9 +245,9 @@ func (r *TFJobReconciler) SetupWithManager(mgr ctrl.Manager) error {
 			IsController: true,
 			OwnerType:    &kubeflowv1.TFJob{},
 		}, predicate.Funcs{
-			CreateFunc: util.OnDependentCreateFunc(r.Expectations),
-			UpdateFunc: util.OnDependentUpdateFunc(&r.JobController),
-			DeleteFunc: util.OnDependentDeleteFunc(r.Expectations),
+			CreateFunc: util.OnDependentCreateFuncGeneric(r.Expectations),
+			UpdateFunc: util.OnDependentUpdateFuncGeneric(&r.JobController),
+			DeleteFunc: util.OnDependentDeleteFuncGeneric(r.Expectations),
 		}); err != nil {
 			return err
 		}
diff --git a/pkg/controller.v1/xgboost/xgboostjob_controller.go b/pkg/controller.v1/xgboost/xgboostjob_controller.go
index 2b9295ec54..396ee82357 100644
--- a/pkg/controller.v1/xgboost/xgboostjob_controller.go
+++ b/pkg/controller.v1/xgboost/xgboostjob_controller.go
@@ -231,9 +231,9 @@ func (r *XGBoostJobReconciler) SetupWithManager(mgr ctrl.Manager) error {
 			IsController: true,
 			OwnerType:    &kubeflowv1.XGBoostJob{},
 		}, predicate.Funcs{
-			CreateFunc: util.OnDependentCreateFunc(r.Expectations),
-			UpdateFunc: util.OnDependentUpdateFunc(&r.JobController),
-			DeleteFunc: util.OnDependentDeleteFunc(r.Expectations),
+			CreateFunc: util.OnDependentCreateFuncGeneric(r.Expectations),
+			UpdateFunc: util.OnDependentUpdateFuncGeneric(&r.JobController),
+			DeleteFunc: util.OnDependentDeleteFuncGeneric(r.Expectations),
 		}); err != nil {
 			return err
 		}


Test patch:
diff --git a/pkg/common/util/reconciler_test.go b/pkg/common/util/reconciler_test.go
new file mode 100644
index 0000000000..22984117f3
--- /dev/null
+++ b/pkg/common/util/reconciler_test.go
@@ -0,0 +1,58 @@
+package util
+
+import (
+	"testing"
+
+	commonv1 "github.com/kubeflow/common/pkg/apis/common/v1"
+	"github.com/kubeflow/common/pkg/controller.v1/expectation"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+)
+
+func TestOnDependentXXXFunc(t *testing.T) {
+	createfunc := OnDependentCreateFunc(expectation.NewControllerExpectations())
+	deletefunc := OnDependentDeleteFunc(expectation.NewControllerExpectations())
+
+	for _, testCase := range []struct {
+		object client.Object
+		expect bool
+	}{
+		{
+			// pod object with label is allowed
+			object: &corev1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{
+						commonv1.ReplicaTypeLabel: "Worker",
+					},
+				},
+			},
+			expect: true,
+		},
+		{
+			// service object without label is not allowed
+			object: &corev1.Service{},
+			expect: false,
+		},
+		{
+			// objects other than pod/service are not allowed
+			object: &corev1.ConfigMap{},
+			expect: false,
+		},
+	} {
+		ret := createfunc(event.CreateEvent{
+			Object: testCase.object,
+		})
+		if ret != testCase.expect {
+			t.Errorf("expect %t, but get %t", testCase.expect, ret)
+		}
+		ret = deletefunc(event.DeleteEvent{
+			Object: testCase.object,
+		})
+		if ret != testCase.expect {
+			t.Errorf("expect %t, but get %t", testCase.expect, ret)
+		}
+
+	}
+}