---
title: "Adding Structured Logging to Celery Tasks Without Touching Every Task"
slug: structured-logging-celery-tasks
date: 2025-08-08
updated: 2025-08-08
tags:
  - code
description: "How a single signal handler gave us argument-level observability across 50+ background jobs."
status: published
---

*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.

```python
@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.

```python
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:

```python
# 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

```python
@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:

```python
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:

```python
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:

```python
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:

```python
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

```python
# 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.