and some safety check (kubeflow#683)

pkg/controller.v2/controller.go

@@ -328,6 +328,7 @@ func (tc *TFJobController) enqueueTFJob(tfjob interface{}) {
 		return
 	}
 
+	// TODO: we may need add backoff here
 	tc.workQueue.Add(key)
 }
 
@@ -344,6 +345,9 @@ func (tc *TFJobController) syncTFJob(key string) (bool, error) {
 	if err != nil {
 		return false, err
 	}
+	if len(namespace) == 0 || len(name) == 0 {
+		return false, fmt.Errorf("invalid tfjob key %q: either namespace or name is missing", key)
+	}
 
 	sharedTFJob, err := tc.getTFJobFromName(namespace, name)
 	if err != nil {

pkg/controller.v2/controller_status.go

@@ -77,7 +77,7 @@ func updateStatusSingle(tfjob *tfv1alpha2.TFJob, rtype tfv1alpha2.TFReplicaType,
 	failed := int(tfjob.Status.TFReplicaStatuses[rtype].Failed)
 
 	// All workers are running, set StartTime.
-	if running == replicas {
+	if running == replicas && tfjob.Status.StartTime == nil {
 		now := metav1.Now()
 		tfjob.Status.StartTime = &now
 	}
@@ -94,8 +94,6 @@ func updateStatusSingle(tfjob *tfv1alpha2.TFJob, rtype tfv1alpha2.TFReplicaType,
 			}
 			if expected == 0 {
 				msg := fmt.Sprintf("TFJob %s is successfully completed.", tfjob.Name)
-				now := metav1.Now()
-				tfjob.Status.CompletionTime = &now
 				err := updateTFJobConditions(tfjob, tfv1alpha2.TFJobSucceeded, tfJobSucceededReason, msg)
 				if err != nil {
 					loggerForTFJob(tfjob).Infof("Append tfjob condition error: %v", err)
@@ -129,8 +127,6 @@ func updateStatusSingle(tfjob *tfv1alpha2.TFJob, rtype tfv1alpha2.TFReplicaType,
 			// All workers are succeeded, leave a succeeded condition.
 			if expected == 0 {
 				msg := fmt.Sprintf("TFJob %s is successfully completed.", tfjob.Name)
-				now := metav1.Now()
-				tfjob.Status.CompletionTime = &now
 				err := updateTFJobConditions(tfjob, tfv1alpha2.TFJobSucceeded, tfJobSucceededReason, msg)
 				if err != nil {
 					loggerForTFJob(tfjob).Infof("Append tfjob condition error: %v", err)
@@ -415,6 +411,7 @@ func getCondition(status tfv1alpha2.TFJobStatus, condType tfv1alpha2.TFJobCondit
 // setCondition updates the tfjob to include the provided condition.
 // If the condition that we are about to add already exists
 // and has the same status and reason then we are not going to update.
+// If condition is TFJobSucceeded, set CompletionTime.
 func setCondition(status *tfv1alpha2.TFJobStatus, condition tfv1alpha2.TFJobCondition) {
 	currentCond := getCondition(*status, condition.Type)
 
@@ -428,6 +425,12 @@ func setCondition(status *tfv1alpha2.TFJobStatus, condition tfv1alpha2.TFJobCond
 		condition.LastTransitionTime = currentCond.LastTransitionTime
 	}
 
+	// if success, update with complete time
+	if condition.Type == tfv1alpha2.TFJobSucceeded && status.CompletionTime == nil {
+		now := metav1.Now()
+		status.CompletionTime = &now
+	}
+
 	// Append the updated condition to the
 	newConditions := filterOutCondition(status.Conditions, condition.Type)
 	status.Conditions = append(newConditions, condition)