taskctl

Documentation

Overview

Package taskctl contains custom implementations and extensions for the github.com/taskctl/taskctl packages

Index

• Constants
• type FileOutputStore
  • func NewOutputStore(path string) (*FileOutputStore, error)
  • func (s *FileOutputStore) Reader(jobID string, taskName string, outputName string) (io.ReadCloser, error)
  • func (s *FileOutputStore) Remove(jobID string) error
  • func (s *FileOutputStore) Writer(jobID string, taskName string, outputName string) (io.WriteCloser, error)
• type LineWriter
  • func (l *LineWriter) Finish()
  • func (l *LineWriter) Lines() [][]byte
  • func (l *LineWriter) Write(p []byte) (n int, err error)
• type Opts
  • func WithContexts(contexts map[string]*runner.ExecutionContext) Opts
  • func WithEnv(env variables.Container) Opts
  • func WithKillTimeout(killTimeout time.Duration) Opts
  • func WithVariables(variables variables.Container) Opts
• type OutputStore
• type PgidExecutor
  • func NewPgidExecutor(stdin io.Reader, stdout, stderr io.Writer, killTimeout time.Duration) (*PgidExecutor, error)
  • func (e *PgidExecutor) Execute(ctx context.Context, job *executor.Job) ([]byte, error)
• type Runner
• type Scheduler
  • func NewScheduler(r runner.Runner) *Scheduler
  • func (s *Scheduler) Cancel()
  • func (s *Scheduler) Finish()
  • func (s *Scheduler) OnStageChange(f func(stage *scheduler.Stage))
  • func (s *Scheduler) Schedule(g *scheduler.ExecutionGraph) error
• type TaskRunner
  • func NewTaskRunner(outputStore OutputStore, opts ...Opts) (*TaskRunner, error)
  • func (r *TaskRunner) Cancel()
  • func (r *TaskRunner) Finish()
  • func (r *TaskRunner) Run(t *task.Task) error
  • func (r *TaskRunner) SetContexts(contexts map[string]*runner.ExecutionContext) *TaskRunner
  • func (r *TaskRunner) SetOnTaskChange(f func(t *task.Task))
  • func (r *TaskRunner) SetVariables(vars variables.Container) *TaskRunner
  • func (r *TaskRunner) WithVariable(key, value string) *TaskRunner

Examples

• Scheduler.Schedule
• TaskRunner.Run

Constants

const JobIDVariableName = "__jobID"

Types

type FileOutputStore

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

func NewOutputStore

func NewOutputStore(path string) (*FileOutputStore, error)

func (*FileOutputStore) Reader

func (s *FileOutputStore) Reader(jobID string, taskName string, outputName string) (io.ReadCloser, error)

func (*FileOutputStore) Remove

func (s *FileOutputStore) Remove(jobID string) error

func (*FileOutputStore) Writer

func (s *FileOutputStore) Writer(jobID string, taskName string, outputName string) (io.WriteCloser, error)

type LineWriter

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

LineWriter splits written data into lines This is not used yet but prepared for streaming output later.

func (*LineWriter) Finish

func (l *LineWriter) Finish()

func (*LineWriter) Lines

func (l *LineWriter) Lines() [][]byte

func (*LineWriter) Write

func (l *LineWriter) Write(p []byte) (n int, err error)

type Opts

type Opts func(*TaskRunner)

Opts is a task runner configuration function.

func WithContexts

func WithContexts(contexts map[string]*runner.ExecutionContext) Opts

WithContexts adds provided contexts to task runner

func WithEnv

func WithEnv(env variables.Container) Opts

WithEnv adds provided environment variables to task runner

func WithKillTimeout

func WithKillTimeout(killTimeout time.Duration) Opts

WithKillTimeout sets the kill timeout for execution of commands

func WithVariables

func WithVariables(variables variables.Container) Opts

WithVariables adds provided variables to task runner

type OutputStore

type OutputStore interface {
	Writer(jobID string, taskName string, outputName string) (io.WriteCloser, error)
	Reader(jobID string, taskName string, outputName string) (io.ReadCloser, error)
	Remove(jobID string) error
}

type PgidExecutor

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

PgidExecutor is an executor that starts commands in a new process group (if supported by OS)

