From 54380976489f0abbc6d7690da5d358cadbe673ed Mon Sep 17 00:00:00 2001 From: Denis Arh Date: Thu, 11 Feb 2021 16:06:26 +0100 Subject: [PATCH] Refactor, add generic step, add error&termination logic --- automation/service/workflow.go | 76 ++++++++++++++++++++++++++++++- automation/types/debug.go | 2 +- automation/types/error.go | 23 ++++++++++ automation/types/error_handler.go | 2 +- automation/types/expr.go | 2 +- automation/types/function.go | 4 +- automation/types/step.go | 8 ---- automation/types/workflow.go | 3 +- pkg/wfexec/gateways.go | 8 ++-- pkg/wfexec/graph.go | 12 ----- pkg/wfexec/graph_test.go | 2 +- pkg/wfexec/response.go | 6 +++ pkg/wfexec/session.go | 22 +++++++-- pkg/wfexec/session_test.go | 4 +- pkg/wfexec/step.go | 32 +++++++++++++ 15 files changed, 166 insertions(+), 40 deletions(-) create mode 100644 automation/types/error.go delete mode 100644 automation/types/step.go create mode 100644 pkg/wfexec/step.go diff --git a/automation/service/workflow.go b/automation/service/workflow.go index a6aa6ff45..a29cf04b3 100644 --- a/automation/service/workflow.go +++ b/automation/service/workflow.go @@ -533,8 +533,11 @@ func (svc *workflow) workflowStepDefConv(g *wfexec.Graph, s *types.WorkflowStep, case types.WorkflowStepKindFunction, types.WorkflowStepKindIterator: return svc.convFunctionStep(g, s, out) - //case types.WorkflowStepKindMessage: - // return svc.convMessageStep(s) + case types.WorkflowStepKindError: + return svc.convErrorStep(s, out) + + case types.WorkflowStepKindTermination: + return svc.convTerminationStep(out) case types.WorkflowStepKindPrompt: return svc.convPromptStep(s) @@ -734,6 +737,69 @@ func (svc *workflow) convFunctionStep(g *wfexec.Graph, s *types.WorkflowStep, ou } } +// creates error step +// +// Expects ZERO outgoing paths and +func (svc *workflow) convErrorStep(s *types.WorkflowStep, out types.WorkflowPathSet) (wfexec.Step, error) { + const ( + argName = "message" + ) + + if len(out) > 0 { + return nil, errors.Internal("error step must be last step in branch") + } + + var ( + args = types.ExprSet(s.Arguments) + ) + + if msgArg := args.GetByTarget(argName); msgArg == nil { + return nil, errors.Internal("error step must have %s argument", argName) + } else if msgArg.Type != (expr.String{}).Type() { + return nil, errors.Internal("%s argument on error step must be string, got type '%s'", argName, msgArg.Type) + } else if len(args) > 1 { + return nil, errors.Internal("too many arguments on error step") + } + + if err := svc.parseExpressions(args...); err != nil { + return nil, err + } + + return wfexec.NewGenericStep(func(ctx context.Context, r *wfexec.ExecRequest) (wfexec.ExecResponse, error) { + var ( + msg string + result, err = args.Eval(ctx, r.Scope) + ) + if err != nil { + return nil, err + } + + if result.Has(argName) { + str, _ := expr.NewString(expr.Must(result.Select(argName))) + msg = str.GetValue() + } else { + if aux, is := args.GetByTarget(argName).Value.(string); is { + msg = aux + } else { + msg = "ERROR" + } + } + + return nil, errors.Automation(msg) + }), nil +} + +// converts prompt definition to wfexec.Step +func (svc *workflow) convTerminationStep(out types.WorkflowPathSet) (wfexec.Step, error) { + if len(out) > 0 { + return nil, errors.Internal("termination step must be last step in branch") + } + + return wfexec.NewGenericStep(func(ctx context.Context, r *wfexec.ExecRequest) (wfexec.ExecResponse, error) { + return wfexec.Termination(), nil + }), nil +} + // converts prompt definition to wfexec.Step func (svc *workflow) convPromptStep(s *types.WorkflowStep) (wfexec.Step, error) { if err := svc.parseExpressions(s.Arguments...); err != nil { @@ -821,6 +887,12 @@ func validateSteps(ss ...*types.WorkflowStep) error { case types.WorkflowStepKindGateway: checks = append(checks, noArgs, noResults) + case types.WorkflowStepKindError: + checks = append(checks, noResults) + + case types.WorkflowStepKindTermination: + checks = append(checks, noArgs, noResults) + case types.WorkflowStepKindFunction, types.WorkflowStepKindIterator: case types.WorkflowStepKindPrompt: diff --git a/automation/types/debug.go b/automation/types/debug.go index 1b95526d1..0754c42f8 100644 --- a/automation/types/debug.go +++ b/automation/types/debug.go @@ -9,7 +9,7 @@ import ( type ( debugStep struct { - identifiableStep + wfexec.StepIdentifier log *zap.Logger } ) diff --git a/automation/types/error.go b/automation/types/error.go new file mode 100644 index 000000000..8be5432f9 --- /dev/null +++ b/automation/types/error.go @@ -0,0 +1,23 @@ +package types + +import ( + "context" + "fmt" + "github.com/cortezaproject/corteza-server/pkg/wfexec" +) + +type ( + errorStep struct { + wfexec.StepIdentifier + message string + } +) + +func ErrorStep(msg string) *errorStep { + return &errorStep{message: msg} +} + +// Executes prompt step +func (e *errorStep) Exec(context.Context, *wfexec.ExecRequest) (wfexec.ExecResponse, error) { + return nil, fmt.Errorf(e.message) +} diff --git a/automation/types/error_handler.go b/automation/types/error_handler.go index cd6bb9310..0c04b6c89 100644 --- a/automation/types/error_handler.go +++ b/automation/types/error_handler.go @@ -7,7 +7,7 @@ import ( type ( errorHandlerStep struct { - identifiableStep + wfexec.StepIdentifier handler wfexec.Step } ) diff --git a/automation/types/expr.go b/automation/types/expr.go index 4799d7058..167c85150 100644 --- a/automation/types/expr.go +++ b/automation/types/expr.go @@ -44,7 +44,7 @@ type ( // WorkflowStepExpression is created from WorkflowStep with kind=expressions expressionsStep struct { - identifiableStep + wfexec.StepIdentifier Set ExprSet } ) diff --git a/automation/types/function.go b/automation/types/function.go index 1726f39ba..53057b6d9 100644 --- a/automation/types/function.go +++ b/automation/types/function.go @@ -33,14 +33,14 @@ type ( } functionStep struct { - identifiableStep + wfexec.StepIdentifier def *Function arguments ExprSet results ExprSet } iteratorStep struct { - identifiableStep + wfexec.StepIdentifier def *Function arguments ExprSet results ExprSet diff --git a/automation/types/step.go b/automation/types/step.go deleted file mode 100644 index ca4de52c4..000000000 --- a/automation/types/step.go +++ /dev/null @@ -1,8 +0,0 @@ -package types - -type ( - identifiableStep struct{ id uint64 } -) - -func (i *identifiableStep) ID() uint64 { return i.id } -func (i *identifiableStep) SetID(id uint64) { i.id = id } diff --git a/automation/types/workflow.go b/automation/types/workflow.go index 37c1eef50..c03fb3607 100644 --- a/automation/types/workflow.go +++ b/automation/types/workflow.go @@ -120,7 +120,8 @@ const ( WorkflowStepKindGateway WorkflowStepKind = "gateway" // ref = join|fork|excl|incl WorkflowStepKindFunction WorkflowStepKind = "function" // ref = WorkflowStepKindIterator WorkflowStepKind = "iterator" // ref = - WorkflowStepKindMessage WorkflowStepKind = "message" // ref = error|warning|info, ... + WorkflowStepKindError WorkflowStepKind = "error" // no ref + WorkflowStepKindTermination WorkflowStepKind = "termination" // no ref WorkflowStepKindPrompt WorkflowStepKind = "prompt" // ref = WorkflowStepKindErrHandler WorkflowStepKind = "error-handler" // no ref WorkflowStepKindVisual WorkflowStepKind = "visual" // ref = <*> diff --git a/pkg/wfexec/gateways.go b/pkg/wfexec/gateways.go index 480ac6f7e..5bc44ebe6 100644 --- a/pkg/wfexec/gateways.go +++ b/pkg/wfexec/gateways.go @@ -29,7 +29,7 @@ func NewGatewayPath(s Step, t pathTester) (gwp *GatewayPath, err error) { // joinGateway handles merging/joining of multiple paths into // a single path forward type joinGateway struct { - stepIdentifier + StepIdentifier paths Steps scopes map[Step]*expr.Vars l sync.Mutex @@ -75,7 +75,7 @@ func (gw *joinGateway) Exec(_ context.Context, r *ExecRequest) (ExecResponse, er // forkGateway handles forking to multiple paths type forkGateway struct { - stepIdentifier + StepIdentifier } // ForkGateway fn initializes fork gateway @@ -93,7 +93,7 @@ func (gw forkGateway) Exec(context.Context, *ExecRequest) (ExecResponse, error) // inclGateway is an inclusive gateway that can return one or more paths type inclGateway struct { - stepIdentifier + StepIdentifier paths []*GatewayPath } @@ -134,7 +134,7 @@ func (gw inclGateway) Exec(ctx context.Context, r *ExecRequest) (ExecResponse, e // exclGateway is an exclusive gateway that can return exactly one path type exclGateway struct { - stepIdentifier + StepIdentifier paths []*GatewayPath } diff --git a/pkg/wfexec/graph.go b/pkg/wfexec/graph.go index 8425e2852..f302298cc 100644 --- a/pkg/wfexec/graph.go +++ b/pkg/wfexec/graph.go @@ -5,13 +5,6 @@ import ( ) type ( - Steps []Step - Step interface { - ID() uint64 - SetID(uint64) - Exec(context.Context, *ExecRequest) (ExecResponse, error) - } - // list of Graph steps with relations Graph struct { steps []Step @@ -19,8 +12,6 @@ type ( parents map[Step][]Step index map[uint64]Step } - - stepIdentifier struct{ id uint64 } ) func NewGraph() *Graph { @@ -34,9 +25,6 @@ func NewGraph() *Graph { return wf } -func (i *stepIdentifier) ID() uint64 { return i.id } -func (i *stepIdentifier) SetID(id uint64) { i.id = id } - func (g *Graph) AddStep(s Step, cc ...Step) { g.steps = append(g.steps, s) diff --git a/pkg/wfexec/graph_test.go b/pkg/wfexec/graph_test.go index dd1ea0f17..3f329f94f 100644 --- a/pkg/wfexec/graph_test.go +++ b/pkg/wfexec/graph_test.go @@ -8,7 +8,7 @@ import ( type ( wfTestStep struct { - stepIdentifier + StepIdentifier name string } ) diff --git a/pkg/wfexec/response.go b/pkg/wfexec/response.go index 8fed69c18..20bf82ce3 100644 --- a/pkg/wfexec/response.go +++ b/pkg/wfexec/response.go @@ -11,6 +11,8 @@ type ( errHandler struct { handler Step } + + termination struct{} ) func DelayExecution(until time.Time) *suspended { @@ -25,6 +27,10 @@ func ErrorHandler(h Step) *errHandler { return &errHandler{handler: h} } +func Termination() *termination { + return &termination{} +} + type ( loopBreak struct{} loopContinue struct{} diff --git a/pkg/wfexec/session.go b/pkg/wfexec/session.go index 9315f1ccc..0ad8156e5 100644 --- a/pkg/wfexec/session.go +++ b/pkg/wfexec/session.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "github.com/cortezaproject/corteza-server/pkg/auth" + "github.com/cortezaproject/corteza-server/pkg/errors" "github.com/cortezaproject/corteza-server/pkg/expr" "github.com/cortezaproject/corteza-server/pkg/id" "github.com/cortezaproject/corteza-server/pkg/logger" @@ -335,7 +336,7 @@ func (s *Session) worker(ctx context.Context) { if st == nil { // stop worker s.log.Debug("worker done") - break + return } s.log.Debug("pulled state from queue", zap.Uint64("stateID", st.stateId)) @@ -482,7 +483,7 @@ func (s *Session) exec(ctx context.Context, st *State) { zap.Error(st.err), ) - expr.Assign(scope, "error", expr.Must(expr.NewString(st.err.Error()))) + _ = expr.Assign(scope, "error", expr.Must(expr.NewString(st.err.Error()))) // copy error handler & disable it on state to prevent inf. loop // in case of another error in the error-handling branch @@ -494,8 +495,13 @@ func (s *Session) exec(ctx context.Context, st *State) { return } else { - log.Error("step execution failed", zap.Error(st.err)) - s.qErr <- fmt.Errorf("session %d step %d execution failed: %w", s.id, st.step.ID(), st.err) + if errors.IsAutomation(st.err) { + s.qErr <- st.err + } else { + log.Error("step execution failed", zap.Error(st.err)) + s.qErr <- fmt.Errorf("session %d step %d execution failed: %w", s.id, st.step.ID(), st.err) + } + return } } @@ -572,6 +578,13 @@ func (s *Session) exec(ctx context.Context, st *State) { // it's used mainly for join gateway step that should be called multiple times (one for each parent path) return + case *termination: + // terminate all activities, all suspended tasks and exit right away + log.Debug("termination", zap.Int("suspended", len(s.suspended))) + s.suspended = nil + s.qState <- FinalState(s, scope) + return + case *suspended: // suspend execution because of delay or pending user input // either way, it breaks execution loop for the current path @@ -586,7 +599,6 @@ func (s *Session) exec(ctx context.Context, st *State) { log.Debug("suspended, temporal", zap.Timep("at", result.resumeAt)) } else { log.Debug("suspended, prompt") - } result.state = st diff --git a/pkg/wfexec/session_test.go b/pkg/wfexec/session_test.go index 54b75c6fb..dcd019dd2 100644 --- a/pkg/wfexec/session_test.go +++ b/pkg/wfexec/session_test.go @@ -11,13 +11,13 @@ import ( type ( sesTestStep struct { - stepIdentifier + StepIdentifier name string exec func(context.Context, *ExecRequest) (ExecResponse, error) } sesTestTemporal struct { - stepIdentifier + StepIdentifier delay time.Duration until time.Time } diff --git a/pkg/wfexec/step.go b/pkg/wfexec/step.go new file mode 100644 index 000000000..6bfe3846f --- /dev/null +++ b/pkg/wfexec/step.go @@ -0,0 +1,32 @@ +package wfexec + +import "context" + +type ( + Steps []Step + Step interface { + ID() uint64 + SetID(uint64) + Exec(context.Context, *ExecRequest) (ExecResponse, error) + } + + StepIdentifier struct{ id uint64 } + + execFn func(context.Context, *ExecRequest) (ExecResponse, error) + + genericStep struct { + StepIdentifier + fn execFn + } +) + +func (i *StepIdentifier) ID() uint64 { return i.id } +func (i *StepIdentifier) SetID(id uint64) { i.id = id } + +func NewGenericStep(fn execFn) *genericStep { + return &genericStep{fn: fn} +} + +func (g *genericStep) Exec(ctx context.Context, r *ExecRequest) (ExecResponse, error) { + return g.fn(ctx, r) +}