Courses › Low-Level Design › Problem Walkthroughs

Design a Logging Framework

Loggers, handlers and formatters, level inheritance, propagation and thread safety.

"Design a logging framework" (something like Python's logging or Java's Log4j) is a favorite because everyone has used one. It tests clean separation of concerns: what gets logged, where it goes and how it's formatted, plus a hierarchy of loggers, and thread safety in the follow-ups.

Requirements

  • Log messages at levels: DEBUG < INFO < WARNING < ERROR.
  • Each logger has a minimum level; lower-level messages are dropped cheaply.
  • Messages can go to several destinations (handlers): the console, a file, a remote service, an in-memory buffer for tests.
  • Each handler has its own minimum level and its own formatter.
  • Loggers are named in a hierarchy (app, app.db, app.db.pool). A child with no level of its own inherits its parent's, and records propagate up to the parent's handlers.
  • Safe to use from many threads; logging must never crash the application.

Model

Level (enum)            DEBUG, INFO, WARNING, ERROR
LogRecord               logger name, level, message, timestamp, thread, extra fields
Logger                  name, level (optional), handlers, parent, propagate
                        debug/info/warning/error -> log(level, msg)
Handler (abstract)      level, formatter, filters; handle(record) -> emit(text)
  ConsoleHandler / FileHandler / MemoryHandler
Formatter               format(record) -> str
LogManager              get_logger(name): one logger per name, builds the parent chain

Separating Logger, Handler and Formatter means each can vary on its own: a JSON formatter works with any handler, and a new destination is one new Handler subclass. The parent chain is a form of Chain of Responsibility: each logger handles a record, then passes it up.

Implementation

import threading
import time
from abc import ABC, abstractmethod
from enum import IntEnum


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


class LogRecord:
    def __init__(self, name, level, msg, **extra):
        self.name, self.level, self.msg, self.extra = name, level, msg, extra
        self.created = time.time()
        self.thread = threading.current_thread().name


class Formatter:
    def __init__(self, fmt="{level} {name}: {msg}"):
        self.fmt = fmt

    def format(self, r):
        return self.fmt.format(level=r.level.name, name=r.name, msg=r.msg,
                               thread=r.thread, time=r.created)


class Handler(ABC):
    def __init__(self, level=Level.DEBUG, formatter=None):
        self.level = level
        self.formatter = formatter or Formatter()
        self.filters = []
        self._lock = threading.Lock()          # one emit at a time per handler

    def handle(self, record):
        if record.level < self.level or not all(f(record) for f in self.filters):
            return
        try:
            text = self.formatter.format(record)
            with self._lock:
                self.emit(text)
        except Exception as e:                 # logging must never crash the caller
            print(f"logging error: {e!r}")

    @abstractmethod
    def emit(self, text): ...


class ConsoleHandler(Handler):
    def emit(self, text):
        print(text)


class MemoryHandler(Handler):
    def __init__(self, **kw):
        super().__init__(**kw)
        self.lines = []

    def emit(self, text):
        self.lines.append(text)


class Logger:
    def __init__(self, name, parent=None):
        self.name, self.parent = name, parent
        self.level = None                      # None = inherit from parent
        self.handlers = []
        self.propagate = True

    def effective_level(self):
        logger = self
        while logger:
            if logger.level is not None:
                return logger.level
            logger = logger.parent
        return Level.WARNING

    def log(self, level, msg, **extra):
        if level < self.effective_level():     # cheap early exit
            return
        record = LogRecord(self.name, level, msg, **extra)
        logger = self
        while logger:
            for h in logger.handlers:
                h.handle(record)
            if not logger.propagate:
                break
            logger = logger.parent

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


class LogManager:
    def __init__(self):
        self.root = Logger("root")
        self._loggers = {"root": self.root}
        self._lock = threading.RLock()         # re-entrant: get_logger recurses for parents

    def get_logger(self, name):
        with self._lock:
            if name not in self._loggers:
                parent_name = name.rpartition(".")[0]
                parent = self.get_logger(parent_name) if parent_name else self.root
                self._loggers[name] = Logger(name, parent)
            return self._loggers[name]

A quick demo:

manager = LogManager()
mem = MemoryHandler(formatter=Formatter("[{level}] {name} - {msg}"))
manager.root.handlers.append(mem)
manager.root.level = Level.INFO

db = manager.get_logger("app.db")
db.debug("connection pool stats")              # dropped: inherits INFO
db.info("connected")

errors = MemoryHandler(level=Level.ERROR)
manager.get_logger("app").handlers.append(errors)
db.error("query failed")                       # reaches app's handler, then root's

print(mem.lines)      # ['[INFO] app.db - connected', '[ERROR] app.db - query failed']
print(errors.lines)   # ['ERROR app.db: query failed']
assert manager.get_logger("app.db") is db

Design decisions to explain

  • Check the level before building the record. Most debug calls in production are disabled, so they should cost almost nothing. Mention lazy formatting too: pass ("user %s", user_id) rather than an f-string, so the message string is only built if it's emitted.
  • Levels inherit, handlers propagate. This lets you configure handlers once at the root and turn up the detail for one subsystem (app.db at DEBUG) without touching anything else.
  • The handler owns its lock. Lines from two threads never interleave, and a slow file handler doesn't block the console handler.
  • Logging never raises. A full disk or a broken formatter shouldn't take down the request that was logging.
  • Filters are small predicates, for example "only records with user_id", or dropping secrets. They add flexibility without new subclasses.

[!NOTE] Python's own logging module This design closely matches Python's standard logging module and Log4j. Saying so is fine, and shows you know the real thing. Interviewers still expect you to design it yourself.

Follow-ups

  • Asynchronous logging: handlers put formatted records on a bounded queue, and a background thread writes them. Decide what happens when the queue is full: block, drop low-level messages, or drop the oldest. Flush the queue on shutdown.
  • File rotation: a RotatingFileHandler that starts a new file at a size or time limit and keeps the last N files.
  • Structured logging: a JsonFormatter that writes extra fields as JSON, so a log platform can search them.
  • Sampling and rate limits: log one in N debug messages, or at most K identical errors per minute.
  • Configuration from a file: build loggers, handlers and formatters from a dictionary, and let operators change levels at runtime.
  • Sensitive data: a filter that masks tokens, emails or card numbers before formatting.

[!TIP] Practice Design a Logging Framework · Design Patterns · Design an In-Memory Pub/Sub Broker