management

Documentation

Overview

Package management provides increased observability and control of the state of amboy queues.

Index

• Constants
• type DBQueueManagerOptions
  • func (o *DBQueueManagerOptions) Validate() error
• type ErrorFilter
  • func (t ErrorFilter) Validate() error
• type JobCounters
• type JobErrorsForType
• type JobErrorsReport
• type JobReportIDs
• type JobRuntimeReport
• type JobRuntimes
• type JobStatusReport
• type Manager
  • func MakeDBQueueManager(ctx context.Context, opts DBQueueManagerOptions, client *mongo.Client) (Manager, error)
  • func NewDBQueueManager(ctx context.Context, opts DBQueueManagerOptions) (Manager, error)
  • func NewQueueManager(q amboy.Queue) Manager
• type RuntimeFilter
  • func (t RuntimeFilter) Validate() error
• type StatusFilter
  • func (t StatusFilter) Validate() error

Constants

const (
	InProgress StatusFilter = "in-progress"
	Pending                 = "pending"
	Stale                   = "stale"
	Completed               = "completed"
	All                     = "all"
)

nolint

const (
	Duration RuntimeFilter = "completed"
	Latency                = "latency"
	Running                = "running"
)

nolint

const (
	UniqueErrors ErrorFilter = "unique-errors"
	AllErrors                = "all-errors"
	StatsOnly                = "stats-only"
)

nolint

Types

type DBQueueManagerOptions

type DBQueueManagerOptions struct {
	Name        string
	Group       string
	SingleGroup bool
	ByGroups    bool
	Options     queue.MongoDBOptions
}

DBQueueManagerOptions describes the arguments to the operations to construct queue managers, and accommodates both group-backed queues and conventional queues.

func (*DBQueueManagerOptions) Validate

func (o *DBQueueManagerOptions) Validate() error

Validate checks the state of the manager configuration, preventing logically invalid options.

type ErrorFilter

type ErrorFilter string

ErrorFilter defines post-processing on errors as returned to users in error queries.

func (ErrorFilter) Validate

func (t ErrorFilter) Validate() error

Validate returns an error if a filter value is not valid.

type JobCounters

type JobCounters struct {
	ID    string `bson:"_id" json:"type" yaml:"type"`
	Count int    `bson:"count" json:"count" yaml:"count"`
	Group string `bson:"group,omitempty" json:"group,omitempty" yaml:"group,omitempty"`
}

JobCounters holds data for counts of jobs by type.

type JobErrorsForType

type JobErrorsForType struct {
	ID      string   `bson:"_id" json:"type" yaml:"type"`
	Count   int      `bson:"count" json:"count" yaml:"count"`
	Total   int      `bson:"total" json:"total" yaml:"total"`
	Average float64  `bson:"average" json:"average" yaml:"average"`
	Errors  []string `bson:"errors,omitempty" json:"errors,omitempty" yaml:"errors,omitempty"`
	Group   string   `bson:"group,omitempty" json:"group,omitempty" yaml:"group,omitempty"`
}

JobErrorsForType holds data about the errors for a specific type of jobs.

type JobErrorsReport

type JobErrorsReport struct {
	Period         time.Duration      `bson:"period" json:"period" yaml:"period"`
	FilteredByType bool               `bson:"filtered" json:"filtered" yaml:"filtered"`
	Data           []JobErrorsForType `bson:"data" json:"data" yaml:"data"`
}

JobErrorsReport contains data for all job errors, by job type, for a specific period.

type JobReportIDs

type JobReportIDs struct {
	Type   string   `bson:"_id" json:"type" yaml:"type"`
	Filter string   `bson:"filter" json:"filter" yaml:"filter"`
	IDs    []string `bson:"jobs" json:"jobs" yaml:"jobs"`
	Group  string   `bson:"group,omitempty" json:"group,omitempty" yaml:"group,omitempty"`
}

JobReportIDs contains the IDs of all jobs of a specific type.

type JobRuntimeReport

type JobRuntimeReport struct {
	Filter string        `bson:"filter" json:"filter" yaml:"filter"`
	Period time.Duration `bson:"period" json:"period" yaml:"period"`
	Stats  []JobRuntimes `bson:"data" json:"data" yaml:"data"`
}

JobRuntimeReport contains data for the runtime of jobs, by type, that have run a specified period.

type JobRuntimes

type JobRuntimes struct {
	ID       string        `bson:"_id" json:"type" yaml:"type"`
	Duration time.Duration `bson:"duration" json:"duration" yaml:"duration"`
	Group    string        `bson:"group,omitempty" json:"group,omitempty" yaml:"group,omitempty"`
}

JobRuntimes holds data for runtimes of jobs by type.

type JobStatusReport

type JobStatusReport struct {
	Filter string        `bson:"filter" json:"filter" yaml:"filter"`
	Stats  []JobCounters `bson:"data" json:"data" yaml:"data"`
}

JobStatusReport contains data for the numbers of jobs that exist for a specified type.

type Manager

type Manager interface {
	JobStatus(context.Context, StatusFilter) (*JobStatusReport, error)
	RecentTiming(context.Context, time.Duration, RuntimeFilter) (*JobRuntimeReport, error)
	JobIDsByState(context.Context, string, StatusFilter) (*JobReportIDs, error)
	RecentErrors(context.Context, time.Duration, ErrorFilter) (*JobErrorsReport, error)
	RecentJobErrors(context.Context, string, time.Duration, ErrorFilter) (*JobErrorsReport, error)
	CompleteJobsByType(context.Context, StatusFilter, string) error
	CompleteJob(context.Context, string) error
	CompleteJobs(context.Context, StatusFilter) error
	CompleteJobsByPrefix(context.Context, StatusFilter, string) error
}

Manager is an interface that describes queue introspection tools and utility for queue management that make it possible to get more details about the running jobs in an amboy queue and gives users broader capabilities than the Queue interface itself.

func MakeDBQueueManager

func MakeDBQueueManager(ctx context.Context, opts DBQueueManagerOptions, client *mongo.Client) (Manager, error)

MakeDBQueueManager make it possible to produce a queue manager with an existing database Connection. This operations runs the "ping" command and and will return an error if there is no session or no active server.

func NewDBQueueManager

func NewDBQueueManager(ctx context.Context, opts DBQueueManagerOptions) (Manager, error)

NewDBQueueManager produces a queue manager for (remote) queues that persist jobs in MongoDB. This implementation does not interact with the queue directly, and manages by interacting with the database directly.

func NewQueueManager

func NewQueueManager(q amboy.Queue) Manager

NewQueueManager returns a queue manager that provides the supported Management interface by calling the output of amboy.Queue.Results() and amboy.Queue.JobStats(), iterating over jobs directly. Use this to manage in-memory queue implementations more generically.

The management algorithms may impact performance of jobs, as queues may require some locking to their Jobs function. Additionally, the speed of these operations will necessarily degrade with the number of jobs. Do pass contexts with timeouts to in these cases.

type RuntimeFilter

type RuntimeFilter string

RuntimeFilter provides ways to filter the timing data returned by the reporting interface.

func (RuntimeFilter) Validate

func (t RuntimeFilter) Validate() error

Validate returns an error if a filter value is not valid.

type StatusFilter

type StatusFilter string

StatusFilter defines a number of dimensions with which to filter current jobs in a queue by status

func (StatusFilter) Validate

func (t StatusFilter) Validate() error

Validate returns an error if a filter value is not valid.