From 1a54155d831d6f63dbf39aa380f694fc993be51d Mon Sep 17 00:00:00 2001 From: Denis Arh Date: Thu, 1 Apr 2021 21:07:17 +0200 Subject: [PATCH] Improve workflow session handling & storing --- automation/service/session.go | 106 ++++++++++++++++++++++++++++------ pkg/wfexec/session.go | 39 ++++++++----- 2 files changed, 112 insertions(+), 33 deletions(-) diff --git a/automation/service/session.go b/automation/service/session.go index e74a58093..3a3a991c9 100644 --- a/automation/service/session.go +++ b/automation/service/session.go @@ -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 { diff --git a/pkg/wfexec/session.go b/pkg/wfexec/session.go index a9d0e6437..b14215c17 100644 --- a/pkg/wfexec/session.go +++ b/pkg/wfexec/session.go @@ -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 }