modules

Documentation

Overview

Package modules provides a full module and task management ecosystem to cleanly put all big and small moving parts of a service together.

Modules are started in a multi-stage process and may depend on other modules: - Go's init(): register flags - prep: check flags, register config variables - start: start actual work, access config - stop: gracefully shut down

Workers: A simple function that is run by the module while catching panics and reporting them. Ideal for long running (possibly) idle goroutines. Can be automatically restarted if execution ends with an error.

Tasks: Functions that take somewhere between a couple seconds and a couple minutes to execute and should be queued, scheduled or repeated.

MicroTasks: Functions that take less than a second to execute, but require lots of resources. Running such functions as MicroTasks will reduce concurrent execution and shall improve performance.

Ideally, _any_ execution by a module is done through these methods. This will not only ensure that all panics are caught, but will also give better insights into how your service performs.

Index

• Constants
• Variables
• func GetExitStatusCode() int
• func SetErrorReportingChannel(reportingChannel chan *ModuleError)
• func SetExitStatusCode(n int)
• func SetMaxConcurrentMicroTasks(n int)
• func SetStdErrReporting(on bool)
• func Shutdown() error
• func ShuttingDown() <-chan struct{}
• func Start() error
• func StartCompleted() bool
• func WaitForStartCompletion() <-chan struct{}
• type Module
  • func Register(name string, prep, start, stop func() error, dependencies ...string) *Module
  • func (m *Module) NewErrorMessage(taskName string, err error) *ModuleError
  • func (m *Module) NewInfoMessage(message string) *ModuleError
  • func (m *Module) NewPanicError(taskName, taskType string, panicValue interface{}) *ModuleError
  • func (m *Module) NewTask(name string, fn func(context.Context, *Task)) *Task
  • func (m *Module) ReadyToPrep() bool
  • func (m *Module) ReadyToStart() bool
  • func (m *Module) ReadyToStop() bool
  • func (m *Module) RegisterEvent(event string)
  • func (m *Module) RegisterEventHook(module string, event string, description string, ...) error
  • func (m *Module) RunLowPriorityMicroTask(name *string, fn func(context.Context) error) error
  • func (m *Module) RunMediumPriorityMicroTask(name *string, fn func(context.Context) error) error
  • func (m *Module) RunMicroTask(name *string, fn func(context.Context) error) error
  • func (m *Module) RunWorker(name string, fn func(context.Context) error) error
  • func (m *Module) ShutdownInProgress() bool
  • func (m *Module) ShuttingDown() <-chan struct{}
  • func (m *Module) StartLowPriorityMicroTask(name *string, fn func(context.Context) error)
  • func (m *Module) StartMediumPriorityMicroTask(name *string, fn func(context.Context) error)
  • func (m *Module) StartMicroTask(name *string, fn func(context.Context) error)
  • func (m *Module) StartServiceWorker(name string, backoffDuration time.Duration, fn func(context.Context) error)
  • func (m *Module) StartWorker(name string, fn func(context.Context) error)
  • func (m *Module) TriggerEvent(event string, data interface{})
• type ModuleError
  • func IsPanic(err error) (bool, *ModuleError)
  • func (me *ModuleError) Error() string
  • func (me *ModuleError) Report()
• type Task
  • func (t *Task) Cancel()
  • func (t *Task) MaxDelay(maxDelay time.Duration) *Task
  • func (t *Task) Prioritize() *Task
  • func (t *Task) Queue() *Task
  • func (t *Task) Repeat(interval time.Duration) *Task
  • func (t *Task) Schedule(executeAt time.Time) *Task
  • func (t *Task) StartASAP() *Task

Constants

const (
	DefaultBackoffDuration = 2 * time.Second
)

Worker Default Configuration

Variables

var (

	// ErrCleanExit is returned by Start() when the program is interrupted before starting. This can happen for example, when using the "--help" flag.
	ErrCleanExit = errors.New("clean exit requested")
)

var (
	// ErrRestartNow may be returned (wrapped) by service workers to request an immediate restart.
	ErrRestartNow = errors.New("requested restart")
)

var (
	// HelpFlag triggers printing flag.Usage. It's exported for custom help handling.
	HelpFlag bool
)

Functions

func GetExitStatusCode

func GetExitStatusCode() int

GetExitStatusCode waits for the shutdown to complete and then returns the exit code

func SetErrorReportingChannel

func SetErrorReportingChannel(reportingChannel chan *ModuleError)

