add job suspend run Policy

PeterChg · PeterChg · commit 66db21b7efd3 · 2022-05-17T10:37:33.000+08:00
add job partial success status
diff --git a/go.sum b/go.sum
@@ -564,6 +564,7 @@ golang.org/x/mod v0.0.0-20190513183733-4bf6d317e70e/go.mod h1:mXi4GBBbnImb6dmsKG
 golang.org/x/mod v0.1.0/go.mod h1:0QHyrYULN0/3qlju5TqG8bIK38QM8yzMo5ekMj3DlcY=
 golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg=
 golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
+golang.org/x/mod v0.3.0 h1:RM4zey1++hCTbCVQfnWeKs9/IEsaBLA8vTkd0WVtmH4=
 golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
@@ -804,6 +805,7 @@ k8s.io/component-base v0.19.2/go.mod h1:g5LrsiTiabMLZ40AR6Hl45f088DevyGY+cCE2agE
 k8s.io/component-base v0.19.6 h1:V76d3rIEWvP95peWgRycKslQnEwlaPy4UORvh3+YBbU=
 k8s.io/component-base v0.19.6/go.mod h1:8Btsf8J00/fVDa/YFmXjei7gVkcFrlKZXjSeP4SZNJg=
 k8s.io/gengo v0.0.0-20200413195148-3a45101e95ac/go.mod h1:ezvh/TsK7cY6rbqRK0oQQ8IAqLxYwwyPxAX1Pzy0ii0=
+k8s.io/gengo v0.0.0-20200428234225-8167cfdcfc14 h1:t4L10Qfx/p7ASH3gXCdIUtPbbIuegCoUJf3TMSFekjw=
 k8s.io/gengo v0.0.0-20200428234225-8167cfdcfc14/go.mod h1:ezvh/TsK7cY6rbqRK0oQQ8IAqLxYwwyPxAX1Pzy0ii0=
 k8s.io/klog v1.0.0 h1:Pt+yjF5aB1xDSVbau4VsWe+dQNzA0qv1LlXdC2dF6Q8=
 k8s.io/klog v1.0.0/go.mod h1:4Bi6QPql/J/LkTDqv7R/cd3hPo4k2DG6Ptcz060Ez5I=
diff --git a/pkg/apis/common/v1/interface.go b/pkg/apis/common/v1/interface.go
@@ -64,7 +64,7 @@ type ControllerInterface interface {
 	// It will requeue the job in case of an error while creating/deleting pods.
 	// Common implementation will be provided and User can still override this to implement their own reconcile logic
 	ReconcilePods(job interface{}, jobStatus *JobStatus, pods []*v1.Pod, rtype ReplicaType, spec *ReplicaSpec,
-		replicas map[ReplicaType]*ReplicaSpec) error
+		replicas map[ReplicaType]*ReplicaSpec, runPolicy *RunPolicy) error
 
 	// ReconcileServices checks and updates services for each given ReplicaSpec.
 	// It will requeue the job in case of an error while creating/deleting services.
diff --git a/pkg/apis/common/v1/openapi_generated.go b/pkg/apis/common/v1/openapi_generated.go
diff --git a/pkg/apis/common/v1/types.go b/pkg/apis/common/v1/types.go
@@ -135,6 +135,17 @@ const (
 	// reached phase failed with no restarting.
 	// The training has failed its execution.
 	JobFailed JobConditionType = "Failed"
+
+	// JobSuspended means sub-resources (e.g. services/pods) of this job
+	// has been terminated.
+	JobSuspended JobConditionType = "Suspended"
+
+	// JobResumed means job Resumed from suspended
+	JobResumed JobConditionType = "Resumed"
+
+	// JobPartialSucceed means all sub-resources (e.g. services/pods) of this job's one worker
+	// reached phase have terminated in success.
+	JobPartialSucceeded JobConditionType = "PartialSucceeded"
 )
 
 // +k8s:openapi-gen=true
@@ -196,6 +207,17 @@ type RunPolicy struct {
 	// SchedulingPolicy defines the policy related to scheduling, e.g. gang-scheduling
 	// +optional
 	SchedulingPolicy *SchedulingPolicy `json:"schedulingPolicy,omitempty"`
+
+	// Suspend specifies whether the Job controller should create Pods or not. If
+	// a Job is created with suspend set to true, no Pods are created by the Job
+	// controller. If a Job is suspended after creation (i.e. the flag goes from
+	// false to true), the Job controller will delete all active Pods associated
+	// with this Job. Users must design their workload to gracefully handle this.
+	// Suspending a Job will reset the StartTime field of the Job, effectively
+	// resetting the ActiveDeadlineSeconds timer too.
+	// Defaults to false.
+	// +optional
+	Suspend *bool `json:"suspend,omitempty"`
 }
 
 // +k8s:openapi-gen=true
