executor

Documentation

Index

• Constants
• Variables
• func IsReboot(exec interface{ ... }) bool
• func JoinSchedules(execs []*Executable) map[*cronlex.Schedule][]*Executable
• func Multi(ctx context.Context, now time.Time, execs ...Executor) error
• func NoOp() noOpExecutor
• func WithLogHandler(handler slog.Handler) cfg.Option[*Executable]
• func WithLogger(logger *slog.Logger) cfg.Option[*Executable]
• func WithMetrics(m Metrics) cfg.Option[*Executable]
• func WithSchedule(cron string, loc *time.Location) cfg.Option[*Executable]
• func WithScheduler(sched Scheduler) cfg.Option[*Executable]
• func WithTimeout(dur time.Duration) cfg.Option[*Executable]
• func WithTrace(tracer trace.Tracer) cfg.Option[*Executable]
• type Executable
  • func New(id string, runners []Runner, options ...cfg.Option[*Executable]) (*Executable, error)
  • func (e *Executable) Exec(ctx context.Context, now time.Time) error
  • func (e *Executable) ID() string
  • func (e *Executable) Next(ctx context.Context, now time.Time) time.Duration
• type Executor
• type Metrics
• type Runnable
  • func (r Runnable) Run(ctx context.Context) error
• type Runner
• type Scheduler

Constants

const (
	ErrEmpty           = errs.Kind("empty")
	ErrAlreadyExecuted = errs.Kind("already executed")

	ErrRunnerList = errs.Entity("runners list")
	ErrScheduler  = errs.Entity("scheduler")
	ErrReboot     = errs.Entity("reboot")
)

Variables

var (
	ErrEmptyRunnerList       = errs.WithDomain(errDomain, ErrEmpty, ErrRunnerList)
	ErrEmptyScheduler        = errs.WithDomain(errDomain, ErrEmpty, ErrScheduler)
	ErrAlreadyExecutedReboot = errs.WithDomain(errDomain, ErrAlreadyExecuted, ErrReboot)
)

Functions

func IsReboot

func IsReboot(exec interface {
	Exec(ctx context.Context, now time.Time) error
	Next(ctx context.Context, now time.Time) time.Duration
	ID() string
}) bool

func JoinSchedules

func JoinSchedules(execs []*Executable) map[*cronlex.Schedule][]*Executable

func Multi

func Multi(ctx context.Context, now time.Time, execs ...Executor) error

Multi calls the Executor.Exec on a set of Executor that are meant to be executed at the same time (e.g. when scheduled for the same time).

This call will execute the input (set of) action(s) in parallel, collecting any error(s) that the Executor(s) raises.

The returned error is a joined error, for any failing executions. The executions are synchronized in a sync.WaitGroup, and are bound to the input context.Context's lifetime.

func NoOp

func NoOp() noOpExecutor

NoOp returns a no-op Executor.

func WithLogHandler

func WithLogHandler(handler slog.Handler) cfg.Option[*Executable]

WithLogHandler decorates the Executor with logging using the input log handler.

func WithLogger

func WithLogger(logger *slog.Logger) cfg.Option[*Executable]

WithLogger decorates the Executor with the input logger.

func WithMetrics

func WithMetrics(m Metrics) cfg.Option[*Executable]

WithMetrics decorates the Executor with the input metrics registry.

func WithSchedule

func WithSchedule(cron string, loc *time.Location) cfg.Option[*Executable]

WithSchedule configures the Executor with a schedule.Scheduler using the input cron string.

This call returns a cfg.NoOp cfg.Option if the cron string is empty.

This option can be followed by a WithLocation option.

func WithScheduler

func WithScheduler(sched Scheduler) cfg.Option[*Executable]

WithScheduler configures the Executor with the input schedule.Scheduler.

This call returns a cfg.NoOp cfg.Option if the input schedule.Scheduler is either nil or a no-op.

Using this option does not require passing WithSchedule nor WithLocation options.

func WithTimeout

func WithTimeout(dur time.Duration) cfg.Option[*Executable]

func WithTrace

func WithTrace(tracer trace.Tracer) cfg.Option[*Executable]

WithTrace decorates the Executor with the input trace.Tracer.

Types

type Executable

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

Executable is an implementation of the Executor interface. It uses a schedule.Scheduler to mark the next job's execution time, and supports multiple Runner.

func New

func New(id string, runners []Runner, options ...cfg.Option[*Executable]) (*Executable, error)

New creates an Executor with the input cfg.Option(s), also returning an error if raised.

The minimum requirements to create an Executor is to supply at least one Runner, be it an implementation of this interface or as a Runnable using the WithRunners option, as well as a schedule.Scheduler using the WithScheduler option -- alternatively, callers can simply pass a cron string directly using the WithSchedule option.

