1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
|
Background Tasks in AWX
=======================
AWX runs a lot of Python code asynchronously _in the background_ - meaning
_outside_ of the context of an HTTP request, such as:
* Any time a Job is launched in AWX (a Job Template, an Adhoc Command, a Project
Update, an Inventory Update, a System Job), a background process retrieves
metadata _about_ that job from the database and forks some process (e.g.,
`ansible-playbook`, `awx-manage inventory_import`)
* Certain expensive or time consuming tasks run in the background
asynchronously (like deleting an inventory).
* AWX runs a variety of periodic background tasks on a schedule. Some examples
are:
- AWX's "Task Manager/Scheduler" wakes up periodically and looks for
`pending` jobs that have been launched and are ready to start running.
- AWX periodically runs code that looks for scheduled jobs and launches
them.
- AWX runs a variety of periodic tasks that clean up temporary files, and
perform various administrative checks
- Every node in an AWX cluster runs a periodic task that serves as
a heartbeat and capacity check
Tasks, Queues and Workers
----------------
To accomplish this, AWX makes use of a "Task Queue" abstraction. Task Queues
are used as a mechanism to distribute work across machines in an AWX
installation. A Task Queue's input is a unit of work called a Task. Dedicated
worker processes running on every AWX node constantly monitor these queues for
new work to perform.
AWX communicates with these worker processes via AMQP - using RabbitMQ,
specifically - to mediate between clients and workers. To initiate a task, the
client (generally, Python code in the AWX API) publishes a message to a queue,
and RabbitMQ then delivers that message to one or more workers.
Clustered AWX installations consist of multiple workers spread across every
node, giving way to high availability and horizontal scaling.
Direct vs Fanout Messages
-------------------------
AWX publishes tasks in two distinct ways.
*Direct* messages are bound _directly_ to a specific named queue. When you launch
a Job Template in AWX, it looks at the available capacity of the various nodes
in your cluster and chooses an `Execution Node` where the playbook will run.
In this scenario, AWX publishes a message to a direct queue associated with
that AWX node. The dispatcher process running on that AWX node is specifically
bound to _listen_ for new events on their instance-specific queue.
Certain direct queues in AWX are bound to by _every_ AWX node. For example,
when an inventory deletion task is published, any available node across the
entire AWX may perform the work. Under _direct_ exchanges, every published
message is consumed and handled by *one* worker process.
*Fanout* messages are sent out in a broadcast fashion. When you change
a setting value in the AWX API, a fanout message is broadcast to _every_ AWX
node in your cluster, and code runs on _every_ node.
Defining and Running Tasks
--------------------------
Tasks are defined in AWX's source code, and generally live in the
`awx.main.tasks` module. Tasks can be defined as simple functions:
from awx.main.dispatch.publish import task
@task()
def add(a, b):
return a + b
...or classes that define a `run` method:
@task()
class Adder:
def run(self, a, b):
return a + b
To publish a task and run it in the background, use the `apply_async()`
function:
add.apply_async([1, 1])
Adder.apply_async([1, 1])
When you run this function, a JSON message is composed and published to the
appropriate AMQP queue:
{
"uuid": "<some_unique_string>",
"args": [1, 1],
"kwargs": {},
"task": "awx.main.tasks.add"
}
When a background worker receives the message, it deserializes it and runs the
associated Python code:
awx.main.tasks.add(123)
Dispatcher Implementation
-------------------------
Every node in an AWX install runs `awx-manage run_dispatcher`, a Python process
that uses the `kombu` library to consume messages from the appropriate queues
for that node (the default shared queue, a queue specific to the node's
hostname, and the broadcast queue). The Dispatcher process manages a pool of
child processes that it distributes inbound messages to. These worker
processes perform the actual work of deserializing published tasks and running
the associated Python code.
Heartbeats, Capacity, and Job Reaping
------------------------------------
One of the most important tasks in a clustered AWX installation is the periodic
heartbeat task. This task runs periodically on _every_ node, and is used to
record a heartbeat and system capacity for that node (which is used by the
scheduler when determining where to placed launched jobs).
If a node in an AWX cluster discovers that one of its peers has not updated its
heartbeat within a certain grace period, it is assumed to be offline, and its
capacity is set to zero to avoid scheduling new tasks on that node.
Additionally, jobs allegedly running or scheduled to run on that node are
assumed to be lost, and "reaped", or marked as failed.
Debugging
---------
`awx-manage run_dispatcher` includes a few flags that allow interaction and
debugging:
```
[root@awx /]# awx-manage run_dispatcher --status
2018-09-14 18:39:22,223 WARNING awx.main.dispatch checking dispatcher status for awx
awx[pid:9610] workers total=4 min=4 max=60
. worker[pid:9758] sent=12 finished=12 qsize=0 rss=106.730MB [IDLE]
. worker[pid:9769] sent=5 finished=5 qsize=0 rss=105.141MB [IDLE]
. worker[pid:9782] sent=5 finished=4 qsize=1 rss=110.430MB
- running 0c1deb4d-25ae-49a9-804f-a8afd05aff29 RunJob(*[9])
. worker[pid:9787] sent=3 finished=3 qsize=0 rss=101.824MB [IDLE]
```
This outputs running and queued task UUIDs handled by a specific dispatcher
(which corresponds to `main_unifiedjob.celery_task_id` in the database):
```
[root@awx /]# awx-manage run_dispatcher --running
2018-09-14 18:39:22,223 WARNING awx.main.dispatch checking dispatcher running for awx
['eb3b0a83-86da-413d-902a-16d7530a6b25', 'f447266a-23da-42b4-8025-fe379d2db96f']
```
Additionally, you can tell the local running dispatcher to recycle all of the
workers in its pool. It will wait for any running jobs to finish and exit when
work has completed, spinning up replacement workers.
```
awx-manage run_dispatcher --reload
```
|
