Repository: argoproj/argo
Pull request #2573 (2020-04-02 19:08:25)

refactor: Refactor Template context interfaces
Closes: #2570 

codecov[bot]: # [Codecov](https://codecov.io/gh/argoproj/argo/pull/2573?src=pr&el=h1) Report
> :exclamation: No coverage uploaded for pull request base (`master@cb739a6`). [Click here to learn what that means](https://docs.codecov.io/docs/error-reference#section-missing-base-commit).
> The diff coverage is `44.11%`.

[![Impacted file tree graph](https://codecov.io/gh/argoproj/argo/pull/2573/graphs/tree.svg?width=650&token=4c3DPSTq8o&height=150&src=pr)](https://codecov.io/gh/argoproj/argo/pull/2573?src=pr&el=tree)

```diff
@@            Coverage Diff            @@
##             master    #2573   +/-   ##
=========================================
  Coverage          ?   11.11%           
=========================================
  Files             ?       83           
  Lines             ?    32605           
  Branches          ?        0           
=========================================
  Hits              ?     3624           
  Misses            ?    28482           
  Partials          ?      499
```


| [Impacted Files](https://codecov.io/gh/argoproj/argo/pull/2573?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [pkg/apis/workflow/v1alpha1/workflow\_types.go](https://codecov.io/gh/argoproj/argo/pull/2573/diff?src=pr&el=tree#diff-cGtnL2FwaXMvd29ya2Zsb3cvdjFhbHBoYTEvd29ya2Zsb3dfdHlwZXMuZ28=) | `11.14% <0%> (ø)` | |
| [...rkflow/v1alpha1/cluster\_workflow\_template\_types.go](https://codecov.io/gh/argoproj/argo/pull/2573/diff?src=pr&el=tree#diff-cGtnL2FwaXMvd29ya2Zsb3cvdjFhbHBoYTEvY2x1c3Rlcl93b3JrZmxvd190ZW1wbGF0ZV90eXBlcy5nbw==) | `0% <0%> (ø)` | |
| [.../apis/workflow/v1alpha1/workflow\_template\_types.go](https://codecov.io/gh/argoproj/argo/pull/2573/diff?src=pr&el=tree#diff-cGtnL2FwaXMvd29ya2Zsb3cvdjFhbHBoYTEvd29ya2Zsb3dfdGVtcGxhdGVfdHlwZXMuZ28=) | `40% <0%> (ø)` | |
| [workflow/common/util.go](https://codecov.io/gh/argoproj/argo/pull/2573/diff?src=pr&el=tree#diff-d29ya2Zsb3cvY29tbW9uL3V0aWwuZ28=) | `5.83% <0%> (ø)` | |
| [workflow/validate/validate.go](https://codecov.io/gh/argoproj/argo/pull/2573/diff?src=pr&el=tree#diff-d29ya2Zsb3cvdmFsaWRhdGUvdmFsaWRhdGUuZ28=) | `74.13% <100%> (ø)` | |
| [workflow/controller/dag.go](https://codecov.io/gh/argoproj/argo/pull/2573/diff?src=pr&el=tree#diff-d29ya2Zsb3cvY29udHJvbGxlci9kYWcuZ28=) | `51.09% <28.57%> (ø)` | |
| [workflow/templateresolution/context.go](https://codecov.io/gh/argoproj/argo/pull/2573/diff?src=pr&el=tree#diff-d29ya2Zsb3cvdGVtcGxhdGVyZXNvbHV0aW9uL2NvbnRleHQuZ28=) | `60% <32.55%> (ø)` | |
| [workflow/controller/steps.go](https://codecov.io/gh/argoproj/argo/pull/2573/diff?src=pr&el=tree#diff-d29ya2Zsb3cvY29udHJvbGxlci9zdGVwcy5nbw==) | `65.34% <66.66%> (ø)` | |
| [workflow/controller/operator.go](https://codecov.io/gh/argoproj/argo/pull/2573/diff?src=pr&el=tree#diff-d29ya2Zsb3cvY29udHJvbGxlci9vcGVyYXRvci5nbw==) | `60.86% <76%> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/argoproj/argo/pull/2573?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/argoproj/argo/pull/2573?src=pr&el=footer). Last update [cb739a6...3048a67](https://codecov.io/gh/argoproj/argo/pull/2573?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).


Patch:
diff --git a/go.mod b/go.mod
index f651e7425..912684138 100644
--- a/go.mod
+++ b/go.mod
@@ -80,7 +80,7 @@ require (
 	golang.org/x/crypto v0.0.0-20200311171314-f7b00557c8c4
 	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-20200407191807-cd5a53e07f8a // indirect
+	golang.org/x/tools v0.0.0-20200408014516-4d14fc9c00ce // 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 efafb7d5e..9012a886f 100644
--- a/go.sum
+++ b/go.sum
@@ -618,8 +618,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-20200407191807-cd5a53e07f8a h1:3CGHZdxu1QXyxpZvKA3QHjIyj6ia66IIHQ3O0lKHgyQ=
-golang.org/x/tools v0.0.0-20200407191807-cd5a53e07f8a/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
+golang.org/x/tools v0.0.0-20200408014516-4d14fc9c00ce h1:Zc5zydnQGVfragUHga1rdKr1tk2yKSlLIK3winWj5sc=
+golang.org/x/tools v0.0.0-20200408014516-4d14fc9c00ce/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
 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/apis/workflow/v1alpha1/cluster_workflow_template_types.go b/pkg/apis/workflow/v1alpha1/cluster_workflow_template_types.go
index b8322d55b..4374ca9d6 100644
--- a/pkg/apis/workflow/v1alpha1/cluster_workflow_template_types.go
+++ b/pkg/apis/workflow/v1alpha1/cluster_workflow_template_types.go
@@ -39,7 +39,7 @@ type ClusterWorkflowTemplateList struct {
 	Items           ClusterWorkflowTemplates `json:"items" protobuf:"bytes,2,rep,name=items"`
 }
 
-var _ TemplateGetter = &ClusterWorkflowTemplate{}
+var _ TemplateHolder = &ClusterWorkflowTemplate{}
 
 // GetTemplateByName retrieves a defined template by its name
 func (cwftmpl *ClusterWorkflowTemplate) GetTemplateByName(name string) *Template {
@@ -51,12 +51,12 @@ func (cwftmpl *ClusterWorkflowTemplate) GetTemplateByName(name string) *Template
 	return nil
 }
 
-// GetTemplateScope returns the template scope of workflow template.
-func (cwftmpl *ClusterWorkflowTemplate) GetTemplateScope() string {
-	return "cluster/" + cwftmpl.Name
+// GetResourceScope returns the template scope of workflow template.
+func (cwftmpl *ClusterWorkflowTemplate) GetResourceScope() ResourceScope {
+	return ResourceScopeCluster
 }
 
-// GetAllTemplates returns the list of templates of cluster workflow template
-func (cwftmpl *ClusterWorkflowTemplate) GetAllTemplates() []Template {
+// GetTemplates returns the list of templates of cluster workflow template
+func (cwftmpl *ClusterWorkflowTemplate) GetTemplates() []Template {
 	return cwftmpl.Spec.Templates
 }
diff --git a/pkg/apis/workflow/v1alpha1/common.go b/pkg/apis/workflow/v1alpha1/common.go
index da7ce4cfa..8651f7a29 100644
--- a/pkg/apis/workflow/v1alpha1/common.go
+++ b/pkg/apis/workflow/v1alpha1/common.go
@@ -4,31 +4,26 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 )
 
-// TemplateGetter is an interface to get templates.
-type TemplateGetter interface {
+type ResourceScope string
+
+const (
+	ResourceScopeLocal      ResourceScope = "local"
+	ResourceScopeNamespaced ResourceScope = "namespaced"
+	ResourceScopeCluster    ResourceScope = "cluster"
+)
+
+// TemplateHolder is an object that holds templates; e.g. Workflow, WorkflowTemplate, and ClusterWorkflowTemplate
+type TemplateHolder interface {
 	GetNamespace() string
 	GetName() string
 	GroupVersionKind() schema.GroupVersionKind
 	GetTemplateByName(name string) *Template
-	GetTemplateScope() string
-	GetAllTemplates() []Template
+	GetResourceScope() ResourceScope
+	GetTemplates() []Template
 }
 
-// TemplateHolder is an interface for holders of templates.
-type TemplateHolder interface {
+// TemplateReferenceHolder is an object that holds a reference to other templates; e.g. WorkflowStep, DAGTask, and NodeStatus
+type TemplateReferenceHolder interface {
 	GetTemplateName() string
 	GetTemplateRef() *TemplateRef
-	IsResolvable() bool
-}
-
-// TemplateStorage is an interface of template storage getter and setter.
-type TemplateStorage interface {
-	GetStoredTemplate(templateScope string, holder TemplateHolder) *Template
-	SetStoredTemplate(templateScope string, holder TemplateHolder, tmpl *Template) (bool, error)
-}
-
-// WorkflowTemplateInterface is an simplified TemplateGetter
-type WorkflowTemplateInterface interface {
-	GetTemplateByName(name string) *Template
-	GetTemplateScope() string
 }
diff --git a/pkg/apis/workflow/v1alpha1/workflow_template_types.go b/pkg/apis/workflow/v1alpha1/workflow_template_types.go
index 671fd64dc..c022251d3 100644
--- a/pkg/apis/workflow/v1alpha1/workflow_template_types.go
+++ b/pkg/apis/workflow/v1alpha1/workflow_template_types.go
@@ -38,7 +38,7 @@ type WorkflowTemplateList struct {
 	Items           WorkflowTemplates `json:"items" protobuf:"bytes,2,rep,name=items"`
 }
 
-var _ TemplateGetter = &WorkflowTemplate{}
+var _ TemplateHolder = &WorkflowTemplate{}
 
 // WorkflowTemplateSpec is a spec of WorkflowTemplate.
 type WorkflowTemplateSpec struct {
@@ -55,12 +55,12 @@ func (wftmpl *WorkflowTemplate) GetTemplateByName(name string) *Template {
 	return nil
 }
 
-// GetTemplateScope returns the template scope of workflow template.
-func (wftmpl *WorkflowTemplate) GetTemplateScope() string {
-	return "namespaced/" + wftmpl.Name
+// GetResourceScope returns the template scope of workflow template.
+func (wftmpl *WorkflowTemplate) GetResourceScope() ResourceScope {
+	return ResourceScopeNamespaced
 }
 
-// GetAllTemplates returns the list of templates of workflow template
-func (wftmpl *WorkflowTemplate) GetAllTemplates() []Template {
+// GetTemplates returns the list of templates of workflow template
+func (wftmpl *WorkflowTemplate) GetTemplates() []Template {
 	return wftmpl.Spec.Templates
 }
diff --git a/pkg/apis/workflow/v1alpha1/workflow_types.go b/pkg/apis/workflow/v1alpha1/workflow_types.go
index cbf7c47bd..f8a395e23 100644
--- a/pkg/apis/workflow/v1alpha1/workflow_types.go
+++ b/pkg/apis/workflow/v1alpha1/workflow_types.go
@@ -107,8 +107,7 @@ type WorkflowList struct {
 	Items           Workflows `json:"items" protobuf:"bytes,2,opt,name=items"`
 }
 
-var _ TemplateGetter = &Workflow{}
-var _ TemplateStorage = &Workflow{}
+var _ TemplateHolder = &Workflow{}
 
 // TTLStrategy is the strategy for the time to live depending on if the workflow succeeded or failed
 type TTLStrategy struct {
@@ -465,9 +464,10 @@ type Template struct {
 	Metrics *Metrics `json:"metrics,omitempty" protobuf:"bytes,35,opt,name=metrics"`
 }
 
-var _ TemplateHolder = &Template{}
+// DEPRECATED: Templates should not be used as TemplateReferenceHolder
+var _ TemplateReferenceHolder = &Template{}
 
-// DEPRECATED: Templates should not be used at TemplateHolders
+// DEPRECATED: Templates should not be used as TemplateReferenceHolder
 func (tmpl *Template) GetTemplateName() string {
 	if tmpl.Template != "" {
 		return tmpl.Template
@@ -476,16 +476,11 @@ func (tmpl *Template) GetTemplateName() string {
 	}
 }
 
-// DEPRECATED: Templates should not be used at TemplateHolders
+// DEPRECATED: Templates should not be used as TemplateReferenceHolder
 func (tmpl *Template) GetTemplateRef() *TemplateRef {
 	return tmpl.TemplateRef
 }
 
-// DEPRECATED: Templates should not be used at TemplateHolders
-func (tmpl *Template) IsResolvable() bool {
-	return tmpl.Template != "" || tmpl.TemplateRef != nil
-}
-
 // GetBaseTemplate returns a base template content.
 func (tmpl *Template) GetBaseTemplate() *Template {
 	baseTemplate := tmpl.DeepCopy()
@@ -748,7 +743,7 @@ type WorkflowStep struct {
 	OnExit string `json:"onExit,omitempty" protobuf:"bytes,11,opt,name=onExit"`
 }
 
-var _ TemplateHolder = &WorkflowStep{}
+var _ TemplateReferenceHolder = &WorkflowStep{}
 
 func (step *WorkflowStep) GetTemplateName() string {
 	return step.Template
@@ -758,10 +753,6 @@ func (step *WorkflowStep) GetTemplateRef() *TemplateRef {
 	return step.TemplateRef
 }
 
-func (step *WorkflowStep) IsResolvable() bool {
-	return true
-}
-
 // Sequence expands a workflow step into numeric range
 type Sequence struct {
 	// Count is number of elements in the sequence (default: 0). Not to be used with end
@@ -1203,7 +1194,21 @@ func (n NodeStatus) CanRetry() bool {
 	return n.Completed() && !n.Successful()
 }
 
-var _ TemplateHolder = &NodeStatus{}
+func (n NodeStatus) GetTemplateScope() (ResourceScope, string) {
+	// For compatibility: an empty TemplateScope is a local scope
+	if n.TemplateScope == "" {
+		return ResourceScopeLocal, ""
+	}
+	split := strings.Split(n.TemplateScope, "/")
+	// For compatibility: an unspecified ResourceScope in a TemplateScope is a namespaced scope
+	if len(split) == 1 {
+		return ResourceScopeNamespaced, split[0]
+	}
+	resourceScope, resourceName := split[0], split[1]
+	return ResourceScope(resourceScope), resourceName
+}
+
+var _ TemplateReferenceHolder = &NodeStatus{}
 
 func (n *NodeStatus) GetTemplateName() string {
 	return n.TemplateName
@@ -1213,10 +1218,6 @@ func (n *NodeStatus) GetTemplateRef() *TemplateRef {
 	return n.TemplateRef
 }
 
-func (n *NodeStatus) IsResolvable() bool {
-	return true
-}
-
 // IsActiveSuspendNode returns whether this node is an active suspend node
 func (n *NodeStatus) IsActiveSuspendNode() bool {
 	return n.Type == NodeTypeSuspend && n.Phase == NodeRunning
@@ -1579,7 +1580,7 @@ type DAGTask struct {
 	OnExit string `json:"onExit,omitempty" protobuf:"bytes,11,opt,name=onExit"`
 }
 
-var _ TemplateHolder = &DAGTask{}
+var _ TemplateReferenceHolder = &DAGTask{}
 
 func (t *DAGTask) GetTemplateName() string {
 	return t.Template
@@ -1589,10 +1590,6 @@ func (t *DAGTask) GetTemplateRef() *TemplateRef {
 	return t.TemplateRef
 }
 
-func (t *DAGTask) IsResolvable() bool {
-	return true
-}
-
 // SuspendTemplate is a template subtype to suspend a workflow at a predetermined point in time
 type SuspendTemplate struct {
 	// Duration is the seconds to wait before automatically resuming a template
@@ -1680,13 +1677,13 @@ func (wf *Workflow) GetTemplateByName(name string) *Template {
 	return nil
 }
 
-// GetTemplateScope returns the template scope of workflow.
-func (wf *Workflow) GetTemplateScope() string {
-	return ""
+// GetResourceScope returns the template scope of workflow.
+func (wf *Workflow) GetResourceScope() ResourceScope {
+	return ResourceScopeLocal
 }
 
-// GetAllTemplates returns the list of templates of workflow.
-func (wf *Workflow) GetAllTemplates() []Template {
+// GetTemplates returns the list of templates of workflow.
+func (wf *Workflow) GetTemplates() []Template {
 	return wf.Spec.Templates
 }
 
@@ -1701,26 +1698,26 @@ func (wf *Workflow) NodeID(name string) string {
 }
 
 // GetStoredTemplate retrieves a template from stored templates of the workflow.
-func (wf *Workflow) GetStoredTemplate(templateScope string, holder TemplateHolder) *Template {
-	tmplID := wf.getStoredTemplateName(templateScope, holder)
-	if tmplID == "" {
+func (wf *Workflow) GetStoredTemplate(scope ResourceScope, resourceName string, caller TemplateReferenceHolder) *Template {
+	tmplID, storageNeeded := resolveTemplateReference(scope, resourceName, caller)
+	if !storageNeeded {
+		// Local templates aren't stored
 		return nil
 	}
-	tmpl, ok := wf.Status.StoredTemplates[tmplID]
-	if !ok {
-		return nil
+	if tmpl, ok := wf.Status.StoredTemplates[tmplID]; ok {
+		return &tmpl
 	}
-	return &tmpl
+	return nil
 }
 
 // SetStoredTemplate stores a new template in stored templates of the workflow.
-func (wf *Workflow) SetStoredTemplate(templateScope string, holder TemplateHolder, tmpl *Template) (bool, error) {
-	tmplID := wf.getStoredTemplateName(templateScope, holder)
-	if tmplID == "" {
+func (wf *Workflow) SetStoredTemplate(scope ResourceScope, resourceName string, caller TemplateReferenceHolder, tmpl *Template) (bool, error) {
+	tmplID, storageNeeded := resolveTemplateReference(scope, resourceName, caller)
+	if !storageNeeded {
+		// Don't need to store local templates
 		return false, nil
 	}
-	_, ok := wf.Status.StoredTemplates[tmplID]
-	if !ok {
+	if _, ok := wf.Status.StoredTemplates[tmplID]; !ok {
 		if wf.Status.StoredTemplates == nil {
 			wf.Status.StoredTemplates = map[string]Template{}
 		}
@@ -1730,16 +1727,24 @@ func (wf *Workflow) SetStoredTemplate(templateScope string, holder TemplateHolde
 	return false, nil
 }
 
-// getStoredTemplateName returns the stored template name of a given template holder on the template scope.
-func (wf *Workflow) getStoredTemplateName(templateScope string, holder TemplateHolder) string {
-	tmplRef := holder.GetTemplateRef()
+// resolveTemplateReference resolves the stored template name of a given template holder on the template scope and determines
+// if it should be stored
+func resolveTemplateReference(callerScope ResourceScope, resourceName string, caller TemplateReferenceHolder) (string, bool) {
+	tmplRef := caller.GetTemplateRef()
 	if tmplRef != nil {
-		return fmt.Sprintf("%s/%s", tmplRef.Name, tmplRef.Template)
-	} else if templateScope != "" {
-		return fmt.Sprintf("%s/%s", templateScope, holder.GetTemplateName())
+		// We are calling an external WorkflowTemplate or ClusterWorkflowTemplate. Template storage is needed
+		// We need to determine if we're calling a WorkflowTemplate or a ClusterWorkflowTemplate
+		referenceScope := ResourceScopeNamespaced
+		if tmplRef.ClusterScope {
+			referenceScope = ResourceScopeCluster
+		}
+		return fmt.Sprintf("%s/%s/%s", referenceScope, tmplRef.Name, tmplRef.Template), true
+	} else if callerScope != ResourceScopeLocal {
+		// Either a WorkflowTemplate or a ClusterWorkflowTemplate is calling a template inside itself. Template storage is needed
+		return fmt.Sprintf("%s/%s/%s", callerScope, resourceName, caller.GetTemplateName()), true
 	} else {
-		// Do not store workflow-local templates.
-		return ""
+		// A Workflow is calling a template inside itself. Template storage is not needed
+		return "", false
 	}
 }
 
diff --git a/workflow/common/util.go b/workflow/common/util.go
index 387ee6bf8..4a4b1ca3f 100644
--- a/workflow/common/util.go
+++ b/workflow/common/util.go
@@ -653,13 +653,13 @@ func MergeReferredTemplate(tmpl *wfv1.Template, referred *wfv1.Template) (*wfv1.
 	return newTmpl, nil
 }
 
-// GetTemplateGetterString returns string of TemplateGetter.
-func GetTemplateGetterString(getter wfv1.TemplateGetter) string {
+// GetTemplateGetterString returns string of TemplateHolder.
+func GetTemplateGetterString(getter wfv1.TemplateHolder) string {
 	return fmt.Sprintf("%T (namespace=%s,name=%s)", getter, getter.GetNamespace(), getter.GetName())
 }
 
-// GetTemplateHolderString returns string of TemplateHolder.
-func GetTemplateHolderString(tmplHolder wfv1.TemplateHolder) string {
+// GetTemplateHolderString returns string of TemplateReferenceHolder.
+func GetTemplateHolderString(tmplHolder wfv1.TemplateReferenceHolder) string {
 	tmplName := tmplHolder.GetTemplateName()
 	tmplRef := tmplHolder.GetTemplateRef()
 	if tmplRef != nil {
diff --git a/workflow/controller/dag.go b/workflow/controller/dag.go
index a6ff6691f..47ba9244b 100644
--- a/workflow/controller/dag.go
+++ b/workflow/controller/dag.go
@@ -194,7 +194,7 @@ func (d *dagContext) hasMoreRetries(node *wfv1.NodeStatus) bool {
 	if !ok {
 		return false
 	}
-	_, tmpl, err := d.tmplCtx.ResolveTemplate(&childNode)
+	_, tmpl, _, err := d.tmplCtx.ResolveTemplate(&childNode)
 	if err != nil {
 		return false
 	}
@@ -204,7 +204,7 @@ func (d *dagContext) hasMoreRetries(node *wfv1.NodeStatus) bool {
 	return true
 }
 
-func (woc *wfOperationCtx) executeDAG(nodeName string, tmplCtx *templateresolution.Context, templateScope string, tmpl *wfv1.Template, orgTmpl wfv1.TemplateHolder, opts *executeTemplateOpts) (*wfv1.NodeStatus, error) {
+func (woc *wfOperationCtx) executeDAG(nodeName string, tmplCtx *templateresolution.Context, templateScope string, tmpl *wfv1.Template, orgTmpl wfv1.TemplateReferenceHolder, opts *executeTemplateOpts) (*wfv1.NodeStatus, error) {
 	node := woc.getNodeByName(nodeName)
 	if node == nil {
 		node = woc.initializeExecutableNode(nodeName, wfv1.NodeTypeDAG, templateScope, tmpl, orgTmpl, opts.boundaryID, wfv1.NodeRunning)
@@ -376,7 +376,7 @@ func (woc *wfOperationCtx) executeDAGTask(dagCtx *dagContext, taskName string) {
 	}
 
 	// The template scope of this dag.
-	dagTemplateScope := dagCtx.tmplCtx.GetCurrentTemplateBase().GetTemplateScope()
+	dagTemplateScope := dagCtx.tmplCtx.GetTemplateScope()
 
 	// First resolve/substitute params/artifacts from our dependencies
 	newTask, err := woc.resolveDependencyReferences(dagCtx, task)
@@ -468,10 +468,15 @@ func (woc *wfOperationCtx) buildLocalScopeFromTask(dagCtx *dagContext, task *wfv
 					ancestorNodes = append(ancestorNodes, node)
 				}
 			}
-			_, tmpl, err := dagCtx.tmplCtx.ResolveTemplate(ancestorNode)
+			_, tmpl, templateStored, err := dagCtx.tmplCtx.ResolveTemplate(ancestorNode)
 			if err != nil {
 				return nil, errors.InternalWrapError(err)
 			}
+			// A new template was stored during resolution, persist it
+			if templateStored {
+				woc.updated = true
+			}
+
 			err = woc.processAggregateNodeOutputs(tmpl, &scope, prefix, ancestorNodes)
 			if err != nil {
 				return nil, errors.InternalWrapError(err)
diff --git a/workflow/controller/operator.go b/workflow/controller/operator.go
index ec8617e50..3eb75370e 100644
--- a/workflow/controller/operator.go
+++ b/workflow/controller/operator.go
@@ -89,8 +89,6 @@ type wfOperationCtx struct {
 	preExecutionNodePhases map[string]wfv1.NodePhase
 }
 
-var _ wfv1.TemplateStorage = &wfOperationCtx{}
-
 var (
 	// ErrDeadlineExceeded indicates the operation exceeded its deadline for execution
 	ErrDeadlineExceeded = errors.New(errors.CodeTimeout, "Deadline exceeded")
@@ -273,7 +271,7 @@ func (woc *wfOperationCtx) operate() {
 	}
 
 	// Create a starting template context.
-	tmplCtx, err := woc.createTemplateContext("")
+	tmplCtx, err := woc.createTemplateContext(wfv1.ResourceScopeLocal, "")
 	if err != nil {
 		msg := fmt.Sprintf("Failed to create a template context: %+v", err)
 		woc.log.Errorf(msg)
@@ -1323,17 +1321,21 @@ type executeTemplateOpts struct {
 // for the created node (if created). Nodes may not be created if parallelism or deadline exceeded.
 // nodeName is the name to be used as the name of the node, and boundaryID indicates which template
 // boundary this node belongs to.
-func (woc *wfOperationCtx) executeTemplate(nodeName string, orgTmpl wfv1.TemplateHolder, tmplCtx *templateresolution.Context, args wfv1.Arguments, opts *executeTemplateOpts) (*wfv1.NodeStatus, error) {
+func (woc *wfOperationCtx) executeTemplate(nodeName string, orgTmpl wfv1.TemplateReferenceHolder, tmplCtx *templateresolution.Context, args wfv1.Arguments, opts *executeTemplateOpts) (*wfv1.NodeStatus, error) {
 	woc.log.Debugf("Evaluating node %s: template: %s, boundaryID: %s", nodeName, common.GetTemplateHolderString(orgTmpl), opts.boundaryID)
 
 	node := woc.getNodeByName(nodeName)
 
 	// Set templateScope from which the template resolution starts.
-	templateScope := tmplCtx.GetCurrentTemplateBase().GetTemplateScope()
-	newTmplCtx, resolvedTmpl, err := tmplCtx.ResolveTemplate(orgTmpl)
+	templateScope := tmplCtx.GetTemplateScope()
+	newTmplCtx, resolvedTmpl, templateStored, err := tmplCtx.ResolveTemplate(orgTmpl)
 	if err != nil {
 		return woc.initializeNodeOrMarkError(node, nodeName, wfv1.NodeTypeSkipped, templateScope, orgTmpl, opts.boundaryID, err), err
 	}
+	// A new template was stored during resolution, persist it
+	if templateStored {
+		woc.updated = true
+	}
 
 	if node != nil {
 		if node.Completed() {
@@ -1583,7 +1585,7 @@ func (woc *wfOperationCtx) markWorkflowError(err error, markCompleted bool) {
 var stepsOrDagSeparator = regexp.MustCompile(`^(\[\d+\])?\.`)
 
 // initializeExecutableNode initializes a node and stores the template.
-func (woc *wfOperationCtx) initializeExecutableNode(nodeName string, nodeType wfv1.NodeType, templateScope string, executeTmpl *wfv1.Template, orgTmpl wfv1.TemplateHolder, boundaryID string, phase wfv1.NodePhase, messages ...string) *wfv1.NodeStatus {
+func (woc *wfOperationCtx) initializeExecutableNode(nodeName string, nodeType wfv1.NodeType, templateScope string, executeTmpl *wfv1.Template, orgTmpl wfv1.TemplateReferenceHolder, boundaryID string, phase wfv1.NodePhase, messages ...string) *wfv1.NodeStatus {
 	node := woc.initializeNode(nodeName, nodeType, templateScope, orgTmpl, boundaryID, phase)
 
 	// Set the input values to the node.
@@ -1599,14 +1601,14 @@ func (woc *wfOperationCtx) initializeExecutableNode(nodeName string, nodeType wf
 }
 
 // initializeNodeOrMarkError initializes an error node or mark a node if it already exists.
-func (woc *wfOperationCtx) initializeNodeOrMarkError(node *wfv1.NodeStatus, nodeName string, nodeType wfv1.NodeType, templateScope string, orgTmpl wfv1.TemplateHolder, boundaryID string, err error) *wfv1.NodeStatus {
+func (woc *wfOperationCtx) initializeNodeOrMarkError(node *wfv1.NodeStatus, nodeName string, nodeType wfv1.NodeType, templateScope string, orgTmpl wfv1.TemplateReferenceHolder, boundaryID string, err error) *wfv1.NodeStatus {
 	if node != nil {
 		return woc.markNodeError(nodeName, err)
 	}
 	return woc.initializeNode(nodeName, wfv1.NodeTypeSkipped, templateScope, orgTmpl, boundaryID, wfv1.NodeError, err.Error())
 }
 
-func (woc *wfOperationCtx) initializeNode(nodeName string, nodeType wfv1.NodeType, templateScope string, orgTmpl wfv1.TemplateHolder, boundaryID string, phase wfv1.NodePhase, messages ...string) *wfv1.NodeStatus {
+func (woc *wfOperationCtx) initializeNode(nodeName string, nodeType wfv1.NodeType, templateScope string, orgTmpl wfv1.TemplateReferenceHolder, boundaryID string, phase wfv1.NodePhase, messages ...string) *wfv1.NodeStatus {
 	woc.log.Debugf("Initializing node %s: template: %s, boundaryID: %s", nodeName, common.GetTemplateHolderString(orgTmpl), boundaryID)
 
 	nodeID := woc.wf.NodeID(nodeName)
@@ -1616,14 +1618,15 @@ func (woc *wfOperationCtx) initializeNode(nodeName string, nodeType wfv1.NodeTyp
 	}
 
 	node := wfv1.NodeStatus{
-		ID:           nodeID,
-		Name:         nodeName,
-		TemplateName: orgTmpl.GetTemplateName(),
-		TemplateRef:  orgTmpl.GetTemplateRef(),
-		Type:         nodeType,
-		BoundaryID:   boundaryID,
-		Phase:        phase,
-		StartedAt:    metav1.Time{Time: time.Now().UTC()},
+		ID:            nodeID,
+		Name:          nodeName,
+		TemplateName:  orgTmpl.GetTemplateName(),
+		TemplateRef:   orgTmpl.GetTemplateRef(),
+		TemplateScope: templateScope,
+		Type:          nodeType,
+		BoundaryID:    boundaryID,
+		Phase:         phase,
+		StartedAt:     metav1.Time{Time: time.Now().UTC()},
 	}
 
 	if boundaryNode, ok := woc.wf.Status.Nodes[boundaryID]; ok {
@@ -1635,8 +1638,6 @@ func (woc *wfOperationCtx) initializeNode(nodeName string, nodeType wfv1.NodeTyp
 		node.DisplayName = nodeName
 	}
 
-	node.TemplateScope = templateScope
-
 	if node.Completed() && node.FinishedAt.IsZero() {
 		node.FinishedAt = node.StartedAt
 	}
@@ -1735,14 +1736,19 @@ func (woc *wfOperationCtx) checkParallelism(tmpl *wfv1.Template, node *wfv1.Node
 			if !ok {
 				return errors.InternalError("boundaryNode not found")
 			}
-			tmplCtx, err := woc.createTemplateContext(boundaryNode.TemplateScope)
+			tmplCtx, err := woc.createTemplateContext(boundaryNode.GetTemplateScope())
 			if err != nil {
 				return err
 			}
-			_, boundaryTemplate, err := tmplCtx.ResolveTemplate(&boundaryNode)
+			_, boundaryTemplate, templateStored, err := tmplCtx.ResolveTemplate(&boundaryNode)
 			if err != nil {
 				return err
 			}
+			// A new template was stored during resolution, persist it
+			if templateStored {
+				woc.updated = true
+			}
+
 			if boundaryTemplate != nil && boundaryTemplate.Parallelism != nil {
 				activeSiblings := woc.countActiveChildren(boundaryID)
 				woc.log.Debugf("counted %d/%d active children in boundary %s", activeSiblings, *boundaryTemplate.Parallelism, boundaryID)
@@ -1756,7 +1762,7 @@ func (woc *wfOperationCtx) checkParallelism(tmpl *wfv1.Template, node *wfv1.Node
 	return nil
 }
 
-func (woc *wfOperationCtx) executeContainer(nodeName string, templateScope string, tmpl *wfv1.Template, orgTmpl wfv1.TemplateHolder, opts *executeTemplateOpts) (*wfv1.NodeStatus, error) {
+func (woc *wfOperationCtx) executeContainer(nodeName string, templateScope string, tmpl *wfv1.Template, orgTmpl wfv1.TemplateReferenceHolder, opts *executeTemplateOpts) (*wfv1.NodeStatus, error) {
 	node := woc.getNodeByName(nodeName)
 	if node == nil {
 		node = woc.initializeExecutableNode(nodeName, wfv1.NodeTypePod, templateScope, tmpl, orgTmpl, opts.boundaryID, wfv1.NodePending)
@@ -1887,7 +1893,7 @@ func getStepOrDAGTaskName(nodeName string) string {
 	return nodeName
 }
 
-func (woc *wfOperationCtx) executeScript(nodeName string, templateScope string, tmpl *wfv1.Template, orgTmpl wfv1.TemplateHolder, opts *executeTemplateOpts) (*wfv1.NodeStatus, error) {
+func (woc *wfOperationCtx) executeScript(nodeName string, templateScope string, tmpl *wfv1.Template, orgTmpl wfv1.TemplateReferenceHolder, opts *executeTemplateOpts) (*wfv1.NodeStatus, error) {
 	node := woc.getNodeByName(nodeName)
 	if node != nil {
 		return node, nil
@@ -1896,14 +1902,19 @@ func (woc *wfOperationCtx) executeScript(nodeName string, templateScope string,
 
 	includeScriptOutput := false
 	if boundaryNode, ok := woc.wf.Status.Nodes[opts.boundaryID]; ok {
-		tmplCtx, err := woc.createTemplateContext(boundaryNode.TemplateScope)
+		tmplCtx, err := woc.createTemplateContext(boundaryNode.GetTemplateScope())
 		if err != nil {
 			return node, err
 		}
-		_, parentTemplate, err := tmplCtx.ResolveTemplate(&boundaryNode)
+		_, parentTemplate, templateStored, err := tmplCtx.ResolveTemplate(&boundaryNode)
 		if err != nil {
 			return node, err
 		}
+		// A new template was stored during resolution, persist it
+		if templateStored {
+			woc.updated = true
+		}
+
 		name := getStepOrDAGTaskName(nodeName)
 		includeScriptOutput = hasOutputResultRef(name, parentTemplate)
 	}
@@ -2143,7 +2154,7 @@ func (woc *wfOperationCtx) addChildNode(parent string, child string) {
 }
 
 // executeResource is runs a kubectl command against a manifest
-func (woc *wfOperationCtx) executeResource(nodeName string, templateScope string, tmpl *wfv1.Template, orgTmpl wfv1.TemplateHolder, opts *executeTemplateOpts) (*wfv1.NodeStatus, error) {
+func (woc *wfOperationCtx) executeResource(nodeName string, templateScope string, tmpl *wfv1.Template, orgTmpl wfv1.TemplateReferenceHolder, opts *executeTemplateOpts) (*wfv1.NodeStatus, error) {
 	node := woc.getNodeByName(nodeName)
 	if node != nil {
 		return node, nil
@@ -2175,7 +2186,7 @@ func (woc *wfOperationCtx) executeResource(nodeName string, templateScope string
 	return node, err
 }
 
-func (woc *wfOperationCtx) executeSuspend(nodeName string, templateScope string, tmpl *wfv1.Template, orgTmpl wfv1.TemplateHolder, opts *executeTemplateOpts) (*wfv1.NodeStatus, error) {
+func (woc *wfOperationCtx) executeSuspend(nodeName string, templateScope string, tmpl *wfv1.Template, orgTmpl wfv1.TemplateReferenceHolder, opts *executeTemplateOpts) (*wfv1.NodeStatus, error) {
 	node := woc.getNodeByName(nodeName)
 	if node == nil {
 		node = woc.initializeExecutableNode(nodeName, wfv1.NodeTypeSuspend, templateScope, tmpl, orgTmpl, opts.boundaryID, wfv1.NodePending)
@@ -2325,20 +2336,6 @@ func expandSequence(seq *wfv1.Sequence) ([]wfv1.Item, error) {
 	return items, nil
 }
 
-// GetStoredTemplate retrieves a template from stored templates of the workflow.
-func (woc *wfOperationCtx) GetStoredTemplate(templateScope string, holder wfv1.TemplateHolder) *wfv1.Template {
-	return woc.wf.GetStoredTemplate(templateScope, holder)
-}
-
-// SetStoredTemplate stores a new template in stored templates of the workflow.
-func (woc *wfOperationCtx) SetStoredTemplate(templateScope string, holder wfv1.TemplateHolder, tmpl *wfv1.Template) (bool, error) {
-	stored, err := woc.wf.SetStoredTemplate(templateScope, holder, tmpl)
-	if stored {
-		woc.updated = true
-	}
-	return stored, err
-}
-
 func (woc *wfOperationCtx) substituteParamsInVolumes(params map[string]string) error {
 	if woc.volumes == nil {
 		return nil
@@ -2364,12 +2361,17 @@ func (woc *wfOperationCtx) substituteParamsInVolumes(params map[string]string) e
 }
 
 // createTemplateContext creates a new template context.
-func (woc *wfOperationCtx) createTemplateContext(templateScope string) (*templateresolution.Context, error) {
-	ctx := templateresolution.NewContext(woc.controller.wftmplInformer.Lister().WorkflowTemplates(woc.wf.Namespace), woc.controller.cwftmplInformer.Lister(), woc.wf, woc)
-	if templateScope != "" {
-		return ctx.WithWorkflowTemplate(templateScope)
+func (woc *wfOperationCtx) createTemplateContext(scope wfv1.ResourceScope, resourceName string) (*templateresolution.Context, error) {
+	ctx := templateresolution.NewContext(woc.controller.wftmplInformer.Lister().WorkflowTemplates(woc.wf.Namespace), woc.controller.cwftmplInformer.Lister(), woc.wf, woc.wf)
+
+	switch scope {
+	case wfv1.ResourceScopeNamespaced:
+		return ctx.WithWorkflowTemplate(resourceName)
+	case wfv1.ResourceScopeCluster:
+		return ctx.WithClusterWorkflowTemplate(resourceName)
+	default:
+		return ctx, nil
 	}
-	return ctx, nil
 }
 
 func (woc *wfOperationCtx) runOnExitNode(parentName, templateRef, boundaryID string, tmplCtx *templateresolution.Context) (bool, *wfv1.NodeStatus, error) {
diff --git a/workflow/controller/steps.go b/workflow/controller/steps.go
index 046bc7bfb..4dfbd1ff0 100644
--- a/workflow/controller/steps.go
+++ b/workflow/controller/steps.go
@@ -27,7 +27,7 @@ type stepsContext struct {
 	tmplCtx *templateresolution.Context
 }
 
-func (woc *wfOperationCtx) executeSteps(nodeName string, tmplCtx *templateresolution.Context, templateScope string, tmpl *wfv1.Template, orgTmpl wfv1.TemplateHolder, opts *executeTemplateOpts) (*wfv1.NodeStatus, error) {
+func (woc *wfOperationCtx) executeSteps(nodeName string, tmplCtx *templateresolution.Context, templateScope string, tmpl *wfv1.Template, orgTmpl wfv1.TemplateReferenceHolder, opts *executeTemplateOpts) (*wfv1.NodeStatus, error) {
 	node := woc.getNodeByName(nodeName)
 	if node == nil {
 		node = woc.initializeExecutableNode(nodeName, wfv1.NodeTypeSteps, templateScope, tmpl, orgTmpl, opts.boundaryID, wfv1.NodeRunning)
@@ -40,7 +40,7 @@ func (woc *wfOperationCtx) executeSteps(nodeName string, tmplCtx *templateresolu
 	}()
 
 	// The template scope of this step.
-	stepTemplateScope := tmplCtx.GetCurrentTemplateBase().GetTemplateScope()
+	stepTemplateScope := tmplCtx.GetTemplateScope()
 
 	stepsCtx := stepsContext{
 		boundaryID: node.ID,
@@ -114,10 +114,15 @@ func (woc *wfOperationCtx) executeSteps(nodeName string, tmplCtx *templateresolu
 				}
 				if len(childNodes) > 0 {
 					// Expanded child nodes should be created from the same template.
-					_, tmpl, err := stepsCtx.tmplCtx.ResolveTemplate(&childNodes[0])
+					_, tmpl, templateStored, err := stepsCtx.tmplCtx.ResolveTemplate(&childNodes[0])
 					if err != nil {
 						return node, err
 					}
+					// A new template was stored during resolution, persist it
+					if templateStored {
+						woc.updated = true
+					}
+
 					err = woc.processAggregateNodeOutputs(tmpl, stepsCtx.scope, prefix, childNodes)
 					if err != nil {
 						return node, err
@@ -196,7 +201,7 @@ func (woc *wfOperationCtx) executeStepGroup(stepGroup []wfv1.WorkflowStep, sgNod
 	nodeSteps := make(map[string]wfv1.WorkflowStep)
 
 	// The template scope of this step group.
-	stepTemplateScope := stepsCtx.tmplCtx.GetCurrentTemplateBase().GetTemplateScope()
+	stepTemplateScope := stepsCtx.tmplCtx.GetTemplateScope()
 
 	// Kick off all parallel steps in the group
 	for _, step := range stepGroup {
@@ -382,7 +387,7 @@ func (woc *wfOperationCtx) expandStepGroup(sgNodeName string, stepGroup []wfv1.W
 			// Empty list
 			childNodeName := fmt.Sprintf("%s.%s", sgNodeName, step.Name)
 			if woc.getNodeByName(childNodeName) == nil {
-				stepTemplateScope := stepsCtx.tmplCtx.GetCurrentTemplateBase().GetTemplateScope()
+				stepTemplateScope := stepsCtx.tmplCtx.GetTemplateScope()
 				skipReason := fmt.Sprint("Skipped, empty params")
 				woc.log.Infof("Skipping %s: %s", childNodeName, skipReason)
 				woc.initializeNode(childNodeName, wfv1.NodeTypeSkipped, stepTemplateScope, &step, stepsCtx.boundaryID, wfv1.NodeSkipped, skipReason)
diff --git a/workflow/templateresolution/context.go b/workflow/templateresolution/context.go
index 3613d8ddf..b9b96e8b7 100644
--- a/workflow/templateresolution/context.go
+++ b/workflow/templateresolution/context.go
@@ -1,8 +1,6 @@
 package templateresolution
 
 import (
-	"strings"
-
 	"github.com/sirupsen/logrus"
 	log "github.com/sirupsen/logrus"
 	apierr "k8s.io/apimachinery/pkg/api/errors"
@@ -65,31 +63,31 @@ type Context struct {
 	// cwftmplGetter is an interface to get ClusterWorkflowTemplates
 	cwftmplGetter ClusterWorkflowTemplateGetter
 	// tmplBase is the base of local template search.
-	tmplBase wfv1.TemplateGetter
-	// storage is an implementation of TemplateStorage.
-	storage wfv1.TemplateStorage
+	tmplBase wfv1.TemplateHolder
+	// workflow is the Workflow where templates will be stored
+	workflow *wfv1.Workflow
 	// log is a logrus entry.
 	log *logrus.Entry
 }
 
 // NewContext returns new Context.
-func NewContext(wftmplGetter WorkflowTemplateNamespacedGetter, cwftmplGetter ClusterWorkflowTemplateGetter, tmplBase wfv1.TemplateGetter, storage wfv1.TemplateStorage) *Context {
+func NewContext(wftmplGetter WorkflowTemplateNamespacedGetter, cwftmplGetter ClusterWorkflowTemplateGetter, tmplBase wfv1.TemplateHolder, workflow *wfv1.Workflow) *Context {
 	return &Context{
 		wftmplGetter:  wftmplGetter,
 		cwftmplGetter: cwftmplGetter,
 		tmplBase:      tmplBase,
-		storage:       storage,
+		workflow:      workflow,
 		log:           log.WithFields(logrus.Fields{}),
 	}
 }
 
 // NewContext returns new Context.
-func NewContextFromClientset(wftmplClientset typed.WorkflowTemplateInterface, clusterWftmplClient typed.ClusterWorkflowTemplateInterface, tmplBase wfv1.TemplateGetter, storage wfv1.TemplateStorage) *Context {
+func NewContextFromClientset(wftmplClientset typed.WorkflowTemplateInterface, clusterWftmplClient typed.ClusterWorkflowTemplateInterface, tmplBase wfv1.TemplateHolder, workflow *wfv1.Workflow) *Context {
 	return &Context{
 		wftmplGetter:  WrapWorkflowTemplateInterface(wftmplClientset),
 		cwftmplGetter: WrapClusterWorkflowTemplateInterface(clusterWftmplClient),
 		tmplBase:      tmplBase,
-		storage:       storage,
+		workflow:      workflow,
 		log:           log.WithFields(logrus.Fields{}),
 	}
 }
@@ -105,7 +103,7 @@ func (ctx *Context) GetTemplateByName(name string) (*wfv1.Template, error) {
 	return tmpl.DeepCopy(), nil
 }
 
-func (ctx *Context) GetTemplateGetterFromRef(tmplRef *wfv1.TemplateRef) (wfv1.TemplateGetter, error) {
+func (ctx *Context) GetTemplateGetterFromRef(tmplRef *wfv1.TemplateRef) (wfv1.TemplateHolder, error) {
 	if tmplRef.ClusterScope {
 		return ctx.cwftmplGetter.Get(tmplRef.Name)
 	}
@@ -116,7 +114,7 @@ func (ctx *Context) GetTemplateGetterFromRef(tmplRef *wfv1.TemplateRef) (wfv1.Te
 func (ctx *Context) GetTemplateFromRef(tmplRef *wfv1.TemplateRef) (*wfv1.Template, error) {
 	ctx.log.Debug("Getting the template from ref")
 	var template *wfv1.Template
-	var wftmpl wfv1.WorkflowTemplateInterface
+	var wftmpl wfv1.TemplateHolder
 	var err error
 	if tmplRef.ClusterScope {
 		wftmpl, err = ctx.cwftmplGetter.Get(tmplRef.Name)
@@ -140,7 +138,7 @@ func (ctx *Context) GetTemplateFromRef(tmplRef *wfv1.TemplateRef) (*wfv1.Templat
 }
 
 // GetTemplate returns a template found by template name or template ref.
-func (ctx *Context) GetTemplate(tmplHolder wfv1.TemplateHolder) (*wfv1.Template, error) {
+func (ctx *Context) GetTemplate(tmplHolder wfv1.TemplateReferenceHolder) (*wfv1.Template, error) {
 	ctx.log.Debug("Getting the template")
 
 	tmplName := tmplHolder.GetTemplateName()
@@ -161,13 +159,17 @@ func (ctx *Context) GetTemplate(tmplHolder wfv1.TemplateHolder) (*wfv1.Template,
 }
 
 // GetCurrentTemplateBase returns the current template base of the context.
-func (ctx *Context) GetCurrentTemplateBase() wfv1.TemplateGetter {
+func (ctx *Context) GetCurrentTemplateBase() wfv1.TemplateHolder {
 	return ctx.tmplBase
 }
 
+func (ctx *Context) GetTemplateScope() string {
+	return string(ctx.tmplBase.GetResourceScope()) + "/" + ctx.tmplBase.GetName()
+}
+
 // ResolveTemplate digs into referenes and returns a merged template.
 // This method is the public start point of template resolution.
-func (ctx *Context) ResolveTemplate(tmplHolder wfv1.TemplateHolder) (*Context, *wfv1.Template, error) {
+func (ctx *Context) ResolveTemplate(tmplHolder wfv1.TemplateReferenceHolder) (*Context, *wfv1.Template, bool, error) {
 	return ctx.resolveTemplateImpl(tmplHolder, 0)
 }
 
@@ -175,7 +177,7 @@ func (ctx *Context) ResolveTemplate(tmplHolder wfv1.TemplateHolder) (*Context, *
 // This method processes inputs and arguments so the inputs of the final
 //  resolved template include intermediate parameter passing.
 // The other fields are just merged and shallower templates overwrite deeper.
-func (ctx *Context) resolveTemplateImpl(tmplHolder wfv1.TemplateHolder, depth int) (*Context, *wfv1.Template, error) {
+func (ctx *Context) resolveTemplateImpl(tmplHolder wfv1.TemplateReferenceHolder, depth int) (*Context, *wfv1.Template, bool, error) {
 	ctx.log = ctx.log.WithFields(logrus.Fields{
 		"depth": depth,
 		"base":  common.GetTemplateGetterString(ctx.tmplBase),
@@ -183,15 +185,18 @@ func (ctx *Context) resolveTemplateImpl(tmplHolder wfv1.TemplateHolder, depth in
 	})
 	// Avoid infinite references
 	if depth > maxResolveDepth {
-		return nil, nil, errors.Errorf(errors.CodeBadRequest, "template reference exceeded max depth (%d)", maxResolveDepth)
+		return nil, nil, false, errors.Errorf(errors.CodeBadRequest, "template reference exceeded max depth (%d)", maxResolveDepth)
 	}
 
 	ctx.log.Debug("Resolving the template")
 
+	templateStored := false
 	var tmpl *wfv1.Template
-	if ctx.storage != nil {
+	if ctx.workflow != nil {
 		// Check if the template has been stored.
-		tmpl = ctx.storage.GetStoredTemplate(ctx.tmplBase.GetTemplateScope(), tmplHolder)
+		scope := ctx.tmplBase.GetResourceScope()
+		resourceName := ctx.tmplBase.GetName()
+		tmpl = ctx.workflow.GetStoredTemplate(scope, resourceName, tmplHolder)
 	}
 	if tmpl != nil {
 		ctx.log.Debug("Found stored template")
@@ -199,16 +204,19 @@ func (ctx *Context) resolveTemplateImpl(tmplHolder wfv1.TemplateHolder, depth in
 		// Find newly appeared template.
 		newTmpl, err := ctx.GetTemplate(tmplHolder)
 		if err != nil {
-			return nil, nil, err
+			return nil, nil, false, err
 		}
 		// Stored the found template.
-		if ctx.storage != nil {
-			stored, err := ctx.storage.SetStoredTemplate(ctx.tmplBase.GetTemplateScope(), tmplHolder, newTmpl)
+		if ctx.workflow != nil {
+			scope := ctx.tmplBase.GetResourceScope()
+			resourceName := ctx.tmplBase.GetName()
+			stored, err := ctx.workflow.SetStoredTemplate(scope, resourceName, tmplHolder, newTmpl)
 			if err != nil {
-				return nil, nil, err
+				return nil, nil, false, err
 			}
 			if stored {
 				ctx.log.Debug("Stored the template")
+				templateStored = true
 			}
 		}
 		tmpl = newTmpl
@@ -217,68 +225,65 @@ func (ctx *Context) resolveTemplateImpl(tmplHolder wfv1.TemplateHolder, depth in
 	// Update the template base of the context.
 	newTmplCtx, err := ctx.WithTemplateHolder(tmplHolder)
 	if err != nil {
-		return nil, nil, err
+		return nil, nil, false, err
 	}
 
 	// Return a concrete template without digging into it.
 	if tmpl.GetType() != wfv1.TemplateTypeUnknown {
-		return newTmplCtx, tmpl, nil
+		return newTmplCtx, tmpl, templateStored, nil
 	}
 
 	// Dig into nested references with new template base.
-	finalTmplCtx, resolvedTmpl, err := newTmplCtx.resolveTemplateImpl(tmpl, depth+1)
+	finalTmplCtx, resolvedTmpl, templateStoredInCall, err := newTmplCtx.resolveTemplateImpl(tmpl, depth+1)
 	if err != nil {
-		return nil, nil, err
+		return nil, nil, false, err
+	}
+	if templateStoredInCall {
+		templateStored = true
 	}
 
 	// Merge the referred template into the original.
 	mergedTmpl, err := common.MergeReferredTemplate(tmpl, resolvedTmpl)
 	if err != nil {
-		return nil, nil, err
+		return nil, nil, false, err
 	}
 
-	return finalTmplCtx, mergedTmpl, nil
+	return finalTmplCtx, mergedTmpl, templateStored, nil
 }
 
 // WithTemplateHolder creates new context with a template base of a given template holder.
-func (ctx *Context) WithTemplateHolder(tmplHolder wfv1.TemplateHolder) (*Context, error) {
+func (ctx *Context) WithTemplateHolder(tmplHolder wfv1.TemplateReferenceHolder) (*Context, error) {
 	tmplRef := tmplHolder.GetTemplateRef()
 	if tmplRef != nil {
 		tmplName := tmplRef.Name
 		if tmplRef.ClusterScope {
-			tmplName = "cluster/" + tmplName
+			return ctx.WithClusterWorkflowTemplate(tmplName)
 		} else {
-			tmplName = "namespaced/" + tmplName
+			return ctx.WithWorkflowTemplate(tmplName)
 		}
-		return ctx.WithWorkflowTemplate(tmplName)
 	}
 	return ctx.WithTemplateBase(ctx.tmplBase), nil
 }
 
-// WithTemplateBase creates new context with a wfv1.TemplateGetter.
-func (ctx *Context) WithTemplateBase(tmplBase wfv1.TemplateGetter) *Context {
-	return NewContext(ctx.wftmplGetter, ctx.cwftmplGetter, tmplBase, ctx.storage)
+// WithTemplateBase creates new context with a wfv1.TemplateHolder.
+func (ctx *Context) WithTemplateBase(tmplBase wfv1.TemplateHolder) *Context {
+	return NewContext(ctx.wftmplGetter, ctx.cwftmplGetter, tmplBase, ctx.workflow)
 }
 
-// WithWorkflowTemplate creates new context with a wfv1.TemplateGetter.
+// WithWorkflowTemplate creates new context with a wfv1.TemplateHolder.
 func (ctx *Context) WithWorkflowTemplate(name string) (*Context, error) {
-	wfTmplnames := strings.Split(name, "/")
-	if len(wfTmplnames) < 1 {
-		return nil, errors.Errorf(errors.CodeBadRequest, "Invalid template name. %s", name)
-	}
-	if wfTmplnames[0] == "cluster" {
-		cwftmpl, err := ctx.cwftmplGetter.Get(wfTmplnames[1])
-		if err != nil {
-			return nil, err
-		}
-		return ctx.WithTemplateBase(cwftmpl), nil
+	wftmpl, err := ctx.wftmplGetter.Get(name)
+	if err != nil {
+		return nil, err
 	}
-	if wfTmplnames[0] == "namespaced" {
-		wftmpl, err := ctx.wftmplGetter.Get(wfTmplnames[1])
-		if err != nil {
-			return nil, err
-		}
-		return ctx.WithTemplateBase(wftmpl), nil
+	return ctx.WithTemplateBase(wftmpl), nil
+}
+
+// WithWorkflowTemplate creates new context with a wfv1.TemplateHolder.
+func (ctx *Context) WithClusterWorkflowTemplate(name string) (*Context, error) {
+	cwftmpl, err := ctx.cwftmplGetter.Get(name)
+	if err != nil {
+		return nil, err
 	}
-	return ctx, nil
+	return ctx.WithTemplateBase(cwftmpl), nil
 }
diff --git a/workflow/validate/validate.go b/workflow/validate/validate.go
index 98b25ad43..31c2edc05 100644
--- a/workflow/validate/validate.go
+++ b/workflow/validate/validate.go
@@ -199,12 +199,12 @@ func ValidateWorkflow(wftmplGetter templateresolution.WorkflowTemplateNamespaced
 }
 
 // ValidateWorkflow accepts a workflow template and performs validation against it.
-func ValidateWorkflowTemplate(wftmplGetter templateresolution.WorkflowTemplateNamespacedGetter, cwftmplGetter templateresolution.ClusterWorkflowTemplateGetter, wftmpl wfv1.TemplateGetter) error {
+func ValidateWorkflowTemplate(wftmplGetter templateresolution.WorkflowTemplateNamespacedGetter, cwftmplGetter templateresolution.ClusterWorkflowTemplateGetter, wftmpl wfv1.TemplateHolder) error {
 	ctx := newTemplateValidationCtx(nil, ValidateOpts{})
 	tmplCtx := templateresolution.NewContext(wftmplGetter, cwftmplGetter, wftmpl, nil)
 
 	// Check if all templates can be resolved.
-	for _, template := range wftmpl.GetAllTemplates() {
+	for _, template := range wftmpl.GetTemplates() {
 		_, err := ctx.validateTemplateHolder(&wfv1.WorkflowStep{Template: template.Name}, tmplCtx, &FakeArguments{}, map[string]interface{}{})
 		if err != nil {
 			return errors.Errorf(errors.CodeBadRequest, "templates.%s %s", template.Name, err.Error())
@@ -363,7 +363,7 @@ func (ctx *templateValidationCtx) validateTemplate(tmpl *wfv1.Template, tmplCtx
 }
 
 // validateTemplateHolder validates a template holder and returns the validated template.
-func (ctx *templateValidationCtx) validateTemplateHolder(tmplHolder wfv1.TemplateHolder, tmplCtx *templateresolution.Context, args wfv1.ArgumentsProvider, extraScope map[string]interface{}) (*wfv1.Template, error) {
+func (ctx *templateValidationCtx) validateTemplateHolder(tmplHolder wfv1.TemplateReferenceHolder, tmplCtx *templateresolution.Context, args wfv1.ArgumentsProvider, extraScope map[string]interface{}) (*wfv1.Template, error) {
 	tmplRef := tmplHolder.GetTemplateRef()
 	tmplName := tmplHolder.GetTemplateName()
 	if tmplRef != nil {
@@ -396,7 +396,7 @@ func (ctx *templateValidationCtx) validateTemplateHolder(tmplHolder wfv1.Templat
 		}
 	}
 
-	tmplCtx, resolvedTmpl, err := tmplCtx.ResolveTemplate(tmplHolder)
+	tmplCtx, resolvedTmpl, _, err := tmplCtx.ResolveTemplate(tmplHolder)
 	if err != nil {
 		if argoerr, ok := err.(errors.ArgoError); ok && argoerr.Code() == errors.CodeNotFound {
 			if tmplRef != nil {


Test patch:
diff --git a/workflow/controller/operator_template_scope_test.go b/workflow/controller/operator_template_scope_test.go
index 1b0e1ab6b..2d56ff6a0 100644
--- a/workflow/controller/operator_template_scope_test.go
+++ b/workflow/controller/operator_template_scope_test.go
@@ -496,7 +496,7 @@ func TestTemplateClusterScope(t *testing.T) {
 	node := findNodeByName(wf.Status.Nodes, "test-template-scope")
 	if assert.NotNil(t, node, "Node %s not found", "test-templte-scope") {
 		assert.Equal(t, wfv1.NodeTypeSteps, node.Type)
-		assert.Equal(t, "", node.TemplateScope)
+		assert.Equal(t, "local/test-template-scope", node.TemplateScope)
 	}
 
 	node = findNodeByName(wf.Status.Nodes, "test-template-scope[0]")
diff --git a/workflow/controller/workflowpod_test.go b/workflow/controller/workflowpod_test.go
index 4b8613f84..9bfb5735c 100644
--- a/workflow/controller/workflowpod_test.go
+++ b/workflow/controller/workflowpod_test.go
@@ -186,9 +186,9 @@ func TestScriptTemplateWithoutVolumeOptionalArtifact(t *testing.T) {
 func TestWFLevelServiceAccount(t *testing.T) {
 	woc := newWoc()
 	woc.wf.Spec.ServiceAccountName = "foo"
-	tmplCtx, err := woc.createTemplateContext("")
+	tmplCtx, err := woc.createTemplateContext(wfv1.ResourceScopeLocal, "")
 	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{})
+	_, err = woc.executeContainer(woc.wf.Spec.Entrypoint, tmplCtx.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)
@@ -203,9 +203,9 @@ func TestTmplServiceAccount(t *testing.T) {
 	woc := newWoc()
 	woc.wf.Spec.ServiceAccountName = "foo"
 	woc.wf.Spec.Templates[0].ServiceAccountName = "tmpl"
-	tmplCtx, err := woc.createTemplateContext("")
+	tmplCtx, err := woc.createTemplateContext(wfv1.ResourceScopeLocal, "")
 	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{})
+	_, err = woc.executeContainer(woc.wf.Spec.Entrypoint, tmplCtx.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)
@@ -223,9 +223,9 @@ func TestWFLevelAutomountServiceAccountToken(t *testing.T) {
 	falseValue := false
 	woc.wf.Spec.AutomountServiceAccountToken = &falseValue
 	woc.wf.Spec.Executor = &wfv1.ExecutorConfig{ServiceAccountName: "foo"}
-	tmplCtx, err := woc.createTemplateContext("")
+	tmplCtx, err := woc.createTemplateContext(wfv1.ResourceScopeLocal, "")
 	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{})
+	_, err = woc.executeContainer(woc.wf.Spec.Entrypoint, tmplCtx.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)
@@ -245,9 +245,9 @@ func TestTmplLevelAutomountServiceAccountToken(t *testing.T) {
 	woc.wf.Spec.AutomountServiceAccountToken = &trueValue
 	woc.wf.Spec.Executor = &wfv1.ExecutorConfig{ServiceAccountName: "foo"}
 	woc.wf.Spec.Templates[0].AutomountServiceAccountToken = &falseValue
-	tmplCtx, err := woc.createTemplateContext("")
+	tmplCtx, err := woc.createTemplateContext(wfv1.ResourceScopeLocal, "")
 	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{})
+	_, err = woc.executeContainer(woc.wf.Spec.Entrypoint, tmplCtx.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)
@@ -273,9 +273,9 @@ func TestWFLevelExecutorServiceAccountName(t *testing.T) {
 	assert.NoError(t, err)
 
 	woc.wf.Spec.Executor = &wfv1.ExecutorConfig{ServiceAccountName: "foo"}
-	tmplCtx, err := woc.createTemplateContext("")
+	tmplCtx, err := woc.createTemplateContext(wfv1.ResourceScopeLocal, "")
 	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{})
+	_, err = woc.executeContainer(woc.wf.Spec.Entrypoint, tmplCtx.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)
@@ -298,9 +298,9 @@ func TestTmplLevelExecutorServiceAccountName(t *testing.T) {
 
 	woc.wf.Spec.Executor = &wfv1.ExecutorConfig{ServiceAccountName: "foo"}
 	woc.wf.Spec.Templates[0].Executor = &wfv1.ExecutorConfig{ServiceAccountName: "tmpl"}
-	tmplCtx, err := woc.createTemplateContext("")
+	tmplCtx, err := woc.createTemplateContext(wfv1.ResourceScopeLocal, "")
 	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{})
+	_, err = woc.executeContainer(woc.wf.Spec.Entrypoint, tmplCtx.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)
@@ -324,9 +324,9 @@ func TestTmplLevelExecutorSecurityContext(t *testing.T) {
 
 	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("")
+	tmplCtx, err := woc.createTemplateContext(wfv1.ResourceScopeLocal, "")
 	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{})
+	_, err = woc.executeContainer(woc.wf.Spec.Entrypoint, tmplCtx.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)
@@ -351,9 +351,9 @@ func TestImagePullSecrets(t *testing.T) {
 			Name: "secret-name",
 		},
 	}
-	tmplCtx, err := woc.createTemplateContext("")
+	tmplCtx, err := woc.createTemplateContext(wfv1.ResourceScopeLocal, "")
 	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{})
+	_, err = woc.executeContainer(woc.wf.Spec.Entrypoint, tmplCtx.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)
@@ -385,9 +385,9 @@ func TestAffinity(t *testing.T) {
 			},
 		},
 	}
-	tmplCtx, err := woc.createTemplateContext("")
+	tmplCtx, err := woc.createTemplateContext(wfv1.ResourceScopeLocal, "")
 	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{})
+	_, err = woc.executeContainer(woc.wf.Spec.Entrypoint, tmplCtx.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)
@@ -404,9 +404,9 @@ func TestTolerations(t *testing.T) {
 		Operator: "Exists",
 		Effect:   "NoSchedule",
 	}}
-	tmplCtx, err := woc.createTemplateContext("")
+	tmplCtx, err := woc.createTemplateContext(wfv1.ResourceScopeLocal, "")
 	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{})
+	_, err = woc.executeContainer(woc.wf.Spec.Entrypoint, tmplCtx.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)
@@ -419,9 +419,9 @@ func TestTolerations(t *testing.T) {
 // TestMetadata verifies ability to carry forward annotations and labels
 func TestMetadata(t *testing.T) {
 	woc := newWoc()
-	tmplCtx, err := woc.createTemplateContext("")
+	tmplCtx, err := woc.createTemplateContext(wfv1.ResourceScopeLocal, "")
 	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{})
+	_, err = woc.executeContainer(woc.wf.Spec.Entrypoint, tmplCtx.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)
@@ -572,9 +572,9 @@ func TestVolumeAndVolumeMounts(t *testing.T) {
 		woc.wf.Spec.Templates[0].Container.VolumeMounts = volumeMounts
 		woc.controller.Config.ContainerRuntimeExecutor = common.ContainerRuntimeExecutorDocker
 
-		tmplCtx, err := woc.createTemplateContext("")
+		tmplCtx, err := woc.createTemplateContext(wfv1.ResourceScopeLocal, "")
 		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{})
+		_, err = woc.executeContainer(woc.wf.Spec.Entrypoint, tmplCtx.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)
@@ -595,9 +595,9 @@ func TestVolumeAndVolumeMounts(t *testing.T) {
 		woc.wf.Spec.Templates[0].Container.VolumeMounts = volumeMounts
 		woc.controller.Config.ContainerRuntimeExecutor = common.ContainerRuntimeExecutorKubelet
 
-		tmplCtx, err := woc.createTemplateContext("")
+		tmplCtx, err := woc.createTemplateContext(wfv1.ResourceScopeLocal, "")
 		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{})
+		_, err = woc.executeContainer(woc.wf.Spec.Entrypoint, tmplCtx.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)
@@ -617,9 +617,9 @@ func TestVolumeAndVolumeMounts(t *testing.T) {
 		woc.wf.Spec.Templates[0].Container.VolumeMounts = volumeMounts
 		woc.controller.Config.ContainerRuntimeExecutor = common.ContainerRuntimeExecutorK8sAPI
 
-		tmplCtx, err := woc.createTemplateContext("")
+		tmplCtx, err := woc.createTemplateContext(wfv1.ResourceScopeLocal, "")
 		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{})
+		_, err = woc.executeContainer(woc.wf.Spec.Entrypoint, tmplCtx.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)
@@ -664,9 +664,9 @@ func TestVolumesPodSubstitution(t *testing.T) {
 	woc.wf.Spec.Templates[0].Inputs.Parameters = inputParameters
 	woc.controller.Config.ContainerRuntimeExecutor = common.ContainerRuntimeExecutorDocker
 
-	tmplCtx, err := woc.createTemplateContext("")
+	tmplCtx, err := woc.createTemplateContext(wfv1.ResourceScopeLocal, "")
 	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{})
+	_, err = woc.executeContainer(woc.wf.Spec.Entrypoint, tmplCtx.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)
@@ -702,9 +702,9 @@ func TestOutOfCluster(t *testing.T) {
 			SecretKey:  "bar",
 		}
 
-		tmplCtx, err := woc.createTemplateContext("")
+		tmplCtx, err := woc.createTemplateContext(wfv1.ResourceScopeLocal, "")
 		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{})
+		_, err = woc.executeContainer(woc.wf.Spec.Entrypoint, tmplCtx.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)
@@ -727,9 +727,9 @@ func TestOutOfCluster(t *testing.T) {
 			VolumeName: "kube-config-secret",
 		}
 
-		tmplCtx, err := woc.createTemplateContext("")
+		tmplCtx, err := woc.createTemplateContext(wfv1.ResourceScopeLocal, "")
 		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{})
+		_, err = woc.executeContainer(woc.wf.Spec.Entrypoint, tmplCtx.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)
@@ -750,9 +750,9 @@ func TestPriority(t *testing.T) {
 	woc := newWoc()
 	woc.wf.Spec.Templates[0].PriorityClassName = "foo"
 	woc.wf.Spec.Templates[0].Priority = &priority
-	tmplCtx, err := woc.createTemplateContext("")
+	tmplCtx, err := woc.createTemplateContext(wfv1.ResourceScopeLocal, "")
 	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{})
+	_, err = woc.executeContainer(woc.wf.Spec.Entrypoint, tmplCtx.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)
@@ -766,9 +766,9 @@ func TestPriority(t *testing.T) {
 func TestSchedulerName(t *testing.T) {
 	woc := newWoc()
 	woc.wf.Spec.Templates[0].SchedulerName = "foo"
-	tmplCtx, err := woc.createTemplateContext("")
+	tmplCtx, err := woc.createTemplateContext(wfv1.ResourceScopeLocal, "")
 	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{})
+	_, err = woc.executeContainer(woc.wf.Spec.Entrypoint, tmplCtx.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)
@@ -820,9 +820,9 @@ func TestInitContainers(t *testing.T) {
 		},
 	}
 
-	tmplCtx, err := woc.createTemplateContext("")
+	tmplCtx, err := woc.createTemplateContext(wfv1.ResourceScopeLocal, "")
 	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{})
+	_, err = woc.executeContainer(woc.wf.Spec.Entrypoint, tmplCtx.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)
@@ -881,9 +881,9 @@ func TestSidecars(t *testing.T) {
 		},
 	}
 
-	tmplCtx, err := woc.createTemplateContext("")
+	tmplCtx, err := woc.createTemplateContext(wfv1.ResourceScopeLocal, "")
 	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{})
+	_, err = woc.executeContainer(woc.wf.Spec.Entrypoint, tmplCtx.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)
@@ -935,9 +935,9 @@ func TestTemplateLocalVolumes(t *testing.T) {
 	woc.wf.Spec.Templates[0].Container.VolumeMounts = volumeMounts
 	woc.wf.Spec.Templates[0].Volumes = localVolumes
 
-	tmplCtx, err := woc.createTemplateContext("")
+	tmplCtx, err := woc.createTemplateContext(wfv1.ResourceScopeLocal, "")
 	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{})
+	_, err = woc.executeContainer(woc.wf.Spec.Entrypoint, tmplCtx.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)
@@ -958,9 +958,9 @@ func TestWFLevelHostAliases(t *testing.T) {
 		{IP: "127.0.0.1"},
 		{IP: "127.0.0.1"},
 	}
-	tmplCtx, err := woc.createTemplateContext("")
+	tmplCtx, err := woc.createTemplateContext(wfv1.ResourceScopeLocal, "")
 	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{})
+	_, err = woc.executeContainer(woc.wf.Spec.Entrypoint, tmplCtx.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)
@@ -977,9 +977,9 @@ func TestTmplLevelHostAliases(t *testing.T) {
 		{IP: "127.0.0.1"},
 		{IP: "127.0.0.1"},
 	}
-	tmplCtx, err := woc.createTemplateContext("")
+	tmplCtx, err := woc.createTemplateContext(wfv1.ResourceScopeLocal, "")
 	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{})
+	_, err = woc.executeContainer(woc.wf.Spec.Entrypoint, tmplCtx.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)
@@ -996,9 +996,9 @@ func TestWFLevelSecurityContext(t *testing.T) {
 	woc.wf.Spec.SecurityContext = &apiv1.PodSecurityContext{
 		RunAsUser: &runAsUser,
 	}
-	tmplCtx, err := woc.createTemplateContext("")
+	tmplCtx, err := woc.createTemplateContext(wfv1.ResourceScopeLocal, "")
 	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{})
+	_, err = woc.executeContainer(woc.wf.Spec.Entrypoint, tmplCtx.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)
@@ -1015,9 +1015,9 @@ func TestTmplLevelSecurityContext(t *testing.T) {
 	woc.wf.Spec.Templates[0].SecurityContext = &apiv1.PodSecurityContext{
 		RunAsUser: &runAsUser,
 	}
-	tmplCtx, err := woc.createTemplateContext("")
+	tmplCtx, err := woc.createTemplateContext(wfv1.ResourceScopeLocal, "")
 	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{})
+	_, err = woc.executeContainer(woc.wf.Spec.Entrypoint, tmplCtx.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)
diff --git a/workflow/templateresolution/context_test.go b/workflow/templateresolution/context_test.go
index a49d6914f..d78661c15 100644
--- a/workflow/templateresolution/context_test.go
+++ b/workflow/templateresolution/context_test.go
@@ -214,7 +214,7 @@ func TestWithTemplateHolder(t *testing.T) {
 	wftmpl := unmarshalWftmpl(baseWorkflowTemplateYaml)
 	ctx := NewContextFromClientset(wfClientset.ArgoprojV1alpha1().WorkflowTemplates(metav1.NamespaceDefault), wfClientset.ArgoprojV1alpha1().ClusterWorkflowTemplates(), wftmpl, nil)
 
-	var tmplGetter wfv1.TemplateGetter
+	var tmplGetter wfv1.TemplateHolder
 	// Get the template base of existing template name.
 	tmplHolder := wfv1.WorkflowStep{Template: "whalesay"}
 	newCtx, err := ctx.WithTemplateHolder(&tmplHolder)
@@ -266,7 +266,7 @@ func TestResolveTemplate(t *testing.T) {
 
 	// Get the template of template name.
 	tmplHolder := wfv1.WorkflowStep{Template: "whalesay"}
-	ctx, tmpl, err := ctx.ResolveTemplate(&tmplHolder)
+	ctx, tmpl, _, err := ctx.ResolveTemplate(&tmplHolder)
 	if !assert.NoError(t, err) {
 		t.Fatal(err)
 	}
@@ -277,10 +277,10 @@ func TestResolveTemplate(t *testing.T) {
 	assert.Equal(t, "base-workflow-template", wftmpl.Name)
 	assert.Equal(t, "whalesay", tmpl.Name)
 
-	var tmplGetter wfv1.TemplateGetter
+	var tmplGetter wfv1.TemplateHolder
 	// Get the template of template reference.
 	tmplHolder = wfv1.WorkflowStep{TemplateRef: &wfv1.TemplateRef{Name: "some-workflow-template", Template: "whalesay"}}
-	ctx, tmpl, err = ctx.ResolveTemplate(&tmplHolder)
+	ctx, tmpl, _, err = ctx.ResolveTemplate(&tmplHolder)
 	if !assert.NoError(t, err) {
 		t.Fatal(err)
 	}
@@ -294,7 +294,7 @@ func TestResolveTemplate(t *testing.T) {
 
 	// Get the template of local nested template reference.
 	tmplHolder = wfv1.WorkflowStep{TemplateRef: &wfv1.TemplateRef{Name: "some-workflow-template", Template: "local-whalesay"}}
-	ctx, tmpl, err = ctx.ResolveTemplate(&tmplHolder)
+	ctx, tmpl, _, err = ctx.ResolveTemplate(&tmplHolder)
 	if !assert.NoError(t, err) {
 		t.Fatal(err)
 	}
@@ -308,7 +308,7 @@ func TestResolveTemplate(t *testing.T) {
 
 	// Get the template of nested template reference.
 	tmplHolder = wfv1.WorkflowStep{TemplateRef: &wfv1.TemplateRef{Name: "some-workflow-template", Template: "another-whalesay"}}
-	ctx, tmpl, err = ctx.ResolveTemplate(&tmplHolder)
+	ctx, tmpl, _, err = ctx.ResolveTemplate(&tmplHolder)
 	if !assert.NoError(t, err) {
 		t.Fatal(err)
 	}
@@ -324,7 +324,7 @@ func TestResolveTemplate(t *testing.T) {
 	tmplHolder = wfv1.WorkflowStep{
 		TemplateRef: &wfv1.TemplateRef{Name: "some-workflow-template", Template: "whalesay-with-arguments"},
 	}
-	ctx, tmpl, err = ctx.ResolveTemplate(&tmplHolder)
+	ctx, tmpl, _, err = ctx.ResolveTemplate(&tmplHolder)
 	if !assert.NoError(t, err) {
 		t.Fatal(err)
 	}
@@ -341,7 +341,7 @@ func TestResolveTemplate(t *testing.T) {
 	tmplHolder = wfv1.WorkflowStep{
 		TemplateRef: &wfv1.TemplateRef{Name: "some-workflow-template", Template: "nested-whalesay-with-arguments"},
 	}
-	ctx, tmpl, err = ctx.ResolveTemplate(&tmplHolder)
+	ctx, tmpl, _, err = ctx.ResolveTemplate(&tmplHolder)
 	if !assert.NoError(t, err) {
 		t.Fatal(err)
 	}
@@ -356,12 +356,12 @@ func TestResolveTemplate(t *testing.T) {
 
 	// Get the template of infinite loop template reference.
 	tmplHolder = wfv1.WorkflowStep{TemplateRef: &wfv1.TemplateRef{Name: "some-workflow-template", Template: "infinite-loop-whalesay"}}
-	_, _, err = ctx.ResolveTemplate(&tmplHolder)
+	_, _, _, err = ctx.ResolveTemplate(&tmplHolder)
 	assert.EqualError(t, err, "template reference exceeded max depth (10)")
 
 	// Get the template of local infinite loop template.
 	tmplHolder = wfv1.WorkflowStep{TemplateRef: &wfv1.TemplateRef{Name: "some-workflow-template", Template: "infinite-local-loop-whalesay"}}
-	_, _, err = ctx.ResolveTemplate(&tmplHolder)
+	_, _, _, err = ctx.ResolveTemplate(&tmplHolder)
 	assert.EqualError(t, err, "template reference exceeded max depth (10)")
 }