Repository: argoproj/argo
Pull request #3464 (2020-07-13 17:57:24)

docs: Add controller-level metrics
Closes https://github.com/argoproj/argo/issues/3057


Patch:
diff --git a/cmd/argo/commands/list.go b/cmd/argo/commands/list.go
index 3fdf9bd61..4f43ba676 100644
--- a/cmd/argo/commands/list.go
+++ b/cmd/argo/commands/list.go
@@ -26,6 +26,7 @@ type listFlags struct {
 	status        []string
 	completed     bool
 	running       bool
+	resubmitted   bool
 	prefix        string
 	output        string
 	createdSince  string
@@ -66,6 +67,7 @@ func NewListCommand() *cobra.Command {
 	command.Flags().StringSliceVar(&listArgs.status, "status", []string{}, "Filter by status (comma separated)")
 	command.Flags().BoolVar(&listArgs.completed, "completed", false, "Show only completed workflows")
 	command.Flags().BoolVar(&listArgs.running, "running", false, "Show only running workflows")
+	command.Flags().BoolVar(&listArgs.resubmitted, "resubmitted", false, "Show only resubmitted workflows")
 	command.Flags().StringVarP(&listArgs.output, "output", "o", "", "Output format. One of: wide|name")
 	command.Flags().StringVar(&listArgs.createdSince, "since", "", "Show only workflows created after than a relative duration")
 	command.Flags().Int64VarP(&listArgs.chunkSize, "chunk-size", "", 0, "Return large lists in chunks rather than all at once. Pass 0 to disable.")
@@ -95,6 +97,10 @@ func listWorkflows(ctx context.Context, serviceClient workflowpkg.WorkflowServic
 		req, _ := labels.NewRequirement(common.LabelKeyCompleted, selection.NotEquals, []string{"true"})
 		labelSelector = labelSelector.Add(*req)
 	}
+	if flags.resubmitted {
+		req, _ := labels.NewRequirement(common.LabelKeyPreviousWorkflowName, selection.Exists, []string{})
+		labelSelector = labelSelector.Add(*req)
+	}
 	listOpts.LabelSelector = labelSelector.String()
 	listOpts.FieldSelector = flags.fields
 	var workflows wfv1.Workflows
diff --git a/cmd/argo/commands/logs.go b/cmd/argo/commands/logs.go
index a7feb16b6..1fc0e78f6 100644
--- a/cmd/argo/commands/logs.go
+++ b/cmd/argo/commands/logs.go
@@ -92,6 +92,7 @@ func NewLogsCommand() *cobra.Command {
 	}
 	command.Flags().StringVarP(&logOptions.Container, "container", "c", "main", "Print the logs of this container")
 	command.Flags().BoolVarP(&logOptions.Follow, "follow", "f", false, "Specify if the logs should be streamed.")
+	command.Flags().BoolVarP(&logOptions.Previous, "previous", "p", false, "Specify if the previously terminated container logs should be returned.")
 	command.Flags().DurationVar(&since, "since", 0, "Only return logs newer than a relative duration like 5s, 2m, or 3h. Defaults to all logs. Only one of since-time / since may be used.")
 	command.Flags().StringVar(&sinceTime, "since-time", "", "Only return logs after a specific date (RFC3339). Defaults to all logs. Only one of since-time / since may be used.")
 	command.Flags().Int64Var(&tailLines, "tail", -1, "If set, the number of lines from the end of the logs to show. If not specified, logs are shown from the creation of the container or sinceSeconds or sinceTime")
diff --git a/cmd/argo/commands/submit.go b/cmd/argo/commands/submit.go
index fcb701224..37cfe1be7 100644
--- a/cmd/argo/commands/submit.go
+++ b/cmd/argo/commands/submit.go
@@ -257,6 +257,7 @@ func waitWatchOrLog(ctx context.Context, serviceClient workflowpkg.WorkflowServi
 			logWorkflow(ctx, serviceClient, namespace, workflow, "", &corev1.PodLogOptions{
 				Container: "main",
 				Follow:    true,
+				Previous:  false,
 			})
 		}
 	}
diff --git a/docs/cli/argo_logs.md b/docs/cli/argo_logs.md
index f032e22ee..82ba82743 100644
--- a/docs/cli/argo_logs.md
+++ b/docs/cli/argo_logs.md
@@ -45,6 +45,7 @@ argo logs WORKFLOW [POD] [flags]
   -f, --follow              Specify if the logs should be streamed.
   -h, --help                help for logs
       --no-color            Disable colorized output
+  -p, --previous            Specify if the previously terminated container logs should be returned.
       --since duration      Only return logs newer than a relative duration like 5s, 2m, or 3h. Defaults to all logs. Only one of since-time / since may be used.
       --since-time string   Only return logs after a specific date (RFC3339). Defaults to all logs. Only one of since-time / since may be used.
       --tail int            If set, the number of lines from the end of the logs to show. If not specified, logs are shown from the creation of the container or sinceSeconds or sinceTime (default -1)
diff --git a/docs/metrics.md b/docs/metrics.md
index db7a15475..ddaf192c7 100644
--- a/docs/metrics.md
+++ b/docs/metrics.md
@@ -6,6 +6,9 @@
 
 ## Introduction
 
+Argo emits a certain number of controller metrics that inform on the state of the controller at any given time. Furthermore,
+users can also define their own custom metrics to inform on the state of their Workflows.
+
 Custom prometheus metrics can be defined to be emitted on a `Workflow`- and `Template`-level basis. These can be useful
 for many cases; some examples:
 
@@ -18,7 +21,12 @@ best way to define metrics in Argo to avoid problems such as [cardinality explos
 
 ## Metrics and metrics in Argo
 
-Emitting metrics is the responsibility of the emitter owner. Since the user defines Workflows in Argo, the user is responsible
+There are two kinds of metrics emitted by Argo: **controller metrics** and **custom metrics**.
+
+* Controller metrics are metrics that inform on the state of the controller; i.e., they answer the question "What is the state of the controller right now?".
+* Custom metrics are metrics that inform on the state of a Workflow, or a series of Workflows. These custom metrics are defined by the user in the Workflow spec.
+
+Emitting custom metrics is the responsibility of the emitter owner. Since the user defines Workflows in Argo, the user is responsible
 for emitting metrics correctly.
 
 ### What is and isn't a Prometheus metric
@@ -46,6 +54,17 @@ at the current time, they should not be used to report historical data such as:
 Metrics are also ephemeral, meaning there is no guarantee that they will be persisted for any amount of time. If you need
 a way to view and analyze historical data, consider the [workflow archive](workflow-archive.md) or reporting to logs.
 
+### Default Controller Metrics
+
+There are several controller-level metrics. These include:
+
+* `workflows_processed_count`: a count of all Workflow updates processed by the controller
+* `count`: a count of all workflows currently accessible by the controller by status
+* `operation_duration_seconds`: a histogram of durations of operations
+* `error_count`: a count of certain errors incurred by the controller
+* `queue_depth_count`: the depth of the queue of workflows or cron workflows to be processed by the controller
+* `queue_adds_count`: the number of adds to the queue of workflows or cron workflows
+* `queue_latency`: the time workflows or cron workflows spend in the queue waiting to be processed
 
 ### Metric types
 
diff --git a/docs/quick-start.md b/docs/quick-start.md
index ba425eaba..8e7224b1a 100644
--- a/docs/quick-start.md
+++ b/docs/quick-start.md
@@ -26,7 +26,7 @@ If you are running Argo Workflows locally (e.g. using Minikube or Docker for Des
 kubectl -n argo port-forward deployment/argo-server 2746:2746
 ```
 
-This will serve the user interface on https://localhost:2746
+This will serve the user interface on http://localhost:2746
 
 If you're using running Argo Workflows on a remote cluster (e.g. on EKS or GKE) then [follow these instructions](argo-server.md#access-the-argo-workflows-ui). 
 
diff --git a/workflow/common/util.go b/workflow/common/util.go
index 6998dc3ca..44089adb0 100644
--- a/workflow/common/util.go
+++ b/workflow/common/util.go
@@ -368,6 +368,16 @@ func Replace(fstTmpl *fasttemplate.Template, replaceMap map[string]string, allow
 	replacedTmpl := fstTmpl.ExecuteFuncString(func(w io.Writer, tag string) (int, error) {
 		replacement, ok := replaceMap[tag]
 		if !ok {
+			// Attempt to resolve nested tags, if possible
+			if index := strings.LastIndex(tag, "{{"); index > 0 {
+				nestedTagPrefix := tag[:index]
+				nestedTag := tag[index+2:]
+				if replacement, ok := replaceMap[nestedTag]; ok {
+					replacement = strconv.Quote(replacement)
+					replacement = replacement[1 : len(replacement)-1]
+					return w.Write([]byte("{{" + nestedTagPrefix + replacement))
+				}
+			}
 			if allowUnresolved {
 				// just write the same string back
 				return w.Write([]byte(fmt.Sprintf("{{%s}}", tag)))
diff --git a/workflow/controller/controller.go b/workflow/controller/controller.go
index 54db26bf5..99a2aaa54 100644
--- a/workflow/controller/controller.go
+++ b/workflow/controller/controller.go
@@ -726,30 +726,34 @@ func (wfc *WorkflowController) newPodInformer() cache.SharedIndexInformer {
 		cache.ResourceEventHandlerFuncs{
 			AddFunc: func(obj interface{}) {
 				key, err := cache.MetaNamespaceKeyFunc(obj)
-				if err == nil {
-					wfc.podQueue.Add(key)
+				if err != nil {
+					return
 				}
+				wfc.podQueue.Add(key)
 			},
 			UpdateFunc: func(old, new interface{}) {
+				key, err := cache.MetaNamespaceKeyFunc(new)
+				if err != nil {
+					return
+				}
 				oldPod, newPod := old.(*apiv1.Pod), new.(*apiv1.Pod)
 				if oldPod.ResourceVersion == newPod.ResourceVersion {
 					return
 				}
 				if !pod.SignificantPodChange(oldPod, newPod) {
+					log.WithField("key", key).Info("insignificant pod change")
 					return
 				}
-				key, err := cache.MetaNamespaceKeyFunc(new)
-				if err == nil {
-					wfc.podQueue.Add(key)
-				}
+				wfc.podQueue.Add(key)
 			},
 			DeleteFunc: func(obj interface{}) {
 				// IndexerInformer uses a delta queue, therefore for deletes we have to use this
 				// key function.
 				key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
-				if err == nil {
-					wfc.podQueue.Add(key)
+				if err != nil {
+					return
 				}
+				wfc.podQueue.Add(key)
 			},
 		},
 	)
diff --git a/workflow/controller/operator.go b/workflow/controller/operator.go
index cbd818b9a..2226b4363 100644
--- a/workflow/controller/operator.go
+++ b/workflow/controller/operator.go
@@ -87,7 +87,10 @@ type wfOperationCtx struct {
 	// preExecutionNodePhases contains the phases of all the nodes before the current operation. Necessary to infer
 	// changes in phase for metric emission
 	preExecutionNodePhases map[string]wfv1.NodePhase
-	// wfSpec holds the WorkflowSpec for use in execution
+	// wfSpec holds the WorkflowSpec for use in execution.
+	// This should usually be used instead `wf.Spec`, with two exceptions for user editable fields:
+	// 1. `wf.Spec.Suspend`
+	// 2. `wf.Spec.Shutdown`
 	wfSpec *wfv1.WorkflowSpec
 }
 
@@ -422,11 +425,11 @@ func (woc *wfOperationCtx) getWorkflowDeadline() *time.Time {
 func (woc *wfOperationCtx) setGlobalParameters(executionParameters wfv1.Arguments) {
 	woc.globalParams[common.GlobalVarWorkflowName] = woc.wf.ObjectMeta.Name
 	woc.globalParams[common.GlobalVarWorkflowNamespace] = woc.wf.ObjectMeta.Namespace
-	woc.globalParams[common.GlobalVarWorkflowServiceAccountName] = woc.wf.Spec.ServiceAccountName
+	woc.globalParams[common.GlobalVarWorkflowServiceAccountName] = woc.wfSpec.ServiceAccountName
 	woc.globalParams[common.GlobalVarWorkflowUID] = string(woc.wf.ObjectMeta.UID)
 	woc.globalParams[common.GlobalVarWorkflowCreationTimestamp] = woc.wf.ObjectMeta.CreationTimestamp.Format(time.RFC3339)
-	if woc.wf.Spec.Priority != nil {
-		woc.globalParams[common.GlobalVarWorkflowPriority] = strconv.Itoa(int(*woc.wf.Spec.Priority))
+	if woc.wfSpec.Priority != nil {
+		woc.globalParams[common.GlobalVarWorkflowPriority] = strconv.Itoa(int(*woc.wfSpec.Priority))
 	}
 	for char := range strftime.FormatChars {
 		cTimeVar := fmt.Sprintf("%s.%s", common.GlobalVarWorkflowCreationTimestamp, string(char))
@@ -1482,6 +1485,16 @@ func (woc *wfOperationCtx) executeTemplate(nodeName string, orgTmpl wfv1.Templat
 		retryParentNode = processedRetryParentNode
 		// The retry node might have completed by now.
 		if retryParentNode.Fulfilled() {
+			if resolvedTmpl.Metrics != nil {
+				// In this check, a completed node may or may not have existed prior to this execution. If it did exist, ensure that it wasn't
+				// completed before this execution. If it did not exist prior, then we can infer that it was completed during this execution.
+				// The statement "(!ok || !prevNodeStatus.Fulfilled())" checks for this behavior and represents the material conditional
+				// "ok -> !prevNodeStatus.Fulfilled()" (https://en.wikipedia.org/wiki/Material_conditional)
+				if prevNodeStatus, ok := woc.preExecutionNodePhases[retryParentNode.ID]; (!ok || !prevNodeStatus.Fulfilled()) && retryParentNode.Fulfilled() {
+					localScope, realTimeScope := woc.prepareMetricScope(node)
+					woc.computeMetrics(resolvedTmpl.Metrics.Prometheus, localScope, realTimeScope, false)
+				}
+			}
 			return retryParentNode, nil
 		}
 		lastChildNode := getChildNodeIndex(retryParentNode, woc.wf.Status.Nodes, -1)
@@ -1558,9 +1571,9 @@ func (woc *wfOperationCtx) executeTemplate(nodeName string, orgTmpl wfv1.Templat
 		//
 		// In this check, a completed node may or may not have existed prior to this execution. If it did exist, ensure that it wasn't
 		// completed before this execution. If it did not exist prior, then we can infer that it was completed during this execution.
-		// The statement "(!ok || prevNodeStatus.Completed())" checks for this behavior and represents the material conditional
-		// "ok -> prevNodeStatus.Completed()" (https://en.wikipedia.org/wiki/Material_conditional)
-		if prevNodeStatus, ok := woc.preExecutionNodePhases[node.ID]; (!ok || prevNodeStatus.Fulfilled()) && node.Fulfilled() {
+		// The statement "(!ok || !prevNodeStatus.Fulfilled())" checks for this behavior and represents the material conditional
+		// "ok -> !prevNodeStatus.Fulfilled()" (https://en.wikipedia.org/wiki/Material_conditional)
+		if prevNodeStatus, ok := woc.preExecutionNodePhases[node.ID]; (!ok || !prevNodeStatus.Fulfilled()) && node.Fulfilled() {
 			localScope, realTimeScope := woc.prepareMetricScope(node)
 			woc.computeMetrics(resolvedTmpl.Metrics.Prometheus, localScope, realTimeScope, false)
 		}
@@ -1894,7 +1907,7 @@ func (woc *wfOperationCtx) executeContainer(nodeName string, templateScope strin
 	})
 
 	if err != nil {
-		return woc.checkForbiddenErrorAndResbmitAllowed(err, node.Name, tmpl)
+		return woc.checkForbiddenErrorAndResubmitAllowed(err, node.Name, tmpl)
 	}
 
 	return node, err
@@ -2033,12 +2046,12 @@ func (woc *wfOperationCtx) executeScript(nodeName string, templateScope string,
 		executionDeadline:   opts.executionDeadline,
 	})
 	if err != nil {
-		return woc.checkForbiddenErrorAndResbmitAllowed(err, node.Name, tmpl)
+		return woc.checkForbiddenErrorAndResubmitAllowed(err, node.Name, tmpl)
 	}
 	return node, err
 }
 
-func (woc *wfOperationCtx) checkForbiddenErrorAndResbmitAllowed(err error, nodeName string, tmpl *wfv1.Template) (*wfv1.NodeStatus, error) {
+func (woc *wfOperationCtx) checkForbiddenErrorAndResubmitAllowed(err error, nodeName string, tmpl *wfv1.Template) (*wfv1.NodeStatus, error) {
 	if apierr.IsForbidden(err) && isResubmitAllowed(tmpl) {
 		// Our error was most likely caused by a lack of resources. If pod resubmission is allowed, keep the node pending
 		woc.requeue(0)
@@ -2309,7 +2322,7 @@ func (woc *wfOperationCtx) executeResource(nodeName string, templateScope string
 	mainCtr.Command = []string{"argoexec", "resource", tmpl.Resource.Action}
 	_, err = woc.createWorkflowPod(nodeName, *mainCtr, tmpl, &createWorkflowPodOpts{onExitPod: opts.onExitTemplate, executionDeadline: opts.executionDeadline})
 	if err != nil {
-		return woc.checkForbiddenErrorAndResbmitAllowed(err, node.Name, tmpl)
+		return woc.checkForbiddenErrorAndResubmitAllowed(err, node.Name, tmpl)
 	}
 
 	return node, err
@@ -2822,6 +2835,9 @@ func (woc *wfOperationCtx) loadExecutionSpec() (wfv1.TemplateReferenceHolder, wf
 	if entrypoint == "" {
 		entrypoint = woc.wfSpec.Entrypoint
 	}
+
+	woc.volumes = woc.wfSpec.DeepCopy().Volumes
+
 	tmplRef := &wfv1.WorkflowStep{TemplateRef: woc.wf.Spec.WorkflowTemplateRef.ToTemplateRef(entrypoint)}
 
 	if len(woc.wfSpec.Arguments.Parameters) > 0 {
diff --git a/workflow/controller/pod/significant.go b/workflow/controller/pod/significant.go
index f42ec7dec..e9c2b7f7d 100644
--- a/workflow/controller/pod/significant.go
+++ b/workflow/controller/pod/significant.go
@@ -9,6 +9,7 @@ func SignificantPodChange(from *apiv1.Pod, to *apiv1.Pod) bool {
 		from.Status.Phase != to.Status.Phase ||
 		from.Status.Message != to.Status.Message ||
 		from.Status.PodIP != to.Status.PodIP ||
+		from.GetDeletionTimestamp() != to.GetDeletionTimestamp() ||
 		significantContainerStatusesChange(from.Status.ContainerStatuses, to.Status.ContainerStatuses) ||
 		significantContainerStatusesChange(from.Status.InitContainerStatuses, to.Status.InitContainerStatuses)
 }


Test patch:
diff --git a/cmd/argo/commands/list_test.go b/cmd/argo/commands/list_test.go
index dfb87bb17..936279606 100644
--- a/cmd/argo/commands/list_test.go
+++ b/cmd/argo/commands/list_test.go
@@ -5,6 +5,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/argoproj/argo/workflow/common"
+
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -39,6 +41,12 @@ func Test_listWorkflows(t *testing.T) {
 			assert.NotNil(t, workflows)
 		}
 	})
+	t.Run("Resubmitted", func(t *testing.T) {
+		workflows, err := list(&metav1.ListOptions{LabelSelector: common.LabelKeyPreviousWorkflowName}, listFlags{resubmitted: true})
+		if assert.NoError(t, err) {
+			assert.NotNil(t, workflows)
+		}
+	})
 	t.Run("Labels", func(t *testing.T) {
 		workflows, err := list(&metav1.ListOptions{LabelSelector: "foo"}, listFlags{labels: "foo"})
 		if assert.NoError(t, err) {
@@ -70,6 +78,11 @@ func list(listOptions *metav1.ListOptions, flags listFlags) (wfv1.Workflows, err
 	c.On("ListWorkflows", mock.Anything, &workflow.WorkflowListRequest{ListOptions: listOptions}).Return(&wfv1.WorkflowList{Items: wfv1.Workflows{
 		{ObjectMeta: metav1.ObjectMeta{Name: "foo-", CreationTimestamp: metav1.Time{Time: time.Now().Add(-2 * time.Hour)}}, Status: wfv1.WorkflowStatus{FinishedAt: metav1.Time{Time: time.Now().Add(-2 * time.Hour)}}},
 		{ObjectMeta: metav1.ObjectMeta{Name: "bar-", CreationTimestamp: metav1.Time{Time: time.Now()}}},
+		{ObjectMeta: metav1.ObjectMeta{
+			Name:              "baz-",
+			CreationTimestamp: metav1.Time{Time: time.Now().Add(-2 * time.Hour)},
+			Labels:            map[string]string{common.LabelKeyPreviousWorkflowName: "foo-"},
+		}},
 	}}, nil)
 	workflows, err := listWorkflows(context.Background(), c, flags)
 	return workflows, err
diff --git a/test/e2e/argo_server_test.go b/test/e2e/argo_server_test.go
index 53845847f..32478fbd4 100644
--- a/test/e2e/argo_server_test.go
+++ b/test/e2e/argo_server_test.go
@@ -592,6 +592,16 @@ func (s *ArgoServerSuite) TestWorkflowService() {
 			Equal("Stopped with strategy 'Terminate'")
 	})
 
+	s.Run("Resubmit", func() {
+		s.e().PUT("/api/v1/workflows/argo/test/resubmit").
+			WithBytes([]byte(`{"memoized": true}`)).
+			Expect().
+			Status(200).
+			JSON().
+			Path("$.metadata.name").
+			NotNull()
+	})
+
 	s.Run("Delete", func() {
 		s.e().DELETE("/api/v1/workflows/argo/test").
 			Expect().
diff --git a/workflow/common/util_test.go b/workflow/common/util_test.go
index f5b46fb36..0b66a00c7 100644
--- a/workflow/common/util_test.go
+++ b/workflow/common/util_test.go
@@ -3,6 +3,8 @@ package common
 import (
 	"testing"
 
+	"github.com/valyala/fasttemplate"
+
 	"github.com/stretchr/testify/assert"
 	corev1 "k8s.io/api/core/v1"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -84,3 +86,74 @@ func TestDeletePod(t *testing.T) {
 		assert.NoError(t, err)
 	})
 }
+
+func TestNestedReplaceString(t *testing.T) {
+
+	replaceMap := map[string]string{"inputs.parameters.message": "hello world"}
+
+	test := `{{- with secret "{{inputs.parameters.message}}" -}}
+    {{ .Data.data.gitcreds }}
+  {{- end }}`
+	fstTmpl, err := fasttemplate.NewTemplate(test, "{{", "}}")
+	if assert.NoError(t, err) {
+		replacement, err := Replace(fstTmpl, replaceMap, true)
+		if assert.NoError(t, err) {
+			assert.Equal(t, "{{- with secret \"hello world\" -}}\n    {{ .Data.data.gitcreds }}\n  {{- end }}", replacement)
+		}
+	}
+
+	test = `{{- with {{ secret "{{inputs.parameters.message}}" -}}
+    {{ .Data.data.gitcreds }}
+  {{- end }}`
+	fstTmpl, err = fasttemplate.NewTemplate(test, "{{", "}}")
+	if assert.NoError(t, err) {
+		replacement, err := Replace(fstTmpl, replaceMap, true)
+		if assert.NoError(t, err) {
+			assert.Equal(t, "{{- with {{ secret \"hello world\" -}}\n    {{ .Data.data.gitcreds }}\n  {{- end }}", replacement)
+		}
+	}
+
+	test = `{{- with {{ secret "{{inputs.parameters.message}}" -}} }}
+    {{ .Data.data.gitcreds }}
+  {{- end }}`
+	fstTmpl, err = fasttemplate.NewTemplate(test, "{{", "}}")
+	if assert.NoError(t, err) {
+		replacement, err := Replace(fstTmpl, replaceMap, true)
+		if assert.NoError(t, err) {
+			assert.Equal(t, "{{- with {{ secret \"hello world\" -}} }}\n    {{ .Data.data.gitcreds }}\n  {{- end }}", replacement)
+		}
+	}
+
+	test = `{{- with secret "{{inputs.parameters.message}}" -}} }}
+    {{ .Data.data.gitcreds }}
+  {{- end }}`
+	fstTmpl, err = fasttemplate.NewTemplate(test, "{{", "}}")
+	if assert.NoError(t, err) {
+		replacement, err := Replace(fstTmpl, replaceMap, true)
+		if assert.NoError(t, err) {
+			assert.Equal(t, "{{- with secret \"hello world\" -}} }}\n    {{ .Data.data.gitcreds }}\n  {{- end }}", replacement)
+		}
+	}
+
+	test = `{{- with {{ {{ }} secret "{{inputs.parameters.message}}" -}} }}
+    {{ .Data.data.gitcreds }}
+  {{- end }}`
+	fstTmpl, err = fasttemplate.NewTemplate(test, "{{", "}}")
+	if assert.NoError(t, err) {
+		replacement, err := Replace(fstTmpl, replaceMap, true)
+		if assert.NoError(t, err) {
+			assert.Equal(t, "{{- with {{ {{ }} secret \"hello world\" -}} }}\n    {{ .Data.data.gitcreds }}\n  {{- end }}", replacement)
+		}
+	}
+
+	test = `{{- with {{ {{ }} secret "{{does-not-exist}}" -}} }}
+    {{ .Data.data.gitcreds }}
+  {{- end }}`
+	fstTmpl, err = fasttemplate.NewTemplate(test, "{{", "}}")
+	if assert.NoError(t, err) {
+		replacement, err := Replace(fstTmpl, replaceMap, true)
+		if assert.NoError(t, err) {
+			assert.Equal(t, test, replacement)
+		}
+	}
+}
diff --git a/workflow/controller/controller_test.go b/workflow/controller/controller_test.go
index 171a2d065..22c5dd460 100644
--- a/workflow/controller/controller_test.go
+++ b/workflow/controller/controller_test.go
@@ -316,7 +316,8 @@ func TestNamespacedController(t *testing.T) {
 		}, nil
 	})
 
-	_, controller := newController()
+	cancel, controller := newController()
+	defer cancel()
 	controller.kubeclientset = kubernetes.Interface(&kubeClient)
 	controller.cwftmplInformer = nil
 	controller.createClusterWorkflowTemplateInformer(context.TODO())
@@ -332,7 +333,8 @@ func TestClusterController(t *testing.T) {
 		}, nil
 	})
 
-	_, controller := newController()
+	cancel, controller := newController()
+	defer cancel()
 	controller.kubeclientset = kubernetes.Interface(&kubeClient)
 	controller.cwftmplInformer = nil
 	controller.createClusterWorkflowTemplateInformer(context.TODO())
@@ -368,6 +370,8 @@ metadata:
   name: workflow-template-whalesay-template
   namespace: default
 spec:
+  serviceAccountName: my-sa
+  priority: 77
   templates:
   - name: whalesay-template
     inputs:
@@ -377,12 +381,16 @@ spec:
       image: docker/whalesay
       command: [cowsay]
       args: ["{{inputs.parameters.message}}"]
+  volumes:
+  - name: data
+    empty: {}
 `
 
 func TestCheckAndInitWorkflowTmplRef(t *testing.T) {
 	wf := unmarshalWF(wfWithTmplRef)
 	wftmpl := unmarshalWFTmpl(wfTmpl)
-	_, controller := newController(wf, wftmpl)
+	cancel, controller := newController(wf, wftmpl)
+	defer cancel()
 	woc := wfOperationCtx{controller: controller,
 		wf: wf}
 	_, _, err := woc.loadExecutionSpec()
@@ -391,7 +399,8 @@ func TestCheckAndInitWorkflowTmplRef(t *testing.T) {
 }
 
 func TestIsArchivable(t *testing.T) {
-	_, controller := newController()
+	cancel, controller := newController()
+	defer cancel()
 	var lblSelector metav1.LabelSelector
 	lblSelector.MatchLabels = make(map[string]string)
 	lblSelector.MatchLabels["workflows.argoproj.io/archive-strategy"] = "true"
diff --git a/workflow/controller/operator_metrics_test.go b/workflow/controller/operator_metrics_test.go
index 03eba95be..d1cbc4f0e 100644
--- a/workflow/controller/operator_metrics_test.go
+++ b/workflow/controller/operator_metrics_test.go
@@ -4,6 +4,8 @@ import (
 	"fmt"
 	"testing"
 
+	"github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
+
 	"github.com/prometheus/client_golang/prometheus"
 	dto "github.com/prometheus/client_model/go"
 	"github.com/stretchr/testify/assert"
@@ -215,3 +217,97 @@ func TestMetricEmissionSameOperationCreationAndFailure(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Contains(t, metricErrorCounterString, `counter:<value:1 > `)
 }
+
+var testRetryStrategyMetric = `
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  name: workflow-template-whalesay-9pk8f
+spec:
+  arguments: {}
+  entrypoint: whalesay
+  templates:
+  - arguments: {}
+    inputs: {}
+    metadata: {}
+    metrics:
+      prometheus:
+      - counter:
+          value: "1"
+        help: number of times the outer workflow was invoked
+        name: workflow_counter
+    name: whalesay
+    outputs: {}
+    steps:
+    - - arguments:
+          parameters:
+          - name: message
+            value: hello world
+        name: call-whalesay-template
+        template: whalesay-template
+  - arguments: {}
+    container:
+      args:
+      - '{{inputs.parameters.message}}'
+      command:
+      - cowsay
+      image: docker/whalesay
+      name: ""
+      resources: {}
+    inputs:
+      parameters:
+      - name: message
+    metadata: {}
+    metrics:
+      prometheus:
+      - counter:
+          value: "1"
+        help: number of times the template was executed
+        name: template_counter
+    name: whalesay-template
+    outputs: {}
+    retryStrategy:
+      limit: 2
+status:
+  phase: Running
+  startedAt: "2020-07-14T16:26:21Z"
+`
+
+func TestRetryStrategyMetric(t *testing.T) {
+	cancel, controller := newController()
+	defer cancel()
+	wfcset := controller.wfclientset.ArgoprojV1alpha1().Workflows("")
+	wf := unmarshalWF(testRetryStrategyMetric)
+	_, err := wfcset.Create(wf)
+	assert.NoError(t, err)
+	woc := newWorkflowOperationCtx(wf, controller)
+
+	woc.operate()
+
+	// Ensure no metrics have been emitted yet
+	metricErrorDesc := wf.Spec.Templates[0].Metrics.Prometheus[0].GetDesc()
+	assert.Nil(t, controller.metrics.GetCustomMetric(metricErrorDesc))
+	metricErrorDesc = wf.Spec.Templates[1].Metrics.Prometheus[0].GetDesc()
+	assert.Nil(t, controller.metrics.GetCustomMetric(metricErrorDesc))
+
+	// Simulate pod succeeded
+	podNode := woc.wf.Status.Nodes["workflow-template-whalesay-9pk8f-1966833540"]
+	podNode.Phase = v1alpha1.NodeSucceeded
+	woc.wf.Status.Nodes["workflow-template-whalesay-9pk8f-1966833540"] = podNode
+	woc.operate()
+
+	metricErrorDesc = wf.Spec.Templates[0].Metrics.Prometheus[0].GetDesc()
+	assert.NotNil(t, controller.metrics.GetCustomMetric(metricErrorDesc))
+	metricErrorCounter := controller.metrics.GetCustomMetric(metricErrorDesc).(prometheus.Counter)
+	metricErrorCounterString, err := getMetricStringValue(metricErrorCounter)
+	assert.NoError(t, err)
+	assert.Contains(t, metricErrorCounterString, `counter:<value:1 > `)
+
+	metricErrorDesc = wf.Spec.Templates[1].Metrics.Prometheus[0].GetDesc()
+	assert.NotNil(t, controller.metrics.GetCustomMetric(metricErrorDesc))
+	metricErrorCounter = controller.metrics.GetCustomMetric(metricErrorDesc).(prometheus.Counter)
+	metricErrorCounterString, err = getMetricStringValue(metricErrorCounter)
+	assert.NoError(t, err)
+	assert.Contains(t, metricErrorCounterString, `counter:<value:1 > `)
+
+}
diff --git a/workflow/controller/operator_test.go b/workflow/controller/operator_test.go
index 5dd077e14..6a95b2d0b 100644
--- a/workflow/controller/operator_test.go
+++ b/workflow/controller/operator_test.go
@@ -3965,15 +3965,58 @@ func TestCheckForbiddenErrorAndResbmitAllowed(t *testing.T) {
 	forbiddenErr := apierr.NewForbidden(schema.GroupResource{Group: "test", Resource: "test1"}, "test", nil)
 	nonForbiddenErr := apierr.NewBadRequest("badrequest")
 	t.Run("ForbiddenError", func(t *testing.T) {
-		node, err := woc.checkForbiddenErrorAndResbmitAllowed(forbiddenErr, "resubmit-pending-wf", &wf.Spec.Templates[0])
+		node, err := woc.checkForbiddenErrorAndResubmitAllowed(forbiddenErr, "resubmit-pending-wf", &wf.Spec.Templates[0])
 		assert.NotNil(t, node)
 		assert.NoError(t, err)
 		assert.Equal(t, wfv1.NodePending, node.Phase)
 	})
 	t.Run("NonForbiddenError", func(t *testing.T) {
-		node, err := woc.checkForbiddenErrorAndResbmitAllowed(nonForbiddenErr, "resubmit-pending-wf", &wf.Spec.Templates[0])
+		node, err := woc.checkForbiddenErrorAndResubmitAllowed(nonForbiddenErr, "resubmit-pending-wf", &wf.Spec.Templates[0])
 		assert.Error(t, err)
 		assert.Nil(t, node)
 	})
 
 }
+
+func TestResubmitMemoization(t *testing.T) {
+	wf := unmarshalWF(`apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  name: my-wf
+spec:
+  entrypoint: main
+  templates:
+  - name: main
+    container:
+      image: busybox
+status:
+  phase: Failed
+  nodes:
+    my-wf:
+      name: my-wf
+      phase: Failed
+`)
+	wf, err := util.FormulateResubmitWorkflow(wf, true)
+	if assert.NoError(t, err) {
+		cancel, controller := newController(wf)
+		defer cancel()
+		woc := newWorkflowOperationCtx(wf, controller)
+		woc.operate()
+		assert.Equal(t, wfv1.NodePending, woc.wf.Status.Phase)
+		for _, node := range woc.wf.Status.Nodes {
+			switch node.TemplateName {
+			case "main":
+				assert.Equal(t, wfv1.NodePending, node.Phase)
+				assert.False(t, node.StartTime().IsZero())
+				assert.Equal(t, woc.wf.Labels[common.LabelKeyPreviousWorkflowName], "my-wf")
+			case "":
+			default:
+				assert.Fail(t, "invalid template")
+			}
+		}
+		list, err := controller.kubeclientset.CoreV1().Pods("").List(metav1.ListOptions{})
+		if assert.NoError(t, err) {
+			assert.Len(t, list.Items, 1)
+		}
+	}
+}
diff --git a/workflow/controller/operator_workflow_template_ref_test.go b/workflow/controller/operator_workflow_template_ref_test.go
index 73788b3b2..c88a23a60 100644
--- a/workflow/controller/operator_workflow_template_ref_test.go
+++ b/workflow/controller/operator_workflow_template_ref_test.go
@@ -11,15 +11,16 @@ import (
 )
 
 func TestWorkflowTemplateRef(t *testing.T) {
-	wf := unmarshalWF(wfWithTmplRef)
-	wftmpl := unmarshalWFTmpl(wfTmpl)
-
-	t.Run("ExecuteWorkflowWithTmplRef", func(t *testing.T) {
-		_, controller := newController(wf, wftmpl)
-		woc := newWorkflowOperationCtx(wf, controller)
-		woc.operate()
-		assert.Equal(t, &wftmpl.Spec.WorkflowSpec, woc.wfSpec)
-	})
+	cancel, controller := newController(unmarshalWF(wfWithTmplRef), unmarshalWFTmpl(wfTmpl))
+	defer cancel()
+	woc := newWorkflowOperationCtx(unmarshalWF(wfWithTmplRef), controller)
+	woc.operate()
+	assert.Equal(t, &unmarshalWFTmpl(wfTmpl).Spec.WorkflowSpec, woc.wfSpec)
+	// verify we copy these values
+	assert.Len(t, woc.volumes, 1, "volumes from workflow template")
+	// and these
+	assert.Equal(t, "my-sa", woc.globalParams["workflow.serviceAccountName"])
+	assert.Equal(t, "77", woc.globalParams["workflow.priority"])
 }
 
 func TestWorkflowTemplateRefWithArgs(t *testing.T) {
@@ -35,7 +36,8 @@ func TestWorkflowTemplateRefWithArgs(t *testing.T) {
 			},
 		}
 		wf.Spec.Arguments.Parameters = util.MergeParameters(wf.Spec.Arguments.Parameters, args)
-		_, controller := newController(wf, wftmpl)
+		cancel, controller := newController(wf, wftmpl)
+		defer cancel()
 		woc := newWorkflowOperationCtx(wf, controller)
 		woc.operate()
 		assert.Equal(t, "test", woc.globalParams["workflow.parameters.param1"])
@@ -55,7 +57,8 @@ func TestWorkflowTemplateRefWithWorkflowTemplateArgs(t *testing.T) {
 			},
 		}
 		wftmpl.Spec.Arguments.Parameters = util.MergeParameters(wf.Spec.Arguments.Parameters, args)
-		_, controller := newController(wf, wftmpl)
+		cancel, controller := newController(wf, wftmpl)
+		defer cancel()
 		woc := newWorkflowOperationCtx(wf, controller)
 		woc.operate()
 		assert.Equal(t, "test", woc.globalParams["workflow.parameters.param1"])
@@ -122,7 +125,8 @@ status:
 func TestWorkflowTemplateRefGetFromStored(t *testing.T) {
 	wf := unmarshalWF(wfWithStatus)
 	t.Run("ProcessWFWithStoredWFT", func(t *testing.T) {
-		_, controller := newController(wf)
+		cancel, controller := newController(wf)
+		defer cancel()
 		woc := newWorkflowOperationCtx(wf, controller)
 		_, execArgs, err := woc.loadExecutionSpec()
 		assert.NoError(t, err)
@@ -147,7 +151,8 @@ spec:
 func TestWorkflowTemplateRefInvalidWF(t *testing.T) {
 	wf := unmarshalWF(invalidWF)
 	t.Run("ProcessWFWithStoredWFT", func(t *testing.T) {
-		_, controller := newController(wf)
+		cancel, controller := newController(wf)
+		defer cancel()
 		woc := newWorkflowOperationCtx(wf, controller)
 		_, _, err := woc.loadExecutionSpec()
 		assert.Error(t, err)
diff --git a/workflow/controller/pod/significant_test.go b/workflow/controller/pod/significant_test.go
index eae08ea55..534d1b7fa 100644
--- a/workflow/controller/pod/significant_test.go
+++ b/workflow/controller/pod/significant_test.go
@@ -5,10 +5,16 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
 func Test_SgnificantPodChange(t *testing.T) {
 	assert.False(t, SignificantPodChange(&corev1.Pod{}, &corev1.Pod{}), "No change")
+
+	t.Run("DeletionTimestamp", func(t *testing.T) {
+		now := metav1.Now()
+		assert.True(t, SignificantPodChange(&corev1.Pod{}, &corev1.Pod{ObjectMeta: metav1.ObjectMeta{DeletionTimestamp: &now}}), "deletion timestamp change")
+	})
 	t.Run("Spec", func(t *testing.T) {
 		assert.True(t, SignificantPodChange(&corev1.Pod{}, &corev1.Pod{Spec: corev1.PodSpec{NodeName: "from"}}), "Node name change")