Помощь с лесозаготовками сельдерея

Question

Итак, Celery - супер отличная библиотека, но ее документация для секции журналирования не самая лучшая, что приводит меня сюда с просьбой о помощи.

Мой сценарий на данный момент таков (хорошо в итоге):

import logging
from celery import Celery
from celery.utils.log import get_logger
from task import process
import config

logger = get_logger(__name__)
timber_handler = timber.TimberHandler(api_key=config.key,
                                  level=logging.INFO)
logger.addHandler(timber_handler)



app = Celery('task',
             broker=config.url,
             backend='rpc://')

@app.task
def run_task():
    status = get_status() # get alive or dead status
    if status == 1:
        logger.info("Task is running")
        process()


@app.on_after_configure.connect
def task_periodic(**kwargs):
    app.add_periodic_task(2.0, run_task.s(), name="Run Constantly")
    # More tasks

Функция процесса в файле tasks.py является очень простой функцией, позволяющей получить доступ к API и БД для получения некоторой информации, и я хочу иметь возможность записать это в логгер (timber.io), который подключается к библиотеке журналов python и является онлайн-хранилище для журналов.

Однако моя главная проблема заключается в том, что бревна отправляются на стандартный вывод, а не на бревна. Я посмотрел на celery.signals, но документация не очень хорошая. Любая помощь здесь будет принята с благодарностью. Спасибо.

Answer 1

Можете ли вы попробовать это?

import logging
import os
import sys
from celery import Celery
from celery.signals import after_setup_logger

app = Celery('app')
app.conf.update({
    'broker_url': 'filesystem://',
    'broker_transport_options': {
        'data_folder_in': './broker/out',
        'data_folder_out': './broker/out',
        'data_folder_processed': './broker/processed'
    },
    'result_persistent': False,
    'task_serializer': 'json',
    'result_serializer': 'json',
    'accept_content': ['json']})


logger = logging.getLogger(__name__)


for f in ['./broker/out', './broker/processed']:
    if not os.path.exists(f):
        os.makedirs(f)

@after_setup_logger.connect
def setup_loggers(logger, *args, **kwargs):
    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')

    # add filehandler
    fh = logging.FileHandler('logs.log')
    fh.setLevel(logging.DEBUG)
    fh.setFormatter(formatter)
    logger.addHandler(fh)


@app.task()
def add(x, y):
    logger.info('Found addition')
    logger.info('Added {0} and {1} to result, '.format(x,y))
    return x+y


if __name__ == '__main__':
    task = add.s(x=2, y=3).delay()

Запустите работника так:

celery worker --app=app.app --concurrency=1 --loglevel=INFO

И запустить задачу асинхронно:

python app.py

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

Это записывает журналы в logs.log (замените обработчик файла на обработчик timber.io, и это должно решить вашу проблему).

Мне было немного трудно, так как я не мог заставить его работать с worker_hijack_root_logger=False и пользовательским логгером, определенным в setup_logging.

Однако, после пересмотра документации, я пришел к выводу, что лучше не переписывать регистратор, а просто расширять его:

Если вы хотите дополнить настройку конфигурации регистрации Celery тогда вы можете использовать after_setup_logger и after_setup_task_logger сигналы. * * тысяча двадцать-две

Смотри также: http://docs.celeryproject.org/en/latest/userguide/signals.html#after-setup-logger