feat: add terminalState to jobset status

api/jobset/v1alpha2/jobset_types.go

@@ -50,6 +50,8 @@ type JobSetConditionType string
 
 // These are built-in conditions of a JobSet.
 const (
+	// JobSetRunning means the job is running.
+	JobSetRunning JobSetConditionType = "Running"
 	// JobSetCompleted means the job has completed its execution.
 	JobSetCompleted JobSetConditionType = "Completed"
 	// JobSetFailed means the job has failed its execution.
@@ -134,6 +136,10 @@ type JobSetStatus struct {
 	// RestartsCountTowardsMax tracks the number of times the JobSet has restarted that counts towards the maximum allowed number of restarts.
 	RestartsCountTowardsMax int32 `json:"restartsCountTowardsMax,omitempty"`
 
+	// Phase of the JobSet.
+	// +kubebuilder:default="Running"
+	Phase string `json:"phase,omitempty"`
+
 	// ReplicatedJobsStatus track the number of JobsReady for each replicatedJob.
 	// +optional
 	// +listType=map
@@ -169,6 +175,7 @@ type ReplicatedJobStatus struct {
 // +k8s:openapi-gen=true
 // +kubebuilder:object:root=true
 // +kubebuilder:subresource:status
+// +kubebuilder:printcolumn:name="Phase",JSONPath=".status.phase",type=string,description="Phase of the JobSet"
 // +kubebuilder:printcolumn:name="Restarts",JSONPath=".status.restarts",type=string,description="Number of restarts"
 // +kubebuilder:printcolumn:name="Completed",type="string",priority=0,JSONPath=".status.conditions[?(@.type==\"Completed\")].status"
 // +kubebuilder:printcolumn:name="Suspended",type="string",JSONPath=".spec.suspend",description="JobSet suspended"

config/components/crd/bases/jobset.x-k8s.io_jobsets.yaml

@@ -15,6 +15,10 @@ spec:
   scope: Namespaced
   versions:
   - additionalPrinterColumns:
+    - description: Phase of the JobSet
+      jsonPath: .status.phase
+      name: Phase
+      type: string
     - description: Number of restarts
       jsonPath: .status.restarts
       name: Restarts
@@ -8487,6 +8491,10 @@ spec:
                 x-kubernetes-list-map-keys:
                 - type
                 x-kubernetes-list-type: map
+              phase:
+                default: Running
+                description: Phase of the JobSet.
+                type: string
               replicatedJobsStatus:
                 description: ReplicatedJobsStatus track the number of JobsReady for
                   each replicatedJob.

hack/python-sdk/swagger.json

@@ -179,6 +179,10 @@
           ],
           "x-kubernetes-list-type": "map"
         },
+        "phase": {
+          "description": "Phase of the JobSet.",
+          "type": "string"
+        },
         "replicatedJobsStatus": {
           "description": "ReplicatedJobsStatus track the number of JobsReady for each replicatedJob.",
           "type": "array",

pkg/controllers/failure_policy.go

@@ -250,6 +250,7 @@ func makeFailedConditionOpts(reason, msg string) *conditionOpts {
 			Reason:  reason,
 			Message: msg,
 		},
+		phase:     string(jobset.JobSetFailed),
 		eventType: corev1.EventTypeWarning,
 	}
 }

pkg/controllers/jobset_controller.go

