scheduler

Documentation

Index

• Variables
• type Config
• type ExecutionGraph
  • func NewExecutionGraph(lg logger.Logger, steps ...dag.Step) (*ExecutionGraph, error)
  • func NewExecutionGraphForRetry(lg logger.Logger, nodes ...*Node) (*ExecutionGraph, error)
  • func (g *ExecutionGraph) Duration() time.Duration
  • func (g *ExecutionGraph) Finish()
  • func (g *ExecutionGraph) FinishAt() time.Time
  • func (g *ExecutionGraph) IsFinished() bool
  • func (g *ExecutionGraph) IsRunning() bool
  • func (g *ExecutionGraph) IsStarted() bool
  • func (g *ExecutionGraph) NodeData() []NodeData
  • func (g *ExecutionGraph) Nodes() []*Node
  • func (g *ExecutionGraph) Start()
  • func (g *ExecutionGraph) StartAt() time.Time
• type Node
  • func NewNode(step dag.Step, state NodeState) *Node
  • func (n *Node) Data() NodeData
  • func (n *Node) Execute(ctx context.Context) error
  • func (n *Node) SetError(err error)
  • func (n *Node) State() NodeState
• type NodeData
• type NodeState
• type NodeStatus
  • func (s NodeStatus) String() string
• type Scheduler
  • func New(cfg *Config) *Scheduler
  • func (sc *Scheduler) Cancel(g *ExecutionGraph)
  • func (sc *Scheduler) HandlerNode(name dag.HandlerType) *Node
  • func (sc *Scheduler) Schedule(ctx context.Context, g *ExecutionGraph, done chan *Node) error
  • func (sc *Scheduler) Signal(g *ExecutionGraph, sig os.Signal, done chan bool, allowOverride bool)
  • func (sc *Scheduler) Status(g *ExecutionGraph) Status
• type Status
  • func (s Status) String() string

Variables

var (
	ErrWorkingDirNotExist = fmt.Errorf("working directory does not exist")
)

Types

type Config

type Config struct {
	LogDir        string
	Logger        logger.Logger
	MaxActiveRuns int
	Timeout       time.Duration
	Delay         time.Duration
	Dry           bool
	OnExit        *dag.Step
	OnSuccess     *dag.Step
	OnFailure     *dag.Step
	OnCancel      *dag.Step
	ReqID         string
}

type ExecutionGraph

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

ExecutionGraph represents a graph of steps.

func NewExecutionGraph

func NewExecutionGraph(lg logger.Logger, steps ...dag.Step) (*ExecutionGraph, error)

NewExecutionGraph creates a new execution graph with the given steps.

func NewExecutionGraphForRetry

func NewExecutionGraphForRetry(lg logger.Logger, nodes ...*Node) (*ExecutionGraph, error)

NewExecutionGraphForRetry creates a new execution graph for retry with given nodes.

func (*ExecutionGraph) Duration

func (g *ExecutionGraph) Duration() time.Duration

Duration returns the duration of the execution.

func (*ExecutionGraph) Finish

func (g *ExecutionGraph) Finish()

func (*ExecutionGraph) FinishAt

func (g *ExecutionGraph) FinishAt() time.Time

func (*ExecutionGraph) IsFinished

func (g *ExecutionGraph) IsFinished() bool

func (*ExecutionGraph) IsRunning

func (g *ExecutionGraph) IsRunning() bool

func (*ExecutionGraph) IsStarted

func (g *ExecutionGraph) IsStarted() bool

func (*ExecutionGraph) NodeData

func (g *ExecutionGraph) NodeData() []NodeData

func (*ExecutionGraph) Nodes

func (g *ExecutionGraph) Nodes() []*Node

Nodes returns the nodes of the execution graph.

func (*ExecutionGraph) Start

func (g *ExecutionGraph) Start()

func (*ExecutionGraph) StartAt

func (g *ExecutionGraph) StartAt() time.Time

type Node

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

Node is a node in a DAG. It executes a command.

func NewNode

func NewNode(step dag.Step, state NodeState) *Node

func (*Node) Data

func (n *Node) Data() NodeData

func (*Node) Execute

func (n *Node) Execute(ctx context.Context) error

Execute runs the command synchronously and returns error if any.

func (*Node) SetError

func (n *Node) SetError(err error)

func (*Node) State

func (n *Node) State() NodeState

type NodeData

type NodeData struct {
	Step  dag.Step
	State NodeState
}

type NodeState

type NodeState struct {
	Status     NodeStatus
	Log        string
	StartedAt  time.Time
	FinishedAt time.Time
	RetryCount int
	RetriedAt  time.Time
	DoneCount  int
	Error      error
}

NodeState contains the state of a node.

type NodeStatus

type NodeStatus int

NodeStatus represents the status of a node.

const (
	NodeStatusNone NodeStatus = iota
	NodeStatusRunning
	NodeStatusError
	NodeStatusCancel
	NodeStatusSuccess
	NodeStatusSkipped
)

func (NodeStatus) String

func (s NodeStatus) String() string

type Scheduler

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

Scheduler is a scheduler that runs a graph of steps.

func New

func New(cfg *Config) *Scheduler

func (*Scheduler) Cancel

func (sc *Scheduler) Cancel(g *ExecutionGraph)

Cancel sends -1 signal to all nodes.

func (*Scheduler) HandlerNode

func (sc *Scheduler) HandlerNode(name dag.HandlerType) *Node

HandlerNode returns the handler node with the given name.

func (*Scheduler) Schedule

func (sc *Scheduler) Schedule(ctx context.Context, g *ExecutionGraph, done chan *Node) error

Schedule runs the graph of steps.

func (*Scheduler) Signal

func (sc *Scheduler) Signal(
	g *ExecutionGraph, sig os.Signal, done chan bool, allowOverride bool,
)

Signal sends a signal to the scheduler. for a node with repeat policy, it does not stop the node and wait to finish current run.

func (*Scheduler) Status

func (sc *Scheduler) Status(g *ExecutionGraph) Status

Status returns the status of the scheduler.

type Status

type Status int

const (
	StatusNone Status = iota
	StatusRunning
	StatusError
	StatusCancel
	StatusSuccess
)

func (Status) String

func (s Status) String() string