Repository: argoproj/argo (go)

Issue:
workflow controller is unable to write events in other namespaces for versions >= v2.9-rc1
**What happened**:
The workflow-controller is unable to write events for workflows running in other namespaces for workflow-controller verions after 2.8.2.

**What you expected to happen**:
That events are written successfully.

**How to reproduce it (as minimally and precisely as possible)**:
* Start with fresh cluster
* Follow the get started guide [here](https://github.com/argoproj/argo/blob/master/docs/getting-started.md)
* Create a new namespace `kubectl create ns test`
* Grant admin to default sa `kubectl create rolebinding default-admin --clusterrole=admin --serviceaccount=test:default -n test`
* Submit workflow `argo -n test submit examples/hello-world.yaml --watch`

The workflow runs just fine. The only thing not working is events. workflow-controller reports `Unable to write event: 'can't create an event with namespace 'test' in namespace 'argo'`.

This is only an issue in versions > 2.8.2. By changing `latest` to `v2.8.2` in `install.yaml` everything works just fine.

**Anything else we need to know?**:
Don't think so. Ask!

**Environment**:

- Argo version:
```
argo: v2.9.0-rc4
  BuildDate: 2020-06-26T21:55:04Z
  GitCommit: 5b109bcb9257653ecbae46e6315c8d65842de58a
  GitTreeState: clean
  GitTag: v2.9.0-rc4
  GoVersion: go1.13.4
  Compiler: gc
  Platform: darwin/amd64
```
- Kubernetes version :
```
clientVersion:
  buildDate: "2020-04-16T11:44:51Z"
  compiler: gc
  gitCommit: a17149e1a189050796ced469dbd78d380f2ed5ef
  gitTreeState: clean
  gitVersion: v1.16.9
  goVersion: go1.13.9
  major: "1"
  minor: "16"
  platform: darwin/amd64
serverVersion:
  buildDate: "2020-05-30T02:07:03Z"
  compiler: gc
  gitCommit: 14bc8ad5f8c245f1240a8e4eab128c4d51bfeffe
  gitTreeState: clean
  gitVersion: v1.16.9-gke.6
  goVersion: go1.13.9b4
  major: "1"
  minor: 16+
  platform: linux/amd64
```

**Other debugging information (if applicable)**:

- workflow result:

```
DEBU[0000] CLI version                                   version="{v2.9.0-rc4 2020-06-26T21:55:04Z 5b109bcb9257653ecbae46e6315c8d65842de58a v2.9.0-rc4 clean go1.13.4 gc darwin/amd64}"
DEBU[0000] Client options                                opts="{{ false false}  0x2277680 0xc0008b55e0}"
Name:                hello-world-fzdxb
Namespace:           test
ServiceAccount:      default
Status:              Succeeded
Conditions:
 Completed           True
Created:             Tue Jun 30 00:04:41 +0200 (3 minutes ago)
Started:             Tue Jun 30 00:04:41 +0200 (3 minutes ago)
Finished:            Tue Jun 30 00:04:46 +0200 (2 minutes ago)
Duration:            5 seconds
ResourcesDuration:   1s*(100Mi memory),1s*(1 cpu)

STEP                  TEMPLATE  PODNAME            DURATION  MESSAGE
 ✔ hello-world-fzdxb  whalesay  hello-world-fzdxb  3s
```

- workflow-controller logs:

```
time="2020-06-29T21:56:56Z" level=info msg="config map" name=workflow-controller-configmap
time="2020-06-29T21:56:56Z" level=info msg="Configuration:\nartifactRepository: {}\nmetricsConfig: {}\nnodeEvents: {}\npodSpecLogStrategy: {}\nsso:\n  clientId:\n    key: \"\"\n  clientSecret:\n    key: \"\"\n  issuer: \"\"\n  redirectUrl: \"\"\ntelemetryConfig: {}\n"
time="2020-06-29T21:56:56Z" level=info msg="Persistence configuration disabled"
time="2020-06-29T21:56:56Z" level=info msg="Starting Workflow Controller" version=8f4945f5
time="2020-06-29T21:56:56Z" level=info msg="Workers: workflow: 32, pod: 32"
time="2020-06-29T21:56:56Z" level=info msg="Performing periodic GC every 5m0s"
time="2020-06-29T21:56:56Z" level=info msg="Persistence disabled - so archived workflow GC disabled - you must restart the controller if you enable this"
time="2020-06-29T21:56:56Z" level=info msg="Starting workflow TTL controller (resync 20m0s)"
time="2020-06-29T21:56:56Z" level=info msg="Starting CronWorkflow controller"
time="2020-06-29T21:56:56Z" level=info msg="Starting prometheus metrics server at localhost:9090/metrics"
time="2020-06-29T21:56:56Z" level=info msg="Started workflow TTL worker"
time="2020-06-29T21:57:05Z" level=info msg="Processing workflow" namespace=test workflow=hello-world-kdvnr
time="2020-06-29T21:57:05Z" level=info msg="Updated phase  -> Running" namespace=test workflow=hello-world-kdvnr
time="2020-06-29T21:57:05Z" level=info msg="Pod node hello-world-kdvnr initialized Pending" namespace=test workflow=hello-world-kdvnr
E0629 21:57:05.825240       1 event.go:249] Unable to write event: 'can't create an event with namespace 'test' in namespace 'argo'' (may retry after sleeping)
time="2020-06-29T21:57:05Z" level=info msg="Created pod: hello-world-kdvnr (hello-world-kdvnr)" namespace=test workflow=hello-world-kdvnr
time="2020-06-29T21:57:05Z" level=info msg="Workflow update successful" namespace=test phase=Running resourceVersion=10542 workflow=hello-world-kdvnr
time="2020-06-29T21:57:05Z" level=info msg="Processing workflow" namespace=test workflow=hello-world-kdvnr
time="2020-06-29T21:57:05Z" level=info msg="Updating node hello-world-kdvnr message: ContainerCreating"
time="2020-06-29T21:57:05Z" level=info msg="Workflow update successful" namespace=test phase=Running resourceVersion=10543 workflow=hello-world-kdvnr
time="2020-06-29T21:57:05Z" level=info msg="Processing workflow" namespace=test workflow=hello-world-kdvnr
E0629 21:57:06.308725       1 event.go:249] Unable to write event: 'can't create an event with namespace 'test' in namespace 'argo'' (may retry after sleeping)
time="2020-06-29T21:57:06Z" level=info msg="Processing workflow" namespace=test workflow=hello-world-kdvnr
time="2020-06-29T21:57:10Z" level=info msg="Processing workflow" namespace=test workflow=hello-world-kdvnr
time="2020-06-29T21:57:10Z" level=info msg="Setting node hello-world-kdvnr outputs"
time="2020-06-29T21:57:10Z" level=info msg="Updating node hello-world-kdvnr status Pending -> Running"
time="2020-06-29T21:57:10Z" level=info msg="Workflow update successful" namespace=test phase=Running resourceVersion=10575 workflow=hello-world-kdvnr
time="2020-06-29T21:57:10Z" level=info msg="Processing workflow" namespace=test workflow=hello-world-kdvnr
time="2020-06-29T21:57:11Z" level=info msg="Processing workflow" namespace=test workflow=hello-world-kdvnr
time="2020-06-29T21:57:11Z" level=info msg="Updating node hello-world-kdvnr status Running -> Succeeded"
time="2020-06-29T21:57:11Z" level=info msg="Updated phase Running -> Succeeded" namespace=test workflow=hello-world-kdvnr
time="2020-06-29T21:57:11Z" level=info msg="Marking workflow completed" namespace=test workflow=hello-world-kdvnr
time="2020-06-29T21:57:11Z" level=info msg="Checking daemoned children of " namespace=test workflow=hello-world-kdvnr
time="2020-06-29T21:57:11Z" level=info msg="Workflow update successful" namespace=test phase=Succeeded resourceVersion=10586 workflow=hello-world-kdvnr
time="2020-06-29T21:57:11Z" level=info msg="Labeled pod test/hello-world-kdvnr completed"
E0629 21:57:16.308986       1 event.go:249] Unable to write event: 'can't create an event with namespace 'test' in namespace 'argo'' (may retry after sleeping)
E0629 21:57:26.309176       1 event.go:249] Unable to write event: 'can't create an event with namespace 'test' in namespace 'argo'' (may retry after sleeping)
E0629 21:57:36.309519       1 event.go:249] Unable to write event: 'can't create an event with namespace 'test' in namespace 'argo'' (may retry after sleeping)
E0629 21:57:46.309742       1 event.go:249] Unable to write event: 'can't create an event with namespace 'test' in namespace 'argo'' (may retry after sleeping)
E0629 21:57:56.310011       1 event.go:249] Unable to write event: 'can't create an event with namespace 'test' in namespace 'argo'' (may retry after sleeping)
E0629 21:58:06.310199       1 event.go:249] Unable to write event: 'can't create an event with namespace 'test' in namespace 'argo'' (may retry after sleeping)
E0629 21:58:16.310436       1 event.go:249] Unable to write event: 'can't create an event with namespace 'test' in namespace 'argo'' (may retry after sleeping)
E0629 21:58:26.310650       1 event.go:249] Unable to write event: 'can't create an event with namespace 'test' in namespace 'argo'' (may retry after sleeping)
```

---
<!-- Issue Author: Don't delete this message to encourage 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(controller): Allow events to be sent to non-argo namespace. Fixes #3342
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).

