amboy: Index | Files | Directories

package amboy

import ""

Package amboy provides basic infrastructure for running and describing tasks and task workflows with, potentially, minimal overhead and additional complexity.

Overview and Motivation

Amboy works with 4 basic logical objects: jobs, or descriptions of tasks; runnners, which are responsible for executing tasks; queues, that represent pipelines and offline workflows of tasks (e.g. not real time, processes that run outside of the primary execution path of a program); and dependencies that represent relationships between jobs.

The inspiration for amboy was to be able to provide a unified way to define and run jobs, that would feel equally "native" for distributed applications and distributed web application, and move easily between different architectures.

While amboy users will generally implement their own Job and dependency implementations, Amboy itself provides several example Queue implementations, as well as several generic examples and prototypes of Job and dependency.Manager objects.

Generally speaking you should be able to use included amboy components to provide the queue and runner components, in conjunction with custom and generic job and dependency variations.

Consider the following example:

   queue := queue.SimpleQueue(12) // pass the number of worker threads
   job := job.NewShellJob("make compile")

   err := queue.Put(job)
   if err != nil {
      // handle error case

   err = queue.Start(ctx) // the queue starts a SimpleRunner object and
		       // creates required channels.
   if err != nil {
      // handle error case

   Wait(queue) // waits for all tasks to finish.
   queue.Close() // waits for all tasks to finish and releases
		 // all resources.

Waiting for Jobs to Complete

The amboy package proves a number of generic methods that, using the Queue.Stats() method, block until all jobs are complete. They provide different semantics, which may be useful in different circumstances. All of these functions wait until the total number of jobs submitted to the queue is equal to the number of completed jobs, and as a result these methods don't prevent other threads from adding jobs to the queue after beginning to wait.

Additionally, there are a set of methods that allow callers to wait for a specific job to complete.


Package Files

doc.go formats.go interface.go meta.go periodic.go stats.go wait.go

func IntervalQueueOperation Uses

func IntervalQueueOperation(ctx context.Context, q Queue, interval time.Duration, startAt time.Time, conf QueueOperationConfig, op QueueOperation)

IntervalQueueOperation runs a queue scheduling operation on a regular interval, starting at specific time. Use this method to schedule jobs every hour, or similar use-cases.

func PeriodicQueueOperation Uses

func PeriodicQueueOperation(ctx context.Context, q Queue, interval time.Duration, conf QueueOperationConfig, op QueueOperation)

PeriodicQueueOperation launches a goroutine that runs the QueueOperation on the specified Queue at the specified interval. If ignoreErrors is true, then a QueueOperation that returns an error will *not* interrupt the background process. Otherwise, the background process will exit if a QueueOperation fails. Use the context to terminate the background process.

func PopulateQueue Uses

func PopulateQueue(ctx context.Context, q Queue, jobs <-chan Job) error

PopulateQueue adds jobs from a channel to a queue and returns an error with the aggregated results of these operations.

func ResolveErrors Uses

func ResolveErrors(ctx context.Context, q Queue) error

ResolveErrors takes a queue object and iterates over the results and returns a single aggregated error for the queue's job. The completeness of this operation depends on the implementation of a the queue implementation's Results() method.

func RunJob Uses

func RunJob(ctx context.Context, job Job) error

RunJob executes a single job directly, without a queue, with similar semantics as it would execute in a queue: MaxTime is respected, and it uses similar logging as is present in the queue, with errors propogated functionally.

func Wait Uses

func Wait(q Queue)

Wait takes a queue and blocks until all tasks are completed. This operation runs in a tight-loop, which means that the Wait will return *as soon* as possible all tasks or complete. Conversely, it's also possible that frequent repeated calls to Stats() may contend with resources needed for dispatching jobs or marking them complete.

func WaitCtx Uses

func WaitCtx(ctx context.Context, q Queue) bool

WaitCtx make it possible to cancel, either directly or using a deadline or timeout, a Wait operation using a context object. The return value is true if all tasks are complete, and false if the operation returns early because it was canceled.

func WaitCtxInterval Uses

func WaitCtxInterval(ctx context.Context, q Queue, interval time.Duration) bool

WaitCtxInterval provides the Wait operation and accepts a context for cancellation while also waiting for an interval between stats calls. The return value reports if the operation was canceled or if all tasks are complete.

func WaitCtxIntervalNum Uses

func WaitCtxIntervalNum(ctx context.Context, q Queue, interval time.Duration, num int) bool

WaitCtxIntervalNum waits for a certain number of jobs to complete, with the same semantics as WaitCtxInterval.

func WaitInterval Uses

func WaitInterval(q Queue, interval time.Duration)

WaitInterval adds a sleep between stats calls, as a way of throttling the impact of repeated Stats calls to the queue.

func WaitJob Uses

func WaitJob(j Job, q Queue) bool

WaitJob blocks until the job, based on its ID, is marked complete in the queue. The return value is false if the job does not exist (or is removed) and true when the job completes. This operation could block indefinitely.

func WaitJobCtx Uses

func WaitJobCtx(ctx context.Context, j Job, q Queue) bool

WaitJobCtx blocks until the job, based on its ID, is marked complete in the queue. This operation blocks indefinitely, unless the context is canceled or reaches its timeout. The return value is false if the job does not exist or if the context is canceled, and only returns true when the job is complete.

func WaitJobCtxInterval Uses

func WaitJobCtxInterval(ctx context.Context, j Job, q Queue, interval time.Duration) bool

WaitJobCtxInterval waits for a job in a queue to complete. Returns false if the context has been canceled, or if the job does not exist in the queue, and true only after the job is marked complete.

func WaitJobInterval Uses

func WaitJobInterval(j Job, q Queue, interval time.Duration) bool

WaitJobInterval takes a job and queue object and waits for the job to be marked complete. The interval parameter controls how long the operation waits between checks, and can be used to limit the impact of waiting on a busy queue. The operation returns false if the job is not registered in the queue, and true when the job completes.

type AbortableRunner Uses

type AbortableRunner interface {

    IsRunning(string) bool
    RunningJobs() []string
    Abort(context.Context, string) error

AbortableRunner provides a superset of the Runner interface but allows callers to abort jobs by ID.

type Format Uses

type Format int

Format defines a sequence of constants used to distinguish between different serialization formats for job objects used in the amboy.ConvertTo and amboy.ConvertFrom functions, which support the functionality of the Export and Import methods in the job interface.

const (
    BSON Format = iota

Supported values of the Format type, which represent different supported serialization methods..

func (Format) String Uses

func (f Format) String() string

String implements fmt.Stringer and pretty prints the format name.

type Job Uses

type Job interface {
    // Provides a unique identifier for a job. Queues may error if
    // two jobs have different IDs.
    ID() string

    // The primary execution method for the job. Should toggle the
    // completed state for the job.

    // Returns a pointer to a JobType object that Queue
    // implementations can use to de-serialize tasks.
    Type() JobType

    // Provides access to the job's dependency information, and
    // allows queues to override a dependency (e.g. in a force
    // build state, or as part of serializing dependency objects
    // with jobs.)
    Dependency() dependency.Manager

    // Provides access to the JobStatusInfo object for the job,
    // which reports the current state.
    Status() JobStatusInfo

    // TimeInfo reports the start/end time of jobs, as well as
    // providing for a "wait until" functionality that queues can
    // use to schedule jobs in the future. The update method, only
    // updates non-zero methods.
    TimeInfo() JobTimeInfo

    // Provides access to the job's priority value, which some
    // queues may use to order job dispatching. Most Jobs
    // implement these values by composing the
    // amboy/priority.Value type.
    Priority() int

    // AddError allows another actor to annotate the job with an
    // error.
    // Error returns an error object if the task was an
    // error. Typically if the job has not run, this is nil.
    Error() error

Job describes a unit of work. Implementations of Job instances are the content of the Queue. The amboy/job package contains several general purpose and example implementations. Jobs are responsible, primarily via their Dependency property, for determining: if they need to run, and what Jobs they depend on. Actual use of the dependency system is the responsibility of the Queue implementation.

In most cases, applications only need to implement the Run() method, all additional functionality is provided by the job.Base type, which can be embedded anonymously in implementations of the Job.

type JobStatusInfo Uses

type JobStatusInfo struct {
    ID                string    `bson:"id,omitempty" json:"id,omitempty" yaml:"id,omitempty"`
    Owner             string    `bson:"owner" json:"owner" yaml:"owner"`
    Completed         bool      `bson:"completed" json:"completed" yaml:"completed"`
    InProgress        bool      `bson:"in_prog" json:"in_progress" yaml:"in_progress"`
    ModificationTime  time.Time `bson:"mod_ts" json:"mod_time" yaml:"mod_time"`
    ModificationCount int       `bson:"mod_count" json:"mod_count" yaml:"mod_count"`
    ErrorCount        int       `bson:"err_count" json:"err_count" yaml:"err_count"`
    Errors            []string  `bson:"errors,omitempty" json:"errors,omitempty" yaml:"errors,omitempty"`

JobStatusInfo contains information about the current status of a job and is reported by the Status and set by the SetStatus methods in the Job interface.e

type JobTimeInfo Uses

type JobTimeInfo struct {
    Created   time.Time     `bson:"created,omitempty" json:"created,omitempty" yaml:"created,omitempty"`
    Start     time.Time     `bson:"start" json:"start,omitempty" yaml:"start,omitempty"`
    End       time.Time     `bson:"end" json:"end,omitempty" yaml:"end,omitempty"`
    WaitUntil time.Time     `bson:"wait_until" json:"wait_until,omitempty" yaml:"wait_until,omitempty"`
    MaxTime   time.Duration `bson:"max_time" json:"max_time,omitempty" yaml:"max_time,omitempty"`

JobTimeInfo stores timing information for a job and is used by both the Runner and Job implementations to track how long jobs take to execute. Additionally, the Queue implementations __may__ use this data to delay execution of a job when WaitUntil refers to a time in the future.

func (JobTimeInfo) Duration Uses

func (j JobTimeInfo) Duration() time.Duration

Duration is a convenience function to return a duration for a job.

type JobType Uses

type JobType struct {
    Name    string `json:"name" bson:"name" yaml:"name"`
    Version int    `json:"version" bson:"version" yaml:"version"`

JobType contains information about the type of a job, which queues can use to serialize objects. All Job implementations must store and produce instances of this type that identify the type and implementation version.

type Queue Uses

type Queue interface {
    // Used to add a job to the queue. Should only error if the
    // Queue cannot accept jobs.
    Put(Job) error

    // Given a job id, get that job. The second return value is a
    // Boolean, which indicates if the named job had been
    // registered by a Queue.
    Get(string) (Job, bool)

    // Returns the next job in the queue. These calls are
    // blocking, but may be interrupted with a canceled context.
    Next(context.Context) Job

    // Makes it possible to detect if a Queue has started
    // dispatching jobs to runners.
    Started() bool

    // Used to mark a Job complete and remove it from the pending
    // work of the queue.
    Complete(context.Context, Job)

    // Returns a channel that produces completed Job objects.
    Results(context.Context) <-chan Job

    // Returns a channel that produces the status objects for all
    // jobs in the queue, completed and otherwise.
    JobStats(context.Context) <-chan JobStatusInfo

    // Returns an object that contains statistics about the
    // current state of the Queue.
    Stats() QueueStats

    // Getter for the Runner implementation embedded in the Queue
    // instance.
    Runner() Runner

    // Setter for the Runner implementation embedded in the Queue
    // instance. Permits runtime substitution of interfaces, but
    // implementations are not expected to permit users to change
    // runner implementations after starting the Queue.
    SetRunner(Runner) error

    // Begins the execution of the job Queue, using the embedded
    // Runner.
    Start(context.Context) error

Queue describes a very simple Job queue interface that allows users to define Job objects, add them to a worker queue and execute tasks from that queue. Queue implementations may run locally or as part of a distributed application, with multiple workers and submitter Queue instances, which can support different job dispatching and organization properties.

type QueueGroup Uses

type QueueGroup interface {
    // Get a queue with the given index.
    Get(context.Context, string) (Queue, error)

    // Put a queue at the given index.
    Put(context.Context, string, Queue) error

    // Prune old queues.
    Prune(context.Context) error

    // Close the queues.
    Close(context.Context) error

    // Len returns the number of active queues managed in the
    // group.
    Len() int

    // Queues returns all currently registered and running queues
    Queues(context.Context) []string

QueueGroup describes a group of queues. Each queue is indexed by a string. Users can use these queues if there are many different types of work or if the types of work are only knowable at runtime.

type QueueOperation Uses

type QueueOperation func(Queue) error

QueueOperation is a named function literal for use in the PeriodicQueueOperation function. Typically these functions add jobs to a queue, or could be used to perform periodic maintenance (e.g. removing stale jobs or removing stuck jobs in a dependency queue.)

func GroupQueueOperationFactory Uses

func GroupQueueOperationFactory(first QueueOperation, ops ...QueueOperation) QueueOperation

GroupQueueOperationFactory produces a QueueOperation that aggregates and runs one or more QueueOperations. The QueueOperation has continue-on-error semantics, and returns an error if any of the QueueOperations fail, but attempts to run all specified QueueOperations before propagating errors.

func ScheduleJobFactory Uses

func ScheduleJobFactory(op func() Job) QueueOperation

ScheduleJobFactory produces a QueueOpertion that calls a single function which returns a Job and puts that job into the queue.

func ScheduleJobsFromGeneratorFactory Uses

func ScheduleJobsFromGeneratorFactory(op func() <-chan Job) QueueOperation

ScheduleJobsFromGeneratorFactory produces a queue operation that calls a single generator function which returns channel of Jobs and puts those jobs into the queue. The QueueOperation attempts to add all jobs in the slice and returns an error if the Queue.Put opertion failed for any (e.g. continue-on-error semantics). The error returned aggregates all errors encountered.

func ScheduleManyJobsFactory Uses

func ScheduleManyJobsFactory(op func() []Job) QueueOperation

ScheduleManyJobsFactory produces a queue operation that calls a single function which returns a slice of jobs and puts those jobs into the queue. The QueueOperation attempts to add all jobs in the slice and returns an error if the Queue.Put opertion failed for any (e.g. continue-on-error semantics). The error returned aggregates all errors encountered.

type QueueOperationConfig Uses

type QueueOperationConfig struct {
    ContinueOnError bool `bson:"continue_on_error" json:"continue_on_error" yaml:"continue_on_error"`
    LogErrors       bool `bson:"log_errors" json:"log_errors" yaml:"log_errors"`
    DebugLogging    bool `bson:"debug_logging" json:"debug_logging" yaml:"debug_logging"`

QueueOperationConfig describes the behavior of the periodic interval schedulers.

type QueueReport Uses

type QueueReport struct {
    Completed  []string `json:"completed"`
    InProgress []string `json:"in_progress"`
    Pending    []string `json:"pending"`

QueueReport holds the ids of all tasks in a queue by state.

func Report Uses

func Report(ctx context.Context, q Queue, limit int) QueueReport

Report returns a QueueReport status for the state of a queue.

type QueueStats Uses

type QueueStats struct {
    Running   int            `bson:"running" json:"running" yaml:"running"`
    Completed int            `bson:"completed" json:"completed" yaml:"completed"`
    Pending   int            `bson:"pending" json:"pending" yaml:"pending"`
    Blocked   int            `bson:"blocked" json:"blocked" yaml:"blocked"`
    Total     int            `bson:"total" json:"total" yaml:"total"`
    Context   message.Fields `bson:"context,omitempty" json:"context,omitempty" yaml:"context,omitempty"`
    // contains filtered or unexported fields

QueueStats is a simple structure that the Stats() method in the Queue interface returns and tracks the state of the queue, and provides a common format for different Queue implementations to report on their state.

Implement's grip's message.Composer interface when passed as a pointer.

func (*QueueStats) Annotate Uses

func (s *QueueStats) Annotate(key string, value interface{}) error

Annotate is part of the grip/message.Composer interface and allows the logging infrastructure to inject content and context into log messages.

func (QueueStats) IsComplete Uses

func (s QueueStats) IsComplete() bool

IsComplete reutrns true when the total number of tasks are equal to the number completed, or if the number of completed and blocked are greater than or equal to total. This method is used by the Wait<> functions to determine when a queue has completed all actionable work.

func (QueueStats) Loggable Uses

func (s QueueStats) Loggable() bool

Loggable is part of the grip/message.Composer interface and only returns true if the queue has at least one job.

func (QueueStats) Priority Uses

func (s QueueStats) Priority() level.Priority

Priority is part of the grip/message.Composer interface and returns the priority of the message.

func (QueueStats) Raw Uses

func (s QueueStats) Raw() interface{}

Raw is part of the grip/message.Composer interface and simply returns the QueueStats object.

func (*QueueStats) SetPriority Uses

func (s *QueueStats) SetPriority(l level.Priority) error

SetPriority is part of the grip/message.Composer interface and allows the caller to configure the piroity of the message.

func (QueueStats) String Uses

func (s QueueStats) String() string

String prints a long form report of the queue for human consumption.

type Runner Uses

type Runner interface {
    // Reports if the pool has started.
    Started() bool

    // Provides a method to change or set the pointer to the
    // enclosing Queue object after instance creation. Runner
    // implementations may not be able to change their Queue
    // association after starting.
    SetQueue(Queue) error

    // Prepares the runner implementation to begin doing work, if
    // any is required (e.g. starting workers.) Typically called
    // by the enclosing Queue object's Start() method.
    Start(context.Context) error

    // Termaintes all in progress work and waits for processes to
    // return.

Runner describes a simple worker interface for executing jobs in the context of a Queue. Used by queue implementations to run tasks. Generally Queue implementations will spawn a runner as part of their constructor or Start() methods, but client code can inject alternate Runner implementations, as required.


buildscriptsThe current vendoring solution supports both new and old style vendoring, via a trick: We commit all vendored code to the "vendor" directory, and then, if we're on a version/deployment of go that doesn't support new style vendoring, we symlink to "build/vendor/src" and add "build/vendor" to the gopath, which the render-gopath program generates inside of the makefile.
buildscripts/vendoringPackage vendoring provides a several variables used in vendoring buildscripts and function that reports (without any external dependencies) if the current environment requires legacy-style vendoring, or if its safe to use new-style vendoring.
dependencyPackage dependency contains the Manager interface, along with several implementations for different kinds of dependency checks.
jobPackage job provides tools and generic implementations of jobs for amboy Queues.
loggerPackage logqueue is a set of implementations to support amboy.Queue backed grip/send.Senders for asynchronous and (generally) non-blocking log message delivery.
poolPackage pool provides specific implementations of the amboy.Runner interface that serve as the worker pools for tasks in work queues.
queuePackage queue provides several implementations of the amboy.Queue interface capable of processing amboy.Job implementations.
registryPackage registry contains infrastructure to support the persistence of Job definitions.
reportingPackage reporting provides increased observability of the state of MongoDB-backed amboy queues.

Package amboy imports 9 packages (graph) and is imported by 73 packages. Updated 2019-04-25. Refresh now. Tools for package owners.