Repository: argoproj/argo (go)

Issue:
support optional Input artifacts when artifacts not exist
# Summary

support optional Input artifacts when artifacts not exist.

# Motivation

![image](https://user-images.githubusercontent.com/24826076/87668140-281d7200-c79e-11ea-84d1-4d04feac51dc.png)

1. `example_a` output `condition_b1:true` and `condition_b2:false`
2. `condition_1` satisfied and execute `exapmle_b`, while `condition_2` not satisfied
3. `example_c` get `condition_b1` and `condition_b2` outputs with optional field: 
```
 inputs:
      artifacts:
      - {name: example-b-result_b, optional: true,  path: /tmp/inputs/result_b1/data}
      - {name: example-b-2-result_b,  optional: true, path: /tmp/inputs/result_b2/data}
```

because `condition_2` not satisfied, so get error:
`Unable to resolve: {{tasks.condition-2.outputs.artifacts.example-b-2-result_b}}`

detail workflow yaml:
```
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: test-
spec:
  entrypoint: test
  templates:
  - name: condition-1
    outputs:
      artifacts:
      - {name: example-b-result_b, from: '{{tasks.example-b.outputs.artifacts.example-b-result_b}}'}
    dag:
      tasks:
      - {name: example-b, template: example-b}
  - name: condition-2
    outputs:
      artifacts:
      - {name: example-b-2-result_b, from: '{{tasks.example-b-2.outputs.artifacts.example-b-2-result_b}}'}
    dag:
      tasks:
      - {name: example-b-2, template: example-b-2}
  - name: example-a
    container:
      args: [components.train.test.runa, run, --condition_b1, /tmp/outputs/condition_b1/data,
        --condition_b2, /tmp/outputs/condition_b2/data]
      command: [python, -u, -m, components.launcher]
      image: xxx:test
      workingDir: /pipelines
      volumeMounts:
      - {mountPath: /pipelines, name: pipeline}
    outputs:
      parameters:
      - name: example-a-condition_b1
        valueFrom: {path: /tmp/outputs/condition_b1/data}
      - name: example-a-condition_b2
        valueFrom: {path: /tmp/outputs/condition_b2/data}
      artifacts:
      - {name: example-a-condition_b1, path: /tmp/outputs/condition_b1/data}
      - {name: example-a-condition_b2, path: /tmp/outputs/condition_b2/data}
    volumes:
    - hostPath: {path: /pipeline, type: Directory}
      name: pipeline
  - name: example-b
    container:
      args: [components.train.test.runb, run, --type_b, bbb1, --result_b, /tmp/outputs/result_b/data]
      command: [python, -u, -m, components.launcher]
      image: xxx:test
      workingDir: /pipelines
      volumeMounts:
      - {mountPath: /pipelines, name: pipeline}
    outputs:
      artifacts:
      - {name: example-b-result_b, path: /tmp/outputs/result_b/data}
    volumes:
    - hostPath: {path: /pipeline, type: Directory}
      name: pipeline
  - name: example-b-2
    container:
      args: [components.train.test.runb, run, --type_b, bbb2, --result_b, /tmp/outputs/result_b/data]
      command: [python, -u, -m, components.launcher]
      image: xxx:test
      workingDir: /pipelines
      volumeMounts:
      - {mountPath: /pipelines, name: pipeline}
    outputs:
      artifacts:
      - {name: example-b-2-result_b, path: /tmp/outputs/result_b/data}
    volumes:
    - hostPath: {path: /pipeline, type: Directory}
      name: pipeline
  - name: example-c
    container:
      args: [components.train.test.runc, run, --result_b1, /tmp/inputs/result_b1/data,
        --result_b2, /tmp/inputs/result_b2/data]
      command: [python, -u, -m, components.launcher]
      image: xxx:test
      workingDir: /pipelines
      volumeMounts:
      - {mountPath: /pipelines, name: pipeline}
    inputs:
      artifacts:
      - {name: example-b-result_b, optional: true,  path: /tmp/inputs/result_b1/data}
      - {name: example-b-2-result_b,  optional: true, path: /tmp/inputs/result_b2/data}
    volumes:
    - hostPath: {path: /pipeline, type: Directory}
      name: pipeline
  - name: test
    dag:
      tasks:
      - name: condition-1
        template: condition-1
        when: '"{{tasks.example-a.outputs.parameters.example-a-condition_b1}}" ==
          "true"'
        dependencies: [example-a]
      - name: condition-2
        template: condition-2
        when: '"{{tasks.example-a.outputs.parameters.example-a-condition_b2}}" ==
          "true"'
        dependencies: [example-a]
      - {name: example-a, template: example-a}
      - name: example-c
        template: example-c
        dependencies: [condition-1, condition-2]
        arguments:
          artifacts:
          - {name: example-b-2-result_b, from: '{{tasks.condition-2.outputs.artifacts.example-b-2-result_b}}'}
          - {name: example-b-result_b, from: '{{tasks.condition-1.outputs.artifacts.example-b-result_b}}'}
  arguments:
    parameters: []
```

# Proposal

How do you think this should be implemented?

when artifact not exist and specified with `optional: true` :  return nil instead report error. 

---
<!-- Issue Author: Don't delete this message to encourage other users to support your issue! -->
**Message from the maintainers**:

If you wish to see this enhancement implemented please add a 👍 reaction to this issue! We often sort issues this way to know what to prioritize.


Pull request description:
fix(artifacts): support optional input artifacts, Fixes #3491
Closes: https://github.com/argoproj/argo/issues/3491

Patch:
diff --git a/workflow/controller/dag.go b/workflow/controller/dag.go
index 8165205a0..baced0118 100644
--- a/workflow/controller/dag.go
+++ b/workflow/controller/dag.go
@@ -539,6 +539,10 @@ func (woc *wfOperationCtx) resolveDependencyReferences(dagCtx *dagContext, task
 		}
 		resolvedArt, err := scope.resolveArtifact(art.From)
 		if err != nil {
+			if strings.Contains(err.Error(), "Unable to resolve") && art.Optional {
+				woc.log.Warnf("Optional artifact '%s' was not found; it won't be available as an input", art.Name)
+				continue
+			}
 			return nil, err
 		}
 		resolvedArt.Name = art.Name


Test patch:
diff --git a/workflow/controller/dag_test.go b/workflow/controller/dag_test.go
index ab72311c7..2c71470b8 100644
--- a/workflow/controller/dag_test.go
+++ b/workflow/controller/dag_test.go
@@ -1772,3 +1772,106 @@ func TestOnExitNonLeaf(t *testing.T) {
 	}
 	assert.Equal(t, wfv1.NodeRunning, woc.wf.Status.Phase)
 }
