Implement suspend semantics #1859
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -26,6 +26,7 @@ import ( | |
| "github.com/kubeflow/training-operator/pkg/core" | ||
| commonutil "github.com/kubeflow/training-operator/pkg/util" | ||
| "github.com/kubeflow/training-operator/pkg/util/k8sutil" | ||
| trainutil "github.com/kubeflow/training-operator/pkg/util/train" | ||
|
|
||
| log "github.com/sirupsen/logrus" | ||
| corev1 "k8s.io/api/core/v1" | ||
|
|
@@ -38,28 +39,30 @@ import ( | |
| volcanov1beta1 "volcano.sh/apis/pkg/apis/scheduling/v1beta1" | ||
| ) | ||
|
|
||
| // DeletePodsAndServices deletes pods and services considering cleanPodPolicy. | ||
| // However, if the job doesn't have Succeeded or Failed condition, it ignores cleanPodPolicy. | ||
| func (jc *JobController) DeletePodsAndServices(runtimeObject runtime.Object, runPolicy *apiv1.RunPolicy, jobStatus apiv1.JobStatus, pods []*corev1.Pod) error { | ||
| if len(pods) == 0 { | ||
| return nil | ||
| } | ||
|
|
||
| // Delete nothing when the cleanPodPolicy is None and the job has Succeeded or Failed condition. | ||
| if commonutil.IsFinished(jobStatus) && *runPolicy.CleanPodPolicy == apiv1.CleanPodPolicyNone { | ||
| return nil | ||
| } | ||
|
|
||
| for _, pod := range pods { | ||
| // Note that pending pod will turn into running once schedulable, | ||
| // not cleaning it may leave orphan running pod in the future, | ||
| // we should treat it equivalent to running phase here. | ||
| if commonutil.IsFinished(jobStatus) && *runPolicy.CleanPodPolicy == apiv1.CleanPodPolicyRunning && pod.Status.Phase != corev1.PodRunning && pod.Status.Phase != corev1.PodPending { | ||
| continue | ||
| } | ||
| if err := jc.PodControl.DeletePod(pod.Namespace, pod.Name, runtimeObject); err != nil { | ||
| return err | ||
| } | ||
| // Pod and service have the same name, thus the service could be deleted using pod's name. | ||
|
There was a problem hiding this comment. side question: why is there a service per pod? That sounds like unnecessary load. There was a problem hiding this comment. IIUC, ml framework configs need different FQDN for each pod. For example, tensorflow There was a problem hiding this comment. wouldn't a single headless Service allow that? similar to this https://kubernetes.io/docs/tasks/job/job-with-pod-to-pod-communication/ There was a problem hiding this comment. Ah, right. |
||
| if err := jc.ServiceControl.DeleteService(pod.Namespace, pod.Name, runtimeObject); err != nil { | ||
| return err | ||
| } | ||
| } | ||
|
|
@@ -117,23 +120,9 @@ func (jc *JobController) ReconcileJobs( | |
| } | ||
|
|
||
| oldStatus := jobStatus.DeepCopy() | ||
| if commonutil.IsFinished(jobStatus) { | ||
| // If the Job is succeed or failed, delete all pods, services, and podGroup. | ||
| if err = jc.CleanUpResources(runPolicy, runtimeObject, metaObject, jobStatus, pods); err != nil { | ||
| return err | ||
| } | ||
|
|
||
|
|
@@ -155,6 +144,44 @@ func (jc *JobController) ReconcileJobs( | |
| return nil | ||
| } | ||
|
|
||
| if trainutil.IsJobSuspended(runPolicy) { | ||
| if err = jc.CleanUpResources(runPolicy, runtimeObject, metaObject, jobStatus, pods); err != nil { | ||
| return err | ||
| } | ||
| for rType := range jobStatus.ReplicaStatuses { | ||
| jobStatus.ReplicaStatuses[rType].Active = 0 | ||
|
There was a problem hiding this comment. I guess this would be set anyway by the other code once the replica pods are cleaned up. This is the approach we take in MPIJob and Job. I would like if we could apply here the same approach There was a problem hiding this comment. We don't have any other codes to reset the Active field, and the training-operator/pkg/controller.v1/common/job.go Lines 143 to 148 in 72f2512 So we need to reset the However, since I think we should reset the There was a problem hiding this comment. is there any code that sets Active to non zero? There was a problem hiding this comment.
Yes, here: -> ->training-operator/pkg/core/status.go Lines 34 to 50 in 72f2512 There was a problem hiding this comment. so those functions wouldn't be called in the next reconcile, essentially resetting the number of active pods? There was a problem hiding this comment.
Yes. If the job is suspended, JobController never calls There was a problem hiding this comment. Instead of setting counts manually, why can't it be derived from the status of all pods? Since we have already cleaned up, active pods will be zero. We can do this refactoring separately as well. There was a problem hiding this comment. We don't have functions to decrease Active count, only have functions to reset the count. So I guess we should refactor ReconcileJob(). However, the refactor will affect Succeeded and Failed conditions, too. So I would like to work on another PR. |
||
| } | ||
| msg := fmt.Sprintf("%s %s is suspended.", jobKind, jobName) | ||
| if commonutil.IsRunning(jobStatus) { | ||
| if err = commonutil.UpdateJobConditions(&jobStatus, apiv1.JobRunning, corev1.ConditionFalse, | ||
| commonutil.NewReason(jobKind, commonutil.JobSuspendedReason), msg); err != nil { | ||
| return err | ||
| } | ||
| } | ||
| // We add the suspended condition to the job only when the job doesn't have a suspended condition. | ||
| if !commonutil.IsSuspended(jobStatus) { | ||
| if err = commonutil.UpdateJobConditions(&jobStatus, apiv1.JobSuspended, corev1.ConditionTrue, | ||
| commonutil.NewReason(jobKind, commonutil.JobSuspendedReason), msg); err != nil { | ||
| return err | ||
| } | ||
| } | ||
| jc.Recorder.Event(runtimeObject, corev1.EventTypeNormal, commonutil.NewReason(jobKind, commonutil.JobSuspendedReason), msg) | ||
| if !reflect.DeepEqual(*oldStatus, jobStatus) { | ||
| return jc.Controller.UpdateJobStatusInApiServer(job, &jobStatus) | ||
|
There was a problem hiding this comment. Would it be possible to let the reconcile function continue, so that other status fields are updated, such as There was a problem hiding this comment.
That makes sense. As I say above, we need to refactor the So I would do your suggestion in follow-ups. |
||
| } | ||
| return nil | ||
| } | ||
| if commonutil.IsSuspended(jobStatus) { | ||
tenzen-y marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| msg := fmt.Sprintf("%s %s is resumed.", jobKind, jobName) | ||
| if err = commonutil.UpdateJobConditions(&jobStatus, apiv1.JobSuspended, corev1.ConditionFalse, | ||
| commonutil.NewReason(jobKind, commonutil.JobResumedReason), msg); err != nil { | ||
| return err | ||
| } | ||
| now := metav1.Now() | ||
| jobStatus.StartTime = &now | ||
| jc.Recorder.Eventf(runtimeObject, corev1.EventTypeNormal, commonutil.NewReason(jobKind, commonutil.JobResumedReason), msg) | ||
| } | ||
|
|
||
| // retrieve the previous number of retry | ||
| previousRetry := jc.WorkQueue.NumRequeues(jobKey) | ||
|
|
||
|
|
@@ -205,7 +232,7 @@ func (jc *JobController) ReconcileJobs( | |
|
|
||
| // If the Job exceeds backoff limit or is past active deadline | ||
| // delete all pods and services, then set the status to failed | ||
| if err := jc.DeletePodsAndServices(runtimeObject, runPolicy, jobStatus, pods); err != nil { | ||
| return err | ||
| } | ||
|
|
||
|
|
@@ -225,7 +252,7 @@ func (jc *JobController) ReconcileJobs( | |
|
|
||
| jc.Recorder.Event(runtimeObject, corev1.EventTypeNormal, commonutil.NewReason(jobKind, commonutil.JobFailedReason), failureMessage) | ||
|
|
||
| if err = commonutil.UpdateJobConditions(&jobStatus, apiv1.JobFailed, corev1.ConditionTrue, commonutil.NewReason(jobKind, commonutil.JobFailedReason), failureMessage); err != nil { | ||
| log.Infof("Append job condition error: %v", err) | ||
| return err | ||
| } | ||
|
|
@@ -344,6 +371,32 @@ func (jc *JobController) ReconcileJobs( | |
| return nil | ||
| } | ||
|
|
||
| func (jc *JobController) CleanUpResources( | ||
| runPolicy *apiv1.RunPolicy, | ||
| runtimeObject runtime.Object, | ||
| metaObject metav1.Object, | ||
| jobStatus apiv1.JobStatus, | ||
| pods []*v1.Pod, | ||
| ) error { | ||
| if err := jc.DeletePodsAndServices(runtimeObject, runPolicy, jobStatus, pods); err != nil { | ||
| return err | ||
| } | ||
| if jc.Config.EnableGangScheduling() { | ||
|
|
||
| jc.Recorder.Event(runtimeObject, corev1.EventTypeNormal, "JobTerminated", "Job has been terminated. Deleting PodGroup") | ||
| if err := jc.DeletePodGroup(metaObject); err != nil { | ||
| jc.Recorder.Eventf(runtimeObject, corev1.EventTypeWarning, "FailedDeletePodGroup", "Error deleting: %v", err) | ||
| return err | ||
| } else { | ||
| jc.Recorder.Eventf(runtimeObject, corev1.EventTypeNormal, "SuccessfulDeletePodGroup", "Deleted PodGroup: %v", metaObject.GetName()) | ||
| } | ||
| } | ||
| if err := jc.CleanupJob(runPolicy, jobStatus, runtimeObject); err != nil { | ||
| return err | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| // ResetExpectations reset the expectation for creates and deletes of pod/service to zero. | ||
| func (jc *JobController) ResetExpectations(jobKey string, replicas map[apiv1.ReplicaType]*apiv1.ReplicaSpec) error { | ||
| var allErrs error | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Here is the core logic for suspend semantics.