Store locally referenced template properly (#1670)

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 {
 }

workflow/controller/operator.go

@@ -1322,8 +1322,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()