Руководство по воркерам

Завершение работы воркера

Для описания различных этапов завершения работы воркера мы будем использовать термины мягкое, постепенное, холодное и принудительное завершение. Воркер начинает процесс завершения работы, получив сигнал TERM или QUIT. Сигнал INT (Ctrl-C) также обрабатывается во время завершения работы и всегда запускает следующий этап этого процесса.

[INFO/MainProcess] Task myapp.long_running_task[6f748357-b2c7-456a-95de-f05c00504042] received
[WARNING/ForkPoolWorker-8] long_running_task is running, sleeping 1/2000s
[WARNING/ForkPoolWorker-8] long_running_task is running, sleeping 2/2000s
[WARNING/ForkPoolWorker-8] long_running_task is running, sleeping 3/2000s
^C
worker: Hitting Ctrl+C again will initiate cold shutdown, terminating all running tasks!

worker: Warm shutdown (MainProcess)
[WARNING/ForkPoolWorker-8] long_running_task is running, sleeping 4/2000s
[WARNING/ForkPoolWorker-8] long_running_task is running, sleeping 5/2000s
[WARNING/ForkPoolWorker-8] long_running_task is running, sleeping 6/2000s
^C
worker: Hitting Ctrl+C again will terminate all running tasks!
[WARNING/MainProcess] Initiating Soft Shutdown, terminating in 3 seconds
[WARNING/ForkPoolWorker-8] long_running_task is running, sleeping 7/2000s
[WARNING/ForkPoolWorker-8] long_running_task is running, sleeping 8/2000s
[WARNING/ForkPoolWorker-8] long_running_task is running, sleeping 9/2000s
[WARNING/MainProcess] Restoring 1 unacknowledged message(s)

[INFO/MainProcess] Task myapp.long_running_task[7235ac16-543d-4fd5-a9e1-2d2bb8ab630a] received
[WARNING/ForkPoolWorker-8] long_running_task is running, sleeping 1/2000s
[WARNING/ForkPoolWorker-8] long_running_task is running, sleeping 2/2000s
^C
worker: Hitting Ctrl+C again will initiate cold shutdown, terminating all running tasks!

worker: Warm shutdown (MainProcess)
[WARNING/ForkPoolWorker-8] long_running_task is running, sleeping 3/2000s
[WARNING/ForkPoolWorker-8] long_running_task is running, sleeping 4/2000s
^C
worker: Hitting Ctrl+C again will terminate all running tasks!
[WARNING/MainProcess] Initiating Soft Shutdown, terminating in 10 seconds
[WARNING/ForkPoolWorker-8] long_running_task is running, sleeping 5/2000s
[WARNING/ForkPoolWorker-8] long_running_task is running, sleeping 6/2000s
^C
Waiting gracefully for cold shutdown to complete...

worker: Cold shutdown (MainProcess)
^C[WARNING/MainProcess] Restoring 1 unacknowledged message(s)

Подстановки имени узла

• %p: полное имя узла.

• %h: имя хоста, включая доменное имя.

• %n: только имя хоста.

• %d: только доменное имя.

• %i: индекс процесса пула prefork или 0 для MainProcess.

• %I: индекс процесса пула prefork с разделителем.

Например, если текущее имя хоста — george@foo.example.com, подстановка даст следующие значения:

• --logfile=%p.log -> george@foo.example.com.log

• --logfile=%h.log -> foo.example.com.log

• --logfile=%n.log -> george.log

• --logfile=%d.log -> example.com.log

Индекс процесса пула prefork

Спецификаторы индекса процесса пула prefork подставляются в различные имена файлов в зависимости от процесса, которому в итоге потребуется открыть файл.

Это можно использовать, чтобы задать отдельный файл журнала для каждого дочернего процесса.

Обратите внимание: номера останутся в пределах лимита процессов, даже если процессы завершаются или используются автомасштабирование, maxtasksperchild или временные лимиты. Иными словами, это индекс процесса, а не количество процессов и не pid.

• %i - индекс процесса пула или 0 для MainProcess.

  В этом случае -n worker1@example.com -c2 -f %n-%i.log создаст три файла журнала:

  • worker1-0.log (основной процесс)

  • worker1-1.log (процесс пула 1)

  • worker1-2.log (процесс пула 2)