SetErrorReportingChannel sets the channel to report module errors through. By default only panics are reported, all other errors need to be manually wrapped into a *ModuleError and reported.

func SetExitStatusCode

func SetExitStatusCode(n int)

SetExitStatusCode sets the exit code that the program shell return to the host after shutdown.

func SetMaxConcurrentMicroTasks

func SetMaxConcurrentMicroTasks(n int)

SetMaxConcurrentMicroTasks sets the maximum number of microtasks that should be run concurrently.

func SetStdErrReporting

func SetStdErrReporting(on bool)

SetStdErrReporting controls error reporting to stderr.

func Shutdown

func Shutdown() error

Shutdown stops all modules in the correct order.

func ShuttingDown

func ShuttingDown() <-chan struct{}

ShuttingDown returns a channel read on the global shutdown signal.

func Start

func Start() error

Start starts all modules in the correct order. In case of an error, it will automatically shutdown again.

func StartCompleted

func StartCompleted() bool

StartCompleted returns whether starting has completed.

func WaitForStartCompletion

func WaitForStartCompletion() <-chan struct{}

WaitForStartCompletion returns as soon as starting has completed.

Types

type Module

type Module struct {
	Name string

	// lifecycle mgmt
	Prepped *abool.AtomicBool
	Started *abool.AtomicBool
	Stopped *abool.AtomicBool

	// shutdown mgmt
	Ctx context.Context
	// contains filtered or unexported fields
}

Module represents a module.

func Register

func Register(name string, prep, start, stop func() error, dependencies ...string) *Module

Register registers a new module. The control functions `prep`, `start` and `stop` are technically optional. `stop` is called _after_ all added module workers finished.

func (*Module) NewErrorMessage

func (m *Module) NewErrorMessage(taskName string, err error) *ModuleError

NewErrorMessage creates a new, reportable, error message (including a stack trace).

func (*Module) NewInfoMessage

func (m *Module) NewInfoMessage(message string) *ModuleError

NewInfoMessage creates a new, reportable, info message (including a stack trace).

func (*Module) NewPanicError

func (m *Module) NewPanicError(taskName, taskType string, panicValue interface{}) *ModuleError

NewPanicError creates a new, reportable, panic error message (including a stack trace).

func (*Module) NewTask

func (m *Module) NewTask(name string, fn func(context.Context, *Task)) *Task

NewTask creates a new task with a descriptive name (non-unique), a optional deadline, and the task function to be executed. You must call one of Queue, Prioritize, StartASAP, Schedule or Repeat in order to have the Task executed.

func (*Module) ReadyToPrep

func (m *Module) ReadyToPrep() bool

ReadyToPrep returns whether all dependencies are ready for this module to prep.

func (*Module) ReadyToStart

func (m *Module) ReadyToStart() bool

ReadyToStart returns whether all dependencies are ready for this module to start.

func (*Module) ReadyToStop

func (m *Module) ReadyToStop() bool

ReadyToStop returns whether all dependencies are ready for this module to stop.

func (*Module) RegisterEvent

func (m *Module) RegisterEvent(event string)

RegisterEvent registers a new event to allow for registering hooks.

func (*Module) RegisterEventHook

func (m *Module) RegisterEventHook(module string, event string, description string, fn func(context.Context, interface{}) error) error

RegisterEventHook registers a hook function with (another) modules' event. Whenever a hook is triggered and the receiving module has not yet fully started, hook execution will be delayed until all modules completed starting.

func (*Module) RunLowPriorityMicroTask

func (m *Module) RunLowPriorityMicroTask(name *string, fn func(context.Context) error) error

RunLowPriorityMicroTask runs a new MicroTask with low priority. It will wait until a slot becomes available (max 15 seconds). The call blocks until finished. The given function will be executed and panics caught. The supplied name must not be changed.

func (*Module) RunMediumPriorityMicroTask

func (m *Module) RunMediumPriorityMicroTask(name *string, fn func(context.Context) error) error

RunMediumPriorityMicroTask runs a new MicroTask with medium priority. It will wait until a slot becomes available (max 3 seconds). The call blocks until finished. The given function will be executed and panics caught. The supplied name must not be changed.

func (*Module) RunMicroTask

func (m *Module) RunMicroTask(name *string, fn func(context.Context) error) error

RunMicroTask runs a new MicroTask with high priority. It will start immediately. The call blocks until finished. The given function will be executed and panics caught. The supplied name must not be changed.

