Repository: argoproj/argo
Pull request #3261 (2020-06-19 20:23:04)

feat: Append previous workflow name as label to resubmitted workflow
Append an additional label so it's easy for user to see the name of the original workflow that has been resubmitted.

Checklist:

* [x] Either (a) I've created an [enhancement proposal](https://github.com/argoproj/argo/issues/new/choose) and discussed it with the community, (b) this is a bug fix, or (c) this is a chore.
* [x] The title of the PR is (a) [conventional](https://www.conventionalcommits.org/en/v1.0.0/), (b) states what changed, and (c) suffixes the related issues number. E.g. `"fix(controller): Updates such and such. Fixes #1234"`.  
* [x] I've signed the CLA.
* [x] I have written unit and/or e2e tests for my change. PRs without these are unlikely to be merged.
* [x] My builds are green. Try syncing with master if they are not. 
* [x] My organization is added to [USERS.md](https://github.com/argoproj/argo/blob/master/USERS.md).



Patch:
diff --git a/config/config.go b/config/config.go
index abde6d582..1ec28bba1 100644
--- a/config/config.go
+++ b/config/config.go
@@ -85,8 +85,11 @@ type Config struct {
 	// WorkflowDefaults are values that will apply to all Workflows from this controller, unless overridden on the Workflow-level
 	WorkflowDefaults *wfv1.Workflow `json:"workflowDefaults,omitempty"`
 
-	// PodSpecLogStrategy enable the logging of podspec on controller log.
+	// PodSpecLogStrategy enables the logging of podspec on controller log.
 	PodSpecLogStrategy PodSpecLogStrategy `json:"podSpecLogStrategy,omitempty"`
+
+	// WorkflowRestrictions restricts the controller to executing Workflows that meet certain restrictions
+	WorkflowRestrictions *WorkflowRestrictions `json:"workflowRestrictions,omitempty"`
 }
 
 // PodSpecLogStrategy contains the configuration for logging the pod spec in controller log for debugging purpose
@@ -163,6 +166,7 @@ type ConnectionPool struct {
 	MaxOpenConns    int `json:"maxOpenConns,omitempty"`
 	ConnMaxLifetime TTL `json:"connMaxLifetime,omitempty"`
 }
+
 type PostgreSQLConfig struct {
 	Host           string                  `json:"host"`
 	Port           string                  `json:"port"`
@@ -246,3 +250,28 @@ type MetricsConfig struct {
 	// IgnoreErrors is a flag that instructs prometheus to ignore metric emission errors
 	IgnoreErrors bool `json:"ignoreErrors,omitempty"`
 }
+
+type WorkflowRestrictions struct {
+	TemplateReferencing TemplateReferencing `json:"templateReferencing"`
+}
+
+type TemplateReferencing string
+
+const (
+	TemplateReferencingStrict TemplateReferencing = "Strict"
+	TemplateReferencingSecure TemplateReferencing = "Secure"
+)
+
+func (req *WorkflowRestrictions) MustUseReference() bool {
+	if req == nil {
+		return false
+	}
+	return req.TemplateReferencing == TemplateReferencingStrict || req.TemplateReferencing == TemplateReferencingSecure
+}
+
+func (req *WorkflowRestrictions) MustNotChangeSpec() bool {
+	if req == nil {
+		return false
+	}
+	return req.TemplateReferencing == TemplateReferencingSecure
+}
diff --git a/docs/README.md b/docs/README.md
index e10d6db12..a1e6ba165 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -59,4 +59,5 @@ Some use-case specific documentation is available:
 * [Workflow Executors](workflow-executors.md)
 * [Workflow Notifications](workflow-notifications.md)
 * [Workfow RBAC](workflow-rbac.md)
+* [Workflow Restrictions](workflow-requirements.md)
 * [Workflow Templates](workflow-templates.md)
diff --git a/docs/workflow-controller-configmap.yaml b/docs/workflow-controller-configmap.yaml
index 118224c21..5e272bd93 100644
--- a/docs/workflow-controller-configmap.yaml
+++ b/docs/workflow-controller-configmap.yaml
@@ -220,3 +220,13 @@ data:
       # be in the form <argo-server-root-url>/oauth2/callback. It must be
       # browser-accessible.
       redirectUrl: https://argo-server/oauth2/callback
+
+    # workflowRequirements restricts the Workflows that the controller will process.
+    # Current options:
+    #   referenceOnly: Only Workflows using "workflowTemplateRef" will be processed. This allows the administrator of the controller
+    #     to set a "library" of templates that may be run by its opeartor, limiting arbitrary Workflow execution.
+    #   strictReferenceOnly: Only Workflows using "workflowTemplateRef" will be processed and the controller will enforce
+    #     that the WorkflowTemplate that is referenced hasn't changed between operations. If you want to make sure the operator of the
+    #     Workflow cannot run an arbitrary Workflow, use this option.
+    workflowRequirements:
+      referenceOnly: true
diff --git a/docs/workflow-requirements.md b/docs/workflow-requirements.md
new file mode 100644
index 000000000..9aef7c9c0
--- /dev/null
+++ b/docs/workflow-requirements.md
@@ -0,0 +1,31 @@
+# Workflow Restrictions
+
+![alpha](assets/alpha.svg)
+
+> v2.9 and after
+
+## Introduction
+
+As the administrator of the controller, you may want to limit which types of Workflows your users can run. Setting workflow restrictions allows you to ensure that Workflows comply with certain requirements. 
+
+## Available Restrictions
+
+* `templateReferencing: Strict`: Only Workflows using "workflowTemplateRef" will be processed. This allows the administrator of the controller to set a "library" of templates that may be run by its opeartor, limiting arbitrary Workflow execution.
+* `templateReferencing: Secure`: Only Workflows using "workflowTemplateRef" will be processed and the controller will enforce that the WorkflowTemplate that is referenced hasn't changed between operations. If you want to make sure the operator of the Workflow cannot run an arbitrary Workflow, use this option.
+
+## Setting Workflow Restrictions
+
+Workflow Restrictions can be specified by adding them under the `workflowRestrictions` key in the [`workflow-controller-configmap`](./workflow-controller-configmap.yaml).
+
+For example, to specify that Workflows may only run with `workflowTemplateRef`
+
+```yaml
+# This file describes the config settings available in the workflow controller configmap
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: workflow-controller-configmap
+data: |
+  workflowRestrictions:
+    templateReferencing: Secure
+```
diff --git a/pkg/apis/workflow/v1alpha1/workflow_types.go b/pkg/apis/workflow/v1alpha1/workflow_types.go
index fc63dc354..903041373 100644
--- a/pkg/apis/workflow/v1alpha1/workflow_types.go
+++ b/pkg/apis/workflow/v1alpha1/workflow_types.go
@@ -1835,6 +1835,13 @@ func (wf *Workflow) GetTemplateByName(name string) *Template {
 			return &t
 		}
 	}
+	if wf.Status.StoredWorkflowSpec != nil {
+		for _, t := range wf.Status.StoredWorkflowSpec.Templates {
+			if t.Name == name {
+				return &t
+			}
+		}
+	}
 	return nil
 }
 
diff --git a/workflow/common/common.go b/workflow/common/common.go
index f98076b24..eeddf0955 100644
--- a/workflow/common/common.go
+++ b/workflow/common/common.go
@@ -57,6 +57,8 @@ const (
 	LabelKeyWorkflow = workflow.WorkflowFullName + "/workflow"
 	// LabelKeyPhase is a label applied to workflows to indicate the current phase of the workflow (for filtering purposes)
 	LabelKeyPhase = workflow.WorkflowFullName + "/phase"
+	// LabelKeyPreviousWorkflowName is a label applied to resubmitted workflows
+	LabelKeyPreviousWorkflowName = workflow.WorkflowFullName + "/resubmitted-from-workflow"
 	// LabelKeyCronWorkflow is a label applied to Workflows that are started by a CronWorkflow
 	LabelKeyCronWorkflow = workflow.WorkflowFullName + "/cron-workflow"
 	// LabelKeyWorkflowTemplate is a label applied to Workflows that are submitted from Workflowtemplate
diff --git a/workflow/controller/controller.go b/workflow/controller/controller.go
index fee629b86..184dd5411 100644
--- a/workflow/controller/controller.go
+++ b/workflow/controller/controller.go
@@ -96,17 +96,7 @@ const (
 )
 
 // NewWorkflowController instantiates a new WorkflowController
-func NewWorkflowController(
-	restConfig *rest.Config,
-	kubeclientset kubernetes.Interface,
-	wfclientset wfclientset.Interface,
-	namespace string,
-	managedNamespace string,
-	executorImage,
-	executorImagePullPolicy,
-	containerRuntimeExecutor,
-	configMap string,
-) *WorkflowController {
+func NewWorkflowController(restConfig *rest.Config, kubeclientset kubernetes.Interface, wfclientset wfclientset.Interface, namespace string, managedNamespace string, executorImage, executorImagePullPolicy, containerRuntimeExecutor, configMap string) *WorkflowController {
 	eventBroadcaster := record.NewBroadcaster()
 	eventBroadcaster.StartLogging(log.Debugf)
 	eventBroadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{Interface: kubeclientset.CoreV1().Events(namespace)})
diff --git a/workflow/controller/operator.go b/workflow/controller/operator.go
index 5fee42038..14699c270 100644
--- a/workflow/controller/operator.go
+++ b/workflow/controller/operator.go
@@ -2700,20 +2700,22 @@ func (woc *wfOperationCtx) getArtifactRepositoryByRef(arRef *wfv1.ArtifactReposi
 }
 
 func (woc *wfOperationCtx) fetchWorkflowSpec() (*wfv1.WorkflowSpec, error) {
+	if woc.wf.Spec.WorkflowTemplateRef == nil {
+		return nil, fmt.Errorf("cannot fetch workflow spec without workflowTemplateRef")
+	}
+
 	var specHolder wfv1.WorkflowSpecHolder
 	var err error
 	// Logic for workflow refers Workflow template
-	if woc.wf.Spec.WorkflowTemplateRef != nil {
-		if woc.wf.Spec.WorkflowTemplateRef.ClusterScope {
-			specHolder, err = woc.controller.cwftmplInformer.Lister().Get(woc.wf.Spec.WorkflowTemplateRef.Name)
-		} else {
-			specHolder, err = woc.controller.wftmplInformer.Lister().WorkflowTemplates(woc.wf.Namespace).Get(woc.wf.Spec.WorkflowTemplateRef.Name)
-		}
-		if err != nil {
-			return nil, err
-		}
+	if woc.wf.Spec.WorkflowTemplateRef.ClusterScope {
+		specHolder, err = woc.controller.cwftmplInformer.Lister().Get(woc.wf.Spec.WorkflowTemplateRef.Name)
+	} else {
+		specHolder, err = woc.controller.wftmplInformer.Lister().WorkflowTemplates(woc.wf.Namespace).Get(woc.wf.Spec.WorkflowTemplateRef.Name)
 	}
-	return specHolder.GetWorkflowSpec().DeepCopy(), nil
+	if err != nil {
+		return nil, err
+	}
+	return specHolder.GetWorkflowSpec(), nil
 }
 
 func (woc *wfOperationCtx) loadExecutionSpec() (wfv1.TemplateReferenceHolder, wfv1.Arguments, error) {
@@ -2721,17 +2723,30 @@ func (woc *wfOperationCtx) loadExecutionSpec() (wfv1.TemplateReferenceHolder, wf
 	executionParameters := woc.wf.Spec.Arguments
 
 	if woc.wf.Spec.WorkflowTemplateRef == nil {
-		tmplRef := &wfv1.WorkflowStep{Template: woc.wfSpec.Entrypoint}
+		if woc.controller.Config.WorkflowRestrictions.MustUseReference() {
+			return nil, executionParameters, fmt.Errorf("workflows must use workflowTemplateRef to be executed when the controller is in reference mode")
+		}
+
+		tmplRef := &wfv1.WorkflowStep{Template: woc.wf.Spec.Entrypoint}
 		return tmplRef, executionParameters, nil
 	}
 
 	if woc.wf.Status.StoredWorkflowSpec == nil {
 		wftSpec, err := woc.fetchWorkflowSpec()
 		if err != nil {
-			return nil, wfv1.Arguments{}, err
+			return nil, executionParameters, err
 		}
 		woc.wf.Status.StoredWorkflowSpec = wftSpec
 		woc.updated = true
+	} else if woc.controller.Config.WorkflowRestrictions.MustNotChangeSpec() {
+		// If the controller is in reference mode, ensure that the stored spec is identical to the reference spec at every operation
+		wftSpec, err := woc.fetchWorkflowSpec()
+		if err != nil {
+			return nil, executionParameters, err
+		}
+		if woc.wf.Status.StoredWorkflowSpec.String() != wftSpec.String() {
+			return nil, executionParameters, fmt.Errorf("workflowTemplateRef reference may not change during execution when the controller is in reference mode")
+		}
 	}
 
 	woc.wfSpec = woc.wf.Status.StoredWorkflowSpec
diff --git a/workflow/util/util.go b/workflow/util/util.go
index 8abddbbfe..dcba90137 100644
--- a/workflow/util/util.go
+++ b/workflow/util/util.go
@@ -493,15 +493,18 @@ func FormulateResubmitWorkflow(wf *wfv1.Workflow, memoized bool) (*wfv1.Workflow
 
 	// carry over user labels and annotations from previous workflow.
 	// skip any argoproj.io labels except for the controller instanceID label.
+	if newWF.ObjectMeta.Labels == nil {
+		newWF.ObjectMeta.Labels = make(map[string]string)
+	}
 	for key, val := range wf.ObjectMeta.Labels {
 		if strings.HasPrefix(key, workflow.WorkflowFullName+"/") && key != common.LabelKeyControllerInstanceID {
 			continue
 		}
-		if newWF.ObjectMeta.Labels == nil {
-			newWF.ObjectMeta.Labels = make(map[string]string)
-		}
 		newWF.ObjectMeta.Labels[key] = val
 	}
+	// Append an additional label so it's easy for user to see the
+	// name of the original workflow that has been resubmitted.
+	newWF.ObjectMeta.Labels[common.LabelKeyPreviousWorkflowName] = wf.ObjectMeta.Name
 	for key, val := range wf.ObjectMeta.Annotations {
 		if newWF.ObjectMeta.Annotations == nil {
 			newWF.ObjectMeta.Annotations = make(map[string]string)


Test patch:
diff --git a/test/e2e/functional_test.go b/test/e2e/functional_test.go
index a68a3c91e..7c5bcdcd4 100644
--- a/test/e2e/functional_test.go
+++ b/test/e2e/functional_test.go
@@ -582,6 +582,23 @@ func (s *FunctionalSuite) TestOptionalInputArtifacts() {
 		})
 }
 
+func (s *FunctionalSuite) TestWorkflowTemplateRefWithExitHandler() {
+	s.Given().
+		WorkflowTemplate("@smoke/workflow-template-whalesay-template.yaml").
+		When().
+		CreateWorkflowTemplates()
+	s.Given().
+		Workflow("@testdata/workflow-template-ref-exithandler.yaml").
+		When().
+		SubmitWorkflow().
+		WaitForWorkflow(30 * time.Second).
+		Then().
+		ExpectWorkflow(func(t *testing.T, _ *metav1.ObjectMeta, status *wfv1.WorkflowStatus) {
+			assert.Equal(t, wfv1.NodeSucceeded, status.Phase)
+		})
+
+}
+
 func TestFunctionalSuite(t *testing.T) {
 	suite.Run(t, new(FunctionalSuite))
 }
diff --git a/test/e2e/smoke/workflow-template-whalesay-template.yaml b/test/e2e/smoke/workflow-template-whalesay-template.yaml
index 704e79bad..fb810ae81 100644
--- a/test/e2e/smoke/workflow-template-whalesay-template.yaml
+++ b/test/e2e/smoke/workflow-template-whalesay-template.yaml
@@ -6,6 +6,7 @@ metadata:
     argo-e2e: true
 spec:
   entrypoint: whalesay-template
+  onExit: on-exit
   arguments:
     parameters:
       - name: message
@@ -19,3 +20,8 @@ spec:
         image: argoproj/argosay:v2
         args: ["echo", "{{inputs.parameters.message}}"]
         imagePullPolicy: IfNotPresent
+    - name: on-exit
+      container:
+        image: argoproj/argosay:v2
+        args: ["echo", "Exit Handler"]
+        imagePullPolicy: IfNotPresent
\ No newline at end of file
diff --git a/test/e2e/testdata/workflow-template-ref-exithandler.yaml b/test/e2e/testdata/workflow-template-ref-exithandler.yaml
new file mode 100644
index 000000000..4f717bc83
--- /dev/null
+++ b/test/e2e/testdata/workflow-template-ref-exithandler.yaml
@@ -0,0 +1,14 @@
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  generateName: workflow-template-hello-world-
+  labels:
+    argo-e2e: "true"
+spec:
+  entrypoint: whalesay-template
+  arguments:
+    parameters:
+      - name: message
+        value: "hello world"
+  workflowTemplateRef:
+    name: workflow-template-whalesay-template
diff --git a/test/test.go b/test/test.go
index df2634076..836c61743 100644
--- a/test/test.go
+++ b/test/test.go
@@ -51,6 +51,25 @@ func LoadWorkflowFromBytes(yamlBytes []byte) *wfv1.Workflow {
 	return &wf
 }
 
+// LoadTestWorkflow returns a workflow relative to the test file
+func LoadTestWorkflowTemplate(path string) *wfv1.WorkflowTemplate {
+	yamlBytes, err := ioutil.ReadFile(path)
+	if err != nil {
+		panic(err)
+	}
+	return LoadWorkflowTemplateFromBytes(yamlBytes)
+}
+
+// LoadWorkflowFromBytes returns a workflow unmarshalled from an yaml byte array
+func LoadWorkflowTemplateFromBytes(yamlBytes []byte) *wfv1.WorkflowTemplate {
+	var wf wfv1.WorkflowTemplate
+	err := yaml.Unmarshal(yamlBytes, &wf)
+	if err != nil {
+		panic(err)
+	}
+	return &wf
+}
+
 // LoadUnstructuredFromBytes returns an Unstructured unmarshalled from an yaml byte array
 func LoadUnstructuredFromBytes(yamlBytes []byte) *unstructured.Unstructured {
 	var un unstructured.Unstructured
diff --git a/workflow/controller/controller_test.go b/workflow/controller/controller_test.go
index bdb6f737b..b85bc1405 100644
--- a/workflow/controller/controller_test.go
+++ b/workflow/controller/controller_test.go
@@ -370,20 +370,14 @@ spec:
 `
 
 func TestCheckAndInitWorkflowTmplRef(t *testing.T) {
-	//_, controller := newController()
 	wf := unmarshalWF(wfWithTmplRef)
 	wftmpl := unmarshalWFTmpl(wfTmpl)
 	_, controller := newController(wf, wftmpl)
-	//_, err := controller.wfclientset.ArgoprojV1alpha1().WorkflowTemplates("default").Create(wftmpl)
-	//assert.NoError(t, err)
 	woc := wfOperationCtx{controller: controller,
 		wf: wf}
-	t.Run("WithWorkflowTmplRef", func(t *testing.T) {
-		_, _, err := woc.loadExecutionSpec()
-		assert.NoError(t, err)
-		assert.Equal(t, &wftmpl.Spec.WorkflowSpec, woc.wfSpec)
-
-	})
+	_, _, err := woc.loadExecutionSpec()
+	assert.NoError(t, err)
+	assert.Equal(t, &wftmpl.Spec.WorkflowSpec, woc.wfSpec)
 }
 
 func TestIsArchivable(t *testing.T) {
diff --git a/workflow/controller/operator_test.go b/workflow/controller/operator_test.go
index f6a617710..4981189ff 100644
--- a/workflow/controller/operator_test.go
+++ b/workflow/controller/operator_test.go
@@ -3687,3 +3687,55 @@ func TestPanicMetric(t *testing.T) {
 	}
 	assert.True(t, seen)
 }
+
+// Assert Workflows cannot be run without using workflowTemplateRef in reference mode
+func TestControllerReferenceMode(t *testing.T) {
+	wf := unmarshalWF(globalVariablePlaceholders)
+	cancel, controller := newController()
+	defer cancel()
+	controller.Config.WorkflowRestrictions = &config.WorkflowRestrictions{}
+	controller.Config.WorkflowRestrictions.TemplateReferencing = config.TemplateReferencingStrict
+	woc := newWorkflowOperationCtx(wf, controller)
+	woc.operate()
+	assert.Equal(t, wfv1.NodeError, woc.wf.Status.Phase)
+	assert.Equal(t, "workflows must use workflowTemplateRef to be executed when the controller is in reference mode", woc.wf.Status.Message)
+
+	controller.Config.WorkflowRestrictions.TemplateReferencing = config.TemplateReferencingSecure
+	woc = newWorkflowOperationCtx(wf, controller)
+	woc.operate()
+	assert.Equal(t, wfv1.NodeError, woc.wf.Status.Phase)
+	assert.Equal(t, "workflows must use workflowTemplateRef to be executed when the controller is in reference mode", woc.wf.Status.Message)
+
+	controller.Config.WorkflowRestrictions = nil
+	woc = newWorkflowOperationCtx(wf, controller)
+	woc.operate()
+	assert.Equal(t, wfv1.NodeRunning, woc.wf.Status.Phase)
+}
+
+func TestValidReferenceMode(t *testing.T) {
+	wf := test.LoadTestWorkflow("testdata/workflow-template-ref.yaml")
+	wfTmpl := test.LoadTestWorkflowTemplate("testdata/workflow-template-submittable.yaml")
+	cancel, controller := newController(wf, wfTmpl)
+	defer cancel()
+	controller.Config.WorkflowRestrictions = &config.WorkflowRestrictions{}
+	controller.Config.WorkflowRestrictions.TemplateReferencing = config.TemplateReferencingSecure
+	woc := newWorkflowOperationCtx(wf, controller)
+	woc.operate()
+	assert.Equal(t, wfv1.NodeRunning, woc.wf.Status.Phase)
+
+	// Change stored Workflow Spec
+	woc.wf.Status.StoredWorkflowSpec.Entrypoint = "different"
+	woc.operate()
+	assert.Equal(t, wfv1.NodeError, woc.wf.Status.Phase)
+	assert.Equal(t, "workflowTemplateRef reference may not change during execution when the controller is in reference mode", woc.wf.Status.Message)
+
+	controller.Config.WorkflowRestrictions.TemplateReferencing = config.TemplateReferencingStrict
+	woc = newWorkflowOperationCtx(wf, controller)
+	woc.operate()
+	assert.Equal(t, wfv1.NodeRunning, woc.wf.Status.Phase)
+
+	// Change stored Workflow Spec
+	woc.wf.Status.StoredWorkflowSpec.Entrypoint = "different"
+	woc.operate()
+	assert.Equal(t, wfv1.NodeRunning, woc.wf.Status.Phase)
+}
diff --git a/workflow/controller/testdata/workflow-template-ref.yaml b/workflow/controller/testdata/workflow-template-ref.yaml
new file mode 100644
index 000000000..c08109716
--- /dev/null
+++ b/workflow/controller/testdata/workflow-template-ref.yaml
@@ -0,0 +1,8 @@
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  generateName: workflow-template-hello-world-
+  namespace: test
+spec:
+  workflowTemplateRef:
+    name: workflow-template-submittable
diff --git a/workflow/controller/testdata/workflow-template-submittable.yaml b/workflow/controller/testdata/workflow-template-submittable.yaml
new file mode 100644
index 000000000..a6b786863
--- /dev/null
+++ b/workflow/controller/testdata/workflow-template-submittable.yaml
@@ -0,0 +1,13 @@
+apiVersion: argoproj.io/v1alpha1
+kind: WorkflowTemplate
+metadata:
+  name: workflow-template-submittable
+  namespace: test
+spec:
+  entrypoint: whalesay-template
+  templates:
+    - name: whalesay-template
+      container:
+        image: docker/whalesay
+        command: [cowsay]
+        args: ['hello']