Python Observability

Writing Custom structlog Processors

A structlog processor is the smallest unit of logging policy you can write: three arguments, one return value, and complete freedom over the record in between. This page covers the contract in detail, the two processor shapes that appear in real services — enrichers and droppers — and the concurrency rules for the stateful ones. It builds on the structlog processors and pipelines guide, part of the modern Python logging libraries deep dive section.

The contract, and its three legal outcomes A processor's signature and the three things it is allowed to do. It receives three positional arguments: the underlying logger object, the method name such as info or warning, and the event dictionary. The ordinary outcome is to return an event dictionary, usually the same one it was given, possibly with keys added or changed. The second legal outcome is to raise structlog.DropEvent, which discards the record silently and skips the rest of the chain. The third applies only to the final processor: a renderer returns a string, or a tuple of positional and keyword arguments, which is passed to the underlying logger. Returning None, or returning something that is neither a dict nor a renderer result, produces a failure inside the next processor rather than a clear error at the point of the mistake. def processor(logger, method_name, event_dict): what you receive logger — the wrapped object method_name — "info", "error" event_dict — everything else 1 · return the dict the ordinary case add keys, rewrite values, or leave it untouched 2 · raise structlog.DropEvent discarded, chain stops, no error 3 · return a string the renderer only — and it must be last the fourth outcome, which is a bug returning None — because a forgotten return statement is the easiest mistake in a five-line function the next processor receives None and raises AttributeError from inside the logging call, pointing at the wrong function a unit test that calls the processor directly catches it before it reaches a chain
Three legal outcomes, one common bug. A missing return surfaces as an error inside the next processor, which is why these are worth unit-testing in isolation.

Prerequisites

pip install "structlog>=24.1.0,<26.0.0" \
            "pytest>=8.0.0,<9.0.0"

Implementation

Step 1 — Write an enricher. The most common processor adds fields. Use setdefault so a value bound at the call site wins over the default.

import os

SERVICE = os.environ.get("SERVICE_NAME", "unknown")
VERSION = os.environ.get("SERVICE_VERSION", "0")

def add_service_metadata(logger, method_name: str, event_dict: dict) -> dict:
    event_dict.setdefault("service", SERVICE)
    event_dict.setdefault("version", VERSION)
    return event_dict

Step 2 — Write a rewriter. Redaction is the canonical case: walk the dict, replace what matches, return it. Keep the key set explicit rather than pattern-matching every value, for the reasons set out in redacting sensitive data in log records.

SECRET_KEYS = frozenset({"password", "token", "api_key", "authorization", "session"})

def redact(logger, method_name: str, event_dict: dict) -> dict:
    for key in list(event_dict):
        if key.lower() in SECRET_KEYS:
            event_dict[key] = "[redacted]"
    return event_dict

Place this early — before anything that might copy values into a message string, and certainly before the renderer. A redaction processor at the end of the chain protects the rendered output and nothing else.

Step 3 — Write a dropper. DropEvent is the only way to discard. Place the processor as early as it can still make a correct decision, because everything after it is work avoided.

import structlog

NOISY_ROUTES = {"/healthz", "/readyz", "/metrics"}

def drop_probe_traffic(logger, method_name: str, event_dict: dict) -> dict:
    if event_dict.get("route") in NOISY_ROUTES and event_dict.get("status", 200) < 400:
        raise structlog.DropEvent          # a failing probe still gets through
    return event_dict

The status < 400 guard is the part worth copying. A health check that starts returning 503 is exactly the record you want, and a filter written as "drop everything from /healthz" removes it.

Step 4 — Write a stateful processor safely. A sampler or deduplicator holds counters that every worker thread touches. Use a class, guard the state with a lock, and keep the critical section to the counter update.

import threading
import time
import structlog

class BurstSampler:
    """First N of an event pass in full, then one per interval, with a suppressed count."""

    def __init__(self, burst: int = 5, interval: float = 1.0, floor: str = "warning"):
        self.burst, self.interval = burst, interval
        self.floor = floor
        self._state: dict[str, list] = {}
        self._lock = threading.Lock()

    def __call__(self, logger, method_name: str, event_dict: dict) -> dict:
        if method_name in {"warning", "error", "critical", "exception"}:
            return event_dict                       # never sample the rare ones
        key = str(event_dict.get("event", ""))
        now = time.monotonic()
        with self._lock:                            # counters only — nothing expensive in here
            seen, last, dropped = self._state.get(key, (0, 0.0, 0))
            seen += 1
            if seen <= self.burst:
                self._state[key] = (seen, now, 0)
                return event_dict
            if now - last >= self.interval:
                self._state[key] = (seen, now, 0)
                event_dict["suppressed"] = dropped
                return event_dict
            self._state[key] = (seen, last, dropped + 1)
            raise structlog.DropEvent

Keying on event_dict["event"] rather than on the rendered line is what makes this work: the event name is the stable event class, while the rendered message contains the IDs that make every occurrence unique. The same reasoning, in standard-library terms, is in rate limiting and sampling noisy loggers.

