Save stored template ID in nodes

api/openapi-spec/swagger.json

@@ -596,6 +596,10 @@
           "description": "Time at which this node started",
           "$ref": "#/definitions/io.k8s.apimachinery.pkg.apis.meta.v1.Time"
         },
+        "storedTemplateID": {
+          "description": "StoredTemplateID is the ID of stored template.",
+          "type": "string"
+        },
         "templateName": {
           "description": "TemplateName is the template name which this node corresponds to. Not applicable to virtual nodes (e.g. Retry, StepGroup)",
           "type": "string"

pkg/apis/workflow/v1alpha1/workflow_types.go

@@ -721,6 +721,9 @@ type NodeStatus struct {
 	// Not applicable to virtual nodes (e.g. Retry, StepGroup)
 	TemplateRef *TemplateRef `json:"templateRef,omitempty"`
 
+	// StoredTemplateID is the ID of stored template.
+	StoredTemplateID string `json:"storedTemplateID,omitempty"`
+
 	// WorkflowTemplateName is the WorkflowTemplate resource name on which the resolved template of this node is retrieved.
 	WorkflowTemplateName string `json:"workflowTemplateName,omitempty"`
 
@@ -825,16 +828,6 @@ func (n NodeStatus) CanRetry() bool {
 	return n.Completed() && !n.Successful()
 }
 
-// GetBaseTemplateID returns a base template ID if available.
-func (n *NodeStatus) GetBaseTemplateID() string {
-	if n.TemplateRef != nil {
-		return fmt.Sprintf("%s/%s", n.TemplateRef.Name, n.TemplateRef.Template)
-	} else if n.WorkflowTemplateName != "" {
-		return fmt.Sprintf("%s/%s", n.WorkflowTemplateName, n.TemplateName)
-	}
-	return ""
-}
-
 // S3Bucket contains the access information required for interfacing with an S3 bucket
 type S3Bucket struct {
 	// Endpoint is the hostname of the bucket endpoint
@@ -1266,7 +1259,10 @@ func (wf *Workflow) NodeID(name string) string {
 
 // GetStoredTemplate gets a resolved template from stored data.
 func (wf *Workflow) GetStoredTemplate(node *NodeStatus) *Template {
-	id := node.GetBaseTemplateID()
+	id := node.StoredTemplateID
+	if id == "" {
+		return nil
+	}
 	tmpl, ok := wf.Status.StoredTemplates[id]
 	if ok {
 		return &tmpl

workflow/controller/operator.go

@@ -1298,7 +1298,7 @@ func (woc *wfOperationCtx) markWorkflowError(err error, markCompleted bool) {
 // DAG or steps templates. Will match stings with prefix like: [0]. or .
 var stepsOrDagSeparator = regexp.MustCompile(`^(\[\d+\])?\.`)
 
-// initializeExecutableNode initializes a node and stores the base template.
+// initializeExecutableNode initializes a node and stores the template.
 func (woc *wfOperationCtx) initializeExecutableNode(nodeName string, nodeType wfv1.NodeType, tmplCtx *templateresolution.Context, executeTmpl *wfv1.Template, orgTmpl wfv1.TemplateHolder, boundaryID string, phase wfv1.NodePhase, messages ...string) *wfv1.NodeStatus {
 	node := woc.initializeNode(nodeName, nodeType, orgTmpl, boundaryID, phase)
 
@@ -1312,16 +1312,20 @@ func (woc *wfOperationCtx) initializeExecutableNode(nodeName string, nodeType wf
 		node.WorkflowTemplateName = tmplCtx.GetCurrentTemplateBase().GetName()
 	}
 
-	// Store base template for the later use.
-	baseTemplateID := node.GetBaseTemplateID()
-	if baseTemplateID != "" {
+	// 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)
+	}
+	if node.StoredTemplateID != "" {
 		baseTemplate := executeTmpl.GetBaseTemplate()
-		_, exists := woc.wf.Status.StoredTemplates[baseTemplateID]
+		_, exists := woc.wf.Status.StoredTemplates[node.StoredTemplateID]
 		if !exists {
-			woc.log.Infof("Create base template '%s'", baseTemplateID)
-			woc.wf.Status.StoredTemplates[baseTemplateID] = *baseTemplate
+			woc.log.Infof("Create stored template '%s'", node.StoredTemplateID)
+			woc.wf.Status.StoredTemplates[node.StoredTemplateID] = *baseTemplate
 		} else {
-			woc.log.Infof("Base template '%s' already exists", baseTemplateID)
+			woc.log.Infof("Stored template '%s' already exists", node.StoredTemplateID)
 		}
 	}