scheduler

Documentation

Index

• type Scheduler
  • func NewScheduler(logger *zap.SugaredLogger, p *pipeline.Pipeline) *Scheduler
  • func (s *Scheduler) GetTaskInstancesByStatus(status TaskInstanceStatus) []*TaskInstance
  • func (s *Scheduler) Kickstart()
  • func (s *Scheduler) MarkAll(status TaskInstanceStatus)
  • func (s *Scheduler) MarkTask(task *pipeline.Task, status TaskInstanceStatus, downstream bool)
  • func (s *Scheduler) MarkTaskInstance(instance *TaskInstance, status TaskInstanceStatus, downstream bool)
  • func (s *Scheduler) Run(ctx context.Context) []*TaskExecutionResult
  • func (s *Scheduler) Tick(result *TaskExecutionResult) bool
  • func (s *Scheduler) WillRunTaskOfType(taskType string) bool
• type TaskExecutionResult
• type TaskInstance
  • func (t *TaskInstance) Completed() bool
  • func (t *TaskInstance) MarkAs(status TaskInstanceStatus)
• type TaskInstanceStatus

Types

type Scheduler

type Scheduler struct {
	WorkQueue chan *TaskInstance
	Results   chan *TaskExecutionResult
	// contains filtered or unexported fields
}

func NewScheduler

func NewScheduler(logger *zap.SugaredLogger, p *pipeline.Pipeline) *Scheduler

func (*Scheduler) GetTaskInstancesByStatus

func (s *Scheduler) GetTaskInstancesByStatus(status TaskInstanceStatus) []*TaskInstance

func (*Scheduler) Kickstart

func (s *Scheduler) Kickstart()

Kickstart initiates the scheduler process by sending a "start" task for the processing.

func (*Scheduler) MarkAll

func (s *Scheduler) MarkAll(status TaskInstanceStatus)

func (*Scheduler) MarkTask

func (s *Scheduler) MarkTask(task *pipeline.Task, status TaskInstanceStatus, downstream bool)

func (*Scheduler) MarkTaskInstance

func (s *Scheduler) MarkTaskInstance(instance *TaskInstance, status TaskInstanceStatus, downstream bool)

func (*Scheduler) Run

func (s *Scheduler) Run(ctx context.Context) []*TaskExecutionResult

func (*Scheduler) Tick

func (s *Scheduler) Tick(result *TaskExecutionResult) bool

Tick marks an iteration of the scheduler loop. It is called when a result is received. The results are mainly fed from a channel, but Tick allows implementing additional methods of passing Task results and simulating scheduler loops, e.g. time travel. It is also useful for testing purposes.

func (*Scheduler) WillRunTaskOfType

func (s *Scheduler) WillRunTaskOfType(taskType string) bool

type TaskExecutionResult

type TaskExecutionResult struct {
	Instance *TaskInstance
	Error    error
}

type TaskInstance

type TaskInstance struct {
	Pipeline *pipeline.Pipeline
	Task     *pipeline.Task
	// contains filtered or unexported fields
}

func (*TaskInstance) Completed

func (t *TaskInstance) Completed() bool

func (*TaskInstance) MarkAs

func (t *TaskInstance) MarkAs(status TaskInstanceStatus)

type TaskInstanceStatus

type TaskInstanceStatus int

const (
	Pending TaskInstanceStatus = iota
	Queued
	Running
	Failed
	UpstreamFailed
	Succeeded
)