crontask

README

synchronized-cron-task

synchronized-cron-task is a lightweight wrapper around github.com/robfig/cron, synchronizing via a redis lock.

What this means ist that synchronized-cron-task provides a guarantee that with any number of running instances of an application, the given cron task is only ever executed on one instance at a time.

Note: Unlike most my projects, this project is licenced under the Apache License 2.0!

Download:

go get github.com/kernle32dll/synchronized-cron-task

Detailed documentation can be found on GoDoc.

Compatibility

synchronized-cron-task is automatically tested against the following:

• Go 1.17.X, 1.18.X and 1.19.X
• Redis 5.X, 6.X and 7.X.

Getting started

This project consists of two tightly intertwined functionalities. SynchronizedCronTask, and TimeKeeper.

The former is the heart of the project, providing the mentioned synchronized cron execution. The latter can optionally be used to collect data about individual SynchronizedCronTask executions, such as run time or errors (if any).

Examples to use with and without time keeper can be found in the examples directory.

Synchronized cron task

A synchronized cron task can be created via two methods:

crontask.NewSynchronizedCronTask(redisClient, someFunc, options...)

crontask.NewSynchronizedCronTaskWithOptions(redisClient, someFunc, &crontask.TaskOptions{})

The former takes an array of crontask.TaskOption elements. Corresponding functions can be found at the source, or on GoDoc

The synchronized cron task will be executed asynchronously in the background. Nothing more has to be done for it to work.

Its ExecuteNow() and NextTime() functions can be used at any time for some additional control.

A synchronized cron task includes an graceful shutdown method Stop(ctx), which irreversibly shuts down the task. This should be done before application shutdown, to ensure that the current execution - if running - exits gracefully.

Time keeper

A time keeper can be - just like a synchronized cron task - created via two methods:

timekeeper.NewTimeKeeper(redisClient, options...)

timekeeper.NewTimeKeeper(redisClient, &timekeeper.Options{})

The former takes an array of timekeeper.Option elements. Corresponding functions can be found at the source, or on GoDoc

For the time keeper to actually do something, it must wrap task functions via its WrapCronTask function, which are then used in a synchronized cron task.

For example:

timeKeeper, err := timekeeper.NewTimeKeeper(redisClient)
if err != nil {
    // oh no, and error...
}

// New synchronized cron task
task, err := crontask.NewSynchronizedCronTask(
    redisClient,

    timeKeeper.WrapCronTask(
        func(ctx context.Context, task crontask.Task) error {
            // ... do something
            return nil
        },
    ),
)

A time keeper provides some functions, that allow introspection of recorded task executions. E.g. the last execution of a given task, or the total number of executions. These functions provide - ony way or another - ExecutionResult objects, which in themselves contain useful meta information, such as time of last and next execution, or errors (if any occurred).

Just like an synchronized cron task, a time keeper includes an graceful shutdown method Stop(ctx), which irreversibly shuts down the clean up task of an time keeper, if it exists. This should be done before application shutdown, to ensure that the cleanup task - if running - exits gracefully.

Documentation

Index

• Constants
• type SynchronizedCronTask
  • func NewSynchronizedCronTask(client redislock.RedisClient, taskFunc TaskFunc, setters ...TaskOption) (*SynchronizedCronTask, error)
  • func NewSynchronizedCronTaskWithOptions(client redislock.RedisClient, taskFunc TaskFunc, options *TaskOptions) (*SynchronizedCronTask, error)
  • func (synchronizedCronTask *SynchronizedCronTask) ExecuteNow()
  • func (synchronizedCronTask *SynchronizedCronTask) MarshalJSON() ([]byte, error)
  • func (synchronizedCronTask *SynchronizedCronTask) Name() string
  • func (synchronizedCronTask *SynchronizedCronTask) NextTime() time.Time
  • func (synchronizedCronTask *SynchronizedCronTask) Stop(ctx context.Context)
• type Task
• type TaskFunc
• type TaskOption
  • func CronExpression(cronExpression string) TaskOption
  • func LeadershipTimeout(leadershipTimeout time.Duration) TaskOption
  • func LockHeartbeat(lockHeartbeat time.Duration) TaskOption
  • func LockTimeout(lockTimeout time.Duration) TaskOption
  • func Logger(logger *logrus.Logger) TaskOption
  • func TaskName(name string) TaskOption
• type TaskOptions

Constants