Where each kind of processor belongs A chain drawn as a horizontal band with four zones. The first zone holds enrichers: context merging, level and logger name, timestamps, and service metadata. They must run first because everything downstream reads what they write. The second zone holds rewriters such as redaction, which need the full set of fields present but must run before anything renders or copies values. The third zone holds droppers — samplers, probe filters, deduplicators — placed as late as necessary to decide correctly and as early as possible so the work after them is skipped; the diagram marks that tension explicitly. The fourth zone is the renderer, alone, at the end. Underneath, an arrow labelled cost shows that a record dropped in zone three still paid for zones one and two, which is why a dropper that can decide on fewer fields should move left. placement follows from what a processor needs to read 1 · enrichers merge_contextvars add_log_level TimeStamper · metadata everything downstream reads these 2 · rewriters redaction value normalisation needs every field present, must precede the renderer 3 · droppers samplers · probe filters deduplicators as late as correctness needs, as early as cost allows 4 · renderer alone at the end always cost already paid when a dropper fires a record dropped in zone 3 still paid for zones 1 and 2 — several function calls per record, on the calling thread so a sampler that only needs the event name can move left of the enrichers and skip all of it and level filtering does not belong in the chain at all — make_filtering_bound_logger discards before the chain starts
The dropper zone is a negotiation: every field it needs pushes it right, and every field it does not need lets it move left, where the saving is bigger.

Step 5 — Register it in the right position.

import structlog

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.stdlib.add_log_level,
        structlog.processors.TimeStamper(fmt="iso", utc=True),
        add_service_metadata,
        redact,                                   # after enrichment, before the renderer
        drop_probe_traffic,                       # after 'route' exists
        BurstSampler(burst=5, interval=1.0),
        structlog.processors.JSONRenderer(),      # last, always
    ],
    cache_logger_on_first_use=True,
)
How much of a stateful processor needs the lock A stateful processor executing on four worker threads. Each thread arrives with its own event dict and does its own unlocked work first: computing the event key from the dict, and reading the clock. Then a short locked region covers only the shared state — reading the counter for that key, comparing it against the burst allowance and the elapsed interval, and writing the updated tuple back. After the lock is released, each thread continues independently, either returning the dict or raising DropEvent. The diagram marks the anti-pattern explicitly: performing the key computation, the clock read, or any formatting inside the locked region turns a few nanoseconds of contention into a serialisation point that every logging thread queues behind, which costs more than the records the sampler was added to save. four threads through one sampler thread 1 thread 2 thread 3 thread 4 key = event_dict["event"] · time.monotonic() unlocked — every thread at once the lock read counter compare write back tens of ns return the dict, or raise DropEvent unlocked again the anti-pattern computing the key, reading the clock or formatting anything inside the lock — every logging thread then queues on work that did not need to be shared
The locked region holds three integer operations. Everything expensive happens on either side of it, which is what keeps a shared counter from becoming a serialisation point.

Configuration options

Concern Choice Recommended
Shape function vs class class when it has state or configuration
Mutation in place vs copy in place
Drop DropEvent vs None DropEvent, always
Level filtering processor vs wrapper make_filtering_bound_logger
Shared state lock vs none a lock around counters only
Position — after what it reads, before the renderer
I/O — never inside a processor

Verification

Test processors as plain functions. They take three arguments and return a dict, which makes them the easiest part of a logging setup to cover.

import pytest
import structlog

def test_enricher_does_not_override_the_caller():
    out = add_service_metadata(None, "info", {"event": "x", "service": "explicit"})
    assert out["service"] == "explicit"          # setdefault, not assignment

def test_redaction_masks_known_keys():
    out = redact(None, "info", {"event": "login", "token": "sk-live-9f3c"})
    assert out["token"] == "[redacted]"

def test_probe_filter_keeps_failures():
    with pytest.raises(structlog.DropEvent):
        drop_probe_traffic(None, "info", {"route": "/healthz", "status": 200})
    kept = drop_probe_traffic(None, "info", {"route": "/healthz", "status": 503})
    assert kept["status"] == 503                 # a failing probe is the point

def test_sampler_keeps_the_onset():
    sampler = BurstSampler(burst=3, interval=999)
    passed = 0
    for _ in range(50):
        try:
            sampler(None, "info", {"event": "retrying"})
            passed += 1
        except structlog.DropEvent:
            pass
    assert passed == 3                           # the first three, then nothing

Expected Output:

test_processors.py::test_enricher_does_not_override_the_caller PASSED
test_processors.py::test_redaction_masks_known_keys PASSED
test_processors.py::test_probe_filter_keeps_failures PASSED
test_processors.py::test_sampler_keeps_the_onset PASSED

Then verify the position in the assembled chain, so a later edit cannot move the renderer:

def test_renderer_is_last():
    chain = structlog.get_config()["processors"]
    assert isinstance(chain[-1], structlog.processors.JSONRenderer)

Common mistakes

AttributeError: 'NoneType' object has no attribute 'get'

