Merge pull request #616 from johnugeorge/metricfix

Set trial completion status only after metric collection
kubeflow · May 31, 2019 · 48889cb · 48889cb
2 parents 8056907 + df9f22c
commit 48889cb
Show file tree

Hide file tree

Showing 2 changed files with 33 additions and 4 deletions.
diff --git a/pkg/controller/v1alpha2/trial/trial_controller.go b/pkg/controller/v1alpha2/trial/trial_controller.go
@@ -86,6 +86,19 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error {
 		return err
 	}
 
+	// Watch for changes to Cronjob
+	err = c.Watch(
+		&source.Kind{Type: &batchv1beta.CronJob{}},
+		&handler.EnqueueRequestForOwner{
+			IsController: true,
+			OwnerType:    &trialsv1alpha2.Trial{},
+		})
+
+	if err != nil {
+		log.Error(err, "CronJob watch error")
+		return err
+	}
+
 	for _, gvk := range commonv1alpha2.GetSupportedJobList() {
 		unstructuredJob := &unstructured.Unstructured{}
 		unstructuredJob.SetGroupVersionKind(gvk)
@@ -215,14 +228,18 @@ func (r *ReconcileTrial) reconcileTrial(instance *trialsv1alpha2.Trial) error {
 	//Job already exists
 	//TODO Can desired Spec differ from deployedSpec?
 	if deployedJob != nil {
-		if err = r.UpdateTrialStatusCondition(instance, deployedJob); err != nil {
-			logger.Error(err, "Update trial status condition error")
-			return err
-		}
 		if err = r.UpdateTrialStatusObservation(instance, deployedJob); err != nil {
 			logger.Error(err, "Update trial status observation error")
 			return err
 		}
+		// Update Trial job status only if observation field is available.
+		// This will ensure that trial is set to be complete only if metric is collected atleast once
+		if isTrialObservationAvailable(instance) {
+			if err = r.UpdateTrialStatusCondition(instance, deployedJob); err != nil {
+				logger.Error(err, "Update trial status condition error")
+				return err
+			}
+		}
 	}
 	return nil
 }

diff --git a/pkg/controller/v1alpha2/trial/trial_controller_util.go b/pkg/controller/v1alpha2/trial/trial_controller_util.go
@@ -109,6 +109,18 @@ func (r *ReconcileTrial) UpdateTrialStatusObservation(instance *trialsv1alpha2.T
 	return nil
 }
 
+func isTrialObservationAvailable(instance *trialsv1alpha2.Trial) bool {
+	objectiveMetricName := instance.Spec.Objective.ObjectiveMetricName
+	if instance.Status.Observation != nil && instance.Status.Observation.Metrics != nil {
+		for _, metric := range instance.Status.Observation.Metrics {
+			if metric.Name == objectiveMetricName {
+				return true
+			}
+		}
+	}
+	return false
+}
+
 func getBestObjectiveMetricValue(metricLogs []*api_pb.MetricLog, objectiveType commonv1alpha2.ObjectiveType) *float64 {
 	metricLogSize := len(metricLogs)
 	if metricLogSize == 0 {