diff --git a/automation/service/session.go b/automation/service/session.go index 5f0cf36c8..fd82d0953 100644 --- a/automation/service/session.go +++ b/automation/service/session.go @@ -376,7 +376,7 @@ func (svc *session) stateChangeHandler(ctx context.Context) wfexec.StateChangeHa frame.ElapsedTime = uint(frame.CreatedAt.Sub(ses.RuntimeStacktrace[0].CreatedAt) / time.Millisecond) } - ses.RuntimeStacktrace = append(ses.RuntimeStacktrace, frame) + ses.AppendRuntimeStacktrace(frame) switch i { case wfexec.SessionPrompted: diff --git a/automation/types/session.go b/automation/types/session.go index 041ca820b..411ad52b7 100644 --- a/automation/types/session.go +++ b/automation/types/session.go @@ -103,15 +103,15 @@ func NewSession(s *wfexec.Session) *Session { } } -func (s Session) Exec(ctx context.Context, step wfexec.Step, input *expr.Vars) error { +func (s *Session) Exec(ctx context.Context, step wfexec.Step, input *expr.Vars) error { return s.session.Exec(ctx, step, input) } -func (s Session) Resume(ctx context.Context, stateID uint64, input *expr.Vars) (*wfexec.ResumedPrompt, error) { +func (s *Session) Resume(ctx context.Context, stateID uint64, input *expr.Vars) (*wfexec.ResumedPrompt, error) { return s.session.Resume(ctx, stateID, input) } -func (s Session) PendingPrompts(ownerId uint64) []*wfexec.PendingPrompt { +func (s *Session) PendingPrompts(ownerId uint64) []*wfexec.PendingPrompt { return s.session.UserPendingPrompts(ownerId) } @@ -155,6 +155,13 @@ func (s *Session) Apply(ssp SessionStartParams) { } } +func (s *Session) AppendRuntimeStacktrace(frame *wfexec.Frame) { + s.l.RLock() + defer s.l.RUnlock() + + s.RuntimeStacktrace = append(s.RuntimeStacktrace, frame) +} + func (s *Session) CopyRuntimeStacktrace() { s.l.RLock() defer s.l.RUnlock() diff --git a/pkg/wfexec/session.go b/pkg/wfexec/session.go index 85fe7f63e..53fd4eda1 100644 --- a/pkg/wfexec/session.go +++ b/pkg/wfexec/session.go @@ -326,15 +326,24 @@ func (s *Session) Resume(ctx context.Context, stateId uint64, input *expr.Vars) return p.toResumed(), nil } -func (s *Session) enqueue(ctx context.Context, st *State) error { +func (s *Session) canEnqueue(st *State) error { if st == nil { return fmt.Errorf("state is nil") } - if st.step == nil { + // when the step is completed right away, it is considered as special + if st.step == nil && st.completed == nil { return fmt.Errorf("state step is nil") } + return nil +} + +func (s *Session) enqueue(ctx context.Context, st *State) error { + if err := s.canEnqueue(st); err != nil { + return err + } + if st.stateId == 0 { st.stateId = nextID() } @@ -448,7 +457,18 @@ func (s *Session) worker(ctx context.Context) { s.execLock <- struct{}{} go func() { - err := s.exec(ctx, st) + defer func() { + // remove protection that prevents multiple + // steps executing at the same time + <-s.execLock + }() + + var ( + err error + log = s.log.With(zap.Uint64("stateID", st.stateId)) + ) + + nxt, err := s.exec(ctx, log, st) if err != nil && st.err == nil { // override the error from the execution st.err = err @@ -456,10 +476,6 @@ func (s *Session) worker(ctx context.Context) { st.completed = now() - // remove protection that prevents multiple - // steps executing at the same time - <-s.execLock - status := s.Status() if st.err != nil { st.err = fmt.Errorf( @@ -493,6 +509,18 @@ func (s *Session) worker(ctx context.Context) { ) s.eventHandler(status, st, s) + + for _, n := range nxt { + if n.step != nil { + log.Debug("next step queued", zap.Uint64("nextStepId", n.step.ID())) + } else { + log.Debug("next step queued", zap.Uint64("nextStepId", 0)) + } + if err = s.enqueue(ctx, n); err != nil { + log.Error("unable to enqueue", zap.Error(err)) + return + } + } }() case err := <-s.qErr: @@ -547,7 +575,7 @@ func (s *Session) queueScheduledSuspended() { } // executes single step, resolves response and schedule following steps for execution -func (s *Session) exec(ctx context.Context, st *State) (err error) { +func (s *Session) exec(ctx context.Context, log *zap.Logger, st *State) (nxt []*State, err error) { st.created = *now() defer func() { @@ -579,8 +607,6 @@ func (s *Session) exec(ctx context.Context, st *State) (err error) { scope = (&expr.Vars{}).MustMerge(st.scope) currLoop = st.loopCurr() - - log = s.log.With(zap.Uint64("stateID", st.stateId)) ) if st.step != nil { @@ -614,7 +640,7 @@ func (s *Session) exec(ctx context.Context, st *State) (err error) { if st.err != nil { if st.errHandler == nil { // no error handler set - return st.err + return nil, st.err } // handling error with error handling @@ -631,11 +657,7 @@ func (s *Session) exec(ctx context.Context, st *State) (err error) { eh := st.errHandler st.errHandler = nil st.errHandled = true - if err = s.enqueue(ctx, st.Next(eh, scope)); err != nil { - log.Warn("unable to queue", zap.Error(err)) - } - - return nil + return []*State{st.Next(eh, scope)}, nil } switch l := result.(type) { @@ -647,7 +669,7 @@ func (s *Session) exec(ctx context.Context, st *State) (err error) { ) n, result, st.err = l.Next(ctx, scope) if st.err != nil { - return st.err + return nil, st.err } if n == nil { @@ -683,7 +705,7 @@ func (s *Session) exec(ctx context.Context, st *State) (err error) { case *loopBreak: st.action = "loop break" if currLoop == nil { - return fmt.Errorf("break step not inside a loop") + return nil, fmt.Errorf("break step not inside a loop") } // jump out of the loop @@ -693,7 +715,7 @@ func (s *Session) exec(ctx context.Context, st *State) (err error) { case *loopContinue: st.action = "loop continue" if currLoop == nil { - return fmt.Errorf("continue step not inside a loop") + return nil, fmt.Errorf("continue step not inside a loop") } // jump back to iterator @@ -710,9 +732,10 @@ func (s *Session) exec(ctx context.Context, st *State) (err error) { st.action = "termination" // terminate all activities, all delayed tasks and exit right away log.Debug("termination", zap.Int("delayed", len(s.delayed))) + s.mux.Lock() s.delayed = nil - s.qState <- FinalState(s, scope) - return + s.mux.Unlock() + return []*State{FinalState(s, scope)}, nil case *delayed: st.action = "delayed" @@ -731,7 +754,7 @@ func (s *Session) exec(ctx context.Context, st *State) (err error) { case *prompted: st.action = "prompted" if result.ownerId == 0 { - return fmt.Errorf("without an owner") + return nil, fmt.Errorf("without an owner") } result.state = st @@ -753,7 +776,7 @@ func (s *Session) exec(ctx context.Context, st *State) (err error) { st.next = Steps{result} default: - return fmt.Errorf("unknown exec response type %T", result) + return nil, fmt.Errorf("unknown exec response type %T", result) } } @@ -767,7 +790,7 @@ func (s *Session) exec(ctx context.Context, st *State) (err error) { // do a quick sanity check cc := s.g.Children(st.step) if len(cc) > 0 && !cc.Contains(st.next...) { - return fmt.Errorf("inconsistent relationship") + return nil, fmt.Errorf("inconsistent relationship") } } @@ -781,19 +804,21 @@ func (s *Session) exec(ctx context.Context, st *State) (err error) { if len(st.next) == 0 { log.Debug("zero paths, finalizing") // using state to transport results and complete the worker loop - s.qState <- FinalState(s, scope) - return + return []*State{FinalState(s, scope)}, nil } - for _, step := range st.next { - log.Debug("next step queued", zap.Uint64("nextStepId", step.ID())) - if err = s.enqueue(ctx, st.Next(step, scope)); err != nil { + nxt = make([]*State, len(st.next)) + for i, step := range st.next { + nn := st.Next(step, scope) + if err = s.canEnqueue(nn); err != nil { log.Error("unable to queue", zap.Error(err)) return } + + nxt[i] = nn } - return + return nxt, nil } func SetWorkerInterval(i time.Duration) SessionOpt { diff --git a/tests/workflows/0010_stacktrace_test.go b/tests/workflows/0010_stacktrace_test.go new file mode 100644 index 000000000..ffe00ecef --- /dev/null +++ b/tests/workflows/0010_stacktrace_test.go @@ -0,0 +1,55 @@ +package workflows + +import ( + "context" + "fmt" + "testing" + + autTypes "github.com/cortezaproject/corteza-server/automation/types" + "github.com/stretchr/testify/require" +) + +func Test0010_stacktrace(t *testing.T) { + var ( + ctx = bypassRBAC(context.Background()) + req = require.New(t) + ) + + req.NoError(defStore.TruncateComposeRecords(ctx, nil)) + req.NoError(defStore.TruncateComposeModules(ctx)) + req.NoError(defStore.TruncateComposeNamespaces(ctx)) + + loadScenario(ctx, t) + + for rep := 0; rep < 11; rep++ { + t.Run(fmt.Sprintf("iteration %d", rep), func(t *testing.T) { + var ( + _, trace = mustExecWorkflow(ctx, t, "testing", autTypes.WorkflowExecParams{}) + ) + + // 6x iterator, 5x continue, 1x terminator, 1x completed + req.Len(trace, 13) + + steps := []uint64{ + 10, + 11, + 10, + 11, + 10, + 11, + 10, + 11, + 10, + 11, + + 10, + 12, + 0, + } + + for i := 0; i < 13; i++ { + req.Equal(steps[i], trace[i].StepID) + } + }) + } +} diff --git a/tests/workflows/0011_termination_explicit_test.go b/tests/workflows/0011_termination_explicit_test.go new file mode 100644 index 000000000..c87e2ca7a --- /dev/null +++ b/tests/workflows/0011_termination_explicit_test.go @@ -0,0 +1,27 @@ +package workflows + +import ( + "context" + "testing" + + autTypes "github.com/cortezaproject/corteza-server/automation/types" + "github.com/stretchr/testify/require" +) + +func Test0011_termination_explicit(t *testing.T) { + var ( + ctx = bypassRBAC(context.Background()) + req = require.New(t) + ) + + loadScenario(ctx, t) + + var ( + _, trace = mustExecWorkflow(ctx, t, "testing", autTypes.WorkflowExecParams{}) + ) + + req.Len(trace, 3) + req.Equal(uint64(10), trace[0].StepID) + req.Equal(uint64(11), trace[1].StepID) + req.Equal(uint64(0), trace[2].StepID) +} diff --git a/tests/workflows/0012_termination_implicit_test.go b/tests/workflows/0012_termination_implicit_test.go new file mode 100644 index 000000000..2e8b1ccc3 --- /dev/null +++ b/tests/workflows/0012_termination_implicit_test.go @@ -0,0 +1,26 @@ +package workflows + +import ( + "context" + "testing" + + autTypes "github.com/cortezaproject/corteza-server/automation/types" + "github.com/stretchr/testify/require" +) + +func Test0012_termination_implicit(t *testing.T) { + var ( + ctx = bypassRBAC(context.Background()) + req = require.New(t) + ) + + loadScenario(ctx, t) + + var ( + _, trace = mustExecWorkflow(ctx, t, "testing", autTypes.WorkflowExecParams{}) + ) + + req.Len(trace, 2) + req.Equal(uint64(10), trace[0].StepID) + req.Equal(uint64(0), trace[1].StepID) +} diff --git a/tests/workflows/0013_gateway_exclusive_test.go b/tests/workflows/0013_gateway_exclusive_test.go new file mode 100644 index 000000000..69a2941d5 --- /dev/null +++ b/tests/workflows/0013_gateway_exclusive_test.go @@ -0,0 +1,48 @@ +package workflows + +import ( + "context" + "testing" + + autTypes "github.com/cortezaproject/corteza-server/automation/types" + "github.com/stretchr/testify/require" +) + +func Test0013_gateway_exclusive(t *testing.T) { + var ( + ctx = bypassRBAC(context.Background()) + req = require.New(t) + ) + + loadScenario(ctx, t) + + t.Run("first path match", func(t *testing.T) { + _, trace := mustExecWorkflow(ctx, t, "case1", autTypes.WorkflowExecParams{}) + + req.Len(trace, 4) + req.Equal(uint64(10), trace[0].StepID) + req.Equal(uint64(11), trace[1].StepID) + req.Equal(uint64(14), trace[2].StepID) + req.Equal(uint64(0), trace[3].StepID) + }) + + t.Run("second path match", func(t *testing.T) { + _, trace := mustExecWorkflow(ctx, t, "case2", autTypes.WorkflowExecParams{}) + + req.Len(trace, 4) + req.Equal(uint64(10), trace[0].StepID) + req.Equal(uint64(12), trace[1].StepID) + req.Equal(uint64(14), trace[2].StepID) + req.Equal(uint64(0), trace[3].StepID) + }) + + t.Run("default path match", func(t *testing.T) { + _, trace := mustExecWorkflow(ctx, t, "case3", autTypes.WorkflowExecParams{}) + + req.Len(trace, 4) + req.Equal(uint64(10), trace[0].StepID) + req.Equal(uint64(13), trace[1].StepID) + req.Equal(uint64(14), trace[2].StepID) + req.Equal(uint64(0), trace[3].StepID) + }) +} diff --git a/tests/workflows/testdata/S0010_stacktrace/data_model.yaml b/tests/workflows/testdata/S0010_stacktrace/data_model.yaml new file mode 100644 index 000000000..013f3e77c --- /dev/null +++ b/tests/workflows/testdata/S0010_stacktrace/data_model.yaml @@ -0,0 +1,25 @@ +namespaces: + ns1: + name: ns1 name + +modules: + mod1: + name: mod1 name + fields: + f1: + label: f1 label + kind: String + required: false + +records: + mod1: + - values: + f1: 1 + - values: + f1: 2 + - values: + f1: 3 + - values: + f1: 4 + - values: + f1: 5 diff --git a/tests/workflows/testdata/S0010_stacktrace/workflow.yaml b/tests/workflows/testdata/S0010_stacktrace/workflow.yaml new file mode 100644 index 000000000..f12e6d2b8 --- /dev/null +++ b/tests/workflows/testdata/S0010_stacktrace/workflow.yaml @@ -0,0 +1,29 @@ +workflows: + testing: + enabled: true + trace: true + triggers: + - enabled: true + stepID: 10 + + steps: + - stepID: 10 + kind: iterator + ref: composeRecordsEach + arguments: + - { "target": "module", "value": "mod1", "type": "Handle" } + - { "target": "namespace", "value": "ns1", "type": "Handle" } + results: + - { "target": "r", "expr": "record" } + - { "target": "i", "expr": "index" } + - { "target": "t", "expr": "total" } + + - stepID: 11 + kind: continue + + - stepID: 12 + kind: termination + + paths: + - { parentID: 10, childID: 11 } + - { parentID: 10, childID: 12 } diff --git a/tests/workflows/testdata/S0011_termination_explicit/workflow.yaml b/tests/workflows/testdata/S0011_termination_explicit/workflow.yaml new file mode 100644 index 000000000..ba808c4f2 --- /dev/null +++ b/tests/workflows/testdata/S0011_termination_explicit/workflow.yaml @@ -0,0 +1,19 @@ +workflows: + testing: + enabled: true + trace: true + triggers: + - enabled: true + stepID: 10 + + steps: + - stepID: 10 + kind: expressions + arguments: + - { "target": "test", "value": "42", "type": "Any" } + + - stepID: 11 + kind: termination + + paths: + - { parentID: 10, childID: 11 } diff --git a/tests/workflows/testdata/S0012_termination_implicit/workflow.yaml b/tests/workflows/testdata/S0012_termination_implicit/workflow.yaml new file mode 100644 index 000000000..8deef75da --- /dev/null +++ b/tests/workflows/testdata/S0012_termination_implicit/workflow.yaml @@ -0,0 +1,13 @@ +workflows: + testing: + enabled: true + trace: true + triggers: + - enabled: true + stepID: 10 + + steps: + - stepID: 10 + kind: expressions + arguments: + - { "target": "test", "value": "42", "type": "Any" } diff --git a/tests/workflows/testdata/S0013_gateway_exclusive/workflow.yaml b/tests/workflows/testdata/S0013_gateway_exclusive/workflow.yaml new file mode 100644 index 000000000..a720b599e --- /dev/null +++ b/tests/workflows/testdata/S0013_gateway_exclusive/workflow.yaml @@ -0,0 +1,120 @@ +workflows: + case1: + enabled: true + trace: true + triggers: + - enabled: true + stepID: 10 + + steps: + - stepID: 10 + kind: gateway + ref: excl + + - stepID: 11 + kind: function + ref: logInfo + arguments: + - {target: "message", value: "log 1", type: "String"} + - stepID: 12 + kind: function + ref: logInfo + arguments: + - {target: "message", value: "log 2", type: "String"} + - stepID: 13 + kind: function + ref: logInfo + arguments: + - {target: "message", value: "log 3", type: "String"} + + - stepID: 14 + kind: termination + + paths: + - { parentID: 10, childID: 11, expr: "true == true" } + - { parentID: 10, childID: 12, expr: "true == true" } + - { parentID: 10, childID: 13, expr: "" } + + - { parentID: 11, childID: 14 } + - { parentID: 12, childID: 14 } + - { parentID: 13, childID: 14 } + + case2: + enabled: true + trace: true + triggers: + - enabled: true + stepID: 10 + + steps: + - stepID: 10 + kind: gateway + ref: excl + + - stepID: 11 + kind: function + ref: logInfo + arguments: + - {target: "message", value: "log 1", type: "String"} + - stepID: 12 + kind: function + ref: logInfo + arguments: + - {target: "message", value: "log 2", type: "String"} + - stepID: 13 + kind: function + ref: logInfo + arguments: + - {target: "message", value: "log 3", type: "String"} + + - stepID: 14 + kind: termination + + paths: + - { parentID: 10, childID: 11, expr: "true == false" } + - { parentID: 10, childID: 12, expr: "true == true" } + - { parentID: 10, childID: 13, expr: "" } + + - { parentID: 11, childID: 14 } + - { parentID: 12, childID: 14 } + - { parentID: 13, childID: 14 } + + case3: + enabled: true + trace: true + triggers: + - enabled: true + stepID: 10 + + steps: + - stepID: 10 + kind: gateway + ref: excl + + - stepID: 11 + kind: function + ref: logInfo + arguments: + - {target: "message", value: "log 1", type: "String"} + - stepID: 12 + kind: function + ref: logInfo + arguments: + - {target: "message", value: "log 2", type: "String"} + - stepID: 13 + kind: function + ref: logInfo + arguments: + - {target: "message", value: "log 3", type: "String"} + + - stepID: 14 + kind: termination + + paths: + - { parentID: 10, childID: 11, expr: "true == false" } + - { parentID: 10, childID: 12, expr: "true == false" } + - { parentID: 10, childID: 13, expr: "" } + + - { parentID: 11, childID: 14 } + - { parentID: 12, childID: 14 } + - { parentID: 13, childID: 14 } diff --git a/tests/workflows/testdata/S0014_gateway_fork/workflow.yaml b/tests/workflows/testdata/S0014_gateway_fork/workflow.yaml new file mode 100644 index 000000000..88f9f4d46 --- /dev/null +++ b/tests/workflows/testdata/S0014_gateway_fork/workflow.yaml @@ -0,0 +1,46 @@ +workflows: + case1: + enabled: true + trace: true + triggers: + - enabled: true + stepID: 10 + + steps: + - stepID: 10 + kind: gateway + ref: fork + + - stepID: 11 + kind: function + ref: logInfo + arguments: + - {target: "message", value: "log 1", type: "String"} + - stepID: 12 + kind: function + ref: logInfo + arguments: + - {target: "message", value: "log 2", type: "String"} + - stepID: 13 + kind: function + ref: logInfo + arguments: + - {target: "message", value: "log 3", type: "String"} + + - stepID: 14 + kind: gateway + ref: join + + - stepID: 15 + kind: termination + + paths: + - { parentID: 10, childID: 11 } + - { parentID: 10, childID: 12 } + - { parentID: 10, childID: 13 } + + - { parentID: 11, childID: 14 } + - { parentID: 12, childID: 14 } + - { parentID: 13, childID: 14 } + + - { parentID: 14, childID: 15 }