Fixes #3342

Patch:
diff --git a/USERS.md b/USERS.md
index 4a4394f82..fcbe8d0aa 100644
--- a/USERS.md
+++ b/USERS.md
@@ -45,6 +45,7 @@ Currently, the following organizations are **officially** using Argo Workflows:
 1. [InsideBoard](https://www.insideboard.com)
 1. [Interline Technologies](https://www.interline.io/blog/scaling-openstreetmap-data-workflows/)
 1. [Intuit](https://www.intuit.com/)
+1. [InVision](https://www.invisionapp.com/)
 1. [İşbank](https://www.isbank.com.tr/en)
 1. [Karius](https://www.kariusdx.com/)
 1. [KintoHub](https://www.kintohub.com/)
diff --git a/workflow/controller/controller.go b/workflow/controller/controller.go
index 184dd5411..8a0d8ecb1 100644
--- a/workflow/controller/controller.go
+++ b/workflow/controller/controller.go
@@ -22,11 +22,8 @@ import (
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/client-go/kubernetes"
-	"k8s.io/client-go/kubernetes/scheme"
-	typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/cache"
-	"k8s.io/client-go/tools/record"
 	"k8s.io/client-go/util/workqueue"
 	"upper.io/db.v3/lib/sqlbuilder"
 
@@ -84,7 +81,7 @@ type WorkflowController struct {
 	hydrator              hydrator.Interface
 	wfArchive             sqldb.WorkflowArchive
 	metrics               metrics.Metrics
-	eventRecorder         record.EventRecorder
+	eventRecorderManager  EventRecorderManager
 	archiveLabelSelector  labels.Selector
 }
 
@@ -97,10 +94,6 @@ 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 {
-	eventBroadcaster := record.NewBroadcaster()
-	eventBroadcaster.StartLogging(log.Debugf)
-	eventBroadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{Interface: kubeclientset.CoreV1().Events(namespace)})
-	eventRecorder := eventBroadcaster.NewRecorder(scheme.Scheme, apiv1.EventSource{Component: "workflow-controller"})
 	wfc := WorkflowController{
 		restConfig:                 restConfig,
 		kubeclientset:              kubeclientset,
@@ -115,7 +108,7 @@ func NewWorkflowController(restConfig *rest.Config, kubeclientset kubernetes.Int
 		configController:           config.NewController(namespace, configMap, kubeclientset),
 		completedPods:              make(chan string, 512),
 		gcPods:                     make(chan string, 512),
-		eventRecorder:              eventRecorder,
+		eventRecorderManager:       newEventRecorderManager(kubeclientset),
 	}
 	wfc.throttler = NewThrottler(0, wfc.wfQueue)
 	wfc.UpdateConfig()
diff --git a/workflow/controller/event_recorder_manager.go b/workflow/controller/event_recorder_manager.go
new file mode 100644
index 000000000..4988bb093
--- /dev/null
+++ b/workflow/controller/event_recorder_manager.go
@@ -0,0 +1,45 @@
+package controller
+
+import (
+	"sync"
+
+	log "github.com/sirupsen/logrus"
+	apiv1 "k8s.io/api/core/v1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes/scheme"
+	typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1"
+	"k8s.io/client-go/tools/record"
+)
+
+type EventRecorderManager interface {
+	Get(namespace string) record.EventRecorder
+}
+
+type eventRecorderManager struct {
+	kubernetes     kubernetes.Interface
+	lock           sync.Mutex
+	eventRecorders map[string]record.EventRecorder
+}
+
+func (m *eventRecorderManager) Get(namespace string) record.EventRecorder {
+	m.lock.Lock()
+	defer m.lock.Unlock()
+	eventRecorder, ok := m.eventRecorders[namespace]
+	if ok {
+		return eventRecorder
+	}
+	eventBroadcaster := record.NewBroadcaster()
+	eventBroadcaster.StartLogging(log.Debugf)
+	eventBroadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{Interface: m.kubernetes.CoreV1().Events(namespace)})
+	m.eventRecorders[namespace] = eventBroadcaster.NewRecorder(scheme.Scheme, apiv1.EventSource{Component: "workflow-controller"})
+	return m.eventRecorders[namespace]
+
+}
+
+func newEventRecorderManager(kubernetes kubernetes.Interface) EventRecorderManager {
+	return &eventRecorderManager{
+		kubernetes:     kubernetes,
+		lock:           sync.Mutex{},
+		eventRecorders: make(map[string]record.EventRecorder),
+	}
+}
diff --git a/workflow/controller/exec_control.go b/workflow/controller/exec_control.go
index 2f09323af..052954cdd 100644
--- a/workflow/controller/exec_control.go
+++ b/workflow/controller/exec_control.go
@@ -56,7 +56,7 @@ func (woc *wfOperationCtx) applyExecutionControl(pod *apiv1.Pod, wfNodesLock *sy
 					wfNodesLock.Lock()
 					defer wfNodesLock.Unlock()
 					node := woc.wf.Status.Nodes[pod.Name]
-					woc.markNodePhase(node.Name, wfv1.NodeFailed, fmt.Sprintf("step exceeded workflow deadline %s", *woc.workflowDeadline))
+					woc.markNodePhase(node.Name, wfv1.NodeFailed, fmt.Sprintf("Step exceeded its deadline"))
 					return nil
 				}
 				// If we fail to delete the pod, fall back to setting the annotation
@@ -73,31 +73,23 @@ func (woc *wfOperationCtx) applyExecutionControl(pod *apiv1.Pod, wfNodesLock *sy
 		}
 	}
 
-	var newDeadline *time.Time
-
 	if woc.wf.Spec.Shutdown != "" {
-		_, onExitPod := pod.Labels[common.LabelKeyOnExit]
-		if !woc.wf.Spec.Shutdown.ShouldExecute(onExitPod) {
-			newDeadline = &time.Time{}
+		if _, onExitPod := pod.Labels[common.LabelKeyOnExit]; !woc.wf.Spec.Shutdown.ShouldExecute(onExitPod) {
+			podExecCtl.Deadline = &time.Time{}
+			woc.log.Infof("Applying shutdown deadline for pod %s", pod.Name)
+			return woc.updateExecutionControl(pod.Name, podExecCtl)
 		}
-	} else {
-		if podExecCtl.Deadline == nil && woc.workflowDeadline == nil {
-			return nil
-		} else if podExecCtl.Deadline != nil && woc.workflowDeadline != nil && podExecCtl.Deadline.Equal(*woc.workflowDeadline) {
-			return nil
-		} else if podExecCtl.Deadline != nil && podExecCtl.Deadline.IsZero() {
-			// If the pod has already been explicitly signaled to terminate, then do nothing.
-			// This can happen when daemon steps are terminated.
-			woc.log.Infof("Skipping sync of execution control of pod %s. pod has been signaled to terminate", pod.Name)
-			return nil
+	}
+
+	if woc.workflowDeadline != nil {
+		if podExecCtl.Deadline == nil || woc.workflowDeadline.Before(*podExecCtl.Deadline) {
+			podExecCtl.Deadline = woc.workflowDeadline
+			woc.log.Infof("Applying sooner Workflow Deadline for pod %s at: %v", pod.Name, woc.workflowDeadline)
+			return woc.updateExecutionControl(pod.Name, podExecCtl)
 		}
-		newDeadline = woc.workflowDeadline
 	}
-	// Assign new deadline value to PodExeCtl
-	podExecCtl.Deadline = newDeadline
 
-	woc.log.Infof("Execution control for pod %s out-of-sync desired: %v, actual: %v", pod.Name, woc.workflowDeadline, podExecCtl.Deadline)
-	return woc.updateExecutionControl(pod.Name, podExecCtl)
+	return nil
 }
 
 // killDaemonedChildren kill any daemoned pods of a steps or DAG template node.
diff --git a/workflow/controller/operator.go b/workflow/controller/operator.go
index 393f12ae5..c935bbd9b 100644
--- a/workflow/controller/operator.go
+++ b/workflow/controller/operator.go
@@ -131,7 +131,7 @@ func newWorkflowOperationCtx(wf *wfv1.Workflow, wfc *WorkflowController) *wfOper
 		completedPods:          make(map[string]bool),
 		succeededPods:          make(map[string]bool),
 		deadline:               time.Now().UTC().Add(maxOperationTime),
-		eventRecorder:          wfc.eventRecorder,
+		eventRecorder:          wfc.eventRecorderManager.Get(wf.Namespace),
 		preExecutionNodePhases: make(map[string]wfv1.NodePhase),
 	}
 
@@ -614,7 +614,7 @@ func (woc *wfOperationCtx) requeue(afterDuration time.Duration) {
 }
 
 // processNodeRetries updates the retry node state based on the child node state and the retry strategy and returns the node.
-func (woc *wfOperationCtx) processNodeRetries(node *wfv1.NodeStatus, retryStrategy wfv1.RetryStrategy) (*wfv1.NodeStatus, bool, error) {
+func (woc *wfOperationCtx) processNodeRetries(node *wfv1.NodeStatus, retryStrategy wfv1.RetryStrategy, opts *executeTemplateOpts) (*wfv1.NodeStatus, bool, error) {
 	if node.Fulfilled() {
 		return node, true, nil
 	}
@@ -691,6 +691,10 @@ func (woc *wfOperationCtx) processNodeRetries(node *wfv1.NodeStatus, retryStrate
 			retryMessage := fmt.Sprintf("Backoff for %s", humanize.Duration(timeToWait))
 			return woc.markNodePhase(node.Name, node.Phase, retryMessage), false, nil
 		}
+
+		woc.log.WithField("node", node.Name).Infof("node has maxDuration set, setting executionDeadline to: %s", humanize.Timestamp(maxDurationDeadline))
+		opts.executionDeadline = maxDurationDeadline
+
 		node = woc.markNodePhase(node.Name, node.Phase, "")
 	}
 
@@ -865,7 +869,7 @@ func (woc *wfOperationCtx) failSuspendedNodesAfterDeadlineOrShutdown() error {
 				if woc.wfSpec.Shutdown != "" {
 					message = fmt.Sprintf("Stopped with strategy '%s'", woc.wfSpec.Shutdown)
 				} else {
-					message = fmt.Sprintf("step exceeded workflow deadline %s", *woc.workflowDeadline)
+					message = fmt.Sprintf("Step exceeded its deadline")
 				}
 				woc.markNodePhase(node.Name, wfv1.NodeFailed, message)
 			}
@@ -1363,6 +1367,8 @@ type executeTemplateOpts struct {
 	// onExitTemplate signifies that executeTemplate was called as part of an onExit handler.
 	// Necessary for graceful shutdowns
 	onExitTemplate bool
+	// activeDeadlineSeconds is a deadline to set to any pods executed. This is necessary for pods to inherit backoff.maxDuration
+	executionDeadline time.Time
 }
 
 // executeTemplate executes the template with the given arguments and returns the created NodeStatus
@@ -1445,7 +1451,7 @@ func (woc *wfOperationCtx) executeTemplate(nodeName string, orgTmpl wfv1.Templat
 			woc.log.Debugf("Inject a retry node for node %s", retryNodeName)
 			retryParentNode = woc.initializeExecutableNode(retryNodeName, wfv1.NodeTypeRetry, templateScope, processedTmpl, orgTmpl, opts.boundaryID, wfv1.NodeRunning)
 		}
-		processedRetryParentNode, continueExecution, err := woc.processNodeRetries(retryParentNode, *processedTmpl.RetryStrategy)
+		processedRetryParentNode, continueExecution, err := woc.processNodeRetries(retryParentNode, *processedTmpl.RetryStrategy, opts)
 		if err != nil {
 			return woc.markNodeError(retryNodeName, err), err
 		} else if !continueExecution {
@@ -1854,6 +1860,7 @@ func (woc *wfOperationCtx) executeContainer(nodeName string, templateScope strin
 	_, err = woc.createWorkflowPod(nodeName, *tmpl.Container, tmpl, &createWorkflowPodOpts{
 		includeScriptOutput: includeScriptOutput,
 		onExitPod:           opts.onExitTemplate,
+		executionDeadline:   opts.executionDeadline,
 	})
 
 	if apierr.IsForbidden(err) && isResubmitAllowed(tmpl) {
@@ -1992,6 +1999,7 @@ func (woc *wfOperationCtx) executeScript(nodeName string, templateScope string,
 	_, err = woc.createWorkflowPod(nodeName, mainCtr, tmpl, &createWorkflowPodOpts{
 		includeScriptOutput: includeScriptOutput,
 		onExitPod:           opts.onExitTemplate,
+		executionDeadline:   opts.executionDeadline,
 	})
 	return node, err
 }
@@ -2253,7 +2261,7 @@ func (woc *wfOperationCtx) executeResource(nodeName string, templateScope string
 
 	mainCtr := woc.newExecContainer(common.MainContainerName, tmpl)
 	mainCtr.Command = []string{"argoexec", "resource", tmpl.Resource.Action}
-	_, err = woc.createWorkflowPod(nodeName, *mainCtr, tmpl, &createWorkflowPodOpts{onExitPod: opts.onExitTemplate})
+	_, err = woc.createWorkflowPod(nodeName, *mainCtr, tmpl, &createWorkflowPodOpts{onExitPod: opts.onExitTemplate, executionDeadline: opts.executionDeadline})
 	return node, err
 }
 
diff --git a/workflow/controller/workflowpod.go b/workflow/controller/workflowpod.go
index db26f6dbf..25c0782db 100644
--- a/workflow/controller/workflowpod.go
+++ b/workflow/controller/workflowpod.go
@@ -90,6 +90,7 @@ func (woc *wfOperationCtx) hasPodSpecPatch(tmpl *wfv1.Template) bool {
 type createWorkflowPodOpts struct {
 	includeScriptOutput bool
 	onExitPod           bool
+	executionDeadline   time.Time
 }
 
 func (woc *wfOperationCtx) createWorkflowPod(nodeName string, mainCtr apiv1.Container, tmpl *wfv1.Template, opts *createWorkflowPodOpts) (*apiv1.Pod, error) {
@@ -196,7 +197,7 @@ func (woc *wfOperationCtx) createWorkflowPod(nodeName string, mainCtr apiv1.Cont
 	}
 
 	addSchedulingConstraints(pod, wfSpec, tmpl)
-	woc.addMetadata(pod, tmpl, opts.includeScriptOutput)
+	woc.addMetadata(pod, tmpl, opts)
 
 	err = addVolumeReferences(pod, woc.volumes, tmpl, woc.wf.Status.PersistentVolumeClaims)
 	if err != nil {
@@ -528,7 +529,7 @@ func isResourcesSpecified(ctr *apiv1.Container) bool {
 }
 
 // addMetadata applies metadata specified in the template
-func (woc *wfOperationCtx) addMetadata(pod *apiv1.Pod, tmpl *wfv1.Template, includeScriptOutput bool) {
+func (woc *wfOperationCtx) addMetadata(pod *apiv1.Pod, tmpl *wfv1.Template, opts *createWorkflowPodOpts) {
 	for k, v := range tmpl.Metadata.Annotations {
 		pod.ObjectMeta.Annotations[k] = v
 	}
@@ -537,14 +538,20 @@ func (woc *wfOperationCtx) addMetadata(pod *apiv1.Pod, tmpl *wfv1.Template, incl
 	}
 
 	execCtl := common.ExecutionControl{
-		IncludeScriptOutput: includeScriptOutput,
+		IncludeScriptOutput: opts.includeScriptOutput,
 	}
 
 	if woc.workflowDeadline != nil {
 		execCtl.Deadline = woc.workflowDeadline
 	}
 
-	if woc.workflowDeadline != nil || includeScriptOutput {
+	// If we're passed down an executionDeadline, only set it if there isn't one set already, or if it's before than
+	// the one already set.
+	if !opts.executionDeadline.IsZero() && (execCtl.Deadline == nil || opts.executionDeadline.Before(*execCtl.Deadline)) {
+		execCtl.Deadline = &opts.executionDeadline
+	}
+
+	if execCtl.Deadline != nil || opts.includeScriptOutput {
 		execCtlBytes, err := json.Marshal(execCtl)
 		if err != nil {
 			panic(err)
diff --git a/workflow/executor/executor.go b/workflow/executor/executor.go
index c9f2e2b3c..71d122c74 100644
--- a/workflow/executor/executor.go
+++ b/workflow/executor/executor.go
@@ -1085,7 +1085,7 @@ func (we *WorkflowExecutor) monitorDeadline(ctx context.Context, annotationsUpda
 					if we.ExecutionControl.Deadline.IsZero() {
 						message = "terminated"
 					} else {
-						message = fmt.Sprintf("step exceeded workflow deadline %s", *we.ExecutionControl.Deadline)
+						message = fmt.Sprintf("Step exceeded its deadline")
 					}
 					log.Info(message)
 					_ = we.AddAnnotation(common.AnnotationKeyNodeMessage, message)


Test patch:
diff --git a/test/e2e/functional_test.go b/test/e2e/functional_test.go
index 7c5bcdcd4..e09f13764 100644
--- a/test/e2e/functional_test.go
+++ b/test/e2e/functional_test.go
@@ -599,6 +599,45 @@ func (s *FunctionalSuite) TestWorkflowTemplateRefWithExitHandler() {
 
 }
 
+func (s *FunctionalSuite) TestPropagateMaxDuration() {
+	s.Given().
+		Workflow(`
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  name: retry-backoff-2
+  labels:
+    argo-e2e: true
+spec:
+  entrypoint: retry-backoff
+  templates:
+  - name: retry-backoff
+    retryStrategy:
+      limit: 10
+      backoff:
+        duration: "1"
+        factor: 1
+        maxDuration: "10"
+    container:
+      image: python:alpine3.6
+      command: [sh, -c]
+      args: ["sleep $(( {{retries}} * 40 )); exit 1"]
+
+`).
+		When().
+		SubmitWorkflow().
+		WaitForWorkflow(45 * time.Second).
+		Then().
+		ExpectWorkflow(func(t *testing.T, _ *metav1.ObjectMeta, status *wfv1.WorkflowStatus) {
+			assert.Equal(t, wfv1.NodeFailed, status.Phase)
+			assert.Len(t, status.Nodes, 3)
+			node := status.Nodes.FindByDisplayName("retry-backoff-2(1)")
+			if assert.NotNil(t, node) {
+				assert.Equal(t, "Step exceeded its deadline", node.Message)
+			}
+		})
+}
+
 func TestFunctionalSuite(t *testing.T) {
 	suite.Run(t, new(FunctionalSuite))
 }
diff --git a/workflow/controller/controller_test.go b/workflow/controller/controller_test.go
index b85bc1405..171a2d065 100644
--- a/workflow/controller/controller_test.go
+++ b/workflow/controller/controller_test.go
@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/client-go/tools/record"
 
 	"github.com/stretchr/testify/assert"
 	authorizationv1 "k8s.io/api/authorization/v1"
@@ -16,7 +17,6 @@ import (
 	"k8s.io/client-go/kubernetes/fake"
 	k8stesting "k8s.io/client-go/testing"
 	"k8s.io/client-go/tools/cache"
-	"k8s.io/client-go/tools/record"
 	"k8s.io/client-go/util/workqueue"
 	"sigs.k8s.io/yaml"
 
@@ -102,6 +102,16 @@ spec:
       args: ["hello world"]
 `
 
+type testEventRecorderManager struct {
+	eventRecorder *record.FakeRecorder
+}
+
+func (t testEventRecorderManager) Get(string) record.EventRecorder {
+	return t.eventRecorder
+}
+
+var _ EventRecorderManager = &testEventRecorderManager{}
+
 func newController(objects ...runtime.Object) (context.CancelFunc, *WorkflowController) {
 	wfclientset := fakewfclientset.NewSimpleClientset(objects...)
 	informerFactory := wfextv.NewSharedInformerFactory(wfclientset, 10*time.Minute)
@@ -133,7 +143,7 @@ func newController(objects ...runtime.Object) (context.CancelFunc, *WorkflowCont
 		wfArchive:            sqldb.NullWorkflowArchive,
 		hydrator:             hydratorfake.Noop,
 		metrics:              metrics.New(metrics.ServerConfig{}, metrics.ServerConfig{}),
-		eventRecorder:        record.NewFakeRecorder(16),
+		eventRecorderManager: &testEventRecorderManager{eventRecorder: record.NewFakeRecorder(16)},
 		archiveLabelSelector: labels.Everything(),
 	}
 	return cancel, controller
diff --git a/workflow/controller/operator_test.go b/workflow/controller/operator_test.go
index f4d84080a..71ea4d4c0 100644
--- a/workflow/controller/operator_test.go
+++ b/workflow/controller/operator_test.go
@@ -13,7 +13,6 @@ import (
 	apiv1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/client-go/tools/record"
 	"sigs.k8s.io/yaml"
 
 	"github.com/argoproj/argo/config"
@@ -202,13 +201,13 @@ func TestProcessNodesWithRetries(t *testing.T) {
 
 	// Last child is still running. processNodesWithRetries() should return false since
 	// there should be no retries at this point.
-	n, _, err = woc.processNodeRetries(n, retries)
+	n, _, err = woc.processNodeRetries(n, retries, &executeTemplateOpts{})
 	assert.NoError(t, err)
 	assert.Equal(t, n.Phase, wfv1.NodeRunning)
 
 	// Mark lastChild as successful.
 	woc.markNodePhase(lastChild.Name, wfv1.NodeSucceeded)
-	n, _, err = woc.processNodeRetries(n, retries)
+	n, _, err = woc.processNodeRetries(n, retries, &executeTemplateOpts{})
 	assert.NoError(t, err)
 	// The parent node also gets marked as Succeeded.
 	assert.Equal(t, n.Phase, wfv1.NodeSucceeded)
@@ -216,7 +215,7 @@ func TestProcessNodesWithRetries(t *testing.T) {
 	// Mark the parent node as running again and the lastChild as failed.
 	woc.markNodePhase(n.Name, wfv1.NodeRunning)
 	woc.markNodePhase(lastChild.Name, wfv1.NodeFailed)
-	_, _, err = woc.processNodeRetries(n, retries)
+	_, _, err = woc.processNodeRetries(n, retries, &executeTemplateOpts{})
 	assert.NoError(t, err)
 	n = woc.getNodeByName(nodeName)
 	assert.Equal(t, n.Phase, wfv1.NodeRunning)
@@ -226,7 +225,7 @@ func TestProcessNodesWithRetries(t *testing.T) {
 	woc.initializeNode(childNode, wfv1.NodeTypePod, "", &wfv1.Template{}, "", wfv1.NodeFailed)
 	woc.addChildNode(nodeName, childNode)
 	n = woc.getNodeByName(nodeName)
-	n, _, err = woc.processNodeRetries(n, retries)
+	n, _, err = woc.processNodeRetries(n, retries, &executeTemplateOpts{})
 	assert.NoError(t, err)
 	assert.Equal(t, n.Phase, wfv1.NodeFailed)
 }
@@ -274,13 +273,13 @@ func TestProcessNodesWithRetriesOnErrors(t *testing.T) {
 
 	// Last child is still running. processNodesWithRetries() should return false since
 	// there should be no retries at this point.
-	n, _, err = woc.processNodeRetries(n, retries)
+	n, _, err = woc.processNodeRetries(n, retries, &executeTemplateOpts{})
 	assert.Nil(t, err)
 	assert.Equal(t, n.Phase, wfv1.NodeRunning)
 
 	// Mark lastChild as successful.
 	woc.markNodePhase(lastChild.Name, wfv1.NodeSucceeded)
-	n, _, err = woc.processNodeRetries(n, retries)
+	n, _, err = woc.processNodeRetries(n, retries, &executeTemplateOpts{})
 	assert.Nil(t, err)
 	// The parent node also gets marked as Succeeded.
 	assert.Equal(t, n.Phase, wfv1.NodeSucceeded)
@@ -288,7 +287,7 @@ func TestProcessNodesWithRetriesOnErrors(t *testing.T) {
 	// Mark the parent node as running again and the lastChild as errored.
 	n = woc.markNodePhase(n.Name, wfv1.NodeRunning)
 	woc.markNodePhase(lastChild.Name, wfv1.NodeError)
-	_, _, err = woc.processNodeRetries(n, retries)
+	_, _, err = woc.processNodeRetries(n, retries, &executeTemplateOpts{})
 	assert.NoError(t, err)
 	n = woc.getNodeByName(nodeName)
 	assert.Equal(t, n.Phase, wfv1.NodeRunning)
@@ -298,7 +297,7 @@ func TestProcessNodesWithRetriesOnErrors(t *testing.T) {
 	woc.initializeNode(childNode, wfv1.NodeTypePod, "", &wfv1.Template{}, "", wfv1.NodeError)
 	woc.addChildNode(nodeName, childNode)
 	n = woc.getNodeByName(nodeName)
-	n, _, err = woc.processNodeRetries(n, retries)
+	n, _, err = woc.processNodeRetries(n, retries, &executeTemplateOpts{})
 	assert.Nil(t, err)
 	assert.Equal(t, n.Phase, wfv1.NodeError)
 }
@@ -347,13 +346,13 @@ func TestProcessNodesWithRetriesWithBackoff(t *testing.T) {
 
 	// Last child is still running. processNodesWithRetries() should return false since
 	// there should be no retries at this point.
-	n, _, err = woc.processNodeRetries(n, retries)
+	n, _, err = woc.processNodeRetries(n, retries, &executeTemplateOpts{})
 	assert.Nil(t, err)
 	assert.Equal(t, n.Phase, wfv1.NodeRunning)
 
 	// Mark lastChild as successful.
 	woc.markNodePhase(lastChild.Name, wfv1.NodeSucceeded)
-	n, _, err = woc.processNodeRetries(n, retries)
+	n, _, err = woc.processNodeRetries(n, retries, &executeTemplateOpts{})
 	assert.Nil(t, err)
 	// The parent node also gets marked as Succeeded.
 	assert.Equal(t, n.Phase, wfv1.NodeSucceeded)
@@ -402,13 +401,13 @@ func TestProcessNodesNoRetryWithError(t *testing.T) {
 
 	// Last child is still running. processNodesWithRetries() should return false since
 	// there should be no retries at this point.
-	n, _, err = woc.processNodeRetries(n, retries)
+	n, _, err = woc.processNodeRetries(n, retries, &executeTemplateOpts{})
 	assert.Nil(t, err)
 	assert.Equal(t, n.Phase, wfv1.NodeRunning)
 
 	// Mark lastChild as successful.
 	woc.markNodePhase(lastChild.Name, wfv1.NodeSucceeded)
-	n, _, err = woc.processNodeRetries(n, retries)
+	n, _, err = woc.processNodeRetries(n, retries, &executeTemplateOpts{})
 	assert.Nil(t, err)
 	// The parent node also gets marked as Succeeded.
 	assert.Equal(t, n.Phase, wfv1.NodeSucceeded)
@@ -417,7 +416,7 @@ func TestProcessNodesNoRetryWithError(t *testing.T) {
 	// Parent node should also be errored because retry on error is disabled
 	n = woc.markNodePhase(n.Name, wfv1.NodeRunning)
 	woc.markNodePhase(lastChild.Name, wfv1.NodeError)
-	_, _, err = woc.processNodeRetries(n, retries)
+	_, _, err = woc.processNodeRetries(n, retries, &executeTemplateOpts{})
 	assert.NoError(t, err)
 	n = woc.getNodeByName(nodeName)
 	assert.Equal(t, wfv1.NodeError, n.Phase)
@@ -567,7 +566,7 @@ func TestBackoffMessage(t *testing.T) {
 	lastNode.FinishedAt = metav1.Time{Time: time.Now().Add(-1 * time.Second)}
 	woc.wf.Status.Nodes[lastNode.ID] = *lastNode
 
-	newRetryNode, proceed, err := woc.processNodeRetries(retryNode, *woc.wf.Spec.Templates[0].RetryStrategy)
+	newRetryNode, proceed, err := woc.processNodeRetries(retryNode, *woc.wf.Spec.Templates[0].RetryStrategy, &executeTemplateOpts{})
 	assert.NoError(t, err)
 	assert.False(t, proceed)
 	assert.Equal(t, "Backoff for 4 seconds", newRetryNode.Message)
@@ -580,7 +579,7 @@ func TestBackoffMessage(t *testing.T) {
 	lastNode.FinishedAt = metav1.Time{Time: time.Now().Add(-2 * time.Second)}
 	woc.wf.Status.Nodes[lastNode.ID] = *lastNode
 
-	newRetryNode, proceed, err = woc.processNodeRetries(retryNode, *woc.wf.Spec.Templates[0].RetryStrategy)
+	newRetryNode, proceed, err = woc.processNodeRetries(retryNode, *woc.wf.Spec.Templates[0].RetryStrategy, &executeTemplateOpts{})
 	assert.NoError(t, err)
 	assert.False(t, proceed)
 	// Message should not change
@@ -594,7 +593,7 @@ func TestBackoffMessage(t *testing.T) {
 	lastNode.FinishedAt = metav1.Time{Time: time.Now().Add(-5 * time.Second)}
 	woc.wf.Status.Nodes[lastNode.ID] = *lastNode
 
-	newRetryNode, proceed, err = woc.processNodeRetries(retryNode, *woc.wf.Spec.Templates[0].RetryStrategy)
+	newRetryNode, proceed, err = woc.processNodeRetries(retryNode, *woc.wf.Spec.Templates[0].RetryStrategy, &executeTemplateOpts{})
 	assert.NoError(t, err)
 	assert.True(t, proceed)
 	// New node is started, message should be clear
@@ -1239,7 +1238,7 @@ func TestSuspendWithDeadline(t *testing.T) {
 	for _, node := range updatedWf.Status.Nodes {
 		if node.Type == wfv1.NodeTypeSuspend {
 			assert.Equal(t, node.Phase, wfv1.NodeFailed)
-			assert.Contains(t, node.Message, "step exceeded workflow deadline")
+			assert.Contains(t, node.Message, "Step exceeded its deadline")
 			found = true
 		}
 	}
@@ -2736,7 +2735,7 @@ func TestEventInvalidSpec(t *testing.T) {
 }
 
 func getEvents(controller *WorkflowController, num int) []string {
-	c := controller.eventRecorder.(*record.FakeRecorder).Events
+	c := controller.eventRecorderManager.(*testEventRecorderManager).eventRecorder.Events
 	events := make([]string, num)
 	for i := 0; i < num; i++ {
 		events[i] = <-c
@@ -3848,3 +3847,62 @@ func TestWorkflowStatusMetric(t *testing.T) {
 	// Must only be one (completed: true)
 	assert.Len(t, woc.wf.Status.Conditions, 1)
 }
+
+var propagate = `
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  name: retry-backoff
+spec:
+  entrypoint: retry-backoff
+  templates:
+  - name: retry-backoff
+    retryStrategy:
+      limit: 10
+      backoff:
+        duration: "1"
+        factor: 1
+        maxDuration: "20"
+    container:
+      image: alpine
+      command: [sh, -c]
+      args: ["sleep $(( {{retries}} * 100 )); exit 1"]
+`
+
+func TestPropagateMaxDurationProcess(t *testing.T) {
+	cancel, controller := newController()
+	defer cancel()
+	assert.NotNil(t, controller)
+	wf := unmarshalWF(propagate)
+	assert.NotNil(t, wf)
+	woc := newWorkflowOperationCtx(wf, controller)
+	assert.NotNil(t, woc)
+	_, _, err := woc.loadExecutionSpec()
+	assert.NoError(t, err)
+	assert.Zero(t, len(woc.wf.Status.Nodes))
+
+	// Add the parent node for retries.
+	nodeName := "test-node"
+	node := woc.initializeNode(nodeName, wfv1.NodeTypeRetry, "", &wfv1.Template{}, "", wfv1.NodeRunning)
+	retryLimit := int32(2)
+	retries := wfv1.RetryStrategy{
+		Limit: &retryLimit,
+		Backoff: &wfv1.Backoff{
+			Duration:    "0",
+			Factor:      1,
+			MaxDuration: "20",
+		},
+	}
+	woc.wf.Status.Nodes[woc.wf.NodeID(nodeName)] = *node
+
+	childNode := fmt.Sprintf("child-node-%d", 0)
+	woc.initializeNode(childNode, wfv1.NodeTypePod, "", &wfv1.Template{}, "", wfv1.NodeFailed)
+	woc.addChildNode(nodeName, childNode)
+
+	var opts executeTemplateOpts
+	n := woc.getNodeByName(nodeName)
+	_, _, err = woc.processNodeRetries(n, retries, &opts)
+	if assert.NoError(t, err) {
+		assert.Equal(t, n.StartedAt.Add(20*time.Second).Round(time.Second).String(), opts.executionDeadline.Round(time.Second).String())
+	}
+}
diff --git a/workflow/controller/workflowpod_test.go b/workflow/controller/workflowpod_test.go
index ef391dd50..8e49c5425 100644
--- a/workflow/controller/workflowpod_test.go
+++ b/workflow/controller/workflowpod_test.go
@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"path/filepath"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	apiv1 "k8s.io/api/core/v1"
@@ -1109,3 +1110,31 @@ func TestPodSpecPatch(t *testing.T) {
 	assert.Equal(t, "104857600", pod.Spec.Containers[1].Resources.Limits.Memory().AsDec().String())
 
 }
+
+var propagateMaxDuration = `
+name: retry-backoff
+retryStrategy:
+  limit: 10
+  backoff:
+    duration: "1"
+    factor: 1
+    maxDuration: "20"
+container:
+  image: alpine
+  command: [sh, -c]
+  args: ["sleep $(( {{retries}} * 100 )); exit 1"]
+
+`
+
+func TestPropagateMaxDuration(t *testing.T) {
+	// Ensure that volume mount is added when artifact is provided
+	tmpl := unmarshalTemplate(propagateMaxDuration)
+	woc := newWoc()
+	deadline := time.Now()
+	pod, err := woc.createWorkflowPod(tmpl.Name, *tmpl.Container, tmpl, &createWorkflowPodOpts{executionDeadline: deadline})
+	assert.NoError(t, err)
+	out, err := json.Marshal(map[string]time.Time{"deadline": deadline})
+	if assert.NoError(t, err) {
+		assert.Equal(t, string(out), pod.Annotations[common.AnnotationKeyExecutionControl])
+	}
+}