This document describes the current stable version of Celery (3.1). For development docs, go here.
Celery is a task queue with batteries included. It is easy to use so that you can get started without learning the full complexities of the problem it solves. It is designed around best practices so that your product can scale and integrate with other languages, and it comes with the tools and support you need to run such a system in production.
In this tutorial you will learn the absolute basics of using Celery. You will learn about;
Choosing and installing a message transport (broker).
Installing Celery and creating your first task
Starting the worker and calling tasks.
Keeping track of tasks as they transition through different states, and inspecting return values.
Celery may seem daunting at first - but don’t worry - this tutorial will get you started in no time. It is deliberately kept simple, so to not confuse you with advanced features. After you have finished this tutorial it’s a good idea to browse the rest of the documentation, for example the Next Steps tutorial, which will showcase Celery’s capabilities.
Celery requires a solution to send and receive messages, usually this comes in the form of a separate service called a message broker.
There are several choices available, including:
RabbitMQ is feature-complete, stable, durable and easy to install. It’s an excellent choice for a production environment. Detailed information about using RabbitMQ with Celery:
If you are using Ubuntu or Debian install RabbitMQ by executing this command:
$ sudo apt-get install rabbitmq-server
When the command completes the broker is already running in the background, ready to move messages for you: Starting rabbitmq-server: SUCCESS.
And don’t worry if you’re not running Ubuntu or Debian, you can go to this website to find similarly simple installation instructions for other platforms, including Microsoft Windows:
http://www.rabbitmq.com/download.html
Redis is also feature-complete, but is more susceptible to data loss in the event of abrupt termination or power failures. Detailed information about using Redis:
Using a database as a message queue is not recommended, but can be sufficient for very small installations. Your options include:
If you’re already using a Django database for example, using it as your message broker can be convenient while developing even if you use a more robust system in production.
In addition to the above, there are other experimental transport implementations to choose from, including Amazon SQS, Using MongoDB and IronMQ.
See Broker Overview for a full list.
Celery is on the Python Package Index (PyPI), so it can be installed with standard Python tools like pip or easy_install:
$ pip install celery
The first thing you need is a Celery instance, this is called the celery application or just app in short. Since this instance is used as the entry-point for everything you want to do in Celery, like creating tasks and managing workers, it must be possible for other modules to import it.
In this tutorial you will keep everything contained in a single module, but for larger projects you want to create a dedicated module.
Let’s create the file tasks.py:
from celery import Celeryapp = Celery('tasks', broker='amqp://guest@localhost//')@app.taskdef add(x, y): return x + y
The first argument to Celery is the name of the current module, this is needed so that names can be automatically generated, the second argument is the broker keyword argument which specifies the URL of the message broker you want to use, using RabbitMQ here, which is already the default option. See Choosing a Broker above for more choices, e.g. for RabbitMQ you can use amqp://localhost, or for Redis you can use redis://localhost.
You defined a single task, called add, which returns the sum of two numbers.
You now run the worker by executing our program with the worker argument:
$ celery -A tasks worker --loglevel=info
Note
See the Troubleshooting section if the worker does not start.
In production you will want to run the worker in the background as a daemon. To do this you need to use the tools provided by your platform, or something like supervisord (seeRunning the worker as a daemon for more information).
For a complete listing of the command-line options available, do:
$ celery worker --help
There are also several other commands available, and help is also available:
$ celery help
To call our task you can use the delay() method.
This is a handy shortcut to the apply_async() method which gives greater control of the task execution (see Calling Tasks):
>>> from tasks import add>>> add.delay(4, 4)
The task has now been processed by the worker you started earlier, and you can verify that by looking at the workers console output.
Calling a task returns an AsyncResult instance, which can be used to check the state of the task, wait for the task to finish or get its return value (or if the task failed, the exception and traceback). But this isn’t enabled by default, and you have to configure Celery to use a result backend, which is detailed in the next section.
If you want to keep track of the tasks’ states, Celery needs to store or send the states somewhere. There are several built-in result backends to choose from:SQLAlchemy/Django ORM, Memcached, Redis, AMQP (RabbitMQ), and MongoDB – or you can define your own.
For this example you will use the amqp result backend, which sends states as messages. The backend is specified via the backend argument to Celery, (or via theCELERY_RESULT_BACKEND setting if you choose to use a configuration module):
app = Celery('tasks', backend='amqp', broker='amqp://')
or if you want to use Redis as the result backend, but still use RabbitMQ as the message broker (a popular combination):
app = Celery('tasks', backend='redis://localhost', broker='amqp://')
To read more about result backends please see Result Backends.
Now with the result backend configured, let’s call the task again. This time you’ll hold on to the AsyncResult instance returned when you call a task:
>>> result = add.delay(4, 4)
The ready() method returns whether the task has finished processing or not:
>>> result.ready()False
You can wait for the result to complete, but this is rarely used since it turns the asynchronous call into a synchronous one:
>>> result.get(timeout=1)8
In case the task raised an exception, get() will re-raise the exception, but you can override this by specifying the propagate argument:
>>> result.get(propagate=False)
If the task raised an exception you can also gain access to the original traceback:
>>> result.traceback…
See celery.result for the complete result object reference.
Celery, like a consumer appliance doesn’t need much to be operated. It has an input and an output, where you must connect the input to a broker and maybe the output to a result backend if so wanted. But if you look closely at the back there’s a lid revealing loads of sliders, dials and buttons: this is the configuration.
The default configuration should be good enough for most uses, but there’s many things to tweak so Celery works just the way you want it to. Reading about the options available is a good idea to get familiar with what can be configured. You can read about the options in the Configuration and defaults reference.
The configuration can be set on the app directly or by using a dedicated configuration module. As an example you can configure the default serializer used for serializing task payloads by changing the CELERY_TASK_SERIALIZER setting:
app.conf.CELERY_TASK_SERIALIZER = 'json'
If you are configuring many settings at once you can use update:
app.conf.update( CELERY_TASK_SERIALIZER='json', CELERY_ACCEPT_CONTENT=['json'], # Ignore other content CELERY_RESULT_SERIALIZER='json', CELERY_TIMEZONE='Europe/Oslo', CELERY_ENABLE_UTC=True,)
For larger projects using a dedicated configuration module is useful, in fact you are discouraged from hard coding periodic task intervals and task routing options, as it is much better to keep this in a centralized location, and especially for libraries it makes it possible for users to control how they want your tasks to behave, you can also imagine your SysAdmin making simple changes to the configuration in the event of system trouble.
You can tell your Celery instance to use a configuration module, by calling theconfig_from_object() method:
app.config_from_object('celeryconfig')
This module is often called “celeryconfig”, but you can use any module name.
A module named celeryconfig.py must then be available to load from the current directory or on the Python path, it could look like this:
celeryconfig.py:
BROKER_URL = 'amqp://'CELERY_RESULT_BACKEND = 'amqp://'CELERY_TASK_SERIALIZER = 'json'CELERY_RESULT_SERIALIZER = 'json'CELERY_ACCEPT_CONTENT=['json']CELERY_TIMEZONE = 'Europe/Oslo'CELERY_ENABLE_UTC = True
To verify that your configuration file works properly, and doesn’t contain any syntax errors, you can try to import it:
$ python -m celeryconfig
For a complete reference of configuration options, see Configuration and defaults.
To demonstrate the power of configuration files, this how you would route a misbehaving task to a dedicated queue:
celeryconfig.py:
CELERY_ROUTES = { 'tasks.add': 'low-priority',}
Or instead of routing it you could rate limit the task instead, so that only 10 tasks of this type can be processed in a minute (10/m):
celeryconfig.py:
CELERY_ANNOTATIONS = { 'tasks.add': {'rate_limit': '10/m'}}
If you are using RabbitMQ or Redis as the broker then you can also direct the workers to set a new rate limit for the task at runtime:
$ celery control rate_limit tasks.add 10/m worker@example.com: OK new rate limit set successfully
See Routing Tasks to read more about task routing, and the CELERY_ANNOTATIONSsetting for more about annotations, or Monitoring and Management Guide for more about remote control commands, and how to monitor what your workers are doing.
If you want to learn more you should continue to the Next Steps tutorial, and after that you can study the User Guide.
There’s also a troubleshooting section in the Frequently Asked Questions.
If you’re using Debian, Ubuntu or other Debian-based distributions:
Debian recently renamed the /dev/shm special file to /run/shm.
A simple workaround is to create a symbolic link:
# ln -s /run/shm /dev/shm
Others:
If you provide any of the --pidfile, --logfile or --statedb arguments, then you must make sure that they point to a file/directory that is writable and readable by the user starting the worker.
All tasks are PENDING by default, so the state would have been better named “unknown”. Celery does not update any state when a task is sent, and any task with no history is assumed to be pending (you know the task id after all).
Make sure that the task does not have ignore_result enabled.
Enabling this option will force the worker to skip updating states.
Make sure the CELERY_IGNORE_RESULT setting is not enabled.
Make sure that you do not have any old workers still running.
It’s easy to start multiple workers by accident, so make sure that the previous worker is properly shutdown before you start a new one.
An old worker that is not configured with the expected result backend may be running and is hijacking the tasks.
The –pidfile argument can be set to an absolute path to make sure this doesn’t happen.
Make sure the client is configured with the right backend.
If for some reason the client is configured to use a different backend than the worker, you will not be able to receive the result, so make sure the backend is correct by inspecting it:
>>> result = task.delay(…) >>> print(result.backend)
This document describes the current stable version of Celery (3.1). For development docs, go here.
The First Steps with Celery guide is intentionally minimal. In this guide I will demonstrate what Celery offers in more detail, including how to add Celery support for your application and library.
This document does not document all of Celery’s features and best practices, so it’s recommended that you also read the User Guide
Project layout:
proj/__init__.py /celery.py /tasks.py
proj/celery.py
from __future__ import absolute_importfrom celery import Celeryapp = Celery('proj', broker='amqp://', backend='amqp://', include=['proj.tasks'])# Optional configuration, see the application user guide.app.conf.update( CELERY_TASK_RESULT_EXPIRES=3600,)if __name__ == '__main__': app.start()
In this module you created our Celery instance (sometimes referred to as the app). To use Celery within your project you simply import this instance.
The broker argument specifies the URL of the broker to use.
See Choosing a Broker for more information.
The backend argument specifies the result backend to use,
It’s used to keep track of task state and results. While results are disabled by default I use the amqp result backend here because I demonstrate how retrieving results work later, you may want to use a different backend for your application. They all have different strengths and weaknesses. If you don’t need results it’s better to disable them. Results can also be disabled for individual tasks by setting the @task(ignore_result=True) option.
See Keeping Results for more information.
The include argument is a list of modules to import when the worker starts. You need to add our tasks module here so that the worker is able to find our tasks.
proj/tasks.py
from __future__ import absolute_importfrom proj.celery import app@app.taskdef add(x, y): return x + y@app.taskdef mul(x, y): return x * y@app.taskdef xsum(numbers): return sum(numbers)
The celery program can be used to start the worker:
$ celery worker --app=proj -l info
When the worker starts you should see a banner and some messages:
-------------- celery@halcyon.local v3.1 (Cipater)---- **** -------- * *** * -- [Configuration]-- * - **** --- . broker: amqp://guest@localhost:5672//- ** ---------- . app: __main__:0x1012d8590- ** ---------- . concurrency: 8 (processes)- ** ---------- . events: OFF (enable -E to monitor this worker)- ** ----------- *** --- * --- [Queues]-- ******* ---- . celery: exchange:celery(direct) binding:celery--- ***** -----[2012-06-08 16:23:51,078: WARNING/MainProcess] celery@halcyon.local has started.
– The broker is the URL you specifed in the broker argument in our celery module, you can also specify a different broker on the command-line by using the -b option.
– Concurrency is the number of prefork worker process used to process your tasks concurrently, when all of these are busy doing work new tasks will have to wait for one of the tasks to finish before it can be processed.
The default concurrency number is the number of CPU’s on that machine (including cores), you can specify a custom number using -c option. There is no recommended value, as the optimal number depends on a number of factors, but if your tasks are mostly I/O-bound then you can try to increase it, experimentation has shown that adding more than twice the number of CPU’s is rarely effective, and likely to degrade performance instead.
Including the default prefork pool, Celery also supports using Eventlet, Gevent, and threads (see Concurrency).
– Events is an option that when enabled causes Celery to send monitoring messages (events) for actions occurring in the worker. These can be used by monitor programs like celery events, and Flower - the real-time Celery monitor, which you can read about in the Monitoring and Management guide.
– Queues is the list of queues that the worker will consume tasks from. The worker can be told to consume from several queues at once, and this is used to route messages to specific workers as a means for Quality of Service, separation of concerns, and emulating priorities, all described in the Routing Guide.
You can get a complete list of command-line arguments by passing in the –help flag:
$ celery worker --help
These options are described in more detailed in the Workers Guide.
Stopping the worker
To stop the worker simply hit Ctrl+C. A list of signals supported by the worker is detailed in the Workers Guide.
In the background
In production you will want to run the worker in the background, this is described in detail in the daemonization tutorial.
The daemonization scripts uses the celery multi command to start one or more workers in the background:
$ celery multi start w1 -A proj -l info celery multi v3.1.1 (Cipater) > Starting nodes... > w1.halcyon.local: OK
You can restart it too:
$ celery multi restart w1 -A proj -l info celery multi v3.1.1 (Cipater) > Stopping nodes... > w1.halcyon.local: TERM -> 64024 > Waiting for 1 node..... > w1.halcyon.local: OK > Restarting node w1.halcyon.local: OK celery multi v3.1.1 (Cipater) > Stopping nodes... > w1.halcyon.local: TERM -> 64052
or stop it:
$ celery multi stop w1 -A proj -l info
The stop command is asynchronous so it will not wait for the worker to shutdown. You will probably want to use the stopwait command instead which will ensure all currently executing tasks is completed:
$ celery multi stopwait w1 -A proj -l info
Note
celery multi doesn’t store information about workers so you need to use the same command-line arguments when restarting. Only the same pidfile and logfile arguments must be used when stopping.
By default it will create pid and log files in the current directory, to protect against multiple workers launching on top of each other you are encouraged to put these in a dedicated directory:
$ mkdir -p /var/run/celery $ mkdir -p /var/log/celery $ celery multi start w1 -A proj -l info --pidfile=/var/run/celery/%n.pid \ --logfile=/var/log/celery/%n.pid
With the multi command you can start multiple workers, and there is a powerful command-line syntax to specify arguments for different workers too, e.g:
$ celery multi start 10 -A proj -l info -Q:1-3 images,video -Q:4,5 data \ -Q default -L:4,5 debug
For more examples see the multi module in the API reference.
About the --app argument
The --app argument specifies the Celery app instance to use, it must be in the form ofmodule.path:attribute
But it also supports a shortcut form If only a package name is specified, where it’ll try to search for the app instance, in the following order:
With --app=proj:
an attribute named proj.app, or
an attribute named proj.celery, or
any attribute in the module proj where the value is a Celery application, or
If none of these are found it’ll try a submodule named proj.celery:
an attribute named proj.celery.app, or
an attribute named proj.celery.celery, or
Any atribute in the module proj.celery where the value is a Celery application.
This scheme mimics the practices used in the documentation, i.e. proj:app for a single contained module, and proj.celery:app for larger projects.
You can call a task using the delay() method:
>>> add.delay(2, 2)
This method is actually a star-argument shortcut to another method calledapply_async():
>>> add.apply_async((2, 2))
The latter enables you to specify execution options like the time to run (countdown), the queue it should be sent to and so on:
>>> add.apply_async((2, 2), queue='lopri', countdown=10)
In the above example the task will be sent to a queue named lopri and the task will execute, at the earliest, 10 seconds after the message was sent.
Applying the task directly will execute the task in the current process, so that no message is sent:
>>> add(2, 2)4
These three methods - delay(), apply_async(), and applying (__call__), represents the Celery calling API, which are also used for subtasks.
A more detailed overview of the Calling API can be found in the Calling User Guide.
Every task invocation will be given a unique identifier (an UUID), this is the task id.
The delay and apply_async methods return an AsyncResult instance, which can be used to keep track of the tasks execution state. But for this you need to enable a result backend so that the state can be stored somewhere.
Results are disabled by default because of the fact that there is no result backend that suits every application, so to choose one you need to consider the drawbacks of each individual backend. For many tasks keeping the return value isn’t even very useful, so it’s a sensible default to have. Also note that result backends are not used for monitoring tasks and workers, for that Celery uses dedicated event messages (seeMonitoring and Management Guide).
If you have a result backend configured you can retrieve the return value of a task:
>>> res = add.delay(2, 2)>>> res.get(timeout=1)4
You can find the task’s id by looking at the id attribute:
>>> res.idd6b3aea2-fb9b-4ebc-8da4-848818db9114
You can also inspect the exception and traceback if the task raised an exception, in factresult.get() will propagate any errors by default:
>>> res = add.delay(2) >>> res.get(timeout=1) Traceback (most recent call last): File "<stdin>", line 1, in <module> File "/opt/devel/celery/celery/result.py", line 113, in get interval=interval) File "/opt/devel/celery/celery/backends/amqp.py", line 138, in wait_for raise self.exception_to_python(meta['result']) TypeError: add() takes exactly 2 arguments (1 given)
If you don’t wish for the errors to propagate then you can disable that by passing thepropagate argument:
>>> res.get(propagate=False)TypeError('add() takes exactly 2 arguments (1 given)',)
In this case it will return the exception instance raised instead, and so to check whether the task succeeded or failed you will have to use the corresponding methods on the result instance:
>>> res.failed()True>>> res.successful()False
So how does it know if the task has failed or not? It can find out by looking at the tasksstate:
>>> res.state'FAILURE'
A task can only be in a single state, but it can progress through several states. The stages of a typical task can be:
PENDING -> STARTED -> SUCCESS
The started state is a special state that is only recorded if the CELERY_TRACK_STARTEDsetting is enabled, or if the @task(track_started=True) option is set for the task.
The pending state is actually not a recorded state, but rather the default state for any task id that is unknown, which you can see from this example:
>>> from proj.celery import app>>> res = app.AsyncResult('this-id-does-not-exist')>>> res.state'PENDING'
If the task is retried the stages can become even more complex, e.g, for a task that is retried two times the stages would be:
PENDING -> STARTED -> RETRY -> STARTED -> RETRY -> STARTED -> SUCCESS
To read more about task states you should see the States section in the tasks user guide.
Calling tasks is described in detail in the Calling Guide.
You just learned how to call a task using the tasks delay method, and this is often all you need, but sometimes you may want to pass the signature of a task invocation to another process or as an argument to another function, for this Celery uses something called subtasks.
A subtask wraps the arguments and execution options of a single task invocation in a way such that it can be passed to functions or even serialized and sent across the wire.
You can create a subtask for the add task using the arguments (2, 2), and a countdown of 10 seconds like this:
>>> add.subtask((2, 2), countdown=10)tasks.add(2, 2)
There is also a shortcut using star arguments:
>>> add.s(2, 2)tasks.add(2, 2)
Subtask instances also supports the calling API, which means that they have the delayand apply_async methods.
But there is a difference in that the subtask may already have an argument signature specified. The add task takes two arguments, so a subtask specifying two arguments would make a complete signature:
>>> s1 = add.s(2, 2)>>> res = s1.delay()>>> res.get()4
But, you can also make incomplete signatures to create what we call partials:
# incomplete partial: add(?, 2)>>> s2 = add.s(2)
s2 is now a partial subtask that needs another argument to be complete, and this can be resolved when calling the subtask:
# resolves the partial: add(8, 2)>>> res = s2.delay(8)>>> res.get()10
Here you added the argument 8, which was prepended to the existing argument 2 forming a complete signature of add(8, 2).
Keyword arguments can also be added later, these are then merged with any existing keyword arguments, but with new arguments taking precedence:
>>> s3 = add.s(2, 2, debug=True)>>> s3.delay(debug=False) # debug is now False.
As stated subtasks supports the calling API, which means that:
subtask.apply_async(args=(), kwargs={}, **options)
Calls the subtask with optional partial arguments and partial keyword arguments. Also supports partial execution options.
subtask.delay(*args, **kwargs)
Star argument version of apply_async. Any arguments will be prepended to the arguments in the signature, and keyword arguments is merged with any existing keys.
So this all seems very useful, but what can you actually do with these? To get to that I must introduce the canvas primitives…
The primitives are subtasks themselves, so that they can be combined in any number of ways to compose complex workflows.
Note
These examples retrieve results, so to try them out you need to configure a result backend. The example project above already does that (see the backend argument toCelery).
Let’s look at some examples:
Groups
A group calls a list of tasks in parallel, and it returns a special result instance that lets you inspect the results as a group, and retrieve the return values in order.
>>> from celery import group>>> from proj.tasks import add>>> group(add.s(i, i) for i in xrange(10))().get()[0, 2, 4, 6, 8, 10, 12, 14, 16, 18]
Partial group
>>> g = group(add.s(i) for i in xrange(10))>>> g(10).get()[10, 11, 12, 13, 14, 15, 16, 17, 18, 19]
Chains
Tasks can be linked together so that after one task returns the other is called:
>>> from celery import chain>>> from proj.tasks import add, mul# (4 + 4) * 8>>> chain(add.s(4, 4) | mul.s(8))().get()64
or a partial chain:
# (? + 4) * 8>>> g = chain(add.s(4) | mul.s(8))>>> g(4).get()64
Chains can also be written like this:
>>> (add.s(4, 4) | mul.s(8))().get()64
Chords
A chord is a group with a callback:
>>> from celery import chord>>> from proj.tasks import add, xsum>>> chord((add.s(i, i) for i in xrange(10)), xsum.s())().get()90
A group chained to another task will be automatically converted to a chord:
>>> (group(add.s(i, i) for i in xrange(10)) | xsum.s())().get()90
Since these primitives are all of the subtask type they can be combined almost however you want, e.g:
>>> upload_document.s(file) | group(apply_filter.s() for filter in filters)
Be sure to read more about workflows in the Canvas user guide.
Celery supports all of the routing facilities provided by AMQP, but it also supports simple routing where messages are sent to named queues.
The CELERY_ROUTES setting enables you to route tasks by name and keep everything centralized in one location:
app.conf.update( CELERY_ROUTES = { 'proj.tasks.add': {'queue': 'hipri'}, },)
You can also specify the queue at runtime with the queue argument to apply_async:
>>> from proj.tasks import add>>> add.apply_async((2, 2), queue='hipri')
You can then make a worker consume from this queue by specifying the -Q option:
$ celery -A proj worker -Q hipri
You may specify multiple queues by using a comma separated list, for example you can make the worker consume from both the default queue, and the hipri queue, where the default queue is named celery for historical reasons:
$ celery -A proj worker -Q hipri,celery
The order of the queues doesn’t matter as the worker will give equal weight to the queues.
To learn more about routing, including taking use of the full power of AMQP routing, see the Routing Guide.
If you’re using RabbitMQ (AMQP), Redis or MongoDB as the broker then you can control and inspect the worker at runtime.
For example you can see what tasks the worker is currently working on:
$ celery -A proj inspect active
This is implemented by using broadcast messaging, so all remote control commands are received by every worker in the cluster.
You can also specify one or more workers to act on the request using the --destinationoption, which is a comma separated list of worker host names:
$ celery -A proj inspect active --destination=celery@example.com
If a destination is not provided then every worker will act and reply to the request.
The celery inspect command contains commands that does not change anything in the worker, it only replies information and statistics about what is going on inside the worker. For a list of inspect commands you can execute:
$ celery -A proj inspect --help
Then there is the celery control command, which contains commands that actually changes things in the worker at runtime:
$ celery -A proj control --help
For example you can force workers to enable event messages (used for monitoring tasks and workers):
$ celery -A proj control enable_events
When events are enabled you can then start the event dumper to see what the workers are doing:
$ celery -A proj events --dump
or you can start the curses interface:
$ celery -A proj events
when you’re finished monitoring you can disable events again:
$ celery -A proj control disable_events
The celery status command also uses remote control commands and shows a list of online workers in the cluster:
$ celery -A proj status
You can read more about the celery command and monitoring in the Monitoring Guide.
All times and dates, internally and in messages uses the UTC timezone.
When the worker receives a message, for example with a countdown set it converts that UTC time to local time. If you wish to use a different timezone than the system timezone then you must configure that using the CELERY_TIMEZONE setting:
app.conf.CELERY_TIMEZONE = 'Europe/London'
The default configuration is not optimized for throughput by default, it tries to walk the middle way between many short tasks and fewer long tasks, a compromise between throughput and fair scheduling.
If you have strict fair scheduling requirements, or want to optimize for throughput then you should read the Optimizing Guide.
If you’re using RabbitMQ then you should install the librabbitmq module, which is an AMQP client implemented in C:
$ pip install librabbitmq
Now that you have read this document you should continue to the User Guide.
There’s also an API reference if you are so in.
NOTE: We need to add user and give permissions to '/' not just the vhost.
hostnames have to be properly setup per rabitmq.
Broker should be set up as: amqp://<user>:<password>@nomad-monitor//