feat: Support running (kubeflow#894)

* feat: Support running Signed-off-by: Ce Gao <gaoce@caicloud.io> * fix: Do not mark trial running when the job is created Signed-off-by: Ce Gao <gaoce@caicloud.io> * fix: Add nil pointer check Signed-off-by: Ce Gao <gaoce@caicloud.io> * fix: Avoid nil pointer Signed-off-by: Ce Gao <gaoce@caicloud.io> * fix: Update Signed-off-by: Ce Gao <gaoce@caicloud.io>
arrikto · Dec 6, 2019 · 5d46799 · 5d46799
1 parent 83eba02
commit 5d46799
Show file tree

Hide file tree

Showing 5 changed files with 31 additions and 7 deletions.
diff --git a/pkg/controller.v1alpha3/trial/trial_controller.go b/pkg/controller.v1alpha3/trial/trial_controller.go
@@ -271,8 +271,6 @@ func (r *ReconcileTrial) reconcileJob(instance *trialsv1alpha3.Trial, desiredJob
 			}
 			eventMsg := fmt.Sprintf("Job %s has been created", desiredJob.GetName())
 			r.recorder.Eventf(instance, corev1.EventTypeNormal, JobCreatedReason, eventMsg)
-			msg := "Trial is running"
-			instance.MarkTrialStatusRunning(TrialRunningReason, msg)
 		} else {
 			logger.Error(err, "Trial Get error")
 			return nil, err

diff --git a/pkg/controller.v1alpha3/trial/trial_controller_consts.go b/pkg/controller.v1alpha3/trial/trial_controller_consts.go
@@ -17,5 +17,6 @@ const (
 	JobSucceededReason          = "JobSucceeded"
 	JobMetricsUnavailableReason = "MetricsUnavailable"
 	JobFailedReason             = "JobFailed"
+	JobRunningReason            = "JobRunning"
 	ReconcileFailedReason       = "ReconcileFailed"
 )
diff --git a/pkg/controller.v1alpha3/trial/trial_controller_util.go b/pkg/controller.v1alpha3/trial/trial_controller_util.go
@@ -36,6 +36,9 @@ const (
 )
 
 func (r *ReconcileTrial) UpdateTrialStatusCondition(instance *trialsv1alpha3.Trial, deployedJob *unstructured.Unstructured, jobCondition *commonv1.JobCondition) {
+	if jobCondition == nil || instance == nil || deployedJob == nil {
+		return
+	}
 	now := metav1.Now()
 	jobConditionType := (*jobCondition).Type
 	if jobConditionType == commonv1.JobSucceeded {
@@ -63,8 +66,17 @@ func (r *ReconcileTrial) UpdateTrialStatusCondition(instance *trialsv1alpha3.Tri
 		eventMsg := fmt.Sprintf("Job %s has failed: %s", deployedJob.GetName(), jobConditionMessage)
 		r.recorder.Eventf(instance, corev1.EventTypeNormal, JobFailedReason, eventMsg)
 		IncreaseTrialsFailedCount()
+	} else if jobConditionType == commonv1.JobRunning {
+		msg := "Trial is running"
+		instance.MarkTrialStatusRunning(TrialRunningReason, msg)
+		jobConditionMessage := (*jobCondition).Message
+		eventMsg := fmt.Sprintf("Job %s is running: %s",
+			deployedJob.GetName(), jobConditionMessage)
+		r.recorder.Eventf(instance, corev1.EventTypeNormal,
+			JobRunningReason, eventMsg)
+		// TODO(gaocegege): Should we maintain a TrialsRunningCount?
 	}
-	//else nothing to do
+	// else nothing to do
 	return
 }
 
@@ -118,6 +130,9 @@ func (r *ReconcileTrial) updateFinalizers(instance *trialsv1alpha3.Trial, finali
 }
 
 func isTrialObservationAvailable(instance *trialsv1alpha3.Trial) bool {
+	if instance == nil {
+		return false
+	}
 	objectiveMetricName := instance.Spec.Objective.ObjectiveMetricName
 	if instance.Status.Observation != nil && instance.Status.Observation.Metrics != nil {
 		for _, metric := range instance.Status.Observation.Metrics {
@@ -130,6 +145,9 @@ func isTrialObservationAvailable(instance *trialsv1alpha3.Trial) bool {
 }
 
 func isTrialComplete(instance *trialsv1alpha3.Trial, jobCondition *commonv1.JobCondition) bool {
+	if jobCondition == nil || instance == nil {
+		return false
+	}
 	jobConditionType := (*jobCondition).Type
 	if jobConditionType == commonv1.JobSucceeded && isTrialObservationAvailable(instance) {
 		return true
@@ -142,6 +160,9 @@ func isTrialComplete(instance *trialsv1alpha3.Trial, jobCondition *commonv1.JobC
 }
 
 func isJobSucceeded(jobCondition *commonv1.JobCondition) bool {
+	if jobCondition == nil {
+		return false
+	}
 	jobConditionType := (*jobCondition).Type
 	if jobConditionType == commonv1.JobSucceeded {
 		return true

diff --git a/pkg/job/v1alpha3/job/job.go b/pkg/job/v1alpha3/job/job.go
@@ -28,9 +28,10 @@ func (j Job) GetDeployedJobStatus(
 			log.Error(unerr, "NestedFieldCopy unstructured to status error")
 			return nil, unerr
 		}
-		log.Info("NestedFieldCopy unstructured to status error",
-			"err", "Status is not found in job")
-		return &jobCondition, nil
+		// Job does not have the running condition in status, thus we think
+		// the job is running when it is created.
+		log.Info("NestedFieldCopy", "err", "status cannot be found in job")
+		return nil, nil
 	}
 
 	statusMap := status.(map[string]interface{})

diff --git a/pkg/job/v1alpha3/kubeflow/kubeflow.go b/pkg/job/v1alpha3/kubeflow/kubeflow.go
@@ -36,7 +36,7 @@ func (k Kubeflow) GetDeployedJobStatus(
 		}
 		log.Info("NestedFieldCopy unstructured to status error",
 			"err", "Status is not found in job")
-		return &jobCondition, nil
+		return nil, nil
 	}
 
 	statusMap := status.(map[string]interface{})
@@ -46,10 +46,13 @@ func (k Kubeflow) GetDeployedJobStatus(
 		log.Error(err, "Convert unstructured to status error")
 		return nil, err
 	}
+	// Get the latest condition and set it to jobCondition.
 	if len(jobStatus.Conditions) > 0 {
 		lc := jobStatus.Conditions[len(jobStatus.Conditions)-1]
 		jobCondition.Type = lc.Type
 		jobCondition.Message = lc.Message
+		jobCondition.Status = lc.Status
+		jobCondition.Reason = lc.Reason
 	}
 
 	return &jobCondition, nil