job

Documentation

Overview

Package job provides tools and generic implementations of jobs for amboy Queues.

Base Metadata

The Base type provides an implementation of the amboy.Job interface that does *not* have a Run method, and can be embedded in your own job implementations to avoid implemented duplicated common functionality. The type also implements several methods which are not part of the Job interface for error handling (e.g. AddError and HasErrors), and methods for marking tasks complete and setting the ID (e.g. MarkComplete and SetID).

All job implementations should use this functionality, although there are some situations where jobs may want independent implementation of the Job interface, including: easier construction for use from the REST interface, needing or wanting a more constrained public interface, or needing more constrained options for some values (e.g. Dependency, Priority).

Index

• func GetNumber() int
• func RegisterDefaultJobs()
• type Base
  • func (b *Base) AddError(err error)
  • func (b *Base) Dependency() dependency.Manager
  • func (b *Base) Error() error
  • func (b *Base) HasErrors() bool
  • func (b *Base) ID() string
  • func (b *Base) Lock(id string, lockTimeout time.Duration) error
  • func (b *Base) MarkComplete()
  • func (b *Base) Priority() int
  • func (b *Base) Scopes() []string
  • func (b *Base) SetDependency(d dependency.Manager)
  • func (b *Base) SetID(n string)
  • func (b *Base) SetPriority(p int)
  • func (b *Base) SetScopes(scopes []string)
  • func (b *Base) SetStatus(s amboy.JobStatusInfo)
  • func (b *Base) Status() amboy.JobStatusInfo
  • func (b *Base) TimeInfo() amboy.JobTimeInfo
  • func (b *Base) Type() amboy.JobType
  • func (b *Base) Unlock(id string, lockTimeout time.Duration)
  • func (b *Base) UpdateTimeInfo(i amboy.JobTimeInfo)
• type Group
  • func NewGroup(name string) *Group
  • func (g *Group) Add(j amboy.Job) error
  • func (g *Group) Run(ctx context.Context)
  • func (g *Group) SetDependency(d dependency.Manager)
• type ShellJob
  • func NewShellJob(cmd string, creates string) *ShellJob
  • func NewShellJobInstance() *ShellJob
  • func (j *ShellJob) Run(ctx context.Context)

Functions

func GetNumber

func GetNumber() int

GetNumber is a source of safe monotonically increasing integers for use in Job ids.

func RegisterDefaultJobs

func RegisterDefaultJobs()

RegisterDefaultJobs registers all default job types in the amboy Job registry which permits their use in contexts that require serializing jobs to or from a common format (e.g. queues that persist pending and completed jobs outside of the process,) or the REST interface.

In most applications these registrations happen automatically in the context of package init() functions, but for the default/generic jobs, users must explicitly load them into the registry.

Types

type Base

type Base struct {
	TaskID         string        `bson:"name" json:"name" yaml:"name"`
	JobType        amboy.JobType `bson:"job_type" json:"job_type" yaml:"job_type"`
	RequiredScopes []string      `bson:"required_scopes" json:"required_scopes" yaml:"required_scopes"`
	// contains filtered or unexported fields
}

Base is a type that all new checks should compose, and provides an implementation of most common Job methods which most jobs need not implement themselves.

func (*Base) AddError

func (b *Base) AddError(err error)

AddError takes an error object and if it is non-nil, tracks it internally. This operation is thread safe, but not part of the Job interface.

func (*Base) Dependency

func (b *Base) Dependency() dependency.Manager

Dependency returns an amboy Job dependency interface object, and is a component of the Job interface.

func (*Base) Error

func (b *Base) Error() error

Error returns all of the error objects produced by the job.

func (*Base) HasErrors

func (b *Base) HasErrors() bool

HasErrors checks the stored errors in the object and reports if there are any stored errors. This operation is thread safe, but not part of the Job interface.

func (*Base) ID

func (b *Base) ID() string

ID returns the name of the job, and is a component of the Job interface.

func (*Base) Lock

func (b *Base) Lock(id string, lockTimeout time.Duration) error

Lock allows pools to modify the state of a job before saving it to the queue to take the lock. The value of the argument should uniquely identify the runtime instance of the queue that holds the lock, and the method returns an error if the lock cannot be acquired.

func (*Base) MarkComplete

func (b *Base) MarkComplete()

MarkComplete signals that the job is complete, and is not part of the Job interface.

func (*Base) Priority

func (b *Base) Priority() int

