Adding Structured Logging to Celery Tasks Without Touching Every Task

How a single signal handler gave us argument-level observability across 50+ background jobs.

---

The Problem

We had 50+ Celery tasks across our Django application — sending emails, syncing data, generating reports, processing uploads. When something went wrong, the debugging flow looked like this:

1. Alert fires: “Task send_candidate_email failed”
2. Open Sentry → see the traceback
3. Traceback shows a KeyError or TypeError
4. But what arguments were passed? Who was this for? Which candidate? Which job?
5. Grep through application logs → nothing useful
6. Try to reproduce manually → waste 30 minutes guessing inputs

The issue: our tasks logged their existence (start/success/failure) but not their arguments. Without knowing what data was passed, debugging was guesswork.

---

The Naive Solution (Don’t Do This)

The obvious approach: add logging to every task.

@celery_task
def send_candidate_email(candidate_id, template_name, **kwargs):
    logger.info(f"send_candidate_email called with "
                f"candidate_id={candidate_id}, "
                f"template_name={template_name}, "
                f"kwargs={kwargs}")
    # ... actual task logic

This has several problems:

1. You have to modify every task. 50+ tasks means 50+ changes.
2. New tasks forget to add it. No enforcement mechanism.
3. Inconsistent formatting. Each developer logs differently.
4. Verbose and noisy. The logging line is often longer than the task logic.

---

The Better Solution: Celery Signals

Celery provides signals that fire at various points in a task’s lifecycle. The one we want is task_prerun — fires after the task is received by a worker but before execution begins.

from celery.signals import task_prerun

@task_prerun.connect
def on_task_prerun(sender, task_id, args, kwargs, **extra):
    logger.info(
        'Task %s[%s] | args=%r | kwargs=%r',
        sender.name, task_id, args, kwargs
    )

That’s it. Four lines. Every task in your application now logs its arguments before execution.

---

Where to Put It

This goes in your Celery app configuration — the file where you define your Celery instance:

# project/celery.py
import logging
from celery import Celery
from celery.signals import task_prerun

logger = logging.getLogger('celery.tasks')

app = Celery('myproject')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()


@task_prerun.connect
def on_task_prerun(sender, task_id, args, kwargs, **extra):
    logger.info(
        'Task %s[%s] | args=%r | kwargs=%r',
        sender.name, task_id, args, kwargs
    )

Because this is registered at the Celery app level via signals, it applies to every task that runs on this worker — including tasks in third-party packages.

---

What the Output Looks Like

[INFO] celery.tasks: Task send_candidate_email[a3f2b1c4-...] | args=(42851, 'interview_confirmation') | kwargs={'cc': ['hr@company.com']}
[INFO] celery.tasks: Task sync_calendar_events[7e8d9f01-...] | args=() | kwargs={'recruiter_id': 1523, 'force': True}
[INFO] celery.tasks: Task generate_report[b4c5d6e7-...] | args=(891,) | kwargs={'format': 'xlsx', 'date_range': '2024-01-01:2024-12-31'}

Now when a task fails, the arguments are right there in the logs — same timestamp, same request. No guessing.

---

Signal Parameters Explained

@task_prerun.connect
def on_task_prerun(sender, task_id, args, kwargs, **extra):
    # sender: The task class (has .name attribute)
    # task_id: UUID for this specific execution
    # args: Positional arguments tuple
    # kwargs: Keyword arguments dict
    # extra: Additional signal metadata (usually empty)

Other Useful Signals

Signal	Fires When	Use Case
task_prerun	Before task executes	Log arguments
task_postrun	After task completes	Log result/duration
task_success	On successful completion	Metrics/counters
task_failure	On exception	Alert with context
task_received	When worker receives task	Queue monitoring

You can compose these for richer observability:

from celery.signals import task_prerun, task_postrun, task_failure
import time

_task_start_times = {}

@task_prerun.connect
def on_task_prerun(sender, task_id, args, kwargs, **extra):
    _task_start_times[task_id] = time.time()
    logger.info(
        'Task %s[%s] started | args=%r | kwargs=%r',
        sender.name, task_id, args, kwargs
    )

