Repository: argoproj/argo (go)

Issue:
Client max gRPC size does not match server
**What happened**:

Executing a client request that is over the default gRPC size limit results in an error.

**What you expected to happen**:

Client gRPC size limit to match server (100mb)

**How to reproduce it (as minimally and precisely as possible)**:

If you have enough data in the server, then a 'argo list' will result in a message like:
```
FATA[0006] rpc error: code = ResourceExhausted desc = grpc: received message larger than max (40608696 vs. 4194304)
```

**Anything else we need to know?**:


---
<!-- Issue Author: Don't delete this message to ecourage other users to support your issue! -->
**Message from the maintainers**:

If you are impacted by this bug please add a 👍 reaction to this issue! We often sort issues this way to know what to prioritize.


Pull request description:
fix: Increase client gRPC max size to match server. Fixes #2513
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"`.  
* [ ] I have written unit and/or e2e tests for my change. PRs without these are unlike to be merged.
* [X] Optional. I've added My organization is added to the USERS.md.
* [X] I've signed the CLA and required builds are green. 


Patch:
diff --git a/api/openapi-spec/swagger.json b/api/openapi-spec/swagger.json
index 9ab690ffa..f7df4162e 100644
--- a/api/openapi-spec/swagger.json
+++ b/api/openapi-spec/swagger.json
@@ -2954,7 +2954,7 @@
           "format": "int32"
         },
         "ttlStrategy": {
-          "title": "TTLStrategy limits the lifetime of a Workflow that has finished execution depending on if it\nSucceeded or Failed. If this struct is set, once the Workflow finishes, it will be\ndeleted after the time to live expires. If this field is unset,\nthe controller config map will hold the default values\nUpdate",
+          "description": "TTLStrategy limits the lifetime of a Workflow that has finished execution depending on if it\nSucceeded or Failed. If this struct is set, once the Workflow finishes, it will be\ndeleted after the time to live expires. If this field is unset,\nthe controller config map will hold the default values.",
           "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.TTLStrategy"
         },
         "volumeClaimTemplates": {
diff --git a/config/config.go b/config/config.go
index c91763bb7..0f33f2906 100644
--- a/config/config.go
+++ b/config/config.go
@@ -51,11 +51,13 @@ type Config struct {
 	// controller watches workflows and pods that *are not* labeled with an instance id.
 	InstanceID string `json:"instanceID,omitempty"`
 
+	// MetricsConfig specifies configuration for metrics emission
 	MetricsConfig PrometheusConfig `json:"metricsConfig,omitempty"`
 
 	// FeatureFlags for general/experimental features
 	FeatureFlags FeatureFlags `json:"featureFlags,omitempty"`
 
+	// TelemetryConfig specifies configuration for telemetry emission
 	TelemetryConfig PrometheusConfig `json:"telemetryConfig,omitempty"`
 
 	// Parallelism limits the max total parallel workflows that can execute at the same time
@@ -70,8 +72,8 @@ type Config struct {
 	// Config customized Docker Sock path
 	DockerSockPath string `json:"dockerSockPath,omitempty"`
 
-	// Default workflow spec, will be adde to workflow if the parameters are not set in the workflow
-	DefautWorkflowSpec *wfv1.WorkflowSpec `json:"workflowDefaults,omitempty"`
+	// 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 PodSpecLogStrategy `json:"podSpecLogStrategy,omitempty"`
diff --git a/docs/default-workflow-specs.md b/docs/default-workflow-specs.md
index 742661d73..0dce93265 100644
--- a/docs/default-workflow-specs.md
+++ b/docs/default-workflow-specs.md
@@ -4,34 +4,54 @@
 
 > v2.7 and after
 
-It's possible to set default workflow specs which will be written to all workflows if the spec of interest is not set. This can be configurated through the 
-workflow controller config [map](../workflow/config/config.go#L11) and the field [DefaultWorkflowSpec](../workflow/config/config.go#L69). 
+## Introduction
 
+Default Workflow spec values can be set at the controller config map that will apply to all Workflows executed from said controller.
+If a Workflow has a value that also has a default value set in the config map, thw Workflow's value will take precedence.
 
-In order to edit the Default workflow spec for a controller, edit the workflow config map: 
+## Setting Default Workflow Values
 
+Default Workflow values can be specified by adding them under the `workflowDefaults` key in the [`workflow-controller-configmap`](./workflow-controller-configmap.yaml).
+Values can be added as the would under the `Workflow.spec` tag.
 
-```bash 
-kubectl edit cm/workflow-controller-configmap
-```
-
-
-As an example the time for a argo workflow to live after finish can be set, in the spec this field is known as ```secondsAfterCompletion``` in the ```ttlStrategy```. Example of how the config map could look with this filed can be found [here](./workflow-controller-configmap.yaml).
-
-In order to test it a example workflow can be submited, in this case the [coinflip example](../examples/coinflip.yaml), the following can be run:
+For example, to specify default values that would partially produce the following `Workflow`:
 
-```bash 
-argo submit ./examples/coinflip.yaml
+```yaml
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  generateName: gc-ttl-
+  annotations:
+    argo: workflows
+  labels:
+    foo: bar
+spec:
+  ttlStrategy:
+    secondsAfterSuccess: 5     # Time to live after workflow is successful
+  parallelism: 3
 ```
 
-to verify that the the defaultd are set run 
-
-```bash
-argo get [YOUR_ARGO_WORKFLOW_NAME]
-```
+The following would be specified in the Config Map:
 
-You should then see the field, Ttl Strategy populated
 ```yaml
-Ttl Strategy:
-  Seconds After Completion:  10
-```
+# This file describes the config settings available in the workflow controller configmap
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: workflow-controller-configmap
+data:
+  config: |
+
+    # Default values that will apply to all Workflows from this controller, unless overridden on the Workflow-level
+    workflowDefaults:
+      metadata:
+        annotations:
+          argo: workflows
+        labels:
+          foo: bar
+      spec:
+        ttlStrategy:
+          secondsAfterSuccess: 5
+        parallelism: 3
+
+```
\ No newline at end of file
diff --git a/docs/workflow-controller-configmap.yaml b/docs/workflow-controller-configmap.yaml
index e85e0cdb4..7c4a93945 100644
--- a/docs/workflow-controller-configmap.yaml
+++ b/docs/workflow-controller-configmap.yaml
@@ -171,7 +171,15 @@ data:
       #     name: argo-mysql-config
       #     key: password
 
-    # default workflow spec, follow the workflow spec, more info: https://github.com/argoproj/argo/blob/master/examples/README.md#the-structure-of-workflow-specs
+    # Default values that will apply to all Workflows from this controller, unless overridden on the Workflow-level
+    # See more: docs/default-workflow-specs.yaml
     workflowDefaults:
-      ttlStrategy:
-        secondsAfterCompletion: 10
+      metadata:
+        annotations:
+          argo: workflows
+        labels:
+          foo: bar
+      spec:
+        ttlStrategy:
+          secondsAfterSuccess: 5
+        parallelism: 3
diff --git a/go.mod b/go.mod
index 417684699..ddde30372 100644
--- a/go.mod
+++ b/go.mod
@@ -76,7 +76,7 @@ require (
 	golang.org/x/crypto v0.0.0-20200128174031-69ecbb4d6d5d
 	golang.org/x/net v0.0.0-20200301022130-244492dfa37a
 	golang.org/x/oauth2 v0.0.0-20200107190931-bf48bf16ab8d
-	golang.org/x/tools v0.0.0-20200323210725-ef1313dc6d0a // indirect
+	golang.org/x/tools v0.0.0-20200325010219-a49f79bcc224 // indirect
 	google.golang.org/api v0.20.0
 	google.golang.org/genproto v0.0.0-20200317114155-1f3552e48f24
 	google.golang.org/grpc v1.28.0
diff --git a/go.sum b/go.sum
index 7642b5fce..909f588f7 100644
--- a/go.sum
+++ b/go.sum
@@ -607,8 +607,8 @@ golang.org/x/tools v0.0.0-20200207183749-b753a1ba74fa/go.mod h1:TB2adYChydJhpapK
 golang.org/x/tools v0.0.0-20200212150539-ea181f53ac56/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
 golang.org/x/tools v0.0.0-20200224181240-023911ca70b2/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
 golang.org/x/tools v0.0.0-20200317043434-63da46f3035e/go.mod h1:Sl4aGygMT6LrqrWclx+PTx3U+LnKx/seiNR+3G19Ar8=
-golang.org/x/tools v0.0.0-20200323210725-ef1313dc6d0a h1:QpsrlnR31DN2tCItRzamUXN4oAe3U00ru8f0ik9HSYI=
-golang.org/x/tools v0.0.0-20200323210725-ef1313dc6d0a/go.mod h1:Sl4aGygMT6LrqrWclx+PTx3U+LnKx/seiNR+3G19Ar8=
+golang.org/x/tools v0.0.0-20200325010219-a49f79bcc224 h1:azwY/v0y0K4mFHVsg5+UrTgchqALYWpqVo6vL5OmkmI=
+golang.org/x/tools v0.0.0-20200325010219-a49f79bcc224/go.mod h1:Sl4aGygMT6LrqrWclx+PTx3U+LnKx/seiNR+3G19Ar8=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898 h1:/atklqdjdhuosWIl6AIbOeHJjicWYPqR9bpxqxYG2pA=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
diff --git a/pkg/apiclient/argo-server-client.go b/pkg/apiclient/argo-server-client.go
index f1c6633d8..31c95dce5 100644
--- a/pkg/apiclient/argo-server-client.go
+++ b/pkg/apiclient/argo-server-client.go
@@ -12,6 +12,11 @@ import (
 	workflowtemplatepkg "github.com/argoproj/argo/pkg/apiclient/workflowtemplate"
 )
 
+const (
+	// MaxGRPCMessageSize contains max grpc message size supported by the client
+	MaxClientGRPCMessageSize = 100 * 1024 * 1024
+)
+
 type argoServerClient struct {
 	*grpc.ClientConn
 }
@@ -41,7 +46,7 @@ func (a *argoServerClient) NewArchivedWorkflowServiceClient() (workflowarchivepk
 }
 
 func NewClientConn(argoServer string) (*grpc.ClientConn, error) {
-	conn, err := grpc.Dial(argoServer, grpc.WithInsecure())
+	conn, err := grpc.Dial(argoServer, grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(MaxClientGRPCMessageSize)), grpc.WithInsecure())
 	if err != nil {
 		return nil, err
 	}
diff --git a/pkg/apiclient/cronworkflow/cron-workflow.swagger.json b/pkg/apiclient/cronworkflow/cron-workflow.swagger.json
index e49ffd901..649298fb9 100644
--- a/pkg/apiclient/cronworkflow/cron-workflow.swagger.json
+++ b/pkg/apiclient/cronworkflow/cron-workflow.swagger.json
@@ -1699,7 +1699,7 @@
         },
         "ttlStrategy": {
           "$ref": "#/definitions/github.com.argoproj.argo.pkg.apis.workflow.v1alpha1.TTLStrategy",
-          "title": "TTLStrategy limits the lifetime of a Workflow that has finished execution depending on if it\nSucceeded or Failed. If this struct is set, once the Workflow finishes, it will be\ndeleted after the time to live expires. If this field is unset,\nthe controller config map will hold the default values\nUpdate"
+          "description": "TTLStrategy limits the lifetime of a Workflow that has finished execution depending on if it\nSucceeded or Failed. If this struct is set, once the Workflow finishes, it will be\ndeleted after the time to live expires. If this field is unset,\nthe controller config map will hold the default values."
         },
         "activeDeadlineSeconds": {
           "type": "string",
diff --git a/pkg/apiclient/workflow/workflow.swagger.json b/pkg/apiclient/workflow/workflow.swagger.json
index f57bc51de..5e01e27e1 100644
--- a/pkg/apiclient/workflow/workflow.swagger.json
+++ b/pkg/apiclient/workflow/workflow.swagger.json
@@ -2106,7 +2106,7 @@
         },
         "ttlStrategy": {
           "$ref": "#/definitions/github.com.argoproj.argo.pkg.apis.workflow.v1alpha1.TTLStrategy",
-          "title": "TTLStrategy limits the lifetime of a Workflow that has finished execution depending on if it\nSucceeded or Failed. If this struct is set, once the Workflow finishes, it will be\ndeleted after the time to live expires. If this field is unset,\nthe controller config map will hold the default values\nUpdate"
+          "description": "TTLStrategy limits the lifetime of a Workflow that has finished execution depending on if it\nSucceeded or Failed. If this struct is set, once the Workflow finishes, it will be\ndeleted after the time to live expires. If this field is unset,\nthe controller config map will hold the default values."
         },
         "activeDeadlineSeconds": {
           "type": "string",
diff --git a/pkg/apiclient/workflowarchive/workflow-archive.swagger.json b/pkg/apiclient/workflowarchive/workflow-archive.swagger.json
index c0a75a3f0..8024c9686 100644
--- a/pkg/apiclient/workflowarchive/workflow-archive.swagger.json
+++ b/pkg/apiclient/workflowarchive/workflow-archive.swagger.json
@@ -1537,7 +1537,7 @@
         },
         "ttlStrategy": {
           "$ref": "#/definitions/github.com.argoproj.argo.pkg.apis.workflow.v1alpha1.TTLStrategy",
-          "title": "TTLStrategy limits the lifetime of a Workflow that has finished execution depending on if it\nSucceeded or Failed. If this struct is set, once the Workflow finishes, it will be\ndeleted after the time to live expires. If this field is unset,\nthe controller config map will hold the default values\nUpdate"
+          "description": "TTLStrategy limits the lifetime of a Workflow that has finished execution depending on if it\nSucceeded or Failed. If this struct is set, once the Workflow finishes, it will be\ndeleted after the time to live expires. If this field is unset,\nthe controller config map will hold the default values."
         },
         "activeDeadlineSeconds": {
           "type": "string",
diff --git a/pkg/apiclient/workflowtemplate/workflow-template.swagger.json b/pkg/apiclient/workflowtemplate/workflow-template.swagger.json
index 8a61ccf2c..186d0295b 100644
--- a/pkg/apiclient/workflowtemplate/workflow-template.swagger.json
+++ b/pkg/apiclient/workflowtemplate/workflow-template.swagger.json
@@ -1567,7 +1567,7 @@
         },
         "ttlStrategy": {
           "$ref": "#/definitions/github.com.argoproj.argo.pkg.apis.workflow.v1alpha1.TTLStrategy",
-          "title": "TTLStrategy limits the lifetime of a Workflow that has finished execution depending on if it\nSucceeded or Failed. If this struct is set, once the Workflow finishes, it will be\ndeleted after the time to live expires. If this field is unset,\nthe controller config map will hold the default values\nUpdate"
+          "description": "TTLStrategy limits the lifetime of a Workflow that has finished execution depending on if it\nSucceeded or Failed. If this struct is set, once the Workflow finishes, it will be\ndeleted after the time to live expires. If this field is unset,\nthe controller config map will hold the default values."
         },
         "activeDeadlineSeconds": {
           "type": "string",
diff --git a/pkg/apis/workflow/v1alpha1/generated.proto b/pkg/apis/workflow/v1alpha1/generated.proto
index 85a3aaaba..b639a1ce0 100644
--- a/pkg/apis/workflow/v1alpha1/generated.proto
+++ b/pkg/apis/workflow/v1alpha1/generated.proto
@@ -1077,8 +1077,7 @@ message WorkflowSpec {
   // TTLStrategy limits the lifetime of a Workflow that has finished execution depending on if it
   // Succeeded or Failed. If this struct is set, once the Workflow finishes, it will be
   // deleted after the time to live expires. If this field is unset,
-  // the controller config map will hold the default values
-  // Update
+  // the controller config map will hold the default values.
   optional TTLStrategy ttlStrategy = 30;
 
   // Optional duration in seconds relative to the workflow start time which the workflow is
diff --git a/pkg/apis/workflow/v1alpha1/workflow_types.go b/pkg/apis/workflow/v1alpha1/workflow_types.go
index 153423080..0a40895e0 100644
--- a/pkg/apis/workflow/v1alpha1/workflow_types.go
+++ b/pkg/apis/workflow/v1alpha1/workflow_types.go
@@ -239,8 +239,7 @@ type WorkflowSpec struct {
 	// TTLStrategy limits the lifetime of a Workflow that has finished execution depending on if it
 	// Succeeded or Failed. If this struct is set, once the Workflow finishes, it will be
 	// deleted after the time to live expires. If this field is unset,
-	// the controller config map will hold the default values
-	// Update
+	// the controller config map will hold the default values.
 	TTLStrategy *TTLStrategy `json:"ttlStrategy,omitempty" protobuf:"bytes,30,opt,name=ttlStrategy"`
 
 	// Optional duration in seconds relative to the workflow start time which the workflow is
diff --git a/workflow/controller/controller.go b/workflow/controller/controller.go
index 264ac3cc7..eb94f6a80 100644
--- a/workflow/controller/controller.go
+++ b/workflow/controller/controller.go
@@ -366,13 +366,15 @@ func (wfc *WorkflowController) processNextItem() bool {
 		wfc.throttler.Remove(key)
 		return true
 	}
-	err = wfc.addingWorkflowDefaultValueIfValueNotExist(wf)
+
+	err = wfc.setWorkflowDefaults(wf)
 	if err != nil {
-		log.Warnf("Failed to unmarshal key '%s' to workflow object: %v", key, err)
+		log.Warnf("Failed to apply default workflow values to '%s': %v", wf.Name, err)
 		woc := newWorkflowOperationCtx(wf, wfc)
 		woc.markWorkflowFailed(fmt.Sprintf("invalid spec: %s", err.Error()))
 		woc.persistUpdates()
 		wfc.throttler.Remove(key)
+		return true
 	}
 
 	if wf.ObjectMeta.Labels[common.LabelKeyCompleted] == "true" {
@@ -381,7 +383,9 @@ func (wfc *WorkflowController) processNextItem() bool {
 		// but we are still draining the controller's workflow workqueue
 		return true
 	}
+
 	woc := newWorkflowOperationCtx(wf, wfc)
+
 	// Loading running workflow from persistence storage if nodeStatusOffload enabled
 	if wf.Status.IsOffloadNodeStatus() {
 		nodes, err := wfc.offloadNodeStatusRepo.Get(string(wf.UID), wf.GetOffloadNodeStatusVersion())
@@ -434,32 +438,6 @@ func (wfc *WorkflowController) processNextItem() bool {
 	return true
 }
 
-// addingWorkflowDefaultValueIfValueNotExist sets values in the workflow.Spec with defaults from the
-// workflowController. Values in the workflow will be given the upper hand over the defaults.
-// The defaults for the workflow controller is set in the WorkflowController.Config.DefautWorkflowSpec
-func (wfc *WorkflowController) addingWorkflowDefaultValueIfValueNotExist(wf *wfv1.Workflow) error {
-	if wfc.Config.DefautWorkflowSpec != nil {
-		defaultsSpec, err := json.Marshal(*wfc.Config.DefautWorkflowSpec)
-		if err != nil {
-			return err
-		}
-		workflowSpec, err := json.Marshal(wf.Spec)
-		if err != nil {
-			return err
-		}
-		// https: //godoc.org/k8s.io/apimachinery/pkg/util/strategicpatch#StrategicMergePatch
-		new, err := strategicpatch.StrategicMergePatch(defaultsSpec, workflowSpec, wfv1.WorkflowSpec{})
-		if err != nil {
-			return err
-		}
-		err = json.Unmarshal(new, &wf.Spec)
-		if err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
 func (wfc *WorkflowController) podWorker() {
 	for wfc.processNextPodItem() {
 	}
@@ -642,6 +620,31 @@ func (wfc *WorkflowController) newPodInformer() cache.SharedIndexInformer {
 	return informer
 }
 
+// setWorkflowDefaults sets values in the workflow.Spec with defaults from the
+// workflowController. Values in the workflow will be given the upper hand over the defaults.
+// The defaults for the workflow controller are set in the workflow-controller config map
+func (wfc *WorkflowController) setWorkflowDefaults(wf *wfv1.Workflow) error {
+	if wfc.Config.WorkflowDefaults != nil {
+		defaultsSpec, err := json.Marshal(*wfc.Config.WorkflowDefaults)
+		if err != nil {
+			return err
+		}
+		workflowBytes, err := json.Marshal(wf)
+		if err != nil {
+			return err
+		}
+		mergedWf, err := strategicpatch.StrategicMergePatch(defaultsSpec, workflowBytes, wfv1.Workflow{})
+		if err != nil {
+			return err
+		}
+		err = json.Unmarshal(mergedWf, &wf)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 func (wfc *WorkflowController) newWorkflowTemplateInformer() wfextvv1alpha1.WorkflowTemplateInformer {
 	return wfextv.NewSharedInformerFactoryWithOptions(wfc.wfclientset, workflowTemplateResyncPeriod, wfextv.WithNamespace(wfc.GetManagedNamespace())).Argoproj().V1alpha1().WorkflowTemplates()
 }
diff --git a/workflow/controller/workflowpod.go b/workflow/controller/workflowpod.go
index 0c69e789f..abba7e226 100644
--- a/workflow/controller/workflowpod.go
+++ b/workflow/controller/workflowpod.go
@@ -484,6 +484,9 @@ func (woc *wfOperationCtx) newExecContainer(name string, tmpl *wfv1.Template) *a
 	}
 	if woc.controller.Config.Executor != nil {
 		exec.Args = woc.controller.Config.Executor.Args
+		if woc.controller.Config.Executor.SecurityContext != nil {
+			exec.SecurityContext = woc.controller.Config.Executor.SecurityContext.DeepCopy()
+		}
 	}
 	if isResourcesSpecified(woc.controller.Config.Executor) {
 		exec.Resources = woc.controller.Config.Executor.Resources
diff --git a/workflow/metrics/controller_metrics.go b/workflow/metrics/controller_metrics.go
index 49437248d..3e3bab741 100644
--- a/workflow/metrics/controller_metrics.go
+++ b/workflow/metrics/controller_metrics.go
@@ -58,7 +58,9 @@ func (wc *controllerCollector) collectWorkflowStatuses(wfs []*wfv1.Workflow) []p
 	}
 
 	for _, wf := range wfs {
-		gauges[wf.Status.Phase].Inc()
+		if _, ok := gauges[wf.Status.Phase]; ok {
+			gauges[wf.Status.Phase].Inc()
+		}
 	}
 
 	var out []prometheus.Metric
diff --git a/workflow/validate/validate.go b/workflow/validate/validate.go
index 1c85cf5e9..16fe68ccd 100644
--- a/workflow/validate/validate.go
+++ b/workflow/validate/validate.go
@@ -67,7 +67,9 @@ const (
 	// anyItemMagicValue is a magic value set in addItemsToScope() and checked in
 	// resolveAllVariables() to determine if any {{item.name}} can be accepted during
 	// variable resolution (to support withParam)
-	anyItemMagicValue = "item.*"
+	anyItemMagicValue                    = "item.*"
+	anyWorkflowOutputParameterMagicValue = "workflow.outputs.parameters.*"
+	anyWorkflowOutputArtifactMagicValue  = "workflow.outputs.artifacts.*"
 )
 
 var (
@@ -488,6 +490,8 @@ func validateArtifactLocation(errPrefix string, art wfv1.ArtifactLocation) error
 func resolveAllVariables(scope map[string]interface{}, tmplStr string) error {
 	var unresolvedErr error
 	_, allowAllItemRefs := scope[anyItemMagicValue] // 'item.*' is a magic placeholder value set by addItemsToScope
+	_, allowAllWorkflowOutputParameterRefs := scope[anyWorkflowOutputParameterMagicValue]
+	_, allowAllWorkflowOutputArtifactRefs := scope[anyWorkflowOutputArtifactMagicValue]
 	fstTmpl := fasttemplate.New(tmplStr, "{{", "}}")
 
 	fstTmpl.ExecuteFuncString(func(w io.Writer, tag string) (int, error) {
@@ -501,6 +505,10 @@ func resolveAllVariables(scope map[string]interface{}, tmplStr string) error {
 			if (tag == "item" || strings.HasPrefix(tag, "item.")) && allowAllItemRefs {
 				// we are *probably* referencing a undetermined item using withParam
 				// NOTE: this is far from foolproof.
+			} else if strings.HasPrefix(tag, "workflow.outputs.parameters.") && allowAllWorkflowOutputParameterRefs {
+				// Allow runtime resolution of workflow output parameter names
+			} else if strings.HasPrefix(tag, "workflow.outputs.artifacts.") && allowAllWorkflowOutputArtifactRefs {
+				// Allow runtime resolution of workflow output artifact names
 			} else if strings.HasPrefix(tag, "outputs.") {
 				// We are self referencing for metric emission, allow it.
 			} else if strings.HasPrefix(tag, common.GlobalVarWorkflowCreationTimestamp) {
@@ -747,18 +755,28 @@ func (ctx *templateValidationCtx) addOutputsToScope(tmpl *wfv1.Template, prefix
 	}
 	for _, param := range tmpl.Outputs.Parameters {
 		scope[fmt.Sprintf("%s.outputs.parameters.%s", prefix, param.Name)] = true
-		if param.GlobalName != "" && !isParameter(param.GlobalName) {
-			globalParamName := fmt.Sprintf("workflow.outputs.parameters.%s", param.GlobalName)
-			scope[globalParamName] = true
-			ctx.globalParams[globalParamName] = placeholderGenerator.NextPlaceholder()
+		if param.GlobalName != "" {
+			if !isParameter(param.GlobalName) {
+				globalParamName := fmt.Sprintf("workflow.outputs.parameters.%s", param.GlobalName)
+				scope[globalParamName] = true
+				ctx.globalParams[globalParamName] = placeholderGenerator.NextPlaceholder()
+			} else {
+				logrus.Warnf("GlobalName '%s' is a parameter and won't be validated until runtime", param.GlobalName)
+				scope[anyWorkflowOutputParameterMagicValue] = true
+			}
 		}
 	}
 	for _, art := range tmpl.Outputs.Artifacts {
 		scope[fmt.Sprintf("%s.outputs.artifacts.%s", prefix, art.Name)] = true
-		if art.GlobalName != "" && !isParameter(art.GlobalName) {
-			globalArtName := fmt.Sprintf("workflow.outputs.artifacts.%s", art.GlobalName)
-			scope[globalArtName] = true
-			ctx.globalParams[globalArtName] = placeholderGenerator.NextPlaceholder()
+		if art.GlobalName != "" {
+			if !isParameter(art.GlobalName) {
+				globalArtName := fmt.Sprintf("workflow.outputs.artifacts.%s", art.GlobalName)
+				scope[globalArtName] = true
+				ctx.globalParams[globalArtName] = placeholderGenerator.NextPlaceholder()
+			} else {
+				logrus.Warnf("GlobalName '%s' is a parameter and won't be validated until runtime", art.GlobalName)
+				scope[anyWorkflowOutputArtifactMagicValue] = true
+			}
 		}
 	}
 	if aggregate {


Test patch:
diff --git a/workflow/controller/controller_test.go b/workflow/controller/controller_test.go
index 99588870f..18fd4290d 100644
--- a/workflow/controller/controller_test.go
+++ b/workflow/controller/controller_test.go
@@ -50,6 +50,8 @@ apiVersion: argoproj.io/v1alpha1
 kind: Workflow
 metadata:
   name: hello-world
+  labels:
+    foo: bar
 spec:
   entrypoint: whalesay
   serviceAccountName: whalesay
@@ -130,8 +132,10 @@ func newControllerWithDefaults() *WorkflowController {
 	return &WorkflowController{
 		Config: config.Config{
 			ExecutorImage: "executor:latest",
-			DefautWorkflowSpec: &wfv1.WorkflowSpec{
-				HostNetwork: &myBool,
+			WorkflowDefaults: &wfv1.Workflow{
+				Spec: wfv1.WorkflowSpec{
+					HostNetwork: &myBool,
+				},
 			},
 		},
 		kubeclientset:  fake.NewSimpleClientset(),
@@ -158,16 +162,26 @@ func newControllerWithComplexDefaults() *WorkflowController {
 	return &WorkflowController{
 		Config: config.Config{
 			ExecutorImage: "executor:latest",
-			DefautWorkflowSpec: &wfv1.WorkflowSpec{
-				HostNetwork:        &myBool,
-				Entrypoint:         "good_entrypoint",
-				ServiceAccountName: "my_service_account",
-				TTLStrategy: &wfv1.TTLStrategy{
-					SecondsAfterCompletion: &ten,
-					SecondsAfterSuccess:    &ten,
-					SecondsAfterFailure:    &ten,
+			WorkflowDefaults: &wfv1.Workflow{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{
+						"annotation": "value",
+					},
+					Labels: map[string]string{
+						"label": "value",
+					},
+				},
+				Spec: wfv1.WorkflowSpec{
+					HostNetwork:        &myBool,
+					Entrypoint:         "good_entrypoint",
+					ServiceAccountName: "my_service_account",
+					TTLStrategy: &wfv1.TTLStrategy{
+						SecondsAfterCompletion: &ten,
+						SecondsAfterSuccess:    &ten,
+						SecondsAfterFailure:    &ten,
+					},
+					TTLSecondsAfterFinished: &seven,
 				},
-				TTLSecondsAfterFinished: &seven,
 			},
 		},
 		kubeclientset:  fake.NewSimpleClientset(),
@@ -228,16 +242,15 @@ func makePodsPhaseAll(t *testing.T, phase apiv1.PodPhase, kubeclientset kubernet
 }
 
 func TestAddingWorkflowDefaultValueIfValueNotExist(t *testing.T) {
-	assert.Equal(t, "hello", "hello")
 	ans := true
 	controller := newController()
 	workflow := unmarshalWF(helloWorldWf)
-	err := controller.addingWorkflowDefaultValueIfValueNotExist(workflow)
+	err := controller.setWorkflowDefaults(workflow)
 	assert.NoError(t, err)
 	assert.Equal(t, workflow, unmarshalWF(helloWorldWf))
 	controllerDefaults := newControllerWithDefaults()
 	defautWorkflowSpec := unmarshalWF(helloWorldWf)
-	err = controllerDefaults.addingWorkflowDefaultValueIfValueNotExist(defautWorkflowSpec)
+	err = controllerDefaults.setWorkflowDefaults(defautWorkflowSpec)
 	assert.NoError(t, err)
 	assert.Equal(t, defautWorkflowSpec.Spec.HostNetwork, &ans)
 	assert.NotEqual(t, defautWorkflowSpec, unmarshalWF(helloWorldWf))
@@ -245,28 +258,30 @@ func TestAddingWorkflowDefaultValueIfValueNotExist(t *testing.T) {
 }
 
 func TestAddingWorkflowDefaultComplex(t *testing.T) {
-	assert.Equal(t, "hello", "hello")
 	controller := newControllerWithComplexDefaults()
 	workflow := unmarshalWF(testDefaultWf)
 	var ten int32 = 10
 	assert.Equal(t, workflow.Spec.Entrypoint, "whalesay")
 	assert.Nil(t, workflow.Spec.TTLStrategy)
-	err := controller.addingWorkflowDefaultValueIfValueNotExist(workflow)
+	assert.Contains(t, workflow.Labels, "foo")
+	err := controller.setWorkflowDefaults(workflow)
 	assert.NoError(t, err)
 	assert.NotEqual(t, workflow, unmarshalWF(testDefaultWf))
 	assert.Equal(t, workflow.Spec.Entrypoint, "whalesay")
 	assert.Equal(t, workflow.Spec.ServiceAccountName, "whalesay")
 	assert.Equal(t, *workflow.Spec.TTLStrategy.SecondsAfterFailure, ten)
+	assert.Contains(t, workflow.Labels, "foo")
+	assert.Contains(t, workflow.Labels, "label")
+	assert.Contains(t, workflow.Annotations, "annotation")
 }
 
 func TestAddingWorkflowDefaultComplexTwo(t *testing.T) {
-	assert.Equal(t, "hello", "hello")
 	controller := newControllerWithComplexDefaults()
 	workflow := unmarshalWF(testDefaultWfTTL)
 	var ten int32 = 10
 	var seven int32 = 7
 	var five int32 = 5
-	err := controller.addingWorkflowDefaultValueIfValueNotExist(workflow)
+	err := controller.setWorkflowDefaults(workflow)
 	assert.NoError(t, err)
 	assert.NotEqual(t, workflow, unmarshalWF(testDefaultWfTTL))
 	assert.Equal(t, workflow.Spec.Entrypoint, "whalesay")
@@ -274,4 +289,7 @@ func TestAddingWorkflowDefaultComplexTwo(t *testing.T) {
 	assert.Equal(t, *workflow.Spec.TTLStrategy.SecondsAfterCompletion, five)
 	assert.Equal(t, *workflow.Spec.TTLStrategy.SecondsAfterFailure, ten)
 	assert.Equal(t, *workflow.Spec.TTLSecondsAfterFinished, seven)
+	assert.NotContains(t, workflow.Labels, "foo")
+	assert.Contains(t, workflow.Labels, "label")
+	assert.Contains(t, workflow.Annotations, "annotation")
 }
diff --git a/workflow/controller/workflowpod_test.go b/workflow/controller/workflowpod_test.go
index c5a88e617..4b8613f84 100644
--- a/workflow/controller/workflowpod_test.go
+++ b/workflow/controller/workflowpod_test.go
@@ -313,6 +313,36 @@ func TestTmplLevelExecutorServiceAccountName(t *testing.T) {
 	verifyServiceAccountTokenVolumeMount(t, waitCtr, "exec-sa-token", "/var/run/secrets/kubernetes.io/serviceaccount")
 }
 
+// TestTmplLevelExecutorServiceAccountName verifies the ability to carry forward template level AutomountServiceAccountToken to Podspec.
+func TestTmplLevelExecutorSecurityContext(t *testing.T) {
+	var user int64 = 1000
+	woc := newWoc()
+	_, err := util.CreateServiceAccountWithToken(woc.controller.kubeclientset, "", "foo", "foo-token")
+	assert.NoError(t, err)
+	_, err = util.CreateServiceAccountWithToken(woc.controller.kubeclientset, "", "tmpl", "tmpl-token")
+	assert.NoError(t, err)
+
+	woc.controller.Config.Executor = &apiv1.Container{SecurityContext: &apiv1.SecurityContext{RunAsUser: &user}}
+	woc.wf.Spec.Templates[0].Executor = &wfv1.ExecutorConfig{ServiceAccountName: "tmpl"}
+	tmplCtx, err := woc.createTemplateContext("")
+	assert.NoError(t, err)
+	_, err = woc.executeContainer(woc.wf.Spec.Entrypoint, tmplCtx.GetCurrentTemplateBase().GetTemplateScope(), &woc.wf.Spec.Templates[0], &woc.wf.Spec.Templates[0], &executeTemplateOpts{})
+	assert.NoError(t, err)
+	pods, err := woc.controller.kubeclientset.CoreV1().Pods("").List(metav1.ListOptions{})
+	assert.NoError(t, err)
+	assert.Len(t, pods.Items, 1)
+	pod := pods.Items[0]
+
+	waitCtr := pod.Spec.Containers[0]
+	assert.NotNil(t, waitCtr.SecurityContext)
+	if waitCtr.SecurityContext != nil {
+		assert.NotNil(t, waitCtr.SecurityContext.RunAsUser)
+		if waitCtr.SecurityContext.RunAsUser != nil {
+			assert.Equal(t, int64(1000), *waitCtr.SecurityContext.RunAsUser)
+		}
+	}
+}
+
 // TestImagePullSecrets verifies the ability to carry forward imagePullSecrets from workflow.spec
 func TestImagePullSecrets(t *testing.T) {
 	woc := newWoc()
diff --git a/workflow/validate/validate_test.go b/workflow/validate/validate_test.go
index 4e2ae0ab6..12d25138a 100644
--- a/workflow/validate/validate_test.go
+++ b/workflow/validate/validate_test.go
@@ -2027,3 +2027,60 @@ func TestAllowPlaceholderInVariableTakenFromInputs(t *testing.T) {
 		assert.NoError(t, err)
 	}
 }
+
+var runtimeResolutionOfVariableNames = `
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  generateName: global-parameter-passing-
+spec:
+  entrypoint: plan
+  templates:
+  - name: plan
+    steps:
+    - - name: pass-parameter
+        template: global-parameter-passing
+        arguments:
+          parameters:
+          - name: global-parameter-name
+            value: key
+          - name: global-parameter-value
+            value: value
+    - - name: print-parameter
+        template: parameter-printing
+        arguments:
+          parameters:
+          - name: parameter
+            value: "{{workflow.outputs.parameters.key}}"
+
+  - name: global-parameter-passing
+    inputs:
+      parameters:
+      - name: global-parameter-name
+      - name: global-parameter-value
+    container:
+      image: alpine:3.11
+      command: [sh, -c]
+      args: ["exit 0"]
+    outputs:
+      parameters:
+      - name: global-parameter
+        value: "{{inputs.parameters.global-parameter-value}}"
+        globalName: "{{inputs.parameters.global-parameter-name}}"
+
+  - name: parameter-printing
+    inputs:
+      parameters:
+      - name: parameter
+    container:
+      image: alpine:3.11
+      command: [sh, -c]
+      args: ["echo {{inputs.parameters.parameter}}"]
+`
+
+// TestInvalidResourceWorkflow verifies an error against a workflow of an invalid resource.
+func TestRuntimeResolutionOfVariableNames(t *testing.T) {
+	wf := unmarshalWf(runtimeResolutionOfVariableNames)
+	err := ValidateWorkflow(wftmplGetter, wf, ValidateOpts{})
+	assert.NoError(t, err)
+}