beanstalkg/beanstalkg

Beanstalkg is a go implementation of beanstalkd - A fast, general-purpose work queue

Beanstalkg

Beanstalkg is a go implementation of Beanstalkd a fast, general-purpose work queue. Idea is to support the same set of features and protocol with the addition of high availability and failover built in. You can read the plan.md if interested in contributing.

Right now it supports all the basic commands to run producers and workers. i.e "use", "put", "watch", "ignore", "reserve", "delete", "release", "bury", "reserve-with-timeout".

I wish to complete rest of the commands soon but any help is always appreciated.

Advantages compared to beanstalkd

• Extensible design. For example you can replace the backend storage with anything you like, just implement a simple interface and plugin.
• Implemented in golang. More readable code with support for concurrency using awesome go routines.
• Support for clustering(coming soon :)

User guide

Beanstalkg is currently only released as a docker image for users. Latest release is v0.0.3. Assuming you already have a working docker engine installation, you can start a Beanstalkg instance with following steps,

• Run command docker run -p 11300:11300 beanstalkg/beanstalkg:v0.0.3. This will start the beanstalkg server in the foreground. The server starts listening on port 11300.
• Now you can connect to the server with any client library available to beanstalkd. eg: Using official go client

  // Produce jobs:
  c, err := beanstalk.Dial("tcp", "127.0.0.1:11300")
  id, err := c.Put([]byte("hello"), 1, 0, 120*time.Second)
  
  // Consume jobs:
  c, err := beanstalk.Dial("tcp", "127.0.0.1:11300")
  id, body, err := c.Reserve(5 * time.Second)
  
  

Some introductory slides can be found here

Developer guide

Please install golang then with GOPATH set correctly run,

• go get github.com/Beanstalkg/beanstalkg
• Add $GOAPTH/bin to the $PATH and run beanstalkg

Licensing

beanstalkg is licensed under the MIT License. See LICENSE for the full license text.

beanstalkg's Issues

Try ginkgo

http://onsi.github.io/ginkgo/ BDD test framework

reserve-with-timeout seem to be buggy

I observed that reserve with time out produces inconsistent behaviour when clients reconnect. Needs further investigation

implement list-tubes-watched command

The list-tubes-watched command returns a list tubes currently being watched by
the client. Its form is:

list-tubes-watched\r\n

The response is:

OK <bytes>\r\n
<data>\r\n

• is the size of the following data section in bytes.

• is a sequence of bytes of length from the previous line. It
  is a YAML file containing watched tube names as a list of strings.

Delete command does not delete the job sometimes when there are many workers connected to the same tube

Reason for this is probably the queue frequency(job processing) is getting in the way of the direct commands.
May be we should have a way to reliably execute direct commands on time.

implement pause-tube command

The pause-tube command can delay any new job being reserved for a given time. Its form is:

pause-tube <tube-name> <delay>\r\n

• is the tube to pause

• is an integer number of seconds < 2**32 to wait before reserving any more
  jobs from the queue

There are two possible responses:

• "PAUSED\r\n" to indicate success.

• "NOT_FOUND\r\n" if the tube does not exist.

Cannot resolve getConfig

the getConfig cannot be resolved by importing github.com/beanstalkg/beanstalkg/config.
I have already moved config.go, config.json, and config.yml to config directory.

here is a sample error:

please help to resolve this.

Add beanstalkg to queues.io

PR here: lukaszx0/queues.io#120

Queues.io: http://queues.io/

Hopefully this can get you more exposure.

Implement kick commands

kick

The kick command applies only to the currently used tube. It moves jobs into
the ready queue. If there are any buried jobs, it will only kick buried jobs.
Otherwise it will kick delayed jobs. It looks like:

kick <bound>\r\n

• is an integer upper bound on the number of jobs to kick. The server
  will kick no more than jobs.

The response is of the form:

KICKED <count>\r\n

• is an integer indicating the number of jobs actually kicked.

kick-job

