func newWorker(mpiJob *kubeflow.MPIJob, desiredReplicas int32, gpus int) *appsv1.StatefulSet {
	labels := map[string]string{
		labelGroupName:   "kubeflow.org",
		labelMPIJobName:  mpiJob.Name,
		labelMPIRoleType: worker,
	}

	podSpec := mpiJob.Spec.Template.DeepCopy()
	// keep the labels which are set in PodTemplate
	if len(podSpec.Labels) == 0 {
		podSpec.Labels = make(map[string]string)
	}

	for key, value := range labels {
		podSpec.Labels[key] = value
	}
	// always set restartPolicy to restartAlways for statefulset
	podSpec.Spec.RestartPolicy = corev1.RestartPolicyAlways

	container := podSpec.Spec.Containers[0]
	container.Command = []string{"sleep"}
	container.Args = []string{"365d"}
	if container.Resources.Limits == nil {
		container.Resources.Limits = make(corev1.ResourceList)
	}
	container.Resources.Limits[gpuResourceName] = *resource.NewQuantity(int64(gpus), resource.DecimalExponent)

	// We need the kubexec.sh script here because Open MPI checks for the path
	// in every rank.
	container.VolumeMounts = append(container.VolumeMounts, corev1.VolumeMount{
		Name:      configVolumeName,
		MountPath: configMountPath,
	})
	podSpec.Spec.Containers[0] = container

	scriptMode := int32(0555)
	podSpec.Spec.Volumes = append(podSpec.Spec.Volumes, corev1.Volume{
		Name: configVolumeName,
		VolumeSource: corev1.VolumeSource{
			ConfigMap: &corev1.ConfigMapVolumeSource{
				LocalObjectReference: corev1.LocalObjectReference{
					Name: mpiJob.Name + configSuffix,
				},
				Items: []corev1.KeyToPath{
					{
						Key:  kubexecScriptName,
						Path: kubexecScriptName,
						Mode: &scriptMode,
					},
				},
			},
		},
	})

	// set default BackoofLimit
	if mpiJob.Spec.BackoffLimit == nil {
		mpiJob.Spec.BackoffLimit = new(int32)
		*mpiJob.Spec.BackoffLimit = 6
	}

	return &appsv1.StatefulSet{
		ObjectMeta: metav1.ObjectMeta{
			Name:      mpiJob.Name + workerSuffix,
			Namespace: mpiJob.Namespace,
			Labels:    labels,
			OwnerReferences: []metav1.OwnerReference{
				*metav1.NewControllerRef(mpiJob, kubeflow.SchemeGroupVersionKind),
			},
		},
		Spec: appsv1.StatefulSetSpec{
			PodManagementPolicy: appsv1.ParallelPodManagement,
			Replicas:            &desiredReplicas,
			Selector: &metav1.LabelSelector{
				MatchLabels: labels,
			},
			ServiceName: mpiJob.Name + workerSuffix,
			Template:    *podSpec,
		},
	}
}
// newWorker creates a new worker StatefulSet for an MPIJob resource. It also
// sets the appropriate OwnerReferences on the resource so handleObject can
// discover the MPIJob resource that 'owns' it.
newWorker creates a new worker StatefulSet for an MPIJob resource. It also
sets the appropriate OwnerReferences on the resource so handleObject can
discover the MPIJob resource that 'owns' it.
// newWorker creates a new worker StatefulSet for an MPIJob resource . It also // sets the appropriate OwnerReferences on the resource so handleObject can // discover the MPIJob resource that ' owns ' it .
func newWorker(mpiJob *kubeflow.MPIJob, desiredReplicas int32, gpus int) *appsv1.StatefulSet {
	labels := map[string]string{
		labelGroupName:   "kubeflow.org",
		labelMPIJobName:  mpiJob.Name,
		labelMPIRoleType: worker,
	}
	podSpec := mpiJob.Spec.Template.DeepCopy()
	if len(podSpec.Labels) == 0 {
		podSpec.Labels = make(map[string]string)
	}
	for key, value := range labels {
		podSpec.Labels[key] = value
	}
	podSpec.Spec.RestartPolicy = corev1.RestartPolicyAlways
	container := podSpec.Spec.Containers[0]
	container.Command = []string{"sleep"}
	container.Args = []string{"365d"}
	if container.Resources.Limits == nil {
		container.Resources.Limits = make(corev1.ResourceList)
	}
	container.Resources.Limits[gpuResourceName] = *resource.NewQuantity(int64(gpus), resource.DecimalExponent)
	container.VolumeMounts = append(container.VolumeMounts, corev1.VolumeMount{
		Name:      configVolumeName,
		MountPath: configMountPath,
	})
	podSpec.Spec.Containers[0] = container
	scriptMode := int32(0555)
	podSpec.Spec.Volumes = append(podSpec.Spec.Volumes, corev1.Volume{
		Name: configVolumeName,
		VolumeSource: corev1.VolumeSource{
			ConfigMap: &corev1.ConfigMapVolumeSource{
				LocalObjectReference: corev1.LocalObjectReference{
					Name: mpiJob.Name + configSuffix,
				},
				Items: []corev1.KeyToPath{
					{
						Key:  kubexecScriptName,
						Path: kubexecScriptName,
						Mode: &scriptMode,
					},
				},
			},
		},
	})
	if mpiJob.Spec.BackoffLimit == nil {
		mpiJob.Spec.BackoffLimit = new(int32)
		*mpiJob.Spec.BackoffLimit = 6
	}
	return &appsv1.StatefulSet{
		ObjectMeta: metav1.ObjectMeta{
			Name:      mpiJob.Name + workerSuffix,
			Namespace: mpiJob.Namespace,
			Labels:    labels,
			OwnerReferences: []metav1.OwnerReference{
				*metav1.NewControllerRef(mpiJob, kubeflow.SchemeGroupVersionKind),
			},
		},
		Spec: appsv1.StatefulSetSpec{
			PodManagementPolicy: appsv1.ParallelPodManagement,
			Replicas:            &desiredReplicas,
			Selector: &metav1.LabelSelector{
				MatchLabels: labels,
			},
			ServiceName: mpiJob.Name + workerSuffix,
			Template:    *podSpec,
		},
	}
}
func newWorker ( mpiJob * kubeflow . MPIJob , desiredReplicas int32 , gpus int ) * appsv1 . StatefulSet { labels := map [ string ] string { labelGroupName : " " , labelMPIJobName : mpiJob . Name , labelMPIRoleType : worker , } podSpec := mpiJob . Spec . Template . DeepCopy ( ) if len ( podSpec . Labels ) == 0 { podSpec . Labels = make ( map [ string ] string ) } for key , value := range labels { podSpec . Labels [ key ] = value } podSpec . Spec . RestartPolicy = corev1 . RestartPolicyAlways container := podSpec . Spec . Containers [ 0 ] container . Command = [ ] string { " " } container . Args = [ ] string { " " } if container . Resources . Limits == nil { container . Resources . Limits = make ( corev1 . ResourceList ) } container . Resources . Limits [ gpuResourceName ] = * resource . NewQuantity ( int64 ( gpus ) , resource . DecimalExponent ) container . VolumeMounts = append ( container . VolumeMounts , corev1 . VolumeMount { Name : configVolumeName , MountPath : configMountPath , } ) podSpec . Spec . Containers [ 0 ] = container scriptMode := int32 ( 0555 ) podSpec . Spec . Volumes = append ( podSpec . Spec . Volumes , corev1 . Volume { Name : configVolumeName , VolumeSource : corev1 . VolumeSource { ConfigMap : & corev1 . ConfigMapVolumeSource { LocalObjectReference : corev1 . LocalObjectReference { Name : mpiJob . Name + configSuffix , } , Items : [ ] corev1 . KeyToPath { { Key : kubexecScriptName , Path : kubexecScriptName , Mode : & scriptMode , } , } , } , } , } ) if mpiJob . Spec . BackoffLimit == nil { mpiJob . Spec . BackoffLimit = new ( int32 ) * mpiJob . Spec . BackoffLimit = 6 } return & appsv1 . StatefulSet { ObjectMeta : metav1 . ObjectMeta { Name : mpiJob . Name + workerSuffix , Namespace : mpiJob . Namespace , Labels : labels , OwnerReferences : [ ] metav1 . OwnerReference { * metav1 . NewControllerRef ( mpiJob , kubeflow . SchemeGroupVersionKind ) , } , } , Spec : appsv1 . StatefulSetSpec { PodManagementPolicy : appsv1 . ParallelPodManagement , Replicas : & desiredReplicas , Selector : & metav1 . LabelSelector { MatchLabels : labels , } , ServiceName : mpiJob . Name + workerSuffix , Template : * podSpec , } , } }
newWorker creates a new worker StatefulSet for an MPIJob resource.
newWorker creates a new worker StatefulSet for an MPIJob resource .
// keep the labels which are set in PodTemplate // always set restartPolicy to restartAlways for statefulset // We need the kubexec.sh script here because Open MPI checks for the path // in every rank. // set default BackoofLimit