The dispatcher is a service to run python tasks in subprocesses, designed specifically to work well with pg_notify, but intended to be extensible to other message delivery means. Its philosophy is to have a limited scope as a "local" runner of background tasks, but to be composable so that it can be "wrapped" easily to enable clustering and distributed task management by apps using it.
Warning
This project is in initial development. Expect many changes, including name, paths, and CLIs.
Licensed under Apache Software License 2.0
You have a postgres server configured and a python project. You will use dispatcher to trigger a background task over pg_notify. Both your background dispatcher service and your task publisher process must have python configured so that your task is importable. Instructions are broken into 3 steps:
- Library - Configure dispatcher, mark the python methods you will run with it
- Dispatcher service - Start your background task service, it will start listening
- Publisher - From some other script, submit tasks to be ran
In the "Manual Demo" section, an runnable example of this is given.
The dispatcher @task() decorator is used to register tasks.
The tests/data/methods.py module defines some
dispatcher tasks.
The decorator accepts some kwargs (like queue below) that will affect task behavior,
see docs/task_options.md.
from dispatcher.publish import task
@task(queue='test_channel')
def print_hello():
print('hello world!!')Configure dispatcher somewhere in your import path or before running the service. This tells dispatcher how to submit tasks to be ran.
from dispatcher.config import setup
config = {
"producers": {
"brokers": {
"pg_notify": {
"conninfo": "dbname=postgres user=postgres"
"channels": [
"test_channel",
],
},
},
},
"pool": {"max_workers": 4},
}
setup(config)For more on how to set up and the allowed options in the config,
see the section config docs.
The queue passed to @task needs to match a pg_notify channel in the config.
It is often useful to have different workers listen to different sets of channels.
The dispatcher service needs to be running before you submit tasks. This does not make any attempts at message durability or confirmation. If you submit a task in an outage of the service, it will be dropped.
There are 2 ways to run the dispatcher service:
- Importing and running (code snippet below)
- A CLI entrypoint
dispatcherdfor demo purposes
from dispatcher import run_service
# After the setup() method has been called
run_service()Configuration tells how to connect to postgres, and what channel(s) to listen to.
This assumes you configured python so that print_hello is importable
from the test_methods python module.
This method does not take any args or kwargs, but if it did, you would
pass those directly as in, .delay(*args, **kwargs).
The following code will submit print_hello to run in the background dispatcher service.
from test_methods import print_hello
# After the setup() method has been called
print_hello.delay()Also valid:
from test_methods import print_hello
# After the setup() method has been called
print_hello.apply_async(args=[], kwargs={})The difference is that apply_async takes both args and kwargs as kwargs themselves,
and allows for additional configuration parameters to come after those.
Initial setup:
pip install -e .[pg_notify]
To experience running a dispatcherd service, you can try this:
make postgres
dispatcherd
The dispatcherd entrypoint will look for a config file in the current
directory if not otherwise specified, which is dispatcher.yml
in this case. You can see it running some schedules and listening.
Ctl+c to stop that server.
The following will start up postgres, then start up 2 dispatcher services. It should take a few seconds, mainly waiting for postgres.
make demo
After it completes docker ps -a should show dispatcherd1 and dispatcherd2
containers as well as postgres. You can see logs via docker logs dispatcherd1.
These will accept task submissions. Submit a lot of tasks as a python
task publisher with the run_demo.py script. To get accurate replies,
we need to specify that 2 replies are expected because we are
communicating with 2 background task services.
./run_demo.py 2
You can talk to these services over postgres with dispatcherctl,
using the same local dispatcher.yml config.
dispatcherctl running
dispatcherctl workers
The "running" command will likely show scheduled tasks and leftover tasks from the demo.
For demo, the uuid and task options allow doing filtering.
dispatcherctl running --task=tests.data.methods.sleep_function
This would show any specific instance of tests.data.methods.sleep_function currently running.
Most tests (except for tests/unit/) require postgres to be running.
pip install -r requirements_dev.txt
make postgres
pytest tests/
This is intended to be a working space for prototyping a code split of:
https://github.com/ansible/awx/tree/devel/awx/main/dispatch
As a part of doing the split, we also want to resolve a number of long-standing design and sustainability issues, thus, asyncio. For a little more background see docs/design_notes.md.
There is documentation of the message formats used by the dispatcher
in docs/message_formats.md. Some of these are internal,
but some messages are what goes over the user-defined brokers (pg_notify).
You can trigger tasks using your own "publisher" code as an alternative
to attached methods like .apply_async. Doing this requires connecting
to postges and submitting a pg_notify message with JSON data
that conforms to the expected format.
The ./run_demo.py script shows examples of this, but borrows some
connection and config utilities to help.
We ask all of our community members and contributors to adhere to the Ansible code of conduct. If you have questions or need assistance, please reach out to our community team at [email protected]
Refer to the Contributing guide for further information.
See the Communication section of the Contributing guide to find out how to get help and contact us.
For more information about getting in touch, see the Ansible communication guide.
Dispatcher is sponsored by Red Hat, Inc.