diff --git a/pkg/apis/common/v1/zz_generated.deepcopy.go b/pkg/apis/common/v1/zz_generated.deepcopy.go
diff --git a/pkg/apis/common/v1/zz_generated.defaults.go b/pkg/apis/common/v1/zz_generated.defaults.go
diff --git a/pkg/controller.v1/common/job.go b/pkg/controller.v1/common/job.go
@@ -270,7 +270,7 @@ func (jc *JobController) ReconcileJobs(
 
 		// Diff current active pods/services with replicas.
 		for rtype, spec := range replicas {
-			err := jc.Controller.ReconcilePods(metaObject, &jobStatus, pods, rtype, spec, replicas)
+			err := jc.Controller.ReconcilePods(metaObject, &jobStatus, pods, rtype, spec, replicas, runPolicy)
 			if err != nil {
 				log.Warnf("ReconcilePods error %v", err)
 				return err
diff --git a/pkg/controller.v1/common/pod.go b/pkg/controller.v1/common/pod.go
@@ -274,7 +274,8 @@ func (jc *JobController) ReconcilePods(
 	pods []*v1.Pod,
 	rType apiv1.ReplicaType,
 	spec *apiv1.ReplicaSpec,
-	replicas map[apiv1.ReplicaType]*apiv1.ReplicaSpec) error {
+	replicas map[apiv1.ReplicaType]*apiv1.ReplicaSpec,
+	runPolicy *apiv1.RunPolicy) error {
 
 	rt := strings.ToLower(string(rType))
 	metaObject, ok := job.(metav1.Object)
@@ -317,6 +318,10 @@ func (jc *JobController) ReconcilePods(
 		} else if len(podSlice) == 0 {
 			logger.Infof("Need to create new pod: %s-%d", rt, index)
 
+			if JobSuspended(runPolicy) || commonutil.IsSuspended(*jobStatus) {
+				logger.Warningf("job is Suspended %s/%s", metaObject.GetNamespace(), metaObject.GetName())
+				continue
+			}
 			// check if this replica is the master role
 			masterRole = jc.Controller.IsMasterRole(replicas, rType, index)
 			err = jc.createNewPod(job, rt, index, spec, masterRole, replicas)
@@ -328,7 +333,7 @@ func (jc *JobController) ReconcilePods(
 			pod := podSlice[0]
 
 			// check if the index is in the valid range, if not, we should kill the pod
-			if index < 0 || index >= numReplicas {
+			if index < 0 || index >= numReplicas || JobSuspended(runPolicy) {
 				err = jc.PodControl.DeletePod(pod.Namespace, pod.Name, runtimeObject)
 				if err != nil {
 					return err
diff --git a/pkg/controller.v1/common/util.go b/pkg/controller.v1/common/util.go
@@ -143,3 +143,9 @@ func CalcPGMinResources(minMember int32, replicas map[apiv1.ReplicaType]*apiv1.R
 
 	return &minAvailableTasksRes
 }
+
+// JobSuspended returns whether a Job is suspended while taking the feature
+// gate into account.
+func JobSuspended(runPolicy *apiv1.RunPolicy) bool {
+	return runPolicy.Suspend != nil && *runPolicy.Suspend
+}
diff --git a/pkg/core/job.go b/pkg/core/job.go
@@ -64,7 +64,7 @@ func RecordAbnormalPods(activePods []*v1.Pod, object runtime.Object, recorder re
 
 // PastActiveDeadline checks if job has ActiveDeadlineSeconds field set and if it is exceeded.
 func PastActiveDeadline(runPolicy *apiv1.RunPolicy, jobStatus apiv1.JobStatus) bool {
-	if runPolicy.ActiveDeadlineSeconds == nil || jobStatus.StartTime == nil {
+	if runPolicy.ActiveDeadlineSeconds == nil || jobStatus.StartTime == nil || (runPolicy.Suspend != nil && *runPolicy.Suspend) {
 		return false
 	}
 	now := metav1.Now()
diff --git a/pkg/util/status.go b/pkg/util/status.go
@@ -17,7 +17,12 @@ const (
 	JobFailedReason = "JobFailed"
 	// JobRestarting is added in a job when it is restarting.
 	JobRestartingReason = "JobRestarting"
-
+	// JobSuspend is added in a job when it is been suspended.
+	JobSuspendedReason = "JobSuspended"
+	// JobResumed is added in when job Resumed.
+	JobResumedReason = "JobResumed"
+	// JobPartialSucceed is added in when job partial successed.
+	JobPartialSucceededReason = "JobPartialSucceeded"
 	// labels for pods and servers.
 
 )
@@ -32,6 +37,16 @@ func IsFailed(status apiv1.JobStatus) bool {
 	return hasCondition(status, apiv1.JobFailed)
 }
 
+// IsSuspended checks if the job is suspended
+func IsSuspended(status apiv1.JobStatus) bool {
+	return hasCondition(status, apiv1.JobSuspended)
+}
+
+// IsSuspended checks if the job is suspended
+func IsPartialSucceeded(status apiv1.JobStatus) bool {
+	return hasCondition(status, apiv1.JobPartialSucceeded)
+}
+
 // UpdateJobConditions adds to the jobStatus a new condition if needed, with the conditionType, reason, and message
 func UpdateJobConditions(jobStatus *apiv1.JobStatus, conditionType apiv1.JobConditionType, reason, message string) error {
 	condition := newCondition(conditionType, reason, message)
@@ -103,16 +118,29 @@ func filterOutCondition(conditions []apiv1.JobCondition, condType apiv1.JobCondi
 		if condType == apiv1.JobRestarting && c.Type == apiv1.JobRunning {
 			continue
 		}
-		if condType == apiv1.JobRunning && c.Type == apiv1.JobRestarting {
+		if condType == apiv1.JobRunning && (c.Type == apiv1.JobRestarting || c.Type == apiv1.JobResumed) {
+			continue
+		}
+
+		if condType == apiv1.JobSuspended && c.Type == apiv1.JobResumed {
+			continue
+		}
+
+		if condType == apiv1.JobResumed && c.Type == apiv1.JobSuspended {
+			continue
+		}
+
+		if condType == apiv1.JobRestarting && c.Type == apiv1.JobPartialSucceeded {
 			continue
 		}
 
 		if c.Type == condType {
 			continue
 		}
 
-		// Set the running condition status to be false when current condition failed or succeeded
-		if (condType == apiv1.JobFailed || condType == apiv1.JobSucceeded) && c.Type == apiv1.JobRunning {
+		// Set the running condition status to be false when current condition failed, succeeded or suspended
+		if (condType == apiv1.JobFailed || condType == apiv1.JobSucceeded ||
+			condType == apiv1.JobSuspended) && c.Type == apiv1.JobRunning {
 			c.Status = v1.ConditionFalse
 		}
 

Original file line number	Diff line number	Diff line change
`@@ -143,3 +143,9 @@ func CalcPGMinResources(minMember int32, replicas map[apiv1.ReplicaType]*apiv1.R`
`143`	`143`
`144`	`144`	`return &minAvailableTasksRes`
`145`	`145`	`}`
	`146`	`+`
	`147`	`+// JobSuspended returns whether a Job is suspended while taking the feature`
	`148`	`+// gate into account.`
	`149`	`+func JobSuspended(runPolicy *apiv1.RunPolicy) bool {`
	`150`	`+ return runPolicy.Suspend != nil && *runPolicy.Suspend`
	`151`	`+}`
Original file line number	Diff line number	Diff line change
`@@ -64,7 +64,7 @@ func RecordAbnormalPods(activePods []*v1.Pod, object runtime.Object, recorder re`
`64`	`64`
`65`	`65`	`// PastActiveDeadline checks if job has ActiveDeadlineSeconds field set and if it is exceeded.`
`66`	`66`	`func PastActiveDeadline(runPolicy *apiv1.RunPolicy, jobStatus apiv1.JobStatus) bool {`
`67`		`- if runPolicy.ActiveDeadlineSeconds == nil \|\| jobStatus.StartTime == nil {`
	`67`	`+ if runPolicy.ActiveDeadlineSeconds == nil \|\| jobStatus.StartTime == nil \|\| (runPolicy.Suspend != nil && *runPolicy.Suspend) {`
`68`	`68`	`return false`
`69`	`69`	`}`
`70`	`70`	`now := metav1.Now()`