fix(backend) fix run retry for argo (kubeflow#11585)

Signed-off-by: arpechenin <[email protected]>

Author: ntny

backend/src/apiserver/resource/resource_manager_util_test.go

@@ -134,7 +134,8 @@ status:
 	assert.Nil(t, err)
 	newWf, nodes, err := workflow.GenerateRetryExecution()
 
-	newWfString, err := yaml.Marshal(newWf)
+	newWfYaml, err := yaml.Marshal(newWf)
+	actualNewWfString := string(newWfYaml)
 	assert.Nil(t, err)
 	assert.Equal(t, []string{"resubmit-hl9ft-random-fail-3879090716"}, nodes)
 
@@ -189,6 +190,9 @@ spec:
     name: random-fail
     outputs: {}
 status:
+  conditions:
+  - status: "False"
+    type: Completed
   finishedAt: null
   nodes:
     resubmit-hl9ft:
@@ -202,10 +206,8 @@ status:
       startedAt: "2021-05-26T09:14:07Z"
       templateName: rand-fail-dag
       type: DAG
-    resubmit-hl9ft-random-fail-3929423573:
+    resubmit-hl9ft-3929423573:
       boundaryID: resubmit-hl9ft
-      children:
-      - resubmit-hl9ft-3879090716
       displayName: A
       finishedAt: "2021-05-26T09:14:11Z"
       id: resubmit-hl9ft-3929423573
@@ -218,5 +220,5 @@ status:
   startedAt: "2021-05-26T09:14:07Z"
 `
 
-	assert.Equal(t, expectedNewWfString, string(newWfString))
+	assert.Equal(t, expectedNewWfString, actualNewWfString)
 }

backend/src/common/util/workflow.go

@@ -207,6 +207,7 @@ func (w *Workflow) GenerateRetryExecution() (ExecutionSpec, []string, error) {
 	delete(newWF.Labels, LabelKeyWorkflowPersistedFinalState)
 	newWF.ObjectMeta.Labels[common.LabelKeyPhase] = string(workflowapi.NodeRunning)
 	newWF.Status.Phase = workflowapi.WorkflowRunning
+	newWF.Status.Conditions.UpsertCondition(workflowapi.Condition{Status: metav1.ConditionFalse, Type: workflowapi.ConditionTypeCompleted})
 	newWF.Status.Message = ""
 	newWF.Status.FinishedAt = metav1.Time{}
 	if newWF.Spec.ActiveDeadlineSeconds != nil && *newWF.Spec.ActiveDeadlineSeconds == 0 {
@@ -223,8 +224,7 @@ func (w *Workflow) GenerateRetryExecution() (ExecutionSpec, []string, error) {
 		switch node.Phase {
 		case workflowapi.NodeSucceeded, workflowapi.NodeSkipped:
 			if !strings.HasPrefix(node.Name, onExitNodeName) {
-				nodeName := RetrievePodName(*newWF, node)
-				newWF.Status.Nodes[nodeName] = node
+				newWF.Status.Nodes[node.ID] = node
 				continue
 			}
 		case workflowapi.NodeError, workflowapi.NodeFailed, workflowapi.NodeOmitted:
@@ -233,8 +233,7 @@ func (w *Workflow) GenerateRetryExecution() (ExecutionSpec, []string, error) {
 				newNode.Phase = workflowapi.NodeRunning
 				newNode.Message = ""
 				newNode.FinishedAt = metav1.Time{}
-				nodeName := RetrievePodName(*newWF, *newNode)
-				newWF.Status.Nodes[nodeName] = *newNode
+				newWF.Status.Nodes[node.ID] = *newNode
 				continue
 			}
 			// do not add this status to the node. pretend as if this node never existed.
@@ -248,6 +247,23 @@ func (w *Workflow) GenerateRetryExecution() (ExecutionSpec, []string, error) {
 			podsToDelete = append(podsToDelete, oldNodeID)
 		}
 	}
+	for _, node := range newWF.Status.Nodes {
+		var children []string
+		for _, child := range node.Children {
+			if _, ok := newWF.Status.Nodes[child]; ok {
+				children = append(children, child)
+			}
+		}
+		var outboundNodes []string
+		for _, outboundNode := range node.OutboundNodes {
+			if _, ok := newWF.Status.Nodes[outboundNode]; ok {
+				outboundNodes = append(outboundNodes, outboundNode)
+			}
+		}
+		node.Children = children
+		node.OutboundNodes = outboundNodes
+		newWF.Status.Nodes[node.ID] = node
+	}
 	return NewWorkflow(newWF), podsToDelete, nil
 }