Fix inputs and arguments during template resolution (argoproj#1545)

workflow/common/util.go

@@ -537,15 +537,6 @@ func MergeReferredTemplate(tmpl *wfv1.Template, referred *wfv1.Template) (*wfv1.
  newTmpl := referred.DeepCopy()
 
  newTmpl.Name = tmpl.Name
-
- emptymap := map[string]string{}
- newTmpl, err := ProcessArgs(newTmpl, &tmpl.Arguments, emptymap, emptymap, false)
- if err != nil {
- return nil, err
- }
- newTmpl.Arguments = wfv1.Arguments{}
-
- newTmpl.Inputs = *tmpl.Inputs.DeepCopy()
  newTmpl.Outputs = *tmpl.Outputs.DeepCopy()
 
  if len(tmpl.NodeSelector) > 0 {

workflow/controller/operator.go

@@ -1083,12 +1083,12 @@ func (woc *wfOperationCtx) executeTemplate(nodeName string, orgTmpl wfv1.Templat
  return node, ErrDeadlineExceeded
  }
 
- newTmplCtx, tmpl, err := tmplCtx.ResolveTemplate(orgTmpl)
+ localParams := make(map[string]string)
+ newTmplCtx, tmpl, err := tmplCtx.ResolveTemplate(orgTmpl, &args, woc.globalParams, localParams, false)
  if err != nil {
  return woc.initializeNode(nodeName, wfv1.NodeTypeSkipped, orgTmpl, boundaryID, wfv1.NodeError, err.Error()), err
  }
  // Perform parameter substitution of the template.
- localParams := make(map[string]string)
  if tmpl.IsPodType() {
  localParams[common.LocalVarPodName] = woc.wf.NodeID(nodeName)
  }

workflow/templateresolution/context.go

@@ -115,12 +115,12 @@ func (ctx *Context) GetTemplateBase(tmplHolder wfv1.TemplateHolder) (wfv1.Templa
 
 // ResolveTemplate digs into referenes and returns a merged template.
 // This method is the public start point of template resolution.
-func (ctx *Context) ResolveTemplate(tmplHolder wfv1.TemplateHolder) (*Context, *wfv1.Template, error) {
- return ctx.resolveTemplateImpl(tmplHolder, 0)
+func (ctx *Context) ResolveTemplate(tmplHolder wfv1.TemplateHolder, args wfv1.ArgumentsProvider, globalParams, localParams map[string]string, validateOnly bool) (*Context, *wfv1.Template, error) {
+ return ctx.resolveTemplateImpl(tmplHolder, args, globalParams, localParams, validateOnly, 0)
 }
 
 // resolveTemplateImpl digs into referenes and returns a merged template.
-func (ctx *Context) resolveTemplateImpl(tmplHolder wfv1.TemplateHolder, depth int) (*Context, *wfv1.Template, error) {
+func (ctx *Context) resolveTemplateImpl(tmplHolder wfv1.TemplateHolder, args wfv1.ArgumentsProvider, globalParams, localParams map[string]string, validateOnly bool, depth int) (*Context, *wfv1.Template, error) {
  // Avoid infinite references
  if depth > maxResolveDepth {
  return nil, nil, errors.Errorf(errors.CodeBadRequest, "template reference exceeded max depth (%d)", maxResolveDepth)
@@ -139,13 +139,21 @@ func (ctx *Context) resolveTemplateImpl(tmplHolder wfv1.TemplateHolder, depth in
  }
  newTmplCtx := ctx.WithTemplateBase(newTmplBase)
 
+ tmpTmpl := &wfv1.Template{Inputs: *tmpl.Inputs.DeepCopy(), Arguments: *tmpl.Arguments.DeepCopy()}
+ processedTmpl, err := common.ProcessArgs(tmpTmpl, args, globalParams, localParams, validateOnly)
+ if err != nil {
+ return nil, nil, err
+ }
+ tmpl.Inputs = processedTmpl.Inputs
+ tmpl.Arguments = processedTmpl.Arguments
+
  // Return a concrete template without digging into it.
  if tmpl.GetType() != wfv1.TemplateTypeUnknown {
  return newTmplCtx, tmpl, nil
  }
 
  // Dig into nested references with new template base.
- finalTmplCtx, newTmpl, err := newTmplCtx.resolveTemplateImpl(tmpl, depth+1)
+ finalTmplCtx, newTmpl, err := newTmplCtx.resolveTemplateImpl(tmpl, &tmpl.Arguments, globalParams, localParams, validateOnly, depth+1)
  if err != nil {
  return nil, nil, err
  }

workflow/templateresolution/context_test.go

@@ -256,6 +256,7 @@ func TestGetTemplateBase(t *testing.T) {
 }
 
 func TestResolveTemplate(t *testing.T) {
+ emptymap := map[string]string{}
  wfClientset := fakewfclientset.NewSimpleClientset()
  err := createWorkflowTemplate(wfClientset, anotherWorkflowTemplateYaml)
  if err != nil {
@@ -270,7 +271,7 @@ func TestResolveTemplate(t *testing.T) {
 
  // Get the template of template name.
  tmplHolder := wfv1.Template{Template: "whalesay"}
- ctx, tmpl, err := ctx.ResolveTemplate(&tmplHolder)
+ ctx, tmpl, err := ctx.ResolveTemplate(&tmplHolder, &wfv1.Arguments{}, emptymap, emptymap, false)
  if !assert.NoError(t, err) {
  t.Fatal(err)
  }
@@ -283,7 +284,7 @@ func TestResolveTemplate(t *testing.T) {
 
  // Get the template of template reference.
  tmplHolder = wfv1.Template{TemplateRef: &wfv1.TemplateRef{Name: "some-workflow-template", Template: "whalesay"}}
- ctx, tmpl, err = ctx.ResolveTemplate(&tmplHolder)
+ ctx, tmpl, err = ctx.ResolveTemplate(&tmplHolder, &wfv1.Arguments{}, emptymap, emptymap, false)
  if !assert.NoError(t, err) {
  t.Fatal(err)
  }
@@ -297,7 +298,7 @@ func TestResolveTemplate(t *testing.T) {
 
  // Get the template of local nested template reference.
  tmplHolder = wfv1.Template{TemplateRef: &wfv1.TemplateRef{Name: "some-workflow-template", Template: "local-whalesay"}}
- ctx, tmpl, err = ctx.ResolveTemplate(&tmplHolder)
+ ctx, tmpl, err = ctx.ResolveTemplate(&tmplHolder, &wfv1.Arguments{}, emptymap, emptymap, false)
  if !assert.NoError(t, err) {
  t.Fatal(err)
  }
@@ -311,7 +312,7 @@ func TestResolveTemplate(t *testing.T) {
 
  // Get the template of nested template reference.
  tmplHolder = wfv1.Template{TemplateRef: &wfv1.TemplateRef{Name: "some-workflow-template", Template: "another-whalesay"}}
- ctx, tmpl, err = ctx.ResolveTemplate(&tmplHolder)
+ ctx, tmpl, err = ctx.ResolveTemplate(&tmplHolder, &wfv1.Arguments{}, emptymap, emptymap, false)
  if !assert.NoError(t, err) {
  t.Fatal(err)
  }
@@ -327,7 +328,8 @@ func TestResolveTemplate(t *testing.T) {
  tmplHolder = wfv1.Template{
  TemplateRef: &wfv1.TemplateRef{Name: "some-workflow-template", Template: "whalesay-with-arguments"},
  }
- ctx, tmpl, err = ctx.ResolveTemplate(&tmplHolder)
+ msgValue := "from-args"
+ ctx, tmpl, err = ctx.ResolveTemplate(&tmplHolder, &wfv1.Arguments{Parameters: []wfv1.Parameter{{Name: "message", Value: &msgValue}}}, emptymap, emptymap, false)
  if !assert.NoError(t, err) {
  t.Fatal(err)
  }
@@ -337,13 +339,14 @@ func TestResolveTemplate(t *testing.T) {
  }
  assert.Equal(t, "some-workflow-template", wftmpl.Name)
  assert.Equal(t, "whalesay-with-arguments", tmpl.Name)
- assert.Equal(t, []string{"{{inputs.parameters.message}}-foo"}, tmpl.Container.Args)
+ assert.Equal(t, "from-args-foo", *tmpl.Inputs.Parameters[0].Value)
+ assert.Equal(t, []string{"{{inputs.parameters.message}}"}, tmpl.Container.Args)
 
  // Get the template of nested template reference with arguments.
  tmplHolder = wfv1.Template{
  TemplateRef: &wfv1.TemplateRef{Name: "some-workflow-template", Template: "nested-whalesay-with-arguments"},
  }
- ctx, tmpl, err = ctx.ResolveTemplate(&tmplHolder)
+ ctx, tmpl, err = ctx.ResolveTemplate(&tmplHolder, &wfv1.Arguments{Parameters: []wfv1.Parameter{{Name: "message", Value: &msgValue}}}, emptymap, emptymap, false)
  if !assert.NoError(t, err) {
  t.Fatal(err)
  }
@@ -353,15 +356,16 @@ func TestResolveTemplate(t *testing.T) {
  }
  assert.Equal(t, "some-workflow-template", wftmpl.Name)
  assert.Equal(t, "nested-whalesay-with-arguments", tmpl.Name)
- assert.Equal(t, []string{"{{inputs.parameters.message}}-bar-foo"}, tmpl.Container.Args)
+ assert.Equal(t, "from-args-bar-foo", *tmpl.Inputs.Parameters[0].Value)
+ assert.Equal(t, []string{"{{inputs.parameters.message}}"}, tmpl.Container.Args)
 
  // Get the template of infinite loop template reference.
  tmplHolder = wfv1.Template{TemplateRef: &wfv1.TemplateRef{Name: "some-workflow-template", Template: "infinite-loop-whalesay"}}
- _, _, err = ctx.ResolveTemplate(&tmplHolder)
+ _, _, err = ctx.ResolveTemplate(&tmplHolder, &wfv1.Arguments{}, emptymap, emptymap, false)
  assert.EqualError(t, err, "template reference exceeded max depth (10)")
 
  // Get the template of local infinite loop template.
  tmplHolder = wfv1.Template{TemplateRef: &wfv1.TemplateRef{Name: "some-workflow-template", Template: "infinite-local-loop-whalesay"}}
- _, _, err = ctx.ResolveTemplate(&tmplHolder)
+ _, _, err = ctx.ResolveTemplate(&tmplHolder, &wfv1.Arguments{}, emptymap, emptymap, false)
  assert.EqualError(t, err, "template reference exceeded max depth (10)")
 }

workflow/validate/validate.go

@@ -283,7 +283,8 @@ func (ctx *templateValidationCtx) validateTemplateHolder(tmplHolder wfv1.Templat
  }
  }
  }
- tmplCtx, tmpl, err := tmplCtx.ResolveTemplate(tmplHolder)
+
+ tmplCtx, tmpl, err := tmplCtx.ResolveTemplate(tmplHolder, args, ctx.globalParams, map[string]string{}, true)
  if err != nil {
  if argoerr, ok := err.(errors.ArgoError); ok && argoerr.Code() == errors.CodeNotFound {
  if tmplRef != nil {
@@ -565,16 +566,22 @@ func (ctx *templateValidationCtx) validateSteps(scope map[string]interface{}, tm
  if err != nil {
  return err
  }
- resolvedTmpl, err := ctx.validateTemplateHolder(&step, tmplCtx, &step.Arguments, scope)
+ resolvedTmpl, err := ctx.validateTemplateHolder(&step, tmplCtx, &FakeArguments{}, scope)
  if err != nil {
  return errors.Errorf(errors.CodeBadRequest, "templates.%s.steps[%d].%s %s", tmpl.Name, i, step.Name, err.Error())
  }
  resolvedTemplates[step.Name] = resolvedTmpl
  }
- for _, step := range stepGroup {
+ for i, step := range stepGroup {
  aggregate := len(step.WithItems) > 0 || step.WithParam != ""
  resolvedTmpl := resolvedTemplates[step.Name]
  ctx.addOutputsToScope(resolvedTmpl, fmt.Sprintf("steps.%s", step.Name), scope, aggregate)
+
+ // Validate the template again with actual arguments.
+ _, err = ctx.validateTemplateHolder(&step, tmplCtx, &step.Arguments, scope)
+ if err != nil {
+ return errors.Errorf(errors.CodeBadRequest, "templates.%s.steps[%d].%s %s", tmpl.Name, i, step.Name, err.Error())
+ }
  }
  }
  return nil