Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix(controller): Fix template resolution for step groups. Fixes #1868 #1920

Merged
merged 20 commits into from
Jan 28, 2020
Merged

fix(controller): Fix template resolution for step groups. Fixes #1868 #1920

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
20 commits
Select commit Hold shift + click to select a range
80d66ae
Use correct template context
dtaniwaki Jan 8, 2020
609c720
Set template scoe in non-executable nodes
dtaniwaki Jan 8, 2020
732015c
Fix tests
dtaniwaki Jan 8, 2020
d01c00e
Use appropriate context
dtaniwaki Jan 8, 2020
a2550fe
Use appropriate template context in parallelism check
dtaniwaki Jan 8, 2020
1e9fc7d
Avoid template context misuse
dtaniwaki Jan 8, 2020
dc84fef
Merge remote-tracking branch 'upstream/master' into fix-template-reso…
dtaniwaki Jan 8, 2020
2cdb53e
Refactor the arguments
dtaniwaki Jan 8, 2020
642c442
Fix tests
dtaniwaki Jan 8, 2020
d39a6f7
Fix tests
dtaniwaki Jan 9, 2020
396faa0
Fix incorrect node type
dtaniwaki Jan 10, 2020
5355370
Add tests
dtaniwaki Jan 10, 2020
937ba2f
Merge remote-tracking branch 'upstream/master' into fix-template-reso…
dtaniwaki Jan 15, 2020
d3e3f67
Merge remote-tracking branch 'upstream/master' into fix-template-reso…
dtaniwaki Jan 16, 2020
e0a981f
Merge remote-tracking branch 'upstream/master' into fix-template-reso…
dtaniwaki Jan 23, 2020
4b80419
FFix a lint issue
dtaniwaki Jan 23, 2020
6ffa9cd
Merge branch 'master' into fix-template-resolution-in-step-group
alexec Jan 23, 2020
7e647a0
Merge branch 'master' into fix-template-resolution-in-step-group
dtaniwaki Jan 24, 2020
9647545
Merge branch 'master' into fix-template-resolution-in-step-group
alexec Jan 24, 2020
b00f942
Merge branch 'master' into fix-template-resolution-in-step-group
alexec Jan 28, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 11 additions & 8 deletions workflow/controller/dag.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -204,7 +204,7 @@ func (d *dagContext) hasMoreRetries(node *wfv1.NodeStatus) bool {
func (woc *wfOperationCtx) executeDAG(nodeName string, tmplCtx *templateresolution.Context, templateScope string, tmpl *wfv1.Template, orgTmpl wfv1.TemplateHolder, boundaryID string) (*wfv1.NodeStatus, error) {
node := woc.getNodeByName(nodeName)
if node == nil {
node = woc.initializeExecutableNode(nodeName, wfv1.NodeTypeSteps, templateScope, tmpl, orgTmpl, boundaryID, wfv1.NodeRunning)
node = woc.initializeExecutableNode(nodeName, wfv1.NodeTypeDAG, templateScope, tmpl, orgTmpl, boundaryID, wfv1.NodeRunning)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this might fix #2056

}

defer func() {
Expand Down Expand Up @@ -307,7 +307,7 @@ func (woc *wfOperationCtx) executeDAGTask(dagCtx *dagContext, taskName string) {
if node != nil && node.Completed() {
// Run the node's onExit node, if any. Only leaf nodes will have their onExit nodes executed here. Nodes that
// have dependencies will have their onExit nodes executed below
hasOnExitNode, onExitNode, err := woc.runOnExitNode(task.Name, task.OnExit, dagCtx.boundaryID)
hasOnExitNode, onExitNode, err := woc.runOnExitNode(task.Name, task.OnExit, dagCtx.boundaryID, dagCtx.tmplCtx)
if hasOnExitNode && (onExitNode == nil || !onExitNode.Completed() || err != nil) {
// The onExit node is either not complete or has errored out, return.
return
Expand All @@ -325,7 +325,7 @@ func (woc *wfOperationCtx) executeDAGTask(dagCtx *dagContext, taskName string) {
depTask := dagCtx.getTask(depName)
// Run the node's onExit node, if any. Only nodes that have dependencies will have their onExit nodes
// executed here. Leaf nodes will have their onExit nodes executed above
hasOnExitNode, onExitNode, err := woc.runOnExitNode(depTask.Name, depTask.OnExit, dagCtx.boundaryID)
hasOnExitNode, onExitNode, err := woc.runOnExitNode(depTask.Name, depTask.OnExit, dagCtx.boundaryID, dagCtx.tmplCtx)
if hasOnExitNode && (onExitNode == nil || !onExitNode.Completed() || err != nil) {
// The onExit node is either not complete or has errored out, return.
return
Expand Down Expand Up @@ -383,10 +383,13 @@ func (woc *wfOperationCtx) executeDAGTask(dagCtx *dagContext, taskName string) {
}
}

// The template scope of this dag.
dagTemplateScope := dagCtx.tmplCtx.GetCurrentTemplateBase().GetTemplateScope()

// First resolve/substitute params/artifacts from our dependencies
newTask, err := woc.resolveDependencyReferences(dagCtx, task)
if err != nil {
woc.initializeNode(nodeName, wfv1.NodeTypeSkipped, task, dagCtx.boundaryID, wfv1.NodeError, err.Error())
woc.initializeNode(nodeName, wfv1.NodeTypeSkipped, dagTemplateScope, task, dagCtx.boundaryID, wfv1.NodeError, err.Error())
connectDependencies(nodeName)
return
}
Expand All @@ -395,7 +398,7 @@ func (woc *wfOperationCtx) executeDAGTask(dagCtx *dagContext, taskName string) {
// expandedTasks will be a single element list of the same task
expandedTasks, err := woc.expandTask(*newTask)
if err != nil {
woc.initializeNode(nodeName, wfv1.NodeTypeSkipped, task, dagCtx.boundaryID, wfv1.NodeError, err.Error())
woc.initializeNode(nodeName, wfv1.NodeTypeSkipped, dagTemplateScope, task, dagCtx.boundaryID, wfv1.NodeError, err.Error())
connectDependencies(nodeName)
return
}
Expand All @@ -406,7 +409,7 @@ func (woc *wfOperationCtx) executeDAGTask(dagCtx *dagContext, taskName string) {
if len(task.WithItems) > 0 || task.WithParam != "" || task.WithSequence != nil {
if taskGroupNode == nil {
connectDependencies(nodeName)
taskGroupNode = woc.initializeNode(nodeName, wfv1.NodeTypeTaskGroup, task, dagCtx.boundaryID, wfv1.NodeRunning, "")
taskGroupNode = woc.initializeNode(nodeName, wfv1.NodeTypeTaskGroup, dagTemplateScope, task, dagCtx.boundaryID, wfv1.NodeRunning, "")
}
}

Expand All @@ -421,12 +424,12 @@ func (woc *wfOperationCtx) executeDAGTask(dagCtx *dagContext, taskName string) {
// Check the task's when clause to decide if it should execute
proceed, err := shouldExecute(t.When)
if err != nil {
woc.initializeNode(taskNodeName, wfv1.NodeTypeSkipped, task, dagCtx.boundaryID, wfv1.NodeError, err.Error())
woc.initializeNode(taskNodeName, wfv1.NodeTypeSkipped, dagTemplateScope, task, dagCtx.boundaryID, wfv1.NodeError, err.Error())
continue
}
if !proceed {
skipReason := fmt.Sprintf("when '%s' evaluated false", t.When)
woc.initializeNode(taskNodeName, wfv1.NodeTypeSkipped, task, dagCtx.boundaryID, wfv1.NodeSkipped, skipReason)
woc.initializeNode(taskNodeName, wfv1.NodeTypeSkipped, dagTemplateScope, task, dagCtx.boundaryID, wfv1.NodeSkipped, skipReason)
continue
}
}
Expand Down
51 changes: 30 additions & 21 deletions workflow/controller/operator.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,10 +78,6 @@ type wfOperationCtx struct {
// workflowDeadline is the deadline which the workflow is expected to complete before we
// terminate the workflow.
workflowDeadline *time.Time

// tmplCtx is the context of template search.
tmplCtx *templateresolution.Context

// auditLogger is the argo audit logger
auditLogger *argo.AuditLogger
}
Expand Down Expand Up @@ -121,7 +117,6 @@ func newWorkflowOperationCtx(wf *wfv1.Workflow, wfc *WorkflowController) *wfOper
deadline: time.Now().UTC().Add(maxOperationTime),
auditLogger: argo.NewAuditLogger(wf.ObjectMeta.Namespace, wfc.kubeclientset, wf.ObjectMeta.Name),
}
woc.tmplCtx = templateresolution.NewContext(wfc.wftmplInformer.Lister().WorkflowTemplates(wf.Namespace), wf, &woc)

if woc.wf.Status.Nodes == nil {
woc.wf.Status.Nodes = make(map[string]wfv1.NodeStatus)
Expand Down Expand Up @@ -228,9 +223,12 @@ func (woc *wfOperationCtx) operate() {
return
}

// Create a starting template context.
tmplCtx := woc.createTemplateContext("")

var workflowStatus wfv1.NodePhase
var workflowMessage string
node, err := woc.executeTemplate(woc.wf.ObjectMeta.Name, &wfv1.Template{Template: woc.wf.Spec.Entrypoint}, woc.tmplCtx, woc.wf.Spec.Arguments, "")
node, err := woc.executeTemplate(woc.wf.ObjectMeta.Name, &wfv1.Template{Template: woc.wf.Spec.Entrypoint}, tmplCtx, woc.wf.Spec.Arguments, "")
if err != nil {
msg := fmt.Sprintf("%s error in entry template execution: %+v", woc.wf.Name, err)
// the error are handled in the callee so just log it.
Expand Down Expand Up @@ -266,7 +264,7 @@ func (woc *wfOperationCtx) operate() {
}
woc.log.Infof("Running OnExit handler: %s", woc.wf.Spec.OnExit)
onExitNodeName := woc.wf.ObjectMeta.Name + ".onExit"
onExitNode, err = woc.executeTemplate(onExitNodeName, &wfv1.Template{Template: woc.wf.Spec.OnExit}, woc.tmplCtx, woc.wf.Spec.Arguments, "")
onExitNode, err = woc.executeTemplate(onExitNodeName, &wfv1.Template{Template: woc.wf.Spec.OnExit}, tmplCtx, woc.wf.Spec.Arguments, "")
if err != nil {
// the error are handled in the callee so just log it.
woc.log.Errorf("%s error in exit template execution: %+v", woc.wf.Name, err)
Expand Down Expand Up @@ -1241,7 +1239,7 @@ func (woc *wfOperationCtx) executeTemplate(nodeName string, orgTmpl wfv1.Templat

newTmplCtx, resolvedTmpl, err := tmplCtx.ResolveTemplate(orgTmpl)
if err != nil {
return woc.initializeNodeOrMarkError(node, nodeName, wfv1.NodeTypeSkipped, orgTmpl, boundaryID, err), err
return woc.initializeNodeOrMarkError(node, nodeName, wfv1.NodeTypeSkipped, templateScope, orgTmpl, boundaryID, err), err
}

localParams := make(map[string]string)
Expand All @@ -1253,7 +1251,7 @@ func (woc *wfOperationCtx) executeTemplate(nodeName string, orgTmpl wfv1.Templat
// Inputs has been processed with arguments already, so pass empty arguments.
processedTmpl, err := common.ProcessArgs(resolvedTmpl, &args, woc.globalParams, localParams, false)
if err != nil {
return woc.initializeNodeOrMarkError(node, nodeName, wfv1.NodeTypeSkipped, orgTmpl, boundaryID, err), err
return woc.initializeNodeOrMarkError(node, nodeName, wfv1.NodeTypeSkipped, templateScope, orgTmpl, boundaryID, err), err
}

// Check if we exceeded template or workflow parallelism and immediately return if we did
Expand Down Expand Up @@ -1303,7 +1301,7 @@ func (woc *wfOperationCtx) executeTemplate(nodeName string, orgTmpl wfv1.Templat
if processedTmpl.IsPodType() {
processedTmpl, err = common.SubstituteParams(processedTmpl, map[string]string{}, map[string]string{common.LocalVarPodName: woc.wf.NodeID(nodeName)})
if err != nil {
return woc.initializeNodeOrMarkError(node, nodeName, wfv1.NodeTypeSkipped, orgTmpl, boundaryID, err), err
return woc.initializeNodeOrMarkError(node, nodeName, wfv1.NodeTypeSkipped, templateScope, orgTmpl, boundaryID, err), err
}
}
}
Expand All @@ -1323,7 +1321,7 @@ func (woc *wfOperationCtx) executeTemplate(nodeName string, orgTmpl wfv1.Templat
node, err = woc.executeSuspend(nodeName, templateScope, processedTmpl, orgTmpl, boundaryID)
default:
err = errors.Errorf(errors.CodeBadRequest, "Template '%s' missing specification", processedTmpl.Name)
return woc.initializeNode(nodeName, wfv1.NodeTypeSkipped, orgTmpl, boundaryID, wfv1.NodeError, err.Error()), err
return woc.initializeNode(nodeName, wfv1.NodeTypeSkipped, templateScope, orgTmpl, boundaryID, wfv1.NodeError, err.Error()), err
}
if err != nil {
node = woc.markNodeError(node.Name, err)
Expand Down Expand Up @@ -1427,15 +1425,13 @@ var stepsOrDagSeparator = regexp.MustCompile(`^(\[\d+\])?\.`)

// initializeExecutableNode initializes a node and stores the template.
func (woc *wfOperationCtx) initializeExecutableNode(nodeName string, nodeType wfv1.NodeType, templateScope string, executeTmpl *wfv1.Template, orgTmpl wfv1.TemplateHolder, boundaryID string, phase wfv1.NodePhase, messages ...string) *wfv1.NodeStatus {
node := woc.initializeNode(nodeName, nodeType, orgTmpl, boundaryID, phase)
node := woc.initializeNode(nodeName, nodeType, templateScope, orgTmpl, boundaryID, phase)

// Set the input values to the node.
if executeTmpl.Inputs.HasInputs() {
node.Inputs = executeTmpl.Inputs.DeepCopy()
}

node.TemplateScope = templateScope

// Update the node
woc.wf.Status.Nodes[node.ID] = *node
woc.updated = true
Expand All @@ -1444,14 +1440,14 @@ func (woc *wfOperationCtx) initializeExecutableNode(nodeName string, nodeType wf
}

// initializeNodeOrMarkError initializes an error node or mark a node if it already exists.
func (woc *wfOperationCtx) initializeNodeOrMarkError(node *wfv1.NodeStatus, nodeName string, nodeType wfv1.NodeType, orgTmpl wfv1.TemplateHolder, boundaryID string, err error) *wfv1.NodeStatus {
func (woc *wfOperationCtx) initializeNodeOrMarkError(node *wfv1.NodeStatus, nodeName string, nodeType wfv1.NodeType, templateScope string, orgTmpl wfv1.TemplateHolder, boundaryID string, err error) *wfv1.NodeStatus {
if node != nil {
return woc.markNodeError(nodeName, err)
}
return woc.initializeNode(nodeName, wfv1.NodeTypeSkipped, orgTmpl, boundaryID, wfv1.NodeError, err.Error())
return woc.initializeNode(nodeName, wfv1.NodeTypeSkipped, templateScope, orgTmpl, boundaryID, wfv1.NodeError, err.Error())
}

func (woc *wfOperationCtx) initializeNode(nodeName string, nodeType wfv1.NodeType, orgTmpl wfv1.TemplateHolder, boundaryID string, phase wfv1.NodePhase, messages ...string) *wfv1.NodeStatus {
func (woc *wfOperationCtx) initializeNode(nodeName string, nodeType wfv1.NodeType, templateScope string, orgTmpl wfv1.TemplateHolder, boundaryID string, phase wfv1.NodePhase, messages ...string) *wfv1.NodeStatus {
woc.log.Debugf("Initializing node %s: template: %s, boundaryID: %s", nodeName, common.GetTemplateHolderString(orgTmpl), boundaryID)

nodeID := woc.wf.NodeID(nodeName)
Expand Down Expand Up @@ -1480,6 +1476,8 @@ func (woc *wfOperationCtx) initializeNode(nodeName string, nodeType wfv1.NodeTyp
node.DisplayName = nodeName
}

node.TemplateScope = templateScope

if node.Completed() && node.FinishedAt.IsZero() {
node.FinishedAt = node.StartedAt
}
Expand Down Expand Up @@ -1552,7 +1550,8 @@ func (woc *wfOperationCtx) checkParallelism(tmpl *wfv1.Template, node *wfv1.Node
if !ok {
return errors.InternalError("boundaryNode not found")
}
_, boundaryTemplate, err := woc.tmplCtx.ResolveTemplate(&boundaryNode)
tmplCtx := woc.createTemplateContext(boundaryNode.TemplateScope)
_, boundaryTemplate, err := tmplCtx.ResolveTemplate(&boundaryNode)
if err != nil {
return err
}
Expand Down Expand Up @@ -1691,7 +1690,8 @@ func (woc *wfOperationCtx) executeScript(nodeName string, templateScope string,

includeScriptOutput := false
if boundaryNode, ok := woc.wf.Status.Nodes[boundaryID]; ok {
_, parentTemplate, err := woc.tmplCtx.ResolveTemplate(&boundaryNode)
tmplCtx := woc.createTemplateContext(boundaryNode.TemplateScope)
_, parentTemplate, err := tmplCtx.ResolveTemplate(&boundaryNode)
if err != nil {
return node, err
}
Expand Down Expand Up @@ -2133,11 +2133,20 @@ func (woc *wfOperationCtx) substituteParamsInVolumes(params map[string]string) e
return nil
}

func (woc *wfOperationCtx) runOnExitNode(parentName, templateRef, boundaryID string) (bool, *wfv1.NodeStatus, error) {
// createTemplateContext creates a new template context.
func (woc *wfOperationCtx) createTemplateContext(templateScope string) *templateresolution.Context {
ctx := templateresolution.NewContext(woc.controller.wftmplInformer.Lister().WorkflowTemplates(woc.wf.Namespace), woc.wf, woc)
if templateScope != "" {
ctx = ctx.WithLazyWorkflowTemplate(woc.wf.Namespace, templateScope)
}
return ctx
}

func (woc *wfOperationCtx) runOnExitNode(parentName, templateRef, boundaryID string, tmplCtx *templateresolution.Context) (bool, *wfv1.NodeStatus, error) {
if templateRef != "" {
woc.log.Infof("Running OnExit handler: %s", templateRef)
onExitNodeName := parentName + ".onExit"
onExitNode, err := woc.executeTemplate(onExitNodeName, &wfv1.Template{Template: templateRef}, woc.tmplCtx, woc.wf.Spec.Arguments, boundaryID)
onExitNode, err := woc.executeTemplate(onExitNodeName, &wfv1.Template{Template: templateRef}, tmplCtx, woc.wf.Spec.Arguments, boundaryID)
return true, onExitNode, err
}
return false, nil, nil
Expand Down
Loading