scheduler

Documentation

Index

• Constants
• func Bod(t time.Time) time.Time
• type Config
  • func NewConfig() *Config
  • func (c *Config) Validate() error
• type Dispatcher
  • func StartDispatcher(nworkers int, eps map[string]provider.ExecutionProvider, ...) *Dispatcher
  • func (d *Dispatcher) Stop() error
• type Service
  • func NewService(config *Config, logger *logrus.Logger, reporters []adapters.Reporter) *Service
  • func (s *Service) RegisterExecutionProvider(name string, provider provider.ExecutionProvider)
  • func (s *Service) Start() error
  • func (s *Service) Stop() error
  • func (s *Service) Tick(tasks *Tasks) error
• type Tasks
  • func NewSourceConfig(config *Config) *Tasks
  • func (t *Tasks) DebugTasks(log *logrus.Logger)
  • func (t *Tasks) GetTask(name string) (td *provider.Task, err error)
  • func (t *Tasks) Load() error
• type Worker
  • func NewWorker(id int, workerQueue chan chan *provider.Task, ...) Worker
  • func (w *Worker) LogReporters(err error, t *provider.Task, start time.Time, r *provider.Response)
  • func (w *Worker) Start()
  • func (w *Worker) Stop()
  • func (w *Worker) SubmitErr(work *provider.Task, err error)

Constants

const (
	//DefaultTickDuration is default state for metrics
	DefaultTickDuration = 1
)

Functions

func Bod

func Bod(t time.Time) time.Time

Bod returns the beginning of the day

Types

type Config

type Config struct {
	Enabled           bool           `toml:"enabled"`
	TickDuration      toml.Duration  `toml:"tick-duration"`
	LogTasksDetection bool           `toml:"log-task-detection"`
	HTTPExecutor      http.Config    `toml:"executor-http"`
	LocalExecutor     local.Config   `toml:"executor-local"`
	CrontabSource     crontab.Config `toml:"provider-crontab"`
	ETCDSource        etcd.Config    `toml:"provider-etcd"`
	SQLSource         sql.Config     `toml:"provider-sql"`
	NumWorkers        int            `toml:"num-workers"`
}

Config represents the meta configuration.

func NewConfig

func NewConfig() *Config

NewConfig builds a new configuration with default values.

func (*Config) Validate

func (c *Config) Validate() error

Validate validates the configuration

type Dispatcher

type Dispatcher struct {
	WorkerQueue chan chan *provider.Task
	WorkQueue   chan *provider.Task
	Workers     []*Worker
}

Dispatcher for spinning up workers and distributing work

func StartDispatcher

func StartDispatcher(nworkers int, eps map[string]provider.ExecutionProvider, reporters []adapters.Reporter) *Dispatcher

StartDispatcher starts the dispatching provider

func (*Dispatcher) Stop

func (d *Dispatcher) Stop() error

Stop stops all workers

type Service

type Service struct {
	Config             *Config
	Logger             *logrus.Logger
	Cancel             chan struct{}
	Ticker             *time.Ticker
	ExecutionProviders map[string]provider.ExecutionProvider
	Reporters          []adapters.Reporter
	JobDispatcher      *Dispatcher
	TaskRunner         *Tasks
}

Service Returns a config

func NewService

func NewService(config *Config, logger *logrus.Logger, reporters []adapters.Reporter) *Service

NewService creates a new service

func (*Service) RegisterExecutionProvider

func (s *Service) RegisterExecutionProvider(name string, provider provider.ExecutionProvider)

RegisterExecutionProvider registers or overwrites an execution provider

func (*Service) Start

func (s *Service) Start() error

Start starts the actual service

func (*Service) Stop

func (s *Service) Stop() error

Stop cancels any executions immediately

func (*Service) Tick

func (s *Service) Tick(tasks *Tasks) error

Tick runs for a set interval

type Tasks

type Tasks struct {
	BasePath        string
	Tasks           map[string]*provider.Task
	SourceProviders []provider.SourceProvider
}

Tasks store all the structural data needed to run the cronjobs

func NewSourceConfig

func NewSourceConfig(config *Config) *Tasks

NewSourceConfig returns a new Task archive

func (*Tasks) DebugTasks

func (t *Tasks) DebugTasks(log *logrus.Logger)

DebugTasks dumps all tasks inside the task archive

func (*Tasks) GetTask

func (t *Tasks) GetTask(name string) (td *provider.Task, err error)

GetTask returns a single task definition

func (*Tasks) Load

func (t *Tasks) Load() error

Load loads tasks from etcd

type Worker

type Worker struct {
	ID                 int
	Work               chan *provider.Task
	WorkerQueue        chan chan *provider.Task
	Reporters          []adapters.Reporter
	QuitChan           chan bool
	ExecutionProviders map[string]provider.ExecutionProvider
	Callbacks          map[string]func(*provider.Task) error
}

Worker defines a working horse for tasks

func NewWorker

func NewWorker(id int, workerQueue chan chan *provider.Task, eps map[string]provider.ExecutionProvider, reporters []adapters.Reporter) Worker

NewWorker creates, and returns a new Worker object. Its only argument is a channel that the worker can add itself to whenever it is done its work.

func (*Worker) LogReporters

func (w *Worker) LogReporters(err error, t *provider.Task, start time.Time, r *provider.Response)

LogReporters fans out to reporters.

func (*Worker) Start

func (w *Worker) Start()

Start function "starts" the worker by starting a goroutine, that is an infinite "for-select" loop.

func (*Worker) Stop

func (w *Worker) Stop()

Stop tells the worker to stop listening for work requests.

Note that the worker will only stop *after* it has finished its work.

func (*Worker) SubmitErr

func (w *Worker) SubmitErr(work *provider.Task, err error)

SubmitErr delivers an error to a source provider