Implementing Custom Sinks in Loguru for Production Observability

You need to route structured Loguru records to a backend that is not a file or a stream — an OTLP collector, a log aggregator, a message broker — without blocking request threads and without losing events when that backend is unreachable. This page is for backend engineers and SREs who already have Loguru configured and now need it to talk to their telemetry pipeline; it covers the exact callable contract, schema transformation, queue-backed dispatch, and dead-letter routing. It sits within the Loguru configuration and sinks reference, part of the Modern Python Logging Libraries Deep Dive guide.

Prerequisites

Pin Loguru and an async HTTP client for the OTLP example. The transformation example needs only the standard library.

pip install \
  "loguru>=0.7.0,<0.8.0" \
  "httpx>=0.27.0,<1.0.0"

Set the collector endpoint through the environment so the sink stays deployment-agnostic — the same endpoint convention used when exporting OTLP metrics to the collector.

export OTEL_EXPORTER_OTLP_LOGS_ENDPOINT="http://otel-collector:4318/v1/logs"

Implementation

A custom sink is the right tool whenever the destination is not a file or a stream: an HTTP endpoint, a message broker, a database, or any backend that needs a schema Loguru's built-in serialize=True does not produce. The contract is deliberately minimal — a callable taking one Message — but that minimalism puts every concern back on you: threading, back-pressure, schema, retries, and failure isolation. The four steps below address each in turn, and they compose: the queue worker in step two dispatches the schema from step three and falls back to the dead-letter path from step four.

Step 1 — Match the callable contract. Loguru invokes a sink synchronously with a single Message argument. Message is a str subclass carrying the fully formatted line, with the structured payload hanging off its .record attribute — so str(message) gives you the rendered text and message.record gives you the dict. Read the record through message.record and never mutate it in place; the same record object is shared with every other sink, so an in-place change corrupts their output. Extract context defensively with .get and explicit defaults, because extra only contains what a caller happened to bind.

import json
from loguru import logger


def sync_otel_sink(message) -> None:
    """Synchronous sink: safe extraction, flat JSON, isolated failures."""
    record = message.record
    extra = record["extra"]
    payload = {
        "timestamp": record["time"].isoformat(),
        "severity_number": record["level"].no,
        "severity_text": record["level"].name,
        "module": record["name"],
        "body": record["message"],
        "trace_id": extra.get("trace_id"),
        "span_id": extra.get("span_id"),
    }
    try:
        print(json.dumps(payload, default=str, separators=(",", ":")))
    except Exception as exc:  # never let the sink crash the caller
        import sys
        print(f"sink serialization failed: {exc}", file=sys.stderr)


logger.remove()
logger.add(sync_otel_sink, level="INFO", enqueue=True)
logger.bind(
    trace_id="0af7651916cd43dd8448eb211c80319c",
    span_id="b7ad6b7169203331",
).info("Service initialized")

Expected Output:

{"timestamp":"2024-05-12T10:15:30.123456+00:00","severity_number":20,"severity_text":"INFO","module":"__main__","body":"Service initialized","trace_id":"0af7651916cd43dd8448eb211c80319c","span_id":"b7ad6b7169203331"}

The trace_id and span_id values arrive through logger.bind, which is how trace identifiers get onto log records so the aggregator can join a log line back to its span.

Step 2 — Decouple network I/O with a queue-backed worker. A direct HTTP call inside the sink blocks the calling thread for the full round trip. Push the record onto a bounded asyncio.Queue instead and dispatch from a background worker, which enforces back-pressure and keeps logging off the request path. Note what enqueue=True does and does not buy you: it moves the call off your thread, but its internal queue is unbounded, so a stalled backend grows memory without limit. The bound has to live in your sink. See async and non-blocking logging with Loguru enqueue for the full semantics of that flag. Provide an explicit stop coroutine to drain pending records at shutdown; Loguru does not call it for you.

import asyncio
import json
from loguru import logger


class AsyncOTLPSink:
    """Enqueues records; a background worker dispatches them."""

    def __init__(self, maxsize: int = 10000) -> None:
        self.queue: asyncio.Queue = asyncio.Queue(maxsize=maxsize)
        self._task: asyncio.Task | None = None
        self.dropped = 0

    async def _worker(self) -> None:
        while True:
            record = await self.queue.get()
            try:
                payload = json.dumps({
                    "timestamp": record["time"].isoformat(),
                    "severity_text": record["level"].name,
                    "body": record["message"],
                    "traceparent": record["extra"].get("traceparent"),
                }, default=str)
                # await http_client.post(endpoint, content=payload)
                print(f"[DISPATCH] {payload}")
            except Exception as exc:
                import sys
                print(f"async dispatch failed: {exc}", file=sys.stderr)
            finally:
                self.queue.task_done()

    def __call__(self, message) -> None:
        """Called by Loguru; enqueues without blocking, drops when full."""
        try:
            self.queue.put_nowait(message.record)
        except asyncio.QueueFull:
            self.dropped += 1  # count drops; never block the caller
            import sys
            print("log queue full, dropping message", file=sys.stderr)

    def start(self) -> None:
        self._task = asyncio.create_task(self._worker())

    async def stop(self) -> None:
        await self.queue.join()
        if self._task:
            self._task.cancel()
            await asyncio.gather(self._task, return_exceptions=True)


