fix: Properly reset suspended and skipped nodes when retrying (#9422)

* chore: wip Signed-off-by: Yuan Tang <[email protected]> * chore: wip Signed-off-by: Yuan Tang <[email protected]> * chore: clean up Signed-off-by: Yuan Tang <[email protected]> * chore: clean up Signed-off-by: Yuan Tang <[email protected]> * chore: Add test Signed-off-by: Yuan Tang <[email protected]> * chore: fix Signed-off-by: Yuan Tang <[email protected]> Signed-off-by: Yuan Tang <[email protected]>
argoproj · Aug 25, 2022 · 04d1943 · 04d1943
1 parent de6b5ae
commit 04d1943
Show file tree

Hide file tree

Showing 2 changed files with 52 additions and 1 deletion.
diff --git a/workflow/util/util.go b/workflow/util/util.go
@@ -919,7 +919,24 @@ func FormulateRetryWorkflow(ctx context.Context, wf *wfv1.Workflow, restartSucce
 }
 
 func resetNode(node wfv1.NodeStatus) wfv1.NodeStatus {
- node.Phase = wfv1.NodeRunning
+ // The previously supplied parameters needed to be reset. Otherwise, `argo node reset` would not work as expected.
+ if node.Type == wfv1.NodeTypeSuspend {
+ if node.Outputs != nil {
+ for i, param := range node.Outputs.Parameters {
+ node.Outputs.Parameters[i] = wfv1.Parameter{
+ Name: param.Name,
+ Value: nil,
+ ValueFrom: &wfv1.ValueFrom{Supplied: &wfv1.SuppliedValueFrom{}},
+ }
+ }
+ }
+ }
+ if node.Phase == wfv1.NodeSkipped {
+ // The skipped nodes need to be kept as skipped. Otherwise, the workflow will be stuck on running.
+ node.Phase = wfv1.NodeSkipped
+ } else {
+ node.Phase = wfv1.NodeRunning
+ }
  node.Message = ""
  node.StartedAt = metav1.Time{Time: time.Now().UTC()}
  node.FinishedAt = metav1.Time{}

diff --git a/workflow/util/util_test.go b/workflow/util/util_test.go
@@ -901,6 +901,40 @@ func TestFormulateRetryWorkflow(t *testing.T) {
 
  }
  })
+ t.Run("Skipped and Suspended Nodes", func(t *testing.T) {
+ wf := &wfv1.Workflow{
+ ObjectMeta: metav1.ObjectMeta{
+ Name: "wf-with-skipped-and-suspended-nodes",
+ Labels: map[string]string{},
+ },
+ Status: wfv1.WorkflowStatus{
+ Phase: wfv1.WorkflowFailed,
+ Nodes: map[string]wfv1.NodeStatus{
+ "my-nested-dag-1": {ID: "my-nested-dag-1", Phase: wfv1.NodeSucceeded, Type: wfv1.NodeTypeTaskGroup},
+ "suspended": {ID: "suspended", Phase: wfv1.NodeSucceeded, Type: wfv1.NodeTypeSuspend, BoundaryID: "my-nested-dag-1", Outputs: &wfv1.Outputs{Parameters: []wfv1.Parameter{{
+ Name: "param-1",
+ Value: wfv1.AnyStringPtr("3"),
+ ValueFrom: &wfv1.ValueFrom{Supplied: &wfv1.SuppliedValueFrom{}},
+ }}}},
+ "skipped": {ID: "skipped", Phase: wfv1.NodeSkipped, Type: wfv1.NodeTypeSkipped, BoundaryID: "suspended"},
+ }},
+ }
+ _, err := wfClient.Create(ctx, wf, metav1.CreateOptions{})
+ assert.NoError(t, err)
+ wf, _, err = FormulateRetryWorkflow(ctx, wf, true, "id=suspended", nil)
+ if assert.NoError(t, err) {
+ if assert.Len(t, wf.Status.Nodes, 3) {
+ assert.Equal(t, wfv1.NodeSucceeded, wf.Status.Nodes["my-nested-dag-1"].Phase)
+ assert.Equal(t, wfv1.NodeRunning, wf.Status.Nodes["suspended"].Phase)
+ assert.Equal(t, wfv1.Parameter{
+ Name: "param-1",
+ Value: nil,
+ ValueFrom: &wfv1.ValueFrom{Supplied: &wfv1.SuppliedValueFrom{}},
+ }, wf.Status.Nodes["suspended"].Outputs.Parameters[0])
+ assert.Equal(t, wfv1.NodeSkipped, wf.Status.Nodes["skipped"].Phase)
+ }
+ }
+ })
  t.Run("Nested DAG with Non-group Node Selected", func(t *testing.T) {
  wf := &wfv1.Workflow{
  ObjectMeta: metav1.ObjectMeta{