DeforaNetworks/corteza
3
0
Fork 0
Code Activity

Improve workflow session handling & storing

Browse Source
This commit is contained in:
Denis Arh
2021-04-01 21:07:17 +02:00
parent 4b347535e9
commit 1a54155d83
2 changed files with 112 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
automation/service/session.go
+88 -18
View File
@@ -106,12 +106,24 @@ func (svc *session) LookupByID(ctx context.Context, sessionID uint64) (res *type
}
func (svc *session) resumeAll(ctx context.Context) error {
// In theory we could resume active/pending/prompt sessions from persistent store
// so that they can survive server termination
// @todo resume active sessions from storage
// load all active sessions from store and load them into the pool
//
return nil
}
func (svc *session) suspendAll(ctx context.Context) error {
// @todo suspend active sessions to storage
// In theory we could suspend active/pending/prompt sessions to persistent store
// so that they can survive server termination
// @todo suspend active sessions to storage:
// stop watcher queue
// run gc
// stop worker on each session
// flush session to store (like we're doing in the status handler
return nil
}
@@ -169,8 +181,10 @@ func (svc *session) Start(g *wfexec.Graph, i auth.Identifiable, ssp types.Sessio
ses.CreatedBy = i.Identity()
ses.Apply(ssp)
if err = store.CreateAutomationSession(context.TODO(), svc.store, ses); err != nil {
return
if ssp.Trace {
if err = store.CreateAutomationSession(context.TODO(), svc.store, ses); err != nil {
return
}
}
if err = ses.Exec(ctx, start, ssp.Input); err != nil {
@@ -225,8 +239,11 @@ func (svc *session) spawn(g *wfexec.Graph, workflowID uint64, trace bool) (ses *
}
func (svc *session) Watch(ctx context.Context) {
gcTicker := time.NewTicker(time.Second)
go func() {
defer sentry.Recover()
defer gcTicker.Stop()
defer svc.log.Info("stopped")
for {
@@ -234,23 +251,25 @@ func (svc *session) Watch(ctx context.Context) {
case <-ctx.Done():
return
case s := <-svc.spawnQueue:
wfexecSessLog := zap.NewNop()
if svc.opt.ExecDebug {
wfexecSessLog = svc.log.
Named("exec").
With(zap.Uint64("workflowID", s.workflowID))
opts := []wfexec.SessionOpt{
wfexec.SetWorkflowID(s.workflowID),
wfexec.SetHandler(svc.stateChangeHandler(ctx)),
}
//
s.session <- wfexec.NewSession(ctx,
s.graph,
wfexec.SetHandler(svc.stateChangeHandler(ctx)),
wfexec.SetWorkflowID(s.workflowID),
wfexec.SetLogger(wfexecSessLog),
wfexec.SetDumpStacktraceOnPanic(svc.opt.ExecDebug),
)
if svc.opt.ExecDebug {
opts = append(
opts,
wfexec.SetLogger(svc.log.Named("exec").With(zap.Uint64("workflowID", s.workflowID))),
wfexec.SetDumpStacktraceOnPanic(true),
)
}
s.session <- wfexec.NewSession(ctx, s.graph, opts...)
// case time for a pool cleanup
// @todo cleanup pool when sessions are complete
case <-gcTicker.C:
svc.gc()
}
}
@@ -261,6 +280,50 @@ func (svc *session) Watch(ctx context.Context) {
svc.log.Debug("watcher initialized")
}
// garbage collection for stale sessions
func (svc *session) gc() {
defer svc.mux.Unlock()
svc.mux.Lock()
var (
total = len(svc.pool)
removed, pending1m, pending1h, pending1d int
)
for _, s := range svc.pool {
switch {
case s.CreatedAt.Sub(*now()) > time.Minute:
pending1m++
case s.CreatedAt.Sub(*now()) > time.Hour:
pending1h++
case s.CreatedAt.Sub(*now()) > time.Hour*24:
pending1d++
}
if s.CompletedAt == nil {
continue
}
removed++
delete(svc.pool, s.ID)
}
if total > 0 {
svc.log.Info(
"workflow session garbage collector stats",
zap.Int("total", total),
zap.Int("removed", removed),
zap.Int("pending1m", pending1m),
zap.Int("pending1h", pending1h),
zap.Int("pending1d", pending1d),
)
}
}
// stateChangeHandler keeps track of session status changes and frequently stores session into db
func (svc *session) stateChangeHandler(ctx context.Context) wfexec.StateChangeHandler {
return func(i wfexec.SessionStatus, state *wfexec.State, s *wfexec.Session) {
log := svc.log.With(zap.Uint64("sessionID", s.ID()))
@@ -273,14 +336,21 @@ func (svc *session) stateChangeHandler(ctx context.Context) wfexec.StateChangeHa
return
}
const (
// how often do we flush to store
flushFrequency = 10
)
var (
// By default we want to update session when new status is prompted, delayed, completed or failed
//
// But if status is active, we
update = true
frame = state.MakeFrame()
)
// Stacktrace will be set to !nil if frame collection is needed
if ses.Stacktrace != nil {
if len(ses.Stacktrace) > 0 {
// calculate how long it took to get to this step
frame.ElapsedTime = uint(frame.CreatedAt.Sub(ses.Stacktrace[0].CreatedAt) / time.Millisecond)
@@ -311,7 +381,7 @@ func (svc *session) stateChangeHandler(ctx context.Context) wfexec.StateChangeHa
default:
// force update on every 10 new frames but only when stacktrace is not nil
update = ses.Stacktrace != nil && len(ses.Stacktrace)%10 == 0
update = ses.Stacktrace != nil && len(ses.Stacktrace)%flushFrequency == 0
}
if !update {
pkg/wfexec/session.go
+24 -15
View File
@@ -66,7 +66,7 @@ type (
StateChangeHandler func(SessionStatus, *State, *Session)
sessionOpt func(*Session)
SessionOpt func(*Session)
Frame struct {
CreatedAt time.Time `json:"createdAt"`
@@ -152,7 +152,7 @@ func (s SessionStatus) String() string {
return "UNKNOWN-SESSION-STATUS"
}
func NewSession(ctx context.Context, g *Graph, oo ...sessionOpt) *Session {
func NewSession(ctx context.Context, g *Graph, oo ...SessionOpt) *Session {
s := &Session{
g: g,
id: nextID(),
@@ -399,6 +399,10 @@ func (s *Session) worker(ctx context.Context) {
// making sure result != nil
s.result = (&expr.Vars{}).Merge(st.scope)
// Call event handler with completed status
s.eventHandler(SessionCompleted, st, s)
return
}
@@ -418,6 +422,7 @@ func (s *Session) worker(ctx context.Context) {
// remove single
<-s.execLock
status := s.Status()
if st.err != nil {
st.err = fmt.Errorf(
"workflow %d step %d execution failed: %w",
@@ -425,9 +430,19 @@ func (s *Session) worker(ctx context.Context) {
st.step.ID(),
st.err,
)
s.mux.Lock()
s.mux.Unlock()
// We need to force failed session status
// because it's not set early enough to pick it up with s.Status()
status = SessionFailed
// pushing step execution error into error queue
// to break worker loop
s.qErr <- st.err
}
status := s.Status()
s.log.Debug(
"executed",
zap.Uint64("stateID", st.stateId),
@@ -437,12 +452,6 @@ func (s *Session) worker(ctx context.Context) {
// after exec lock is released call event handler with (new) session status
s.eventHandler(status, st, s)
if err != nil {
// pushing step execution error into error queue
// to break worker loop
s.qErr <- st.err
}
}()
case err := <-s.qErr:
@@ -461,7 +470,7 @@ func (s *Session) worker(ctx context.Context) {
}
func (s *Session) Stop() {
s.log.Debug("stopping session")
s.log.Debug("stopping session", zap.Stringer("status", s.Status()))
defer s.workerTicker.Stop()
}
@@ -739,31 +748,31 @@ func (s *Session) exec(ctx context.Context, st *State) (err error) {
return
}
func SetWorkerInterval(i time.Duration) sessionOpt {
func SetWorkerInterval(i time.Duration) SessionOpt {
return func(s *Session) {
s.workerInterval = i
}
}
func SetHandler(fn StateChangeHandler) sessionOpt {
func SetHandler(fn StateChangeHandler) SessionOpt {
return func(s *Session) {
s.eventHandler = fn
}
}
func SetWorkflowID(workflowID uint64) sessionOpt {
func SetWorkflowID(workflowID uint64) SessionOpt {
return func(s *Session) {
s.workflowID = workflowID
}
}
func SetLogger(log *zap.Logger) sessionOpt {
func SetLogger(log *zap.Logger) SessionOpt {
return func(s *Session) {
s.log = log
}
}
func SetDumpStacktraceOnPanic(dump bool) sessionOpt {
func SetDumpStacktraceOnPanic(dump bool) SessionOpt {
return func(s *Session) {
s.dumpStacktraceOnPanic = dump
}