DeforaNetworks/corteza
3
0
Fork 0
Code Activity

Add support for exec-workflow step type

Browse Source
This commit is contained in:
Denis Arh
2022-09-01 15:27:58 +02:00
parent 5d45846fdd
commit a32a580a28
8 changed files with 404 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
automation/service/workflow.go
+38 -27
View File
@@ -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
automation/service/workflow_converter.go
+112 -9
View File
@@ -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)
automation/service/workflow_step_args.go
+75
View File
@@ -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
}
automation/service/workflow_step_args_test.go
+83
View File
@@ -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)
})
}
automation/types/expr.go
+4
View File
@@ -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
automation/types/step.go
+15 -14
View File
@@ -58,19 +58,20 @@ type (
)
const (
WorkflowStepKindExpressions WorkflowStepKind = "expressions" // no ref
WorkflowStepKindGateway WorkflowStepKind = "gateway" // ref = join|fork|excl|incl
WorkflowStepKindFunction WorkflowStepKind = "function" // ref = <function ref>
WorkflowStepKindIterator WorkflowStepKind = "iterator" // ref = <iterator function ref>
WorkflowStepKindError WorkflowStepKind = "error" // no ref
WorkflowStepKindTermination WorkflowStepKind = "termination" // no ref
WorkflowStepKindPrompt WorkflowStepKind = "prompt" // ref = <client function>
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 = <function ref>
WorkflowStepKindIterator WorkflowStepKind = "iterator" // ref = <iterator function ref>
WorkflowStepKindError WorkflowStepKind = "error" // no ref
WorkflowStepKindTermination WorkflowStepKind = "termination" // no ref
WorkflowStepKindPrompt WorkflowStepKind = "prompt" // ref = <client function>
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() {
tests/workflows/0017_subworkflows_test.go
+33
View File
@@ -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)
}
tests/workflows/testdata/S0017_subworkflows/workflow.yaml
Vendored
+44
View File
@@ -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 }