func NewMPIJobController(
	kubeClient kubernetes.Interface,
	kubeflowClient clientset.Interface,
	configMapInformer coreinformers.ConfigMapInformer,
	serviceAccountInformer coreinformers.ServiceAccountInformer,
	roleInformer rbacinformers.RoleInformer,
	roleBindingInformer rbacinformers.RoleBindingInformer,
	statefulSetInformer appsinformers.StatefulSetInformer,
	jobInformer batchinformers.JobInformer,
	mpiJobInformer informers.MPIJobInformer,
	gpusPerNode int,
	kubectlDeliveryImage string) *MPIJobController {

	// Create event broadcaster.
	// Add mpi-job-controller types to the default Kubernetes Scheme so Events
	// can be logged for mpi-job-controller types.
	kubeflowScheme.AddToScheme(scheme.Scheme)
	glog.V(4).Info("Creating event broadcaster")
	eventBroadcaster := record.NewBroadcaster()
	eventBroadcaster.StartLogging(glog.Infof)
	eventBroadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{Interface: kubeClient.CoreV1().Events("")})
	recorder := eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: controllerAgentName})

	controller := &MPIJobController{
		kubeClient:           kubeClient,
		kubeflowClient:       kubeflowClient,
		configMapLister:      configMapInformer.Lister(),
		configMapSynced:      configMapInformer.Informer().HasSynced,
		serviceAccountLister: serviceAccountInformer.Lister(),
		serviceAccountSynced: serviceAccountInformer.Informer().HasSynced,
		roleLister:           roleInformer.Lister(),
		roleSynced:           roleInformer.Informer().HasSynced,
		roleBindingLister:    roleBindingInformer.Lister(),
		roleBindingSynced:    roleBindingInformer.Informer().HasSynced,
		statefulSetLister:    statefulSetInformer.Lister(),
		statefulSetSynced:    statefulSetInformer.Informer().HasSynced,
		jobLister:            jobInformer.Lister(),
		jobSynced:            jobInformer.Informer().HasSynced,
		mpiJobLister:         mpiJobInformer.Lister(),
		mpiJobSynced:         mpiJobInformer.Informer().HasSynced,
		queue:                workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "MPIJobs"),
		recorder:             recorder,
		gpusPerNode:          gpusPerNode,
		kubectlDeliveryImage: kubectlDeliveryImage,
	}

	glog.Info("Setting up event handlers")
	// Set up an event handler for when MPIJob resources change.
	mpiJobInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: controller.enqueueMPIJob,
		UpdateFunc: func(old, new interface{}) {
			controller.enqueueMPIJob(new)
		},
	})

	// Set up an event handler for when dependent resources change. This
	// handler will lookup the owner of the given resource, and if it is
	// owned by an MPIJob resource will enqueue that MPIJob resource for
	// processing. This way, we don't need to implement custom logic for
	// handling dependent resources. More info on this pattern:
	// https://github.com/kubernetes/community/blob/8cafef897a22026d42f5e5bb3f104febe7e29830/contributors/devel/controllers.md
	configMapInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: controller.handleObject,
		UpdateFunc: func(old, new interface{}) {
			newConfigMap := new.(*corev1.ConfigMap)
			oldConfigMap := old.(*corev1.ConfigMap)
			if newConfigMap.ResourceVersion == oldConfigMap.ResourceVersion {
				// Periodic re-sync will send update events for all known
				// ConfigMaps. Two different versions of the same ConfigMap
				// will always have different RVs.
				return
			}
			controller.handleObject(new)
		},
		DeleteFunc: controller.handleObject,
	})
	serviceAccountInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: controller.handleObject,
		UpdateFunc: func(old, new interface{}) {
			newServiceAccount := new.(*corev1.ServiceAccount)
			oldServiceAccount := old.(*corev1.ServiceAccount)
			if newServiceAccount.ResourceVersion == oldServiceAccount.ResourceVersion {
				// Periodic re-sync will send update events for all known
				// ServiceAccounts. Two different versions of the same ServiceAccount
				// will always have different RVs.
				return
			}
			controller.handleObject(new)
		},
		DeleteFunc: controller.handleObject,
	})
	roleInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: controller.handleObject,
		UpdateFunc: func(old, new interface{}) {
			newRole := new.(*rbacv1.Role)
			oldRole := old.(*rbacv1.Role)
			if newRole.ResourceVersion == oldRole.ResourceVersion {
				// Periodic re-sync will send update events for all known
				// Roles. Two different versions of the same Role
				// will always have different RVs.
				return
			}
			controller.handleObject(new)
		},
		DeleteFunc: controller.handleObject,
	})
	roleBindingInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: controller.handleObject,
		UpdateFunc: func(old, new interface{}) {
			newRoleBinding := new.(*rbacv1.RoleBinding)
			oldRoleBinding := old.(*rbacv1.RoleBinding)
			if newRoleBinding.ResourceVersion == oldRoleBinding.ResourceVersion {
				// Periodic re-sync will send update events for all known
				// RoleBindings. Two different versions of the same RoleBinding
				// will always have different RVs.
				return
			}
			controller.handleObject(new)
		},
		DeleteFunc: controller.handleObject,
	})
	statefulSetInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: controller.handleObject,
		UpdateFunc: func(old, new interface{}) {
			newStatefulSet := new.(*appsv1.StatefulSet)
			oldStatefulSet := old.(*appsv1.StatefulSet)
			if newStatefulSet.ResourceVersion == oldStatefulSet.ResourceVersion {
				// Periodic re-sync will send update events for all known
				// StatefulSets. Two different versions of the same StatefulSet
				// will always have different RVs.
				return
			}
			controller.handleObject(new)
		},
		DeleteFunc: controller.handleObject,
	})
	jobInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: controller.handleObject,
		UpdateFunc: func(old, new interface{}) {
			newJob := new.(*batchv1.Job)
			oldJob := old.(*batchv1.Job)
			if newJob.ResourceVersion == oldJob.ResourceVersion {
				// Periodic re-sync will send update events for all known Jobs.
				// Two different versions of the same Job will always have
				// different RVs.
				return
			}
			controller.handleObject(new)
		},
		DeleteFunc: controller.handleObject,
	})

	return controller
}
// NewMPIJobController returns a new MPIJob controller.
NewMPIJobController returns a new MPIJob controller.
// NewMPIJobController returns a new MPIJob controller .
func NewMPIJobController(
	kubeClient kubernetes.Interface,
	kubeflowClient clientset.Interface,
	configMapInformer coreinformers.ConfigMapInformer,
	serviceAccountInformer coreinformers.ServiceAccountInformer,
	roleInformer rbacinformers.RoleInformer,
	roleBindingInformer rbacinformers.RoleBindingInformer,
	statefulSetInformer appsinformers.StatefulSetInformer,
	jobInformer batchinformers.JobInformer,
	mpiJobInformer informers.MPIJobInformer,
	gpusPerNode int,
	kubectlDeliveryImage string) *MPIJobController {
	kubeflowScheme.AddToScheme(scheme.Scheme)
	glog.V(4).Info("Creating event broadcaster")
	eventBroadcaster := record.NewBroadcaster()
	eventBroadcaster.StartLogging(glog.Infof)
	eventBroadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{Interface: kubeClient.CoreV1().Events("")})
	recorder := eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: controllerAgentName})
	controller := &MPIJobController{
		kubeClient:           kubeClient,
		kubeflowClient:       kubeflowClient,
		configMapLister:      configMapInformer.Lister(),
		configMapSynced:      configMapInformer.Informer().HasSynced,
		serviceAccountLister: serviceAccountInformer.Lister(),
		serviceAccountSynced: serviceAccountInformer.Informer().HasSynced,
		roleLister:           roleInformer.Lister(),
		roleSynced:           roleInformer.Informer().HasSynced,
		roleBindingLister:    roleBindingInformer.Lister(),
		roleBindingSynced:    roleBindingInformer.Informer().HasSynced,
		statefulSetLister:    statefulSetInformer.Lister(),
		statefulSetSynced:    statefulSetInformer.Informer().HasSynced,
		jobLister:            jobInformer.Lister(),
		jobSynced:            jobInformer.Informer().HasSynced,
		mpiJobLister:         mpiJobInformer.Lister(),
		mpiJobSynced:         mpiJobInformer.Informer().HasSynced,
		queue:                workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "MPIJobs"),
		recorder:             recorder,
		gpusPerNode:          gpusPerNode,
		kubectlDeliveryImage: kubectlDeliveryImage,
	}
	glog.Info("Setting up event handlers")
	mpiJobInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: controller.enqueueMPIJob,
		UpdateFunc: func(old, new interface{}) {
			controller.enqueueMPIJob(new)
		},
	})
	configMapInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: controller.handleObject,
		UpdateFunc: func(old, new interface{}) {
			newConfigMap := new.(*corev1.ConfigMap)
			oldConfigMap := old.(*corev1.ConfigMap)
			if newConfigMap.ResourceVersion == oldConfigMap.ResourceVersion {
				return
			}
			controller.handleObject(new)
		},
		DeleteFunc: controller.handleObject,
	})
	serviceAccountInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: controller.handleObject,
		UpdateFunc: func(old, new interface{}) {
			newServiceAccount := new.(*corev1.ServiceAccount)
			oldServiceAccount := old.(*corev1.ServiceAccount)
			if newServiceAccount.ResourceVersion == oldServiceAccount.ResourceVersion {
				return
			}
			controller.handleObject(new)
		},
		DeleteFunc: controller.handleObject,
	})
	roleInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: controller.handleObject,
		UpdateFunc: func(old, new interface{}) {
			newRole := new.(*rbacv1.Role)
			oldRole := old.(*rbacv1.Role)
			if newRole.ResourceVersion == oldRole.ResourceVersion {
				return
			}
			controller.handleObject(new)
		},
		DeleteFunc: controller.handleObject,
	})
	roleBindingInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: controller.handleObject,
		UpdateFunc: func(old, new interface{}) {
			newRoleBinding := new.(*rbacv1.RoleBinding)
			oldRoleBinding := old.(*rbacv1.RoleBinding)
			if newRoleBinding.ResourceVersion == oldRoleBinding.ResourceVersion {
				return
			}
			controller.handleObject(new)
		},
		DeleteFunc: controller.handleObject,
	})
	statefulSetInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: controller.handleObject,
		UpdateFunc: func(old, new interface{}) {
			newStatefulSet := new.(*appsv1.StatefulSet)
			oldStatefulSet := old.(*appsv1.StatefulSet)
			if newStatefulSet.ResourceVersion == oldStatefulSet.ResourceVersion {
				return
			}
			controller.handleObject(new)
		},
		DeleteFunc: controller.handleObject,
	})
	jobInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: controller.handleObject,
		UpdateFunc: func(old, new interface{}) {
			newJob := new.(*batchv1.Job)
			oldJob := old.(*batchv1.Job)
			if newJob.ResourceVersion == oldJob.ResourceVersion {
				 Jobs.
				return
			}
			controller.handleObject(new)
		},
		DeleteFunc: controller.handleObject,
	})
	return controller
}
func NewMPIJobController ( kubeClient kubernetes . Interface , kubeflowClient clientset . Interface , configMapInformer coreinformers . ConfigMapInformer , serviceAccountInformer coreinformers . ServiceAccountInformer , roleInformer rbacinformers . RoleInformer , roleBindingInformer rbacinformers . RoleBindingInformer , statefulSetInformer appsinformers . StatefulSetInformer , jobInformer batchinformers . JobInformer , mpiJobInformer informers . MPIJobInformer , gpusPerNode int , kubectlDeliveryImage string ) * MPIJobController { kubeflowScheme . AddToScheme ( scheme . Scheme ) glog . V ( 4 ) . Info ( " " ) eventBroadcaster := record . NewBroadcaster ( ) eventBroadcaster . StartLogging ( glog . Infof ) eventBroadcaster . StartRecordingToSink ( & typedcorev1 . EventSinkImpl { Interface : kubeClient . CoreV1 ( ) . Events ( " " ) } ) recorder := eventBroadcaster . NewRecorder ( scheme . Scheme , corev1 . EventSource { Component : controllerAgentName } ) controller := & MPIJobController { kubeClient : kubeClient , kubeflowClient : kubeflowClient , configMapLister : configMapInformer . Lister ( ) , configMapSynced : configMapInformer . Informer ( ) . HasSynced , serviceAccountLister : serviceAccountInformer . Lister ( ) , serviceAccountSynced : serviceAccountInformer . Informer ( ) . HasSynced , roleLister : roleInformer . Lister ( ) , roleSynced : roleInformer . Informer ( ) . HasSynced , roleBindingLister : roleBindingInformer . Lister ( ) , roleBindingSynced : roleBindingInformer . Informer ( ) . HasSynced , statefulSetLister : statefulSetInformer . Lister ( ) , statefulSetSynced : statefulSetInformer . Informer ( ) . HasSynced , jobLister : jobInformer . Lister ( ) , jobSynced : jobInformer . Informer ( ) . HasSynced , mpiJobLister : mpiJobInformer . Lister ( ) , mpiJobSynced : mpiJobInformer . Informer ( ) . HasSynced , queue : workqueue . NewNamedRateLimitingQueue ( workqueue . DefaultControllerRateLimiter ( ) , " " ) , recorder : recorder , gpusPerNode : gpusPerNode , kubectlDeliveryImage : kubectlDeliveryImage , } glog . Info ( " " ) mpiJobInformer . Informer ( ) . AddEventHandler ( cache . ResourceEventHandlerFuncs { AddFunc : controller . enqueueMPIJob , UpdateFunc : func ( old , new interface { } ) { controller . enqueueMPIJob ( new ) } , } ) configMapInformer . Informer ( ) . AddEventHandler ( cache . ResourceEventHandlerFuncs { AddFunc : controller . handleObject , UpdateFunc : func ( old , new interface { } ) { newConfigMap := new . ( * corev1 . ConfigMap ) oldConfigMap := old . ( * corev1 . ConfigMap ) if newConfigMap . ResourceVersion == oldConfigMap . ResourceVersion { return } controller . handleObject ( new ) } , DeleteFunc : controller . handleObject , } ) serviceAccountInformer . Informer ( ) . AddEventHandler ( cache . ResourceEventHandlerFuncs { AddFunc : controller . handleObject , UpdateFunc : func ( old , new interface { } ) { newServiceAccount := new . ( * corev1 . ServiceAccount ) oldServiceAccount := old . ( * corev1 . ServiceAccount ) if newServiceAccount . ResourceVersion == oldServiceAccount . ResourceVersion { return } controller . handleObject ( new ) } , DeleteFunc : controller . handleObject , } ) roleInformer . Informer ( ) . AddEventHandler ( cache . ResourceEventHandlerFuncs { AddFunc : controller . handleObject , UpdateFunc : func ( old , new interface { } ) { newRole := new . ( * rbacv1 . Role ) oldRole := old . ( * rbacv1 . Role ) if newRole . ResourceVersion == oldRole . ResourceVersion { return } controller . handleObject ( new ) } , DeleteFunc : controller . handleObject , } ) roleBindingInformer . Informer ( ) . AddEventHandler ( cache . ResourceEventHandlerFuncs { AddFunc : controller . handleObject , UpdateFunc : func ( old , new interface { } ) { newRoleBinding := new . ( * rbacv1 . RoleBinding ) oldRoleBinding := old . ( * rbacv1 . RoleBinding ) if newRoleBinding . ResourceVersion == oldRoleBinding . ResourceVersion { return } controller . handleObject ( new ) } , DeleteFunc : controller . handleObject , } ) statefulSetInformer . Informer ( ) . AddEventHandler ( cache . ResourceEventHandlerFuncs { AddFunc : controller . handleObject , UpdateFunc : func ( old , new interface { } ) { newStatefulSet := new . ( * appsv1 . StatefulSet ) oldStatefulSet := old . ( * appsv1 . StatefulSet ) if newStatefulSet . ResourceVersion == oldStatefulSet . ResourceVersion { return } controller . handleObject ( new ) } , DeleteFunc : controller . handleObject , } ) jobInformer . Informer ( ) . AddEventHandler ( cache . ResourceEventHandlerFuncs { AddFunc : controller . handleObject , UpdateFunc : func ( old , new interface { } ) { newJob := new . ( * batchv1 . Job ) oldJob := old . ( * batchv1 . Job ) if newJob . ResourceVersion == oldJob . ResourceVersion { return } controller . handleObject ( new ) } , DeleteFunc : controller . handleObject , } ) return controller }
NewMPIJobController returns a new MPIJob controller.
NewMPIJobController returns a new MPIJob controller .
// Create event broadcaster. // Add mpi-job-controller types to the default Kubernetes Scheme so Events // can be logged for mpi-job-controller types. // Set up an event handler for when MPIJob resources change. // Set up an event handler for when dependent resources change. This // handler will lookup the owner of the given resource, and if it is // owned by an MPIJob resource will enqueue that MPIJob resource for // processing. This way, we don't need to implement custom logic for // handling dependent resources. More info on this pattern: // https://github.com/kubernetes/community/blob/8cafef897a22026d42f5e5bb3f104febe7e29830/contributors/devel/controllers.md // Periodic re-sync will send update events for all known // ConfigMaps. Two different versions of the same ConfigMap // will always have different RVs. // Periodic re-sync will send update events for all known // ServiceAccounts. Two different versions of the same ServiceAccount // will always have different RVs. // Periodic re-sync will send update events for all known // Roles. Two different versions of the same Role // will always have different RVs. // Periodic re-sync will send update events for all known // RoleBindings. Two different versions of the same RoleBinding // will always have different RVs. // Periodic re-sync will send update events for all known // StatefulSets. Two different versions of the same StatefulSet // will always have different RVs. // Periodic re-sync will send update events for all known Jobs. // Two different versions of the same Job will always have // different RVs.