Design a Logging Framework

Ishaan S Published 25 Oct, 2026

Design a logging library that other teams will import into their services. Application code asks for a named logger, such as payments.db, and calls info, error and friends, optionally attaching key-value context like an order id.

Operators configure it without touching that code: they set a minimum level per logger, and a setting on payments should apply to everything beneath it unless overridden. Output can go to several places at once, for example readable lines on the console and JSON lines in a file, each with its own level.

The library runs inside busy multi-threaded servers, so lines from different threads must never interleave, and a slow disk must not slow down request handling. In 45 minutes, design the classes, explain how a log call flows from logger to output, and show how you keep it thread-safe and cheap when a level is disabled.

Use the LLD board for your class diagram and the LLD code tab to write and run your classes.

Design Path · Low-Level Design · problem 63 of 70

Ask before you design. Try answering each yourself, then compare.

Which levels do we support?
Typical answer
DEBUG, INFO, WARNING and ERROR, ordered so a threshold comparison is a single integer check.
How are loggers related?
Typical answer
By dotted name. A logger without its own level inherits the nearest ancestor's level, ending at the root.
Should a record logged on a child also reach sinks attached to its parents?
Typical answer
Yes by default (propagation), with a per-logger switch to turn it off.
Which outputs are needed now?
Typical answer
Console and file. Network collectors should be possible later without changing loggers.
Can we lose log lines under extreme load?
Typical answer
For the async path, dropping and counting lines is acceptable rather than blocking requests. Synchronous sinks never drop.
Is configuration from a file required?
Typical answer
Programmatic configuration is enough; a config loader can come later on top of the same API.

Separate three questions: should this message be logged, what should it look like, and where should it go.

If a logger has no level of its own, where does it get one? Think of the dotted name as a tree.

A disabled DEBUG call may run millions of times a second. What is the very first thing log() should do?

Where is the smallest place you need a lock so lines never interleave?

How could you make any existing sink asynchronous without rewriting it?

Requirements

Functional

Get or create loggers by dotted name, forming a parent-child hierarchy.
Log at four levels with optional key-value context.
Configure levels per logger (inherited when unset) and per sink.
Send records to multiple sinks, each with its own formatter.
Flush and close everything cleanly on shutdown.

Non-functional

A call below the effective level costs well under 1 microsecond and allocates nothing.
Thread-safe: lines from concurrent threads are never interleaved or corrupted.
The async sink adds under 10 microseconds to the caller and never blocks it, even when the disk stalls.

Out of scope

Log shipping, indexing and search.
File rotation and compression (discuss as an extension).
Configuration file formats.

Class design

