diff --git a/automation/service/session.go b/automation/service/session.go index fd82d0953..151b71f35 100644 --- a/automation/service/session.go +++ b/automation/service/session.go @@ -39,6 +39,7 @@ type ( session chan *wfexec.Session graph *wfexec.Graph trace bool + callStack []uint64 } sessionAccessController interface { @@ -182,7 +183,7 @@ func (svc *session) Start(g *wfexec.Graph, i auth.Identifiable, ssp types.Sessio var ( ctx = auth.SetIdentityToContext(context.Background(), i) - ses = svc.spawn(g, ssp.WorkflowID, ssp.Trace) + ses = svc.spawn(g, ssp.WorkflowID, ssp.Trace, ssp.CallStack) ) ses.CreatedAt = *now() @@ -230,12 +231,13 @@ func (svc *session) Resume(sessionID, stateID uint64, i auth.Identifiable, input // // We need initial context for the session because we want to catch all cancellations or timeouts from there // and not from any potential HTTP requests or similar temporary context that can prematurely destroy a workflow session -func (svc *session) spawn(g *wfexec.Graph, workflowID uint64, trace bool) (ses *types.Session) { +func (svc *session) spawn(g *wfexec.Graph, workflowID uint64, trace bool, callStack []uint64) (ses *types.Session) { s := &spawn{ workflowID: workflowID, session: make(chan *wfexec.Session, 1), graph: g, trace: trace, + callStack: callStack, } // Send new-session request @@ -266,6 +268,7 @@ func (svc *session) Watch(ctx context.Context) { case s := <-svc.spawnQueue: opts := []wfexec.SessionOpt{ wfexec.SetWorkflowID(s.workflowID), + wfexec.SetCallStack(s.callStack...), wfexec.SetHandler(svc.stateChangeHandler(ctx)), } diff --git a/automation/service/trigger.go b/automation/service/trigger.go index 2535ddf60..a099dc48e 100644 --- a/automation/service/trigger.go +++ b/automation/service/trigger.go @@ -414,7 +414,7 @@ func (svc trigger) canManageTrigger(ctx context.Context, res *types.Trigger, per // registers all triggers on all given workflows // before registering triggers on a workflow, all workflow triggers are unregistered func (svc *trigger) registerWorkflows(ctx context.Context, workflows ...*types.Workflow) error { - // load ALL workflows directly from store + // load ALL triggers directly from store tt, _, err := store.SearchAutomationTriggers(ctx, svc.store, types.TriggerFilter{ WorkflowID: types.WorkflowSet(workflows).IDs(), Deleted: filter.StateInclusive, diff --git a/automation/service/workflow.go b/automation/service/workflow.go index 21c3edec2..772df7111 100644 --- a/automation/service/workflow.go +++ b/automation/service/workflow.go @@ -32,6 +32,8 @@ type ( triggers *trigger session *session + opt options.WorkflowOpt + log *zap.Logger // maps resolved workflow graphs to workflow ID (key, uint64) @@ -79,9 +81,10 @@ const ( workflowDefChanged workflowChanges = 4 ) -func Workflow(log *zap.Logger, corredorOpt options.CorredorOpt) *workflow { +func Workflow(log *zap.Logger, corredorOpt options.CorredorOpt, opt options.WorkflowOpt) *workflow { return &workflow{ log: log, + opt: opt, actionlog: DefaultActionlog, store: DefaultStore, ac: DefaultAccessControl, @@ -552,11 +555,18 @@ func (svc *workflow) Exec(ctx context.Context, workflowID uint64, p types.Workfl // Start with workflow scope scope := wf.Scope.MustMerge() + callStack := wfexec.GetContextCallStack(ctx) + if len(callStack) > svc.opt.CallStackSize { + return WorkflowErrMaximumCallStackSizeExceeded() + } + ssp := types.SessionStartParams{ WorkflowID: wf.ID, KeepFor: wf.KeepSessions, Trace: wf.Trace || p.Trace, StepID: p.StepID, + + CallStack: callStack, } if !p.Trace { @@ -697,6 +707,11 @@ func makeWorkflowHandler(ac workflowExecController, s *session, t *types.Trigger return WorkflowErrNotAllowedToExecute() } + callStack := wfexec.GetContextCallStack(ctx) + if len(callStack) > s.opt.CallStackSize { + return WorkflowErrMaximumCallStackSizeExceeded() + } + wait, err = s.Start(g, runAs, types.SessionStartParams{ WorkflowID: wf.ID, KeepFor: wf.KeepSessions, @@ -705,6 +720,8 @@ func makeWorkflowHandler(ac workflowExecController, s *session, t *types.Trigger StepID: t.StepID, EventType: t.EventType, ResourceType: t.ResourceType, + + CallStack: callStack, }) if err != nil { diff --git a/automation/service/workflow_actions.gen.go b/automation/service/workflow_actions.gen.go index 9cddb67cd..7110ebda1 100644 --- a/automation/service/workflow_actions.gen.go +++ b/automation/service/workflow_actions.gen.go @@ -979,6 +979,42 @@ func WorkflowErrNotAllowedToExecuteCorredorStep(mm ...*workflowActionProps) *err return e } +// WorkflowErrMaximumCallStackSizeExceeded returns "automation:workflow.maximumCallStackSizeExceeded" as *errors.Error +// +// +// This function is auto-generated. +// +func WorkflowErrMaximumCallStackSizeExceeded(mm ...*workflowActionProps) *errors.Error { + var p = &workflowActionProps{} + if len(mm) > 0 { + p = mm[0] + } + + var e = errors.New( + errors.KindInternal, + + p.Format("maximum call stack size exceeded", nil), + + errors.Meta("type", "maximumCallStackSizeExceeded"), + errors.Meta("resource", "automation:workflow"), + + // action log entry; no formatting, it will be applied inside recordAction fn. + errors.Meta(workflowLogMetaKey{}, "maximum call stack size exceeded"), + errors.Meta(workflowPropsMetaKey{}, p), + + // translation namespace & key + errors.Meta(locale.ErrorMetaNamespace{}, "automation"), + errors.Meta(locale.ErrorMetaKey{}, "workflow.errors.maximumCallStackSizeExceeded"), + + errors.StackSkip(1), + ) + + if len(mm) > 0 { + } + + return e +} + // ********************************************************************************************************************* // ********************************************************************************************************************* diff --git a/automation/service/workflow_actions.yaml b/automation/service/workflow_actions.yaml index 52ff73016..36b822774 100644 --- a/automation/service/workflow_actions.yaml +++ b/automation/service/workflow_actions.yaml @@ -111,3 +111,7 @@ errors: - error: notAllowedToExecuteCorredorStep message: "not allowed to run corredorExec function, corredor is disabled" log: "failed to execute {{workflow}} with corredorExec function step; corredor is disabled" + + - error: maximumCallStackSizeExceeded + message: "maximum call stack size exceeded" + log: "maximum call stack size exceeded" diff --git a/automation/types/session.go b/automation/types/session.go index 411ad52b7..b2ee650a6 100644 --- a/automation/types/session.go +++ b/automation/types/session.go @@ -60,6 +60,8 @@ type ( StepID uint64 EventType string ResourceType string + + CallStack []uint64 } SessionFilter struct { diff --git a/pkg/options/workflow.gen.go b/pkg/options/workflow.gen.go index 2d7081c4b..a19d9c707 100644 --- a/pkg/options/workflow.gen.go +++ b/pkg/options/workflow.gen.go @@ -10,16 +10,18 @@ package options type ( WorkflowOpt struct { - Register bool `env:"WORKFLOW_REGISTER"` - ExecDebug bool `env:"WORKFLOW_EXEC_DEBUG"` + Register bool `env:"WORKFLOW_REGISTER"` + ExecDebug bool `env:"WORKFLOW_EXEC_DEBUG"` + CallStackSize int `env:"WORKFLOW_CALL_STACK_SIZE"` } ) // Workflow initializes and returns a WorkflowOpt with default values func Workflow() (o *WorkflowOpt) { o = &WorkflowOpt{ - Register: true, - ExecDebug: false, + Register: true, + ExecDebug: false, + CallStackSize: 16, } fill(o) diff --git a/pkg/options/workflow.yaml b/pkg/options/workflow.yaml index ce1e32c94..3d07a89d5 100644 --- a/pkg/options/workflow.yaml +++ b/pkg/options/workflow.yaml @@ -11,3 +11,8 @@ props: type: bool default: false description: Enables verbose logging for workflow execution + + - name: callStackSize + type: int + default: 16 + description: Defines the maximum call stack size between workflows diff --git a/pkg/wfexec/session.go b/pkg/wfexec/session.go index 53fd4eda1..23b3a2c72 100644 --- a/pkg/wfexec/session.go +++ b/pkg/wfexec/session.go @@ -61,6 +61,8 @@ type ( dumpStacktraceOnPanic bool eventHandler StateChangeHandler + + callStack []uint64 } StateChangeHandler func(SessionStatus, *State, *Session) @@ -106,6 +108,8 @@ type ( } SessionStatus int + + callStackCtxKey struct{} ) const ( @@ -180,6 +184,8 @@ func NewSession(ctx context.Context, g *Graph, oo ...SessionOpt) *Session { s.log = s.log. With(zap.Uint64("sessionID", s.id)) + s.callStack = append(s.callStack, s.id) + go s.worker(ctx) return s @@ -624,9 +630,10 @@ func (s *Session) exec(ctx context.Context, log *zap.Logger, st *State) (nxt []* // Context received in exec() wil not have the identity we're expecting // so we need to pull it from state owner and add it to new context // that is set to step exec function - ctxWithIdentity := auth.SetIdentityToContext(ctx, st.owner) + stepCtx := auth.SetIdentityToContext(ctx, st.owner) + stepCtx = SetContextCallStack(stepCtx, s.callStack) - result, st.err = st.step.Exec(ctxWithIdentity, st.MakeRequest()) + result, st.err = st.step.Exec(stepCtx, st.MakeRequest()) if iterator, isIterator := result.(Iterator); isIterator && st.err == nil { // Exec fn returned an iterator, adding loop to stack @@ -851,6 +858,12 @@ func SetDumpStacktraceOnPanic(dump bool) SessionOpt { } } +func SetCallStack(id ...uint64) SessionOpt { + return func(s *Session) { + s.callStack = id + } +} + func (ss Steps) hash() map[Step]bool { out := make(map[Step]bool) for _, s := range ss { @@ -883,3 +896,16 @@ func (ss Steps) IDs() []uint64 { return ids } + +func SetContextCallStack(ctx context.Context, ss []uint64) context.Context { + return context.WithValue(ctx, callStackCtxKey{}, ss) +} + +func GetContextCallStack(ctx context.Context) []uint64 { + v := ctx.Value(callStackCtxKey{}) + if v == nil { + return nil + } + + return v.([]uint64) +}