Руководство по мониторингу и управлению

• shell: перейти в оболочку Python.

  В локальной области видимости будет доступна переменная celery: это текущее приложение. Кроме того, все известные задачи будут автоматически добавлены в локальную область видимости (если не указан флаг --without-tasks).

  Если они установлены, последовательно используются https://pypi.org/project/Ipython/, https://pypi.org/project/bpython/ или обычный python. Вы можете принудительно выбрать реализацию с помощью --ipython, --bpython или --python.

• status: вывести список активных узлов кластера

  $ celery-Aprojstatus
  

• result: показать результат задачи

  $ celery-Aprojresult-ttasks.add4e196aa4-0141-4601-8138-7aa33db0f577
  

  Обратите внимание: имя задачи можно не указывать, если для задачи не используется пользовательский бэкенд результатов.

• purge: удалить сообщения из всех настроенных очередей задач.

  Эта команда удалит все сообщения из очередей, настроенных в параметре CELERY_QUEUES:

  Предупреждение

  Операцию нельзя отменить, сообщения будут удалены без возможности восстановления!

  $ celery-Aprojpurge
  

  Очереди для очистки также можно указать с помощью параметра -Q:

  $ celery-Aprojpurge-Qcelery,foo,bar
  

  а очереди, которые не нужно очищать, — с помощью параметра -X:

  $ celery-Aprojpurge-Xcelery
  

• inspect active: вывести список активных задач

  $ celery-Aprojinspectactive
  

  Это все задачи, выполнение которых происходит в данный момент.

• inspect scheduled: вывести список запланированных задач с ETA

  $ celery-Aprojinspectscheduled
  

  Это задачи, зарезервированные исполнителем, для которых задан аргумент eta или countdown.

• inspect reserved: вывести список зарезервированных задач

  $ celery-Aprojinspectreserved
  

  Команда выводит все задачи, предварительно полученные исполнителем и ожидающие выполнения (задачи с заданным значением ETA не включаются).

• inspect revoked: вывести историю отозванных задач

  $ celery-Aprojinspectrevoked
  

• inspect registered: вывести список зарегистрированных задач

  $ celery-Aprojinspectregistered
  

• inspect stats: показать статистику исполнителя (см. Статистика)

  $ celery-Aprojinspectstats
  

• inspect query_task: показать сведения о задаче или задачах по идентификатору.

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

  $ celery-Aprojinspectquery_taske9f6c8f0-fec9-4ae8-a8c6-cf8c8451d4f8
  

  Также можно запросить сведения сразу о нескольких задачах:

  $ celery-Aprojinspectquery_taskid1id2...idN
  

• control enable_events: включить события

  $ celery-Aprojcontrolenable_events
  

• control disable_events: отключить события

  $ celery-Aprojcontroldisable_events
  

• migrate: перенести задачи с одного брокера на другой (ЭКСПЕРИМЕНТАЛЬНО).

  $ celery-Aprojmigrateredis://localhostamqp://localhost
  

  Эта команда перенесет все задачи с одного брокера на другой. Поскольку команда новая и экспериментальная, перед ее выполнением обязательно создайте резервную копию данных.

По умолчанию команды inspect и control выполняются для всех исполнителей. С помощью аргумента --destination можно указать одного или нескольких исполнителей:

$ celery-Aprojinspect-dw1@e.com,w2@e.comreserved

$ celery-Aprojcontrol-dw1@e.com,w2@e.comenable_events

• Мониторинг в реальном времени с использованием событий Celery

  • Ход выполнения задач и история

  • Возможность просматривать подробные сведения о задачах (аргументы, время начала, время выполнения и другое)

  • Графики и статистика

• Удаленное управление

  • Просмотр состояния и статистики исполнителей

  • Остановка и перезапуск экземпляров исполнителей

  • Управление размером пула исполнителя и настройками автомасштабирования

  • Просмотр и изменение очередей, из которых получает задачи экземпляр исполнителя

  • Просмотр выполняющихся в данный момент задач

  • Просмотр запланированных задач (ETA/countdown)

  • Просмотр зарезервированных и отозванных задач

  • Настройка ограничений по времени и частоте выполнения

  • Просмотр конфигурации

  • Отзыв или принудительное завершение задач

• HTTP API

  • Вывод списка исполнителей

  • Остановка исполнителя

  • Перезапуск пула исполнителя

  • Увеличение пула исполнителя

  • Уменьшение пула исполнителя

  • Автомасштабирование пула исполнителя

  • Начало получения задач из очереди

  • Прекращение получения задач из очереди

  • Вывод списка задач

  • Вывод списка типов задач (с которыми уже работали)

  • Получение сведений о задаче

  • Запуск задачи

  • Запуск задачи по имени

  • Получение результата задачи

  • Изменение мягкого и жесткого ограничений по времени выполнения задачи

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

  • Отзыв задачи

• Аутентификация OpenID

Снимки экрана

Другие снимки экрана:

Для установки Flower можно использовать pip:

$ pipinstallflower

Команда flower запускает веб-сервер, который можно открыть в браузере:

