Support nested steps workflow parallelism

[WeiTang114]

Related issue: #1035

What is solved:

• Parallelism for nested steps (StepGroup) and DAGs

What is not solved:
- Parallelism for nested DAG

This commit make nested StepGroup workflow parallelism on the outer
workflow able to limit inner workflow execution. This is done by making
checkParallelism called by the inner workflow check if the number of
its running siblings (the nodes with the same parent node) is >=
the parent node's parallelism.

Update 2018/10/27: after the changes thanks to @jessesuen's suggestion, parallelism for nested DAG is supported too!

[CLAassistant]

All committers have signed the CLA.

[jessesuen]

It shouldn't be necessary to change the method signature to include parentName. parentName is the same as woc.wf.Status.Nodes[boundaryID].Name.

[WeiTang114]

Thanks for the reply. I tried to use boundaryID to find the parent node, but I found it not necessarily the case. An example is in the example workflow I used in the issue #1035.

The seq-step (template B) 's boundaryID is the top node (template A), while its parent node is a StepGroup node between it and the top node. I could only find the children nodes with the parentNode, but not with the top node.

I'm afraid I can't express it precisely, so let me show the figure.

Some related logs are here if it helps (see the lines starting with "Evaluating node "):

time="2018-10-27T03:53:03Z" level=info msg="Processing workflow" namespace=default workflow=seq-test-ms4fj
time="2018-10-27T03:53:03Z" level=info msg="Updated phase  -> Running" namespace=default workflow=seq-test-ms4fj
time="2018-10-27T03:53:03Z" level=debug msg="Evaluating node seq-test-ms4fj: template: A, boundaryID: , parentName: " namespace=default workflow=seq-test-ms4fj
time="2018-10-27T03:53:03Z" level=info msg="tmpl type: Steps, parallelism:0xc4207ea8c0, node:<nil>" namespace=default workflow=seq-test-ms4fj
time="2018-10-27T03:53:03Z" level=info msg="Steps node seq-test-ms4fj (seq-test-ms4fj) initialized Running" namespace=default workflow=seq-test-ms4fj
time="2018-10-27T03:53:03Z" level=info msg="StepGroup node seq-test-ms4fj[0] (seq-test-ms4fj-3412503640) initialized Running" namespace=default workflow=seq-test-ms4fj
time="2018-10-27T03:53:03Z" level=error msg="shouldExecute , proceed: true, error: <nil>" namespace=default workflow=seq-test-ms4fj
time="2018-10-27T03:53:03Z" level=debug msg="Evaluating node seq-test-ms4fj[0].seq-step(0:a): template: B, boundaryID: seq-test-ms4fj, parentName: seq-test-ms4fj[0]" namespace=default workflow=seq-test-ms4fj
time="2018-10-27T03:53:03Z" level=info msg="tmpl type: Steps, parallelism:<nil>, node:<nil>" namespace=default workflow=seq-test-ms4fj
time="2018-10-27T03:53:03Z" level=debug msg="counted 0/1 active children in boundary seq-test-ms4fj of parent seq-test-ms4fj[0]" namespace=default workflow=seq-test-ms4fj
time="2018-10-27T03:53:03Z" level=info msg="Steps node seq-test-ms4fj[0].seq-step(0:a) (seq-test-ms4fj-643033778) initialized Running" namespace=default workflow=seq-test-ms4fj
time="2018-10-27T03:53:03Z" level=info msg="StepGroup node seq-test-ms4fj[0].seq-step(0:a)[0] (seq-test-ms4fj-3735306292) initialized Running" namespace=default workflow=seq-test-ms4fj
time="2018-10-27T03:53:03Z" level=error msg="shouldExecute , proceed: true, error: <nil>" namespace=default workflow=seq-test-ms4fj
time="2018-10-27T03:53:03Z" level=debug msg="Evaluating node seq-test-ms4fj[0].seq-step(0:a)[0].jobs(0:1): template: one-job, boundaryID: seq-test-ms4fj-643033778, parentName: seq-test-ms4fj[0].seq-step(0:a)[0]" namespace=default workflow
=seq-test-ms4fj
time="2018-10-27T03:53:03Z" level=info msg="tmpl type: Container, parallelism:<nil>, node:<nil>" namespace=default workflow=seq-test-ms4fj
time="2018-10-27T03:53:03Z" level=debug msg="Executing node seq-test-ms4fj[0].seq-step(0:a)[0].jobs(0:1) with container template: &{one-job {[{seq-id <nil> 0xc42088bdc0 <nil> }] []} {[] [] <nil>} map[] nil {map[] map[]} <nil> [] &Containe
r{Name:,Image:alpine,Command:[/bin/sh -c],Args:[echo a; sleep 10],WorkingDir:,Ports:[],Env:[],Resources:ResourceRequirements{Limits:ResourceList{},Requests:ResourceList{},},VolumeMounts:[],LivenessProbe:nil,ReadinessProbe:nil,Lifecycle:ni
l,TerminationMessagePath:,ImagePullPolicy:,SecurityContext:nil,Stdin:false,StdinOnce:false,TTY:false,EnvFrom:[],TerminationMessagePolicy:,VolumeDevices:[],} <nil> <nil> <nil> <nil> [] <nil> <nil> <nil> <nil> []}\n" namespace=default workf
low=seq-test-ms4fj
time="2018-10-27T03:53:03Z" level=debug msg="Creating Pod: seq-test-ms4fj[0].seq-step(0:a)[0].jobs(0:1) (seq-test-ms4fj-2529398666)" namespace=default workflow=seq-test-ms4fj
time="2018-10-27T03:53:03Z" level=info msg="Created pod: seq-test-ms4fj[0].seq-step(0:a)[0].jobs(0:1) (seq-test-ms4fj-2529398666)" namespace=default workflow=seq-test-ms4fj
time="2018-10-27T03:53:03Z" level=info msg="Pod node seq-test-ms4fj[0].seq-step(0:a)[0].jobs(0:1) (seq-test-ms4fj-2529398666) initialized Pending" namespace=default workflow=seq-test-ms4fj
time="2018-10-27T03:53:03Z" level=error msg="shouldExecute , proceed: true, error: <nil>" namespace=default workflow=seq-test-ms4fj
time="2018-10-27T03:53:03Z" level=debug msg="Evaluating node seq-test-ms4fj[0].seq-step(0:a)[0].jobs(1:2): template: one-job, boundaryID: seq-test-ms4fj-643033778, parentName: seq-test-ms4fj[0].seq-step(0:a)[0]" namespace=default workflow
=seq-test-ms4fj
time="2018-10-27T03:53:03Z" level=info msg="tmpl type: Container, parallelism:<nil>, node:<nil>" namespace=default workflow=seq-test-ms4fj
time="2018-10-27T03:53:03Z" level=debug msg="Executing node seq-test-ms4fj[0].seq-step(0:a)[0].jobs(1:2) with container template: &{one-job {[{seq-id <nil> 0xc420b09c30 <nil> }] []} {[] [] <nil>} map[] nil {map[] map[]} <nil> [] &Containe
r{Name:,Image:alpine,Command:[/bin/sh -c],Args:[echo a; sleep 10],WorkingDir:,Ports:[],Env:[],Resources:ResourceRequirements{Limits:ResourceList{},Requests:ResourceList{},},VolumeMounts:[],LivenessProbe:nil,ReadinessProbe:nil,Lifecycle:ni
l,TerminationMessagePath:,ImagePullPolicy:,SecurityContext:nil,Stdin:false,StdinOnce:false,TTY:false,EnvFrom:[],TerminationMessagePolicy:,VolumeDevices:[],} <nil> <nil> <nil> <nil> [] <nil> <nil> <nil> <nil> []}\n" namespace=default workf
low=seq-test-ms4fj
time="2018-10-27T03:53:03Z" level=debug msg="Creating Pod: seq-test-ms4fj[0].seq-step(0:a)[0].jobs(1:2) (seq-test-ms4fj-765137104)" namespace=default workflow=seq-test-ms4fj
time="2018-10-27T03:53:03Z" level=info msg="Created pod: seq-test-ms4fj[0].seq-step(0:a)[0].jobs(1:2) (seq-test-ms4fj-765137104)" namespace=default workflow=seq-test-ms4fj

