LLD: Logger¶

Difficulty: Advanced | Time: 35–45 minutes

1. Problem Statement¶

Design a logging library. Callers log messages at different severity levels (DEBUG, INFO, WARN, ERROR). Each log record can be written to multiple output sinks simultaneously — console, a local file, a remote log-aggregation service — and each sink has its own configurable minimum level (e.g. console shows INFO+, a remote sink only ships WARN+ to save bandwidth). A slow sink, especially the network one, must never block the calling thread. The logger is called from many application threads concurrently and must not corrupt output or lose writes.

2. Requirements¶

Functional (in scope):

• log(level, message) (and convenience debug()/info()/warn()/error()) accepts a record and routes it to all configured sinks
• Each sink has its own minimum level filter, independent of the others
• At least three sink types: console, file, remote/network
• log() returns immediately — the calling thread never blocks on I/O, even if a sink (e.g. the network sink) is slow or temporarily unavailable
• Safe to call from many threads at once without corrupting a sink's output or the logger's internal state

Explicitly out of scope for v1: structured/JSON log querying, log aggregation/search on the server side, distributed tracing correlation IDs (a real extension, noted below), guaranteed delivery across a process crash (a fundamental limit of async logging, discussed in Edge Cases).

3. Entities¶

The nouns in the problem statement: LogLevel, LogRecord, LogSink (interface, with ConsoleSink, FileSink, RemoteSink implementations), LogFormatter, Logger.

4. Class Design¶

classDiagram
    class LogLevel {
        <<enumeration>>
        DEBUG
        INFO
        WARN
        ERROR
    }

    class LogRecord {
        +level: LogLevel
        +message: str
        +logger_name: str
        +timestamp: datetime
    }

    class LogFormatter {
        +format(record) str
    }

    class LogSink {
        <<interface>>
        +min_level: LogLevel
        +write(record)* void
    }
    class ConsoleSink
    class FileSink
    class RemoteSink

    class Logger {
        -name: str
        -level: LogLevel
        -sinks: List~LogSink~
        -queue: Queue~LogRecord~
        +log(level, message) void
        +debug(message) void
        +info(message) void
        +warn(message) void
        +error(message) void
    }

    class LoggerFactory {
        -_loggers: Dict~str, Logger~
        +get_logger(name) Logger
    }

    LogSink <|.. ConsoleSink
    LogSink <|.. FileSink
    LogSink <|.. RemoteSink
    LogSink --> LogFormatter : uses (injected)
    Logger o-- LogSink : holds a configured list
    Logger --> LogRecord : creates
    LoggerFactory --> Logger : creates & caches, keyed by name

The Singleton trap, resolved. The index page's warning is right to flag Logger as the textbook Singleton example done badly: a global Logger.instance() hides a dependency (any class can silently reach for it), can't be swapped for a fake in a test, and — the moment logging is concurrent, which this exercise requires — turns any accidental shared mutable state on that global into a bug that's hard to reproduce. So Logger here is never accessed through a global accessor. It's a normal object: constructed with a name, a level, and a list of sinks, and handed to whatever needs it via constructor injection, exactly as PricingStrategy was injected into ParkingLot — see Dependency Inversion. A class under test receives a fake Logger (or a real one pointed at an in-memory sink) with zero global state to reset between tests.

The wrinkle is ergonomics: nobody wants to thread a Logger through every constructor by hand for something this cross-cutting, and logging.getLogger(__name__) / SLF4J's LoggerFactory.getLogger(Foo.class) are popular for a reason — call-site convenience matters. LoggerFactory.get_logger(name) gives that convenience without collapsing back into a Singleton: it's a cache keyed by name, not a single hidden instance. Each named logger ("payments", "auth", "db.pool") is independently configurable — different levels, different sinks — and, critically, still injectable: production code calls get_logger(__name__) at the module level for convenience, but a class that wants to be strictly unit-testable still accepts a Logger in its constructor and defaults to get_logger(...) only if none is supplied. That default-with-override is the same pattern behind Design Patterns' discussion of avoiding hidden global state while keeping the common case terse.

5. Patterns Applied¶

