Update Template Resolution Logic

api/openapi-spec/swagger.json

@@ -636,7 +636,7 @@
           "$ref": "#/definitions/io.k8s.apimachinery.pkg.apis.meta.v1.Time"
         },
         "storedTemplateID": {
-          "description": "StoredTemplateID is the ID of stored template.",
+          "description": "StoredTemplateID is the ID of stored template. DEPRECATED: This value is not used anymore.",
           "type": "string"
         },
         "templateName": {
@@ -647,12 +647,16 @@
           "description": "TemplateRef is the reference to the template resource which this node corresponds to. Not applicable to virtual nodes (e.g. Retry, StepGroup)",
           "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.TemplateRef"
         },
+        "templateScope": {
+          "description": "TemplateScope is the template scope in which the template of this node was retrieved.",
+          "type": "string"
+        },
         "type": {
           "description": "Type indicates type of node",
           "type": "string"
         },
         "workflowTemplateName": {
-          "description": "WorkflowTemplateName is the WorkflowTemplate resource name on which the resolved template of this node is retrieved.",
+          "description": "WorkflowTemplateName is the WorkflowTemplate resource name on which the resolved template of this node is retrieved. DEPRECATED: This value is not used anymore.",
           "type": "string"
         }
       }

examples/workflow-template/hello-world.yaml

@@ -9,7 +9,7 @@ spec:
     templateRef:
       name: workflow-template-whalesay-template
       template: whalesay-template
-    arguments:
+    inputs:
       parameters:
       - name: message
         value: "hello world"

examples/workflow-template/local-ref.yaml

@@ -7,10 +7,10 @@ spec:
   templates:
   - name: whalesay
     template: whalesay-template
-    arguments:
+    inputs:
       parameters:
       - name: message
-        value: "hello world"
+        value: "hello from local ref"
   - name: whalesay-template
     inputs:
       parameters:

examples/workflow-template/nested.yaml

@@ -6,10 +6,10 @@ spec:
   entrypoint: whalesay
   templates:
   - name: whalesay
+    inputs:
+      parameters:
+      - name: message
+        value: hello from nested
     templateRef:
       name: workflow-template-nested-template
       template: whalesay-template
-    arguments:
-      parameters:
-      - name: message
-        value: "hello"

examples/workflow-template/templates.yaml

@@ -26,19 +26,11 @@ spec:
     inputs:
       parameters:
       - name: message
-    arguments:
-      parameters:
-      - name: message
-        value: "{{inputs.parameters.message}} hello"
   - name: whalesay-template
     template: whalesay-inner-template
     inputs:
       parameters:
       - name: message
-    arguments:
-      parameters:
-      - name: message
-        value: "{{inputs.parameters.message}} hello"
 ---
 apiVersion: argoproj.io/v1alpha1
 kind: WorkflowTemplate

pkg/apis/workflow/v1alpha1/workflow_template_types.go

@@ -41,3 +41,8 @@ func (wftmpl *WorkflowTemplate) GetTemplateByName(name string) *Template {
 	}
 	return nil
 }
+
+// GetTemplateScope returns the template scope of workflow template.
+func (wftmpl *WorkflowTemplate) GetTemplateScope() string {
+	return wftmpl.Name
+}

pkg/apis/workflow/v1alpha1/workflow_types.go

@@ -69,6 +69,7 @@ type TemplateGetter interface {
 	GetName() string
 	GroupVersionKind() schema.GroupVersionKind
 	GetTemplateByName(name string) *Template
+	GetTemplateScope() string
 }
 
 // TemplateHolder is an interface for holders of templates.
@@ -78,6 +79,12 @@ type TemplateHolder interface {
 	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)
+}
+
 // Workflow is the definition of a workflow resource
 // +genclient
 // +genclient:noStatus
@@ -98,6 +105,7 @@ type WorkflowList struct {
 }
 
 var _ TemplateGetter = &Workflow{}
