DeforaNetworks/corteza
3
0
Fork 0
Code Activity

Return workflow session ID when executing wf

Browse Source 
It also fixes missimplemented async exec.
This commit is contained in:
Denis Arh
2022-07-21 20:35:54 +02:00
parent bba02ec9df
commit 8e42876010
9 changed files with 56 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
automation/rest/workflow.go
+7 -6
View File
@@ -24,7 +24,7 @@ type (
Update(ctx context.Context, upd *types.Workflow) (*types.Workflow, error)
DeleteByID(ctx context.Context, workflowID uint64) error
UndeleteByID(ctx context.Context, workflowID uint64) error
Exec(ctx context.Context, workflowID uint64, p types.WorkflowExecParams) (*expr.Vars, types.Stacktrace, error)
Exec(ctx context.Context, workflowID uint64, p types.WorkflowExecParams) (*expr.Vars, uint64, types.Stacktrace, error)
}
// cross-link with compose service to load module on resolved records
@@ -64,9 +64,10 @@ type (
}
workflowExecPayload struct {
Results *expr.Vars `json:"results"`
Trace types.Stacktrace `json:"trace,omitempty"`
Error string `json:"error,omitempty"`
Results *expr.Vars `json:"results"`
Trace types.Stacktrace `json:"trace,omitempty"`
SessionID uint64 `json:"sessionID,string,omitempty"`
Error string `json:"error,omitempty"`
}
)
@@ -178,7 +179,7 @@ func (ctrl Workflow) Exec(ctx context.Context, r *request.WorkflowExec) (interfa
}
}
// Now that all types are resolved we have to load modules and link them to records
// Now when all types are resolved we have to load modules and link them to records
//
// Very naive approach for now.
execParams.Input.Each(func(k string, v expr.TypedValue) error {
@@ -199,7 +200,7 @@ func (ctrl Workflow) Exec(ctx context.Context, r *request.WorkflowExec) (interfa
return nil
})
wep.Results, wep.Trace, err = ctrl.svc.Exec(ctx, r.WorkflowID, execParams)
wep.Results, wep.SessionID, wep.Trace, err = ctrl.svc.Exec(ctx, r.WorkflowID, execParams)
if err != nil && wep.Trace != nil && r.Trace {
// in case of an error & trace enabled (and stacktrace present)
automation/service/session.go
+14 -12
View File
@@ -49,7 +49,7 @@ type (
CanManageSessionsOnWorkflow(context.Context, *types.Workflow) bool
}
WaitFn func(ctx context.Context) (*expr.Vars, wfexec.SessionStatus, types.Stacktrace, error)
WaitFn func(ctx context.Context) (*expr.Vars, uint64, wfexec.SessionStatus, types.Stacktrace, error)
)
const (
@@ -175,23 +175,23 @@ func (svc *session) PendingPrompts(ctx context.Context) (pp []*wfexec.PendingPro
// used for the execution of the workflow. See watch function!
//
// It does not check user's permissions to execute workflow(s) so it should be used only when !
func (svc *session) Start(ctx context.Context, g *wfexec.Graph, ssp types.SessionStartParams) (wait WaitFn, err error) {
func (svc *session) Start(ctx context.Context, g *wfexec.Graph, ssp types.SessionStartParams) (wait WaitFn, _ uint64, err error) {
var (
start wfexec.Step
)
if g == nil {
return nil, errors.InvalidData("cannot start workflow, uninitialized graph")
return nil, 0, errors.InvalidData("cannot start workflow, uninitialized graph")
}
if len(ssp.CallStack) > svc.opt.CallStackSize {
return nil, WorkflowErrMaximumCallStackSizeExceeded()
return nil, 0, WorkflowErrMaximumCallStackSizeExceeded()
}
ssp.CallStack = append(ssp.CallStack, ssp.WorkflowID)
if ssp.Invoker == nil {
return nil, errors.InvalidData("cannot start workflow without user")
return nil, 0, errors.InvalidData("cannot start workflow without user")
}
if ssp.Runner == nil {
@@ -205,14 +205,14 @@ func (svc *session) Start(ctx context.Context, g *wfexec.Graph, ssp types.Sessio
case 1:
start = oo[0]
case 0:
return nil, errors.InvalidData("could not find starting step")
return nil, 0, errors.InvalidData("could not find starting step")
default:
return nil, errors.InvalidData("cannot start workflow session multiple starting steps found")
return nil, 0, errors.InvalidData("cannot start workflow session multiple starting steps found")
}
} else if start = g.StepByID(ssp.StepID); start == nil {
return nil, errors.InvalidData("trigger staring step references non-existing step")
return nil, 0, errors.InvalidData("trigger staring step references non-existing step")
} else if len(g.Parents(g.StepByID(ssp.StepID))) > 0 {
return nil, errors.InvalidData("cannot start workflow on a step with parents")
return nil, 0, errors.InvalidData("cannot start workflow on a step with parents")
}
var (
@@ -233,9 +233,11 @@ func (svc *session) Start(ctx context.Context, g *wfexec.Graph, ssp types.Sessio
return
}
return func(ctx context.Context) (*expr.Vars, wfexec.SessionStatus, types.Stacktrace, error) {
return ses.WaitResults(ctx)
}, nil
return func(ctx context.Context) (scope *expr.Vars, sessionID uint64, status wfexec.SessionStatus, stacktrace types.Stacktrace, err error) {
sessionID = ses.ID
scope, status, stacktrace, err = ses.WaitResults(ctx)
return
}, ses.ID, nil
}
// Resume resumes suspended session/state
automation/service/session_test.go
+4 -4
View File
@@ -20,16 +20,16 @@ func TestSession_Start(t *testing.T) {
err error
)
_, err = ses.Start(ctx, g, types.SessionStartParams{Invoker: auth.Anonymous()})
_, _, err = ses.Start(ctx, g, types.SessionStartParams{Invoker: auth.Anonymous()})
req.EqualError(err, "could not find starting step")
g.AddStep(wfexec.NewGenericStep(nil))
_, err = ses.Start(ctx, g, types.SessionStartParams{StepID: 4321, Invoker: auth.Anonymous()})
_, _, err = ses.Start(ctx, g, types.SessionStartParams{StepID: 4321, Invoker: auth.Anonymous()})
req.EqualError(err, "trigger staring step references non-existing step")
// Adding another orphaned step and starting session w/o explicitly specifying the starting step
g.AddStep(wfexec.NewGenericStep(nil))
_, err = ses.Start(ctx, g, types.SessionStartParams{Invoker: auth.Anonymous()})
_, _, err = ses.Start(ctx, g, types.SessionStartParams{Invoker: auth.Anonymous()})
req.EqualError(err, "cannot start workflow session multiple starting steps found")
// add a generic step with a known ID so we can use it as a starting point
@@ -38,7 +38,7 @@ func TestSession_Start(t *testing.T) {
g.AddStep(s)
// add parents to the 42 step
g.AddStep(wfexec.NewGenericStep(nil), s)
_, err = ses.Start(ctx, g, types.SessionStartParams{StepID: 42, Invoker: auth.Anonymous()})
_, _, err = ses.Start(ctx, g, types.SessionStartParams{StepID: 42, Invoker: auth.Anonymous()})
req.EqualError(err, "cannot start workflow on a step with parents")
}
automation/service/workflow.go
+15 -15
View File
@@ -531,14 +531,14 @@ func (svc *workflow) updateCache(wf *types.Workflow, runAs intAuth.Identifiable,
return
}
func (svc *workflow) Exec(ctx context.Context, workflowID uint64, p types.WorkflowExecParams) (*expr.Vars, types.Stacktrace, error) {
func (svc *workflow) Exec(ctx context.Context, workflowID uint64, p types.WorkflowExecParams) (*expr.Vars, uint64, types.Stacktrace, error) {
var (
wap = &workflowActionProps{}
t *types.Trigger
results *expr.Vars
wait WaitFn
wap = &workflowActionProps{}
t *types.Trigger
results *expr.Vars
wait WaitFn
stacktrace types.Stacktrace
sessionID uint64
)
err := func() (err error) {
@@ -610,13 +610,13 @@ func (svc *workflow) Exec(ctx context.Context, workflowID uint64, p types.Workfl
p.EventType = "onTrace"
}
wait, err = svc.exec(ctx, wf, p)
wait, sessionID, err = svc.exec(ctx, wf, p)
if err != nil {
return err
}
if !p.Async {
if p.Async {
if !p.Wait && wf.CheckDeferred() {
// deferred workflow, return right away and keep the workflow session
// running without waiting for the execution
@@ -627,11 +627,11 @@ func (svc *workflow) Exec(ctx context.Context, workflowID uint64, p types.Workfl
// wait for the workflow to complete
// reuse scope for results
// this will be decoded back to event properties
results, _, stacktrace, err = wait(ctx)
results, sessionID, _, stacktrace, err = wait(ctx)
return err
}()
return results, stacktrace, svc.recordAction(ctx, wap, WorkflowActionExecute, err)
return results, sessionID, stacktrace, svc.recordAction(ctx, wap, WorkflowActionExecute, err)
}
// validates workflow by trying to convert it to graph and checking assigned triggers
@@ -675,16 +675,16 @@ func (svc *workflow) validateWorkflow(ctx context.Context, wf *types.Workflow) (
return
}
func (svc *workflow) exec(ctx context.Context, wf *types.Workflow, p types.WorkflowExecParams) (WaitFn, error) {
func (svc *workflow) exec(ctx context.Context, wf *types.Workflow, p types.WorkflowExecParams) (WaitFn, uint64, error) {
if wf.Issues != nil {
return nil, wf.Issues
return nil, 0, wf.Issues
}
defer svc.mux.Unlock()
svc.mux.Lock()
if svc.cache[wf.ID] == nil {
return nil, WorkflowErrInvalidID()
return nil, 0, WorkflowErrInvalidID()
}
var (
@@ -726,7 +726,7 @@ func makeWorkflowHandler(svc *workflow, wf *types.Workflow, t *types.Trigger) ev
}
}
wait, err := svc.exec(ctx, wf, types.WorkflowExecParams{
wait, _, err := svc.exec(ctx, wf, types.WorkflowExecParams{
StepID: t.StepID,
EventType: t.EventType,
ResourceType: t.ResourceType,
@@ -749,7 +749,7 @@ func makeWorkflowHandler(svc *workflow, wf *types.Workflow, t *types.Trigger) ev
// wait for the workflow to complete
// reuse scope for results
// this will be decoded back to event properties
scope, _, _, err = wait(ctx)
scope, _, _, _, err = wait(ctx)
if err != nil {
return
}
pkg/apigw/filter/processer.go
+2 -2
View File
@@ -31,7 +31,7 @@ type (
WfExecer interface {
Load(ctx context.Context) error
Exec(ctx context.Context, workflowID uint64, p atypes.WorkflowExecParams) (*expr.Vars, atypes.Stacktrace, error)
Exec(ctx context.Context, workflowID uint64, p atypes.WorkflowExecParams) (*expr.Vars, uint64, atypes.Stacktrace, error)
}
processerPayload struct {
@@ -120,7 +120,7 @@ func (h workflow) Handler() types.HandlerFunc {
Input: in,
}
out, _, err := h.d.Exec(ctx, h.params.Workflow, wp)
out, _, _, err := h.d.Exec(ctx, h.params.Workflow, wp)
if err != nil {
return pe.Internal("could not exec workflow: %v", err)
pkg/apigw/filter/processer_test.go
+6 -6
View File
@@ -24,7 +24,7 @@ import (
type (
wfServicer struct {
load func(ctx context.Context) error
exec func(ctx context.Context, workflowID uint64, p atypes.WorkflowExecParams) (*expr.Vars, atypes.Stacktrace, error)
exec func(ctx context.Context, workflowID uint64, p atypes.WorkflowExecParams) (*expr.Vars, uint64, atypes.Stacktrace, error)
}
)
@@ -50,8 +50,8 @@ func Test_processerWorkflow(t *testing.T) {
load: func(ctx context.Context) error {
return nil
},
exec: func(ctx context.Context, workflowID uint64, p atypes.WorkflowExecParams) (*expr.Vars, atypes.Stacktrace, error) {
return must(expr.NewVars(map[string]interface{}{"foo": "bar"})), make([]*wfexec.Frame, 0), errors.New("mocked error")
exec: func(ctx context.Context, workflowID uint64, p atypes.WorkflowExecParams) (*expr.Vars, uint64, atypes.Stacktrace, error) {
return must(expr.NewVars(map[string]interface{}{"foo": "bar"})), 0, make([]*wfexec.Frame, 0), errors.New("mocked error")
},
},
},
@@ -63,8 +63,8 @@ func Test_processerWorkflow(t *testing.T) {
load: func(ctx context.Context) error {
return nil
},
exec: func(ctx context.Context, workflowID uint64, p atypes.WorkflowExecParams) (*expr.Vars, atypes.Stacktrace, error) {
return must(expr.NewVars(map[string]interface{}{"foo": "bar"})), make([]*wfexec.Frame, 0), nil
exec: func(ctx context.Context, workflowID uint64, p atypes.WorkflowExecParams) (*expr.Vars, uint64, atypes.Stacktrace, error) {
return must(expr.NewVars(map[string]interface{}{"foo": "bar"})), 0, make([]*wfexec.Frame, 0), nil
},
},
},
@@ -215,7 +215,7 @@ func (f wfServicer) Load(ctx context.Context) error {
return f.load(ctx)
}
func (f wfServicer) Exec(ctx context.Context, workflowID uint64, p atypes.WorkflowExecParams) (*expr.Vars, atypes.Stacktrace, error) {
func (f wfServicer) Exec(ctx context.Context, workflowID uint64, p atypes.WorkflowExecParams) (*expr.Vars, uint64, atypes.Stacktrace, error) {
return f.exec(ctx, workflowID, p)
}
tests/workflows/0015_step_issue_test.go
+4 -4
View File
@@ -18,7 +18,7 @@ func Test0015_step_issue(t *testing.T) {
loadScenario(ctx, t)
t.Run("exclusive gateway step issue", func(t *testing.T) {
_, _, err := execWorkflow(ctx, "case1", types.WorkflowExecParams{})
_, _, _, err := execWorkflow(ctx, "case1", types.WorkflowExecParams{})
issues, is := err.(types.WorkflowIssueSet)
req.True(is)
@@ -30,7 +30,7 @@ func Test0015_step_issue(t *testing.T) {
})
t.Run("inclusive gateway step issue", func(t *testing.T) {
_, _, err := execWorkflow(ctx, "case2", types.WorkflowExecParams{})
_, _, _, err := execWorkflow(ctx, "case2", types.WorkflowExecParams{})
issues, is := err.(types.WorkflowIssueSet)
req.True(is)
@@ -42,7 +42,7 @@ func Test0015_step_issue(t *testing.T) {
})
t.Run("function step issue", func(t *testing.T) {
_, _, err := execWorkflow(ctx, "case3", types.WorkflowExecParams{})
_, _, _, err := execWorkflow(ctx, "case3", types.WorkflowExecParams{})
issues, is := err.(types.WorkflowIssueSet)
req.True(is)
@@ -53,7 +53,7 @@ func Test0015_step_issue(t *testing.T) {
})
t.Run("iterator step issue", func(t *testing.T) {
_, _, err := execWorkflow(ctx, "case4", types.WorkflowExecParams{})
_, _, _, err := execWorkflow(ctx, "case4", types.WorkflowExecParams{})
issues, is := err.(types.WorkflowIssueSet)
req.True(is)
tests/workflows/exec_permissions_test.go
+1 -1
View File
@@ -56,7 +56,7 @@ func Test_exec_permissions(t *testing.T) {
t.Run("exec denied", func(t *testing.T) {
req = require.New(t)
ctx = auth.SetIdentityToContext(ctx, execDenied)
_, _, err = execWorkflow(ctx, "wf", types.WorkflowExecParams{})
_, _, _, err = execWorkflow(ctx, "wf", types.WorkflowExecParams{})
req.ErrorIs(err, service.WorkflowErrNotAllowedToExecute())
})
}
tests/workflows/main_test.go
+3 -3
View File
@@ -165,10 +165,10 @@ func bypassRBAC(ctx context.Context) context.Context {
return auth.SetIdentityToContext(ctx, u)
}
func execWorkflow(ctx context.Context, name string, p autTypes.WorkflowExecParams) (*expr.Vars, autTypes.Stacktrace, error) {
func execWorkflow(ctx context.Context, name string, p autTypes.WorkflowExecParams) (*expr.Vars, uint64, autTypes.Stacktrace, error) {
wf, err := defStore.LookupAutomationWorkflowByHandle(ctx, name)
if err != nil {
return nil, nil, err
return nil, 0, nil, err
}
return service.DefaultWorkflow.Exec(ctx, wf.ID, p)
@@ -176,7 +176,7 @@ func execWorkflow(ctx context.Context, name string, p autTypes.WorkflowExecParam
func mustExecWorkflow(ctx context.Context, t *testing.T, name string, p autTypes.WorkflowExecParams) (vars *expr.Vars, strace autTypes.Stacktrace) {
var err error
vars, strace, err = execWorkflow(ctx, name, p)
vars, _, strace, err = execWorkflow(ctx, name, p)
if err != nil {
if issues, is := err.(autTypes.WorkflowIssueSet); is {
for _, i := range issues {