• Strategy for LogSink — the requirement ("write to console, file, and network simultaneously, each independently configured") is a real variation point that's explicitly named, so Logger holds a List[LogSink] and depends only on the interface. Adding a new sink (e.g. shipping to Datadog) is a new class with zero edits to Logger. See Design Patterns.
• Builder, lightly, for Logger construction — a logger has several optional, order-independent configuration knobs (level, formatter, list of sinks, queue size) and a builder (LoggerBuilder().with_level(...).add_sink(...).build()) keeps the constructor from becoming a five-positional-argument mess as sinks accumulate. See Design Patterns. This is optional polish, not load-bearing — a plain constructor taking sinks: list[LogSink] is equally correct for the interview; mention the builder if config keeps growing.
• Singleton — when it's actually correct here, and when it isn't. The Logger object itself should never be a singleton, for the reasons above. But there's a narrower thing that is legitimately process-wide: a single file handle registry for FileSink. If two independently-constructed Logger instances both write to app.log, two open file handles interleaving writes at the OS level can corrupt lines (one handle's partial write() gets scheduled between another's). The fix isn't making Logger a singleton — it's making the file handle a singleton resource, shared via a small process-wide registry (FileHandleRegistry.get(path) returns the same handle + lock for a given path, however many FileSink or Logger instances reference it). That's the legitimate case for a true singleton: one physical, non-shareable resource (a file descriptor, a network socket, a hardware lock) that genuinely must have exactly one owner in the process — not a convenience object like Logger that has no such physical constraint and every reason to be many independently-testable instances.

6. Core Code¶

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from enum import IntEnum
from queue import Queue, Full, Empty
from threading import Event, Lock, Thread
import sys


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


@dataclass
class LogRecord:
    level: LogLevel
    message: str
    logger_name: str
    timestamp: datetime = field(default_factory=datetime.now)


class LogFormatter:
    def format(self, record: LogRecord) -> str:
        return f"{record.timestamp.isoformat()} [{record.level.name}] {record.logger_name}: {record.message}"


class LogSink(ABC):
    def __init__(self, min_level: LogLevel = LogLevel.DEBUG, formatter: LogFormatter | None = None):
        self.min_level = min_level
        self.formatter = formatter or LogFormatter()

    def accepts(self, record: LogRecord) -> bool:
        return record.level >= self.min_level

    @abstractmethod
    def write(self, record: LogRecord) -> None: ...


class ConsoleSink(LogSink):
    def write(self, record: LogRecord) -> None:
        stream = sys.stderr if record.level >= LogLevel.WARN else sys.stdout
        print(self.formatter.format(record), file=stream)


class FileHandleRegistry:
    """One handle + lock per path, process-wide. Two FileSinks writing app.log
    must not open two descriptors that interleave writes."""

    _handles: dict[str, tuple] = {}
    _lock = Lock()

    @classmethod
    def get(cls, path: str):
        with cls._lock:
            if path not in cls._handles:
                cls._handles[path] = (open(path, "a", encoding="utf-8"), Lock())
            return cls._handles[path]


class FileSink(LogSink):
    def __init__(self, path: str, min_level: LogLevel = LogLevel.DEBUG, formatter: LogFormatter | None = None):
        super().__init__(min_level, formatter)
        self._path = path
        self._file, self._write_lock = FileHandleRegistry.get(path)

    def write(self, record: LogRecord) -> None:
        line = self.formatter.format(record) + "\n"
        with self._write_lock:
            self._file.write(line)
            self._file.flush()

    def close(self) -> None:
        pass  # registry owns the handle; other sinks may still be using it


class RemoteSink(LogSink):
    """Ships records to a network log collector. Slow/flaky by nature — this is
    exactly why sinks are never called on the caller's thread (see Logger below)."""

    def __init__(self, endpoint: str, min_level: LogLevel = LogLevel.WARN, formatter: LogFormatter | None = None):
        super().__init__(min_level, formatter)
        self._endpoint = endpoint

    def write(self, record: LogRecord) -> None:
        payload = self.formatter.format(record)
        self._send(payload)                 # network I/O — can block or raise; handled by the worker, not the caller

    def _send(self, payload: str) -> None:
        ...  # actual HTTP/gRPC call to self._endpoint