• %I - индекс процесса пула с разделителем.

  В этом случае -n worker1@example.com -c2 -f %n%I.log создаст три файла журнала:

  • worker1.log (основной процесс)

  • worker1-1.log (процесс пула 1)

  • worker1-2.log (процесс пула 2)

Функция broadcast()

Эта функция клиента используется для отправки команд воркерам. Для некоторых команд удалённого управления также доступны интерфейсы более высокого уровня, использующие в фоновом режиме broadcast(), например rate_limit() и ping().

Отправка команды rate_limit и именованных аргументов:

>>> app.control.broadcast('rate_limit',
...                          arguments={'task_name': 'myapp.mytask',
...                                     'rate_limit': '200/m'})

Команда будет отправлена асинхронно, без ожидания ответа. Чтобы запросить ответ, используйте аргумент reply:

>>> app.control.broadcast('rate_limit', {
...     'task_name': 'myapp.mytask', 'rate_limit': '200/m'}, reply=True)
[{'worker1.example.com': 'New rate limit set successfully'},
 {'worker2.example.com': 'New rate limit set successfully'},
 {'worker3.example.com': 'New rate limit set successfully'}]

С помощью аргумента destination можно указать список воркеров, которым будет отправлена команда:

>>> app.control.broadcast('rate_limit', {
...     'task_name': 'myapp.mytask',
...     'rate_limit': '200/m'}, reply=True,
...                             destination=['worker1@example.com'])
[{'worker1.example.com': 'New rate limit set successfully'}]

Разумеется, для задания ограничений скорости гораздо удобнее использовать интерфейс более высокого уровня, однако некоторые команды можно отправить только с помощью broadcast().

revoke: отзыв задач

поддержка пула:

все; terminate поддерживается только в prefork, eventlet и gevent

поддержка брокеров:

amqp, redis

команда:

celery -A proj control revoke <task_id>

Все узлы воркеров хранят сведения об отозванных идентификаторах задач — в памяти или на диске (см. раздел Постоянное хранение сведений об отзыве).

Получив запрос на отзыв, воркер пропустит выполнение задачи, но не завершит уже выполняющуюся задачу, если не задан параметр terminate.

Если задан параметр terminate, дочерний процесс воркера, выполняющий задачу, будет завершён. По умолчанию отправляется сигнал TERM, но его можно указать с помощью аргумента signal. В качестве сигнала можно указать имя в верхнем регистре любого сигнала, определённого в модуле signal стандартной библиотеки Python.

Завершение задачи также отзывает её.

Пример

>>> result.revoke()

>>> AsyncResult(id).revoke()

>>> app.control.revoke('d9078da5-9915-40a0-bfa1-392c7bde42ed')

>>> app.control.revoke('d9078da5-9915-40a0-bfa1-392c7bde42ed',
...                    terminate=True)

>>> app.control.revoke('d9078da5-9915-40a0-bfa1-392c7bde42ed',
...                    terminate=True, signal='SIGKILL')

Отзыв нескольких задач

Метод отзыва также принимает список, позволяя отозвать несколько задач одновременно.

Пример

>>> app.control.revoke([
...    '7993b0aa-1f0b-4780-9af0-c47c0858b3f2',
...    'f565793e-b041-4b2b-9ca4-dca22762a55d',
...    'd9d35e03-2997-42d0-a13e-64a66b88a618',
])

Начиная с версии 3.1, метод GroupResult.revoke использует эту возможность.

Постоянное хранение сведений об отзыве

Для отзыва задач всем воркерам отправляется широковещательное сообщение, после чего воркеры хранят список отозванных задач в памяти. При запуске воркер синхронизирует сведения об отозванных задачах с другими воркерами кластера.

Список отозванных задач хранится в памяти, поэтому при перезапуске всех воркеров список отозванных идентификаторов исчезнет. Чтобы сохранять этот список между перезапусками, укажите файл для его хранения с помощью аргумента –statedb команды celery worker:

$ celery-Aprojworker-lINFO--statedb=/var/run/celery/worker.state

Если вы используете celery multi, создайте отдельный файл для каждого экземпляра воркера и используйте формат %n для подстановки имени текущего узла:

celery multi start 2 -l INFO --statedb=/var/run/celery/%n.state

См. также раздел Переменные в путях к файлам

Для работы отзыва должны функционировать команды удалённого управления. На данный момент их поддерживают только RabbitMQ (amqp) и Redis.

revoke_by_stamped_headers: отзыв задач по помеченным заголовкам

поддержка пула:

