faktory

README

Faktory

At a high level, Faktory is a work server. It is the repository for background jobs within your application. Jobs have a type and a set of arguments and are placed into queues for workers to fetch and execute.

You can use this server to distribute jobs to one or hundreds of machines. Jobs can be executed with any language by clients using the Faktory API to fetch a job from a queue.

Basic Features

• Jobs are represented as JSON hashes.
• Jobs are pushed to and fetched from queues.
• Jobs are reserved with a timeout, 30 min by default.
• Jobs FAIL'd or not ACK'd within the reservation timeout are requeued.
• FAIL'd jobs trigger a retry workflow with exponential backoff.
• Contains a comprehensive Web UI for management and monitoring.

Installation

See the Installation wiki page for current installation methods.

Documentation

Please see the Faktory wiki for full documentation.

Support

You can find help in the contribsys/faktory chat channel. Stop by and say hi!

Author

Mike Perham, @mperham, mike @ contribsys.com

Documentation

Index

• Constants
• Variables
• type Client
  • func Dial(srv *Server, password string) (*Client, error)
  • func Open() (*Client, error)
  • func (c *Client) Ack(jid string) error
  • func (c *Client) Beat() (string, error)
  • func (c *Client) Close() error
  • func (c *Client) Fail(jid string, err error, backtrace []byte) error
  • func (c *Client) Fetch(q ...string) (*Job, error)
  • func (c *Client) Flush() error
  • func (c *Client) Generic(cmdline string) (string, error)
  • func (c *Client) Info() (map[string]interface{}, error)
  • func (c *Client) Push(job *Job) error
• type ClientData
• type Failure
• type Job
  • func NewJob(jobtype string, args ...interface{}) *Job
  • func (j *Job) EnsureValidPriority()
  • func (j *Job) GetPriority() uint8
• type ProtocolError
  • func (pe *ProtocolError) Error() string
• type Server
  • func DefaultServer() *Server

Constants

const (
	// This is the protocol version supported by this client.
	// The server might be running an older or newer version.
	ExpectedProtocolVersion = 2
)

Variables

var (
	Name      = "Faktory"
	License   = "GPLv3"
	Licensing = "Licensed under the GNU Public License 3.0"
	Version   = "0.6.0"
)

var (
	// Set this to a non-empty value in a consumer process
	RandomProcessWid = ""
)

Types

type Client

type Client struct {
	Location string
	Options  *ClientData
	// contains filtered or unexported fields
}

func Dial

func Dial(srv *Server, password string) (*Client, error)

Dial connects to the remote faktory server.

faktory.Dial(faktory.Localhost, "topsecret")

func Open

func Open() (*Client, error)

Open connects to a Faktory server based on the environment variable conventions:

• Use FAKTORY_PROVIDER to point to a custom URL variable.

• Use FAKTORY_URL as a catch-all default.

func (*Client) Ack

func (c *Client) Ack(jid string) error

func (*Client) Beat

func (c *Client) Beat() (string, error)

func (*Client) Close

func (c *Client) Close() error

func (*Client) Fail

func (c *Client) Fail(jid string, err error, backtrace []byte) error

Fail notifies Faktory that a job failed with the given error. If backtrace is non-nil, it is assumed to be the output from runtime/debug.Stack().

func (*Client) Fetch

func (c *Client) Fetch(q ...string) (*Job, error)

func (*Client) Flush

func (c *Client) Flush() error

func (*Client) Generic

func (c *Client) Generic(cmdline string) (string, error)

func (*Client) Info

func (c *Client) Info() (map[string]interface{}, error)

func (*Client) Push

func (c *Client) Push(job *Job) error

type ClientData

type ClientData struct {
	Hostname string   `json:"hostname"`
	Wid      string   `json:"wid"`
	Pid      int      `json:"pid"`
	Labels   []string `json:"labels"`
	// Hash is hex(sha256(password + nonce))
	PasswordHash string `json:"pwdhash"`
	// The protocol version used by this client.
	// The server can reject this connection if the version will not work
	// The server advertises its protocol version in the HI.
	Version int `json:"v"`
}

ClientData is serialized to JSON and sent with the HELLO command. PasswordHash is required if the server is not listening on localhost. The WID (worker id) must be random and unique for each worker process. It can be a UUID, etc. Non-worker processes should leave WID empty.

The other elements can be useful for debugging and are displayed on the Busy tab.

type Failure

type Failure struct {
	RetryCount   int      `json:"retry_count"`
	FailedAt     string   `json:"failed_at"`
	NextAt       string   `json:"next_at,omitempty"`
	ErrorMessage string   `json:"message,omitempty"`
	ErrorType    string   `json:"errtype,omitempty"`
	Backtrace    []string `json:"backtrace,omitempty"`
}

type Job

type Job struct {
	// required
	Jid   string        `json:"jid"`
	Queue string        `json:"queue"`
	Type  string        `json:"jobtype"`
	Args  []interface{} `json:"args"`

	// optional
	Priority   uint8                  `json:"priority,omitempty"`
	CreatedAt  string                 `json:"created_at,omitempty"`
	EnqueuedAt string                 `json:"enqueued_at,omitempty"`
	At         string                 `json:"at,omitempty"`
	ReserveFor int                    `json:"reserve_for,omitempty"`
	Retry      int                    `json:"retry,omitempty"`
	Backtrace  int                    `json:"backtrace,omitempty"`
	Failure    *Failure               `json:"failure,omitempty"`
	Custom     map[string]interface{} `json:"custom,omitempty"`
}

func NewJob

func NewJob(jobtype string, args ...interface{}) *Job

func (*Job) EnsureValidPriority

func (j *Job) EnsureValidPriority()

func (*Job) GetPriority

func (j *Job) GetPriority() uint8

Accessor so that if priority isn't specified we don't persist it to disk

type ProtocolError

type ProtocolError struct {
	// contains filtered or unexported fields
}

func (*ProtocolError) Error

func (pe *ProtocolError) Error() string

type Server

type Server struct {
	Network string
	Address string
	Timeout time.Duration
}

func DefaultServer

func DefaultServer() *Server