Error signature: the traceback points at a structlog-provided processor, not at yours. Root cause: your processor fell off the end of a branch without returning the dict. Remediation: return event_dict on every path, and unit-test the processor directly — the error surfaces one step downstream from the actual mistake.

The filter never fires

Error signature: volume is unchanged and the dropped records still appear. Root cause: the processor reads a key that a later processor adds. Remediation: move it after the enricher that provides the field, and assert the ordering in a test.

The sampler drops errors

Error signature: an incident timeline is missing the failures that triggered it. Root cause: the sampler ran for every method name. Remediation: return immediately for warning, error, critical and exception.

Composing processors well

A chain of ten small processors is easier to reason about than three large ones, and the discipline that keeps it that way is worth stating: each processor should have one reason to change.

One concern per processor. A processor that adds service metadata and redacts secrets and drops probe traffic is three policies in one function, and changing any of them means re-reading all three. Splitting them costs three function calls per record — a few hundred nanoseconds — and buys independent testing, independent ordering, and the ability to disable one without touching the others.

Configuration in the constructor, not the environment. A processor that reads os.environ inside __call__ re-reads it per record and cannot be tested with different settings in the same process. Take the settings as constructor arguments and build the instance where the chain is assembled, which is also where a reviewer expects to see the configuration.

class SampleByLevel:
    def __init__(self, rates: dict[str, float]):
        self.rates = rates                      # decided once, at configuration time

    def __call__(self, logger, method_name: str, event_dict: dict) -> dict:
        rate = self.rates.get(method_name, 1.0)
        if rate < 1.0 and _deterministic_hash(event_dict) % 100 >= rate * 100:
            raise structlog.DropEvent
        return event_dict

Deterministic decisions where possible. A sampler that uses a random number produces a different result for the same event on every replica, which makes the aggregate rate correct and any single investigation frustrating. Hashing a stable field instead means the same event is kept or dropped consistently across the fleet, so a trace that survives on one service survives on all of them — the same reasoning that makes trace-ID-based sampling the default for distributed tracing.

Processors that need state elsewhere

Occasionally a processor needs something that is expensive to compute: a feature-flag value, a tenant's configuration, a mapping loaded from a database. None of that belongs in the chain, because the chain runs on the calling thread inside the log call.

The pattern that works is a cache populated elsewhere and read cheaply here. A background task refreshes the mapping on a timer; the processor does a dictionary lookup with a default. If the lookup misses, the processor proceeds without the field rather than blocking to fetch it — a missing field is a small loss, and a network call inside a log statement is a large one.

class TenantEnricher:
    def __init__(self, cache: dict[str, str]):
        self.cache = cache                       # refreshed by a background task

    def __call__(self, logger, method_name: str, event_dict: dict) -> dict:
        tenant = event_dict.get("tenant_id")
        if tenant:
            name = self.cache.get(tenant)        # a dict lookup, never a fetch
            if name:
                event_dict["tenant_name"] = name
        return event_dict
Need Do Do not
Configuration constructor arguments read the environment per record
Expensive lookups a cache refreshed elsewhere fetch inside __call__
Sampling hash a stable field use a random number
Shared counters a lock around the counter only a lock around the whole body
Failure inside the processor let it raise in tests, guard in production swallow it silently either way

What happens when a processor raises

An exception in a processor propagates out of the logging call and into the code that was logging, which is almost never what you want in production and exactly what you want in tests. The safe shape is a guard that is explicit about which exceptions it tolerates, placed only around the parts that can genuinely fail — a serialisation of an unknown object, a regex over user-supplied text — rather than around the whole function, which turns a bug into silence.

def enrich(logger, method_name: str, event_dict: dict) -> dict:
    try:
        event_dict["payload_size"] = len(json.dumps(event_dict.get("payload", {})))
    except (TypeError, ValueError):
        event_dict["payload_size"] = -1          # explicit, and visible in a query
    return event_dict

The -1 matters more than it looks: a field that silently disappears when serialisation fails is indistinguishable from one that was never set, while a sentinel value shows up in a query and tells you the enrichment is failing.

Frequently Asked Questions

Can a processor be a class instead of a function?

Yes — anything callable with the three-argument signature works, and a class is the natural choice when the processor has configuration or state. structlog's own TimeStamper and JSONRenderer are classes. Give it a __call__ method with the same signature and construct it in the processors list.

Should a processor mutate the event dict or return a new one?

Mutate it. The dict is created per log call and belongs to the chain, so copying it at every step multiplies allocations for no benefit. The exception is a processor that conditionally rewrites a value bound by the caller: returning a modified copy there makes it obvious the original binding is unchanged for anyone else holding that logger.

How do I make a processor skip certain loggers?

Read the logger name from the event dict, which stdlib.add_logger_name puts there — but only if that processor runs first. The logger argument your processor receives is the underlying logger object, which under some factories has no useful name, so the event dict is the reliable source.

Is it safe to do I/O in a processor?

No. Processors run on the calling thread inside the log call, so a network lookup or a file read there blocks whatever was logging, and under asyncio it blocks the event loop. Anything expensive belongs behind a cache populated elsewhere, or in a handler on the far side of a queue.

Related Guides