core

Documentation

Overview

Package core This is core package for the scheduler which includes

• scheduler interface
• scheduler implementation using gocron https://github.com/robfig/cron
• updater which updates the schedules in the scheduler by reading periodically from the DB
• snapshot runner which snapshot the schedules with there last exec times so that it can be used as check point in case of a crash. After a crash the scheduler replays the schedules from the last recorded snapshot. It relies on the admin idempotency aspect to fail executions if the execution with a scheduled time already exists with it.

Index

• func GetCatchUpTimes(s models.SchedulableEntity, from time.Time, to time.Time) ([]time.Time, error)
• func GetScheduledTime(s models.SchedulableEntity, fromTime time.Time) (time.Time, error)
• type GoCronJob
  • func (g *GoCronJob) Run(t time.Time)
• type GoCronScheduler
  • func (g *GoCronScheduler) AddCronJob(ctx context.Context, job *GoCronJob) error
  • func (g *GoCronScheduler) AddFixedIntervalJob(ctx context.Context, job *GoCronJob) error
  • func (g *GoCronScheduler) BootStrapSchedulesFromSnapShot(ctx context.Context, schedules []models.SchedulableEntity, ...)
  • func (g *GoCronScheduler) CalculateSnapshot(ctx context.Context) snapshoter.Snapshot
  • func (g *GoCronScheduler) CatchUpSingleSchedule(ctx context.Context, s models.SchedulableEntity, fromTime time.Time, ...) error
  • func (g *GoCronScheduler) CatchupAll(ctx context.Context, until time.Time) bool
  • func (g *GoCronScheduler) DeScheduleJob(ctx context.Context, schedule models.SchedulableEntity)
  • func (g *GoCronScheduler) GetTimedFuncWithSchedule() TimedFuncWithSchedule
  • func (g *GoCronScheduler) RemoveCronJob(ctx context.Context, job *GoCronJob)
  • func (g *GoCronScheduler) RemoveFixedIntervalJob(ctx context.Context, job *GoCronJob)
  • func (g *GoCronScheduler) ScheduleJob(ctx context.Context, schedule models.SchedulableEntity, ...) error
  • func (g *GoCronScheduler) UpdateSchedules(ctx context.Context, schedules []models.SchedulableEntity)
• type Scheduler
  • func NewGoCronScheduler(ctx context.Context, schedules []models.SchedulableEntity, ...) Scheduler
• type Snapshotrunner
  • func NewSnapshotRunner(snapshoter sImpl.Persistence, scheduler Scheduler) Snapshotrunner
  • func (u Snapshotrunner) Run(ctx context.Context)
• type TimedFuncWithSchedule
• type Updater
  • func NewUpdater(db repositories.SchedulerRepoInterface, scheduler Scheduler) Updater
  • func (u Updater) UpdateGoCronSchedules(ctx context.Context)

Functions

func GetCatchUpTimes

func GetCatchUpTimes(s models.SchedulableEntity, from time.Time, to time.Time) ([]time.Time, error)

func GetScheduledTime

func GetScheduledTime(s models.SchedulableEntity, fromTime time.Time) (time.Time, error)

Types

type GoCronJob

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

GoCronJob this provides a wrapper around the go cron libraries job function.

func (*GoCronJob) Run

func (g *GoCronJob) Run(t time.Time)

type GoCronScheduler

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

GoCronScheduler this provides a scheduler functionality using the https://github.com/robfig/cron library.

func (*GoCronScheduler) AddCronJob

func (g *GoCronScheduler) AddCronJob(ctx context.Context, job *GoCronJob) error

func (*GoCronScheduler) AddFixedIntervalJob

func (g *GoCronScheduler) AddFixedIntervalJob(ctx context.Context, job *GoCronJob) error

func (*GoCronScheduler) BootStrapSchedulesFromSnapShot

func (g *GoCronScheduler) BootStrapSchedulesFromSnapShot(ctx context.Context, schedules []models.SchedulableEntity,
	snapshot snapshoter.Snapshot)

func (*GoCronScheduler) CalculateSnapshot

