Repository: argoproj/argo (go)

Issue:
Cannot submit workflow from workflow template with input artifact argument
## Summary 
I've converted https://github.com/argoproj/argo/blob/master/examples/arguments-artifacts.yaml to workflow template:
```
apiVersion: argoproj.io/v1alpha1
kind: WorkflowTemplate
metadata:
  name: arguments-artifacts
spec:
  entrypoint: kubectl-input-artifact
  arguments:
    artifacts:
    - name: kubectl
      http:
        url: https://storage.googleapis.com/kubernetes-release/release/v1.8.0/bin/linux/amd64/kubectl

  templates:
  - name: kubectl-input-artifact
    inputs:
      artifacts:
      - name: kubectl
        path: /usr/local/bin/kubectl
        mode: 0755
    container:
      image: debian:9.4
      command: [sh, -c]
      args: ["kubectl version"]
```
 But when I try and submit any workflow using this template, submission fails
```
❯ argo  submit --from workflowtemplate/arguments-artifacts
2020/10/15 16:46:00 Failed to submit workflow: templates.kubectl-input-artifact inputs.artifacts.kubectl was not supplied
```

## Server/CLI version
2.11.3



Pull request description:
fix: merge artifact arguments from workflow template. Fixes #4296
Fixes #4296 

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"`.  
* [ ] My organization is added to [USERS.md](https://github.com/argoproj/argo/blob/master/USERS.md).
* [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.
* [ ] My builds are green. Try syncing with master if they are not. 


Patch:
diff --git a/USERS.md b/USERS.md
index 578e5c377..bb66900f5 100644
--- a/USERS.md
+++ b/USERS.md
@@ -34,6 +34,7 @@ Currently, the following organizations are **officially** using Argo Workflows:
 1. [Data4Risk](https://www.data4risk.com/)
 1. [Datadog](https://www.datadoghq.com/)
 1. [DataStax](https://www.datastax.com/)
+1. [DevSamurai](https://www.devsamurai.com/)
 1. [EBSCO Information Services](https://www.ebsco.com/)
 1. [Equinor](https://www.equinor.com/)
 1. [Fairwinds](https://fairwinds.com/)
diff --git a/docs/configure-artifact-repository.md b/docs/configure-artifact-repository.md
index 47c8ee298..ef65ace54 100644
--- a/docs/configure-artifact-repository.md
+++ b/docs/configure-artifact-repository.md
@@ -14,7 +14,7 @@ Subsequent sections will show how to use it.
 | HTTP | Yes | No | 2% |
 | OSS | Yes | Yes | - |
 | Raw | Yes | No | 5% |
-| S3 | Yes | Yes | 86% | 
+| S3 | Yes | Yes | 86% |
 
 ## Configuring Minio
 
@@ -142,7 +142,7 @@ access is on a per project rather than per bucket basis.
   (https://console.cloud.google.com/storage/settings).
 - Enable interoperability access if needed.
 - Create a new key if needed.
-- Configure `s3` artifact as following exmaple.
+- Configure `s3` artifact as following example.
 
 ```yaml
 artifacts:
@@ -168,7 +168,7 @@ artifacts:
 
 To configure artifact storage for Alibaba Cloud OSS, please first follow
 the [official documentation](https://www.alibabacloud.com/product/oss) to set up
-an OSS account and bucket. 
+an OSS account and bucket.
 
 Once it's set up, you can find endpoint and bucket
 information on your OSS dashboard and then use them like the following to
@@ -267,7 +267,7 @@ that you can specify a `keyFormat`.
 
 Argo also can use native GCS APIs to access a Google Cloud Storage bucket.
 
-`serviceAccountKeySecret` refereces to a k8 secret which stores a Google Cloud
+`serviceAccountKeySecret` references to a k8 secret which stores a Google Cloud
 service account key to access the bucket.
 
 Example:
diff --git a/docs/synchronization.md b/docs/synchronization.md
index 11c7bcec6..9a8880fd7 100644
--- a/docs/synchronization.md
+++ b/docs/synchronization.md
@@ -14,7 +14,7 @@ from a workflow or template within a workflow.
 For example:
 ```yaml
 apiVersion: v1
- kind: ConfigMap
+kind: ConfigMap
 metadata:
  name: my-config
 data:
diff --git a/util/util.go b/util/util.go
index 4dd56adf1..ade027ea6 100644
--- a/util/util.go
+++ b/util/util.go
@@ -83,6 +83,22 @@ func MergeParameters(params ...[]wfv1.Parameter) []wfv1.Parameter {
 	return resultParams
 }
 
+// MergeArtifacts merges artifact argument slices
+// Merge the slices based on arguments order (first is high priority).
+func MergeArtifacts(artifactSlices ...[]wfv1.Artifact) []wfv1.Artifact {
+	var result []wfv1.Artifact
+	alreadyMerged := make(map[string]bool)
+	for _, artifacts := range artifactSlices {
+		for _, item := range artifacts {
+			if !alreadyMerged[item.Name] {
+				result = append(result, item)
+				alreadyMerged[item.Name] = true
+			}
+		}
+	}
+	return result
+}
+
 func RecoverIndexFromNodeName(name string) int {
 	startIndex := strings.Index(name, "(")
 	endIndex := strings.Index(name, ":")
diff --git a/workflow/controller/operator.go b/workflow/controller/operator.go
index 78d1cb1ac..140385a68 100644
--- a/workflow/controller/operator.go
+++ b/workflow/controller/operator.go
@@ -3136,6 +3136,7 @@ func (woc *wfOperationCtx) loadExecutionSpec() (wfv1.TemplateReferenceHolder, wf
 	if len(woc.execWf.Spec.Arguments.Parameters) > 0 {
 		executionParameters.Parameters = util.MergeParameters(executionParameters.Parameters, woc.execWf.Spec.Arguments.Parameters)
 	}
+	executionParameters.Artifacts = util.MergeArtifacts(executionParameters.Artifacts, woc.execWf.Spec.Arguments.Artifacts)
 
 	return tmplRef, executionParameters, nil
 }
diff --git a/workflow/cron/controller.go b/workflow/cron/controller.go
index 6c013055f..fa14c2429 100644
--- a/workflow/cron/controller.go
+++ b/workflow/cron/controller.go
@@ -40,8 +40,7 @@ type Controller struct {
 	managedNamespace     string
 	instanceId           string
 	cron                 *cron.Cron
-	nameEntryIDMap       map[string]cron.EntryID
-	nameEntryIDMapLock   *sync.Mutex
+	nameEntryIDMap       sync.Map
 	wfClientset          versioned.Interface
 	wfInformer           cache.SharedIndexInformer
 	wfLister             util.WorkflowLister
@@ -69,8 +68,7 @@ func NewCronController(wfclientset versioned.Interface, restConfig *rest.Config,
 		cron:                 cron.New(),
 		restConfig:           restConfig,
 		dynamicInterface:     dynamicInterface,
-		nameEntryIDMap:       make(map[string]cron.EntryID),
-		nameEntryIDMapLock:   &sync.Mutex{},
+		nameEntryIDMap:       sync.Map{},
 		wfQueue:              workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "wf_cron_queue"),
 		cronWfQueue:          workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "cron_wf_queue"),
 		metrics:              metrics,
@@ -126,7 +124,7 @@ func (cc *Controller) processNextCronItem() bool {
 		return false
 	}
 	defer cc.cronWfQueue.Done(key)
-	logCtx := log.WithField("key", key)
+	logCtx := log.WithField("cronWorkflow", key)
 	logCtx.Infof("Processing %s", key)
 
 	obj, exists, err := cc.cronWfInformer.Informer().GetIndexer().GetByKey(key.(string))
@@ -134,13 +132,11 @@ func (cc *Controller) processNextCronItem() bool {
 		logCtx.WithError(err).Error(fmt.Sprintf("Failed to get CronWorkflow '%s' from informer index", key))
 		return true
 	}
-	cc.nameEntryIDMapLock.Lock()
-	defer cc.nameEntryIDMapLock.Unlock()
 	if !exists {
-		if entryId, ok := cc.nameEntryIDMap[key.(string)]; ok {
+		if entryId, ok := cc.nameEntryIDMap.Load(key.(string)); ok {
 			logCtx.Infof("Deleting '%s'", key)
-			cc.cron.Remove(entryId)
-			delete(cc.nameEntryIDMap, key.(string))
+			cc.cron.Remove(entryId.(cron.EntryID))
+			cc.nameEntryIDMap.Delete(key.(string))
 		}
 		return true
 	}
@@ -173,9 +169,9 @@ func (cc *Controller) processNextCronItem() bool {
 	}
 
 	// The job is currently scheduled, remove it and re add it.
-	if entryId, ok := cc.nameEntryIDMap[key.(string)]; ok {
-		cc.cron.Remove(entryId)
-		delete(cc.nameEntryIDMap, key.(string))
+	if entryId, ok := cc.nameEntryIDMap.Load(key.(string)); ok {
+		cc.cron.Remove(entryId.(cron.EntryID))
+		cc.nameEntryIDMap.Delete(key.(string))
 	}
 
 	cronSchedule := cronWf.Spec.Schedule
@@ -188,7 +184,7 @@ func (cc *Controller) processNextCronItem() bool {
 		logCtx.WithError(err).Error("could not schedule CronWorkflow")
 		return true
 	}
-	cc.nameEntryIDMap[key.(string)] = entryId
+	cc.nameEntryIDMap.Store(key.(string), entryId)
 
 	logCtx.Infof("CronWorkflow %s added", key.(string))
 
@@ -242,10 +238,8 @@ func (cc *Controller) processNextWorkflowItem() bool {
 	// Workflows are run in the same namespace as CronWorkflow
 	nameEntryIdMapKey := wf.Namespace + "/" + wf.OwnerReferences[0].Name
 	var woc *cronWfOperationCtx
-	cc.nameEntryIDMapLock.Lock()
-	defer cc.nameEntryIDMapLock.Unlock()
-	if entryId, ok := cc.nameEntryIDMap[nameEntryIdMapKey]; ok {
-		woc, ok = cc.cron.Entry(entryId).Job.(*cronWfOperationCtx)
+	if entryId, ok := cc.nameEntryIDMap.Load(nameEntryIdMapKey); ok {
+		woc, ok = cc.cron.Entry(entryId.(cron.EntryID)).Job.(*cronWfOperationCtx)
 		if !ok {
 			log.Errorf("Parent CronWorkflow '%s' is malformed", nameEntryIdMapKey)
 			return true
diff --git a/workflow/executor/pns/pns.go b/workflow/executor/pns/pns.go
index 7d20666a0..f3a0f5845 100644
--- a/workflow/executor/pns/pns.go
+++ b/workflow/executor/pns/pns.go
@@ -393,21 +393,38 @@ func parseContainerID(pid int) (string, error) {
 	defer func() { _ = cgroupFile.Close() }()
 	sc := bufio.NewScanner(cgroupFile)
 	for sc.Scan() {
-		// See https://www.systutorials.com/docs/linux/man/5-proc/ for /proc/XX/cgroup format. e.g.:
-		// 5:cpuacct,cpu,cpuset:/daemons
 		line := sc.Text()
 		log.Debugf("pid %d: %s", pid, line)
-		parts := strings.Split(line, "/")
-		if len(parts) > 1 {
-			if containerID := parts[len(parts)-1]; containerID != "" {
-				// need to check for empty string because the line may look like: 5:rdma:/
-
-				// for crio we need to get rid of "crio-" prefix and ".scope" suffix
-				// e.g. crio-7a92a067289f6197148912be1c15f20f0330c7f3c541473d3b9c4043ca137b42.scope
-				containerID := strings.TrimSuffix(strings.TrimPrefix(containerID, "crio-"), ".scope")
-				return containerID, nil
-			}
+		containerID := parseContainerIDFromCgroupLine(line)
+		if containerID != "" {
+			return containerID, nil
 		}
 	}
 	return "", errors.InternalErrorf("Failed to parse container ID from %s", cgroupPath)
 }
+
+func parseContainerIDFromCgroupLine(line string) string {
+	// See https://www.systutorials.com/docs/linux/man/5-proc/ for /proc/XX/cgroup format. e.g.:
+	// 5:cpuacct,cpu,cpuset:/daemons
+	parts := strings.Split(line, "/")
+	if len(parts) > 1 {
+		if containerID := parts[len(parts)-1]; containerID != "" {
+			// need to check for empty string because the line may look like: 5:rdma:/
+
+			// for crio we need to get rid of "crio-" prefix and ".scope" suffix
+			// e.g. crio-7a92a067289f6197148912be1c15f20f0330c7f3c541473d3b9c4043ca137b42.scope
+			containerID := strings.TrimSuffix(strings.TrimPrefix(containerID, "crio-"), ".scope")
+
+			// for compatibility with cri-containerd record format when using systemd cgroup path
+			// example record in /proc/{pid}/cgroup:
+			// 9:cpuset:/kubepods-besteffort-pod30556cce_0f92_11eb_b36d_02623cf324c8.slice:cri-containerd:c688c856b21cfb29c1dbf6c14793435e44a1299dfc12add33283239bffed2620
+			if strings.Contains(containerID, "cri-containerd") {
+				strList := strings.Split(containerID, ":")
+				containerID = strList[len(strList)-1]
+				containerID = strings.TrimPrefix(containerID, "cri-containerd-")
+			}
+			return containerID
+		}
+	}
+	return ""
+}
diff --git a/workflow/validate/validate.go b/workflow/validate/validate.go
index 614a1280f..f304dc329 100644
--- a/workflow/validate/validate.go
+++ b/workflow/validate/validate.go
@@ -140,6 +140,7 @@ func ValidateWorkflow(wftmplGetter templateresolution.WorkflowTemplateNamespaced
 
 	if wf.Spec.WorkflowTemplateRef != nil {
 		wfArgs.Parameters = util.MergeParameters(wfArgs.Parameters, wfSpecHolder.GetWorkflowSpec().Arguments.Parameters)
+		wfArgs.Artifacts = util.MergeArtifacts(wfArgs.Artifacts, wfSpecHolder.GetWorkflowSpec().Arguments.Artifacts)
 	}
 	if err != nil {
 		return nil, errors.Errorf(errors.CodeBadRequest, "spec.templates%s", err.Error())


Test patch:
diff --git a/workflow/controller/operator_workflow_template_ref_test.go b/workflow/controller/operator_workflow_template_ref_test.go
index f7d58c6d8..29a2fcf7e 100644
--- a/workflow/controller/operator_workflow_template_ref_test.go
+++ b/workflow/controller/operator_workflow_template_ref_test.go
@@ -277,3 +277,76 @@ func TestWorkflowTemplateRefParamMerge(t *testing.T) {
 	})
 
 }
+
+var wftWithArtifact = `
+apiVersion: argoproj.io/v1alpha1
+kind: WorkflowTemplate
+metadata:
+  name: artifact-test-1
+  namespace: test-namespace
+spec:
+  entrypoint: main
+  arguments:
+    artifacts:
+    - name: binary-file
+      http:
+        url: https://a.server.io/file
+    - name: data-file
+      http:
+        url: https://b.server.io/data
+
+  templates:
+    - name: main
+      steps:
+        - - name: process-data
+            template: process
+
+    - name: process
+      inputs:
+        artifacts:
+          - name: binary-file
+            path: /usr/local/bin/binfile
+            mode: 0755
+          - name: data-file
+            path: /tmp/data
+            mode: 0755
+      container:
+        image: busybox
+        command: [sh, -c]
+        args: ["binary-file /tmp/data"]
+`
+
+const wfWithTemplateWithArtifact = `
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  generateName: workflow-from-artifact-test-1-
+  namespace: test-namespace
+spec:
+  arguments:
+    artifacts:
+    - name: own-file
+      http:
+        url: https://local/blob
+  workflowTemplateRef:
+    name: artifact-test-1
+`
+
+func TestWorkflowTemplateRefGetArtifactsFromTemplate(t *testing.T) {
+	wf := unmarshalWF(wfWithTemplateWithArtifact)
+	wftmpl := unmarshalWFTmpl(wftWithArtifact)
+
+	t.Run("CheckArtifactArgumentFromWF", func(t *testing.T) {
+		cancel, controller := newController(wf, wftmpl)
+		defer cancel()
+		woc := newWorkflowOperationCtx(wf, controller)
+		_, execArgs, err := woc.loadExecutionSpec()
+		assert.NoError(t, err)
+		assert.Equal(t, wf.Spec.Arguments.Artifacts, woc.wf.Spec.Arguments.Artifacts)
+		assert.Len(t, execArgs.Artifacts, 3)
+
+		assert.Equal(t, "own-file", execArgs.Artifacts[0].Name)
+		assert.Equal(t, "binary-file", execArgs.Artifacts[1].Name)
+		assert.Equal(t, "data-file", execArgs.Artifacts[2].Name)
+	})
+}
diff --git a/workflow/executor/pns/pns_test.go b/workflow/executor/pns/pns_test.go
new file mode 100644
index 000000000..4c9aa37e1
--- /dev/null
+++ b/workflow/executor/pns/pns_test.go
@@ -0,0 +1,40 @@
+package pns
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestPNSExecutor_parseContainerIDFromCgroupLine(t *testing.T) {
+	testCases := []struct {
+		line     string
+		expected string
+	}{
+		{
+			line:     "",
+			expected: "",
+		},
+		{
+			line:     "5:rdma:/",
+			expected: "",
+		},
+		{
+			line:     "8:cpu,cpuacct:/kubepods/besteffort/pod2fad8aad-dcd0-4fef-b45a-151630b9a4b5/b844ef90162af07ad3fb2961ffb2f528f8bf7c9edb8c45465fd8651fab8a78c1",
+			expected: "b844ef90162af07ad3fb2961ffb2f528f8bf7c9edb8c45465fd8651fab8a78c1",
+		},
+		{
+			line:     "2:cpu,cpuacct:/system.slice/containerd.service/kubepods-burstable-podf61fae9b_d7a7_11ea_bb0c_12a065621d2b.slice:cri-containerd:b6b894a028b7ec8e0f98c57a5f7b9735ad792276c3ce52bc795fcd367d829de9",
+			expected: "b6b894a028b7ec8e0f98c57a5f7b9735ad792276c3ce52bc795fcd367d829de9",
+		},
+		{
+			line:     "8:cpu,cpuacct:/kubepods/besteffort/pod2fad8aad-dcd0-4fef-b45a-151630b9a4b5/crio-7a92a067289f6197148912be1c15f20f0330c7f3c541473d3b9c4043ca137b42.scope",
+			expected: "7a92a067289f6197148912be1c15f20f0330c7f3c541473d3b9c4043ca137b42",
+		},
+	}
+
+	for _, testCase := range testCases {
+		containerID := parseContainerIDFromCgroupLine(testCase.line)
+		assert.Equal(t, containerID, testCase.expected)
+	}
+}
diff --git a/workflow/validate/validate_test.go b/workflow/validate/validate_test.go
index 93f98ef92..c963ac2c3 100644
--- a/workflow/validate/validate_test.go
+++ b/workflow/validate/validate_test.go
@@ -2645,3 +2645,68 @@ func TestWorkflowTemplateLabels(t *testing.T) {
 	err := validateWorkflowTemplate(testWorkflowTemplateLabels)
 	assert.NoError(t, err)
 }
+
+const templateRefWithArtifactArgument = `
+apiVersion: argoproj.io/v1alpha1
+kind: WorkflowTemplate
+metadata:
+  name: template-ref-with-artifact
+spec:
+  entrypoint: A
+  arguments:
+    artifacts:
+    - name: binary-file
+      http:
+        url: https://a.server.io/file
+  templates:
+  - name: A
+    inputs:
+      artifacts:
+      - name: binary-file
+        path: /usr/local/bin/binfile
+        mode: 0755
+    container:
+      image: alpine:latest
+      command: [echo, hello]
+`
+
+const wfWithWFTRefAndNoOwnArtifact = `
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  generateName: hello-world-
+  namespace: default
+spec:
+  workflowTemplateRef:
+    name: template-ref-with-artifact
+`
+
+func TestWorkflowWithWFTRefWithOutOwnArtifactArgument(t *testing.T) {
+	err := createWorkflowTemplate(templateRefWithArtifactArgument)
+	assert.NoError(t, err)
+	_, err = validate(wfWithWFTRefAndNoOwnArtifact)
+	assert.NoError(t, err)
+}
+
+const wfWithWFTRefAndOwnArtifactArgument = `
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  generateName: hello-world-
+  namespace: default
+spec:
+  arguments:
+    artifacts:
+    - name: binary-file
+      http:
+        url: http://localserver/file
+  workflowTemplateRef:
+    name: template-ref-with-artifact
+`
+
+func TestWorkflowWithWFTRefWithArtifactArgument(t *testing.T) {
+	err := createWorkflowTemplate(templateRefWithArtifactArgument)
+	assert.NoError(t, err)
+	_, err = validate(wfWithWFTRefAndOwnArtifactArgument)
+	assert.NoError(t, err)
+}