scheduler

Documentation

Overview

Package scheduler allows callers to register recurring jobs on the controller. The scheduler will periodically query the repository for registered jobs that should be run.

Before a job can be invoked by the scheduler, it must be made known to the scheduler by being registered on the instance of the scheduler that is running.

Index

• type Job
• type JobStatus
• type Option
  • func WithInterruptThreshold(l time.Duration) Option
  • func WithMonitorInterval(l time.Duration) Option
  • func WithNextRunIn(d time.Duration) Option
  • func WithRunJobsInterval(l time.Duration) Option
  • func WithRunJobsLimit(l int) Option
  • func WithRunNow(b bool) Option
• type Scheduler
  • func New(serverId string, jobRepoFn jobRepoFactory, opt ...Option) (*Scheduler, error)
  • func TestScheduler(t testing.TB, conn *db.DB, wrapper wrapping.Wrapper, opt ...Option) *Scheduler
  • func (s *Scheduler) RegisterJob(ctx context.Context, j Job, opt ...Option) error
  • func (s *Scheduler) RunNow()
  • func (s *Scheduler) Start(ctx context.Context, wg *sync.WaitGroup) error
  • func (s *Scheduler) UpdateJobNextRunInAtLeast(ctx context.Context, name string, nextRunInAtLeast time.Duration, ...) error

Types

type Job

type Job interface {
	// Status reports the job’s current status.  The status is periodically persisted by
	// the scheduler when a job is running, and will be used to verify a job is making progress.
	Status() JobStatus

	// Run performs the required work depending on the implementation.
	// The context is used to notify the job that it should exit early.
	Run(ctx context.Context) error

	// NextRunIn returns the duration until the next job run should be scheduled.  This
	// method is invoked after a run has successfully completed and the next run time
	// is being persisted by the scheduler.  If an error is returned, the error will be logged
	// but the duration returned will still be used in scheduling.  If a zero duration is returned
	// the job will be scheduled to run again immediately.
	NextRunIn(context.Context) (time.Duration, error)

	// Name is the unique name of the job.
	Name() string

	// Description is the human readable description of the job.
	Description() string
}

Job defines an interface for jobs that can be invoked by the scheduler.

type JobStatus

type JobStatus struct {
	// Completed and Total are used to indicate job progress,
	// each job implementation will determine the definition of
	// progress by calculating both Completed and Total.
	Completed, Total int
}

JobStatus defines the struct that must be returned by the Job.Status() method.

type Option

type Option func(*options)

Option - how Options are passed as arguments

func WithInterruptThreshold

func WithInterruptThreshold(l time.Duration) Option

WithInterruptThreshold provides an option to provide the duration after which a controller will interrupt a running job that is not updating its status. If WithInterruptThreshold == 0, then default duration is used.

func WithMonitorInterval

func WithMonitorInterval(l time.Duration) Option

WithMonitorInterval provides an option to provide the interval at which the scheduler will query running jobs for status and update the repository accordingly. If WithMonitorInterval == 0, then default interval is used.

func WithNextRunIn

func WithNextRunIn(d time.Duration) Option

WithNextRunIn provides an option to provide the duration until the next run is scheduled. If this option is not provided the NextScheduledRun of the job will default to the current database time, and be available to run immediately.

func WithRunJobsInterval

func WithRunJobsInterval(l time.Duration) Option

WithRunJobsInterval provides an option to provide the interval at which the scheduler will query the repository for jobs to run. If WithRunJobsInterval == 0, then default interval is used.

func WithRunJobsLimit

func WithRunJobsLimit(l int) Option

WithRunJobsLimit provides an option to provide the number of jobs that will be requested by the scheduler when querying for jobs to run. If WithRunJobsLimit == 0, then default run jobs limit is used. If WithRunJobsLimit < 0, then no limit is used.

func WithRunNow

func WithRunNow(b bool) Option

WithRunNow provides an option to trigger the scheduling loop after updating the next run time of a specific job. Note this does not guarantee the job will run on the scheduler that updated the job run time.

type Scheduler

type Scheduler struct {
	// contains filtered or unexported fields
}

Scheduler is used to register and run recurring jobs on the controller.

func New

func New(serverId string, jobRepoFn jobRepoFactory, opt ...Option) (*Scheduler, error)

New creates a new Scheduler

• serverId must be provided and is the private id of the server that will run the scheduler

• jobRepoFn must be provided and is a function that returns the job repository

WithRunJobsLimit, WithRunJobsInterval, WithMonitorInterval and WithInterruptThreshold are the only valid options.

func TestScheduler

func TestScheduler(t testing.TB, conn *db.DB, wrapper wrapping.Wrapper, opt ...Option) *Scheduler

TestScheduler creates a mock controller and a new Scheduler attached to that controller id. The Scheduler returned should only be used for tests. The mock controller is not run.

WithRunJobsLimit, WithRunJobsInterval, WithMonitorInterval and WithInterruptThreshold are the only valid options.

func (*Scheduler) RegisterJob

func (s *Scheduler) RegisterJob(ctx context.Context, j Job, opt ...Option) error

RegisterJob registers a job with the scheduler and persists the job into the repository.

• job must be provided and is an implementer of the Job interface.

WithNextRunIn is the only valid options.

func (*Scheduler) RunNow

func (s *Scheduler) RunNow()

RunNow attempts to trigger the scheduling loop, if the scheduling loop is actively running it will cause the loop to run again immediately after finishing.

func (*Scheduler) Start

func (s *Scheduler) Start(ctx context.Context, wg *sync.WaitGroup) error

Start begins the scheduling loop that will query the repository for jobs to run and run them in a goroutine, the scheduler will stop all running jobs and stop requesting new jobs once the ctx past in is canceled. The scheduler cannot be started again once the ctx is canceled, a new scheduler will need to be instantiated in order to begin scheduling again.

func (*Scheduler) UpdateJobNextRunInAtLeast

func (s *Scheduler) UpdateJobNextRunInAtLeast(ctx context.Context, name string, nextRunInAtLeast time.Duration, opt ...Option) error

UpdateJobNextRunInAtLeast updates the next scheduled run time for the provided name, setting the job's NextScheduledRun time to either the current database time incremented by the nextRunInAtLeast parameter or the current NextScheduledRun time value, which ever is sooner. If nextRunInAtLeast == 0 the job will be available to run immediately.

WithRunNow is the only supported option.