From 599120cb6c2dbdc9af33ce92c0e79aae30215b8b Mon Sep 17 00:00:00 2001
From: Daisuke Taniwaki <daisuketaniwaki@gmail.com>
Date: Sat, 12 Oct 2019 17:42:02 +0900
Subject: [PATCH] Store locally referenced template properly

---
 pkg/apis/workflow/v1alpha1/workflow_types.go | 23 +++++++++++---------
 workflow/controller/operator.go              | 10 +++++++--
 2 files changed, 21 insertions(+), 12 deletions(-)

diff --git a/pkg/apis/workflow/v1alpha1/workflow_types.go b/pkg/apis/workflow/v1alpha1/workflow_types.go
index e10ee4c1846d..6792cefe8410 100644
--- a/pkg/apis/workflow/v1alpha1/workflow_types.go
+++ b/pkg/apis/workflow/v1alpha1/workflow_types.go
@@ -76,6 +76,7 @@ type TemplateGetter interface {
 type TemplateHolder interface {
 	GetTemplateName() string
 	GetTemplateRef() *TemplateRef
+	IsResolvable() bool
 }
 
 // Workflow is the definition of a workflow resource
@@ -367,6 +368,10 @@ func (tmpl *Template) GetTemplateRef() *TemplateRef {
 	return tmpl.TemplateRef
 }
 
+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()
@@ -567,6 +572,10 @@ func (step *WorkflowStep) GetTemplateRef() *TemplateRef {
 	return step.TemplateRef
 }
 
+func (step *WorkflowStep) IsResolvable() bool {
+	return true
+}
+
 // Item expands a single workflow step into multiple parallel steps
 // The value of Item can be a map, string, bool, or number
 type Item struct {
@@ -773,16 +782,6 @@ type NodeStatus struct {
 	OutboundNodes []string `json:"outboundNodes,omitempty"`
 }
 
-var _ TemplateHolder = &NodeStatus{}
-
-func (n *NodeStatus) GetTemplateName() string {
-	return n.TemplateName
-}
-
-func (n *NodeStatus) GetTemplateRef() *TemplateRef {
-	return n.TemplateRef
-}
-
 func (n NodeStatus) String() string {
 	return fmt.Sprintf("%s (%s)", n.Name, n.ID)
 }
@@ -1158,6 +1157,10 @@ 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 {
 }
diff --git a/workflow/controller/operator.go b/workflow/controller/operator.go
index 5731552bd662..c1a1846c3fc0 100644
--- a/workflow/controller/operator.go
+++ b/workflow/controller/operator.go
@@ -1315,8 +1315,14 @@ func (woc *wfOperationCtx) initializeExecutableNode(nodeName string, nodeType wf
 	// Store the template for the later use.
 	if node.TemplateRef != nil {
 		node.StoredTemplateID = fmt.Sprintf("%s/%s", node.TemplateRef.Name, node.TemplateRef.Template)
-	} else if node.WorkflowTemplateName != "" {
-		node.StoredTemplateID = fmt.Sprintf("%s/%s", node.WorkflowTemplateName, node.TemplateName)
+	} else if node.TemplateName != "" {
+		if node.WorkflowTemplateName != "" {
+			// Locally resolvable in workflow template level.
+			node.StoredTemplateID = fmt.Sprintf("%s/%s", node.WorkflowTemplateName, node.TemplateName)
+		} else if orgTmpl.IsResolvable() {
+			// Locally resolvable in workflow level.
+			node.StoredTemplateID = fmt.Sprintf("/%s", node.TemplateName)
+		}
 	}
 	if node.StoredTemplateID != "" {
 		baseTemplate := executeTmpl.GetBaseTemplate()