feat: cli allow retry successful workflow if nodeFieldSelector is set. …

…Fixes #11020 (#11409)

Signed-off-by: or-shachar <or.shachar@wiz.io>

cmd/argo/commands/retry.go

@@ -71,6 +71,9 @@ func NewRetryCommand() *cobra.Command {
 # Retry the latest workflow:
 
   argo retry @latest
+
+# Restart node with id 5 on successful workflow, using node-field-selector
+  argo retry my-wf --restart-successful --node-field-selector id=5
 `,
 		Run: func(cmd *cobra.Command, args []string) {
 			if len(args) == 0 && !retryOpts.hasSelector() {

docs/cli/argo_retry.md

@@ -41,6 +41,9 @@ argo retry [WORKFLOW...] [flags]
 
   argo retry @latest
 
+# Restart node with id 5 on successful workflow, using node-field-selector
+  argo retry my-wf --restart-successful --node-field-selector id=5
+
 ```
 
 ### Options

workflow/util/util.go

@@ -810,8 +810,12 @@ func resetConnectedParentGroupNodes(oldWF *wfv1.Workflow, newWF *wfv1.Workflow,
 func FormulateRetryWorkflow(ctx context.Context, wf *wfv1.Workflow, restartSuccessful bool, nodeFieldSelector string, parameters []string) (*wfv1.Workflow, []string, error) {
 	switch wf.Status.Phase {
 	case wfv1.WorkflowFailed, wfv1.WorkflowError:
+	case wfv1.WorkflowSucceeded:
+		if !(restartSuccessful && len(nodeFieldSelector) > 0) {
+			return nil, nil, errors.Errorf(errors.CodeBadRequest, "To retry a succeeded workflow, set the options restartSuccessful and nodeFieldSelector")
+		}
 	default:
-		return nil, nil, errors.Errorf(errors.CodeBadRequest, "workflow must be Failed/Error to retry")
+		return nil, nil, errors.Errorf(errors.CodeBadRequest, "Cannot retry a workflow in phase %s", wf.Status.Phase)
 	}
 
 	newWF := wf.DeepCopy()

workflow/util/util_test.go

@@ -468,7 +468,7 @@ func TestUpdateSuspendedNode(t *testing.T) {
 		err = updateSuspendedNode(ctx, wfIf, hydratorfake.Noop, "suspend-template", "name=suspend-template-kgfn7[0].approve", SetOperationValues{OutputParameters: map[string]string{"message2": "Hello World 2"}})
 		assert.NoError(t, err)
 
-		//make sure global variable was updated
+		// make sure global variable was updated
 		wf, err := wfIf.Get(ctx, "suspend-template", metav1.GetOptions{})
 		assert.NoError(t, err)
 		assert.Equal(t, "Hello World 2", wf.Status.Outputs.Parameters[0].Value.String())
@@ -1215,6 +1215,91 @@ func TestFormulateRetryWorkflow(t *testing.T) {
 			assert.Equal(t, "modified", wf.Spec.Arguments.Parameters[0].Value.String())
 		}
 	})
+
+	t.Run("Fail on running workflow", func(t *testing.T) {
+		wf := &wfv1.Workflow{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:   "running-workflow-1",
+				Labels: map[string]string{},
+			},
+			Status: wfv1.WorkflowStatus{
+				Phase: wfv1.WorkflowRunning,
+				Nodes: map[string]wfv1.NodeStatus{},
+			},
+		}
+		_, err := wfClient.Create(ctx, wf, metav1.CreateOptions{})
+		assert.NoError(t, err)
+		_, _, err = FormulateRetryWorkflow(ctx, wf, false, "", nil)
+		assert.Error(t, err)
+	})
+
+	t.Run("Fail on pending workflow", func(t *testing.T) {
+		wf := &wfv1.Workflow{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:   "pending-workflow-1",
+				Labels: map[string]string{},
+			},
+			Status: wfv1.WorkflowStatus{
+				Phase: wfv1.WorkflowPending,
+				Nodes: map[string]wfv1.NodeStatus{},
+			},
+		}
+		_, err := wfClient.Create(ctx, wf, metav1.CreateOptions{})
+		assert.NoError(t, err)
+		_, _, err = FormulateRetryWorkflow(ctx, wf, false, "", nil)
+		assert.Error(t, err)
+	})
+
+	t.Run("Fail on successful workflow without restartSuccessful and nodeFieldSelector", func(t *testing.T) {
+		wf := &wfv1.Workflow{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:   "successful-workflow-1",
+				Labels: map[string]string{},
+			},
+			Status: wfv1.WorkflowStatus{
+				Phase: wfv1.WorkflowSucceeded,
+				Nodes: map[string]wfv1.NodeStatus{
+					"successful-workflow-1": {ID: "successful-workflow-1", Phase: wfv1.NodeSucceeded, Type: wfv1.NodeTypeTaskGroup, Children: []string{"1"}},
+					"1":                     {ID: "1", Phase: wfv1.NodeSucceeded, Type: wfv1.NodeTypeTaskGroup, BoundaryID: "successful-workflow-1", Children: []string{"2"}},
+					"2":                     {ID: "2", Phase: wfv1.NodeSucceeded, Type: wfv1.NodeTypePod, BoundaryID: "1"}},
+			},
+		}
+		_, err := wfClient.Create(ctx, wf, metav1.CreateOptions{})
+		assert.NoError(t, err)
+		_, _, err = FormulateRetryWorkflow(ctx, wf, false, "", nil)
+		assert.Error(t, err)
+	})
+
+	t.Run("Retry successful workflow with restartSuccessful and nodeFieldSelector", func(t *testing.T) {
+		wf := &wfv1.Workflow{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:   "successful-workflow-2",
+				Labels: map[string]string{},
+			},
+			Status: wfv1.WorkflowStatus{
+				Phase: wfv1.WorkflowSucceeded,
+				Nodes: map[string]wfv1.NodeStatus{
+					"successful-workflow-2": {ID: "successful-workflow-2", Phase: wfv1.NodeSucceeded, Type: wfv1.NodeTypeTaskGroup, Children: []string{"1"}},
+					"1":                     {ID: "1", Phase: wfv1.NodeSucceeded, Type: wfv1.NodeTypeTaskGroup, BoundaryID: "successful-workflow-2", Children: []string{"2", "4"}},
+					"2":                     {ID: "2", Phase: wfv1.NodeSucceeded, Type: wfv1.NodeTypeTaskGroup, BoundaryID: "1", Children: []string{"3"}},
+					"3":                     {ID: "3", Phase: wfv1.NodeSucceeded, Type: wfv1.NodeTypePod, BoundaryID: "2"},
+					"4":                     {ID: "4", Phase: wfv1.NodeSucceeded, Type: wfv1.NodeTypePod, BoundaryID: "1"}},
+			},
+		}
+		_, err := wfClient.Create(ctx, wf, metav1.CreateOptions{})
+		assert.NoError(t, err)
+		wf, _, err = FormulateRetryWorkflow(ctx, wf, true, "id=4", nil)
+		if assert.NoError(t, err) {
+			// Node #4 is deleted and will be recreated so only 4 nodes left in wf.Status.Nodes
+			if assert.Len(t, wf.Status.Nodes, 4) {
+				assert.Equal(t, wfv1.NodeSucceeded, wf.Status.Nodes["successful-workflow-2"].Phase)
+				// The parent group nodes should be running.
+				assert.Equal(t, wfv1.NodeRunning, wf.Status.Nodes["1"].Phase)
+				assert.Equal(t, wfv1.NodeSucceeded, wf.Status.Nodes["2"].Phase)
+				assert.Equal(t, wfv1.NodeSucceeded, wf.Status.Nodes["3"].Phase)
+			}
+		}
+	})
 }
 
 func TestFromUnstructuredObj(t *testing.T) {