If an ID is not supplied, then the default ID of `micron.executor` is set.

func (*Executable) Exec

func (e *Executable) Exec(ctx context.Context, now time.Time) error

Exec runs the task when on its scheduled time.

For this, Exec leverages the Executor's underlying schedule.Scheduler to retrieve the job's next execution time, waits for it, and calls Runner.Run on each configured Runner. All raised errors are joined and returned at the end of this call.

func (*Executable) ID

func (e *Executable) ID() string

ID returns this Executor's ID.

func (*Executable) Next

func (e *Executable) Next(ctx context.Context, now time.Time) time.Duration

Next calls the Executor's underlying schedule.Scheduler Next method.

type Executor

type Executor interface {
	// Exec runs the task when on its scheduled time.
	//
	// For this, Exec leverages the Executor's underlying schedule.Scheduler to retrieve the job's next execution time,
	// waits for it, and calls Runner.Run on each configured Runner. All raised errors are joined and returned at the end
	// of this call.
	Exec(ctx context.Context, now time.Time) error
	// Next calls the Executor's underlying schedule.Scheduler Next method.
	Next(ctx context.Context, now time.Time) time.Duration
	// ID returns this Executor's ID.
	ID() string
}

Executor describes the capabilities of cron job's executor component, which is based on fetching the next execution's time, Next; as well as running the job, Exec. It also exposes an ID method to allow access to this Executor's configured ID or name.

Implementations of Executor must focus on the logic of the Exec method, which should contain the logic of the Next method as well. It should not be the responsibility of other components to wait until it is time to execute the job; but actually the Executor's responsibility to consider it in its Exec method. That being said, its Next method (just like its ID method) allows access to some of the details of the executor if the caller needs that information; as helpers.

The logic behind Next and generally calculating the time for the next job execution should be deferred to a schedule.Scheduler, which should be part of the Executor.

One Executor may contain multiple Runner, as a job may be composed of several (smaller) tasks. However, an Executor is identified by a single ID.

type Metrics

type Metrics interface {
	// IncExecutorExecCalls increases the count of Exec calls, by the Executor.
	IncExecutorExecCalls(ctx context.Context, id string)
	// IncExecutorExecErrors increases the count of Exec call errors, by the Executor.
	IncExecutorExecErrors(ctx context.Context, id string)
	// ObserveExecLatency registers the duration of an Exec call, by the Executor.
	ObserveExecLatency(ctx context.Context, id string, dur time.Duration)
	// IncExecutorNextCalls increases the count of Next calls, by the Executor.
	IncExecutorNextCalls(ctx context.Context, id string)
}

Metrics describes the actions that register Executor-related metrics.

type Runnable

type Runnable func(ctx context.Context) error

Runnable is a custom type for any function that takes in a context.Context and returns an error. This type of function can be perceived as a Runner type. For that, this custom type will implement Runner by exposing a Run method that invokes the actual Runnable function.

func (Runnable) Run

func (r Runnable) Run(ctx context.Context) error

Run executes the job or task.

This call takes in a context.Context which may be used to denote cancellation or closure (e.g. with a timeout)

The returned error denotes the success state of the execution. A nil error means that the execution was successful, where a non-nil error must signal a failed execution.

type Runner

type Runner interface {
	// Run executes the job or task.
	//
	// This call takes in a context.Context which may be used to denote cancellation or closure (e.g. with a timeout)
	//
	// The returned error denotes the success state of the execution. A nil error means that the execution was successful,
	// where a non-nil error must signal a failed execution.
	Run(ctx context.Context) error
}

Runner describes a type that executes a job or task. It contains only one method, Run, that is called with a context as input and returns an error.

Implementations of Runner only need to comply with this method, where the logic within Run is completely up to the actual implementation. These implementations need to be aware of the state of the input context.Context, which may denote cancellation or closure (e.g. with a timeout).

The returned error denotes the success state of the execution. A nil error means that the execution was successful, where a non-nil error must signal a failed execution.

type Scheduler

type Scheduler interface {
	// Next calculates and returns the following scheduled time, from the input time.Time.
	Next(ctx context.Context, now time.Time) time.Time
}

Scheduler describes the capabilities of a cron job scheduler. Its sole responsibility is to provide the timestamp for the next job's execution, after calculating its frequency from its configuration.

Scheduler exposes one method, Next, that takes in a context.Context and a time.Time. It is implied that the input time is the time.Now value, however it is open to any input that the caller desires to pass to it. The returned time.Time value must always be the following occurrence according to the schedule, in the context of the input time.

Implementations of Next should take into consideration the cron specification; however the interface allows a custom approach to the scheduler, especially if added functionality is necessary (e.g. frequency overriding schedulers, dynamic frequencies, and pipeline-approaches where the frequency is evaluated after a certain check).