feat: More control over resuming suspended nodes Fixes #1893 (#1904)

api/openapi-spec/swagger.json

@@ -5798,6 +5798,9 @@
         },
         "namespace": {
           "type": "string"
+        },
+        "nodeFieldSelector": {
+          "type": "string"
         }
       }
     },
@@ -5815,11 +5818,17 @@
     "io.argoproj.workflow.v1alpha1.WorkflowStopRequest": {
       "type": "object",
       "properties": {
+        "message": {
+          "type": "string"
+        },
         "name": {
           "type": "string"
         },
         "namespace": {
           "type": "string"
+        },
+        "nodeFieldSelector": {
+          "type": "string"
         }
       }
     },

cmd/argo/commands/resume.go

@@ -5,23 +5,39 @@ import (
 	"log"
 
 	"github.com/spf13/cobra"
+	"k8s.io/apimachinery/pkg/fields"
 
 	"github.com/argoproj/argo/cmd/argo/commands/client"
 	workflowpkg "github.com/argoproj/argo/pkg/apiclient/workflow"
 )
 
+type resumeOps struct {
+	nodeFieldSelector string // --node-field-selector
+}
+
 func NewResumeCommand() *cobra.Command {
+	var (
+		resumeArgs resumeOps
+	)
+
 	var command = &cobra.Command{
 		Use:   "resume WORKFLOW1 WORKFLOW2...",
 		Short: "resume zero or more workflows",
 		Run: func(cmd *cobra.Command, args []string) {
 			ctx, apiClient := client.NewAPIClient()
 			serviceClient := apiClient.NewWorkflowServiceClient()
 			namespace := client.Namespace()
+
+			selector, err := fields.ParseSelector(resumeArgs.nodeFieldSelector)
+			if err != nil {
+				log.Fatalf("Unable to parse node field selector '%s': %s", resumeArgs.nodeFieldSelector, err)
+			}
+
 			for _, wfName := range args {
 				_, err := serviceClient.ResumeWorkflow(ctx, &workflowpkg.WorkflowResumeRequest{
-					Name:      wfName,
-					Namespace: namespace,
+					Name:              wfName,
+					Namespace:         namespace,
+					NodeFieldSelector: selector.String(),
 				})
 				if err != nil {
 					log.Fatalf("Failed to resume %s: %+v", wfName, err)
@@ -31,5 +47,6 @@ func NewResumeCommand() *cobra.Command {
 
 		},
 	}
+	command.Flags().StringVar(&resumeArgs.nodeFieldSelector, "node-field-selector", "", "selector of node to resume, eg: --node-field-selector inputs.paramaters.myparam.value=abc")
 	return command
 }

cmd/argo/commands/stop.go

@@ -2,15 +2,26 @@ package commands
 
 import (
 	"fmt"
+	"log"
 
 	"github.com/argoproj/pkg/errors"
 	"github.com/spf13/cobra"
+	"k8s.io/apimachinery/pkg/fields"
 
 	"github.com/argoproj/argo/cmd/argo/commands/client"
 	workflowpkg "github.com/argoproj/argo/pkg/apiclient/workflow"
 )
 
+type stopOps struct {
+	message           string // --message
+	nodeFieldSelector string // --node-field-selector
+}
+
 func NewStopCommand() *cobra.Command {
+	var (
+		stopArgs stopOps
+	)
+
 	var command = &cobra.Command{
 		Use:   "stop WORKFLOW WORKFLOW2...",
 		Short: "stop zero or more workflows",
@@ -19,15 +30,25 @@ func NewStopCommand() *cobra.Command {
 			ctx, apiClient := client.NewAPIClient()
 			serviceClient := apiClient.NewWorkflowServiceClient()
 			namespace := client.Namespace()
+
+			selector, err := fields.ParseSelector(stopArgs.nodeFieldSelector)
+			if err != nil {
+				log.Fatalf("Unable to parse node field selector '%s': %s", stopArgs.nodeFieldSelector, err)
+			}
+
 			for _, name := range args {
 				wf, err := serviceClient.StopWorkflow(ctx, &workflowpkg.WorkflowStopRequest{
-					Name:      name,
-					Namespace: namespace,
+					Name:              name,
+					Namespace:         namespace,
+					NodeFieldSelector: selector.String(),
+					Message:           stopArgs.message,
 				})
 				errors.CheckError(err)
 				fmt.Printf("workflow %s stopped\n", wf.Name)
 			}
 		},
 	}
+	command.Flags().StringVar(&stopArgs.message, "message", "", "Message to add to previously running nodes")
+	command.Flags().StringVar(&stopArgs.nodeFieldSelector, "node-field-selector", "", "selector of node to stop, eg: --node-field-selector inputs.paramaters.myparam.value=abc")
 	return command
 }

pkg/apiclient/workflow/workflow.proto

@@ -48,6 +48,7 @@ message WorkflowRetryRequest {
 message WorkflowResumeRequest {
     string name = 1;
     string namespace = 2;
+    string nodeFieldSelector = 3;
 }
 
 message WorkflowTerminateRequest {
@@ -58,6 +59,8 @@ message WorkflowTerminateRequest {
 message WorkflowStopRequest {
      string name = 1;
      string namespace = 2;
+     string nodeFieldSelector = 3;
+     string message = 4;
 }
 
 message WorkflowSuspendRequest {

pkg/apiclient/workflow/workflow.swagger.json

@@ -4903,6 +4903,9 @@
         },
         "namespace": {
           "type": "string"
+        },
+        "nodeFieldSelector": {
+          "type": "string"
         }
       }
     },
@@ -4925,6 +4928,12 @@
         },
         "namespace": {
           "type": "string"
+        },
+        "nodeFieldSelector": {
+          "type": "string"
+        },
+        "message": {
+          "type": "string"
         }
       }
     },

server/workflow/workflow_server.go

@@ -256,7 +256,7 @@ func (s *workflowServer) ResumeWorkflow(ctx context.Context, req *workflowpkg.Wo
 		return nil, err
 	}
 
-	err = util.ResumeWorkflow(wfClient.ArgoprojV1alpha1().Workflows(req.Namespace), req.Name)
+	err = util.ResumeWorkflow(wfClient.ArgoprojV1alpha1().Workflows(req.Namespace), req.Name, req.NodeFieldSelector)
 	if err != nil {
 		log.Warnf("Failed to resume %s: %+v", req.Name, err)
 		return nil, err
@@ -313,7 +313,7 @@ func (s *workflowServer) TerminateWorkflow(ctx context.Context, req *workflowpkg
 
 func (s *workflowServer) StopWorkflow(ctx context.Context, req *workflowpkg.WorkflowStopRequest) (*v1alpha1.Workflow, error) {
 	wfClient := auth.GetWfClient(ctx)
-	err := util.StopWorkflow(wfClient.ArgoprojV1alpha1().Workflows(req.Namespace), req.Name)
+	err := util.StopWorkflow(wfClient.ArgoprojV1alpha1().Workflows(req.Namespace), req.Name, req.NodeFieldSelector, req.Message)
 	if err != nil {
 		return nil, err
 	}

test/e2e/cli_test.go

@@ -277,6 +277,41 @@ func (s *CLISuite) TestWorkflowSuspendResume() {
 		})
 }
 
+func (s *CLISuite) TestNodeSuspendResume() {
+	s.Given().
+		Workflow("@testdata/node-suspend.yaml").
+		When().
+		SubmitWorkflow().
+		WaitForWorkflowCondition(func(wf *wfv1.Workflow) bool {
+			return wf.Status.AnyActiveSuspendNode()
+		}, "suspended node", 20*time.Second).
+		RunCli([]string{"resume", "node-suspend", "--node-selector", "inputs.parameters.tag.value=suspend1-tag1"}, func(t *testing.T, output string, err error) {
+			if assert.NoError(t, err) {
+				assert.Contains(t, output, "workflow node-suspend resumed")
+			}
+		}).
+		WaitForWorkflowCondition(func(wf *wfv1.Workflow) bool {
+			return wf.Status.AnyActiveSuspendNode()
+		}, "suspended node", 10*time.Second).
+		RunCli([]string{"stop", "node-suspend", "--node-selector", "inputs.parameters.tag.value=suspend2-tag1", "--message", "because"}, func(t *testing.T, output string, err error) {
+			if assert.NoError(t, err) {
+				assert.Contains(t, output, "workflow node-suspend stopped")
+			}
+		}).
+		WaitForWorkflowCondition(func(wf *wfv1.Workflow) bool {
+			return wf.Status.Phase == wfv1.NodeFailed
+		}, "suspended node", 10*time.Second).
+		Then().
+		ExpectWorkflow(func(t *testing.T, _ *corev1.ObjectMeta, status *wfv1.WorkflowStatus) {
+			if assert.Equal(t, wfv1.NodeFailed, status.Phase) {
+				r := regexp.MustCompile(`child '(node-suspend-[0-9]+)' failed`)
+				res := r.FindStringSubmatch(status.Message)
+				assert.Equal(t, len(res), 2)
+				assert.Equal(t, status.Nodes[res[1]].Message, "because")
+			}
+		})
+}
+
 func (s *CLISuite) TestWorkflowDelete() {
 	s.Run("DeleteByName", func() {
 		s.Given().

test/e2e/testdata/node-suspend.yaml

@@ -0,0 +1,37 @@
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  name: node-suspend
+  labels:
+    argo-e2e: true
+spec:
+  entrypoint: node-suspend
+  templates:
+    - name: node-suspend
+      steps:
+      - - name: step1
+          template: whalesay
+      - - name: suspend1
+          template: suspend
+          arguments:
+            parameters:
+              - name: "tag"
+                value: "suspend1-{{steps.step1.outputs.result}}"
+      - - name: suspend2
+          template: suspend
+          arguments:
+            parameters:
+              - name: "tag"
+                value: "suspend2-{{steps.step1.outputs.result}}"
+
+    - name: whalesay
+      script:
+        image: python:alpine3.6
+        command: [sh]
+        source: echo tag1
+
+    - name: suspend
+      inputs:
+        parameters:
+          - name: tag
+      suspend: {}

test/e2e/testdata/workflow-template-nested-template.yaml

@@ -17,4 +17,4 @@ spec:
       template: whalesay-inner-template
       inputs:
         parameters:
-          - name: message
+          - name: message

workflow/controller/operator_test.go

@@ -847,7 +847,7 @@ func TestSuspendResume(t *testing.T) {
 	assert.Equal(t, 0, len(pods.Items))
 
 	// resume the workflow and operate again. two pods should be able to be scheduled
-	err = util.ResumeWorkflow(wfcset, wf.ObjectMeta.Name)
+	err = util.ResumeWorkflow(wfcset, wf.ObjectMeta.Name, "")
 	assert.NoError(t, err)
 	wf, err = wfcset.Get(wf.ObjectMeta.Name, metav1.GetOptions{})
 	assert.NoError(t, err)
@@ -1124,10 +1124,17 @@ spec:
     steps:
     - - name: approve
         template: approve
+        arguments:
+          parameters:
+          - name: param1
+            value: value1
     - - name: release
         template: whalesay
 
   - name: approve
+    inputs:
+      parameters:
+      - name: param1
     suspend: {}
 
   - name: whalesay
@@ -1159,7 +1166,92 @@ func TestSuspendTemplate(t *testing.T) {
 	assert.Equal(t, 0, len(pods.Items))
 
 	// resume the workflow. verify resume workflow edits nodestatus correctly
-	err = util.ResumeWorkflow(wfcset, wf.ObjectMeta.Name)
+	err = util.ResumeWorkflow(wfcset, wf.ObjectMeta.Name, "")
+	assert.NoError(t, err)
+	wf, err = wfcset.Get(wf.ObjectMeta.Name, metav1.GetOptions{})
+	assert.NoError(t, err)
+	assert.False(t, util.IsWorkflowSuspended(wf))
+
+	// operate the workflow. it should reach the second step
+	woc = newWorkflowOperationCtx(wf, controller)
+	woc.operate()
+	pods, err = controller.kubeclientset.CoreV1().Pods("").List(metav1.ListOptions{})
+	assert.NoError(t, err)
+	assert.Equal(t, 1, len(pods.Items))
+}
+
+func TestSuspendTemplateWithFailedResume(t *testing.T) {
+	controller := newController()
+	wfcset := controller.wfclientset.ArgoprojV1alpha1().Workflows("")
+
+	// operate the workflow. it should become in a suspended state after
+	wf := unmarshalWF(suspendTemplate)
+	wf, err := wfcset.Create(wf)
+	assert.NoError(t, err)
+	woc := newWorkflowOperationCtx(wf, controller)
+	woc.operate()
+	wf, err = wfcset.Get(wf.ObjectMeta.Name, metav1.GetOptions{})
+	assert.NoError(t, err)
+	assert.True(t, util.IsWorkflowSuspended(wf))
+
+	// operate again and verify no pods were scheduled
+	woc = newWorkflowOperationCtx(wf, controller)
+	woc.operate()
+	pods, err := controller.kubeclientset.CoreV1().Pods("").List(metav1.ListOptions{})
+	assert.NoError(t, err)
+	assert.Equal(t, 0, len(pods.Items))
+
+	// resume the workflow. verify resume workflow edits nodestatus correctly
+	err = util.StopWorkflow(wfcset, wf.ObjectMeta.Name, "inputs.parameters.param1.value=value1", "Step failed!")
+	assert.NoError(t, err)
+	wf, err = wfcset.Get(wf.ObjectMeta.Name, metav1.GetOptions{})
+	assert.NoError(t, err)
+	assert.False(t, util.IsWorkflowSuspended(wf))
+
+	// operate the workflow. it should be failed and not reach the second step
+	woc = newWorkflowOperationCtx(wf, controller)
+	woc.operate()
+	assert.Equal(t, wfv1.NodeFailed, woc.wf.Status.Phase)
+	pods, err = controller.kubeclientset.CoreV1().Pods("").List(metav1.ListOptions{})
+	assert.NoError(t, err)
+	assert.Equal(t, 0, len(pods.Items))
+}
+
+func TestSuspendTemplateWithFilteredResume(t *testing.T) {
+	controller := newController()
+	wfcset := controller.wfclientset.ArgoprojV1alpha1().Workflows("")
+
+	// operate the workflow. it should become in a suspended state after
+	wf := unmarshalWF(suspendTemplate)
+	wf, err := wfcset.Create(wf)
+	assert.NoError(t, err)
+	woc := newWorkflowOperationCtx(wf, controller)
+	woc.operate()
+	wf, err = wfcset.Get(wf.ObjectMeta.Name, metav1.GetOptions{})
+	assert.NoError(t, err)
+	assert.True(t, util.IsWorkflowSuspended(wf))
+
+	// operate again and verify no pods were scheduled
+	woc = newWorkflowOperationCtx(wf, controller)
+	woc.operate()
+	pods, err := controller.kubeclientset.CoreV1().Pods("").List(metav1.ListOptions{})
+	assert.NoError(t, err)
+	assert.Equal(t, 0, len(pods.Items))
+
+	// resume the workflow, but with non-matching selector
+	err = util.ResumeWorkflow(wfcset, wf.ObjectMeta.Name, "inputs.paramaters.param1.value=value2")
+	assert.Error(t, err)
+
+	// operate the workflow. nothing should have happened
+	woc = newWorkflowOperationCtx(wf, controller)
+	woc.operate()
+	pods, err = controller.kubeclientset.CoreV1().Pods("").List(metav1.ListOptions{})
+	assert.NoError(t, err)
+	assert.Equal(t, 0, len(pods.Items))
+	assert.True(t, util.IsWorkflowSuspended(wf))
+
+	// resume the workflow, but with matching selector
+	err = util.ResumeWorkflow(wfcset, wf.ObjectMeta.Name, "inputs.parameters.param1.value=value1")
 	assert.NoError(t, err)
 	wf, err = wfcset.Get(wf.ObjectMeta.Name, metav1.GetOptions{})
 	assert.NoError(t, err)