async def main() -> None:
    sink = AsyncOTLPSink(maxsize=5000)
    logger.remove()
    logger.add(sink, level="DEBUG")
    sink.start()
    logger.info(
        "Async pipeline active",
        traceparent="00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01",
    )
    await asyncio.sleep(0.1)
    await sink.stop()


if __name__ == "__main__":
    asyncio.run(main())

Expected Output:

[DISPATCH] {"timestamp": "2024-05-12T10:15:30.123456+00:00", "severity_text": "INFO", "body": "Async pipeline active", "traceparent": "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01"}

Two constraints on this design. asyncio.Queue is not thread-safe, so if any records originate outside the loop's thread the enqueue must go through loop.call_soon_threadsafe, or you should use a queue.Queue with a thread worker instead — the same single-writer discipline behind the standard library's non-blocking QueueHandler pattern. And because this sink owns a task and a queue, do not combine it with enqueue=True: that would put a second queue and a second thread in front of yours, with a pickling boundary in between.

Step 3 — Transform to a vendor-agnostic schema. Aggregators want flat JSON. Map Loguru's severity onto the OpenTelemetry severity_number range, strip ANSI escape sequences that break strict parsers, and flatten nested context with dot notation for Elasticsearch or ClickHouse.

import re
from loguru import logger

ANSI = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])")

OTEL_SEVERITY = {  # Loguru level number -> OTel severity_number
    5: 1, 10: 5, 20: 9, 25: 9, 30: 13, 40: 17, 50: 21,
}


def flatten(d: dict, prefix: str = "", sep: str = ".") -> dict:
    out = {}
    for k, v in d.items():
        key = f"{prefix}{sep}{k}" if prefix else k
        out.update(flatten(v, key, sep) if isinstance(v, dict) else {key: v})
    return out


def to_otel_schema(message) -> dict:
    record = message.record
    extra = record["extra"]
    return {
        "resource": {
            "service.name": extra.get("service_name", "unknown"),
            "deployment.environment": extra.get("env", "production"),
        },
        "severity_number": OTEL_SEVERITY.get(record["level"].no, 9),
        "severity_text": record["level"].name,
        "body": ANSI.sub("", record["message"]),
        "trace_id": extra.get("trace_id"),
        "span_id": extra.get("span_id"),
        "attributes": flatten(
            {k: v for k, v in extra.items() if k not in ("trace_id", "span_id")}
        ),
    }

The severity mapping deserves a note, because it is the single most common source of wrong alerts downstream. Loguru's numeric levels (DEBUG=10, INFO=20, WARNING=30, ERROR=40, CRITICAL=50) do not line up with the OpenTelemetry severity_number range of 1–24, where INFO is 9 and ERROR is 17. Passing record["level"].no through unchanged makes an INFO record arrive as severity 20 — which the OTel scale reads as FATAL. The explicit lookup above avoids that, and it is the same translation problem covered in mapping Python log levels to syslog. If you register custom levels with logger.level(), add their numbers to the table or they fall through to the default.

Step 4 — Isolate failures with a dead-letter fallback. A sink must never crash the host process. Wrap dispatch in try/except, retry transient errors with exponential backoff and jitter, and on exhaustion route the record to a local dead-letter file for later replay. The dead-letter logger must be a separate Loguru instance writing to a plain file; routing failures back through the same network sink that just failed produces an infinite retry storm. Creating it lazily on first failure, as below, also avoids paying for the file handle in the common case where the backend is healthy — and that file should carry rotation and retention, per log rotation best practices, or a long outage will fill the disk.

import random
import time

_dlq = None  # lazily created to avoid recursion at import


def resilient_sink(message, max_retries: int = 3) -> None:
    global _dlq
    payload = to_otel_schema(message)
    for attempt in range(max_retries):
        try:
            raise ConnectionError("backend unreachable")  # simulate dispatch
        except ConnectionError:
            if attempt == max_retries - 1:
                if _dlq is None:
                    from loguru import logger as _l
                    _l.add("dlq.jsonl", mode="a", serialize=True)
                    _dlq = _l
                _dlq.bind(**payload).warning("dispatch exhausted, routed to DLQ")
                return
            time.sleep((2 ** attempt) + random.uniform(0, 1))