все; terminate поддерживается только в prefork и eventlet

поддержка брокеров:

amqp, redis

команда:

celery -A proj control revoke_by_stamped_headers <header=value>

Эта команда похожа на revoke(), но вместо идентификаторов задач указываются помеченные заголовки в виде пар «ключ-значение». Будет отозвана каждая задача, у которой помеченный заголовок совпадает с указанной парой «ключ-значение».

Пример

>>> app.control.revoke_by_stamped_headers({'header': 'value'})

>>> app.control.revoke_by_stamped_headers({'header': 'value'}, terminate=True)

>>> app.control.revoke_by_stamped_headers({'header': 'value'}, terminate=True, signal='SIGKILL')

Отзыв нескольких задач по помеченным заголовкам

Метод revoke_by_stamped_headers также принимает список, позволяя выполнять отзыв по нескольким заголовкам или нескольким значениям.

Пример

>> app.control.revoke_by_stamped_headers({
...    'header_A': 'value_1',
...    'header_B': ['value_2', 'value_3'],
})

Будут отозваны все задачи с помеченным заголовком header_A со значением value_1, а также все задачи с помеченным заголовком header_B со значениями value_2 или value_3.

Пример для CLI

$ celery-Aprojcontrolrevoke_by_stamped_headersstamped_header_key_A=stamped_header_value_1stamped_header_key_B=stamped_header_value_2

$ celery-Aprojcontrolrevoke_by_stamped_headersstamped_header_key_A=stamped_header_value_1stamped_header_key_B=stamped_header_value_2--terminate

$ celery-Aprojcontrolrevoke_by_stamped_headersstamped_header_key_A=stamped_header_value_1stamped_header_key_B=stamped_header_value_2--terminate--signal=SIGKILL

Изменение ограничений времени во время выполнения

поддержка брокеров:

amqp, redis

Существует команда удалённого управления, позволяющая менять мягкие и жёсткие ограничения времени для задачи — она называется time_limit.

Пример изменения ограничения времени для задачи tasks.crawl_the_web: мягкое ограничение — одна минута, жёсткое — две минуты:

>>> app.control.time_limit('tasks.crawl_the_web',
                           soft=60, hard=120, reply=True)
[{'worker1.example.com': {'ok': 'time limits set successfully'}}]

Изменение повлияет только на задачи, выполнение которых начнётся после изменения ограничения времени.

Изменение ограничений частоты выполнения во время работы

Пример изменения ограничения частоты выполнения для задачи myapp.mytask: не более 200 задач этого типа в минуту:

>>> app.control.rate_limit('myapp.mytask', '200/m')

В примере выше не указано назначение, поэтому запрос на изменение затронет все экземпляры рабочих процессов в кластере. Если нужно затронуть только определённый список рабочих процессов, можно указать аргумент destination:

>>> app.control.rate_limit('myapp.mytask', '200/m',
...            destination=['celery@worker1.example.com'])

Очереди: добавление получателей

Команда управления add_consumer указывает одному или нескольким рабочим процессам начать получать сообщения из очереди. Эта операция идемпотентна.

Чтобы указать всем рабочим процессам кластера начать получать сообщения из очереди с именем «foo», используйте программу celery control:

$ celery-Aprojcontroladd_consumerfoo
-> worker1.local: OK
    started consuming from u'foo'

Чтобы указать конкретный рабочий процесс, используйте аргумент --destination:

$ celery-Aprojcontroladd_consumerfoo-dcelery@worker1.local

То же самое можно выполнить динамически с помощью метода app.control.add_consumer():

>>> app.control.add_consumer('foo', reply=True)
[{u'worker1.local': {u'ok': u"already consuming from u'foo'"}}]

>>> app.control.add_consumer('foo', reply=True,
...                          destination=['worker1@example.com'])
[{u'worker1.local': {u'ok': u"already consuming from u'foo'"}}]

До сих пор мы приводили примеры только с автоматическими очередями. Если вам нужен больший контроль, можно также указать обменник, ключ маршрутизации и другие параметры:

>>> app.control.add_consumer(
...     queue='baz',
...     exchange='ex',
...     exchange_type='topic',
...     routing_key='media.*',
...     options={
...         'queue_durable': False,
...         'exchange_durable': False,
...     },
...     reply=True,
...     destination=['w1@example.com', 'w2@example.com'])

Очереди: отмена получателей