[jessesuen]

It looks like it's possible to miscalculate the parallelism since we are checking active pods independently of child steps/dag template invocations, and not summing up the counts of each to compare against the parallelism limit. I think the calculation needs to be the summation of both pods, as well as dag/step templates for the parallelism calculation to be accurate.

[WeiTang114]

Good idea. I put the two countings together.

[jessesuen]

I think countActiveChildren() should replace the use of countActivePods() and sum the number of all NodeTypePod, NodeTypeSteps, NodeTypeDAG nodes within a boundaryID as an aggregate. It can have similar logic to countActivePods() but modified slightly. Something like:

	activeChildren := 0
	for _, node := range woc.wf.Status.Nodes {
		if boundaryID != "" && node.BoundaryID != boundaryID {
			continue
		}
		switch node.Type {
		case wfv1.NodeTypePod, wfv1.NodeTypeSteps, wfv1.NodeTypeDAG:
		default:
			continue
		}
		switch node.Phase {
		case wfv1.NodePending, wfv1.NodeRunning:
			activeChildren++
		}
	}
	return activeChildren

Then, the existing calls to countActivePods() in checkParallelism() would be replaced with the call to countActiveChildren().

We would then only use countActivePods() when checking against the global parallelism limit.

[WeiTang114]

Thanks! Finding children nodes with the same boundaryID should be able to solve my problem in #1046 (comment). (I was stuck at using the Children field to find children 😅). I'll let you know if it works.

[WeiTang114]

It's working! 😀

[WeiTang114]

Update: After the changes thanks to @jessesuen's suggestion, parallelism for nested DAG is supported too!