The blocking time.sleep here is only acceptable because this example runs on a worker, not on a request thread. In the queue-backed design of step two the equivalent is await asyncio.sleep(...) inside _worker, so retries stall one dispatch rather than the whole loop.

Configuration options

Two further logger.add arguments matter for callable sinks. catch=True (the default) makes Loguru print a traceback when your callable raises instead of letting the exception surface at the call site — useful as a backstop, but not a substitute for handling errors inside the sink, since a caught exception still means the record was lost. And filter= is how you keep a network sink cheap: give it a predicate so only the records that belong in the aggregator ever reach the queue, rather than filtering after the transform has already run.

For where callable sinks sit in the broader configuration surface, the Loguru configuration and sinks reference covers rotation, retention, and the level-routing topology a custom sink slots into.

Verification

Run the queue-backed example and confirm the dispatch line is emitted from the worker rather than the caller by observing that it appears after the await asyncio.sleep, not inline with the logger.info call. To assert the schema in CI without a live collector, build a stub message and check the OTel fields directly.

from types import SimpleNamespace
from datetime import datetime, timezone

msg = SimpleNamespace(record={
    "time": datetime(2024, 5, 12, tzinfo=timezone.utc),
    "level": SimpleNamespace(no=20, name="INFO"),
    "name": "__main__",
    "message": "ok",
    "extra": {"trace_id": "0af7", "span_id": "b7ad", "user": "u1"},
})
out = to_otel_schema(msg)
assert out["severity_number"] == 9        # OTel INFO, not Loguru's 20
assert out["severity_text"] == "INFO"
assert out["trace_id"] == "0af7"
assert out["attributes"] == {"user": "u1"}
print("schema assertions passed")

Expected Output:

schema assertions passed

Then verify the failure path, which is the part that only ever runs in an incident. Point the endpoint at a closed port, emit a record, and confirm three things: the process stays up, one line lands in dlq.jsonl with the full payload bound as structured fields, and the elapsed time matches the backoff schedule rather than hanging. Finally, load-test the drop policy by filling the queue faster than the worker drains it and checking that sink.dropped increases while request latency stays flat — a sink that silently blocks under load will show up here as a latency spike instead.

Common mistakes

• Error signature: request latency tracks the log volume and p99 climbs whenever the aggregator is slow, even though enqueue=True is set. Root cause: the callable still runs on Loguru's queue thread, so a synchronous requests.post inside it serializes every record behind one round trip and the internal queue simply grows. Remediation: dispatch from your own async worker or a concurrent.futures.ThreadPoolExecutor, keeping the callable itself to a single non-blocking enqueue.

• Error signature: a second sink starts emitting fields it never bound, or context leaks between unrelated requests. Root cause: the sink mutated message.record (or record["extra"]) in place, and that same object is handed to every other registered sink. Remediation: never write to the record; build a new payload dict, and copy with dict(record["extra"]) before transforming.

• Error signature: logs stop reaching the backend, nothing appears in the application log, and a bare traceback shows on stderr. Root cause: an exception escaped the callable; Loguru's catch printed it and dropped the record, so the failure is invisible to your own monitoring. Remediation: wrap the whole body in try/except Exception and route failures to a dead-letter file, then alert on the dead-letter file's growth rather than on the absence of logs.

• Error signature: every record arrives in the backend tagged FATAL and pages the on-call team. Root cause: record["level"].no was forwarded as severity_number, so Loguru's INFO=20 was read on the OpenTelemetry 1–24 scale, where 20 sits in the fatal band. Remediation: translate through an explicit lookup table, as in step three, and assert the mapping in a unit test so a new custom level cannot silently regress it.

• Error signature: RuntimeError: no running event loop, or records vanish, when logging from a thread pool or a forked child. Root cause: the sink holds an asyncio.Queue bound to one loop, and calls arriving from another thread or process never reach it. Remediation: enqueue via loop.call_soon_threadsafe, or use a thread-safe queue.Queue; across processes, re-register the sink in each child, as covered in thread-safe logging in multiprocessing.

Related

• Loguru configuration and sinks — the parent reference covering sink topology, rotation, retention and structured output.
• Async and non-blocking logging with Loguru enqueue — what the enqueue flag gives you before you write any queue plumbing of your own.
• Non-blocking logging with QueueHandler — the standard-library equivalent of this dispatch pattern, with a listener you own.
• Mapping Python log levels to syslog — the same severity-translation problem against a different target scale.
• Adding trace IDs to log records — how the trace and span identifiers this sink forwards get onto the record in the first place.

Frequently Asked Questions