Fix: refactor the step group in workflow (#3956)

* Fix: refactor the step group in workflow

Signed-off-by: FogDong <dongtianxin.tx@alibaba-inc.com>

* Feat: add tests for discover

Signed-off-by: FogDong <dongtianxin.tx@alibaba-inc.com>

pkg/workflow/tasks/custom/task.go

@@ -103,11 +103,6 @@ func (tr *taskRunner) Pending(ctx wfContext.Context) bool {
 	return tr.checkPending(ctx)
 }
 
-// SubTaskRunners return child step names. it could be null if the step have no sub-step
-func (tr *taskRunner) SubTaskRunners() []wfTypes.TaskRunner {
-	return nil
-}
-
 func (t *TaskLoader) makeTaskGenerator(templ string) (wfTypes.TaskGenerator, error) {
 	return func(wfStep v1beta1.WorkflowStep, genOpt *wfTypes.GeneratorOptions) (wfTypes.TaskRunner, error) {
 

pkg/workflow/tasks/discover.go

@@ -87,7 +87,8 @@ func suspend(step v1beta1.WorkflowStep, opt *types.GeneratorOptions) (types.Task
 	return tr, nil
 }
 
-func stepGroup(step v1beta1.WorkflowStep, opt *types.GeneratorOptions) (types.TaskRunner, error) {
+// StepGroup is the step group runner
+func StepGroup(step v1beta1.WorkflowStep, opt *types.GeneratorOptions) (types.TaskRunner, error) {
 	return &stepGroupTaskRunner{
 		id:             opt.ID,
 		name:           step.Name,
@@ -104,7 +105,7 @@ func newTaskDiscover(ctx monitorContext.Context, providerHandlers providers.Prov
 	return &taskDiscover{
 		builtins: map[string]types.TaskGenerator{
 			types.WorkflowStepTypeSuspend:   suspend,
-			types.WorkflowStepTypeStepGroup: stepGroup,
+			types.WorkflowStepTypeStepGroup: StepGroup,
 		},
 		remoteTaskDiscover: custom.NewTaskLoader(templateLoader.LoadTaskTemplate, pd, providerHandlers, 0, pCtx),
 		templateLoader:     templateLoader,
@@ -149,11 +150,6 @@ func (tr *suspendTaskRunner) Pending(ctx wfContext.Context) bool {
 	return false
 }
 
-// SubTaskRunners return child step names. it could be null if the step have no sub-step
-func (tr *suspendTaskRunner) SubTaskRunners() []types.TaskRunner {
-	return nil
-}
-
 type stepGroupTaskRunner struct {
 	id             string
 	name           string
@@ -165,15 +161,38 @@ func (tr *stepGroupTaskRunner) Name() string {
 	return tr.name
 }
 
-// SubTaskRunners return child step runners. it could be null if the step have no sub-step
-func (tr *stepGroupTaskRunner) SubTaskRunners() []types.TaskRunner {
-	return tr.subTaskRunners
-}
-
 // Run make workflow step group.
 func (tr *stepGroupTaskRunner) Run(ctx wfContext.Context, options *types.TaskRunOptions) (common.StepStatus, *types.Operation, error) {
-	phase := common.WorkflowStepPhaseRunning
-	if tr.subTaskRunners == nil {
+	e := options.Engine
+	if len(tr.subTaskRunners) > 0 {
+		// set sub steps to dag mode for now
+		e.SetParentRunner(tr.name)
+		if err := e.Run(tr.subTaskRunners, true); err != nil {
+			return common.StepStatus{
+				ID:    tr.id,
+				Name:  tr.name,
+				Type:  types.WorkflowStepTypeStepGroup,
+				Phase: common.WorkflowStepPhaseRunning,
+			}, e.GetOperation(), err
+		}
+		e.SetParentRunner("")
+	}
+	stepStatus := e.GetStepStatus(tr.name)
+	var phase common.WorkflowStepPhase
+	subStepPhases := make(map[common.WorkflowStepPhase]int)
+	for _, subStepsStatus := range stepStatus.SubStepsStatus {
+		subStepPhases[subStepsStatus.Phase]++
+	}
+	switch {
+	case len(stepStatus.SubStepsStatus) < len(tr.subTaskRunners):
+		phase = common.WorkflowStepPhaseRunning
+	case subStepPhases[common.WorkflowStepPhaseRunning] > 0:
+		phase = common.WorkflowStepPhaseRunning
+	case subStepPhases[common.WorkflowStepPhaseStopped] > 0:
+		phase = common.WorkflowStepPhaseStopped
+	case subStepPhases[common.WorkflowStepPhaseFailed] > 0:
+		phase = common.WorkflowStepPhaseFailed
+	default:
 		phase = common.WorkflowStepPhaseSucceeded
 	}
 	return common.StepStatus{
@@ -181,7 +200,7 @@ func (tr *stepGroupTaskRunner) Run(ctx wfContext.Context, options *types.TaskRun
 		Name:  tr.name,
 		Type:  types.WorkflowStepTypeStepGroup,
 		Phase: phase,
-	}, &types.Operation{}, nil
+	}, e.GetOperation(), nil
 }
 
 // Pending check task should be executed or not.

pkg/workflow/tasks/discover_test.go

@@ -20,19 +20,18 @@ import (
 	"context"
 	"testing"
 
+	"github.com/pkg/errors"
+	"github.com/stretchr/testify/require"
+
 	"github.com/oam-dev/kubevela/apis/core.oam.dev/common"
 	"github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1"
-
-	"github.com/pkg/errors"
-	"gotest.tools/assert"
-
 	"github.com/oam-dev/kubevela/pkg/cue/process"
 	"github.com/oam-dev/kubevela/pkg/workflow/tasks/custom"
 	"github.com/oam-dev/kubevela/pkg/workflow/types"
 )
 
 func TestDiscover(t *testing.T) {
-
+	r := require.New(t)
 	makeErr := func(name string) error {
 		return errors.Errorf("template %s not found", name)
 	}
@@ -56,60 +55,202 @@ func TestDiscover(t *testing.T) {
 	discover := &taskDiscover{
 		builtins: map[string]types.TaskGenerator{
 			"suspend":   suspend,
-			"stepGroup": stepGroup,
+			"stepGroup": StepGroup,
 		},
 		remoteTaskDiscover: custom.NewTaskLoader(loadTemplate, nil, nil, 0, pCtx),
 	}
 
 	_, err := discover.GetTaskGenerator(context.Background(), "suspend")
-	assert.NilError(t, err)
+	r.NoError(err)
 	_, err = discover.GetTaskGenerator(context.Background(), "stepGroup")
-	assert.NilError(t, err)
+	r.NoError(err)
 	_, err = discover.GetTaskGenerator(context.Background(), "foo")
-	assert.NilError(t, err)
+	r.NoError(err)
 	_, err = discover.GetTaskGenerator(context.Background(), "crazy")
-	assert.NilError(t, err)
+	r.NoError(err)
 	_, err = discover.GetTaskGenerator(context.Background(), "fly")
-	assert.Equal(t, err.Error(), makeErr("fly").Error())
+	r.Equal(err.Error(), makeErr("fly").Error())
 
 }
 
 func TestSuspendStep(t *testing.T) {
+	r := require.New(t)
 	discover := &taskDiscover{
 		builtins: map[string]types.TaskGenerator{
 			"suspend": suspend,
 		},
 	}
 	gen, err := discover.GetTaskGenerator(context.Background(), "suspend")
-	assert.NilError(t, err)
+	r.NoError(err)
 	runner, err := gen(v1beta1.WorkflowStep{Name: "test"}, &types.GeneratorOptions{ID: "124"})
-	assert.NilError(t, err)
-	assert.Equal(t, runner.Name(), "test")
-	assert.Equal(t, runner.Pending(nil), false)
+	r.NoError(err)
+	r.Equal(runner.Name(), "test")
+	r.Equal(runner.Pending(nil), false)
 	status, act, err := runner.Run(nil, nil)
-	assert.NilError(t, err)
-	assert.Equal(t, act.Suspend, true)
-	assert.Equal(t, status.ID, "124")
-	assert.Equal(t, status.Name, "test")
-	assert.Equal(t, status.Phase, common.WorkflowStepPhaseSucceeded)
+	r.NoError(err)
+	r.Equal(act.Suspend, true)
+	r.Equal(status.ID, "124")
+	r.Equal(status.Name, "test")
+	r.Equal(status.Phase, common.WorkflowStepPhaseSucceeded)
+}
+
+type testEngine struct {
+	stepStatus common.WorkflowStepStatus
+	operation  *types.Operation
+}
+
+func (e *testEngine) Run(taskRunners []types.TaskRunner, dag bool) error {
+	return nil
+}
+
+func (e *testEngine) GetStepStatus(stepName string) common.WorkflowStepStatus {
+	return e.stepStatus
+}
+
+func (e *testEngine) SetParentRunner(name string) {
+}
+
+func (e *testEngine) GetOperation() *types.Operation {
+	return e.operation
 }
 
 func TestStepGroupStep(t *testing.T) {
+	r := require.New(t)
 	discover := &taskDiscover{
 		builtins: map[string]types.TaskGenerator{
-			"stepGroup": stepGroup,
+			"stepGroup": StepGroup,
 		},
 	}
+	genSub, err := discover.GetTaskGenerator(context.Background(), "stepGroup")
+	r.NoError(err)
+	subRunner, err := genSub(v1beta1.WorkflowStep{Name: "sub"}, &types.GeneratorOptions{ID: "1"})
+	r.NoError(err)
 	gen, err := discover.GetTaskGenerator(context.Background(), "stepGroup")
-	assert.NilError(t, err)
-	runner, err := gen(v1beta1.WorkflowStep{Name: "test"}, &types.GeneratorOptions{ID: "124"})
-	assert.NilError(t, err)
-	assert.Equal(t, runner.Name(), "test")
-	assert.Equal(t, runner.Pending(nil), false)
-	status, act, err := runner.Run(nil, nil)
-	assert.NilError(t, err)
-	assert.Equal(t, act.Suspend, false)
-	assert.Equal(t, status.ID, "124")
-	assert.Equal(t, status.Name, "test")
-	assert.Equal(t, status.Phase, common.WorkflowStepPhaseSucceeded)
+	r.NoError(err)
+	runner, err := gen(v1beta1.WorkflowStep{Name: "test"}, &types.GeneratorOptions{ID: "124", SubTaskRunners: []types.TaskRunner{subRunner}})
+	r.NoError(err)
+	r.Equal(runner.Name(), "test")
+	r.Equal(runner.Pending(nil), false)
+
+	testCases := []struct {
+		name          string
+		engine        *testEngine
+		expectedPhase common.WorkflowStepPhase
+	}{
+		{
+			name: "running1",
+			engine: &testEngine{
+				stepStatus: common.WorkflowStepStatus{},
+				operation:  &types.Operation{},
+			},
+			expectedPhase: common.WorkflowStepPhaseRunning,
+		},
+		{
+			name: "running2",
+			engine: &testEngine{
+				stepStatus: common.WorkflowStepStatus{
+					SubStepsStatus: []common.WorkflowSubStepStatus{
+						{
+							StepStatus: common.StepStatus{
+								Phase: common.WorkflowStepPhaseRunning,
+							},
+						},
+					},
+				},
+				operation: &types.Operation{},
+			},
+			expectedPhase: common.WorkflowStepPhaseRunning,
+		},
+		{
+			name: "stop",
+			engine: &testEngine{
+				stepStatus: common.WorkflowStepStatus{
+					SubStepsStatus: []common.WorkflowSubStepStatus{
+						{
+							StepStatus: common.StepStatus{
+								Phase: common.WorkflowStepPhaseStopped,
+							},
+						},
+						{
+							StepStatus: common.StepStatus{
+								Phase: common.WorkflowStepPhaseFailed,
+							},
+						},
+					},
+				},
+				operation: &types.Operation{},
+			},
+			expectedPhase: common.WorkflowStepPhaseStopped,
+		},
+		{
+			name: "fail",
+			engine: &testEngine{
+				stepStatus: common.WorkflowStepStatus{
+					SubStepsStatus: []common.WorkflowSubStepStatus{
+						{
+							StepStatus: common.StepStatus{
+								Phase: common.WorkflowStepPhaseFailed,
+							},
+						},
+						{
+							StepStatus: common.StepStatus{
+								Phase: common.WorkflowStepPhaseSucceeded,
+							},
+						},
+					},
+				},
+				operation: &types.Operation{},
+			},
+			expectedPhase: common.WorkflowStepPhaseFailed,
+		},
+		{
+			name: "success",
+			engine: &testEngine{
+				stepStatus: common.WorkflowStepStatus{
+					SubStepsStatus: []common.WorkflowSubStepStatus{
+						{
+							StepStatus: common.StepStatus{
+								Phase: common.WorkflowStepPhaseSucceeded,
+							},
+						},
+					},
+				},
+				operation: &types.Operation{},
+			},
+			expectedPhase: common.WorkflowStepPhaseSucceeded,
+		},
+		{
+			name: "operation",
+			engine: &testEngine{
+				stepStatus: common.WorkflowStepStatus{
+					SubStepsStatus: []common.WorkflowSubStepStatus{
+						{
+							StepStatus: common.StepStatus{
+								Phase: common.WorkflowStepPhaseSucceeded,
+							},
+						},
+					},
+				},
+				operation: &types.Operation{
+					Suspend:            true,
+					Terminated:         true,
+					FailedAfterRetries: true,
+					Waiting:            true,
+				},
+			},
+			expectedPhase: common.WorkflowStepPhaseSucceeded,
+		},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			status, act, err := runner.Run(nil, &types.TaskRunOptions{
+				Engine: tc.engine,
+			})
+			r.NoError(err)
+			r.Equal(status.ID, "124")
+			r.Equal(status.Name, "test")
+			r.Equal(act.Suspend, tc.engine.operation.Suspend)
+			r.Equal(status.Phase, tc.expectedPhase)
+		})
+	}
 }

pkg/workflow/types/types.go

@@ -30,7 +30,6 @@ import (
 // TaskRunner is a task runner.
 type TaskRunner interface {
 	Name() string
-	SubTaskRunners() []TaskRunner
 	Pending(ctx wfContext.Context) bool
 	Run(ctx wfContext.Context, options *TaskRunOptions) (common.StepStatus, *Operation, error)
 }
@@ -40,6 +39,14 @@ type TaskDiscover interface {
 	GetTaskGenerator(ctx context.Context, name string) (TaskGenerator, error)
 }
 
+// Engine is the engine to run workflow
+type Engine interface {
+	Run(taskRunners []TaskRunner, dag bool) error
+	GetStepStatus(stepName string) common.WorkflowStepStatus
+	SetParentRunner(name string)
+	GetOperation() *Operation
+}
+
 // TaskRunOptions is the options for task run.
 type TaskRunOptions struct {
 	Data          *value.Value
@@ -49,6 +56,8 @@ type TaskRunOptions struct {
 	GetTracer     func(id string, step v1beta1.WorkflowStep) monitorCtx.Context
 	RunSteps      func(isDag bool, runners ...TaskRunner) (*common.WorkflowStatus, error)
 	Debug         func(step string, v *value.Value) error
+	Engine        Engine
+	ParentRunner  string
 }
 
 // TaskPreStartHook run before task execution.

pkg/workflow/workflow.go

@@ -173,7 +173,6 @@ func (w *workflow) ExecuteSteps(ctx monitorContext.Context, appRev *oamcore.Appl
 
 	e := &engine{
 		status:     wfStatus,
-		dagMode:    w.dagMode,
 		monitorCtx: ctx,
 		app:        w.app,
 		wfCtx:      wfCtx,
@@ -182,7 +181,7 @@ func (w *workflow) ExecuteSteps(ctx monitorContext.Context, appRev *oamcore.Appl
 		rk:         w.rk,
 	}
 
-	err = e.run(taskRunners)
+	err = e.Run(taskRunners, w.dagMode)
 	if err != nil {
 		ctx.Error(err, "run steps")
 		StepStatusCache.Store(cacheKey, len(wfStatus.Steps))
@@ -458,7 +457,7 @@ func (e *engine) runAsDAG(taskRunners []wfTypes.TaskRunner) error {
 	}
 
 	if len(todoTasks) > 0 {
-		err := e.steps(todoTasks)
+		err := e.steps(todoTasks, true)
 		if err != nil {
 			return err
 		}
@@ -475,16 +474,12 @@ func (e *engine) runAsDAG(taskRunners []wfTypes.TaskRunner) error {
 
 }
 
-func (e *engine) runAsStepByStep(taskRunners []wfTypes.TaskRunner) error {
-	return e.steps(e.todoByIndex(taskRunners))
-}
-
-func (e *engine) run(taskRunners []wfTypes.TaskRunner) error {
+func (e *engine) Run(taskRunners []wfTypes.TaskRunner, dag bool) error {
 	var err error
-	if e.dagMode {
+	if dag {
 		err = e.runAsDAG(taskRunners)
 	} else {
-		err = e.runAsStepByStep(taskRunners)
+		err = e.steps(e.todoByIndex(taskRunners), dag)
 	}
 
 	e.setNextExecuteTime()
@@ -519,139 +514,68 @@ func (e *engine) todoByIndex(taskRunners []wfTypes.TaskRunner) []wfTypes.TaskRun
 	}
 	return taskRunners[index:]
 }
-func (e *engine) steps(taskRunners []wfTypes.TaskRunner) error {
+
+func (e *engine) steps(taskRunners []wfTypes.TaskRunner, dag bool) error {
+	wfCtx := e.wfCtx
 	for _, runner := range taskRunners {
-		var err error
-		var needStop bool
-		if runner.SubTaskRunners() == nil {
-			needStop, err = e.step(runner, e.isDag(), "")
-		} else {
-			needStop, err = e.stepWithSubSteps(runner)
+		options := &wfTypes.TaskRunOptions{
+			GetTracer: func(id string, stepStatus oamcore.WorkflowStep) monitorContext.Context {
+				return e.monitorCtx.Fork(id, monitorContext.DurationMetric(func(v float64) {
+					metrics.StepDurationHistogram.WithLabelValues("application", stepStatus.Type).Observe(v)
+				}))
+			},
+			Engine: e,
 		}
-		if needStop || err != nil {
+		if e.debug {
+			options.Debug = func(step string, v *value.Value) error {
+				debugContext := debug.NewContext(e.cli, e.rk, e.app, step)
+				if err := debugContext.Set(v); err != nil {
+					return err
+				}
+				return nil
+			}
+		}
+		status, operation, err := runner.Run(wfCtx, options)
+		if err != nil {
 			return err
 		}
+
+		e.updateStepStatus(status)
+
+		e.failedAfterRetries = e.failedAfterRetries || operation.FailedAfterRetries
+		e.waiting = e.waiting || operation.Waiting
+		if status.Phase == common.WorkflowStepPhaseSucceeded || (status.Phase == common.WorkflowStepPhaseRunning && status.Type == wfTypes.WorkflowStepTypeSuspend) {
+			wfCtx.DeleteValueInMemory(wfTypes.ContextPrefixBackoffTimes, status.ID)
+			wfCtx.DeleteValueInMemory(wfTypes.ContextPrefixBackoffReason, status.ID)
+			if err := wfCtx.Commit(); err != nil {
+				return errors.WithMessage(err, "commit workflow context")
+			}
+
+			e.finishStep(operation)
+			if e.needStop() {
+				return nil
+			}
+			continue
+		}
+
+		if val, exists := wfCtx.GetValueInMemory(wfTypes.ContextPrefixBackoffReason, status.ID); !exists || val != status.Message {
+			wfCtx.SetValueInMemory(status.Message, wfTypes.ContextPrefixBackoffReason, status.ID)
+			wfCtx.DeleteValueInMemory(wfTypes.ContextPrefixBackoffTimes, status.ID)
+		}
+		wfCtx.IncreaseCountValueInMemory(wfTypes.ContextPrefixBackoffTimes, status.ID)
+		if err := wfCtx.Commit(); err != nil {
+			return errors.WithMessage(err, "commit workflow context")
+		}
+		if dag {
+			continue
+		}
+		e.checkFailedAfterRetries()
+		return nil
 	}
 	return nil
 }
 
-func (e *engine) step(runner wfTypes.TaskRunner, dag bool, parentStepName string) (bool, error) {
-	wfCtx := e.wfCtx
-	options := &wfTypes.TaskRunOptions{
-		GetTracer: func(id string, stepStatus oamcore.WorkflowStep) monitorContext.Context {
-			return e.monitorCtx.Fork(id, monitorContext.DurationMetric(func(v float64) {
-				metrics.StepDurationHistogram.WithLabelValues("application", stepStatus.Type).Observe(v)
-			}))
-		},
-	}
-	if e.debug {
-		options.Debug = func(step string, v *value.Value) error {
-			debugContext := debug.NewContext(e.cli, e.rk, e.app, step)
-			if err := debugContext.Set(v); err != nil {
-				return err
-			}
-			return nil
-		}
-	}
-	status, operation, err := runner.Run(wfCtx, options)
-	if err != nil {
-		return true, err
-	}
-	if parentStepName != "" {
-		subStepStatus := common.WorkflowSubStepStatus{
-			StepStatus: status,
-		}
-		e.updateSubStepStatus(parentStepName, subStepStatus)
-	} else {
-		workflowStepStatus := common.WorkflowStepStatus{
-			StepStatus: status,
-		}
-		e.updateStepStatus(workflowStepStatus)
-	}
-
-	e.failedAfterRetries = e.failedAfterRetries || operation.FailedAfterRetries
-	e.waiting = e.waiting || operation.Waiting
-	if status.Phase == common.WorkflowStepPhaseSucceeded || (status.Phase == common.WorkflowStepPhaseRunning && status.Type == wfTypes.WorkflowStepTypeSuspend) {
-		wfCtx.DeleteValueInMemory(wfTypes.ContextPrefixBackoffTimes, status.ID)
-		wfCtx.DeleteValueInMemory(wfTypes.ContextPrefixBackoffReason, status.ID)
-		if err := wfCtx.Commit(); err != nil {
-			return true, errors.WithMessage(err, "commit workflow context")
-		}
-
-		e.finishStep(operation)
-		if e.needStop() {
-			return true, nil
-		}
-		return false, nil
-	}
-
-	if val, exists := wfCtx.GetValueInMemory(wfTypes.ContextPrefixBackoffReason, status.ID); !exists || val != status.Message {
-		wfCtx.SetValueInMemory(status.Message, wfTypes.ContextPrefixBackoffReason, status.ID)
-		wfCtx.DeleteValueInMemory(wfTypes.ContextPrefixBackoffTimes, status.ID)
-	}
-	wfCtx.IncreaseCountValueInMemory(wfTypes.ContextPrefixBackoffTimes, status.ID)
-	if err := wfCtx.Commit(); err != nil {
-		return true, errors.WithMessage(err, "commit workflow context")
-	}
-	if dag {
-		return false, nil
-	}
-	e.checkFailedAfterRetries()
-	return true, nil
-}
-
-func (e *engine) stepWithSubSteps(runner wfTypes.TaskRunner) (bool, error) {
-	stepName := runner.Name()
-	var err error
-	var needStop bool
-	var status common.StepStatus
-	status, _, err = runner.Run(e.wfCtx, &wfTypes.TaskRunOptions{})
-	if err != nil {
-		return true, err
-	}
-	workflowStepStatus := common.WorkflowStepStatus{
-		StepStatus:     status,
-		SubStepsStatus: e.getStepStatus(stepName).SubStepsStatus,
-	}
-	e.updateStepStatus(workflowStepStatus)
-
-	todoSubTaskRunners, pendingSubTaskRunners := e.getTodoSubTaskRunners(runner)
-	for _, todoSubTaskRunner := range todoSubTaskRunners {
-		// subStep only use dag mode
-		dag := true
-		needStop, err = e.step(todoSubTaskRunner, dag, stepName)
-	}
-
-	stepStatus := e.getStepStatus(stepName)
-	subStepPhaseToCount := make(map[common.WorkflowStepPhase]int)
-	for _, subStepsStatus := range stepStatus.SubStepsStatus {
-		subStepPhaseToCount[subStepsStatus.Phase]++
-	}
-	switch {
-	case subStepPhaseToCount[common.WorkflowStepPhaseRunning] != 0:
-		stepStatus.Phase = common.WorkflowStepPhaseRunning
-	case subStepPhaseToCount[common.WorkflowStepPhaseStopped] != 0:
-		stepStatus.Phase = common.WorkflowStepPhaseStopped
-	case subStepPhaseToCount[common.WorkflowStepPhaseFailed] != 0:
-		stepStatus.Phase = common.WorkflowStepPhaseFailed
-	case len(pendingSubTaskRunners) != 0:
-		stepStatus.Phase = common.WorkflowStepPhaseRunning
-	default:
-		stepStatus.Phase = common.WorkflowStepPhaseSucceeded
-	}
-	e.updateStepStatus(stepStatus)
-	e.checkFailedAfterRetries()
-	if needStop || err != nil {
-		return true, err
-	}
-	if stepStatus.Phase == common.WorkflowStepPhaseSucceeded {
-		return false, nil
-	}
-	return true, nil
-}
-
 type engine struct {
-	dagMode            bool
 	failedAfterRetries bool
 	waiting            bool
 	debug              bool
@@ -661,10 +585,7 @@ type engine struct {
 	app                *oamcore.Application
 	cli                client.Client
 	rk                 resourcekeeper.ResourceKeeper
-}
-
-func (e *engine) isDag() bool {
-	return e.dagMode
+	parentRunner       string
 }
 
 func (e *engine) finishStep(operation *wfTypes.Operation) {
@@ -674,63 +595,59 @@ func (e *engine) finishStep(operation *wfTypes.Operation) {
 	}
 }
 
-func (e *engine) updateStepStatus(status common.WorkflowStepStatus) {
-	var now = metav1.NewTime(time.Now())
+func (e *engine) updateStepStatus(status common.StepStatus) {
+	var (
+		conditionUpdated bool
+		now              = metav1.NewTime(time.Now())
+	)
+
+	parentRunner := e.parentRunner
+	stepName := status.Name
+	if parentRunner != "" {
+		stepName = parentRunner
+	}
 	e.wfCtx.SetValueInMemory(now.Unix(), wfTypes.ContextKeyLastExecuteTime)
-	e.status.Steps = updateStepStatusByName(e.status.Steps, status, now)
-}
-
-func updateStepStatusByName(stepStatus []common.WorkflowStepStatus, status common.WorkflowStepStatus, now metav1.Time) []common.WorkflowStepStatus {
-	var (
-		conditionUpdated bool
-	)
-
 	status.LastExecuteTime = now
-	for i := range stepStatus {
-		if stepStatus[i].Name == status.Name {
-			status.FirstExecuteTime = stepStatus[i].FirstExecuteTime
-			stepStatus[i] = status
-			conditionUpdated = true
-			break
+	index := -1
+	for i, ss := range e.status.Steps {
+		if ss.Name == stepName {
+			index = i
+			if parentRunner != "" {
+				// update the sub steps status
+				for j, sub := range ss.SubStepsStatus {
+					if sub.Name == status.Name {
+						status.FirstExecuteTime = sub.FirstExecuteTime
+						e.status.Steps[i].SubStepsStatus[j].StepStatus = status
+						conditionUpdated = true
+						break
+					}
+				}
+			} else {
+				// update the parent steps status
+				status.FirstExecuteTime = ss.FirstExecuteTime
+				e.status.Steps[i].StepStatus = status
+				conditionUpdated = true
+				break
+			}
 		}
 	}
 	if !conditionUpdated {
 		status.FirstExecuteTime = now
-		stepStatus = append(stepStatus, status)
-	}
-	return stepStatus
-}
-
-func (e *engine) updateSubStepStatus(stepName string, subStepStatus common.WorkflowSubStepStatus) {
-	var now = metav1.NewTime(time.Now())
-	for i := range e.status.Steps {
-		if e.status.Steps[i].Name == stepName {
-			e.status.Steps[i].SubStepsStatus = updateSubStepStatusByName(e.status.Steps[i].SubStepsStatus, subStepStatus, now)
+		if parentRunner != "" {
+			if index < 0 {
+				e.status.Steps = append(e.status.Steps, common.WorkflowStepStatus{
+					StepStatus: common.StepStatus{
+						Name: parentRunner,
+					}})
+				index = len(e.status.Steps) - 1
+			}
+			e.status.Steps[index].SubStepsStatus = append(e.status.Steps[index].SubStepsStatus, common.WorkflowSubStepStatus{StepStatus: status})
+		} else {
+			e.status.Steps = append(e.status.Steps, common.WorkflowStepStatus{StepStatus: status})
 		}
 	}
 }
 
-func updateSubStepStatusByName(stepStatus []common.WorkflowSubStepStatus, status common.WorkflowSubStepStatus, now metav1.Time) []common.WorkflowSubStepStatus {
-	var (
-		conditionUpdated bool
-	)
-
-	status.LastExecuteTime = now
-	for i := range stepStatus {
-		if stepStatus[i].Name == status.Name {
-			status.FirstExecuteTime = stepStatus[i].FirstExecuteTime
-			stepStatus[i] = status
-			conditionUpdated = true
-			break
-		}
-	}
-	if !conditionUpdated {
-		status.FirstExecuteTime = now
-		stepStatus = append(stepStatus, status)
-	}
-	return stepStatus
-}
-
 func (e *engine) checkFailedAfterRetries() {
 	if !e.waiting && e.failedAfterRetries {
 		e.status.Suspend = true
@@ -763,47 +680,7 @@ func IsFailedAfterRetry(app *oamcore.Application) bool {
 	return app.Status.Workflow != nil && app.Status.Workflow.Message == MessageFailedAfterRetries
 }
 
-func (e *engine) getTodoSubTaskRunners(taskRunner wfTypes.TaskRunner) ([]wfTypes.TaskRunner, []wfTypes.TaskRunner) {
-	wfCtx := e.wfCtx
-	var (
-		todoSubTasks    []wfTypes.TaskRunner
-		pendingSubTasks []wfTypes.TaskRunner
-	)
-	for _, subTRunner := range taskRunner.SubTaskRunners() {
-		ready := false
-		var stepID string
-		subStepStatus := e.getSubStepStatus(taskRunner.Name(), subTRunner.Name())
-		ready = subStepStatus.Phase == common.WorkflowStepPhaseSucceeded
-		stepID = subStepStatus.ID
-
-		if !ready {
-			if subTRunner.Pending(wfCtx) {
-				pendingSubTasks = append(pendingSubTasks, subTRunner)
-				continue
-			}
-			todoSubTasks = append(todoSubTasks, subTRunner)
-		} else {
-			wfCtx.DeleteValueInMemory(wfTypes.ContextPrefixBackoffTimes, stepID)
-		}
-	}
-	return todoSubTasks, pendingSubTasks
-}
-
-func (e *engine) getSubStepStatus(stepName string, subStepName string) common.WorkflowSubStepStatus {
-	// ss is step status
-	for _, ss := range e.status.Steps {
-		if ss.Name == stepName {
-			// sss is subStepStatus
-			for _, sss := range ss.SubStepsStatus {
-				if sss.Name == subStepName {
-					return sss
-				}
-			}
-		}
-	}
-	return common.WorkflowSubStepStatus{}
-}
-func (e *engine) getStepStatus(stepName string) common.WorkflowStepStatus {
+func (e *engine) GetStepStatus(stepName string) common.WorkflowStepStatus {
 	// ss is step status
 	for _, ss := range e.status.Steps {
 		if ss.Name == stepName {
@@ -812,3 +689,16 @@ func (e *engine) getStepStatus(stepName string) common.WorkflowStepStatus {
 	}
 	return common.WorkflowStepStatus{}
 }
+
+func (e *engine) SetParentRunner(name string) {
+	e.parentRunner = name
+}
+
+func (e *engine) GetOperation() *wfTypes.Operation {
+	return &wfTypes.Operation{
+		Suspend:            e.status.Suspend,
+		Terminated:         e.status.Terminated,
+		Waiting:            e.waiting,
+		FailedAfterRetries: e.failedAfterRetries,
+	}
+}

pkg/workflow/workflow_test.go

@@ -36,6 +36,7 @@ import (
 	"github.com/oam-dev/kubevela/pkg/cue/model/value"
 	monitorContext "github.com/oam-dev/kubevela/pkg/monitor/context"
 	wfContext "github.com/oam-dev/kubevela/pkg/workflow/context"
+	"github.com/oam-dev/kubevela/pkg/workflow/tasks"
 	wfTypes "github.com/oam-dev/kubevela/pkg/workflow/types"
 )
 
@@ -154,9 +155,11 @@ var _ = Describe("Test Workflow", func() {
 				},
 			}},
 		})).Should(BeEquivalentTo(""))
+	})
 
+	It("Workflow test failed with sub steps", func() {
 		By("Test failed with step group")
-		app, runners = makeTestCase([]oamcore.WorkflowStep{
+		app, runners := makeTestCase([]oamcore.WorkflowStep{
 			{
 				Name: "s1",
 				Type: "success",
@@ -180,9 +183,9 @@ var _ = Describe("Test Workflow", func() {
 				Type: "success",
 			},
 		})
-		wf = NewWorkflow(app, k8sClient, common.WorkflowModeStep, false, nil)
-		ctx = monitorContext.NewTraceContext(context.Background(), "test-app")
-		state, err = wf.ExecuteSteps(ctx, revision, runners)
+		wf := NewWorkflow(app, k8sClient, common.WorkflowModeStep, false, nil)
+		ctx := monitorContext.NewTraceContext(context.Background(), "test-app")
+		state, err := wf.ExecuteSteps(ctx, revision, runners)
 		Expect(err).ToNot(HaveOccurred())
 		Expect(state).Should(BeEquivalentTo(common.WorkflowStateInitializing))
 		state, err = wf.ExecuteSteps(ctx, revision, runners)
@@ -221,9 +224,11 @@ var _ = Describe("Test Workflow", func() {
 				}},
 			}},
 		})).Should(BeEquivalentTo(""))
+	})
 
+	It("Workflow test success with sub steps", func() {
 		By("Test success with step group")
-		app, runners = makeTestCase([]oamcore.WorkflowStep{
+		app, runners := makeTestCase([]oamcore.WorkflowStep{
 			{
 				Name: "s1",
 				Type: "success",
@@ -247,9 +252,9 @@ var _ = Describe("Test Workflow", func() {
 				Type: "success",
 			},
 		})
-		wf = NewWorkflow(app, k8sClient, common.WorkflowModeStep, false, nil)
-		ctx = monitorContext.NewTraceContext(context.Background(), "test-app")
-		state, err = wf.ExecuteSteps(ctx, revision, runners)
+		wf := NewWorkflow(app, k8sClient, common.WorkflowModeStep, false, nil)
+		ctx := monitorContext.NewTraceContext(context.Background(), "test-app")
+		state, err := wf.ExecuteSteps(ctx, revision, runners)
 		Expect(err).ToNot(HaveOccurred())
 		Expect(state).Should(BeEquivalentTo(common.WorkflowStateInitializing))
 		state, err = wf.ExecuteSteps(ctx, revision, runners)
@@ -449,9 +454,11 @@ var _ = Describe("Test Workflow", func() {
 				},
 			}},
 		})).Should(BeEquivalentTo(""))
+	})
 
+	It("Test failed after retries with sub steps", func() {
 		By("Test failed-after-retries with step group in StepByStep mode")
-		app, runners = makeTestCase([]oamcore.WorkflowStep{
+		app, runners := makeTestCase([]oamcore.WorkflowStep{
 			{
 				Name: "s1",
 				Type: "success",
@@ -475,9 +482,9 @@ var _ = Describe("Test Workflow", func() {
 				Type: "success",
 			},
 		})
-		wf = NewWorkflow(app, k8sClient, common.WorkflowModeStep, false, nil)
-		ctx = monitorContext.NewTraceContext(context.Background(), "test-app")
-		state, err = wf.ExecuteSteps(ctx, revision, runners)
+		wf := NewWorkflow(app, k8sClient, common.WorkflowModeStep, false, nil)
+		ctx := monitorContext.NewTraceContext(context.Background(), "test-app")
+		state, err := wf.ExecuteSteps(ctx, revision, runners)
 		Expect(err).ToNot(HaveOccurred())
 		Expect(state).Should(BeEquivalentTo(common.WorkflowStateInitializing))
 		state, err = wf.ExecuteSteps(ctx, revision, runners)
@@ -663,9 +670,11 @@ var _ = Describe("Test Workflow", func() {
 		state, err = wf.ExecuteSteps(ctx, revision, runners)
 		Expect(err).ToNot(HaveOccurred())
 		Expect(state).Should(BeEquivalentTo(common.WorkflowStateSucceeded))
+	})
 
+	It("test for suspend with sub steps", func() {
 		By("Test suspend with step group")
-		app, runners = makeTestCase([]oamcore.WorkflowStep{
+		app, runners := makeTestCase([]oamcore.WorkflowStep{
 			{
 				Name: "s1",
 				Type: "success",
@@ -689,9 +698,9 @@ var _ = Describe("Test Workflow", func() {
 				Type: "success",
 			},
 		})
-		wf = NewWorkflow(app, k8sClient, common.WorkflowModeStep, false, nil)
-		ctx = monitorContext.NewTraceContext(context.Background(), "test-app")
-		state, err = wf.ExecuteSteps(ctx, revision, runners)
+		wf := NewWorkflow(app, k8sClient, common.WorkflowModeStep, false, nil)
+		ctx := monitorContext.NewTraceContext(context.Background(), "test-app")
+		state, err := wf.ExecuteSteps(ctx, revision, runners)
 		Expect(err).ToNot(HaveOccurred())
 		Expect(state).Should(BeEquivalentTo(common.WorkflowStateInitializing))
 		state, err = wf.ExecuteSteps(ctx, revision, runners)
@@ -777,9 +786,12 @@ var _ = Describe("Test Workflow", func() {
 		state, err = wf.ExecuteSteps(ctx, revision, runners)
 		Expect(err).ToNot(HaveOccurred())
 		Expect(state).Should(BeEquivalentTo(common.WorkflowStateTerminated))
+	})
+
+	It("test for terminate with sub steps", func() {
 
 		By("Test terminate with step group")
-		app, runners = makeTestCase([]oamcore.WorkflowStep{
+		app, runners := makeTestCase([]oamcore.WorkflowStep{
 			{
 				Name: "s1",
 				Type: "success",
@@ -799,9 +811,9 @@ var _ = Describe("Test Workflow", func() {
 				},
 			},
 		})
-		ctx = monitorContext.NewTraceContext(context.Background(), "test-app")
-		wf = NewWorkflow(app, k8sClient, common.WorkflowModeStep, false, nil)
-		state, err = wf.ExecuteSteps(ctx, revision, runners)
+		ctx := monitorContext.NewTraceContext(context.Background(), "test-app")
+		wf := NewWorkflow(app, k8sClient, common.WorkflowModeStep, false, nil)
+		state, err := wf.ExecuteSteps(ctx, revision, runners)
 		Expect(err).ToNot(HaveOccurred())
 		Expect(state).Should(BeEquivalentTo(common.WorkflowStateInitializing))
 		state, err = wf.ExecuteSteps(ctx, revision, runners)
@@ -1099,13 +1111,8 @@ func makeRunner(name string, tpy string, subTaskRunners []wfTypes.TaskRunner) wf
 			}, &wfTypes.Operation{}, err
 		}
 	case "step-group":
-		run = func(ctx wfContext.Context, options *wfTypes.TaskRunOptions) (common.StepStatus, *wfTypes.Operation, error) {
-			return common.StepStatus{
-				Name:  name,
-				Type:  "step-group",
-				Phase: common.WorkflowStepPhaseRunning,
-			}, &wfTypes.Operation{}, nil
-		}
+		group, _ := tasks.StepGroup(oamcore.WorkflowStep{Name: name}, &wfTypes.GeneratorOptions{SubTaskRunners: subTaskRunners})
+		run = group.Run
 	default:
 		run = func(ctx wfContext.Context, options *wfTypes.TaskRunOptions) (common.StepStatus, *wfTypes.Operation, error) {
 			return common.StepStatus{
@@ -1162,7 +1169,7 @@ func (tr *testTaskRunner) Name() string {
 
 // Run execute task.
 func (tr *testTaskRunner) Run(ctx wfContext.Context, options *wfTypes.TaskRunOptions) (common.StepStatus, *wfTypes.Operation, error) {
-	return tr.run(ctx, nil)
+	return tr.run(ctx, options)
 }
 
 // Pending check task should be executed or not.