class Logger:
    """Never a global. Constructed with its sinks and handed out via
    constructor injection or LoggerFactory.get_logger() — see Class Design."""

    def __init__(self, name: str, level: LogLevel, sinks: list[LogSink], queue_size: int = 10_000):
        self.name = name
        self.level = level
        self.sinks = sinks
        # +1 reserved so shutdown can put_nowait the sentinel even when the
        # producer-facing capacity (queue_size) is full. _shutdown covers the
        # race where the reserved slot is already taken.
        self._queue: Queue[LogRecord | None] = Queue(maxsize=queue_size + 1)
        self._queue_size = queue_size
        self._shutdown = Event()
        self._worker = Thread(target=self._drain_loop, name=f"logger-{name}", daemon=True)
        self._worker.start()

    # --- producer side: called from arbitrary application threads ---

    def log(self, level: LogLevel, message: str) -> None:
        if level < self.level:
            return                          # cheap global filter before touching the queue at all
        record = LogRecord(level=level, message=message, logger_name=self.name)
        try:
            if self._queue.qsize() >= self._queue_size:
                raise Full
            self._queue.put_nowait(record)  # never blocks the caller — see Edge Cases for overflow policy
        except Full:
            sys.stderr.write(f"[logger:{self.name}] queue full, dropping record\n")

    def debug(self, message: str) -> None:
        self.log(LogLevel.DEBUG, message)

    def info(self, message: str) -> None:
        self.log(LogLevel.INFO, message)

    def warn(self, message: str) -> None:
        self.log(LogLevel.WARN, message)

    def error(self, message: str) -> None:
        self.log(LogLevel.ERROR, message)

    # --- consumer side: the single background thread ---

    def _drain_loop(self) -> None:
        while True:
            try:
                record = self._queue.get(timeout=0.5)
            except Empty:
                if self._shutdown.is_set():
                    break
                continue
            if record is None:              # shutdown sentinel
                break
            for sink in self.sinks:
                if not sink.accepts(record):
                    continue
                try:
                    sink.write(record)
                except Exception as exc:      # a broken sink must not take down the worker or the other sinks
                    sys.stderr.write(f"[logger:{self.name}] sink {sink!r} failed: {exc!r}\n")

    def shutdown(self) -> None:
        # put_nowait into the reserved slot — a blocking put(None) can deadlock
        # forever if the queue is already full of records. _shutdown lets the
        # worker exit even if the sentinel loses the race for that slot.
        self._shutdown.set()
        try:
            self._queue.put_nowait(None)
        except Full:
            pass
        self._worker.join(timeout=5)


class LoggerFactory:
    """Gives get_logger(name) convenience without a hidden global instance —
    each name is independently configured and still just an injectable object."""

    _loggers: dict[str, Logger] = {}
    _lock = Lock()

    @classmethod
    def get_logger(cls, name: str, level: LogLevel = LogLevel.INFO, sinks: list[LogSink] | None = None) -> Logger:
        with cls._lock:
            if name not in cls._loggers:
                cls._loggers[name] = Logger(name, level, sinks or [ConsoleSink()])
            return cls._loggers[name]

sequenceDiagram
    participant App as Application thread
    participant Log as Logger.log()
    participant Q as Queue (thread-safe)
    participant W as _drain_loop (worker thread)
    participant Sink as ConsoleSink / FileSink / RemoteSink

    App->>Log: log(level, message)
    activate Log
    Log->>Log: level < self.level? (cheap filter)
    Log->>Q: put_nowait(record)
    Log-->>App: return (non-blocking)
    deactivate Log
    Note over App,Log: caller never waits on I/O

    loop drain loop, single background thread
        W->>Q: get(timeout=0.5)
        Q-->>W: record
        loop for each configured sink
            W->>Sink: sink.accepts(record)?
            alt accepted
                W->>Sink: write(record)
                Sink-->>W: ok, or raises (caught, logged to stderr)
            end
        end
    end

The producer/consumer split is the crux: log() (producer, called from N application threads) only validates the level and does a non-blocking queue.put_nowait, then returns — no I/O happens on the caller's thread. _drain_loop() (consumer, exactly one background thread per Logger) is the only code that ever calls a sink's write(), so all sink I/O — including a slow RemoteSink._send() — happens off the critical path of every caller.

7. Edge Cases¶

8. Concurrency¶

Many application threads call log() on the same Logger concurrently. Two things must not happen: the queue must not be corrupted by concurrent puts, and two records must never interleave into a partial write on the same sink (e.g. two log lines' bytes interleaving mid-line in a file).

The producer-consumer queue solves both, for a different reason each:

• Queue safety: queue.Queue is internally thread-safe — it wraps its own mutex and condition variables — so N threads calling put_nowait concurrently is already correctly synchronized without this design writing a single lock itself. This is the "thread-safety without hand-rolled locks" case from Concurrency Basics: reach for a higher-level, already-correct primitive before reaching for Lock() directly.
• No interleaved writes to a sink: because there is exactly one consumer thread per Logger (the _drain_loop worker), and that worker processes records one at a time, no two threads ever call sink.write() concurrently for sinks owned by that logger — the ordering is serialized by construction, not by a lock around write(). FileSink still takes its own _write_lock (see Core Code) defensively, in case the same underlying file handle is ever shared across more than one Logger instance (exactly the scenario the FileHandleRegistry singleton in Patterns Applied is meant to prevent) — but under the single-consumer-per-logger design, that lock is uncontended in the common case, matching the locks discussion on preferring the smallest lock that's actually needed.
• What a naive alternative would get wrong: if log() called sink.write() directly from the caller's thread instead of enqueueing, every sink would need its own lock, held for the full duration of potentially slow I/O — turning log() into a blocking call and reintroducing exactly the throughput and latency problem the queue exists to remove.

9. Extensibility¶

Interview Questions¶

Key Takeaways¶

Previous: Rate Limiter (LLD) | Next: Notification System (LLD)