diff --git a/server/automation/service/workflow_converter.go b/server/automation/service/workflow_converter.go index ccfc5a18a..2cf8cb624 100644 --- a/server/automation/service/workflow_converter.go +++ b/server/automation/service/workflow_converter.go @@ -3,8 +3,6 @@ package service import ( "context" "fmt" - "strings" - "github.com/cortezaproject/corteza/server/automation/types" "github.com/cortezaproject/corteza/server/pkg/auth" "github.com/cortezaproject/corteza/server/pkg/errors" @@ -12,6 +10,7 @@ import ( "github.com/cortezaproject/corteza/server/pkg/wfexec" "github.com/cortezaproject/corteza/server/system/automation" "go.uber.org/zap" + "strings" ) type ( @@ -200,7 +199,7 @@ func (svc workflowConverter) workflowStepDefConv(g *wfexec.Graph, def *types.Wor return svc.convDelayStep(s) case types.WorkflowStepKindErrHandler: - return svc.convErrorHandlerStep(g, out) + return svc.convErrorHandlerStep(g, s, out) case types.WorkflowStepKindBreak: return svc.convBreakStep() @@ -299,13 +298,25 @@ func (svc workflowConverter) convGateway(g *wfexec.Graph, s *types.WorkflowStep, return nil, errors.Internal("unexpected workflow configuration") } -func (svc workflowConverter) convErrorHandlerStep(g *wfexec.Graph, out []*types.WorkflowPath) (wfexec.Step, error) { +func (svc workflowConverter) convErrorHandlerStep(g *wfexec.Graph, s *types.WorkflowStep, out []*types.WorkflowPath) (wfexec.Step, error) { + var ( + results = &expr.Vars{} + ) + + // evaluate results and set target path for each expression + for _, r := range s.Results { + err := results.Set(r.Expr, r.Target) + if err != nil { + return nil, err + } + } + switch len(out) { case 0: return nil, fmt.Errorf("expecting at least one path out of error handling step") case 1: // remove error handler - return types.ErrorHandlerStep(nil), nil + return types.ErrorHandlerStep(nil, results), nil case 2: errorHandler := g.StepByID(out[1].ChildID) if errorHandler == nil { @@ -313,7 +324,7 @@ func (svc workflowConverter) convErrorHandlerStep(g *wfexec.Graph, out []*types. return nil, nil } - return types.ErrorHandlerStep(errorHandler), nil + return types.ErrorHandlerStep(errorHandler, results), nil default: // this might be extended in the future to allow different paths using expression @@ -515,8 +526,6 @@ 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 @@ -671,9 +680,9 @@ func verifyStep(s *types.WorkflowStep, in, out types.WorkflowPathSet) types.Work // check if reference is set on the step numericRef = func() error { // @todo SUBWF enable this back - //if 0 == cast.ToUint64(s.Ref) { + // if 0 == cast.ToUint64(s.Ref) { // return errors.Internal("%s step expects workflow reference ID", s.Kind) - //} + // } return nil } @@ -751,7 +760,6 @@ func verifyStep(s *types.WorkflowStep, in, out types.WorkflowPathSet) types.Work checks = append(checks, noRef, zero(arguments), - zero(results), count(1, 2, outbound), ) diff --git a/server/automation/types/error_handler.go b/server/automation/types/error_handler.go index d992ae5e7..8e1164aea 100644 --- a/server/automation/types/error_handler.go +++ b/server/automation/types/error_handler.go @@ -2,6 +2,7 @@ package types import ( "context" + "github.com/cortezaproject/corteza/server/pkg/expr" "github.com/cortezaproject/corteza/server/pkg/wfexec" ) @@ -9,14 +10,15 @@ type ( errorHandlerStep struct { wfexec.StepIdentifier handler wfexec.Step + results *expr.Vars } ) -func ErrorHandlerStep(h wfexec.Step) *errorHandlerStep { - return &errorHandlerStep{handler: h} +func ErrorHandlerStep(h wfexec.Step, rr *expr.Vars) *errorHandlerStep { + return &errorHandlerStep{handler: h, results: rr} } -// Executes prompt step -func (h errorHandlerStep) Exec(_ context.Context, _ *wfexec.ExecRequest) (wfexec.ExecResponse, error) { - return wfexec.ErrorHandler(h.handler), nil +// Exec errorHandler step +func (h errorHandlerStep) Exec(ctx context.Context, r *wfexec.ExecRequest) (wfexec.ExecResponse, error) { + return wfexec.ErrorHandler(h.handler, h.results), nil } diff --git a/server/pkg/wfexec/response.go b/server/pkg/wfexec/response.go index ca2ff548b..c5a16b34a 100644 --- a/server/pkg/wfexec/response.go +++ b/server/pkg/wfexec/response.go @@ -1,6 +1,7 @@ package wfexec import ( + "github.com/cortezaproject/corteza/server/pkg/expr" "time" ) @@ -10,6 +11,7 @@ type ( errHandler struct { handler Step + results *expr.Vars } termination struct{} @@ -35,8 +37,8 @@ func Resume() *resumed { return &resumed{} } -func ErrorHandler(h Step) *errHandler { - return &errHandler{handler: h} +func ErrorHandler(h Step, results *expr.Vars) *errHandler { + return &errHandler{handler: h, results: results} } func Termination() *termination { diff --git a/server/pkg/wfexec/session.go b/server/pkg/wfexec/session.go index ec18df5e2..a42553f51 100644 --- a/server/pkg/wfexec/session.go +++ b/server/pkg/wfexec/session.go @@ -171,7 +171,7 @@ func NewSession(ctx context.Context, g *Graph, oo ...SessionOpt) *Session { delayed: make(map[uint64]*delayed), prompted: make(map[uint64]*prompted), - //workerInterval: time.Millisecond, + // workerInterval: time.Millisecond, workerInterval: time.Millisecond * 250, // debug mode rate workerLock: make(chan struct{}, 1), @@ -395,15 +395,14 @@ func (s *Session) enqueue(ctx context.Context, st *State) error { } // Wait does not wait for the whole wf to be complete but until: -// - context timeout -// - idle state -// - error in error queue +// - context timeout +// - idle state +// - error in error queue func (s *Session) Wait(ctx context.Context) error { return s.WaitUntil(ctx, SessionFailed, SessionDelayed, SessionCompleted) } // WaitUntil blocks until workflow session gets into expected status -// func (s *Session) WaitUntil(ctx context.Context, expected ...SessionStatus) error { indexed := make(map[SessionStatus]bool) for _, status := range expected { @@ -689,7 +688,10 @@ func (s *Session) exec(ctx context.Context, log *zap.Logger, st *State) (nxt []* zap.Error(st.err), ) - _ = expr.Assign(scope, "error", expr.Must(expr.NewString(st.err.Error()))) + err = setErrorHandlerResultsToScope(scope, st.results, st.err, st.step.ID()) + if err != nil { + return nil, err + } // copy error handler & disable it on state to prevent inf. loop // in case of another error in the error-handling branch @@ -731,6 +733,7 @@ func (s *Session) exec(ctx context.Context, log *zap.Logger, st *State) (nxt []* // this step sets error handling step on current state // and continues on the current path st.errHandler = result.handler + st.results = st.results.MustMerge(result.results) // find step that's not error handler and // use it for the next step @@ -941,3 +944,30 @@ func GetContextCallStack(ctx context.Context) []uint64 { return v.([]uint64) } + +func setErrorHandlerResultsToScope(scope *expr.Vars, result *expr.Vars, e error, stepID uint64) (err error) { + var ( + ehr = struct { + Error string `json:"error"` + ErrorMessage string `json:"errorMessage"` + ErrorStepID string `json:"errorStepID"` + }{} + ) + + err = result.Decode(&ehr) + if err != nil { + return + } + + if len(ehr.Error) > 0 { + _ = expr.Assign(scope, ehr.Error, expr.Must(expr.NewAny(e))) + } + if len(ehr.ErrorMessage) > 0 { + _ = expr.Assign(scope, ehr.ErrorMessage, expr.Must(expr.NewString(e.Error()))) + } + if len(ehr.ErrorStepID) > 0 { + _ = expr.Assign(scope, ehr.ErrorStepID, expr.Must(expr.NewInteger(stepID))) + } + + return +} diff --git a/server/pkg/wfexec/session_test.go b/server/pkg/wfexec/session_test.go index 9f3705c9b..426318521 100644 --- a/server/pkg/wfexec/session_test.go +++ b/server/pkg/wfexec/session_test.go @@ -203,14 +203,14 @@ func TestSession_ErrHandler(t *testing.T) { wf, // enable if you need to see what is going on - //SetLogger(logger.MakeDebugLogger()), + // SetLogger(logger.MakeDebugLogger()), // enable if you need to see what is going on - //SetHandler(func(status SessionStatus, state *State, session *Session) { + // SetHandler(func(status SessionStatus, state *State, session *Session) { // if state.step != nil { // println(state.step.(*sesTestStep).name) // } - //}), + // }), ) cb_1_1 = &sesTestStep{name: "catch-branch-1-1"} @@ -218,7 +218,7 @@ func TestSession_ErrHandler(t *testing.T) { tb_1_1 = &sesTestStep{name: "try-branch-1-1"} eh_1 = &sesTestStep{name: "err-handler", exec: func(ctx context.Context, request *ExecRequest) (ExecResponse, error) { - return ErrorHandler(cb_1_1), nil + return ErrorHandler(cb_1_1, nil), nil }} er_1 = &sesTestStep{name: "err-raiser", exec: func(ctx context.Context, request *ExecRequest) (ExecResponse, error) { return nil, fmt.Errorf("would-be-handled-error") @@ -229,7 +229,7 @@ func TestSession_ErrHandler(t *testing.T) { tb_2_1 = &sesTestStep{name: "try-branch-2-1"} eh_2 = &sesTestStep{name: "err-handler", exec: func(ctx context.Context, request *ExecRequest) (ExecResponse, error) { - return ErrorHandler(cb_2_1), nil + return ErrorHandler(cb_2_1, nil), nil }} er_2 = &sesTestStep{name: "err-raiser", exec: func(ctx context.Context, request *ExecRequest) (ExecResponse, error) { return nil, fmt.Errorf("would-be-handled-error") diff --git a/server/pkg/wfexec/state.go b/server/pkg/wfexec/state.go index 41c94cf9f..10ed2aad8 100644 --- a/server/pkg/wfexec/state.go +++ b/server/pkg/wfexec/state.go @@ -88,6 +88,7 @@ func (s State) Next(current Step, scope *expr.Vars) *State { sessionId: s.sessionId, parent: s.step, errHandler: s.errHandler, + results: s.results, loops: s.loops, step: current, diff --git a/server/tests/workflows/0018_error_handler_step_test.go b/server/tests/workflows/0018_error_handler_step_test.go new file mode 100644 index 000000000..a15230e19 --- /dev/null +++ b/server/tests/workflows/0018_error_handler_step_test.go @@ -0,0 +1,50 @@ +package workflows + +import ( + "context" + "github.com/cortezaproject/corteza/server/pkg/errors" + "github.com/cortezaproject/corteza/server/pkg/expr" + "github.com/stretchr/testify/require" + "testing" + + "github.com/cortezaproject/corteza/server/automation/types" +) + +func Test0018_error_handler_step(t *testing.T) { + var ( + ctx = bypassRBAC(context.Background()) + req = require.New(t) + ) + + loadScenario(ctx, t) + + t.Run("error handler with result", func(t *testing.T) { + type ( + testInput struct { + E expr.TypedValue + EM expr.TypedValue + EsID expr.TypedValue + } + ) + + var ( + aux = testInput{} + vars, _ = mustExecWorkflow(ctx, t, "error_handler_results", types.WorkflowExecParams{}) + expected = testInput{ + E: expr.Must(expr.NewAny(errors.New(errors.KindAutomation, "TestingError"))), + EM: expr.Must(expr.NewString("TestingError")), + EsID: expr.Must(expr.NewInteger(2)), + } + ) + + req.NoError(vars.Decode(&aux)) + + errA := expr.Must(expr.NewAny(aux.E.Get())) + req.Error(errors.Unwrap(errA.Get().(*errors.Error))) + req.Equal(errors.Unwrap(errA.Get().(*errors.Error)).Error(), "TestingError") + req.True(errors.IsKind(errA.Get().(*errors.Error), errors.KindAutomation)) + + req.Equal(expected.EM, aux.EM) + req.Equal(expected.EsID, aux.EsID) + }) +} diff --git a/server/tests/workflows/testdata/S0018_error_handler_step/workflow.yaml b/server/tests/workflows/testdata/S0018_error_handler_step/workflow.yaml new file mode 100644 index 000000000..205583e60 --- /dev/null +++ b/server/tests/workflows/testdata/S0018_error_handler_step/workflow.yaml @@ -0,0 +1,31 @@ +workflows: + error_handler_results: + enabled: true + trace: true + triggers: + - enabled: true + stepID: 1 + + steps: + - stepID: 1 + kind: error-handler + results: + - { target: e, type: Any, expr: "error" } + - { target: eM, type: String, expr: "errorMessage" } + - { target: esID, type: Integer, expr: "errorStepID" } + + - stepID: 2 + kind: error + arguments: + - { target: message, type: String, value: "TestingError" } + + - stepID: 3 + kind: function + ref: logInfo + arguments: + - { target: message, type: String, expr: "format(\"%s, %s, %d\", e, eM, esID)" } + + paths: + - { parentID: 1, childID: 2 } + - { parentID: 1, childID: 3 } +