const (
	// DefaultName is the default name of a synchronized cron task.
	DefaultName = "Default Synchronized Task"

	// DefaultCronExpression is the default cron expression of a
	// synchronized cron task. 0 * * * * * means every minute.
	DefaultCronExpression = "0 * * * * *"

	// DefaultLeadershipTimeout is the default timeout of a single
	// execution of a synchronized cron task.
	DefaultLeadershipTimeout = 30 * time.Second

	// DefaultLockTimeout is the default timeout for the lock of a
	// single execution of a synchronized cron task.
	DefaultLockTimeout = 5 * time.Second

	// DefaultLockHeartbeat is the default interval, in which an
	// acquired lock should be renewed (to the total of the leadership
	// timeout).
	DefaultLockHeartbeat = 1 * time.Second
)

Types

type SynchronizedCronTask

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

SynchronizedCronTask describes a task, which is identified by a cron expression and a redis client it uses to synchronize execution across running instances.

It supports graceful shutdowns via its Stop() function.

func NewSynchronizedCronTask

func NewSynchronizedCronTask(client redislock.RedisClient, taskFunc TaskFunc, setters ...TaskOption) (*SynchronizedCronTask, error)

NewSynchronizedCronTask creates a new SynchronizedCronTask instance, or errors out if the provided cron expression was invalid.

func NewSynchronizedCronTaskWithOptions

func NewSynchronizedCronTaskWithOptions(client redislock.RedisClient, taskFunc TaskFunc, options *TaskOptions) (*SynchronizedCronTask, error)

NewSynchronizedCronTaskWithOptions creates a new SynchronizedCronTask instance, or errors out if the provided cron expression was invalid.

func (*SynchronizedCronTask) ExecuteNow

func (synchronizedCronTask *SynchronizedCronTask) ExecuteNow()

ExecuteNow forces the cron to fire immediately. Locking is still honored, so no concurrent task execution can be forced this way.

func (*SynchronizedCronTask) MarshalJSON

func (synchronizedCronTask *SynchronizedCronTask) MarshalJSON() ([]byte, error)

func (*SynchronizedCronTask) Name

func (synchronizedCronTask *SynchronizedCronTask) Name() string

Name returns the name of the task.

func (*SynchronizedCronTask) NextTime

func (synchronizedCronTask *SynchronizedCronTask) NextTime() time.Time

NextTime returns the next time the cron task will fire.

func (*SynchronizedCronTask) Stop

func (synchronizedCronTask *SynchronizedCronTask) Stop(ctx context.Context)

Stop gracefully stops the task, while also freeing most of its underlying resources.

type Task

type Task interface {
	Name() string
	NextTime() time.Time
}

Task is an abstraction of a running task.

type TaskFunc

type TaskFunc func(ctx context.Context, task Task) error

TaskFunc is a function, that is called upon the cron firing.

type TaskOption

type TaskOption func(*TaskOptions)

TaskOption represents an option for a synchronized cron task.

func CronExpression

func CronExpression(cronExpression string) TaskOption

CronExpression sets the cron expression of the synchronized cron task. The default is crontask.DefaultCronExpression.

func LeadershipTimeout

func LeadershipTimeout(leadershipTimeout time.Duration) TaskOption

LeadershipTimeout sets the timeout of a single execution of the synchronized cron task. The default is crontask.DefaultLeadershipTimeout.

func LockHeartbeat

func LockHeartbeat(lockHeartbeat time.Duration) TaskOption

LockHeartbeat sets the interval, in which an acquired lock should be renewed (to the total of LeadershipTimeout). This should be smaller than the LockTimeout, otherwise the task will have no chance to ever renew the lock. The default is crontask.DefaultLockHeartbeat.

func LockTimeout

func LockTimeout(lockTimeout time.Duration) TaskOption

LockTimeout sets the timeout for the lock of a single execution of the synchronized cron task. It is good practice to keep the timeout small, for fast failure detection. The default is crontask.DefaultLockTimeout.

func Logger

func Logger(logger *logrus.Logger) TaskOption

Logger sets the logger of the synchronized cron task. The default is the logrus global default logger.

func TaskName

func TaskName(name string) TaskOption

TaskName sets the name of the synchronized cron task. The default is crontask.DefaultName.

type TaskOptions

type TaskOptions struct {
	Name           string
	CronExpression string

	Logger *logrus.Logger

	LeadershipTimeout time.Duration
	LockTimeout       time.Duration
	LockHeartbeat     time.Duration
}

TaskOptions bundles all available configuration properties for a synchronized cron task.