pool

Documentation

Overview

Package pool contains a generic implementation of the worker pool pattern for concurrent processing

Index

• type Job
• type MetricsOptions
• type Option
• type Options
• type Phase
  • func (p Phase) String() string
• type Pool
  • func New(jobs []Job, options ...Option) Pool
• type Status
  • func (s Status) Dict() *zerolog.Event
• type StatusReporter
• type SummarizerOptions

Types

type Job

type Job struct {
	// Name short text description for this job to use in log messages
	Name string
	// Run function that performs work
	Run func(StatusReporter) error
	// Labels optional set of metric labels to add to job metrics
	Labels map[string]string
}

Job a unit of work that can be executed by a Pool

type MetricsOptions

type MetricsOptions struct {
	// Enabled if true, record metrics
	Enabled bool
	// PoolName optional name prefix for job metrics
	PoolName string
}

type Option

type Option func(*Options)

Option function for configuring Options

type Options

type Options struct {
	// NumWorkers number of jobs to execute concurrently
	NumWorkers int
	// StopProcessingOnError whether to stop processing work items in the event a job returns an error
	StopProcessingOnError bool
	// Summarizer options for printing periodic processing summaries to the log
	Summarizer SummarizerOptions
	// Metrics options for recording metrics
	Metrics MetricsOptions
}

type Phase

type Phase int64

Phase is an enum type representing the execution phase of a work item

const (
	Queued Phase = iota
	Running
	Success
	Error
)

func (Phase) String

func (p Phase) String() string

type Pool

type Pool interface {
	// Execute starts execution of the pool, returning an error that aggregates errors from all jobs (if any were encountered)
	Execute() error
	// NumWorkers returns the number of workers in the pool
	NumWorkers() int
}

Pool implements the worker pool pattern for concurrent processing

func New

func New(jobs []Job, options ...Option) Pool

type Status

type Status struct {
	// Message a short message summarizing job status
	Message string
	// Context contextual field to add to status summary in log messages
	Context map[string]interface{}
}

Status is for representing job status in log messages

func (Status) Dict

func (s Status) Dict() *zerolog.Event

Dict returns a representation of this Status as a zerolog Dict for inclusion in log messages

type StatusReporter

type StatusReporter interface {
	// Update report a job status update
	Update(status Status)
}

StatusReporter is an interface for reporting job status updates in logs. Note that its use is _optional_ -- a job's Run() function is free to ignore it.

type SummarizerOptions

type SummarizerOptions struct {
	// Enabled if true, print a periodic summary of pool status while items are being processed. For example:
	//
	// 2/5 items processed queued=1 running=2 success=1 error=1
	// foo:    error   err="something bad happened" dur=2m30s
	// bar:    running status="downloading file" dur=30s
	// baz:    running status="uploading file" dur=1m53s
	// quux:   queued
	// blergh: success status="finished transfer"
	//
	Enabled bool
	// Interval how frequently summary messages should be printed to the log
	Interval time.Duration
	// LogLevel level at which summary messages should be logged
	LogLevel zerolog.Level
	// WorkDescription description to use in summary header (defaults to "items processed")
	WorkDescription string
	// Footer an optional footer string to include after printing the summary
	Footer string
	// MaxLineItems an optional number of maximum line items to include in the summary
	MaxLineItems int
	// contains filtered or unexported fields
}