From 8e42876010d8578f457c104f275f9460aea572ff Mon Sep 17 00:00:00 2001 From: Denis Arh Date: Thu, 21 Jul 2022 20:35:54 +0200 Subject: [PATCH] Return workflow session ID when executing wf It also fixes missimplemented async exec. --- automation/rest/workflow.go | 13 +++++----- automation/service/session.go | 26 ++++++++++---------- automation/service/session_test.go | 8 +++---- automation/service/workflow.go | 30 ++++++++++++------------ pkg/apigw/filter/processer.go | 4 ++-- pkg/apigw/filter/processer_test.go | 12 +++++----- tests/workflows/0015_step_issue_test.go | 8 +++---- tests/workflows/exec_permissions_test.go | 2 +- tests/workflows/main_test.go | 6 ++--- 9 files changed, 56 insertions(+), 53 deletions(-) diff --git a/automation/rest/workflow.go b/automation/rest/workflow.go index a16787ab9..423da34b8 100644 --- a/automation/rest/workflow.go +++ b/automation/rest/workflow.go @@ -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) diff --git a/automation/service/session.go b/automation/service/session.go index e623e3bdb..4791105f7 100644 --- a/automation/service/session.go +++ b/automation/service/session.go @@ -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 diff --git a/automation/service/session_test.go b/automation/service/session_test.go index a2f97260d..67943880f 100644 --- a/automation/service/session_test.go +++ b/automation/service/session_test.go @@ -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") } diff --git a/automation/service/workflow.go b/automation/service/workflow.go index 510046a49..472e154d8 100644 --- a/automation/service/workflow.go +++ b/automation/service/workflow.go @@ -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 } diff --git a/pkg/apigw/filter/processer.go b/pkg/apigw/filter/processer.go index 489d5425b..2f40a107a 100644 --- a/pkg/apigw/filter/processer.go +++ b/pkg/apigw/filter/processer.go @@ -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) diff --git a/pkg/apigw/filter/processer_test.go b/pkg/apigw/filter/processer_test.go index 12a53387f..3129fc117 100644 --- a/pkg/apigw/filter/processer_test.go +++ b/pkg/apigw/filter/processer_test.go @@ -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) } diff --git a/tests/workflows/0015_step_issue_test.go b/tests/workflows/0015_step_issue_test.go index 1dfdfbb53..0b8bfbd40 100644 --- a/tests/workflows/0015_step_issue_test.go +++ b/tests/workflows/0015_step_issue_test.go @@ -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) diff --git a/tests/workflows/exec_permissions_test.go b/tests/workflows/exec_permissions_test.go index 31dbdf6d7..3d1ab4d23 100644 --- a/tests/workflows/exec_permissions_test.go +++ b/tests/workflows/exec_permissions_test.go @@ -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()) }) } diff --git a/tests/workflows/main_test.go b/tests/workflows/main_test.go index a1f9c5310..db239e694 100644 --- a/tests/workflows/main_test.go +++ b/tests/workflows/main_test.go @@ -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 {