Celery Logging: Structured Logging for Tasks and Workers

from celery import Celery, signals
import logging
import sys

app = Celery('logging_demo', broker='redis://localhost:6379/0')

@signals.setup_logging.connect
def setup_celery_logging(loglevel=None, **kwargs):
    """Configure Celery logging with custom format."""
    formatter = logging.Formatter(
        fmt='%(asctime)s [%(levelname)s] %(name)s: %(message)s',
        datefmt='%Y-%m-%d %H:%M:%S'
    )
    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(formatter)

    root_logger = logging.getLogger()
    root_logger.addHandler(handler)
    root_logger.setLevel(loglevel or logging.INFO)

    celery_logger = logging.getLogger('celery')
    celery_logger.setLevel(logging.DEBUG)

@signals.task_prerun.connect
def log_task_start(sender=None, task_id=None, task=None, **kwargs):
    logger = logging.getLogger('celery.task')
    logger.info(f"Starting task {task.name} [{task_id}]")

@signals.task_success.connect
def log_task_success(sender=None, result=None, **kwargs):
    logger = logging.getLogger('celery.task')
    logger.info(f"Task completed with result: {result}")

@signals.task_failure.connect
def log_task_failure(sender=None, task_id=None, exception=None, **kwargs):
    logger = logging.getLogger('celery.task')
    logger.error(f"Task {task_id} failed: {exception}", exc_info=True)

@app.task
def sample_logging_task(item_id, count=1):
    import time
    logger = logging.getLogger('celery.task')
    logger.debug(f"Processing item {item_id} with count {count}")
    time.sleep(0.1)
    logger.info(f"Item {item_id} processed successfully")
    return {"item_id": item_id, "status": "done"}

import time
print("Worker logging example:")
result = sample_logging_task.delay(42, count=3)
time.sleep(0.3)

from celery import Celery, signals
import logging
import json
import sys
import time

app = Celery('logging_demo', broker='redis://localhost:6379/0')

class StructuredJSONFormatter(logging.Formatter):
    def format(self, record):
        log_entry = {
            "timestamp": self.formatTime(record, "%Y-%m-%dT%H:%M:%S"),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "module": record.module,
            "line": record.lineno,
        }

        if hasattr(record, 'task_id'):
            log_entry["task_id"] = record.task_id
        if hasattr(record, 'task_name'):
            log_entry["task_name"] = record.task_name
        if record.exc_info and record.exc_info[0]:
            log_entry["exception"] = self.formatException(record.exc_info)

        return json.dumps(log_entry)

@signals.setup_logging.connect
def setup_json_logging(**kwargs):
    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(StructuredJSONFormatter())
    logging.getLogger('celery').addHandler(handler)
    logging.getLogger('celery').setLevel(logging.DEBUG)

class TaskAdapter(logging.LoggerAdapter):
    def process(self, msg, kwargs):
        kwargs.setdefault('extra', {})
        if 'extra' in kwargs:
            kwargs['extra'].update({
                'task_id': self.extra.get('task_id', ''),
                'task_name': self.extra.get('task_name', ''),
            })
        return msg, kwargs

@signals.task_prerun.connect
def setup_task_logging(sender=None, task_id=None, task=None, **kwargs):
    logger = logging.getLogger('celery.task')
    adapter = TaskAdapter(logger, {'task_id': task_id, 'task_name': task.name})
    task.request.logger = adapter
    adapter.info(f"Task started")

@app.task
def json_logging_task(data):
    logger = getattr(getattr(json_logging_task, 'request', None), 'logger', None)
    if not logger:
        logger = logging.getLogger('celery.task')
    logger.info(f"Processing data: {data}")
    time.sleep(0.1)
    logger.info("Processing complete")
    return {"result": "ok"}

import time
print("Structured JSON logging:")
json_logging_task.delay({"key": "value"})
time.sleep(0.3)

from celery import Celery, signals
import logging
import sys
import json

app = Celery('logging_demo', broker='redis://localhost:6379/0')

class AggregationFormatter(logging.Formatter):
    def format(self, record):
        import socket
        base = {
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "message": record.getMessage(),
            "worker": socket.gethostname(),
            "process": record.process,
        }
        if hasattr(record, 'task_id'):
            base["task_id"] = record.task_id
        if hasattr(record, 'task_name'):
            base["task_name"] = record.task_name
        return json.dumps(base)

@signals.setup_logging.connect
def setup_aggregation_logging(**kwargs):
    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(AggregationFormatter())
    logging.getLogger('celery').addHandler(handler)

    import socket
    hostname = socket.gethostname()
    print(f"[INIT] Worker starting on {hostname}", file=sys.stderr)

@app.task
def aggregation_task(task_num):
    import logging
    import time

    logger = logging.getLogger('celery.task')
    logger.info(f"Processing task number {task_num}")
    time.sleep(0.05)
    logger.info(f"Task {task_num} complete")

    return task_num

import time
print("Log aggregation demo:")
for i in range(3):
    aggregation_task.delay(i)

time.sleep(0.5)

import socket
print(f"\nSearch all logs for: worker={socket.gethostname()}")
print("(In production, use Elasticsearch or Loki for cross-worker search)")