+var _ TemplateStorage = &Workflow{}
 
 // WorkflowSpec is the specification of a Workflow.
 type WorkflowSpec struct {
@@ -762,11 +770,16 @@ type NodeStatus struct {
 	TemplateRef *TemplateRef `json:"templateRef,omitempty" protobuf:"bytes,6,opt,name=templateRef"`
 
 	// StoredTemplateID is the ID of stored template.
+	// DEPRECATED: This value is not used anymore.
 	StoredTemplateID string `json:"storedTemplateID,omitempty" protobuf:"bytes,18,opt,name=storedTemplateID"`
 
 	// WorkflowTemplateName is the WorkflowTemplate resource name on which the resolved template of this node is retrieved.
+	// DEPRECATED: This value is not used anymore.
 	WorkflowTemplateName string `json:"workflowTemplateName,omitempty" protobuf:"bytes,19,opt,name=workflowTemplateName"`
 
+	// TemplateScope is the template scope in which the template of this node was retrieved.
+	TemplateScope string `json:"templateScope,omitempty" protobuf:"bytes,20,opt,name=templateScope"`
+
 	// Phase a simple, high-level summary of where the node is in its lifecycle.
 	// Can be used as a state machine.
 	Phase NodePhase `json:"phase,omitempty" protobuf:"bytes,7,opt,name=phase,casttype=NodePhase"`
@@ -858,6 +871,20 @@ func (n NodeStatus) CanRetry() bool {
 	return n.Completed() && !n.Successful()
 }
 
+var _ TemplateHolder = &NodeStatus{}
+
+func (n *NodeStatus) GetTemplateName() string {
+	return n.TemplateName
+}
+
+func (n *NodeStatus) GetTemplateRef() *TemplateRef {
+	return n.TemplateRef
+}
+
+func (n *NodeStatus) IsResolvable() bool {
+	return true
+}
+
 // S3Bucket contains the access information required for interfacing with an S3 bucket
 type S3Bucket struct {
 	// Endpoint is the hostname of the bucket endpoint
@@ -1271,6 +1298,11 @@ func (wf *Workflow) GetTemplateByName(name string) *Template {
 	return nil
 }
 
+// GetTemplateScope returns the template scope of workflow.
+func (wf *Workflow) GetTemplateScope() string {
+	return ""
+}
+
 // NodeID creates a deterministic node ID based on a node name
 func (wf *Workflow) NodeID(name string) string {
 	if name == wf.ObjectMeta.Name {
@@ -1281,34 +1313,47 @@ func (wf *Workflow) NodeID(name string) string {
 	return fmt.Sprintf("%s-%v", wf.ObjectMeta.Name, h.Sum32())
 }
 
-// GetStoredTemplate gets a resolved template from stored data.
-func (wf *Workflow) GetStoredTemplate(node *NodeStatus) *Template {
-	id := node.StoredTemplateID
-	if id == "" {
+// 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 == "" {
 		return nil
 	}
-	tmpl, ok := wf.Status.StoredTemplates[id]
-	if ok {
-		return &tmpl
+	tmpl, ok := wf.Status.StoredTemplates[tmplID]
+	if !ok {
+		return nil
 	}
-	return nil
+	return &tmpl
 }
 
-// GetStoredOrLocalTemplate gets a resolved template from stored data or local template.
-func (wf *Workflow) GetStoredOrLocalTemplate(node *NodeStatus) *Template {
-	// Try to find a template from stored data.
-	tmpl := wf.GetStoredTemplate(node)
-	if tmpl != nil {
-		return tmpl
+// 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 == "" {
+		return false, nil
 	}
-	// Try to get template from Workflow.
-	if node.WorkflowTemplateName == "" && node.TemplateName != "" {
-		tmpl := wf.GetTemplateByName(node.TemplateName)
-		if tmpl != nil {
-			return tmpl
+	_, ok := wf.Status.StoredTemplates[tmplID]
+	if !ok {
+		if wf.Status.StoredTemplates == nil {
+			wf.Status.StoredTemplates = map[string]Template{}
 		}
+		wf.Status.StoredTemplates[tmplID] = *tmpl
+		return true, nil
+	}
+	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()
+	if tmplRef != nil {
+		return fmt.Sprintf("%s/%s", tmplRef.Name, tmplRef.Template)
+	} else if templateScope != "" {
+		return fmt.Sprintf("%s/%s", templateScope, holder.GetTemplateName())
+	} else {
+		// Do not store workflow-local templates.
+		return ""
 	}
-	return nil
 }
 
 // ContinueOn defines if a workflow should continue even if a task or step fails/errors.

workflow/common/util.go

@@ -537,7 +537,39 @@ func MergeReferredTemplate(tmpl *wfv1.Template, referred *wfv1.Template) (*wfv1.
 	newTmpl := referred.DeepCopy()
 
 	newTmpl.Name = tmpl.Name
-	newTmpl.Outputs = *tmpl.Outputs.DeepCopy()
+
+	if len(tmpl.Inputs.Parameters) > 0 {
+		parameters := make([]wfv1.Parameter, len(tmpl.Inputs.Parameters))
+		copy(parameters, tmpl.Inputs.Parameters)
+		newTmpl.Inputs.Parameters = parameters
+	}
+	if len(tmpl.Inputs.Artifacts) > 0 {
+		artifacts := make([]wfv1.Artifact, len(tmpl.Inputs.Artifacts))
+		copy(artifacts, tmpl.Inputs.Artifacts)
+		newTmpl.Inputs.Artifacts = artifacts
+	}
+
+	if len(tmpl.Outputs.Parameters) > 0 {
+		parameters := make([]wfv1.Parameter, len(tmpl.Outputs.Parameters))
+		copy(parameters, tmpl.Outputs.Parameters)
+		newTmpl.Outputs.Parameters = parameters
+	}
+	if len(tmpl.Outputs.Artifacts) > 0 {
+		artifacts := make([]wfv1.Artifact, len(tmpl.Outputs.Artifacts))
+		copy(artifacts, tmpl.Outputs.Artifacts)
+		newTmpl.Outputs.Artifacts = artifacts
+	}
+
+	if len(tmpl.Arguments.Parameters) > 0 {
+		parameters := make([]wfv1.Parameter, len(tmpl.Arguments.Parameters))
+		copy(parameters, tmpl.Arguments.Parameters)
+		newTmpl.Arguments.Parameters = parameters
+	}
+	if len(tmpl.Arguments.Artifacts) > 0 {
+		artifacts := make([]wfv1.Artifact, len(tmpl.Arguments.Artifacts))
+		copy(artifacts, tmpl.Arguments.Artifacts)
+		newTmpl.Arguments.Artifacts = artifacts
+	}
 
 	if len(tmpl.NodeSelector) > 0 {
 		m := make(map[string]string, len(tmpl.NodeSelector))