Skip to content

Commit

Permalink
Tune workflow session flushing frequency
Browse files Browse the repository at this point in the history
  • Loading branch information
darh committed Dec 8, 2021
1 parent b0590d2 commit 486a575
Showing 1 changed file with 13 additions and 10 deletions.
23 changes: 13 additions & 10 deletions automation/service/session.go
Expand Up @@ -50,6 +50,16 @@ type (
WaitFn func(ctx context.Context) (*expr.Vars, wfexec.SessionStatus, types.Stacktrace, error)
)

const (
// when the state changes, state-change-handler is called and for non-fatal,
// non-interactive or non-delay steps (that are much more frequent) we need
// to limit how often the store is updated with the updated session info
//
// We use the size of the stacktrace and for every F (see the value of the constant)
// we flush the session info to the store.
sessionStateFlushFrequency = 1000
)

func Session(log *zap.Logger, opt options.WorkflowOpt, ps promptSender) *session {
return &session{
log: log,
Expand Down Expand Up @@ -360,14 +370,9 @@ 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'll flush it every X frames (flushFrequency)
// But if status is active, we'll flush it every X frames (sessionStateFlushFrequency)
update = true

frame = state.MakeFrame()
Expand Down Expand Up @@ -411,8 +416,8 @@ func (svc *session) stateChangeHandler(ctx context.Context) wfexec.StateChangeHa
ses.Status = types.SessionFailed

default:
// force update on every 10 new frames but only when stacktrace is not nil
update = ses.RuntimeStacktrace != nil && len(ses.RuntimeStacktrace)%flushFrequency == 0
// force update on every F new frames (F=sessionStateFlushFrequency) but only when stacktrace is not nil
update = ses.RuntimeStacktrace != nil && len(ses.RuntimeStacktrace)%sessionStateFlushFrequency == 0
}

if !update {
Expand All @@ -423,8 +428,6 @@ func (svc *session) stateChangeHandler(ctx context.Context) wfexec.StateChangeHa

if err := svc.store.UpsertAutomationSession(ctx, ses); err != nil {
log.Error("failed to update session", zap.Error(err))
} else {
log.Debug("session updated", zap.Stringer("status", ses.Status))
}
}
}
Expand Down

0 comments on commit 486a575

Please sign in to comment.