Как я могу захватить все записи журнала Python, сгенерированные во время выполнения серии задач Celery?

Question

Я хочу преобразовать свою доморощенную систему очередей задач в очередь задач на основе Celery, но в настоящее время у меня есть одна особенность, которая вызывает у меня некоторое беспокойство.

Сейчас моя очередь задач работает очень грубо;Я запускаю задание (которое генерирует данные и загружает их на другой сервер), собираю протоколирование, используя вариант в библиотеке захвата журналов Nose, и затем сохраняю протоколирование для задачи как подробную запись результата в базе данных приложения.

Я хотел бы разбить это на три задачи:

1. собрать данные
2. загрузить данные
3. результаты отчета (включая все записи из предыдущих двух задач))

Настоящим кикером здесь является сбор журналов.Прямо сейчас, используя захват журнала, у меня есть серия записей журнала для каждого вызова журнала, сделанного во время процесса генерации и загрузки данных.Они необходимы для диагностических целей.Учитывая, что задачи даже не гарантированно выполняются в одном и том же процессе, неясно, как бы я выполнял это в очереди задач Celery.

Моим идеальным решением этой проблемы будет тривиальный и в идеале минимально инвазивный метод.захвата всех журналов во время задач предшественника (1, 2) и обеспечения их доступности для задачи репортера (3)

Лучше ли я оставаться довольно грубым с моим определением задачи и помещать всю эту работув одном задании?или есть способ передать существующее захваченное логирование, чтобы собрать его в конце?

Answer 1

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

в task.py:

import logging

@task
step1(*args, **kwargs):
    # `key` is some unique identifier common for a piece of data in all steps of processing
    logger = logging.getLogger("myapp.tasks.processing.%s"%key)
    # ...
    logger.info(...) # log something

@task
step2(*args, **kwargs):
    logger = logging.getLogger("myapp.tasks.processing.%s"%key)
    # ...
    logger.info(...) # log something

Здесь все записи были отправлены в одноименный регистратор. Теперь вы можете использовать 2 подхода для извлечения этих записей:

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

2. Создайте пользовательский прослушиватель, который будет накапливать записи в памяти, а затем возвращать их все, когда будет сказано Я бы использовал memcached для хранения здесь, это проще, чем создавать свое собственное межпроцессное хранилище.

Answer 2

Django Sentry - это утилита ведения журнала для Python (и Django), которая поддерживает Celery.

Answer 3

Звучит так, будто какой-то «наблюдатель» был бы идеальным. Если вы можете просматривать и использовать журналы как поток, вы можете отбрасывать результаты по мере их поступления. Так как наблюдатель будет работать отдельно и, следовательно, не будет зависеть от того, что он просматривает, я считаю, что это удовлетворит ваши требования для других пользователей. -инвазивное решение.