DeforaNetworks/corteza
3
0
Fork 0
Code Activity

Refactored and cleanedup workflow exection

Browse Source
This commit is contained in:
Denis Arh
2021-11-11 15:01:44 +01:00
parent 2d9e62ae27
commit 922243605d
9 changed files with 341 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files
automation/service/session.go
+50 -7
View File
@@ -38,6 +38,8 @@ type (
workflowID uint64
session chan *wfexec.Session
graph *wfexec.Graph
invoker auth.Identifiable
runner auth.Identifiable
trace bool
callStack []uint64
}
@@ -169,14 +171,36 @@ func (svc *session) PendingPrompts(ctx context.Context) (pp []*wfexec.PendingPro
//
// Start is an asynchronous operation
//
// Please note that context passed to the function is NOT the the one that is
// 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(g *wfexec.Graph, i auth.Identifiable, ssp types.SessionStartParams) (wait WaitFn, err error) {
func (svc *session) Start(ctx context.Context, g *wfexec.Graph, ssp types.SessionStartParams) (wait WaitFn, err error) {
var (
start wfexec.Step
)
if g == nil {
return nil, errors.InvalidData("cannot start workflow, uninitialized graph")
}
if len(ssp.CallStack) > svc.opt.CallStackSize {
return nil, WorkflowErrMaximumCallStackSizeExceeded()
}
ssp.CallStack = append(ssp.CallStack, ssp.WorkflowID)
if ssp.Invoker == nil {
return nil, errors.InvalidData("cannot start workflow without user")
}
if ssp.Runner == nil {
ssp.Runner = ssp.Invoker
}
if ssp.StepID == 0 {
// starting step is not explicitly workflows on trigger, find orphan step
// starting step is not explicitly set
// find orphan step
switch oo := g.Orphans(); len(oo) {
case 1:
start = oo[0]
@@ -192,15 +216,19 @@ 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, ssp.CallStack)
ses = svc.spawn(g, ssp.WorkflowID, ssp.Trace, ssp.CallStack, ssp.Runner, ssp.Invoker)
)
ses.CreatedAt = *now()
ses.CreatedBy = i.Identity()
ses.CreatedBy = ssp.Invoker.Identity()
ses.Status = types.SessionStarted
ses.Apply(ssp)
_ = ssp.Input.AssignFieldValue("eventType", expr.Must(expr.NewString(ssp.EventType)))
_ = ssp.Input.AssignFieldValue("resourceType", expr.Must(expr.NewString(ssp.ResourceType)))
_ = ssp.Input.AssignFieldValue("invoker", expr.Must(expr.NewAny(ssp.Invoker)))
_ = ssp.Input.AssignFieldValue("runner", expr.Must(expr.NewAny(ssp.Runner)))
if err = ses.Exec(ctx, start, ssp.Input); err != nil {
return
}
@@ -241,13 +269,15 @@ 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, callStack []uint64) (ses *types.Session) {
func (svc *session) spawn(g *wfexec.Graph, workflowID uint64, trace bool, callStack []uint64, invoker, runner auth.Identifiable) (ses *types.Session) {
s := &spawn{
workflowID: workflowID,
session: make(chan *wfexec.Session, 1),
graph: g,
trace: trace,
callStack: callStack,
invoker: invoker,
runner: runner,
}
// Send new-session request
@@ -262,6 +292,7 @@ func (svc *session) spawn(g *wfexec.Graph, workflowID uint64, trace bool, callSt
return ses
}
// Watch looks over session's spawn queue
func (svc *session) Watch(ctx context.Context) {
gcTicker := time.NewTicker(time.Second)
lpTicker := time.NewTicker(time.Second * 30)
@@ -276,6 +307,8 @@ func (svc *session) Watch(ctx context.Context) {
case <-ctx.Done():
return
case s := <-svc.spawnQueue:
var execCtx = context.Background()
opts := []wfexec.SessionOpt{
wfexec.SetWorkflowID(s.workflowID),
wfexec.SetCallStack(s.callStack...),
@@ -290,7 +323,15 @@ func (svc *session) Watch(ctx context.Context) {
)
}
s.session <- wfexec.NewSession(ctx, s.graph, opts...)
// Encode runner into execution context
// runner is used as identity and for access control
execCtx = auth.SetIdentityToContext(execCtx, s.runner)
// Encode invoker into execution context
// invoker is used
execCtx = context.WithValue(execCtx, workflowInvokerCtxKey{}, s.invoker)
s.session <- wfexec.NewSession(execCtx, s.graph, opts...)
// case time for a pool cleanup
// @todo cleanup pool when sessions are complete
@@ -370,6 +411,8 @@ func (svc *session) stateChangeHandler(ctx context.Context) wfexec.StateChangeHa
return
}
log = log.With(zap.Uint64("workflowID", ses.WorkflowID))
var (
// By default, we want to update session when new status is prompted, delayed, completed or failed
// But if status is active, we'll flush it every X frames (sessionStateFlushFrequency)
automation/service/session_test.go
+8 -4
View File
@@ -1,9 +1,11 @@
package service
import (
"context"
"testing"
"github.com/cortezaproject/corteza-server/automation/types"
"github.com/cortezaproject/corteza-server/pkg/auth"
"github.com/cortezaproject/corteza-server/pkg/wfexec"
"github.com/stretchr/testify/require"
)
@@ -13,19 +15,21 @@ func TestSession_Start(t *testing.T) {
req = require.New(t)
ses = &session{}
g = wfexec.NewGraph()
ctx = context.Background()
err error
)
_, err = ses.Start(g, nil, types.SessionStartParams{})
_, 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(g, nil, types.SessionStartParams{StepID: 4321})
_, 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(g, nil, types.SessionStartParams{})
_, 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
@@ -34,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(g, nil, types.SessionStartParams{StepID: 42})
_, 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/trigger.go
+6 -2
View File
@@ -528,11 +528,15 @@ func (svc *trigger) registerTriggers(wf *types.Workflow, runAs auth.Identifiable
registerWorkflow = wf.Enabled && wf.DeletedAt == nil
)
// convert only registerable and issuless workflwos
// convert only registrable and workflows without issues
if registerWorkflow && len(wf.Issues) == 0 {
// Convert workflow only when valid (no issues, enable, not delete)
if g, issues = Convert(svc.workflow, wf); len(issues) > 0 {
wfLog.Error("failed to convert workflow to graph", zap.Error(issues))
_ = issues.Walk(func(i *types.WorkflowIssue) error {
wfLog.Debug("workflow issue found: "+i.Description, zap.Any("culprit", i.Culprit))
return nil
})
g = nil
}
}
@@ -577,7 +581,7 @@ func (svc *trigger) registerTriggers(wf *types.Workflow, runAs auth.Identifiable
).Wrap(wf.Issues)
}
} else {
handlerFn = makeWorkflowHandler(svc.ac, svc.session, t, wf, g, runAs)
handlerFn = makeWorkflowHandler(svc.workflow, wf, t)
}
ops = append(
automation/service/workflow.go
+184 -171
View File
@@ -2,6 +2,7 @@ package service
import (
"context"
"fmt"
"reflect"
"sync"
@@ -18,8 +19,6 @@ import (
"github.com/cortezaproject/corteza-server/pkg/rbac"
"github.com/cortezaproject/corteza-server/pkg/wfexec"
"github.com/cortezaproject/corteza-server/store"
sysAutoTypes "github.com/cortezaproject/corteza-server/system/automation"
sysTypes "github.com/cortezaproject/corteza-server/system/types"
"go.uber.org/zap"
)
@@ -36,8 +35,11 @@ type (
log *zap.Logger
// maps resolved workflow graphs to workflow ID (key, uint64)
wfgs map[uint64]*wfexec.Graph
// cache of workflows, graphs to workflow ID (key, uint64)
cache map[uint64]*wfCacheItem
// handle to workflow index
wIndex map[string]uint64
// workflow function registry
reg *registry
@@ -47,6 +49,16 @@ type (
parser expr.Parsable
}
wfCacheItem struct {
wf *types.Workflow
// caching exec graph
g *wfexec.Graph
// caching user we'll executing workflow with
runAs intAuth.Identifiable
}
workflowAccessController interface {
CanSearchWorkflows(context.Context) bool
CanCreateWorkflow(context.Context) bool
@@ -72,6 +84,8 @@ type (
workflowUpdateHandler func(ctx context.Context, ns *types.Workflow) (workflowChanges, error)
workflowChanges uint8
workflowInvokerCtxKey struct{}
)
const (
@@ -91,7 +105,8 @@ func Workflow(log *zap.Logger, corredorOpt options.CorredorOpt, opt options.Work
triggers: DefaultTrigger,
session: DefaultSession,
eventbus: eventbus.Service(),
wfgs: make(map[uint64]*wfexec.Graph),
cache: make(map[uint64]*wfCacheItem),
wIndex: make(map[string]uint64),
mux: &sync.RWMutex{},
parser: expr.NewParser(),
reg: Registry(),
@@ -180,6 +195,8 @@ func (svc *workflow) Create(ctx context.Context, new *types.Workflow) (wf *types
var (
wap = &workflowActionProps{new: new}
cUser = intAuth.GetIdentityFromContext(ctx).Identity()
g *wfexec.Graph
runAs intAuth.Identifiable
)
err = store.Tx(ctx, svc.store, func(ctx context.Context, s store.Storer) (err error) {
@@ -215,9 +232,15 @@ func (svc *workflow) Create(ctx context.Context, new *types.Workflow) (wf *types
CreatedBy: cUser,
}
if wf.Issues, err = svc.validateWorkflow(ctx, wf); err != nil {
if g, runAs, err = svc.validateWorkflow(ctx, wf); err != nil {
return
} else if len(wf.Issues) == 0 {
}
if err = svc.updateCache(wf, runAs, g); err != nil {
return
}
if len(wf.Issues) == 0 {
if err = svc.triggers.registerWorkflows(ctx, wf); err != nil {
return err
}
@@ -256,34 +279,20 @@ func (svc *workflow) DeleteByID(ctx context.Context, workflowID uint64) error {
return workflowUnchanged, err
}
if res.Issues, err = svc.validateWorkflow(ctx, res); err != nil {
return workflowUnchanged, err
} else if len(res.Issues) == 0 {
if err := svc.triggers.registerWorkflows(ctx, res); err != nil {
return workflowUnchanged, err
}
}
return changes, err
}))
}
func (svc *workflow) UndeleteByID(ctx context.Context, workflowID uint64) error {
return trim1st(svc.updater(ctx, workflowID, WorkflowActionUndelete, func(ctx context.Context, res *types.Workflow) (workflowChanges, error) {
changes, err := svc.handleUndelete(ctx, res)
var (
changes, err = svc.handleUndelete(ctx, res)
)
if err != nil {
return workflowUnchanged, err
}
if res.Issues, err = svc.validateWorkflow(ctx, res); err != nil {
return workflowUnchanged, err
} else if len(res.Issues) == 0 {
if err := svc.triggers.registerWorkflows(ctx, res); err != nil {
return workflowUnchanged, err
}
}
return changes, err
}))
@@ -305,6 +314,8 @@ func (svc *workflow) updater(ctx context.Context, workflowID uint64, action func
res *types.Workflow
aProps = &workflowActionProps{workflow: &types.Workflow{ID: workflowID}}
err error
g *wfexec.Graph
runAs intAuth.Identifiable
)
err = store.Tx(ctx, svc.store, func(ctx context.Context, s store.Storer) (err error) {
@@ -324,9 +335,15 @@ func (svc *workflow) updater(ctx context.Context, workflowID uint64, action func
return err
}
if res.Issues, err = svc.validateWorkflow(ctx, res); err != nil {
if g, runAs, err = svc.validateWorkflow(ctx, res); err != nil {
return
} else if len(res.Issues) == 0 {
}
if err = svc.updateCache(res, runAs, g); err != nil {
return
}
if len(res.Issues) == 0 {
if err = svc.triggers.registerWorkflows(ctx, res); err != nil {
return err
}
@@ -344,12 +361,16 @@ func (svc *workflow) updater(ctx context.Context, workflowID uint64, action func
}
}
return err
return
})
return res, svc.recordAction(ctx, aProps, action, err)
}
func (svc *workflow) handleToID(h string) uint64 {
return svc.wIndex[h]
}
func (svc workflow) handleUpdate(upd *types.Workflow) workflowUpdateHandler {
return func(ctx context.Context, res *types.Workflow) (changes workflowChanges, err error) {
if isStale(upd.UpdatedAt, res.UpdatedAt, res.CreatedAt) {
@@ -472,60 +493,78 @@ func (svc workflow) handleUndelete(ctx context.Context, res *types.Workflow) (wo
}
func (svc *workflow) Load(ctx context.Context) error {
wwf, _, err := store.SearchAutomationWorkflows(ctx, svc.store, types.WorkflowFilter{
Deleted: filter.StateInclusive,
Disabled: filter.StateExcluded,
})
var (
wwf, _, err = store.SearchAutomationWorkflows(ctx, svc.store, types.WorkflowFilter{
Deleted: filter.StateInclusive,
Disabled: filter.StateExcluded,
})
g *wfexec.Graph
runAs intAuth.Identifiable
)
if err != nil {
return err
}
_ = wwf.Walk(func(wf *types.Workflow) error {
svc.wIndex[wf.Handle] = wf.ID
if g, runAs, err = svc.validateWorkflow(ctx, wf); err != nil {
return err
}
if err = svc.updateCache(wf, runAs, g); err != nil {
return err
}
return nil
})
return svc.triggers.registerWorkflows(ctx, wwf...)
}
// updateCache
func (svc *workflow) updateCache(wf *types.Workflow, runAs intAuth.Identifiable, g *wfexec.Graph) (err error) {
defer svc.mux.Unlock()
svc.mux.Lock()
if wf.Executable() {
svc.cache[wf.ID] = &wfCacheItem{g: g, wf: wf, runAs: runAs}
} else {
// remove deleted
delete(svc.cache, wf.ID)
}
return
}
func (svc *workflow) Exec(ctx context.Context, workflowID uint64, p types.WorkflowExecParams) (*expr.Vars, types.Stacktrace, error) {
var (
runAs intAuth.Identifiable
wap = &workflowActionProps{}
wf *types.Workflow
t *types.Trigger
results *expr.Vars
wait WaitFn
stacktrace types.Stacktrace
runner, invoker *sysTypes.User
)
err := func() (err error) {
wf, err = loadWorkflow(ctx, svc.store, workflowID)
if err != nil {
return
svc.mux.Lock()
if nil == svc.cache[workflowID] || nil == svc.cache[workflowID].wf {
svc.mux.Unlock()
return WorkflowErrNotFound()
}
wf := svc.cache[workflowID].wf
svc.mux.Unlock()
wap.setWorkflow(wf)
if !svc.ac.CanExecuteWorkflow(ctx, wf) {
return WorkflowErrNotAllowedToExecute()
}
// User wants to trace workflow execution
// This means we'll allow him to specify any (orphaned) step
// even if it's not linked to onManual trigger
if p.Trace && !svc.ac.CanManageSessionsOnWorkflow(ctx, wf) {
return WorkflowErrNotAllowedToExecute()
}
if !wf.Enabled && !p.Trace {
return WorkflowErrDisabled()
}
g, convErr := Convert(svc, wf)
if len(convErr) > 0 {
return convErr
}
// Find the trigger.
// @todo can we cache this as well?
t, err = func() (*types.Trigger, error) {
var tt types.TriggerSet
// Load triggers directly from the store. At this point we do not care
@@ -552,23 +591,6 @@ func (svc *workflow) Exec(ctx context.Context, workflowID uint64, p types.Workfl
return
}
// 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 {
if t == nil {
return WorkflowErrUnknownWorkflowStep()
@@ -579,59 +601,19 @@ func (svc *workflow) Exec(ctx context.Context, workflowID uint64, p types.Workfl
if t != nil {
wap.setTrigger(t)
p.StepID = t.StepID
p.EventType = t.EventType
p.ResourceType = t.ResourceType
// Add trigger's input to scope
scope = scope.MustMerge(t.Input)
_ = scope.AssignFieldValue("eventType", expr.Must(expr.NewString(t.EventType)))
_ = scope.AssignFieldValue("resourceType", expr.Must(expr.NewString(t.ResourceType)))
ssp.StepID = t.StepID
ssp.EventType = t.EventType
ssp.ResourceType = t.ResourceType
// merge with input from trigger
// with trigger input vars are overwritten by input vars
p.Input = t.Input.MustMerge(p.Input)
} else {
ssp.EventType = "onTrace"
ssp.ResourceType = ""
p.EventType = "onTrace"
}
// Returns context with identity set to service user
//
// Current user (identity in the context) might not have
// sufficient privileges to load info about invoker and runner
sysUserCtx := func() context.Context {
return intAuth.SetIdentityToContext(ctx, intAuth.ServiceUser())
}
if invokerId := intAuth.GetIdentityFromContext(ctx).Identity(); invokerId > 0 {
var is bool
if invoker, is = intAuth.GetIdentityFromContext(ctx).(*sysTypes.User); !is {
if invoker, err = DefaultUser.FindByAny(sysUserCtx(), invokerId); err != nil {
return
}
}
runner = invoker
}
if wf.RunAs > 0 {
if runner, err = DefaultUser.FindByAny(sysUserCtx(), wf.RunAs); err != nil {
return
}
}
if runAs == nil {
// Default to current user
runAs = intAuth.GetIdentityFromContext(ctx)
}
// @todo find a better way to typify expression values
// so that we do not have to import automation types from the system component
_ = scope.AssignFieldValue("invoker", expr.Must(sysAutoTypes.NewUser(invoker)))
_ = scope.AssignFieldValue("runner", expr.Must(sysAutoTypes.NewUser(runner)))
// Finally, assign input values
ssp.Input = scope.MustMerge(p.Input)
wait, err = svc.session.Start(g, runAs, ssp)
//wait, err = svc.session.Start(g, ssp)
wait, err = svc.exec(ctx, wf, p)
if err != nil {
return err
@@ -649,87 +631,118 @@ func (svc *workflow) Exec(ctx context.Context, workflowID uint64, p types.Workfl
// reuse scope for results
// this will be decoded back to event properties
results, _, stacktrace, err = wait(ctx)
return
return err
}()
return results, stacktrace, svc.recordAction(ctx, wap, WorkflowActionExecute, err)
}
// validates workflow by trying to convert it to graph and checking assigned triggers
func (svc *workflow) validateWorkflow(ctx context.Context, wf *types.Workflow) (types.WorkflowIssueSet, error) {
_, wis := Convert(svc, wf)
func (svc *workflow) validateWorkflow(ctx context.Context, wf *types.Workflow) (g *wfexec.Graph, runAs intAuth.Identifiable, err error) {
var (
tt []*types.Trigger
)
tt, _, err := store.SearchAutomationTriggers(ctx, svc.store, types.TriggerFilter{
g, wf.Issues = Convert(svc, wf)
tt, _, err = store.SearchAutomationTriggers(ctx, svc.store, types.TriggerFilter{
WorkflowID: types.WorkflowSet{wf}.IDs(),
Deleted: filter.StateExcluded,
Disabled: filter.StateExcluded,
})
if err != nil {
return nil, err
return
}
wis = append(wis, validateWorkflowTriggers(wf, tt...)...)
wf.Issues = append(wf.Issues, validateWorkflowTriggers(wf, tt...)...)
return wis, nil
// Returns context with identity set to service user
//
// Current user (identity in the context) might not have
// sufficient privileges to load info about invoker and runner
sysUserCtx := func() context.Context {
return intAuth.SetIdentityToContext(ctx, intAuth.ServiceUser())
}
// @todo this might not be the smartest thing, users might get invalidated after
// we add cache them as workflow runners
if wf.RunAs > 0 {
if runAs, err = DefaultUser.FindByAny(sysUserCtx(), wf.RunAs); err != nil {
wf.Issues = wf.Issues.Append(fmt.Errorf("failed to load run-as user %d: %w", wf.RunAs, err), nil)
} else if !runAs.Valid() {
wf.Issues = wf.Issues.Append(fmt.Errorf("invalid user %d used for workflow run-as", wf.RunAs), nil)
}
}
return
}
func makeWorkflowHandler(ac workflowExecController, s *session, t *types.Trigger, wf *types.Workflow, g *wfexec.Graph, _runAs intAuth.Identifiable) eventbus.HandlerFn {
func (svc *workflow) exec(ctx context.Context, wf *types.Workflow, p types.WorkflowExecParams) (WaitFn, error) {
if wf.Issues != nil {
return nil, wf.Issues
}
defer svc.mux.Unlock()
svc.mux.Lock()
if svc.cache[wf.ID] == nil {
return nil, WorkflowErrInvalidID()
}
var (
g = svc.cache[wf.ID].g
runAs = svc.cache[wf.ID].runAs
scope *expr.Vars
)
// merge workflow scope with the input
scope = wf.Scope.MustMerge(p.Input)
// User (either invoker or one set in the security descriptor) MUST have
// permissions to execute this workflow
if !svc.ac.CanExecuteWorkflow(ctx, wf) {
return nil, WorkflowErrNotAllowedToExecute()
}
return svc.session.Start(ctx, g, types.SessionStartParams{
Invoker: intAuth.GetIdentityFromContext(ctx),
Runner: runAs,
WorkflowID: wf.ID,
KeepFor: wf.KeepSessions,
Trace: wf.Trace,
Input: scope,
StepID: p.StepID,
EventType: p.EventType,
ResourceType: p.ResourceType,
CallStack: wfexec.GetContextCallStack(ctx),
})
}
func makeWorkflowHandler(svc *workflow, wf *types.Workflow, t *types.Trigger) eventbus.HandlerFn {
return func(ctx context.Context, ev eventbus.Event) (err error) {
var (
// create session scope from predefined workflow scope and trigger input
scope = wf.Scope.MustMerge(t.Input)
evScope *expr.Vars
wait WaitFn
// The returned closure needs to have its own instance, so it doesn't
// affect the instance bound to the workflow handler
runAs = _runAs
scope *expr.Vars
)
if enc, is := ev.(varsEncoder); is {
if evScope, err = enc.EncodeVars(); err != nil {
if dec, is := ev.(varsEncoder); is {
scope, err = dec.EncodeVars()
if err != nil {
return
}
scope = scope.MustMerge(evScope)
}
_ = scope.AssignFieldValue("eventType", expr.Must(expr.NewString(ev.EventType())))
_ = scope.AssignFieldValue("resourceType", expr.Must(expr.NewString(ev.ResourceType())))
if runAs == nil {
// @todo can/should we get alternative identity from Event?
// for example:
// - use http auth header and get username
// - use from/to/replyTo and use that as an identifier
runAs = intAuth.GetIdentityFromContext(ctx)
} else {
// Running workflow with a different security context
ctx = intAuth.SetIdentityToContext(ctx, runAs)
}
// User (either invoker or one set in the security descriptor) MUST have
// permissions to execute this workflow
if !ac.CanExecuteWorkflow(ctx, wf) {
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,
Trace: wf.Trace,
Input: scope,
wait, err := svc.exec(ctx, wf, types.WorkflowExecParams{
StepID: t.StepID,
EventType: t.EventType,
ResourceType: t.ResourceType,
Input: t.Input.MustMerge(scope),
CallStack: callStack,
Trace: wf.Trace,
Async: false,
})
if err != nil {
automation/service/workflow_converter.go
+22 -5
View File
@@ -3,13 +3,14 @@ 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"
"github.com/cortezaproject/corteza-server/pkg/expr"
"github.com/cortezaproject/corteza-server/pkg/wfexec"
"go.uber.org/zap"
"strings"
)
type (
@@ -18,6 +19,10 @@ type (
reg *registry
parser expr.Parsable
log *zap.Logger
graphs interface {
handleToID(string) uint64
}
}
)
@@ -26,6 +31,7 @@ func Convert(wfService *workflow, wf *types.Workflow) (*wfexec.Graph, types.Work
reg: wfService.reg,
parser: wfService.parser,
log: wfService.log,
graphs: wfService,
}
return conv.makeGraph(wf)
@@ -92,7 +98,7 @@ func (svc workflowConverter) makeGraph(def *types.Workflow) (*wfexec.Graph, type
stepIssues := verifyStep(step, inPaths, outPaths)
if resolved, err := svc.workflowStepDefConv(g, step, inPaths, outPaths); err != nil {
if resolved, err := svc.workflowStepDefConv(g, def, step, inPaths, outPaths); err != nil {
switch aux := err.(type) {
case types.WorkflowIssueSet:
stepIssues = append(stepIssues, aux...)
@@ -151,7 +157,7 @@ func (svc workflowConverter) makeGraph(def *types.Workflow) (*wfexec.Graph, type
// converts all step definitions into workflow.Step instances
//
// if this func returns nil for step and error, assume unresolved dependencies
func (svc workflowConverter) workflowStepDefConv(g *wfexec.Graph, s *types.WorkflowStep, in, out []*types.WorkflowPath) (bool, error) {
func (svc workflowConverter) workflowStepDefConv(g *wfexec.Graph, def *types.Workflow, s *types.WorkflowStep, in, out []*types.WorkflowPath) (bool, error) {
if err := svc.parseExpressions(s.Arguments...); err != nil {
return false, errors.Internal("failed to parse step arguments expressions for %s: %s", s.Kind, err).Wrap(err)
}
@@ -314,7 +320,7 @@ func (svc workflowConverter) convExpressionStep(s *types.WorkflowStep) (wfexec.S
}
// internal debug step that can log entire
func (svc workflowConverter) convDebugStep(s *types.WorkflowStep) (wfexec.Step, error) {
func (svc workflowConverter) convDebugStep(_ *types.WorkflowStep) (wfexec.Step, error) {
return types.DebugStep(svc.log), nil
}
@@ -443,7 +449,7 @@ func (svc workflowConverter) convPromptStep(s *types.WorkflowStep) (wfexec.Step,
}
var ownerId uint64 = 0
if i := auth.GetIdentityFromContext(ctx); i != nil {
if i := auth.GetIdentityFromContextWithKey(ctx, workflowInvokerCtxKey{}); i != nil {
ownerId = i.Identity()
}
@@ -558,6 +564,16 @@ func verifyStep(s *types.WorkflowStep, in, out types.WorkflowPathSet) types.Work
return nil
}
// check if reference is set on the step
numericRef = func() error {
// @todo SUBWF enable this back
//if 0 == cast.ToUint64(s.Ref) {
// return errors.Internal("%s step expects workflow reference ID", s.Kind)
//}
return nil
}
// reference should not be set on the step
noRef = func() error {
if s.Ref != "" {
@@ -672,6 +688,7 @@ func verifyStep(s *types.WorkflowStep, in, out types.WorkflowPathSet) types.Work
case types.WorkflowStepKindFunction:
checks = append(checks,
requiredRef,
numericRef,
checkDisabledFunc,
count(0, 1, outbound),
)
automation/types/session.go
+43
View File
@@ -8,6 +8,7 @@ import (
"sync"
"time"
"github.com/cortezaproject/corteza-server/pkg/auth"
"github.com/cortezaproject/corteza-server/pkg/expr"
"github.com/cortezaproject/corteza-server/pkg/filter"
"github.com/cortezaproject/corteza-server/pkg/wfexec"
@@ -53,6 +54,12 @@ type (
}
SessionStartParams struct {
// Always set, users that invoked/started the workflow session
Invoker auth.Identifiable
// Optional, (alternative) user that is running the workflow
Runner auth.Identifiable
WorkflowID uint64
KeepFor int
Trace bool
@@ -194,6 +201,42 @@ func (set Stacktrace) Value() (driver.Value, error) {
return json.Marshal(set)
}
func (set Stacktrace) String() (str string) {
for i, f := range set {
str += fmt.Sprintf(
"[%3d] %-14s %d (",
i,
f.CreatedAt.Format("15:04:05.00000"),
f.StepID,
)
if f.Input.Len() == 0 {
str += "no input, "
}
if f.Scope.Len() == 0 {
str += "no scope, "
}
if f.Results.Len() == 0 {
str += "no results, "
}
str += ")"
if f.Scope.Len() > 0 {
str += fmt.Sprintf(" Scope:\n")
f.Scope.Each(func(k string, v expr.TypedValue) error {
str += fmt.Sprintf(" [%s]: %v\n", k, v)
return nil
})
}
str += "\n"
}
return
}
func (s SessionStatus) String() string {
switch s {
case SessionStarted:
automation/types/step.go
+1
View File
@@ -5,6 +5,7 @@ import (
"database/sql/driver"
"encoding/json"
"fmt"
"github.com/cortezaproject/corteza-server/pkg/expr"
)
automation/types/workflow.go
+17
View File
@@ -79,9 +79,21 @@ type (
}
WorkflowExecParams struct {
// When executed as a sub-workflow
CallerWorkflowID uint64
// When executed as a sub-workflow
CallerSessionID uint64
// When executed as a sub-workflow
CallerStepID uint64
// Start with this specific step
StepID uint64
EventType string
ResourceType string
// Enable execution tracing
Trace bool
@@ -105,6 +117,11 @@ func (r Workflow) CheckDeferred() bool {
return r.Steps.HasDeferred()
}
// Executable returns true if workflow is valid and enabled
func (r Workflow) Executable() bool {
return r.DeletedAt == nil && r.Enabled
}
func (r Workflow) Dict() map[string]interface{} {
return map[string]interface{}{
"ID": r.ID,
pkg/auth/identity_context.go
+10 -2
View File
@@ -16,9 +16,17 @@ func SetIdentityToContext(ctx context.Context, identity Identifiable) context.Co
//
// For anonymous user, it auto appends all anonymous defined on the system
func GetIdentityFromContext(ctx context.Context) Identifiable {
if identity, ok := ctx.Value(identityCtxKey{}).(Identifiable); ok && identity != nil && identity.Valid() {
return identity
if i := GetIdentityFromContextWithKey(ctx, identityCtxKey{}); i != nil && i.Valid() {
return i
} else {
return Anonymous()
}
}
func GetIdentityFromContextWithKey(ctx context.Context, key interface{}) Identifiable {
if i, ok := ctx.Value(key).(Identifiable); ok {
return i
} else {
return nil
}
}