func (*Module) RunWorker

func (m *Module) RunWorker(name string, fn func(context.Context) error) error

RunWorker directly runs a generic worker that does not fit to be a Task or MicroTask, such as long running (and possibly mostly idle) sessions. A call to RunWorker blocks until the worker is finished.

func (*Module) ShutdownInProgress

func (m *Module) ShutdownInProgress() bool

ShutdownInProgress returns whether the module has started shutting down. In most cases, you should use ShuttingDown instead.

func (*Module) ShuttingDown

func (m *Module) ShuttingDown() <-chan struct{}

ShuttingDown lets you listen for the shutdown signal.

func (*Module) StartLowPriorityMicroTask

func (m *Module) StartLowPriorityMicroTask(name *string, fn func(context.Context) error)

StartLowPriorityMicroTask starts a new MicroTask with low priority. The call starts a new goroutine and returns immediately. It will wait until a slot becomes available (max 15 seconds). The given function will be executed and panics caught. The supplied name must not be changed.

func (*Module) StartMediumPriorityMicroTask

func (m *Module) StartMediumPriorityMicroTask(name *string, fn func(context.Context) error)

StartMediumPriorityMicroTask starts a new MicroTask with medium priority. The call starts a new goroutine and returns immediately. It will wait until a slot becomes available (max 3 seconds). The given function will be executed and panics caught. The supplied name must not be changed.

func (*Module) StartMicroTask

func (m *Module) StartMicroTask(name *string, fn func(context.Context) error)

StartMicroTask starts a new MicroTask with high priority. It will start immediately. The call starts a new goroutine and returns immediately. The given function will be executed and panics caught. The supplied name must not be changed.

func (*Module) StartServiceWorker

func (m *Module) StartServiceWorker(name string, backoffDuration time.Duration, fn func(context.Context) error)

StartServiceWorker starts a generic worker, which is automatically restarted in case of an error. A call to StartServiceWorker runs the service-worker in a new goroutine and returns immediately. `backoffDuration` specifies how to long to wait before restarts, multiplied by the number of failed attempts. Pass `0` for the default backoff duration. For custom error remediation functionality, build your own error handling procedure using calls to RunWorker.

func (*Module) StartWorker

func (m *Module) StartWorker(name string, fn func(context.Context) error)

StartWorker directly starts a generic worker that does not fit to be a Task or MicroTask, such as long running (and possibly mostly idle) sessions. A call to StartWorker starts a new goroutine and returns immediately.

func (*Module) TriggerEvent

func (m *Module) TriggerEvent(event string, data interface{})

TriggerEvent executes all hook functions registered to the specified event.

type ModuleError

type ModuleError struct {
	Message string

	ModuleName string
	TaskName   string
	TaskType   string // one of "worker", "task", "microtask" or custom
	Severity   string // one of "info", "error", "panic" or custom

	PanicValue interface{}
	StackTrace string
}

ModuleError wraps a panic, error or message into an error that can be reported.

func IsPanic

func IsPanic(err error) (bool, *ModuleError)

IsPanic returns whether the given error is a wrapped panic by the modules package and additionally returns it, if true.

func (*ModuleError) Error

func (me *ModuleError) Error() string

Error returns the string representation of the error.

func (*ModuleError) Report

func (me *ModuleError) Report()

Report reports the error through the configured reporting channel.

type Task

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

Task is managed task bound to a module.

func (*Task) Cancel

func (t *Task) Cancel()

Cancel cancels the current and any future execution of the Task. This is not reversible by any other functions.

func (*Task) MaxDelay

func (t *Task) MaxDelay(maxDelay time.Duration) *Task

MaxDelay sets a maximum delay within the task should be executed from being queued. Scheduled tasks are queued when they are triggered. The default delay is 3 minutes.

func (*Task) Prioritize

func (t *Task) Prioritize() *Task

Prioritize puts the task in the prioritized queue.

func (*Task) Queue

func (t *Task) Queue() *Task

Queue queues the Task for execution.

func (*Task) Repeat

func (t *Task) Repeat(interval time.Duration) *Task

Repeat sets the task to be executed in endless repeat at the specified interval. First execution will be after interval. Minimum repeat interval is one minute.

func (*Task) Schedule

func (t *Task) Schedule(executeAt time.Time) *Task

Schedule schedules the task for execution at the given time.

func (*Task) StartASAP

func (t *Task) StartASAP() *Task

StartASAP schedules the task to be executed next.