@@ -149,7 +149,7 @@ func (r *JobSetReconciler) reconcile(ctx context.Context, js *jobset.JobSet, upd
 
 	// Calculate JobsReady and update statuses for each ReplicatedJob.
 	rjobStatuses := r.calculateReplicatedJobStatuses(ctx, js, ownedJobs)
-	updateReplicatedJobsStatuses(ctx, js, rjobStatuses, updateStatusOpts)
+	updateReplicatedJobsStatuses(js, rjobStatuses, updateStatusOpts)
 
 	// If JobSet is already completed or failed, clean up active child jobs and requeue if TTLSecondsAfterFinished is set.
 	if jobSetFinished(js) {
@@ -185,7 +185,7 @@ func (r *JobSetReconciler) reconcile(ctx context.Context, js *jobset.JobSet, upd
 
 	// If any jobs have succeeded, execute the JobSet success policy.
 	if len(ownedJobs.successful) > 0 {
-		if completed := executeSuccessPolicy(ctx, js, ownedJobs, updateStatusOpts); completed {
+		if completed := executeSuccessPolicy(js, ownedJobs, updateStatusOpts); completed {
 			return ctrl.Result{}, nil
 		}
 	}
@@ -304,7 +304,7 @@ func (r *JobSetReconciler) getChildJobs(ctx context.Context, js *jobset.JobSet)
 }
 
 // updateReplicatedJobsStatuses updates the replicatedJob statuses if they have changed.
-func updateReplicatedJobsStatuses(ctx context.Context, js *jobset.JobSet, statuses []jobset.ReplicatedJobStatus, updateStatusOpts *statusUpdateOpts) {
+func updateReplicatedJobsStatuses(js *jobset.JobSet, statuses []jobset.ReplicatedJobStatus, updateStatusOpts *statusUpdateOpts) {
 	// If replicated job statuses haven't changed, there's nothing to do here.
 	if replicatedJobStatusesEqual(js.Status.ReplicatedJobsStatus, statuses) {
 		return
@@ -630,7 +630,7 @@ func (r *JobSetReconciler) createHeadlessSvcIfNecessary(ctx context.Context, js
 // executeSuccessPolicy checks the completed jobs against the jobset success policy
 // and updates the jobset status to completed if the success policy conditions are met.
 // Returns a boolean value indicating if the jobset was completed or not.
-func executeSuccessPolicy(ctx context.Context, js *jobset.JobSet, ownedJobs *childJobs, updateStatusOpts *statusUpdateOpts) bool {
+func executeSuccessPolicy(js *jobset.JobSet, ownedJobs *childJobs, updateStatusOpts *statusUpdateOpts) bool {
 	if numJobsMatchingSuccessPolicy(js, ownedJobs.successful) >= numJobsExpectedToSucceed(js) {
 		setJobSetCompletedCondition(js, updateStatusOpts)
 		return true
@@ -867,14 +867,15 @@ func enqueueEvent(updateStatusOpts *statusUpdateOpts, event *eventParams) {
 // function parameters for setCondition
 type conditionOpts struct {
 	eventType string
+	phase     string
 	condition *metav1.Condition
 }
 
-// setCondition will add a new condition to the JobSet status (or update an existing one),
+// setCondition will add a new condition and phase to the JobSet status (or update an existing one),
 // and enqueue an event for emission if the status update succeeds at the end of the reconcile.
 func setCondition(js *jobset.JobSet, condOpts *conditionOpts, updateStatusOpts *statusUpdateOpts) {
-	// Return early if no status update is required for this condition.
-	if !updateCondition(js, condOpts) {
+	// Return early if no status update is required for this condition and phase.
+	if !updateConditionAndPhase(js, condOpts) {
 		return
 	}
 
@@ -897,12 +898,14 @@ func setCondition(js *jobset.JobSet, condOpts *conditionOpts, updateStatusOpts *
 	enqueueEvent(updateStatusOpts, event)
 }
 
-// updateCondition accepts a given condition and does one of the following:
+// updateConditionAndPhase accepts a condition and a phase, and does the following:
 //  1. If an identical condition already exists, do nothing and return false (indicating
 //     no change was made).
 //  2. If a condition of the same type exists but with a different status, update
 //     the condition in place and return true (indicating a condition change was made).
-func updateCondition(js *jobset.JobSet, opts *conditionOpts) bool {
+//  3. If the specified phase is different from the current phase of the JobSet,
+//     update the JobSet Status Phase
+func updateConditionAndPhase(js *jobset.JobSet, opts *conditionOpts) bool {
 	if opts == nil || opts.condition == nil {
 		return false
 	}
@@ -941,6 +944,13 @@ func updateCondition(js *jobset.JobSet, opts *conditionOpts) bool {
 		js.Status.Conditions = append(js.Status.Conditions, newCond)
 		shouldUpdate = true
 	}
+
+	// Update the JobSet Status Phase if necessary.
+	if opts.phase != "" && js.Status.Phase != opts.phase {
+		js.Status.Phase = opts.phase
+		shouldUpdate = true
+	}
+
 	return shouldUpdate
 }
 
@@ -970,6 +980,7 @@ func makeCompletedConditionsOpts() *conditionOpts {
 			Reason:  constants.AllJobsCompletedReason,
 			Message: constants.AllJobsCompletedMessage,
 		},
+		phase: string(jobset.JobSetCompleted),
 	}
 }
 
@@ -984,6 +995,7 @@ func makeSuspendedConditionOpts() *conditionOpts {
 			Reason:             constants.JobSetSuspendedReason,
 			Message:            constants.JobSetSuspendedMessage,
 		},
+		phase: string(jobset.JobSetSuspended),
 	}
 }
 
@@ -998,6 +1010,7 @@ func makeResumedConditionOpts() *conditionOpts {
 			Reason:             constants.JobSetResumedReason,
 			Message:            constants.JobSetResumedMessage,
 		},
+		phase: string(jobset.JobSetRunning),
 	}
 }
 

pkg/controllers/jobset_controller_test.go

@@ -696,7 +696,7 @@ func TestUpdateConditions(t *testing.T) {
 				ReplicatedJob(testutils.MakeReplicatedJob(replicatedJobName).
 					Job(testutils.MakeJobTemplate(jobName, ns).Obj()).
 					Replicas(1).
-					Obj()).Obj(),
+					Obj()).Phase(jobset.JobSetRunning).Obj(),
 			opts:           makeCompletedConditionsOpts(),
 			expectedUpdate: true,
 		},
@@ -706,7 +706,7 @@ func TestUpdateConditions(t *testing.T) {
 				ReplicatedJob(testutils.MakeReplicatedJob(replicatedJobName).
 					Job(testutils.MakeJobTemplate(jobName, ns).Obj()).
 					Replicas(1).
-					Obj()).Obj(),
+					Obj()).Phase(jobset.JobSetRunning).Obj(),
 			opts:           makeSuspendedConditionOpts(),
 			expectedUpdate: true,
 		},
@@ -716,7 +716,7 @@ func TestUpdateConditions(t *testing.T) {
 				ReplicatedJob(testutils.MakeReplicatedJob(replicatedJobName).
 					Job(testutils.MakeJobTemplate(jobName, ns).Obj()).
 					Replicas(1).
-					Obj()).
+					Obj()).Phase(jobset.JobSetRunning).
 				Conditions([]metav1.Condition{
 					// JobSet is currrently suspended.
 					{
@@ -736,7 +736,7 @@ func TestUpdateConditions(t *testing.T) {
 				ReplicatedJob(testutils.MakeReplicatedJob(replicatedJobName).
 					Job(testutils.MakeJobTemplate(jobName, ns).Obj()).
 					Replicas(1).
-					Obj()).
+					Obj()).Phase(jobset.JobSetCompleted).
 				Conditions([]metav1.Condition{
 					// JobSet is completed..
 					{
@@ -752,7 +752,7 @@ func TestUpdateConditions(t *testing.T) {
 	}
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
-			gotUpdate := updateCondition(tc.js, tc.opts)
+			gotUpdate := updateConditionAndPhase(tc.js, tc.opts)
 			if gotUpdate != tc.expectedUpdate {
 				t.Errorf("updateCondition return mismatch (want: %v, got %v)", tc.expectedUpdate, gotUpdate)
 			}

pkg/util/testing/wrappers.go

@@ -159,6 +159,12 @@ func (j *JobSetWrapper) FailedCondition(failedAt metav1.Time) *JobSetWrapper {
 	return j
 }
 
+// Phase sets the value of JobSet.Status.Phase.
+func (j *JobSetWrapper) Phase(phase jobset.JobSetConditionType) *JobSetWrapper {
+	j.Status.Phase = string(phase)
+	return j
+}
+
 func (j *JobSetWrapper) DeletionTimestamp(deletionTimestamp *metav1.Time) *JobSetWrapper {
 	j.ObjectMeta.DeletionTimestamp = deletionTimestamp
 	return j

sdk/python/docs/JobsetV1alpha2JobSetStatus.md

@@ -5,6 +5,7 @@ JobSetStatus defines the observed state of JobSet
 Name | Type | Description | Notes
 ------------ | ------------- | ------------- | -------------
 **conditions** | [**list[V1Condition]**](V1Condition.md) |  | [optional] 
+**phase** | **str** | Phase of the JobSet. | [optional] 
 **replicated_jobs_status** | [**list[JobsetV1alpha2ReplicatedJobStatus]**](JobsetV1alpha2ReplicatedJobStatus.md) | ReplicatedJobsStatus track the number of JobsReady for each replicatedJob. | [optional] 
 **restarts** | **int** | Restarts tracks the number of times the JobSet has restarted (i.e. recreated in case of RecreateAll policy). | [optional] 
 **restarts_count_towards_max** | **int** | RestartsCountTowardsMax tracks the number of times the JobSet has restarted that counts towards the maximum allowed number of restarts. | [optional] 

sdk/python/jobset/models/jobset_v1alpha2_job_set_status.py

@@ -34,32 +34,37 @@ class JobsetV1alpha2JobSetStatus(object):
     """
     openapi_types = {
         'conditions': 'list[V1Condition]',
+        'phase': 'str',
         'replicated_jobs_status': 'list[JobsetV1alpha2ReplicatedJobStatus]',
         'restarts': 'int',
         'restarts_count_towards_max': 'int'
     }
 
     attribute_map = {
         'conditions': 'conditions',
+        'phase': 'phase',
         'replicated_jobs_status': 'replicatedJobsStatus',
         'restarts': 'restarts',
         'restarts_count_towards_max': 'restartsCountTowardsMax'
     }
 
-    def __init__(self, conditions=None, replicated_jobs_status=None, restarts=None, restarts_count_towards_max=None, local_vars_configuration=None):  # noqa: E501
+    def __init__(self, conditions=None, phase=None, replicated_jobs_status=None, restarts=None, restarts_count_towards_max=None, local_vars_configuration=None):  # noqa: E501
         """JobsetV1alpha2JobSetStatus - a model defined in OpenAPI"""  # noqa: E501
         if local_vars_configuration is None:
             local_vars_configuration = Configuration()
         self.local_vars_configuration = local_vars_configuration
 
         self._conditions = None
+        self._phase = None
         self._replicated_jobs_status = None
         self._restarts = None
         self._restarts_count_towards_max = None
         self.discriminator = None
 
         if conditions is not None:
             self.conditions = conditions
+        if phase is not None:
+            self.phase = phase
         if replicated_jobs_status is not None:
             self.replicated_jobs_status = replicated_jobs_status
         if restarts is not None:
@@ -88,6 +93,29 @@ def conditions(self, conditions):
 
         self._conditions = conditions
 
+    @property
+    def phase(self):
+        """Gets the phase of this JobsetV1alpha2JobSetStatus.  # noqa: E501
+
+        Phase of the JobSet.  # noqa: E501
+
+        :return: The phase of this JobsetV1alpha2JobSetStatus.  # noqa: E501
+        :rtype: str
+        """
+        return self._phase
+
+    @phase.setter
+    def phase(self, phase):
+        """Sets the phase of this JobsetV1alpha2JobSetStatus.
+
+        Phase of the JobSet.  # noqa: E501
+
+        :param phase: The phase of this JobsetV1alpha2JobSetStatus.  # noqa: E501
+        :type: str
+        """
+
+        self._phase = phase
+
     @property
     def replicated_jobs_status(self):
         """Gets the replicated_jobs_status of this JobsetV1alpha2JobSetStatus.  # noqa: E501

sdk/python/test/test_jobset_v1alpha2_job_set.py

@@ -76,6 +76,7 @@ def make_instance(self, include_optional):
                     conditions = [
                         None
                         ], 
+                    phase = '0', 
                     replicated_jobs_status = [
                         jobset.models.jobset_v1alpha2_replicated_job_status.JobsetV1alpha2ReplicatedJobStatus(
                             active = 56, 

sdk/python/test/test_jobset_v1alpha2_job_set_list.py

@@ -79,6 +79,7 @@ def make_instance(self, include_optional):
                             conditions = [
                                 None
                                 ], 
+                            phase = '0', 
                             replicated_jobs_status = [
                                 jobset.models.jobset_v1alpha2_replicated_job_status.JobsetV1alpha2ReplicatedJobStatus(
                                     active = 56, 
@@ -136,6 +137,7 @@ def make_instance(self, include_optional):
                             conditions = [
                                 None
                                 ], 
+                            phase = '0', 
                             replicated_jobs_status = [
                                 jobset.models.jobset_v1alpha2_replicated_job_status.JobsetV1alpha2ReplicatedJobStatus(
                                     active = 56, 

sdk/python/test/test_jobset_v1alpha2_job_set_status.py

@@ -41,6 +41,7 @@ def make_instance(self, include_optional):
                 conditions = [
                     None
                     ], 
+                phase = '0', 
                 replicated_jobs_status = [
                     jobset.models.jobset_v1alpha2_replicated_job_status.JobsetV1alpha2ReplicatedJobStatus(
                         active = 56, 

test/integration/controller/jobset_controller_test.go

@@ -1612,6 +1612,7 @@ var _ = ginkgo.Describe("JobSet controller", func() {
 						LastTransitionTime: metav1.Now(),
 					},
 				},
+				Phase:    string(jobset.JobSetRunning),
 				Restarts: 1,
 				ReplicatedJobsStatus: []jobset.ReplicatedJobStatus{
 					{