Design a logging framework

01The design, in the six-step framework

1. Requirements and use cases

A logging framework is a library, so its users are other programmers. It must be cheap when a message is filtered out and must never slow down or crash the code that calls it.

• A developer gets a named logger and calls it at a level: debug, info, warn, error.
• An operator sets a minimum level per logger, and child names inherit from parents, so app.db follows app unless overridden.
• An operator attaches sinks (console, file, in-memory) with their own level and format.
• The library formats a record as plain text or JSON, chosen per sink.
• The application flushes and shuts down without losing queued records.

Out of scope: remote shipping, log rotation policy details and structured query. Non-functional constraints: thread-safe from any number of callers; a slow sink must not block callers, so writes go through a bounded queue; a new sink or format is a new class. Reported twist: a two-part Google walkthrough spends its second half entirely on concurrency and locking, so expect that follow-up.

2. Core entities and relationships

• Level: an ordered enum, so comparison is just an integer compare.
• LogRecord: an immutable event: logger name, level, message, timestamp, thread name.
• Logger: the front door; filters by effective level and passes records to its handlers, then its parent.
• Formatter: turns a record into a string; text and JSON implementations.
• Sink: writes a line somewhere; console, file and memory implementations.
• Handler: pairs a level with a formatter and a sink.
• AsyncDispatcher: a bounded queue and one writer thread between callers and handlers.
• LogManager: the registry that returns one logger per name and builds the hierarchy.

A handler composes its formatter and sink, which exist only to serve it. A logger aggregates handlers because several loggers may share one file handler. The manager composes loggers: it alone creates them, which is what makes two calls with the same name return one object.

3. Interfaces and key methods

class Formatter(Protocol):
    def format(self, record: LogRecord) -> str: ...

class Sink(Protocol):
    def write(self, line: str) -> None: ...
    def flush(self) -> None: ...

class Logger:
    def log(self, level: Level, message: str) -> None:
        """Return immediately if below the effective level; never raises to the caller."""
    def effective_level(self) -> Level:
        """Own level, or the nearest ancestor's, or INFO at the root."""
    def add_handler(self, handler: Handler) -> None: ...

class AsyncDispatcher:
    def submit(self, handlers: list[Handler], record: LogRecord) -> None:
        """Non-blocking; drops the record and counts it when the queue is full."""
    def stop(self) -> None:
        """Drain the queue, flush every sink, join the worker."""

class LogManager:
    def get_logger(self, name: str) -> Logger: ...
    def shutdown(self) -> None: ...

Invariants: records from one thread reach a sink in the order that thread logged them; a sink is only touched by the single writer thread, so sinks need no locks of their own; a failing sink never propagates an exception to the application.

4. Design patterns and the choices behind them

The logger hierarchy as a chain

A record passes through the logger's handlers and then up to its parent's. This is a chain of responsibility where every link may act rather than one link claiming the request, so attaching one handler to the root captures everything. A flat list of loggers would force every new module to be configured by hand.

Formatter and sink vary independently

JSON to a file, text to the console, JSON to memory: formats and destinations multiply. Making each a separate strategy inside Handler is the bridge idea: two hierarchies joined by composition rather than a JsonFileHandler class for every pair. A third-party client is wrapped as a Sink with an adapter.

Decorators for sink behaviour

Buffering is not a new kind of destination. BufferedSink wraps any sink and is itself a sink, the decorator pattern, so retry or sampling wrappers can stack without touching FileSink.

Queue between callers and I/O

The write path is producer-consumer: callers enqueue, one worker writes. Sinks stay lock-free because a single thread owns them, and a slow disk costs queue depth rather than request latency. A lock around each write is simpler and correct, but it makes every request thread wait for the disk.

A manager, not a singleton

A global logger is tempting, and LogManager may well have one default instance, but a hidden singleton makes tests share state. Keeping it an ordinary class with an optional module-level default preserves both convenience and single responsibility: it creates and wires, and does not format or write.

5. Core flow and code

1. log(WARN, msg) compares against effective_level() and returns at once if filtered.
2. It builds a LogRecord, walks itself and its ancestors collecting handlers, and calls submit.
3. submit uses put_nowait; a full queue increments dropped.
4. The worker pops each item and calls handler.emit, which checks the handler's level, formats and writes.
5. stop sends a sentinel, the worker drains, flushes every sink it has seen and exits.

from __future__ import annotations
import json, queue, sys, threading, time
from dataclasses import dataclass, asdict
from enum import IntEnum
from typing import Optional, Protocol


class Level(IntEnum):
    DEBUG = 10; INFO = 20; WARN = 30; ERROR = 40


@dataclass(frozen=True)
class LogRecord:
    name: str
    level: Level
    message: str
    at: float
    thread: str


class Formatter(Protocol):
    def format(self, record: LogRecord) -> str: ...


class TextFormatter:
    def format(self, r: LogRecord) -> str:
        return f"{r.at:.3f} {r.level.name} {r.name} [{r.thread}] {r.message}"


class JsonFormatter:
    def format(self, r: LogRecord) -> str:
        return json.dumps({**asdict(r), "level": r.level.name})


class Sink(Protocol):
    def write(self, line: str) -> None: ...
    def flush(self) -> None: ...


class ConsoleSink:
    def write(self, line: str) -> None: sys.stdout.write(line + "\n")
    def flush(self) -> None: sys.stdout.flush()


class MemorySink:
    def __init__(self) -> None: self.lines: list[str] = []
    def write(self, line: str) -> None: self.lines.append(line)
    def flush(self) -> None: pass


