func (c *MPIJobController) syncHandler(key string) error {
	// Convert the namespace/name string into a distinct namespace and name.
	namespace, name, err := cache.SplitMetaNamespaceKey(key)
	if err != nil {
		runtime.HandleError(fmt.Errorf("invalid resource key: %s", key))
		return nil
	}

	// Get the MPIJob with this namespace/name.
	mpiJob, err := c.mpiJobLister.MPIJobs(namespace).Get(name)
	// The MPIJob may no longer exist, in which case we stop processing.
	if errors.IsNotFound(err) {
		runtime.HandleError(fmt.Errorf("mpi job '%s' in work queue no longer exists", key))
		return nil
	}
	if err != nil {
		return err
	}

	// Get the launcher Job for this MPIJob.
	launcher, err := c.getLauncherJob(mpiJob)
	if err != nil {
		return err
	}
	// We're done if the launcher either succeeded or failed.
	done := launcher != nil && (launcher.Status.Succeeded == 1 || launcher.Status.Failed == 1)

	workerReplicas, gpusPerWorker, err := allocateGPUs(mpiJob, c.gpusPerNode, done)
	if err != nil {
		runtime.HandleError(err)
		return nil
	}

	if !done {
		// Get the ConfigMap for this MPIJob.
		if config, err := c.getOrCreateConfigMap(mpiJob, workerReplicas, gpusPerWorker); config == nil || err != nil {
			return err
		}

		// Get the launcher ServiceAccount for this MPIJob.
		if sa, err := c.getOrCreateLauncherServiceAccount(mpiJob); sa == nil || err != nil {
			return err
		}

		// Get the launcher Role for this MPIJob.
		if r, err := c.getOrCreateLauncherRole(mpiJob, workerReplicas); r == nil || err != nil {
			return err
		}

		// Get the launcher RoleBinding for this MPIJob.
		if rb, err := c.getLauncherRoleBinding(mpiJob); rb == nil || err != nil {
			return err
		}
	}

	worker, err := c.getOrCreateWorkerStatefulSet(mpiJob, workerReplicas, gpusPerWorker)
	if err != nil {
		return err
	}

	// If the worker is ready, start the launcher.
	workerReady := workerReplicas == 0 || int(worker.Status.ReadyReplicas) == workerReplicas
	if workerReady && launcher == nil {
		launcher, err = c.kubeClient.BatchV1().Jobs(namespace).Create(newLauncher(mpiJob, c.kubectlDeliveryImage))
		if err != nil {
			return err
		}
	}

	// Finally, we update the status block of the MPIJob resource to reflect the
	// current state of the world.
	err = c.updateMPIJobStatus(mpiJob, launcher, worker)
	if err != nil {
		return err
	}

	c.recorder.Event(mpiJob, corev1.EventTypeNormal, SuccessSynced, MessageResourceSynced)
	return nil
}
// syncHandler compares the actual state with the desired, and attempts to
// converge the two. It then updates the Status block of the MPIJob resource
// with the current status of the resource.
syncHandler compares the actual state with the desired, and attempts to
converge the two. It then updates the Status block of the MPIJob resource
with the current status of the resource.
// syncHandler compares the actual state with the desired and attempts to // converge the two . It then updates the Status block of the MPIJob resource // with the current status of the resource .
func (c *MPIJobController) syncHandler(key string) error {
	namespace, name, err := cache.SplitMetaNamespaceKey(key)
	if err != nil {
		runtime.HandleError(fmt.Errorf("invalid resource key: %s", key))
		return nil
	}
	mpiJob, err := c.mpiJobLister.MPIJobs(namespace).Get(name)
	if errors.IsNotFound(err) {
		runtime.HandleError(fmt.Errorf("mpi job '%s' in work queue no longer exists", key))
		return nil
	}
	if err != nil {
		return err
	}
	launcher, err := c.getLauncherJob(mpiJob)
	if err != nil {
		return err
	}
	done := launcher != nil && (launcher.Status.Succeeded == 1 || launcher.Status.Failed == 1)
	workerReplicas, gpusPerWorker, err := allocateGPUs(mpiJob, c.gpusPerNode, done)
	if err != nil {
		runtime.HandleError(err)
		return nil
	}
	if !done {
		if config, err := c.getOrCreateConfigMap(mpiJob, workerReplicas, gpusPerWorker); config == nil || err != nil {
			return err
		}
		if sa, err := c.getOrCreateLauncherServiceAccount(mpiJob); sa == nil || err != nil {
			return err
		}
		if r, err := c.getOrCreateLauncherRole(mpiJob, workerReplicas); r == nil || err != nil {
			return err
		}
		if rb, err := c.getLauncherRoleBinding(mpiJob); rb == nil || err != nil {
			return err
		}
	}
	worker, err := c.getOrCreateWorkerStatefulSet(mpiJob, workerReplicas, gpusPerWorker)
	if err != nil {
		return err
	}
	workerReady := workerReplicas == 0 || int(worker.Status.ReadyReplicas) == workerReplicas
	if workerReady && launcher == nil {
		launcher, err = c.kubeClient.BatchV1().Jobs(namespace).Create(newLauncher(mpiJob, c.kubectlDeliveryImage))
		if err != nil {
			return err
		}
	}
	err = c.updateMPIJobStatus(mpiJob, launcher, worker)
	if err != nil {
		return err
	}
	c.recorder.Event(mpiJob, corev1.EventTypeNormal, SuccessSynced, MessageResourceSynced)
	return nil
}
func ( c * MPIJobController ) syncHandler ( key string ) error { namespace , name , err := cache . SplitMetaNamespaceKey ( key ) if err != nil { runtime . HandleError ( fmt . Errorf ( " " , key ) ) return nil } mpiJob , err := c . mpiJobLister . MPIJobs ( namespace ) . Get ( name ) if errors . IsNotFound ( err ) { runtime . HandleError ( fmt . Errorf ( " " , key ) ) return nil } if err != nil { return err } launcher , err := c . getLauncherJob ( mpiJob ) if err != nil { return err } done := launcher != nil && ( launcher . Status . Succeeded == 1 || launcher . Status . Failed == 1 ) workerReplicas , gpusPerWorker , err := allocateGPUs ( mpiJob , c . gpusPerNode , done ) if err != nil { runtime . HandleError ( err ) return nil } if ! done { if config , err := c . getOrCreateConfigMap ( mpiJob , workerReplicas , gpusPerWorker ) ; config == nil || err != nil { return err } if sa , err := c . getOrCreateLauncherServiceAccount ( mpiJob ) ; sa == nil || err != nil { return err } if r , err := c . getOrCreateLauncherRole ( mpiJob , workerReplicas ) ; r == nil || err != nil { return err } if rb , err := c . getLauncherRoleBinding ( mpiJob ) ; rb == nil || err != nil { return err } } worker , err := c . getOrCreateWorkerStatefulSet ( mpiJob , workerReplicas , gpusPerWorker ) if err != nil { return err } workerReady := workerReplicas == 0 || int ( worker . Status . ReadyReplicas ) == workerReplicas if workerReady && launcher == nil { launcher , err = c . kubeClient . BatchV1 ( ) . Jobs ( namespace ) . Create ( newLauncher ( mpiJob , c . kubectlDeliveryImage ) ) if err != nil { return err } } err = c . updateMPIJobStatus ( mpiJob , launcher , worker ) if err != nil { return err } c . recorder . Event ( mpiJob , corev1 . EventTypeNormal , SuccessSynced , MessageResourceSynced ) return nil }
syncHandler compares the actual state with the desired, and attempts to
converge the two.
syncHandler compares the actual state with the desired and attempts to converge the two .
// Convert the namespace/name string into a distinct namespace and name. // Get the MPIJob with this namespace/name. // The MPIJob may no longer exist, in which case we stop processing. // Get the launcher Job for this MPIJob. // We're done if the launcher either succeeded or failed. // Get the ConfigMap for this MPIJob. // Get the launcher ServiceAccount for this MPIJob. // Get the launcher Role for this MPIJob. // Get the launcher RoleBinding for this MPIJob. // If the worker is ready, start the launcher. // Finally, we update the status block of the MPIJob resource to reflect the // current state of the world.