Отменить получение сообщений из очереди можно по её имени с помощью команды управления cancel_consumer.

Чтобы указать всем рабочим процессам кластера прекратить получать сообщения из очереди, используйте программу celery control:

$ celery-Aprojcontrolcancel_consumerfoo

Аргумент --destination позволяет указать рабочий процесс или список рабочих процессов, к которым будет применена команда:

$ celery-Aprojcontrolcancel_consumerfoo-dcelery@worker1.local

Также можно программно отменить получение сообщений с помощью метода app.control.cancel_consumer():

>>> app.control.cancel_consumer('foo', reply=True)
[{u'worker1.local': {u'ok': u"no longer consuming from u'foo'"}}]

Очереди: список активных очередей

Получить список очередей, из которых рабочий процесс получает сообщения, можно с помощью команды управления active_queues:

$ celery-Aprojinspectactive_queues
[...]

Как и все остальные команды удалённого управления, эта команда поддерживает аргумент --destination, который позволяет указать рабочие процессы, от которых нужно получить ответ:

$ celery-Aprojinspectactive_queues-dcelery@worker1.local
[...]

То же самое можно сделать программно с помощью метода active_queues():

>>> app.control.inspect().active_queues()
[...]

>>> app.control.inspect(['worker1.local']).active_queues()
[...]

Список зарегистрированных задач

Получить список задач, зарегистрированных в рабочем процессе, можно с помощью registered():

>>> i.registered()
[{'worker1.example.com': ['tasks.add',
                          'tasks.sleeptask']}]

Список выполняющихся задач

Получить список активных задач можно с помощью active():

>>> i.active()
[{'worker1.example.com':
    [{'name': 'tasks.sleeptask',
      'id': '32666e9b-809c-41fa-8e93-5ae0c80afbbf',
      'args': '(8,)',
      'kwargs': '{}'}]}]

Список запланированных задач (ETA)

Получить список задач, ожидающих запланированного выполнения, можно с помощью scheduled():

>>> i.scheduled()
[{'worker1.example.com':
    [{'eta': '2010-06-07 09:07:52', 'priority': 0,
      'request': {
        'name': 'tasks.sleeptask',
        'id': '1a7980ea-8b19-413e-91d2-0b74f3844c4d',
        'args': '[1]',
        'kwargs': '{}'}},
     {'eta': '2010-06-07 09:07:53', 'priority': 0,
      'request': {
        'name': 'tasks.sleeptask',
        'id': '49661b9a-aa22-4120-94b7-9ee8031d219d',
        'args': '[2]',
        'kwargs': '{}'}}]}]

Список зарезервированных задач

Зарезервированные задачи — это задачи, которые уже получены, но ещё ожидают выполнения.

Получить их список можно с помощью reserved():

>>> i.reserved()
[{'worker1.example.com':
    [{'name': 'tasks.sleeptask',
      'id': '32666e9b-809c-41fa-8e93-5ae0c80afbbf',
      'args': '(8,)',
      'kwargs': '{}'}]}]

Статистика

Команда удалённого управления inspect stats (или stats()) выводит длинный список полезных (или не очень полезных) статистических данных о рабочем процессе:

$ celery-Aprojinspectstats

Подробное описание выходных данных см. в справочной документации по stats().

Удалённое завершение работы

Эта команда корректно завершит работу удалённого рабочего процесса:

>>> app.control.broadcast('shutdown') # shutdown all workers
>>> app.control.broadcast('shutdown', destination='worker1@example.com')

Проверка связи

Эта команда отправляет запрос проверки связи работающим процессам. В ответ они отправляют строку «pong» — и на этом всё. Если не указать собственное время ожидания, для ответа будет использоваться значение по умолчанию — одна секунда:

>>> app.control.ping(timeout=0.5)
[{'worker1.example.com': 'pong'},
 {'worker2.example.com': 'pong'},
 {'worker3.example.com': 'pong'}]

ping() также поддерживает аргумент destination, позволяющий указать рабочие процессы для проверки связи:

>>> ping(['worker2.example.com', 'worker3.example.com'])
[{'worker2.example.com': 'pong'},
 {'worker3.example.com': 'pong'}]

Включение и отключение событий

События можно включать и отключать с помощью команд enable_events и disable_events. Это удобно для временного мониторинга рабочего процесса с помощью celery events/celerymon.

>>> app.control.enable_events()
>>> app.control.disable_events()