class FileSink:
    def __init__(self, path: str) -> None: self.f = open(path, "a", encoding="utf-8")
    def write(self, line: str) -> None: self.f.write(line + "\n")
    def flush(self) -> None: self.f.flush()


class BufferedSink:
    """Decorator: holds lines and writes them to the inner sink in batches."""
    def __init__(self, inner: Sink, size: int = 100) -> None:
        self.inner, self.size, self.buf = inner, size, []
    def write(self, line: str) -> None:
        self.buf.append(line)
        if len(self.buf) >= self.size:
            self.flush()
    def flush(self) -> None:
        for line in self.buf:
            self.inner.write(line)
        self.buf.clear()
        self.inner.flush()


@dataclass
class Handler:
    level: Level
    formatter: Formatter
    sink: Sink

    def emit(self, record: LogRecord) -> None:
        if record.level >= self.level:
            self.sink.write(self.formatter.format(record))


class AsyncDispatcher:
    def __init__(self, capacity: int = 10_000) -> None:
        self.queue: queue.Queue = queue.Queue(capacity)
        self.dropped = 0
        self.errors = 0
        self.seen: dict[int, Handler] = {}
        self.worker = threading.Thread(target=self._run, daemon=True)
        self.worker.start()

    def submit(self, handlers: list[Handler], record: LogRecord) -> None:
        try:
            self.queue.put_nowait((handlers, record))
        except queue.Full:
            self.dropped += 1

    def _run(self) -> None:
        while True:
            item = self.queue.get()
            if item is None:
                break
            handlers, record = item
            for h in handlers:
                self.seen[id(h)] = h
                try:
                    h.emit(record)
                except Exception:
                    self.errors += 1            # a bad sink must not kill the worker
        for h in self.seen.values():
            h.sink.flush()

    def stop(self) -> None:
        self.queue.put(None)
        self.worker.join()


class Logger:
    def __init__(self, name: str, parent: Optional[Logger], dispatcher: AsyncDispatcher) -> None:
        self.name, self.parent, self.dispatcher = name, parent, dispatcher
        self.level: Optional[Level] = None
        self.handlers: list[Handler] = []

    def effective_level(self) -> Level:
        node: Optional[Logger] = self
        while node is not None:
            if node.level is not None:
                return node.level
            node = node.parent
        return Level.INFO

    def add_handler(self, handler: Handler) -> None:
        self.handlers = self.handlers + [handler]     # copy, so readers never see a partial list

    def log(self, level: Level, message: str) -> None:
        if level < self.effective_level():
            return
        record = LogRecord(self.name, level, message, time.time(), threading.current_thread().name)
        chain: list[Handler] = []
        node: Optional[Logger] = self
        while node is not None:
            chain.extend(node.handlers)
            node = node.parent
        self.dispatcher.submit(chain, record)

    def info(self, message: str) -> None: self.log(Level.INFO, message)
    def error(self, message: str) -> None: self.log(Level.ERROR, message)


class LogManager:
    def __init__(self) -> None:
        self.dispatcher = AsyncDispatcher()
        self.root = Logger("root", None, self.dispatcher)
        self.loggers: dict[str, Logger] = {"root": self.root}
        self.lock = threading.Lock()

    def get_logger(self, name: str) -> Logger:
        with self.lock:
            if name in self.loggers:
                return self.loggers[name]
            parent = self.root
            parts = name.split(".")
            for i in range(1, len(parts)):
                prefix = ".".join(parts[:i])
                parent = self.loggers.setdefault(prefix, Logger(prefix, parent, self.dispatcher))
            logger = Logger(name, parent, self.dispatcher)
            self.loggers[name] = logger
            return logger

    def shutdown(self) -> None:
        self.dispatcher.stop()

6. Extensions, concurrency and what interviewers probe

• Full queue policy: dropping is right for debug noise and wrong for errors. A policy object can block on ERROR and drop on DEBUG; the choice belongs in submit, and dropped should itself be reported so loss is visible.
• Locking: if asked for a synchronous version, one lock per sink is enough; a global lock serialises unrelated files. The only other shared state is the registry, guarded in get_logger, and the handler list, which is replaced rather than mutated so log reads it without a lock.
• Ordering: one queue preserves per-thread order. One worker per sink would parallelise I/O but needs a queue per sink, and then ordering holds per sink only.
• Cost when disabled: the level check comes first and the message is not formatted yet. Passing an f-string still pays for building it; a lazy argument form avoids that, which is the detail interviewers fish for.
• Test first: app.db inherits a level set on app; a MemorySink holds every record after shutdown; a sink that raises does not stop later records; eight threads each logging a thousand lines produce exactly 8,000 lines with no torn output.
• Common mistake: letting a logging failure reach the caller. A full disk must not take down a checkout.

02Where this question was reported

• Hello Interview · Hello Interview breakdown 'Logging Service' (Medium, premium)
• LLD Mastery · Google-tagged (Easy); handler chains, formatting strategies
• LLD Mastery · Google-tagged with Amazon/Microsoft/Apple/Splunk; Chain of Responsibility, Decorator
• Codemia · Codemia Easy
• GitHub · awesome-low-level-design, Easy
• systemdesign.academy · systemdesign.academy solved
• PracHub · PracHub OOD practice prompt
• Exponent / Aced · Aced Google-tagged 'Design a metrics and logging service' (system-level framing)
• raw.githubusercontent.com · 'Design a logging library with levels and sinks' (oop)
• YouTube · Video: 'Google low level system design question - Part 1 - Implement a logger' (creator says asked in a Google SWE interview)
• YouTube · Video: 'Low level System Design of a Logger Part 2: Solving for concurrency and locking'