+
+var testDagOptionalInputArtifacts = `
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  name: dag-optional-inputartifacts
+spec:
+  entrypoint: test
+  templates:
+  - name: condition
+    outputs:
+      artifacts:
+      - {name: A-out, from: '{{tasks.A.outputs.artifacts.A-out}}'}
+    dag:
+      tasks:
+      - {name: A, template: A}
+  - name: A
+    container:
+      args: ['mkdir -p /tmp/outputs/A && echo "exist" > /tmp/outputs/A/data']
+      command: [sh, -c]
+      image: alpine:3.7
+    outputs:
+      artifacts:
+      - {name: A-out, path: /tmp/outputs/A/data}
+  - name: B
+    container:
+      args: ['[ -f /tmp/outputs/condition/data ] && cat /tmp/outputs/condition/data || echo not exist']
+      command: [sh, -c]
+      image: alpine:3.7
+    inputs:
+      artifacts:
+      - {name: B-in, optional: true,  path: /tmp/outputs/condition/data}
+  - name: test
+    dag:
+      tasks:
+      - name: condition
+        template: condition
+        when: 'false'
+      - name: B
+        template: B
+        dependencies: [condition]
+        arguments:
+          artifacts:
+          - {name: B-in, optional: true, from: '{{tasks.condition.outputs.artifacts.A-out}}'}
+  arguments:
+    parameters: []
+status:
+  conditions:
+  - status: "True"
+    type: Completed
+  finishedAt: "2020-07-21T01:56:24Z"
+  nodes:
+    dag-optional-inputartifacts:
+      children:
+      - dag-optional-inputartifacts-3418089753
+      displayName: dag-optional-inputartifacts
+      finishedAt: "2020-07-21T01:56:24Z"
+      id: dag-optional-inputartifacts
+      name: dag-optional-inputartifacts
+      outboundNodes:
+      - dag-optional-inputartifacts-1920355018
+      phase: Running
+      startedAt: "2020-07-21T01:56:18Z"
+      templateName: test
+      templateScope: local/dag-optional-inputartifacts
+      type: DAG
+    dag-optional-inputartifacts-3418089753:
+      boundaryID: dag-optional-inputartifacts
+      children:
+      - dag-optional-inputartifacts-1920355018
+      displayName: condition
+      finishedAt: "2020-07-21T01:56:18Z"
+      id: dag-optional-inputartifacts-3418089753
+      message: when 'false' evaluated false
+      name: dag-optional-inputartifacts.condition
+      phase: Skipped
+      startedAt: "2020-07-21T01:56:18Z"
+      templateName: condition
+      templateScope: local/dag-optional-inputartifacts
+      type: Skipped
+  phase: Running
+  resourcesDuration:
+    cpu: 1
+    memory: 0
+  startedAt: "2020-07-21T01:56:18Z"
+`
+
+func TestDagOptionalInputArtifacts(t *testing.T) {
+	cancel, controller := newController()
+	defer cancel()
+	wfcset := controller.wfclientset.ArgoprojV1alpha1().Workflows("")
+
+	wf := unmarshalWF(testDagOptionalInputArtifacts)
+	wf, err := wfcset.Create(wf)
+	assert.NoError(t, err)
+	woc := newWorkflowOperationCtx(wf, controller)
+
+	woc.operate()
+	optionalInputArtifactsNode := woc.wf.GetNodeByName("dag-optional-inputartifacts.B")
+	if assert.NotNil(t, optionalInputArtifactsNode) {
+		assert.Equal(t, wfv1.NodePending, optionalInputArtifactsNode.Phase)
+	}
+}