$ celery-Aprojflower

По умолчанию используется порт http://localhost:5555, но его можно изменить с помощью аргумента –port:

$ celery-Aprojflower--port=5555

URL брокера также можно передать с помощью аргумента --broker:

$ celery--broker=amqp://guest:guest@localhost:5672//flower
or
$ celery--broker=redis://guest:guest@localhost:6379/0flower

Затем откройте Flower в веб-браузере:

$ openhttp://localhost:5555

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

Камеры полезны, если нужно перехватывать события и выполнять с ними какие-либо действия с заданным интервалом. Для обработки событий в реальном времени следует напрямую использовать app.events.Receiver, как показано в разделе Обработка в реальном времени.

Ниже приведен пример камеры, которая выводит снимок состояния на экран:

from pprint import pformat

from celery.events.snapshot import Polaroid

class DumpCam(Polaroid):
    clear_after = True  # clear after flush (incl, state.event_count).

    def on_shutter(self, state):
        if not state.event_count:
            # No new events since last snapshot.
            return
        print('Workers: {0}'.format(pformat(state.workers, indent=4)))
        print('Tasks: {0}'.format(pformat(state.tasks, indent=4)))
        print('Total: {0.event_count} events, {0.task_count} tasks'.format(
            state))

Дополнительные сведения об объектах состояния см. в справочнике API для celery.events.state.

Теперь эту камеру можно использовать с celery events, указав ее с помощью параметра -c:

$ celery-Aprojevents-cmyapp.DumpCam--frequency=2.0

Или можно использовать ее программно следующим образом:

from celery import Celery
from myapp import DumpCam

def main(app, freq=1.0):
    state = app.events.State()
    with app.connection() as connection:
        recv = app.events.Receiver(connection, handlers={'*': state.event})
        with DumpCam(state, freq=freq):
            recv.capture(limit=None, timeout=None)

if __name__ == '__main__':
    app = Celery(broker='amqp://guest@localhost//')
    main(app)

сигнатура:

task-sent(uuid, name, args, kwargs, retries, eta, expires, queue, exchange, routing_key, root_id, parent_id)

Отправляется при публикации сообщения задачи, если включен параметр task_send_sent_event.

сигнатура:

task-received(uuid, name, args, kwargs, retries, eta, hostname, timestamp, root_id, parent_id)

Отправляется при получении задачи исполнителем.

сигнатура:

task-started(uuid, hostname, timestamp, pid)

Отправляется непосредственно перед выполнением задачи исполнителем.

сигнатура:

task-succeeded(uuid, result, runtime, hostname, timestamp)

Отправляется, если задача выполнена успешно.

Время выполнения — это время, затраченное на выполнение задачи с помощью пула (от отправки задачи в пул исполнителей до вызова обработчика результата пула).

сигнатура:

task-failed(uuid, exception, traceback, hostname, timestamp)

Отправляется, если выполнить задачу не удалось.

сигнатура:

task-rejected(uuid, requeue)

Исполнитель отклонил задачу, возможно, чтобы она была повторно поставлена в очередь или перемещена в очередь недоставленных сообщений.

сигнатура:

task-revoked(uuid, terminated, signum, expired)

Отправляется, если задача отозвана (обратите внимание: сообщение, вероятно, будет отправлено несколькими исполнителями).

• terminated принимает значение true, если процесс задачи был завершен принудительно,

  а в поле signum указывается использованный сигнал.

• expired принимает значение true, если срок действия задачи истек.

сигнатура:

task-retried(uuid, exception, traceback, hostname, timestamp)

Отправляется, если задача завершилась с ошибкой, но в дальнейшем будет выполнена повторно.

сигнатура:

worker-online(hostname, timestamp, freq, sw_ident, sw_ver, sw_sys)

Исполнитель подключился к брокеру и находится в сети.

• hostname: имя узла исполнителя.

• timestamp: временная метка события.

• freq: частота контрольных сигналов в секундах (число с плавающей точкой).

• sw_ident: название программного обеспечения исполнителя (например, py-celery).

• sw_ver: версия программного обеспечения (например, 2.2.0).

• sw_sys: операционная система (например, Linux/Darwin).

сигнатура:

worker-heartbeat(hostname, timestamp, freq, sw_ident, sw_ver, sw_sys, active, processed)

Отправляется каждую минуту. Если исполнитель не отправлял контрольный сигнал в течение 2 минут, он считается отключенным.

• hostname: имя узла исполнителя.

• timestamp: временная метка события.

• freq: частота контрольных сигналов в секундах (число с плавающей точкой).

• sw_ident: название программного обеспечения исполнителя (например, py-celery).

• sw_ver: версия программного обеспечения (например, 2.2.0).

• sw_sys: операционная система (например, Linux/Darwin).

• active: количество выполняемых в данный момент задач.

• processed: общее количество задач, обработанных этим исполнителем.

сигнатура:

worker-offline(hostname, timestamp, freq, sw_ident, sw_ver, sw_sys)

Исполнитель отключился от брокера.