diff --git a/automation/service/workflow.go b/automation/service/workflow.go index 472e154d8..3bf94f751 100644 --- a/automation/service/workflow.go +++ b/automation/service/workflow.go @@ -564,6 +564,12 @@ func (svc *workflow) Exec(ctx context.Context, workflowID uint64, p types.Workfl // Find the trigger. // @todo can we cache this as well? t, err = func() (*types.Trigger, error) { + if p.CallerWorkflowID > 0 { + // skip triggers checking when executed as sub-workflow + // @todo be more strict and allow this ONLY when workflow is flagged as a sub-workflow + return nil, nil + } + var tt types.TriggerSet // Load triggers directly from the store. At this point we do not care // about trigger search or read permissions @@ -572,6 +578,10 @@ func (svc *workflow) Exec(ctx context.Context, workflowID uint64, p types.Workfl return nil, err } + if len(tt) == 0 { + return nil, nil + } + if p.StepID == 0 && len(tt) > 0 { return tt[0], nil } else { @@ -582,6 +592,30 @@ func (svc *workflow) Exec(ctx context.Context, workflowID uint64, p types.Workfl } } + if !p.Trace { + // when not doing a trace (designing the workflow) + // we need to be more strict and disallow execution of + // the misconfigured workflows and use of disabled triggers + if t == nil { + return nil, WorkflowErrUnknownWorkflowStep() + } else if !t.Enabled { + return nil, WorkflowErrDisabled() + } + } + + if t != nil { + wap.setTrigger(t) + p.StepID = t.StepID + p.EventType = t.EventType + p.ResourceType = t.ResourceType + + // merge with input from trigger + // with trigger input vars are overwritten by input vars + p.Input = t.Input.MustMerge(p.Input) + } else { + p.EventType = "onTrace" + } + return nil, nil }() @@ -589,39 +623,16 @@ func (svc *workflow) Exec(ctx context.Context, workflowID uint64, p types.Workfl return } - if !p.Trace { - if t == nil { - return WorkflowErrUnknownWorkflowStep() - } else if !t.Enabled { - return WorkflowErrDisabled() - } - } - - if t != nil { - wap.setTrigger(t) - p.StepID = t.StepID - p.EventType = t.EventType - p.ResourceType = t.ResourceType - - // merge with input from trigger - // with trigger input vars are overwritten by input vars - p.Input = t.Input.MustMerge(p.Input) - } else { - p.EventType = "onTrace" - } - wait, sessionID, err = svc.exec(ctx, wf, p) if err != nil { return err } - if p.Async { - if !p.Wait && wf.CheckDeferred() { - // deferred workflow, return right away and keep the workflow session - // running without waiting for the execution - return nil - } + if p.Async && !p.Wait && wf.CheckDeferred() { + // deferred workflow, return right away and keep the workflow session + // running without waiting for the execution + return nil } // wait for the workflow to complete diff --git a/automation/service/workflow_converter.go b/automation/service/workflow_converter.go index 2d980bfb6..4ad1674d2 100644 --- a/automation/service/workflow_converter.go +++ b/automation/service/workflow_converter.go @@ -15,15 +15,18 @@ import ( ) type ( + wfExecutor interface { + handleToID(string) uint64 + Exec(ctx context.Context, workflowID uint64, p types.WorkflowExecParams) (*expr.Vars, uint64, types.Stacktrace, error) + } + workflowConverter struct { // workflow function registry reg *registry parser expr.Parsable log *zap.Logger - graphs interface { - handleToID(string) uint64 - } + graphs wfExecutor } ) @@ -205,6 +208,9 @@ func (svc workflowConverter) workflowStepDefConv(g *wfexec.Graph, def *types.Wor case types.WorkflowStepKindContinue: return svc.convContinueStep() + case types.WorkflowStepKindExecWorkflow: + return svc.convExecWorkflowStep(def, s) + default: return nil, errors.Internal("unsupported step kind %q", s.Kind) } @@ -506,6 +512,82 @@ func (svc workflowConverter) convContinueStep() (wfexec.Step, error) { } +// creates workflow-executor step +// +// Expects max ONE outgoing paths and ONE incoming path +// +// +func (svc workflowConverter) convExecWorkflowStep(wf *types.Workflow, s *types.WorkflowStep) (_ wfexec.Step, err error) { + const ( + // sub-workflow's scope + // expecting expr.Vars + argNameScope = "scope" + + // workflow id or handle + argNameWorkflow = "workflow" + ) + + return wfexec.NewGenericStep(func(ctx context.Context, r *wfexec.ExecRequest) (resp wfexec.ExecResponse, err error) { + var ( + p = types.WorkflowExecParams{ + // sub-workflows should not be executed asynchronously + // might lead to uncontrolled resource consumption + Async: false, + + // @todo figure out how check if trace is enabled + Trace: false, + + // always wait for the sub-workflow even when deferred + Wait: true, + + Input: expr.EmptyVars(), + + // who's calling sub-workflow? + CallerSessionID: r.SessionID, + CallerStepID: s.ID, + CallerWorkflowID: wf.ID, + } + + result *expr.Vars + + workflowID uint64 + workflowHandle string + ) + + if ap, err := processArguments(ctx, s.Arguments, r.Scope); err != nil { + return nil, err + } else { + if ap.string(argNameWorkflow, &workflowHandle) { + if workflowID = svc.graphs.handleToID(workflowHandle); workflowID == 0 { + return nil, errors.NotFound("workflow with handle %s not found", workflowHandle) + } + } else if !ap.uint64(argNameWorkflow, &workflowID) { + return nil, errors.InvalidData("workflow ID must be provided") + } + + // let's make sure we're not calling ourselves + if p.CallerWorkflowID == workflowID { + return nil, errors.InvalidData("recursive workflow call is not allowed") + } + + // scope for the sub-workflow + ap.vars(argNameScope, p.Input) + } + + result, _, _, err = svc.graphs.Exec(ctx, workflowID, p) + if err != nil { + return + } + + result, err = types.ExprSet(s.Results).Eval(ctx, result) + if err != nil { + return + } + + return result, nil + }), nil +} + func (svc workflowConverter) parseExpressions(ee ...*types.Expr) (err error) { for _, e := range ee { @@ -614,25 +696,31 @@ func verifyStep(s *types.WorkflowStep, in, out types.WorkflowPathSet) types.Work } // checks if argument is present - checkArg = func(argName string, typ expr.Type) func() error { + checkArg = func(argName string, tt ...expr.Type) func() error { return func() error { msgArg := types.ExprSet(s.Arguments).GetByTarget(argName) - if msgArg != nil && msgArg.Type != typ.Type() { - return errors.Internal("%s argument on %s step must be %s, got type '%s'", argName, s.Kind, typ.Type(), msgArg.Type) + if msgArg == nil { + return nil } - return nil + for _, typ := range tt { + if msgArg.Type == typ.Type() { + return nil + } + } + + return errors.Internal("unexpected type %q for argument %q no step type %q", msgArg.Type, argName, s.Kind) } } // checks if argument is present - requiredArg = func(argName string, typ expr.Type) func() error { + requiredArg = func(argName string, tt ...expr.Type) func() error { return func() error { if msgArg := types.ExprSet(s.Arguments).GetByTarget(argName); msgArg == nil { return errors.Internal("%s step expects to have '%s' argument", s.Kind, argName) } - return checkArg(argName, typ)() + return checkArg(argName, tt...)() } } @@ -754,6 +842,21 @@ func verifyStep(s *types.WorkflowStep, in, out types.WorkflowPathSet) types.Work last, ) + case types.WorkflowStepKindExecWorkflow: + checks = append(checks, + noRef, + // required "workflow" with handle or ID + requiredArg("workflow", expr.String{}, expr.ID{}), + + // optional "scope" as expr.Vars + checkArg("scope", expr.Vars{}), + + // expecting at least workflow + optional scope, nothing more + count(1, 2, arguments), + + count(0, 1, outbound), + ) + case "": return ii.Append(fmt.Errorf("missing step kind"), nil) diff --git a/automation/service/workflow_step_args.go b/automation/service/workflow_step_args.go new file mode 100644 index 000000000..f8a972755 --- /dev/null +++ b/automation/service/workflow_step_args.go @@ -0,0 +1,75 @@ +package service + +import ( + "context" + "github.com/cortezaproject/corteza-server/automation/types" + "github.com/cortezaproject/corteza-server/pkg/expr" +) + +type ( + // argsProc is a helper to process arguments + // + // when initialised it evaluates expressions using given scope + // it provides methods to extract values from the result + argsProc struct { + expr types.ExprSet + result *expr.Vars + } +) + +func processArguments(ctx context.Context, expr []*types.Expr, scope *expr.Vars) (p *argsProc, err error) { + p = &argsProc{expr: expr} + + if p.result, err = p.expr.Eval(ctx, scope); err != nil { + return nil, err + } + + return +} + +func (p *argsProc) bool(name string, val *bool) bool { return getStepArg(p, name, val) } +func (p *argsProc) string(name string, val *string) bool { return getStepArg(p, name, val) } +func (p *argsProc) uint64(name string, val *uint64) bool { return getStepArg(p, name, val) } + +// vars handles assigning of expr.Vars +// +// not as straightforward as plain types +func (p *argsProc) vars(name string, val *expr.Vars) bool { + // make an auxiliary variable to hold the results q + aux := make(map[string]expr.TypedValue) + + // extract it + if !getStepArg(p, name, &aux) { + return false + } + + if val == nil { + panic("initialize Vars before calling the function") + } + + vars, _ := expr.NewVars(aux) + *val = *vars + return true +} + +// getStepArg is a helper to extract step argument from result +func getStepArg[T any](p *argsProc, arg string, val *T) bool { + var ( + aux any + ) + + if p.result.Has(arg) { + aux = expr.Must(p.result.Select(arg)).Get() + } else if exp := p.expr.GetByTarget(arg); exp != nil { + aux = p.expr.GetByTarget(arg).Value.(T) + } else { + return false + } + + if conv, is := aux.(T); is { + *val = conv + return true + } + + return false +} diff --git a/automation/service/workflow_step_args_test.go b/automation/service/workflow_step_args_test.go new file mode 100644 index 000000000..b08ce49b4 --- /dev/null +++ b/automation/service/workflow_step_args_test.go @@ -0,0 +1,83 @@ +package service + +import ( + "context" + "github.com/cortezaproject/corteza-server/automation/types" + "github.com/cortezaproject/corteza-server/pkg/expr" + "github.com/stretchr/testify/require" + "testing" +) + +func TestProcessArguments(t *testing.T) { + t.Run("empty", func(t *testing.T) { + var ( + req = require.New(t) + p, err = processArguments(context.Background(), []*types.Expr{}, nil) + + tBool bool + tStr string + tUint uint64 + tVars *expr.Vars + ) + + req.NoError(err) + req.NotNil(p) + + req.False(p.bool("not-there", &tBool)) + req.False(tBool) + + req.False(p.string("not-there", &tStr)) + req.Empty(tStr) + + req.False(p.uint64("not-there", &tUint)) + req.Zero(tUint) + + req.False(p.vars("not-there", tVars)) + req.Zero(tVars) + }) + + t.Run("expr", func(t *testing.T) { + var ( + req = require.New(t) + ee = []*types.Expr{ + types.NewTypedExpr("tBool", "true", &expr.Boolean{}), + types.NewTypedExpr("tString", "\"foo\"", &expr.String{}), + types.NewTypedExpr("tUint", "42", &expr.UnsignedInteger{}), + types.NewTypedExpr("tVars", "{}", &expr.Vars{}), + types.NewTypedExpr("tVars.foo", "\"bar\"", &expr.String{}), + } + ) + + req.NoError(expr.NewGvalParser().ParseEvaluators(func() []expr.Evaluator { + oo := make([]expr.Evaluator, len(ee)) + for i, e := range ee { + oo[i] = e + } + return oo + }()...)) + + var ( + p, err = processArguments(context.Background(), ee, nil) + + tBool bool + tStr string + tUint uint64 + tVars = &expr.Vars{} + ) + + req.NoError(err) + req.NotNil(p) + + req.True(p.bool("tBool", &tBool)) + req.True(tBool) + + req.True(p.string("tString", &tStr)) + req.Equal("foo", tStr) + + req.True(p.uint64("tUint", &tUint)) + req.Equal(uint64(42), tUint) + + req.True(p.vars("tVars", tVars)) + req.Equal(expr.Must(expr.NewVars(map[string]any{"foo": "bar"})), tVars) + }) +} diff --git a/automation/types/expr.go b/automation/types/expr.go index 85ac076bd..7eabc6e89 100644 --- a/automation/types/expr.go +++ b/automation/types/expr.go @@ -54,6 +54,10 @@ func NewExpr(target, typ, expr string) (e *Expr, err error) { return &Expr{Expr: expr, Target: target, Type: typ}, nil } +func NewTypedExpr(target string, expr string, typ expr.Type) *Expr { + return &Expr{Expr: expr, Target: target, typ: typ} +} + func (e *Expr) SetType(fn func(string) (expr.Type, error)) error { if typ, err := fn(e.Type); err != nil { return err diff --git a/automation/types/step.go b/automation/types/step.go index 6e83642d3..df0aeffbe 100644 --- a/automation/types/step.go +++ b/automation/types/step.go @@ -58,19 +58,20 @@ type ( ) const ( - WorkflowStepKindExpressions WorkflowStepKind = "expressions" // no ref - WorkflowStepKindGateway WorkflowStepKind = "gateway" // ref = join|fork|excl|incl - WorkflowStepKindFunction WorkflowStepKind = "function" // ref = - WorkflowStepKindIterator WorkflowStepKind = "iterator" // ref = - WorkflowStepKindError WorkflowStepKind = "error" // no ref - WorkflowStepKindTermination WorkflowStepKind = "termination" // no ref - WorkflowStepKindPrompt WorkflowStepKind = "prompt" // ref = - WorkflowStepKindDelay WorkflowStepKind = "delay" // no ref - WorkflowStepKindErrHandler WorkflowStepKind = "error-handler" // no ref - WorkflowStepKindVisual WorkflowStepKind = "visual" // ref = <*> - WorkflowStepKindDebug WorkflowStepKind = "debug" // ref = <*> - WorkflowStepKindBreak WorkflowStepKind = "break" // ref = <*> - WorkflowStepKindContinue WorkflowStepKind = "continue" // ref = <*> + WorkflowStepKindExpressions WorkflowStepKind = "expressions" // no ref + WorkflowStepKindGateway WorkflowStepKind = "gateway" // ref = join|fork|excl|incl + WorkflowStepKindFunction WorkflowStepKind = "function" // ref = + WorkflowStepKindIterator WorkflowStepKind = "iterator" // ref = + WorkflowStepKindError WorkflowStepKind = "error" // no ref + WorkflowStepKindTermination WorkflowStepKind = "termination" // no ref + WorkflowStepKindPrompt WorkflowStepKind = "prompt" // ref = + WorkflowStepKindDelay WorkflowStepKind = "delay" // no ref + WorkflowStepKindErrHandler WorkflowStepKind = "error-handler" // no ref + WorkflowStepKindVisual WorkflowStepKind = "visual" // ref = <*> + WorkflowStepKindDebug WorkflowStepKind = "debug" // ref = <*> + WorkflowStepKindBreak WorkflowStepKind = "break" // ref = <*> + WorkflowStepKindContinue WorkflowStepKind = "continue" // ref = <*> + WorkflowStepKindExecWorkflow WorkflowStepKind = "exec-workflow" // no ref ) // IsDeferred fn returns true if type of step is delay or prompt @@ -84,7 +85,7 @@ func (s WorkflowStep) IsDeferred() (is bool) { return false } -// HasDeferred fn returns true if type of any of workflow's steps is delay or prompt +// HasDeferred fn returns true if wf-step is delay or prompt func (vv WorkflowStepSet) HasDeferred() bool { for _, s := range vv { if s.IsDeferred() { diff --git a/tests/workflows/0017_subworkflows_test.go b/tests/workflows/0017_subworkflows_test.go new file mode 100644 index 000000000..95d9e9663 --- /dev/null +++ b/tests/workflows/0017_subworkflows_test.go @@ -0,0 +1,33 @@ +package workflows + +import ( + "context" + "github.com/stretchr/testify/require" + "testing" + + "github.com/cortezaproject/corteza-server/automation/types" +) + +func Test0017_subworkflows(t *testing.T) { + var ( + ctx = bypassRBAC(context.Background()) + req = require.New(t) + ) + + loadScenario(ctx, t) + + type ( + testInput struct { + Out string + } + ) + var ( + aux = testInput{} + vars, _ = mustExecWorkflow(ctx, t, "main", types.WorkflowExecParams{}) + expected = testInput{Out: "main + sub"} + ) + + req.NoError(vars.Decode(&aux)) + req.Equal(expected, aux) + +} diff --git a/tests/workflows/testdata/S0017_subworkflows/workflow.yaml b/tests/workflows/testdata/S0017_subworkflows/workflow.yaml new file mode 100644 index 000000000..e2d53eed4 --- /dev/null +++ b/tests/workflows/testdata/S0017_subworkflows/workflow.yaml @@ -0,0 +1,44 @@ +workflows: + sub: + enabled: true + trace: true + + steps: + - stepID: 1 + kind: expressions + arguments: + - { target: breadcrumbs, expr: 'breadcrumbs + " + sub"' } + - stepID: 2 + kind: termination + + paths: + - { parentID: 1, childID: 2 } + + main: + enabled: true + trace: true + triggers: + - enabled: true + stepID: 1 + + steps: + - stepID: 1 + kind: expressions + arguments: + - { target: scope4sub, type: Vars } + - { target: scope4sub.breadcrumbs, value: 'main' } + + - stepID: 2 + kind: exec-workflow + arguments: + - { target: workflow, type: Handle, value: "sub" } + - { target: scope, type: Vars, source: scope4sub } + results: + - { target: out, source: breadcrumbs } + + - stepID: 3 + kind: termination + + paths: + - { parentID: 1, childID: 2 } + - { parentID: 2, childID: 3 }