Resolve WorkflowTemplate lazily (argoproj#1655)

workflow/controller/operator.go

@@ -349,7 +349,7 @@ func (woc *wfOperationCtx) getResolvedTemplate(node *wfv1.NodeStatus, tmpl wfv1.
 			woc.log.Debugf("Found a resolved template for node %s", node.Name)
 			if node.WorkflowTemplateName != "" {
 				woc.log.Debugf("Switch the template context to %s", node.WorkflowTemplateName)
-				newTmplCtx, err := tmplCtx.OnWorkflowTemplate(node.WorkflowTemplateName)
+				newTmplCtx, err := tmplCtx.WithLazyWorkflowTemplate(woc.wf.Namespace, node.WorkflowTemplateName)
 				if err != nil {
 					return nil, nil, err
 				}

workflow/controller/steps.go

@@ -98,7 +98,7 @@ func (woc *wfOperationCtx) executeSteps(nodeName string, tmplCtx *templateresolu
 				// We add the aggregate outputs of our children to the scope as a JSON list
 				var childNodes []wfv1.NodeStatus
 				for _, node := range woc.wf.Status.Nodes {
-					if node.BoundaryID == stepsCtx.boundaryID && strings.HasPrefix(node.Name, childNodeName+"(") {
+					if node.BoundaryID == stepsCtx.boundaryID && strings.HasPrefix(node.Name, childNodeName+"(") && node.Type != wfv1.NodeTypeSkipped {
 						childNodes = append(childNodes, node)
 					}
 				}

workflow/templateresolution/context.go

@@ -185,11 +185,7 @@ func (ctx *Context) WithTemplateBase(tmplBase wfv1.TemplateGetter) *Context {
 	return NewContext(ctx.wftmplGetter, tmplBase)
 }
 
-// OnWorkflowTemplate creates new context with the wfv1.WorkflowTemplate of the given name.
-func (ctx *Context) OnWorkflowTemplate(name string) (*Context, error) {
-	wftmpl, err := ctx.wftmplGetter.Get(name)
-	if err != nil {
-		return nil, err
-	}
-	return NewContext(ctx.wftmplGetter, wftmpl), nil
+// WithLazyWorkflowTemplate creates new context with the wfv1.WorkflowTemplate of the given name with lazy loading.
+func (ctx *Context) WithLazyWorkflowTemplate(namespace, name string) (*Context, error) {
+	return NewContext(ctx.wftmplGetter, NewLazyWorkflowTemplate(ctx.wftmplGetter, namespace, name)), nil
 }

workflow/templateresolution/context_test.go

@@ -6,10 +6,10 @@ import (
 	wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
 	wfclientset "github.com/argoproj/argo/pkg/client/clientset/versioned"
 	fakewfclientset "github.com/argoproj/argo/pkg/client/clientset/versioned/fake"
-	"sigs.k8s.io/yaml"
 	"github.com/stretchr/testify/assert"
 	apierr "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/yaml"
 )
 
 func createWorkflowTemplate(wfClientset wfclientset.Interface, yamlStr string) error {
@@ -411,13 +411,10 @@ func TestOnWorkflowTemplate(t *testing.T) {
 	}
 
 	// Get the template base of existing template name.
-	newCtx, err := ctx.OnWorkflowTemplate("another-workflow-template")
+	newCtx, err := ctx.WithLazyWorkflowTemplate("namespace", "another-workflow-template")
 	if err != nil {
 		t.Fatal(err)
 	}
-	wftmpl, ok := newCtx.tmplBase.(*wfv1.WorkflowTemplate)
-	if !assert.True(t, ok) {
-		t.Fatal("tmplBase is not a WorkflowTemplate")
-	}
-	assert.Equal(t, "another-workflow-template", wftmpl.Name)
+	tmpl := newCtx.tmplBase.GetTemplateByName("whalesay")
+	assert.NotNil(t, tmpl)
 }

workflow/templateresolution/lazyworkflowtemplate.go

@@ -0,0 +1,66 @@
+package templateresolution
+
+import (
+	"github.com/argoproj/argo/pkg/apis/workflow"
+	v1alpha1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
+	wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+)
+
+// lazyWorkflowTemplate retrieves WorkflowTemplate lazily.
+type lazyWorkflowTemplate struct {
+	// wftmplGetter is a proxied WorkflowTemplate getter.
+	wftmplGetter WorkflowTemplateNamespacedGetter
+	// wftmpl is a cache of retrieved WorkflowTemplate.
+	wftmpl *wfv1.WorkflowTemplate
+	// namespace is the namespace of the WorkflowTemplate.
+	namespace string
+	// name is the name of the WorkflowTemplate.
+	name string
+}
+
+var _ wfv1.TemplateGetter = &lazyWorkflowTemplate{}
+
+// NewLazyWorkflowTemplate is a public constructor of lazyWorkflowTemplate.
+func NewLazyWorkflowTemplate(wftmplGetter WorkflowTemplateNamespacedGetter, namespace, name string) *lazyWorkflowTemplate {
+	return &lazyWorkflowTemplate{
+		wftmplGetter: wftmplGetter,
+		namespace:    namespace,
+		name:         name,
+	}
+}
+
+// GetNamespace returns the namespace of the WorkflowTemplate.
+func (lwt *lazyWorkflowTemplate) GetNamespace() string {
+	return lwt.namespace
+}
+
+// GetName returns the name of the WorkflowTemplate.
+func (lwt *lazyWorkflowTemplate) GetName() string {
+	return lwt.name
+}
+
+// GroupVersionKind returns a GroupVersionKind of WorkflowTemplate.
+func (lwt *lazyWorkflowTemplate) GroupVersionKind() schema.GroupVersionKind {
+	return v1alpha1.SchemeGroupVersion.WithKind(workflow.WorkflowTemplateKind)
+}
+
+// GetTemplateByName retrieves a defined template by its name
+func (lwt *lazyWorkflowTemplate) GetTemplateByName(name string) *wfv1.Template {
+	err := lwt.ensureWorkflowTemplate()
+	if err != nil {
+		return nil
+	}
+	return lwt.wftmpl.GetTemplateByName(name)
+}
+
+func (lwt *lazyWorkflowTemplate) ensureWorkflowTemplate() error {
+	if lwt.wftmpl == nil {
+		wftmpl, err := lwt.wftmplGetter.Get(lwt.name)
+		if err != nil {
+			return err
+		}
+		lwt.wftmpl = wftmpl
+	}
+	return nil
+}