func (g *GoCronScheduler) CalculateSnapshot(ctx context.Context) snapshoter.Snapshot

func (*GoCronScheduler) CatchUpSingleSchedule

func (g *GoCronScheduler) CatchUpSingleSchedule(ctx context.Context, s models.SchedulableEntity, fromTime time.Time, toTime time.Time) error

func (*GoCronScheduler) CatchupAll

func (g *GoCronScheduler) CatchupAll(ctx context.Context, until time.Time) bool

func (*GoCronScheduler) DeScheduleJob

func (g *GoCronScheduler) DeScheduleJob(ctx context.Context, schedule models.SchedulableEntity)

func (*GoCronScheduler) GetTimedFuncWithSchedule

func (g *GoCronScheduler) GetTimedFuncWithSchedule() TimedFuncWithSchedule

func (*GoCronScheduler) RemoveCronJob

func (g *GoCronScheduler) RemoveCronJob(ctx context.Context, job *GoCronJob)

func (*GoCronScheduler) RemoveFixedIntervalJob

func (g *GoCronScheduler) RemoveFixedIntervalJob(ctx context.Context, job *GoCronJob)

func (*GoCronScheduler) ScheduleJob

func (g *GoCronScheduler) ScheduleJob(ctx context.Context, schedule models.SchedulableEntity,
	funcWithSchedule TimedFuncWithSchedule, lastTime *time.Time) error

func (*GoCronScheduler) UpdateSchedules

func (g *GoCronScheduler) UpdateSchedules(ctx context.Context, schedules []models.SchedulableEntity)

type Scheduler

type Scheduler interface {
	// ScheduleJob allows to schedule a job using the implemented scheduler
	ScheduleJob(ctx context.Context, s models.SchedulableEntity, f TimedFuncWithSchedule, lastT *time.Time) error
	// DeScheduleJob allows to remove a scheduled job using the implemented scheduler
	DeScheduleJob(ctx context.Context, s models.SchedulableEntity)
	// BootStrapSchedulesFromSnapShot allows to initialize the scheduler from a previous snapshot of the schedule executions
	BootStrapSchedulesFromSnapShot(ctx context.Context, schedules []models.SchedulableEntity, snapshot snapshoter.Snapshot)
	// UpdateSchedules updates all the schedules in the schedulers job store
	UpdateSchedules(ctx context.Context, s []models.SchedulableEntity)
	// CalculateSnapshot creates a snapshot of the existing state of the schedules run by the scheduler which can be used in case of failure.
	CalculateSnapshot(ctx context.Context) snapshoter.Snapshot
	// CatchupAll catches up all the schedules in the schedulers job store to the until time
	CatchupAll(ctx context.Context, until time.Time) bool
}

Scheduler is the main scheduler interfaces for scheduling/descheduling jobs, updating the schedules, calculating snapshot of the schedules , bootstrapping the scheduler from the snapshot as well as the catcup functionality

func NewGoCronScheduler

func NewGoCronScheduler(ctx context.Context, schedules []models.SchedulableEntity, scope promutils.Scope,
	snapshot snapshoter.Snapshot, rateLimiter *rate.Limiter, executor executor.Executor) Scheduler

type Snapshotrunner

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

Snapshotrunner allows the ability to snapshot the scheduler state and save it to the db. Its invoked periodically from the scheduledExecutor

func NewSnapshotRunner

func NewSnapshotRunner(snapshoter sImpl.Persistence, scheduler Scheduler) Snapshotrunner

func (Snapshotrunner) Run

func (u Snapshotrunner) Run(ctx context.Context)

type TimedFuncWithSchedule

type TimedFuncWithSchedule func(ctx context.Context, s models.SchedulableEntity, t time.Time) error

type Updater

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

Updater this updates the scheduler with the latest state of schedules from the DB.

func NewUpdater

func NewUpdater(db repositories.SchedulerRepoInterface,
	scheduler Scheduler) Updater

func (Updater) UpdateGoCronSchedules

func (u Updater) UpdateGoCronSchedules(ctx context.Context)