adds an option to specify whether informers should fail on error, which defaults to false to maintain existing compatibility

Author: emsixteeen

cmd/mpi-operator/app/options/options.go

@@ -38,6 +38,7 @@ type ServerOption struct {
 	LockNamespace      string
 	QPS                int
 	Burst              int
+	FailInformersOnErr bool
 }
 
 // NewServerOption creates a new CMServer with a default config.
@@ -75,4 +76,8 @@ func (s *ServerOption) AddFlags(fs *flag.FlagSet) {
 
 	fs.IntVar(&s.QPS, "kube-api-qps", 5, "QPS indicates the maximum QPS to the master from this client.")
 	fs.IntVar(&s.Burst, "kube-api-burst", 10, "Maximum burst for throttle.")
+
+	fs.BoolVar(&s.FailInformersOnErr, "fail-informers-on-error", false,
+		`Exit the mpi-operator if it fails to list or watch objects from the API server due to lack of permissions (instead of retrying indefinitely).
+                Note: This only applies if a list/watch fails with Unauthorized or Forbidden errors.`)
 }

cmd/mpi-operator/app/server.go

@@ -153,7 +153,7 @@ func Run(opt *options.ServerOption) error {
 			kubeInformerFactory.Core().V1().Pods(),
 			kubeInformerFactory.Scheduling().V1().PriorityClasses(),
 			kubeflowInformerFactory.Kubeflow().V2beta1().MPIJobs(),
-			namespace, opt.GangSchedulingName)
+			namespace, opt.GangSchedulingName, opt.FailInformersOnErr)
 		if err != nil {
 			klog.Fatalf("Failed to setup the controller")
 		}

pkg/controller/mpi_job_controller.go

@@ -272,10 +272,10 @@ func NewMPIJobController(
 	podInformer coreinformers.PodInformer,
 	priorityClassInformer schedulinginformers.PriorityClassInformer,
 	mpiJobInformer informers.MPIJobInformer,
-	namespace, gangSchedulingName string) (*MPIJobController, error) {
+	namespace, gangSchedulingName string, failInformersOnErr bool) (*MPIJobController, error) {
 	return NewMPIJobControllerWithClock(kubeClient, kubeflowClient, volcanoClient, schedClient,
 		configMapInformer, secretInformer, serviceInformer, jobInformer, podInformer,
-		priorityClassInformer, mpiJobInformer, &clock.RealClock{}, namespace, gangSchedulingName)
+		priorityClassInformer, mpiJobInformer, &clock.RealClock{}, namespace, gangSchedulingName, failInformersOnErr)
 }
 
 // NewMPIJobControllerWithClock returns a new MPIJob controller.
@@ -292,7 +292,7 @@ func NewMPIJobControllerWithClock(
 	priorityClassInformer schedulinginformers.PriorityClassInformer,
 	mpiJobInformer informers.MPIJobInformer,
 	clock clock.WithTicker,
-	namespace, gangSchedulingName string) (*MPIJobController, error) {
+	namespace, gangSchedulingName string, failInformersOnErr bool) (*MPIJobController, error) {
 
 	// Create event broadcaster.
 	klog.V(4).Info("Creating event broadcaster")
@@ -346,30 +346,32 @@ func NewMPIJobControllerWithClock(
 
 	controller.updateStatusHandler = controller.doUpdateJobStatus
 
-	// Set up error handlers for informers
-	klog.Info("Setting up informer error handlers")
-	informers := map[string]cache.SharedIndexInformer{
-		"configMapInformer":     configMapInformer.Informer(),
-		"secretInformer":        secretInformer.Informer(),
-		"serviceInformer":       serviceInformer.Informer(),
-		"jobInformer":           jobInformer.Informer(),
-		"podInformer":           podInformer.Informer(),
-		"priorityClassInformer": priorityClassInformer.Informer(),
-		"mpiJobInformer":        mpiJobInformer.Informer(),
-	}
-
-	for name, informer := range informers {
-		err := informer.SetWatchErrorHandler(func(r *cache.Reflector, err error) {
-			// Pipe to default handler first, which just logs the error
-			cache.DefaultWatchErrorHandler(r, err)
-
-			if errors.IsUnauthorized(err) || errors.IsForbidden(err) {
-				klog.Fatalf("Unable to sync cache for informer %s: %s. Exiting.", name, err)
-			}
-		})
+	// Set up error handlers for informers if asked to do so
+	if failInformersOnErr {
+		klog.Info("Setting up informer error handlers")
+		informers := map[string]cache.SharedInformer{
+			"configMapInformer":     configMapInformer.Informer(),
+			"secretInformer":        secretInformer.Informer(),
+			"serviceInformer":       serviceInformer.Informer(),
+			"jobInformer":           jobInformer.Informer(),
+			"podInformer":           podInformer.Informer(),
+			"priorityClassInformer": priorityClassInformer.Informer(),
+			"mpiJobInformer":        mpiJobInformer.Informer(),
+		}
 
-		if err != nil {
-			klog.Fatalf("Unable to set error handler for informer %s: %s. Exiting.", name, err)
+		for name, informer := range informers {
+			err := informer.SetWatchErrorHandler(func(r *cache.Reflector, err error) {
+				// Pipe to default handler first, which just logs the error
+				cache.DefaultWatchErrorHandler(r, err)
+
+				if errors.IsUnauthorized(err) || errors.IsForbidden(err) {
+					klog.Fatalf("Unable to sync cache for informer %s: %s. Exiting.", name, err)
+				}
+			})
+
+			if err != nil {
+				klog.Fatalf("Unable to set error handler for informer %s: %s. Exiting.", name, err)
+			}
 		}
 	}