Priority returns the priority value, and is part of the amboy.Job interface.

func (*Base) Scopes

func (b *Base) Scopes() []string

Scopes returns the required scopes for the job.

func (*Base) SetDependency

func (b *Base) SetDependency(d dependency.Manager)

SetDependency allows you to inject a different Job dependency object, and is a component of the Job interface.

func (*Base) SetID

func (b *Base) SetID(n string)

SetID makes it possible to change the ID of an amboy.Job. It is not part of the amboy.Job interface.

func (*Base) SetPriority

func (b *Base) SetPriority(p int)

SetPriority allows users to set the priority of a job, and is part of the amboy.Job interface.

func (*Base) SetScopes

func (b *Base) SetScopes(scopes []string)

SetScopes overrides the jobs current scopes with those from the argument. To unset scopes, pass nil to this method.

func (*Base) SetStatus

func (b *Base) SetStatus(s amboy.JobStatusInfo)

SetStatus resets the Status object of a Job document without. It is part of the Job interface and used by remote queues.

func (*Base) Status

func (b *Base) Status() amboy.JobStatusInfo

Status returns the current state of the job including information useful for locking for compatibility with remote queues that require managing exclusive access to a job.

func (*Base) TimeInfo

func (b *Base) TimeInfo() amboy.JobTimeInfo

TimeInfo returns the job's TimeInfo object. The runner implementations are responsible for updating these values.

func (*Base) Type

func (b *Base) Type() amboy.JobType

Type returns the JobType specification for this object, and is a component of the Job interface.

func (*Base) Unlock

func (b *Base) Unlock(id string, lockTimeout time.Duration)

Unlock attempts to remove the current lock state in the job, if possible.

func (*Base) UpdateTimeInfo

func (b *Base) UpdateTimeInfo(i amboy.JobTimeInfo)

UpdateTimeInfo updates the stored value of time in the job, but does *not* modify fields that are unset in the input document.

type Group

type Group struct {
	Jobs  map[string]*registry.JobInterchange `bson:"jobs" json:"jobs" yaml:"jobs"`
	*Base `bson:"metadata" json:"metadata" yaml:"metadata"`
	// contains filtered or unexported fields
}

Group is a structure for running collections of Job objects at the same time, as a single Job. Use Groups to isolate several Jobs from other Jobs in the queue, and ensure that several Jobs run on a single system.

func NewGroup

func NewGroup(name string) *Group

NewGroup creates a new, empty Group object.

func (*Group) Add

func (g *Group) Add(j amboy.Job) error

Add is not part of the Job interface, but allows callers to append jobs to the Group. Returns an error if a job with the same ID() value already exists in the group.

func (*Group) Run

func (g *Group) Run(ctx context.Context)

Run executes the jobs. Provides "continue on error" semantics for Jobs in the Group. Returns an error if: the Group has already run, or if any of the constituent Jobs produce an error *or* if there are problems with the JobInterchange converters.

func (*Group) SetDependency

func (g *Group) SetDependency(d dependency.Manager)

SetDependency allows you to configure the dependency.Manager instance for this object. If you want to swap different dependency instances you can as long as the new instance is of the "Always" type.

type ShellJob

type ShellJob struct {
	Command    string            `bson:"command" json:"command" yaml:"command"`
	Output     string            `bson:"output" json:"output" yaml:"output"`
	WorkingDir string            `bson:"working_dir" json:"working_dir" yaml:"working_dir"`
	Env        map[string]string `bson:"env" json:"env" yaml:"env"`

	Base `bson:"job_base" json:"job_base" yaml:"job_base"`
}

ShellJob is an amboy.Job implementation that runs shell commands in the context of an amboy.Job object.

func NewShellJob

func NewShellJob(cmd string, creates string) *ShellJob

NewShellJob takes the command, as a string along with the name of a file that the command would create, and returns a pointer to a ShellJob object. If the "creates" argument is an empty string then the command always runs, otherwise only if the file specified does not exist. You can change the dependency with the SetDependency argument.

func NewShellJobInstance

func NewShellJobInstance() *ShellJob

NewShellJobInstance returns a pointer to an initialized ShellJob instance, but does not set the command or the name. Use when the command is not known at creation time.

func (*ShellJob) Run

func (j *ShellJob) Run(ctx context.Context)

Run executes the shell commands. Add keys to the Env map to modify the environment, or change the value of the WorkingDir property to set the working directory for this command. Captures output into the Output attribute, and returns the error value of the command.