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 not ACK'd or FAIL'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.

We need help getting Faktory running easily on OSX. Today you have to git clone this repo and build manually but we hope to get it into Homebrew soon.

Documentation

Please see the Faktory wiki for full documentation.

Support

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

Author

Mike Perham, @mperham, mike @ contribsys.com

Documentation

Index

• 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
• type ProtocolError
  • func (pe *ProtocolError) Error() string
• type Server
  • func DefaultServer() *Server

Variables

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

var (
	RandomProcessWid = strconv.FormatInt(rand.Int63(), 32)
)

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)

* Open a connection to the remote faktory server. * * faktory.Dial(faktory.Localhost, "topsecret") *

func Open

func Open() (*Client, error)

* This function 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

* Notify 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"`
	// Salt should be a random string and
	// must change on every call.
	Salt string `json:"salt"`
	// Hash is hex(sha256(password + salt))
	PasswordHash string `json:"pwdhash"`
}

* This data is serialized to JSON and sent * with the AHOY 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. * * 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
	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

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