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. HasErrors), and methods for marking jobs complete (e.g. MarkComplete) and setting the ID (e.g. 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).

Index

• func GetNumber() int
• func RegisterDefaultJobs()
• type Base
  • func (b *Base) AddError(err error)
  • func (b *Base) AddRetryableError(err error)
  • func (b *Base) Dependency() dependency.Manager
  • func (b *Base) EnqueueAllScopes() bool
  • func (b *Base) EnqueueScopes() []string
  • func (b *Base) Error() error
  • func (b *Base) HasErrors() bool
  • func (b *Base) ID() string
  • func (b *Base) IsLastAttempt() bool
  • func (b *Base) Lock(id string, lockTimeout time.Duration) error
  • func (b *Base) MarkComplete()
  • func (b *Base) RetryInfo() amboy.JobRetryInfo
  • func (b *Base) Scopes() []string
  • func (b *Base) SetDependency(d dependency.Manager)
  • func (b *Base) SetEnqueueAllScopes(val bool)
  • func (b *Base) SetEnqueueScopes(scopes ...string)
  • func (b *Base) SetID(n string)
  • func (b *Base) SetScopes(scopes []string)
  • func (b *Base) SetStatus(s amboy.JobStatusInfo)
  • func (b *Base) SetTimeInfo(i amboy.JobTimeInfo)
  • 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) UpdateRetryInfo(opts amboy.JobRetryOptions)
  • 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 {
	Name           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.

func (*Base) AddRetryableError

func (b *Base) AddRetryableError(err error)

AddRetryableError takes an error object and if it is non-nil, tracks it internally and marks the job as needing to retry. This operation is thread safe.

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. If no dependency manager has been explicitly set, its default value is the always manager.

func (*Base) EnqueueAllScopes

func (b *Base) EnqueueAllScopes() bool

EnqueueAllScopes returns whether all of the job's scopes are applied on enqueue. If false, the subset of scopes that are selected to apply on enqueue with SetEnqueueScopes will apply on enqueue; the remaining scopes will apply on dispatch.

func (*Base) EnqueueScopes

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

EnqueueScopes returns the subset of the job's scopes that will be applied on enqueue, if any.

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) IsLastAttempt

func (b *Base) IsLastAttempt() bool

IsLastAttempt determines if this is the final attempt of a retryable job. If it's not retryable, this always return true.

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) RetryInfo

func (b *Base) RetryInfo() amboy.JobRetryInfo

RetryInfo returns information and options for the job's retry policies.

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. If the given dependency manager is nil, it will default to the always manager.

func (*Base) SetEnqueueAllScopes

func (b *Base) SetEnqueueAllScopes(val bool)

SetEnqueueAllScopes overrides the default behavior of scopes so that they all apply when the job is inserted into the queue rather than when the job is dispatched.

func (*Base) SetEnqueueScopes

func (b *Base) SetEnqueueScopes(scopes ...string)

SetEnqueueScopes overrides the default behavior for the given scopes so that they apply when the job is inserted into the queue rather than when the job is dispatched. This filter will be ignored if EnqueueAllScopes is true.

func (*Base) SetID

func (b *Base) SetID(n string)

SetID makes it possible to change the ID of an amboy.Job.

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) SetTimeInfo

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

SetTimeInfo sets the value of time in the job, including unset fields.

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) UpdateRetryInfo

func (b *Base) UpdateRetryInfo(opts amboy.JobRetryOptions)

UpdateRetryInfo updates the stored retry information and configuration, but does not modify fields that are unset.

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.