Who decides what. A Logger decides whether a message is logged (level check) and where it travels (its own sinks, then its ancestors' while propagate is on). A Formatter turns a LogRecord into text. A Sink filters by its own level and writes the line. LogManager owns the registry and creates intermediate loggers, so app.db always has app as its parent.

Log call flow. log() first compares the level with the effective level, found by walking up to the nearest logger with a level set; disabled calls return before a record is built. Otherwise it creates an immutable record, captures the thread name and timestamp, and hands it to every sink along the chain.

Thread safety. Records are immutable, formatting happens outside locks, and each sink holds a lock only around the write. The registry has its own lock. AsyncSink wraps any sink: callers do a non-blocking put on a bounded queue and one worker thread drains it, so a slow disk never stalls a request. When the queue is full it drops and counts. Shutdown sends a sentinel so everything queued is written first. Where threads cannot start at all, such as Python in the browser, the sink notices at construction and writes synchronously instead.

Extensibility. New outputs subclass Sink; new layouts implement Formatter; neither touches Logger.

Open the LLD board and press Reference to see the class diagram.

Patterns used

Chain of Responsibility: a record climbs the logger hierarchy and each level's sinks get a chance to handle it.
Strategy: formatters are interchangeable, so the same sink can write text or JSON.
Decorator: AsyncSink wraps any Sink and adds buffering on a background thread without changing it.
Registry (a controlled Singleton): LogManager hands out one Logger per name, so configuration applies everywhere it is used.

Reference implementation

import json
import os
import queue
import sys
import tempfile
import threading
import time
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from enum import IntEnum
from typing import Dict, List, Optional


class Level(IntEnum):
    DEBUG, INFO, WARNING, ERROR = 10, 20, 30, 40


@dataclass(frozen=True)
class LogRecord:
    logger: str
    level: Level
    message: str
    context: Dict[str, object] = field(default_factory=dict)
    created: float = field(default_factory=time.time)
    thread: str = field(default_factory=lambda: threading.current_thread().name)


class Formatter(ABC):
    @abstractmethod
    def format(self, record: LogRecord) -> str: ...


class TextFormatter(Formatter):
    def format(self, r):
        ctx = "".join(f" {k}={v}" for k, v in r.context.items())
        stamp = time.strftime("%H:%M:%S", time.localtime(r.created))
        return f"{stamp} {r.level.name:<7} [{r.logger}] {r.message}{ctx}"


class JsonFormatter(Formatter):
    def format(self, r):
        return json.dumps({"ts": round(r.created, 3), "level": r.level.name, "logger": r.logger,
                           "thread": r.thread, "msg": r.message, **r.context})


class Sink(ABC):
    """A destination. Filters by its own level and serialises writes with a lock."""
    def __init__(self, formatter: Formatter, level: Level = Level.DEBUG):
        self.formatter, self.level = formatter, level
        self._lock = threading.Lock()

    def handle(self, record: LogRecord):
        if record.level >= self.level:
            line = self.formatter.format(record)          # format outside the lock
            with self._lock:
                self.write(line)

    @abstractmethod
    def write(self, line: str): ...

    def close(self):
        pass


class ConsoleSink(Sink):
    def write(self, line):
        sys.stdout.write(line + "\n")


class FileSink(Sink):
    def __init__(self, path: str, formatter, level=Level.DEBUG):
        super().__init__(formatter, level)
        self._file = open(path, "a", encoding="utf-8")

    def write(self, line):
        self._file.write(line + "\n")

    def close(self):
        with self._lock:
            self._file.close()


class AsyncSink(Sink):
    """Decorator: callers only enqueue; one background thread writes to the wrapped sink."""
    _STOP = object()

    def __init__(self, inner: Sink, capacity: int = 10_000):
        super().__init__(inner.formatter, inner.level)
        self.inner, self.dropped = inner, 0
        self._queue: "queue.Queue" = queue.Queue(maxsize=capacity)
        self._worker: Optional[threading.Thread] = threading.Thread(target=self._drain, name="log-writer", daemon=True)
        try:
            self._worker.start()
        except RuntimeError:                              # no threads (e.g. browser Python): write directly
            self._worker = None

    def handle(self, record):
        if self._worker is None:
            return self.inner.handle(record)
        try:
            self._queue.put_nowait(record)                # never block the application thread
        except queue.Full:
            self.dropped += 1

    def write(self, line):
        self.inner.write(line)

    def _drain(self):
        while (record := self._queue.get()) is not self._STOP:
            self.inner.handle(record)

    def close(self):
        if self._worker:
            self._queue.put(self._STOP)                   # everything queued before this is written
            self._worker.join()
        self.inner.close()


class Logger:
    def __init__(self, name: str, parent: Optional["Logger"] = None, level: Optional[Level] = None):
        self.name, self.parent = name, parent
        self.level = level                                # None means inherit from parent
        self.sinks: List[Sink] = []
        self.propagate = True

    def effective_level(self) -> Level:
        node = self
        while node.level is None:
            node = node.parent
        return node.level

    def log(self, level: Level, message: str, **context):
        if level < self.effective_level():                # cheap early exit before building a record
            return
        record, node = LogRecord(self.name, level, message, context), self
        while node:
            for sink in node.sinks:
                sink.handle(record)
            node = node.parent if node.propagate else None

    def debug(self, msg, **ctx): self.log(Level.DEBUG, msg, **ctx)
    def info(self, msg, **ctx): self.log(Level.INFO, msg, **ctx)
    def warning(self, msg, **ctx): self.log(Level.WARNING, msg, **ctx)
    def error(self, msg, **ctx): self.log(Level.ERROR, msg, **ctx)


class LogManager:
    """Registry of loggers keyed by dotted name; 'app.db' is a child of 'app', which is a child of root."""
    def __init__(self, root_level: Level = Level.INFO):
        self.root = Logger("root", level=root_level)
        self._loggers: Dict[str, Logger] = {}
        self._lock = threading.Lock()

    def get_logger(self, name: str) -> Logger:
        with self._lock:
            parent, parts = self.root, name.split(".")
            for i in range(1, len(parts) + 1):
                full = ".".join(parts[:i])
                if full not in self._loggers:
                    self._loggers[full] = Logger(full, parent)
                parent = self._loggers[full]
            return parent

    def shutdown(self):
        for lg in [self.root, *self._loggers.values()]:
            for sink in lg.sinks:
                sink.close()


if __name__ == "__main__":
    manager = LogManager(root_level=Level.INFO)
    manager.root.sinks.append(ConsoleSink(TextFormatter(), level=Level.INFO))

    path = os.path.join(tempfile.mkdtemp(), "db.log")
    db = manager.get_logger("app.db")
    db.level = Level.DEBUG
    db.sinks.append(AsyncSink(FileSink(path, JsonFormatter())))
    http = manager.get_logger("app.http")
    manager.get_logger("app").level = Level.WARNING

    http.info("GET /health 200")                          # dropped: 'app' is at WARNING
    http.error("upstream timeout", route="/checkout", ms=3012)
    db.debug("pool warmed", size=8)                       # file only: console sink is INFO+

    def worker(n):
        for i in range(25):
            db.debug("query ok", worker=n, seq=i)
    if sys.platform == "emscripten":                     # browser Python cannot start threads
        print("(browser: no threads here, so the 4 workers run in turn and AsyncSink writes directly)")
        [worker(n) for n in range(4)]
    else:
        threads = [threading.Thread(target=worker, args=(n,), name=f"req-{n}") for n in range(4)]
        [t.start() for t in threads]
        [t.join() for t in threads]
    db.info("migration finished", version=42)             # file and console (propagates to root)

    manager.shutdown()
    lines = open(path).read().splitlines()
    print(f"\n{path.split(os.sep)[-1]}: {len(lines)} lines, all valid JSON:",
          all(json.loads(l)["logger"] == "app.db" for l in lines))

Trade-offs

Async output keeps request latency low but can lose the last lines on a crash; synchronous output is durable but slower.
Dropping when the queue is full protects the application but loses data; blocking keeps every line but couples latency to the disk.
Walking up the hierarchy for every call is simple but repeats work; caching effective levels is faster but must be invalidated on reconfiguration.
Structured JSON is easy for machines to query but harder to read on a terminal than plain text.

What a strong answer shows

Separates filtering, formatting and output into different classes.
Implements level inheritance through the logger hierarchy and explains propagation.
Makes disabled log calls nearly free.
Places locks narrowly (per sink, not global) and keeps formatting outside them.
Adds asynchronous output without changing existing sinks, and explains the full-queue policy and shutdown flush.

Mistakes to avoid

One giant Logger class that formats, filters and writes files itself.
Building the message string before checking the level.
A global lock around every log call, which serialises the whole application.
An unbounded async queue that grows until the process runs out of memory when the disk stalls.
Forgetting to drain the queue on shutdown, losing the last and often most important lines.

Follow-ups to expect

How would you add size-based file rotation? (A RotatingFileSink that checks size inside the locked write and renames files.)
How do you attach a request id to every line without passing it around? (A context variable read when the record is built.)
How would you rate-limit a noisy error? (A filter on the sink or logger that keys on message and counts per window.)
Should ERROR lines ever be dropped by the async sink? (Offer a policy: block or write synchronously for ERROR only.)

Practice next

{"nodes":[{"id":"mgr","type":"class","label":"LogManager","attrs":"- root: Logger\n- _loggers: Dict[str, Logger]\n- _lock: Lock","methods":"+ get_logger(name): Logger\n+ shutdown()","x":0,"y":0},{"id":"logger","type":"class","label":"Logger","attrs":"- name: str\n- parent: Logger\n- level: Level | None\n- sinks: List[Sink]\n- propagate: bool","methods":"+ effective_level(): Level\n+ log(level, msg, **ctx)\n+ debug / info / warning / error","x":320,"y":0},{"id":"record","type":"class","label":"LogRecord","attrs":"- logger: str\n- level: Level\n- message: str\n- context: Dict\n- created: float\n- thread: str","methods":"","x":662,"y":0},{"id":"sink","type":"abstract","label":"Sink","attrs":"- formatter: Formatter\n- level: Level\n- _lock: Lock","methods":"+ handle(record)\n# write(line)\n+ close()","x":0,"y":254},{"id":"fmt","type":"interface","label":"Formatter","attrs":"","methods":"+ format(record): str","x":269,"y":254},{"id":"console","type":"class","label":"ConsoleSink","attrs":"","methods":"+ write(line)","x":0,"y":486},{"id":"file","type":"class","label":"FileSink","attrs":"- _file: TextIO","methods":"+ write(line)\n+ close()","x":250,"y":486},{"id":"async","type":"class","label":"AsyncSink","attrs":"- inner: Sink\n- _queue: Queue\n- _worker: Thread | None\n- dropped: int","methods":"+ handle(record)\n+ close()","x":500,"y":486},{"id":"text","type":"class","label":"TextFormatter","attrs":"","methods":"+ format(record): str","x":0,"y":704},{"id":"json","type":"class","label":"JsonFormatter","attrs":"","methods":"+ format(record): str","x":261,"y":704}],"edges":[{"id":"e1","from":"mgr","to":"logger","kind":"comp","label":"registry"},{"id":"e3","from":"logger","to":"record","kind":"dep","label":"creates"},{"id":"e4","from":"logger","to":"sink","kind":"aggr","label":"0..* sinks"},{"id":"e5","from":"sink","to":"fmt","kind":"assoc","label":"formats with"},{"id":"e6","from":"console","to":"sink","kind":"inherit","label":""},{"id":"e7","from":"file","to":"sink","kind":"inherit","label":""},{"id":"e8","from":"async","to":"sink","kind":"inherit","label":""},{"id":"e9","from":"async","to":"file","kind":"assoc","label":"wraps any sink"},{"id":"e10","from":"text","to":"fmt","kind":"impl","label":""},{"id":"e11","from":"json","to":"fmt","kind":"impl","label":""}]}

# Starter: fill in each TODO. Loggers decide WHETHER to log and where records travel;
# formatters decide HOW a record looks; sinks decide WHERE it goes. Keep those apart.
import json
import os
import queue
import sys
import tempfile
import threading
import time
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from enum import IntEnum
from typing import Dict, List, Optional

class Level(IntEnum):
    DEBUG, INFO, WARNING, ERROR = (10, 20, 30, 40)

@dataclass(frozen=True)
class LogRecord:
    logger: str
    level: Level
    message: str
    context: Dict[str, object] = field(default_factory=dict)
    created: float = field(default_factory=time.time)
    thread: str = field(default_factory=lambda: threading.current_thread().name)

class Formatter(ABC):
    @abstractmethod
    def format(self, record: LogRecord) -> str:
        ...

class TextFormatter(Formatter):
    def format(self, r):
        pass  # TODO

class JsonFormatter(Formatter):
    def format(self, r):
        pass  # TODO

class Sink(ABC):
    """A destination. Filters by its own level and serialises writes with a lock."""

def __init__(self, formatter: Formatter, level: Level=Level.DEBUG):
        self.formatter, self.level = (formatter, level)
        self._lock = threading.Lock()

def handle(self, record: LogRecord):
        pass  # TODO

@abstractmethod
    def write(self, line: str):
        ...

def close(self):
        pass  # TODO

class ConsoleSink(Sink):
    def write(self, line):
        pass  # TODO

class FileSink(Sink):
    def __init__(self, path: str, formatter, level=Level.DEBUG):
        super().__init__(formatter, level)
        self._file = open(path, 'a', encoding='utf-8')

def write(self, line):
        pass  # TODO

def close(self):
        pass  # TODO

class AsyncSink(Sink):
    """Decorator: callers only enqueue; one background thread writes to the wrapped sink."""
    _STOP = object()

def __init__(self, inner: Sink, capacity: int=10000):
        super().__init__(inner.formatter, inner.level)
        self.inner, self.dropped = (inner, 0)
        self._queue: 'queue.Queue' = queue.Queue(maxsize=capacity)
        self._worker: Optional[threading.Thread] = threading.Thread(target=self._drain, name='log-writer', daemon=True)
        try:
            self._worker.start()
        except RuntimeError:
            self._worker = None

def handle(self, record):
        pass  # TODO

def write(self, line):
        pass  # TODO

def _drain(self):
        pass  # TODO

def close(self):
        pass  # TODO

class Logger:
    def __init__(self, name: str, parent: Optional['Logger']=None, level: Optional[Level]=None):
        self.name, self.parent = (name, parent)
        self.level = level
        self.sinks: List[Sink] = []
        self.propagate = True

def effective_level(self) -> Level:
        pass  # TODO

def log(self, level: Level, message: str, **context):
        pass  # TODO

def debug(self, msg, **ctx):
        self.log(Level.DEBUG, msg, **ctx)

def info(self, msg, **ctx):
        self.log(Level.INFO, msg, **ctx)

def warning(self, msg, **ctx):
        self.log(Level.WARNING, msg, **ctx)

def error(self, msg, **ctx):
        self.log(Level.ERROR, msg, **ctx)

class LogManager:
    """Registry of loggers keyed by dotted name; 'app.db' is a child of 'app', which is a child of root."""

def __init__(self, root_level: Level=Level.INFO):
        self.root = Logger('root', level=root_level)
        self._loggers: Dict[str, Logger] = {}
        self._lock = threading.Lock()

def get_logger(self, name: str) -> Logger:
        pass  # TODO

def shutdown(self):
        pass  # TODO

Topics #063 #t-sd-lld Concurrency Design Patterns Medium Object-Oriented Design System Design Practice