The kick-job command is a variant of kick that operates with a single job
identified by its job id. If the given job id exists and is in a buried or
delayed state, it will be moved to the ready queue of the the same tube where it
currently belongs. The syntax is:

kick-job <id>\r\n

• is the job id to kick.

The response is one of:

• "NOT_FOUND\r\n" if the job does not exist or is not in a kickable state. This
  can also happen upon internal errors.

• "KICKED\r\n" when the operation succeeded.

bury command results with BAD_FORMAT error

I was not able to bury the job using FiveBeans client.

It looks like the last parameter is ignored.

Create a proper release process

We need to create a proper release process so that users on a targeted platform can use beanstalkg easily.

Targeted platforms

1. Container environments(build image as explained in https://docs.docker.com/engine/userguide/eng-image/multistage-build/)
2. rpm packages for the linux variants that use it
3. deb packages for the linux variants that use it

We can probably follow something like https://github.com/goshawkdb/packaging for the linux distros. More research is still needed.

Refactor the `operations` package to be more human readable

Implement stats command

The stats command gives statistical information about the system as a whole.
Its form is:

stats\r\n

The server will respond:

OK <bytes>\r\n
<data>\r\n

• is the size of the following data section in bytes.

• is a sequence of bytes of length from the previous line. It
  is a YAML file with statistical information represented a dictionary.

The stats data for the system is a YAML file representing a single dictionary
of strings to scalars. Entries described as "cumulative" are reset when the
beanstalkd process starts; they are not stored on disk with the -b flag.

• "current-jobs-urgent" is the number of ready jobs with priority < 1024.

• "current-jobs-ready" is the number of jobs in the ready queue.

• "current-jobs-reserved" is the number of jobs reserved by all clients.

• "current-jobs-delayed" is the number of delayed jobs.

• "current-jobs-buried" is the number of buried jobs.

• "cmd-put" is the cumulative number of put commands.

• "cmd-peek" is the cumulative number of peek commands.

• "cmd-peek-ready" is the cumulative number of peek-ready commands.

• "cmd-peek-delayed" is the cumulative number of peek-delayed commands.

• "cmd-peek-buried" is the cumulative number of peek-buried commands.

• "cmd-reserve" is the cumulative number of reserve commands.

• "cmd-use" is the cumulative number of use commands.

• "cmd-watch" is the cumulative number of watch commands.

• "cmd-ignore" is the cumulative number of ignore commands.

• "cmd-delete" is the cumulative number of delete commands.

• "cmd-release" is the cumulative number of release commands.

• "cmd-bury" is the cumulative number of bury commands.

• "cmd-kick" is the cumulative number of kick commands.

• "cmd-stats" is the cumulative number of stats commands.

• "cmd-stats-job" is the cumulative number of stats-job commands.

• "cmd-stats-tube" is the cumulative number of stats-tube commands.

• "cmd-list-tubes" is the cumulative number of list-tubes commands.

• "cmd-list-tube-used" is the cumulative number of list-tube-used commands.

• "cmd-list-tubes-watched" is the cumulative number of list-tubes-watched
  commands.

• "cmd-pause-tube" is the cumulative number of pause-tube commands.

• "job-timeouts" is the cumulative count of times a job has timed out.

• "total-jobs" is the cumulative count of jobs created.

• "max-job-size" is the maximum number of bytes in a job.

• "current-tubes" is the number of currently-existing tubes.

• "current-connections" is the number of currently open connections.

• "current-producers" is the number of open connections that have each
  issued at least one put command.

• "current-workers" is the number of open connections that have each issued
  at least one reserve command.

• "current-waiting" is the number of open connections that have issued a
  reserve command but not yet received a response.

• "total-connections" is the cumulative count of connections.

• "pid" is the process id of the server.

• "version" is the version string of the server.

• "rusage-utime" is the cumulative user CPU time of this process in seconds
  and microseconds.

• "rusage-stime" is the cumulative system CPU time of this process in
  seconds and microseconds.

• "uptime" is the number of seconds since this server process started running.

• "binlog-oldest-index" is the index of the oldest binlog file needed to
  store the current jobs.

• "binlog-current-index" is the index of the current binlog file being
  written to. If binlog is not active this value will be 0.

• "binlog-max-size" is the maximum size in bytes a binlog file is allowed
  to get before a new binlog file is opened.

• "binlog-records-written" is the cumulative number of records written
  to the binlog.

• "binlog-records-migrated" is the cumulative number of records written
  as part of compaction.

• "id" is a random id string for this server process, generated when each
  beanstalkd process starts.

• "hostname" the hostname of the machine as determined by uname.

Create a better configuration system based on yaml files

We can try do golang libraries to do this once we agree on which one to use.

Implement quit command

The quit command simply closes the connection. Its form is:

quit\r\n

architecture: log redeclared in this block

queue.go has a var log = logging.MustGetLogger("BEANSTALKG") and job.go imports github.com/labstack/gommon/log, which causes the conflict.

Kill the github.com/labstack/gommon/log import?

Create integration test framework

Right now there are no integration tests being run to make sure that all implemented functionality runs correctly. Need to develop a proper integration test suite to be run in circleci. Ideally it should also be run against the docker image when a release is made.
Links #57

Complete the blog post series

https://vimukthi.com/queue-story-1-introduction-to-beanstalkg-3f60ecbd6477

Implement stats-tube command

The stats-tube command gives statistical information about the specified tube
if it exists. Its form is:

stats-tube <tube>\r\n

• is a name at most 200 bytes. Stats will be returned for this tube.

The response is one of:

• "NOT_FOUND\r\n" if the tube does not exist.

• "OK \r\n\r\n"

  • is the size of the following data section in bytes.

  • is a sequence of bytes of length from the previous line. It
    is a YAML file with statistical information represented a dictionary.

The stats-tube data is a YAML file representing a single dictionary of strings
to scalars. It contains these keys:

• "name" is the tube's name.

• "current-jobs-urgent" is the number of ready jobs with priority < 1024 in
  this tube.

• "current-jobs-ready" is the number of jobs in the ready queue in this tube.

• "current-jobs-reserved" is the number of jobs reserved by all clients in
  this tube.

• "current-jobs-delayed" is the number of delayed jobs in this tube.

• "current-jobs-buried" is the number of buried jobs in this tube.

• "total-jobs" is the cumulative count of jobs created in this tube in
  the current beanstalkd process.

• "current-using" is the number of open connections that are currently
  using this tube.

• "current-waiting" is the number of open connections that have issued a
  reserve command while watching this tube but not yet received a response.

• "current-watching" is the number of open connections that are currently
  watching this tube.

• "pause" is the number of seconds the tube has been paused for.

• "cmd-delete" is the cumulative number of delete commands for this tube

• "cmd-pause-tube" is the cumulative number of pause-tube commands for this
  tube.

• "pause-time-left" is the number of seconds until the tube is un-paused.

implement stats-job command

The stats-job command gives statistical information about the specified job if
it exists. Its form is:

stats-job <id>\r\n

• is a job id.

The response is one of:

• "NOT_FOUND\r\n" if the job does not exist.

• "OK \r\n\r\n"

  • is the size of the following data section in bytes.

  • is a sequence of bytes of length from the previous line. It
    is a YAML file with statistical information represented a dictionary.

The stats-job data is a YAML file representing a single dictionary of strings
to scalars. It contains these keys:

• "id" is the job id

• "tube" is the name of the tube that contains this job

• "state" is "ready" or "delayed" or "reserved" or "buried"

• "pri" is the priority value set by the put, release, or bury commands.

• "age" is the time in seconds since the put command that created this job.

• "delay" is the integer number of seconds to wait before putting this job in
  the ready queue.

• "ttr" -- time to run -- is the integer number of seconds a worker is
  allowed to run this job.

• "time-left" is the number of seconds left until the server puts this job
  into the ready queue. This number is only meaningful if the job is
  reserved or delayed. If the job is reserved and this amount of time
  elapses before its state changes, it is considered to have timed out.

• "file" is the number of the earliest binlog file containing this job.
  If -b wasn't used, this will be 0.

• "reserves" is the number of times this job has been reserved.

• "timeouts" is the number of times this job has timed out during a
  reservation.

• "releases" is the number of times a client has released this job from a
  reservation.

• "buries" is the number of times this job has been buried.

• "kicks" is the number of times this job has been kicked.

Implement list-tubes command

The list-tubes command returns a list of all existing tubes. Its form is:

list-tubes\r\n

The response is:

OK <bytes>\r\n
<data>\r\n

• is the size of the following data section in bytes.

• is a sequence of bytes of length from the previous line. It
  is a YAML file containing all tube names as a list of strings.

Create SQL-based backend.

Implement PriorityQueue interface using PostgreSQL; MySQL is optional for now.

Inspired by: https://github.com/gree/futurama

Support all admin commands supported by beanstalkd

Currently beanstalkg do not support the any of the administrative commands supported by beanstalkd. These are needed to make it more devops friendly.

Will create an issue each if individual commands are too complex to be handled in one pull req.

unit test architecture/tube.go

https://github.com/vimukthi-git/beanstalkg/blob/master/architecture/queue.go needs to have unit tests.

Make stephenwithav a collaborator of the project

@stephenwithav I have made you a collaborator for the project.

Check if existing beanstalkd clients have a problem with job ids being string

I just noticed because beanstalkg uses string ids instead of integers as in the case of beanstalkd some clients available for beanstalkd might not work correctly for beanstalkg. I already found one which doesn't work, go client.

We need to check others and if they have problems we need to fork and modify and publish our own

cc
@stephenwithav @marekyggdrasil

Implement peak commands

The peek commands let the client inspect a job in the system. There are four
variations. All but the first operate only on the currently used tube.

• "peek \r\n" - return job .

• "peek-ready\r\n" - return the next ready job.

• "peek-delayed\r\n" - return the delayed job with the shortest delay left.

• "peek-buried\r\n" - return the next job in the list of buried jobs.

There are two possible responses, either a single line:

• "NOT_FOUND\r\n" if the requested job doesn't exist or there are no jobs in
  the requested state.

Or a line followed by a chunk of data, if the command was successful:

FOUND <id> <bytes>\r\n
<data>\r\n

• is the job id.

• is an integer indicating the size of the job body, not including
  the trailing "\r\n".

• is the job body -- a sequence of bytes of length from the
  previous line.

Rename log instances to logg.

Problem: a pre-save hook runs goimports to format the go files and automatically adds missing imports. When log is referenced outside of the file it is defined, goimports chooses from the packages you have defined. In my case, goimports prefers gommon/log.

Workaround: rename instances of log to logg to prevent goimports from addiing gommon/log every time.

Any objections? :)

dead?

subj

Add a docker file and dockerize beanstalkg

Add a docker file and dockerize beanstalkg, so that it can be run easily in a container environment

debug log at every tick on master branch

If someone clones our repo and simply runs beanstalkg will get flooded with output messages

https://github.com/Beanstalkg/beanstalkg/blob/85042b659fe17081a3345196a0129239d78b41a6/operation/tube_handler.go#L25

implement list-tube-used command

The list-tube-used command returns the tube currently being used by the
client. Its form is:

list-tube-used\r\n

The response is:

USING <tube>\r\n

• is the name of the tube being used.

Disconnected clients create problem with job serving for other clients

I notice that disconnected clients gets jobs served because beanstalkg doesn't reliably detect when they have gone away.

Update the readme to include details on how to run beanstalkg correctly

Update the readme to include details on how to run beanstalkg correctly in both,

1. standalone mode
2. proxy mode

Which interface is required for a new backend?

Is it PriorityQueue?

Add interfaces to allow jobs and job history to be persisted.

PR due later today.

consider introducing a cmd package for main

Introduce validation for command parameters in protocol config

This is in reference to a conversation we had on #55 .