TaskBunny
TaskBunny is a background processing application written in Elixir and uses RabbitMQ as a messaging backend.
Use cases
Although TaskBunny provides similar features to popular background processing libraries in other languages such as Resque, Sidekiq, RQ etc., you might not need it for a same reason. Erlang process or GenServer would always be your first choice for background processing in Elixir.
However you might want to try out TaskBunny in the following cases:
- You want to separate a background processing concern from your phoenix application
- You use container based deployment such as Heroku, Docker etc. and each deploy is immutable and disposable
- You want to have a control on retry and its interval on background job processing
- You want to schedule the job execution time
- You want to use a part of functionalities TaskBunny provides to talk to RabbitMQ
- You want to enqueue jobs from other system via RabbitMQ
- You want to control the concurrency to avoid making too much traffic
Getting started
1. Check requirements
- Elixir 1.4+
- RabbitMQ 3.6.0 or greater
2. Install TaskBunny
Edit mix.exs
and add task_bunny
to your list of dependencies and applications:
def deps do
[{:task_bunny, "~> 0.3.2"}]
end
def application do
[applications: [:task_bunny]]
end
Then run mix deps.get
.
3. Configure TaskBunny
Configure hosts and queues:
config :task_bunny, hosts: [
default: [connect_options: "amqp://localhost?heartbeat=30"]
]
config :task_bunny, queue: [
namespace: "task_bunny.",
queues: [[name: "normal", jobs: :default]]
]
4. Define TaskBunny job
Use TaskBunny.Job
module in your job module and define perform/1
that takes a map as an argument.
defmodule HelloJob do
use TaskBunny.Job
require Logger
def perform(%{"name" => name}) do
Logger.info("Hello #{name}")
:ok
end
end
Make sure you return :ok
or {:ok, something}
when the job was successfully processed.
Otherwise TaskBunny would treat the job as failed and move it to retry queue.
5. Enqueueing TaskBunny job
Then enqueue a job
HelloJob.enqueue!(%{"name" => "Cloud"})
The worker invokes the job with Hello Cloud
in your logger output.
Queues
Worker queue and sub queues
TaskBunny declares four queues for each worker queue on RabbitMQ.
config :task_bunny, queue: [
namespace: "task_bunny."
queues: [
[name: "normal", jobs: :default]
]
]
If have a config like above, TaskBunny will define these four queues on RabbitMQ:
- task_bunny.normal: main worker queue
- task_bunny.normal.retry: queue for retry
- task_bunny.normal.rejected: queue that stores jobs failed more than allowed times
- task_bunny.normal.delay: queue that stores jobs that are performed in the future
Reset queues
TaskBunny provides a mix task to reset queues. This task deletes the queues and creates them again. Existing messages in the queue will be lost so please be aware of this.
% mix task_bunny.queue.reset
You need to redefine a queue when you want to change the retry interval for a queue.
Umbrella app
When you use TaskBunny under an umbrella app and each apps needs a different queue definition, you can prefix config key like below so that it doesn’t overwrite the other configuration.
config :task_bunny, app_a_queue: [
namespace: "app_a.",
queues: [
[name: "normal", jobs: "AppA.*"]
]
]
config :task_bunny, app_b_queue: [
namespace: "app_b.",
queues: [
[name: "normal", jobs: "AppB.*"]
]
]
Enqueue job
Enqueue
TaskBunny.Job
will define enqueue/1
and enqueue!/1
to your job module.
Like other Elixir libraries, enqueue!/1
is similar to enqueue/1
but raises
an exception when it gets an error during the enqueue.
You can also use TaskBunny.Job.enqueue/2
which takes a module as a first
argument. The two examples below will give you the same result.
SampleJob.enqueue!()
TaskBunny.Job.enqueue!(SampleJob)
First expression is concise and preferred but the later expression lets you enqueue the job without defining job module. TaskBunny takes the module just as atom and doesn’t check module existence when it enqueues. It is useful when you have separate applications for enqueueing and performing.
Schedule job
When you don’t want to perform the job immediately you can use delay
options
when you enqueue the job.
SampleJob.enqueue!(delay: 10_000)
It will enqueue the job to the worker queue in 10 seconds.
When you use delay
option it enqueues the job to the delay queue.
The job will be moved to the worker queue after the specific time.
The move between those queues will be handled by RabbitMQ so the job will be enqueued safely even if your application dies after the call.
Enqueue job from other system
The message should be encoded in JSON format and set job and payload(argument). For example:
{
"job": "YourApp.HelloJob",
"payload": {"name": "Aerith"}
}
Then send the message to the worker queue - TaskBunny will process it.
Select queue
TaskBunny looks up config and chooses a right queue for the job.
config :task_bunny, queue: [
queues: [
[name: "default", jobs: :default],
[name: "fast_track", jobs: [YourApp.RushJob, YourApp.HurryJob],
[name: "analytics", jobs: "Analytics.*"]
]
]
You can configure it with module name(atom), string (support wildcard) or list of them. If the job matches one of them the queue will be chosen. If the doesn’t match any the queue with :default will be chosen.
YourApp.RushJob.enqueue(payload) #=> "fast_track"
Analytics.MiningJob.enqueue(payload) #=> "analytics"
YourApp.HelloJob.enqueue(payload) #=> "default"
If you pass the queue option TaskBunny will use it.
YourApp.HelloJob.enqueue(payload, queue: "fast_track") #=> "fast_track"
Workers
What is worker?
TaskBunny worker is a GenServer that processes jobs and handles errors. A worker listens to a single queue, receives messages(jobs) from it and invokes jobs to perform.
Concurrency
By default a TaskBunny worker runs two jobs concurrently. You can change the concurrency with the config.
config :task_bunny, queue: [
namespace: "task_bunny."
queues: [
[name: "default", jobs: :default, worker: [concurrency: 1]],
[name: "analytics", jobs: "Analytics.*", worker: [concurrency: 10]]
]
]
The concurrency is set per an application. If you run your application on five different hosts with above configuration, there can be 55 jobs performing simultaneously in total.
Disable worker
You can disable workers starting with your application by setting 1
, TRUE
or YES
to TASK_BUNNY_DISABLE_WORKER
environment variable.
% TASK_BUNNY_DISABLE_WORKER=1 mix phoenix.server
You can also disable workers in the config.
config :task_bunny, disable_worker: true
You can also disable worker running for a specific queue with the config.
config :task_bunny, queue: [
namespace: "task_bunny."
queues: [
[name: "default", jobs: :default],
[name: "analytics", jobs: "Analytics.*", worker: false]
]
]
With above, TaskBunny starts only a worker for the default queue.
Control job execution
Retry
TaskBunny marks the job failed when:
- job raises an exception or exits during
perform
perform
doesn’t return:ok
or{:ok, something}
perform
times out.
TaskBunny retries the job automatically if the job has failed. By default, it retries 10 times for every 5 minutes.
If you want to change it, you can override the value on a job module.
defmodule FlakyJob do
use TaskBunny.Job
require Logger
def max_retry, do: 100
def retry_interval(_), do: 10_000
...
end
In this example, it will retry 100 times for every 10 seconds. You can also change the retry_interval by the number of failures.
def max_retry, do: 5
def retry_interval(failed_count) do
# failed_count will be between 1 and 5.
# Gradually have longer retry interval
[10, 60, 300, 3_600, 7_200]
|> Enum.map(&(&1 * 1000))
|> Enum.at(failed_count - 1, 1000)
end
If a job fails more than max_retry
times, the payload is sent to jobs.[job_name].rejected
queue.
When a job gets rejected the on_reject
callback is called. By default it does nothing but you can override it.
It’s useful to execute recovery actions when a job fails (like sending an email to a customer for instance)
It receives the body containing the payload of the rejected job plus the full error trace. It returns :ok
defmodule FlakyJob do
use TaskBunny.Job
require Logger
def on_reject(_body) do
...
:ok
end
...
end
Immediately Reject
TaskBunny can mark a job as rejected without retrying when perform
returns :reject
or {:reject, something}
In this case any max_retry
config is ignored.
Timeout
By default, jobs timeout after 2 minutes. If job doesn’t respond for more than 2 minutes, worker kills the process and moves it to retry queue.
You can change the timeout by overriding timeout/0
in your job.
defmodule SlowJob do
use TaskBunny.Job
def timeout, do: 300_000
...
end
Connection management
TaskBunny provides an extra layer on top of the amqp connection module.
Configuration
TaskBunny automatically connects to RabbitMQ hosts in the config at the start of the application.
TaskBunny forwards connect_options
to AMQP.Connection.open/1.
config :task_bunny, hosts: [
default: [
connect_options: "amqp://rabbitmq.example.com?heartbeat=30"
],
legacy: [
connect_options: [
host: "legacy.example.com",
port: 15672,
username: "guest",
password: "bunny"
]
]
]
:default
host has a special meaning on TaskBunny: TaskBunny would select :default
host when you didn’t specify the host.
assert TaskBunny.Connection.get_connection() == TaskBunny.Connection.get_connection(:default)
You can specify the host to the queue:
config :task_bunny, queue: [
queues: [
[name: "normal", jobs: "MainApp.*"], # => :default host
[name: "normal", jobs: "Legacy.*", host: :legacy]
]
]
If you don’t want to start TaskBunny automatically in a specific environment, set true
to disable_auto_start
in the config:
config :task_bunny, disable_auto_start: true
Get connection
TaskBunny provides two ways to access the connections.
Most of time you want to use Connection.get_connection/1
or
Connection.get_connection!/1
that returns the connection synchronously.
conn = TaskBunny.Connection.get_connection()
legacy = TaskBunny.Connection.get_connection(:legacy)
TaskBunny also provides asynchronous API Connection.subscribe_connection/1
.
See the API documentation for more details.
Reconnection
TaskBunny automatically tries reconnecting to RabbitMQ if the connection is gone. All workers will restart automatically once the new connection is established.
TaskBunny aims to provide zero hassle and recover automatically regardless how long the host takes to come back and accessible.
Failure backends
By default, when the error occurs during the job execution TaskBunny reports it to Logger. If you want to report the error to different services, you can configure your custom failure backend.
config :task_bunny, failure_backend: [YourApp.CustomFailureBackend]
You can also report the errors to the multiple backends. For example, if you want to use our default Logger backend with your custom backend you can configure like below:
config :task_bunny, failure_backend: [
TaskBunny.FailureBackend.Logger,
YourApp.CustomFailureBackend
]
Check out the implementation of TaskBunny.FailureBackend.Logger to learn how to write your custom failure backend.
Implementations
(Send us a pull request if you want to add other implementation)
Monitoring
RabbitMQ plugins
RabbitMQ supports a variety of plugins. If you are not familiar with them we recommend you to look into those.
The following plugins will help you use RabbitMQ with TaskBunny.
- Management Plugin: provides an HTTP-based API for management and monitoring of your RabbitMQ server, along with a browser-based UI and a command line tool, rabbitmqadmin.
- Shovel Plugin: helps you to move messages(job) from a queue to another queue.
Wobserver integration
TaskBunny automatically integrates with Wobserver.
All worker and connection information will be added as a page on the web interface.
The current amount of job runners and job success, failure, and reject totals are added to the /metrics
endpoint.
Copyright and License
Copyright (c) 2017, SQUARE ENIX LTD.
TaskBunny code is licensed under the MIT License.