Support configuring argo CLI args when using argo workflow trigger. (#…

…1655)

api/jsonschema/schema.json

@@ -2626,6 +2626,13 @@
     "io.argoproj.sensor.v1alpha1.ArgoWorkflowTrigger": {
       "description": "ArgoWorkflowTrigger is the trigger for the Argo Workflow",
       "properties": {
+        "args": {
+          "description": "Args is the list of arguments to pass to the argo CLI",
+          "items": {
+            "type": "string"
+          },
+          "type": "array"
+        },
         "operation": {
           "description": "Operation refers to the type of operation performed on the argo workflow resource. Default value is Submit.",
           "type": "string"

examples/sensors/special-workflow-trigger.yaml

@@ -14,6 +14,9 @@ spec:
         name: argo-workflow-trigger
         argoWorkflow:
           operation: submit
+          args:
+            - --node-field-selector
+            - phase=abc
           source:
             resource:
               apiVersion: argoproj.io/v1alpha1

pkg/apis/sensor/v1alpha1/types.go

@@ -437,6 +437,8 @@ type ArgoWorkflowTrigger struct {
 	Operation ArgoWorkflowOperation `json:"operation,omitempty" protobuf:"bytes,2,opt,name=operation,casttype=ArgoWorkflowOperation"`
 	// Parameters is the list of parameters to pass to resolved Argo Workflow object
 	Parameters []TriggerParameter `json:"parameters,omitempty" protobuf:"bytes,3,rep,name=parameters"`
+	// Args is the list of arguments to pass to the argo CLI
+	Args []string `json:"args,omitempty" protobuf:"bytes,4,rep,name=args"`
 }
 
 // HTTPTrigger is the trigger for the HTTP request

sensors/triggers/argo-workflow/argo-workflow.go

@@ -55,6 +55,7 @@ type ArgoWorkflowTrigger struct {
 	Logger *zap.SugaredLogger
 
 	namespableDynamicClient dynamic.NamespaceableResourceInterface
+	cmdRunner               func(cmd *exec.Cmd) error
 }
 
 // NewArgoWorkflowTrigger returns a new Argo workflow trigger
@@ -65,6 +66,9 @@ func NewArgoWorkflowTrigger(k8sClient kubernetes.Interface, dynamicClient dynami
 		Sensor:        sensor,
 		Trigger:       trigger,
 		Logger:        logger.With(logging.LabelTriggerType, apicommon.ArgoWorkflowTrigger),
+		cmdRunner: func(cmd *exec.Cmd) error {
+			return cmd.Run()
+		},
 	}
 }
 
@@ -174,7 +178,8 @@ func (t *ArgoWorkflowTrigger) Execute(ctx context.Context, events map[string]*v1
 
 	cmd.Stdout = os.Stdout
 	cmd.Stderr = os.Stderr
-	if err := cmd.Run(); err != nil {
+	cmd.Args = append(cmd.Args, trigger.Template.ArgoWorkflow.Args...)
+	if err := t.cmdRunner(cmd); err != nil {
 		return nil, errors.Wrapf(err, "failed to execute %s command for workflow %s", string(op), name)
 	}
 

sensors/triggers/argo-workflow/argo-workflow_test.go

@@ -17,6 +17,10 @@ package argo_workflow
 
 import (
 	"context"
+	"fmt"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/dynamic"
+	"os/exec"
 	"testing"
 
 	"github.com/argoproj/argo-events/common/logging"
@@ -47,6 +51,10 @@ var sensorObj = &v1alpha1.Sensor{
 	},
 }
 
+var (
+	un = newUnstructured("argoproj.io/v1alpha1", "Workflow", "fake", "test")
+)
+
 func newUnstructured(apiVersion, kind, namespace, name string) *unstructured.Unstructured {
 	return &unstructured.Unstructured{
 		Object: map[string]interface{}{
@@ -63,10 +71,9 @@ func newUnstructured(apiVersion, kind, namespace, name string) *unstructured.Uns
 	}
 }
 
-func getFakeWfTrigger() *ArgoWorkflowTrigger {
+func getFakeWfTrigger(operation v1alpha1.ArgoWorkflowOperation) *ArgoWorkflowTrigger {
 	runtimeScheme := runtime.NewScheme()
 	client := dynamicFake.NewSimpleDynamicClient(runtimeScheme)
-	un := newUnstructured("argoproj.io/v1alpha1", "Workflow", "fake", "test")
 	artifact := apicommon.NewResource(un)
 	trigger := &v1alpha1.Trigger{
 		Template: &v1alpha1.TriggerTemplate{
@@ -75,15 +82,15 @@ func getFakeWfTrigger() *ArgoWorkflowTrigger {
 				Source: &v1alpha1.ArtifactLocation{
 					Resource: &artifact,
 				},
-				Operation: "Submit",
+				Operation: operation,
 			},
 		},
 	}
 	return NewArgoWorkflowTrigger(fake.NewSimpleClientset(), client, sensorObj.DeepCopy(), trigger, logging.NewArgoEventsLogger())
 }
 
 func TestFetchResource(t *testing.T) {
-	trigger := getFakeWfTrigger()
+	trigger := getFakeWfTrigger("submit")
 	resource, err := trigger.FetchResource(context.TODO())
 	assert.Nil(t, err)
 	assert.NotNil(t, resource)
@@ -96,3 +103,42 @@ func TestFetchResource(t *testing.T) {
 func TestApplyResourceParameters(t *testing.T) {
 
 }
+
+func TestExecute(t *testing.T) {
+	t.Run("passes trigger args as flags to argo command", func(t *testing.T) {
+		ctx := context.Background()
+		var actual string
+		firstArg := "--foo"
+		secondArg := "--bar"
+		trigger := storingCmdTrigger(&actual, firstArg, secondArg)
+
+		_, err := namespacedClientFrom(trigger).Namespace(un.GetNamespace()).Create(ctx, un, metav1.CreateOptions{})
+		assert.Nil(t, err)
+
+		_, err = trigger.Execute(ctx, nil, un)
+		assert.Nil(t, err)
+
+		expected := fmt.Sprintf("argo -n %s resume test %s %s", un.GetNamespace(), firstArg, secondArg)
+		assert.Contains(t, actual, expected)
+	})
+}
+
+func storingCmdTrigger(cmdStr *string, wfArgs ...string) *ArgoWorkflowTrigger {
+	trigger := getFakeWfTrigger("resume")
+	f := func(cmd *exec.Cmd) error {
+		*cmdStr = cmd.String()
+		return nil
+	}
+	trigger.cmdRunner = f
+	trigger.Trigger.Template.ArgoWorkflow.Args = wfArgs
+
+	return trigger
+}
+
+func namespacedClientFrom(trigger *ArgoWorkflowTrigger) dynamic.NamespaceableResourceInterface {
+	return trigger.DynamicClient.Resource(schema.GroupVersionResource{
+		Group:    "argoproj.io",
+		Version:  "v1alpha1",
+		Resource: "workflows",
+	})
+}