It is based on executor.DefaultExecutor but uses an exec handler that sets setpgid to start child processes in a new process group (with negative pid). It is used to prevent signals from propagating to children and to kill all descendant processes e.g. when forked by a shell process.

func NewPgidExecutor

func NewPgidExecutor(stdin io.Reader, stdout, stderr io.Writer, killTimeout time.Duration) (*PgidExecutor, error)

NewPgidExecutor creates new pgid executor

func (*PgidExecutor) Execute

func (e *PgidExecutor) Execute(ctx context.Context, job *executor.Job) ([]byte, error)

Execute executes given job with provided context Returns job output

type Runner

type Runner interface {
	runner.Runner
	SetOnTaskChange(func(t *task.Task))
}

Runner extends runner.Runner (from taskctl) to implement additional features: - storage of outputs - callback on finished task (so we can f.e. schedule the next task when one is on the wait-queue) The file https://github.com/taskctl/taskctl/blob/master/pkg/runner/runner.go is the original basis of this file.

type Scheduler

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

Scheduler executes ExecutionGraph

func NewScheduler

func NewScheduler(r runner.Runner) *Scheduler

NewScheduler create new Scheduler instance

func (*Scheduler) Cancel

func (s *Scheduler) Cancel()

Cancel cancels executing tasks

func (*Scheduler) Finish

func (s *Scheduler) Finish()

Finish finishes scheduler's TaskRunner

func (*Scheduler) OnStageChange

func (s *Scheduler) OnStageChange(f func(stage *scheduler.Stage))

func (*Scheduler) Schedule

func (s *Scheduler) Schedule(g *scheduler.ExecutionGraph) error

Schedule starts execution of the given ExecutionGraph

Modified to notify on stage changes

Example

format := task.FromCommands("go fmt ./...")
build := task.FromCommands("go build ./..")
r, _ := runner.NewTaskRunner()
s := NewScheduler(r)

graph, err := scheduler.NewExecutionGraph(
	&scheduler.Stage{Name: "format", Task: format},
	&scheduler.Stage{Name: "build", Task: build, DependsOn: []string{"format"}},
)
if err != nil {
	return
}

err = s.Schedule(graph)
if err != nil {
	fmt.Println(err)
}

type TaskRunner

type TaskRunner struct {
	Stdin          io.Reader
	Stdout, Stderr io.Writer
	OutputFormat   string
	// contains filtered or unexported fields
}

TaskRunner run tasks

func NewTaskRunner

func NewTaskRunner(outputStore OutputStore, opts ...Opts) (*TaskRunner, error)

NewTaskRunner creates new TaskRunner instance

func (*TaskRunner) Cancel

func (r *TaskRunner) Cancel()

Cancel cancels execution

func (*TaskRunner) Finish

func (r *TaskRunner) Finish()

Finish makes cleanup tasks over contexts

func (*TaskRunner) Run

func (r *TaskRunner) Run(t *task.Task) error

Run run provided task -> highly modified from taskctl/runner/runner.go TaskRunner first compiles task into linked list of Jobs, then passes those jobs to Executor

Example

t := task.FromCommands("go fmt ./...", "go build ./..")
r, err := NewTaskRunner(nil)
if err != nil {
	return
}
err = r.Run(t)
if err != nil {
	fmt.Println(err, t.ExitCode, t.ErrorMessage())
}
fmt.Println(t.Output())

func (*TaskRunner) SetContexts

func (r *TaskRunner) SetContexts(contexts map[string]*runner.ExecutionContext) *TaskRunner

SetContexts sets task runner's contexts

func (*TaskRunner) SetOnTaskChange

func (r *TaskRunner) SetOnTaskChange(f func(t *task.Task))

SetOnTaskChange is synchronous; so you can read the task in any way inside the callback because there is no concurrent access of the task

func (*TaskRunner) SetVariables

func (r *TaskRunner) SetVariables(vars variables.Container) *TaskRunner

SetVariables sets task runner's variables

func (*TaskRunner) WithVariable

func (r *TaskRunner) WithVariable(key, value string) *TaskRunner

WithVariable adds variable to task runner's variables list. It creates new instance of variables container.