diff --git a/pkg/workflow/tasks/custom/task.go b/pkg/workflow/tasks/custom/task.go index a25e5dcce..f6c7b6198 100644 --- a/pkg/workflow/tasks/custom/task.go +++ b/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) { diff --git a/pkg/workflow/tasks/discover.go b/pkg/workflow/tasks/discover.go index 05a5a26f6..c5e78fcb9 100644 --- a/pkg/workflow/tasks/discover.go +++ b/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. diff --git a/pkg/workflow/tasks/discover_test.go b/pkg/workflow/tasks/discover_test.go index f36312e08..2a41fb133 100644 --- a/pkg/workflow/tasks/discover_test.go +++ b/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) + }) + } } diff --git a/pkg/workflow/types/types.go b/pkg/workflow/types/types.go index 349cb0b47..7de832a1c 100644 --- a/pkg/workflow/types/types.go +++ b/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. diff --git a/pkg/workflow/workflow.go b/pkg/workflow/workflow.go index f27549079..8efb48329 100644 --- a/pkg/workflow/workflow.go +++ b/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, + } +} diff --git a/pkg/workflow/workflow_test.go b/pkg/workflow/workflow_test.go index 55e3b34f5..f7d5a0da4 100644 --- a/pkg/workflow/workflow_test.go +++ b/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.