Repository: argoproj/argo (go)

Issue:
Support DAG outputs
<!-- Also consider asking your question on our Slack channel. See the README for more info! -->
I did ask in the slack channel, but unfortunately I was not able to get a response.

# Summary

What do you want to know about this project?

Currently it is not possible to have dag-level outputs (referencing a step output) with the k8s api executor, as the validation doesn't allow outputs that don't reference a volumePath. Is this intended or is the validation just too strict for this particular use-case?


# Motivation

I am currently evaluating how we can run argo on OpenShift 3.11 and therefore I am having a look at the k8s api executor.
I am currently using a workflow template which consists of a DAG that outputs an artifact from one of its subtasks (similar to the example below).

```
  templates:
    - name: dag
      outputs:
        artifacts:
          - name: artifact
            from: "{{ tasks.task-2.outputs.artifacts.artifact }}"
      dag:
        tasks:
          - name: task-1
            template: sleep-5-param-output
          - name: task-2
            template: sleep-5-artifact-output
```

I am using an emptyDir within the 'task-2' so the artifact can correctly be outputted by the step (I have verified this also).

However my 'dag' template is rejected by the validation with the message: invalid spec: templates.dag.outputs.artifacts.artifact: k8sapi executor does not support outputs from base image layer. must use emptyDir

I'm not sure if this kind of output-artifact from a dag is simply not supported by the k8s-api executor, or if the validation is too restrictive (a quick look at the code showed that this might be the case as the validation does not seem to consider this use-case).

Thank you for considering my problem! If this is indeed a bug and not intended behavior I would be more than glad to file a bug-report including all necessary information.


Pull request:
fix: DAG level Output Artifacts on K8S and Kubelet executor
Checklist:

* [ ] 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.
* [ ] 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"`.  Fixes #3545 
* [ ] I've signed the CLA.
* [ ] I have written unit and/or e2e tests for my change. PRs without these are unlikely to be merged.
* [ ] My builds are green. Try syncing with master if they are not. 
* [ ] My organization is added to [USERS.md](https://github.com/argoproj/argo/blob/master/USERS.md).


Patch:
diff --git a/workflow/validate/validate.go b/workflow/validate/validate.go
index 29b467496..634d99e9a 100644
--- a/workflow/validate/validate.go
+++ b/workflow/validate/validate.go
@@ -1002,6 +1002,10 @@ func validateOutputs(scope map[string]interface{}, tmpl *wfv1.Template) error {
 
 // validateBaseImageOutputs detects if the template contains an valid output from base image layer
 func (ctx *templateValidationCtx) validateBaseImageOutputs(tmpl *wfv1.Template) error {
+	// This validation is not applicable for DAG and Step Template types
+	if tmpl.GetType() == wfv1.TemplateTypeDAG || tmpl.GetType() == wfv1.TemplateTypeSteps {
+		return nil
+	}
 	switch ctx.ContainerRuntimeExecutor {
 	case "", common.ContainerRuntimeExecutorDocker:
 		// docker executor supports all modes of artifact outputs


Test patch:
diff --git a/workflow/validate/validate_test.go b/workflow/validate/validate_test.go
index 69a8d29b3..8e89fe6c3 100644
--- a/workflow/validate/validate_test.go
+++ b/workflow/validate/validate_test.go
@@ -35,6 +35,11 @@ func validate(yamlStr string) (*wfv1.Conditions, error) {
 	return ValidateWorkflow(wftmplGetter, cwftmplGetter, wf, ValidateOpts{})
 }
 
+func validateWithOptions(yamlStr string, opts ValidateOpts) (*wfv1.Conditions, error) {
+	wf := unmarshalWf(yamlStr)
+	return ValidateWorkflow(wftmplGetter, cwftmplGetter, wf, opts)
+}
+
 // validateWorkflowTemplate is a test helper to accept WorkflowTemplate YAML as a string and return
 // its validation result.
 func validateWorkflowTemplate(yamlStr string) error {
@@ -2542,3 +2547,72 @@ func TestWorkflowWithWFTRefWithOverrideParam(t *testing.T) {
 	_, err = validate(wfWithWFTRefOverrideParam)
 	assert.NoError(t, err)
 }
+
+var dagAndStepLevelOutputArtifacts = `
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  generateName: dag-target-
+spec:
+  entrypoint: main
+  templates:
+  - name: main
+    outputs:
+      artifacts:
+        - name: artifact
+          from: "{{tasks.artifact-svn-retrieve.outputs.artifacts.artifact}}"
+    dag:
+      tasks:
+      - name: artifact-svn-retrieve
+        template: artifact-svn-retrieve
+      - name: step-tmpl
+        template: step
+
+  - name: step
+    outputs:
+      artifacts:
+        - name: artifact
+          from: "{{steps.artifact-svn-retrieve.outputs.artifacts.artifact}}"
+    steps:
+    - - name: artifact-svn-retrieve
+        template: artifact-svn-retrieve
+
+  - name: artifact-svn-retrieve
+    outputs:
+      artifacts:
+      - name: artifact
+        path: "/vol/hello_world.txt"
+    container:
+      image: docker/whalesay:latest
+      command: [sh, -c]
+      args: ["sleep 1; cowsay hello world | tee /vol/hello_world.txt"]
+      volumeMounts:
+      - name: vol
+        mountPath: "/vol"
+    volumes:
+    - name: vol
+      emptyDir: {}
+`
+
+func TestDagAndStepLevelOutputArtifactsForDiffExecutor(t *testing.T) {
+	t.Run("DefaultExecutor", func(t *testing.T) {
+		_, err := validateWithOptions(dagAndStepLevelOutputArtifacts, ValidateOpts{ContainerRuntimeExecutor: ""})
+		assert.NoError(t, err)
+	})
+	t.Run("DockerExecutor", func(t *testing.T) {
+		_, err := validateWithOptions(dagAndStepLevelOutputArtifacts, ValidateOpts{ContainerRuntimeExecutor: common.ContainerRuntimeExecutorDocker})
+		assert.NoError(t, err)
+	})
+	t.Run("PNSExecutor", func(t *testing.T) {
+		_, err := validateWithOptions(dagAndStepLevelOutputArtifacts, ValidateOpts{ContainerRuntimeExecutor: common.ContainerRuntimeExecutorPNS})
+		assert.NoError(t, err)
+	})
+	t.Run("K8SExecutor", func(t *testing.T) {
+		_, err := validateWithOptions(dagAndStepLevelOutputArtifacts, ValidateOpts{ContainerRuntimeExecutor: common.ContainerRuntimeExecutorK8sAPI})
+		assert.NoError(t, err)
+	})
+	t.Run("KubeletExecutor", func(t *testing.T) {
+		_, err := validateWithOptions(dagAndStepLevelOutputArtifacts, ValidateOpts{ContainerRuntimeExecutor: common.ContainerRuntimeExecutorKubelet})
+		assert.NoError(t, err)
+	})
+}