@task_postrun.connect
def on_task_postrun(sender, task_id, retval, state, **extra):
    start = _task_start_times.pop(task_id, None)
    duration = f"{time.time() - start:.2f}s" if start else "unknown"
    logger.info(
        'Task %s[%s] completed | state=%s | duration=%s',
        sender.name, task_id, state, duration
    )

@task_failure.connect
def on_task_failure(sender, task_id, exception, args, kwargs, **extra):
    logger.error(
        'Task %s[%s] FAILED | exception=%s | args=%r | kwargs=%r',
        sender.name, task_id, repr(exception), args, kwargs
    )

---

Handling Sensitive Arguments

Not all arguments should be logged. You might pass passwords, tokens, or PII:

SENSITIVE_TASKS = {
    'auth.reset_password',
    'payments.process_charge',
}

REDACT_KWARGS = {'password', 'token', 'secret', 'credit_card'}

@task_prerun.connect
def on_task_prerun(sender, task_id, args, kwargs, **extra):
    if sender.name in SENSITIVE_TASKS:
        logger.info('Task %s[%s] started (args redacted)', sender.name, task_id)
        return

    safe_kwargs = {
        k: '***' if k in REDACT_KWARGS else v
        for k, v in kwargs.items()
    }

    logger.info(
        'Task %s[%s] | args=%r | kwargs=%r',
        sender.name, task_id, args, safe_kwargs
    )

---

Handling Large Arguments

Some tasks receive large payloads (file contents, bulk data). You don’t want megabytes in your logs:

import sys

MAX_ARG_LOG_SIZE = 1024  # bytes

def truncate_repr(obj, max_size=MAX_ARG_LOG_SIZE):
    r = repr(obj)
    if len(r) > max_size:
        return r[:max_size] + f'... (truncated, {sys.getsizeof(obj)} bytes)'
    return r

@task_prerun.connect
def on_task_prerun(sender, task_id, args, kwargs, **extra):
    logger.info(
        'Task %s[%s] | args=%s | kwargs=%s',
        sender.name, task_id,
        truncate_repr(args),
        truncate_repr(kwargs)
    )

---

Integration with APM Tools

If you use Datadog, New Relic, or similar APM tools, you can enrich the trace span:

from ddtrace import tracer

@task_prerun.connect
def on_task_prerun(sender, task_id, args, kwargs, **extra):
    span = tracer.current_span()
    if span:
        span.set_tag('celery.task_name', sender.name)
        span.set_tag('celery.task_id', task_id)
        # Add searchable tags for key arguments
        if args and isinstance(args[0], int):
            span.set_tag('celery.primary_id', args[0])

Now you can search in Datadog: “Show me all executions of send_email where primary_id = 42851.”

---

Testing the Signal

# tests/test_celery_signals.py
from unittest.mock import patch
from myapp.tasks import send_candidate_email

@patch('project.celery.logger')
def test_task_prerun_logs_arguments(mock_logger):
    send_candidate_email.apply(args=(123, 'welcome'), kwargs={'cc': ['a@b.com']})

    mock_logger.info.assert_called()
    call_args = mock_logger.info.call_args[0]
    assert 'send_candidate_email' in call_args[0] or call_args[1]
    assert '123' in str(call_args)

---

Results

After deploying this single signal handler:

• Mean time to debug task failures: Reduced from ~30 minutes to ~5 minutes
• “Cannot reproduce” bugs: Dropped significantly — we now have the exact inputs
• Zero code changes to existing tasks: The signal applies globally
• Automatic coverage of new tasks: Any new task immediately gets logging

---

Takeaways

1. Celery signals are the right abstraction for cross-cutting concerns. Don’t modify individual tasks when you can hook into the lifecycle.

2. %r (repr) is better than %s (str) for logging arguments. It shows types, distinguishes None from "None", and handles nested structures.

3. Log at the INFO level, not DEBUG. You want this in production. DEBUG often gets filtered out.

4. Handle sensitive data. Add a redaction layer from day one. Logging passwords to your aggregation service is a security incident.

5. This pattern applies beyond Celery. Django signals, middleware, decorators — any framework that provides lifecycle hooks can be instrumented this way without modifying business logic.

The total investment: 4 lines of code in one file. The total return: permanent argument-level observability for every background job in the system.