Skip to content

Design a logging framework

TL;DR

  • You build a dotted logger hierarchy (app.api.auth) where the threshold is inherited upwards and the record is propagated upwards through every ancestor's handlers.
  • Three decisions carry the interview: the two axes (handlers are destinations, formatters are representations — Bridge, not a subclass matrix), the bounded queue in front of slow sinks with a named overflow policy, and failure isolation so a full disk never raises into the caller.
  • Patterns that earn their place: Chain of Responsibility, Bridge, Decorator (async), Builder, Null Object. Singleton appears exactly once, on the registry, and you say why.

Problem statement

"Design a logging framework that an application team would use instead of writing to standard output. Modules should get a logger by name and inherit configuration from their parent; each logger fans out to handlers — console, file, rotating file, an in-memory buffer for tests, a remote shipper — and each handler decides how records are rendered. It has to be safe from many threads, it must not block a request thread on a slow sink, and nothing may be lost at shutdown."

Requirements

Functional

  • Levels DEBUG < INFO < WARNING < ERROR < CRITICAL, filtered at both the logger and the handler.
  • Handlers: console (stream), file, size-based rotating file, in-memory, and a remote shipper behind a transport interface.
  • Formatters: plain human-readable text and one JSON object per line; any handler can use any formatter.
  • A dotted logger hierarchy where a.b.c inherits its level from a.b, then a, then the root, and propagates records to their handlers.
  • Filters as predicates on a record: by name subtree, by level range, sampled, or rate limited.
  • An async handler that moves work off the calling thread through a bounded queue.
  • Thread safety, fluent configuration through a builder, and a shutdown that flushes everything.
  • Structured context (a correlation id) attached without changing call signatures.

Non-functional and constraints

  • A disabled log call is one integer comparison, allocating nothing.
  • A failing handler must never propagate an exception to the application.
  • Contention is per handler, not global: two files being written are two independent locks.
  • In-memory, single process, standard library only. Time and IDs are injected, so tests are deterministic.

Out of scope: log parsing, a query language, alerting rules, and cross-process aggregation — that is the Observability, SLOs and error budgets conversation.

Clarifying questions and assumptions

Question to ask Assumption taken here
Do loggers form a hierarchy, or is a logger a flat named channel? A hierarchy. It is the feature that makes "turn on DEBUG for the whole payments subtree" a one-line change.
When a child and its parent both have handlers, do both run? Yes — that is propagation, and propagate=False is the escape hatch for an audit logger that must not leak to the console.
Is the framework allowed to block the caller? Only when you configure it to. The default async policy sheds load and counts drops; BLOCK is available and is a deliberate choice.
What happens when a handler raises? It is recorded in the manager's diagnostics and the siblings still run. An application must not die because a log volume filled up.
Should there be one global logger registry? Yes, one LogManager per process, reachable through instance(). The constructor stays public so tests never share state.
How is the correlation id passed? Ambiently, through a ContextVar, so no call signature grows a trace_id parameter.

Core entities and relationships

  • LogLevel — an ordered IntEnum. NOTSET = 0 means "ask my parent"; that single value is what makes level inheritance work.
  • LogRecord — the immutable event: id, logger name, level, message, creation time (from the injected Clock), thread name and a context dict. One record is shared by every handler in the chain, so it is frozen.
  • Logger — a node in the dotted tree, holding its own level, its own handlers, its own filters and a propagate flag. 1 → * handlers, 1 → 0..1 parent.
  • LogManager — the registry. It owns the root logger, mints records (clock + ids + ambient context), collects handler failures, and drives shutdown. 1 → * loggers.
  • Handler (abstract) — a destination with its own threshold and filters: NullHandler, StreamHandler, InMemoryHandler, FileHandler, RotatingFileHandler, RemoteHandler, and AsyncHandler, which wraps any other handler.
  • Formatter (protocol) — PlainFormatter and JsonFormatter. A handler has a formatter; it does not inherit one.
  • FileSystem and Transport (protocols) — injected so rotation and shipping are testable without a disk or a socket.
  • Filter (protocol) — LevelRangeFilter, NamePrefixFilter, RateLimitFilter, SamplingFilter, attachable to either.
  • LogContext — ambient key/value pairs bound around a unit of work; LoggerConfigBuilder — the fluent front door.

Class diagram

Structure: the hierarchy, the registry and the record that flows through them.

classDiagram
    class LogManager {
        +root: Logger
        -_lock: RLock
        -_diagnostics_lock: Lock
        +instance() LogManager
        +get_logger(name: str) Logger
        +make_record(name, level, message, context) LogRecord
        +on_handler_error(name, handler, record, exc)
        +shutdown()
    }
    class Logger {
        +name: str
        +level: LogLevel
        +propagate: bool
        +parent: Logger
        -_lock: RLock
        +log(level, message) LogRecord
        +effective_level() LogLevel
        +is_enabled_for(level) bool
        +add_handler(handler) Logger
    }
    class LogRecord {
        +id: str
        +logger_name: str
        +level: LogLevel
        +message: str
        +created: float
        +context: Dict~str,str~
    }
    class LogLevel {
        <<enumeration>>
        NOTSET
        DEBUG
        INFO
        WARNING
        ERROR
        CRITICAL
    }
    class LogContext {
        +current() Dict~str,str~
        +bind(values) Iterator
    }
    class LoggerConfigBuilder {
        +logger(name) LoggerConfigBuilder
        +level(level) LoggerConfigBuilder
        +propagate(enabled) LoggerConfigBuilder
        +handler(handler) LoggerConfigBuilder
        +apply() Logger
    }
    class Filter {
        <<interface>>
        +allows(record: LogRecord) bool
    }
    class Handler {
        <<abstract>>
    }
    LogManager "1" *-- "many" Logger
    Logger "0..1" o-- "many" Logger : parent
    Logger --> "many" Handler
    Logger --> "many" Filter
    LogManager ..> LogRecord : creates
    LogManager ..> LogContext : reads
    LogRecord --> LogLevel
    LoggerConfigBuilder ..> Logger : configures

Behaviour: the Bridge. Destinations on the left, representations on the right, and the async Decorator on top.

classDiagram
    class Handler {
        <<abstract>>
        +name: str
        +level: LogLevel
        +formatter: Formatter
        +accepts(record) bool
        +handle(record) bool
        +emit(record)*
        +flush()
        +close()
    }
    class NullHandler
    class StreamHandler
    class InMemoryHandler
    class FileHandler {
        -_lock: Lock
        -_written: int
        +emit(record)
    }
    class RotatingFileHandler {
        +max_bytes: int
        +backup_count: int
        +rotations: int
    }
    class RemoteHandler {
        +batch_size: int
    }
    class AsyncHandler {
        +inner: Handler
        +policy: OverflowPolicy
        +dropped: int
        +state: WorkerState
        +start() AsyncHandler
        +flush()
        +close()
    }
    class Formatter {
        <<interface>>
        +format(record) str
    }
    class PlainFormatter
    class JsonFormatter
    class FileSystem {
        <<interface>>
        +open_append(path) Stream
        +rename(src, dst)
    }
    class LocalFileSystem
    class MemoryFileSystem
    class OverflowPolicy {
        <<enumeration>>
        BLOCK
        DROP_NEWEST
        DROP_OLDEST
    }
    Handler <|-- NullHandler
    Handler <|-- StreamHandler
    Handler <|-- InMemoryHandler
    Handler <|-- FileHandler
    FileHandler <|-- RotatingFileHandler
    Handler <|-- RemoteHandler
    Handler <|-- AsyncHandler
    AsyncHandler o-- "1" Handler : wraps
    AsyncHandler --> OverflowPolicy
    Handler --> "1" Formatter
    Formatter <|.. PlainFormatter
    Formatter <|.. JsonFormatter
    FileHandler --> "1" FileSystem
    FileSystem <|.. LocalFileSystem
    FileSystem <|.. MemoryFileSystem

Design patterns applied

Pattern Where Why it earns its place
Chain of Responsibility Logger._dispatch walking self → parent → … → root Two behaviours ride the same chain: the threshold is inherited, the record is propagated. Configuring one subtree configures everything under it, and propagate=False cuts the chain at exactly one node.
Bridge Handler × Formatter Destination and representation vary independently. Without it you get JsonFileHandler, PlainFileHandler, JsonRemoteHandler — an n × m matrix. With it, a CsvFormatter is one class every handler can use.
Decorator AsyncHandler(inner) Asynchrony is orthogonal to destination. Any handler becomes non-blocking by being wrapped, and the wrapper is itself a Handler, so the logger cannot tell.
Builder LoggerConfigBuilder Configuration has many optional knobs and one invalid half-built state (no handlers and no propagation). It validates once, and apply() is the only method that mutates a live logger — exactly what config reload needs.
Singleton LogManager.instance(), double-checked The registry is genuinely process-wide: two of them mean two roots and two shutdown paths. The constructor stays public, so tests build isolated managers.
Null Object NullHandler A library attaches one so it never has to ask "did the application configure logging?" and never warns about a missing handler.
Factory Method LogManager.get_logger(name) Idempotent, and it builds the ancestor chain eagerly so a late get_logger("app") never has to re-parent an existing app.api.
Strategy Filter implementations, OverflowPolicy Sampling, rate limiting and shedding are policies you swap, not if branches inside the emit path.
Template Method Handler.handle calling emit handle fixes the policy (threshold, then filters, then emit); subclasses supply only the varying step.

Deliberately not used: an Observer registry for handlers. It looks like Observer — one producer, many listeners — but Observer implies decoupled, unordered, best-effort notification, whereas here handler order on a logger is defined, delivery is synchronous by default, and the caller needs the return value. Also not used: a State class per WorkerState; four values with guarded transitions are an enum, not four objects.

Key flows

The hot path: one call, a threshold check, one record, and the walk up the chain.

sequenceDiagram
    autonumber
    actor App as Request thread
    participant API as Logger app.api
    participant M as LogManager
    participant AH as AsyncHandler
    participant ROOT as Logger root
    participant W as Worker thread
    participant FH as RotatingFileHandler
    App->>API: info(GET /orders, route=/orders)
    API->>API: is_enabled_for(INFO)
    alt below the effective level
        API-->>App: None (nothing allocated)
    else enabled
        API->>M: make_record(name, level, message, context)
        M-->>API: LogRecord with clock.now and LogContext.current
        API->>AH: handle(record)
        AH->>AH: accepts, then put_nowait on the bounded queue
        opt queue full
            AH->>AH: shed per OverflowPolicy and count the drop
        end
        AH-->>API: True
        API->>ROOT: propagate to the parent chain
        ROOT-->>API: handlers of every ancestor ran
        API-->>App: LogRecord
    end
    W->>AH: get from the queue
    W->>FH: handle(record)
    FH->>FH: take the file lock, rotate if needed, write one line

The async worker's lifecycle. DRAINING is what makes shutdown honest: close() puts a sentinel behind the backlog and joins, so every record already queued is written before the file descriptor closes.

stateDiagram-v2
    [*] --> IDLE : AsyncHandler constructed
    IDLE --> RUNNING : start()
    RUNNING --> RUNNING : record consumed, task_done
    RUNNING --> DRAINING : close() enqueues the sentinel
    DRAINING --> STOPPED : sentinel seen, thread joined, inner closed
    IDLE --> STOPPED : close() before start()
    STOPPED --> [*]

Implementation

Write it in the order you would defend it: the vocabulary, then the record, then the two Bridge axes, then the hierarchy, and only then the async wrapper and the builder.

Levels are an IntEnum and not a StrEnum, because threshold filtering must be a single integer comparison on the hot path. NOTSET is the sentinel that makes inheritance work, and OverflowPolicy and WorkerState are the two enums the concurrency section turns into behaviour.

code/lld/logging_framework/models.py — levels and policies
class LogLevel(IntEnum):
    """Ordered on purpose: threshold filtering is one integer comparison.

    ``NOTSET`` means "ask my parent" and is what makes the hierarchy work:
    a logger only overrides the inherited threshold when it sets a real level.
    """

    NOTSET = 0
    DEBUG = 10
    INFO = 20
    WARNING = 30
    ERROR = 40
    CRITICAL = 50

    def __str__(self) -> str:
        return self.name


class OverflowPolicy(StrEnum):
    """What an ``AsyncHandler`` does when its bounded queue is full."""

    BLOCK = "block"  # backpressure: the caller waits (never lose a record)
    DROP_NEWEST = "drop_newest"  # shed load: the caller never waits
    DROP_OLDEST = "drop_oldest"  # keep the freshest window of history


class WorkerState(StrEnum):
    IDLE = "idle"  # constructed, worker thread not started
    RUNNING = "running"  # worker consuming the queue
    DRAINING = "draining"  # close() called, finishing the backlog
    STOPPED = "stopped"  # worker joined, sink closed

The errors subclass the shared hierarchy, so an application can catch ValidationError on a bad config without importing anything from this package:

code/lld/logging_framework/models.py — errors
class LoggingConfigError(ValidationError):
    """A logger name, level or handler wiring that cannot be honoured."""


class QueueOverflowError(ConflictError):
    """A bounded queue rejected a record under the BLOCK policy timeout."""


class LoggerShutdownError(InvalidStateError):
    """The manager has been shut down; handlers no longer accept records."""


class HandlerFailure(HandbookError):
    """Raised by a handler's sink. Captured by the manager, never re-raised at the call site."""

The record is frozen on purpose. One record is handed to every handler in the chain, and two handlers may format it concurrently; a mutable record would be a data race waiting for a busy afternoon.

code/lld/logging_framework/models.py — the record
@dataclass(frozen=True, slots=True)
class LogRecord:
    """One immutable logging event.

    It is created once by ``LogManager.make_record`` and then shared by every
    handler in the chain, so it must never be mutated: two handlers formatting
    the same record concurrently is the common case, not the exception.
    """

    id: str
    logger_name: str
    level: LogLevel
    message: str
    created: float  # epoch seconds from the injected Clock
    thread_name: str
    context: dict[str, str] = field(default_factory=dict)

    def is_at_least(self, level: LogLevel) -> bool:
        return self.level >= level

Everything that touches the outside world is a Protocol, so the tests exercise rotation and remote shipping without a disk or a socket:

code/lld/logging_framework/models.py — sink protocols
class Stream(Protocol):
    """The two calls a file-like destination has to support."""

    def write(self, text: str) -> int: ...
    def flush(self) -> None: ...
    def close(self) -> None: ...


class FileSystem(Protocol):
    """Injected so rotation is testable without touching a real disk."""

    def open_append(self, path: str) -> Stream: ...
    def size(self, path: str) -> int: ...
    def exists(self, path: str) -> bool: ...
    def rename(self, src: str, dst: str) -> None: ...
    def remove(self, path: str) -> None: ...


class Transport(Protocol):
    """Where ``RemoteHandler`` ships a batch of formatted lines."""

    def send(self, lines: list[str]) -> None: ...

Formatters are the second axis of the Bridge. Note that the formatter never asks what time it is — the record carries created, the formatter only renders it, which is why the demo output below is byte-for-byte reproducible.

code/lld/logging_framework/formatters.py
class Formatter(Protocol):
    """Anything that turns a record into a line. No inheritance required."""

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


def _timestamp(created: float) -> str:
    """The record carries the time; the formatter only renders it."""
    return datetime.fromtimestamp(created, tz=UTC).strftime("%Y-%m-%dT%H:%M:%SZ")


class PlainFormatter:
    """Human-readable, column-aligned, context appended as ``key=value`` pairs."""

    def __init__(self, show_thread: bool = False) -> None:
        self.show_thread = show_thread

    def format(self, record: LogRecord) -> str:
        head = f"{_timestamp(record.created)} {record.level!s:<8} {record.logger_name:<12}"
        if self.show_thread:
            head = f"{head} [{record.thread_name}]"
        tail = " ".join(f"{k}={v}" for k, v in sorted(record.context.items()))
        return f"{head} {record.message}{' ' + tail if tail else ''}"


class JsonFormatter:
    """One JSON object per line: what a log shipper actually wants to parse."""

    def __init__(self, sort_keys: bool = True) -> None:
        self.sort_keys = sort_keys

    def format(self, record: LogRecord) -> str:
        payload = {
            "ts": _timestamp(record.created),
            "level": str(record.level),
            "logger": record.logger_name,
            "msg": record.message,
            **record.context,
        }
        return json.dumps(payload, sort_keys=self.sort_keys, separators=(",", ":"))

Handler is the first axis. handle is fixed policy — threshold, then filters, then emit — and subclasses override only emit. NullHandler is the Null Object that makes "no configuration" a valid configuration.

code/lld/logging_framework/handlers.py — the base and the Null Object
class Handler(ABC):
    """Template Method: ``handle`` is fixed policy, ``emit`` is the varying step.

    A handler has its own threshold and its own filters, independent of the
    logger that reached it: one logger can feed a DEBUG file and a WARNING pager.
    """

    def __init__(
        self,
        formatter: Formatter | None = None,
        level: LogLevel = LogLevel.NOTSET,
        name: str = "",
    ) -> None:
        self.name = name or type(self).__name__
        self.level = level
        self.formatter: Formatter = formatter or PlainFormatter()
        self._filters: list[Filter] = []

    def add_filter(self, log_filter: Filter) -> Handler:
        self._filters.append(log_filter)
        return self

    def accepts(self, record: LogRecord) -> bool:
        return record.level >= self.level and all(f.allows(record) for f in self._filters)

    def handle(self, record: LogRecord) -> bool:
        """Returns True when the record was actually written by this handler."""
        if not self.accepts(record):
            return False
        self.emit(record)
        return True

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

    def flush(self) -> None:
        """Push buffered records to the destination. Safe to call repeatedly."""
        return None  # unbuffered handlers have nothing to do

    def close(self) -> None:
        self.flush()


class NullHandler(Handler):
    """Null Object: the default a library attaches so it never warns about missing handlers."""

    def emit(self, record: LogRecord) -> None:
        return None

The concrete handlers are where the locks live. Each one owns a lock over the resource it writes; RotatingFileHandler rotates inside the same lock, so no record can be split across a rename.

code/lld/logging_framework/handlers.py — concrete destinations
class StreamHandler(Handler):
    """Console output. The lock keeps two threads from interleaving one line."""

    def __init__(self, stream: TextIO, formatter: Formatter | None = None, level: LogLevel = LogLevel.NOTSET) -> None:
        super().__init__(formatter, level, name="console")
        self._stream = stream
        self._lock = threading.Lock()

    def emit(self, record: LogRecord) -> None:
        line = self.formatter.format(record)
        with self._lock:
            self._stream.write(line + "\n")

    def flush(self) -> None:
        with self._lock:
            self._stream.flush()


class InMemoryHandler(Handler):
    """Keeps the last ``capacity`` formatted lines. The assertion target in tests."""

    def __init__(self, formatter: Formatter | None = None, level: LogLevel = LogLevel.NOTSET, capacity: int = 1000) -> None:
        super().__init__(formatter, level, name="memory")
        self.capacity = capacity
        self._lock = threading.Lock()
        self._lines: list[str] = []
        self._records: list[LogRecord] = []

    def emit(self, record: LogRecord) -> None:
        line = self.formatter.format(record)
        with self._lock:
            self._lines.append(line)
            self._records.append(record)
            if len(self._lines) > self.capacity:
                del self._lines[: len(self._lines) - self.capacity]
                del self._records[: len(self._records) - self.capacity]

    def lines(self) -> list[str]:
        with self._lock:
            return list(self._lines)

    def records(self) -> list[LogRecord]:
        with self._lock:
            return list(self._records)


class FileHandler(Handler):
    """One lock per file descriptor: concurrent writers append whole lines, never halves."""

    def __init__(
        self,
        path: str,
        fs: FileSystem | None = None,
        formatter: Formatter | None = None,
        level: LogLevel = LogLevel.NOTSET,
    ) -> None:
        super().__init__(formatter, level, name=f"file:{path}")
        self._fs: FileSystem = fs or LocalFileSystem()
        self._path = path
        self._lock = threading.Lock()
        self._stream = self._fs.open_append(path)
        self._written = self._fs.size(path)

    @property
    def path(self) -> str:
        return self._path

    def emit(self, record: LogRecord) -> None:
        line = self.formatter.format(record) + "\n"
        with self._lock:
            self._write_locked(line)

    def _write_locked(self, line: str) -> None:
        self._stream.write(line)
        self._written += len(line)

    def flush(self) -> None:
        with self._lock:
            self._stream.flush()

    def close(self) -> None:
        with self._lock:
            self._stream.flush()
            self._stream.close()


class RotatingFileHandler(FileHandler):
    """Size-based rotation *inside* the write lock, so no record is split by a rename."""

    def __init__(
        self,
        path: str,
        max_bytes: int,
        backup_count: int = 3,
        fs: FileSystem | None = None,
        formatter: Formatter | None = None,
        level: LogLevel = LogLevel.NOTSET,
    ) -> None:
        super().__init__(path, fs, formatter, level)
        if max_bytes <= 0 or backup_count < 1:
            raise ValueError("max_bytes must be positive and backup_count at least 1")
        self.max_bytes, self.backup_count = max_bytes, backup_count
        self.rotations = 0

    def _write_locked(self, line: str) -> None:
        if self._written and self._written + len(line) > self.max_bytes:
            self._rotate_locked()
        super()._write_locked(line)

    def _rotate_locked(self) -> None:
        self._stream.flush()
        self._stream.close()
        oldest = f"{self._path}.{self.backup_count}"
        if self._fs.exists(oldest):
            self._fs.remove(oldest)
        for index in range(self.backup_count - 1, 0, -1):
            src = f"{self._path}.{index}"
            if self._fs.exists(src):
                self._fs.rename(src, f"{self._path}.{index + 1}")
        self._fs.rename(self._path, f"{self._path}.1")
        self._stream = self._fs.open_append(self._path)
        self._written = 0
        self.rotations += 1


class RemoteHandler(Handler):
    """Ships batches over an injected transport; a failing transport raises to the manager."""

    def __init__(
        self,
        transport: Transport,
        batch_size: int = 10,
        formatter: Formatter | None = None,
        level: LogLevel = LogLevel.NOTSET,
    ) -> None:
        super().__init__(formatter, level, name="remote")
        self._transport = transport
        self.batch_size = batch_size
        self._lock = threading.Lock()
        self._buffer: list[str] = []

    def emit(self, record: LogRecord) -> None:
        line = self.formatter.format(record)
        with self._lock:
            self._buffer.append(line)
            batch = self._buffer[:] if len(self._buffer) >= self.batch_size else []
            if batch:
                self._buffer.clear()
        if batch:
            self._transport.send(batch)

    def flush(self) -> None:
        with self._lock:
            batch, self._buffer = self._buffer[:], []
        if batch:
            self._transport.send(batch)

Ambient context is a ContextVar, not a parameter and not a threading.local: it is correct for threads and for coroutines, and bind restores the previous mapping on exit even when the body raises.

code/lld/logging_framework/services.py — ambient context
class LogContext:
    """Ambient structured context (correlation id, user id, tenant).

    A ``ContextVar`` is per-thread *and* per-task, so a request handler binds
    once and every log line inside it carries the id without threading an extra
    argument through ten call frames.
    """

    _var: ClassVar[ContextVar[Mapping[str, str]]] = ContextVar(
        "handbook_log_context", default=_EMPTY_CONTEXT
    )

    @classmethod
    def current(cls) -> dict[str, str]:
        return dict(cls._var.get())

    @classmethod
    @contextmanager
    def bind(cls, **values: str) -> Iterator[None]:
        token = cls._var.set(MappingProxyType({**cls._var.get(), **values}))
        try:
            yield
        finally:
            cls._var.reset(token)

Now the centrepiece. effective_level walks up for the threshold; _dispatch walks up for the record. _emit_local is where failure isolation lives — a raising handler is reported to the manager, and its siblings still run.

code/lld/logging_framework/services.py — the logger and the chain
class Logger:
    """A node in the dotted hierarchy: ``app.api.auth`` is a child of ``app.api``.

    Two things travel up the chain, and they travel differently:

    * the **threshold** is inherited -- ``effective_level`` walks up until it
      finds a logger whose level is not ``NOTSET``;
    * the **record** is propagated -- every ancestor's handlers get a turn,
      until a logger with ``propagate=False`` stops the chain.
    """

    def __init__(
        self,
        name: str,
        manager: LogManager,
        level: LogLevel = LogLevel.NOTSET,
        propagate: bool = True,
    ) -> None:
        self.name = name
        self.level = level
        self.propagate = propagate
        self.parent: Logger | None = None
        self._manager = manager
        self._lock = threading.RLock()  # guards the handler and filter lists
        self._handlers: list[Handler] = []
        self._filters: list[Filter] = []

    def add_handler(self, handler: Handler) -> Logger:
        with self._lock:
            if handler not in self._handlers:
                self._handlers.append(handler)
        return self

    def remove_handler(self, handler: Handler) -> Logger:
        with self._lock:
            if handler in self._handlers:
                self._handlers.remove(handler)
        return self

    def clear_handlers(self) -> Logger:
        with self._lock:
            self._handlers.clear()
        return self

    def handlers(self) -> list[Handler]:
        with self._lock:
            return list(self._handlers)

    def add_filter(self, log_filter: Filter) -> Logger:
        with self._lock:
            self._filters.append(log_filter)
        return self

    def effective_level(self) -> LogLevel:
        node: Logger | None = self
        while node is not None:
            if node.level is not LogLevel.NOTSET:
                return node.level
            node = node.parent
        return DEFAULT_ROOT_LEVEL

    def is_enabled_for(self, level: LogLevel) -> bool:
        return level >= self.effective_level()

    def log(self, level: LogLevel, message: str, **context: object) -> LogRecord | None:
        """The single entry point. Returns the record when it was emitted, else None."""
        if not self.is_enabled_for(level):
            return None  # the cheap path: one integer comparison, no record allocated
        record = self._manager.make_record(self.name, level, message, context)
        with self._lock:
            filters = tuple(self._filters)
        if not all(f.allows(record) for f in filters):
            return None
        self._dispatch(record)
        return record

    def debug(self, message: str, **context: object) -> LogRecord | None:
        return self.log(LogLevel.DEBUG, message, **context)

    def info(self, message: str, **context: object) -> LogRecord | None:
        return self.log(LogLevel.INFO, message, **context)

    def warning(self, message: str, **context: object) -> LogRecord | None:
        return self.log(LogLevel.WARNING, message, **context)

    def error(self, message: str, **context: object) -> LogRecord | None:
        return self.log(LogLevel.ERROR, message, **context)

    def critical(self, message: str, **context: object) -> LogRecord | None:
        return self.log(LogLevel.CRITICAL, message, **context)

    def _dispatch(self, record: LogRecord) -> int:
        """Chain of Responsibility: every ancestor gets a turn unless one stops the chain."""
        node: Logger | None = self
        written = 0
        while node is not None:
            written += node._emit_local(record)
            if not node.propagate:
                break
            node = node.parent
        if written == 0:
            self._manager.on_unhandled(record)
        return written

    def _emit_local(self, record: LogRecord) -> int:
        """Run this logger's own handlers. A raising handler never stops its siblings."""
        written = 0
        for handler in self.handlers():  # snapshot: a handler added mid-loop waits for the next record
            try:
                if handler.handle(record):
                    written += 1
            except Exception as exc:  # failure isolation is the whole point
                self._manager.on_handler_error(self.name, handler, record, exc)
        return written

The manager is the registry and the only Singleton on the page. _get_locked creates ancestors eagerly, which is the simplification CPython avoids with placeholder nodes; say that out loud, because the interviewer who knows the standard library will ask.

code/lld/logging_framework/services.py — the registry
class LogManager:
    """The registry: owns the root logger, mints records, and shuts handlers down.

    This is the one place a Singleton is defensible -- a second registry would
    mean two roots and two shutdown paths for one process. ``instance()`` is the
    process-wide accessor; the constructor stays public so every test builds its
    own isolated manager instead of fighting global state.
    """

    _instance_lock: ClassVar[threading.Lock] = threading.Lock()
    _instance: ClassVar[LogManager | None] = None

    def __init__(
        self,
        clock: Clock | None = None,
        ids: IdGenerator | None = None,
        root_level: LogLevel = DEFAULT_ROOT_LEVEL,
    ) -> None:
        self._clock = clock or SystemClock()
        self._ids = ids or SequentialIdGenerator("R")
        self._lock = threading.RLock()  # guards the logger registry
        self._diagnostics_lock = threading.Lock()  # guards the error log and counters
        self.root = Logger(ROOT_NAME, self, level=root_level)
        self._loggers: dict[str, Logger] = {ROOT_NAME: self.root}
        self._errors: list[tuple[str, str]] = []
        self._unhandled = 0
        self._closed = False

    @classmethod
    def instance(cls) -> LogManager:
        """Double-checked locking: the fast path never takes the lock."""
        if cls._instance is None:
            with cls._instance_lock:
                if cls._instance is None:
                    cls._instance = cls()
        return cls._instance

    @classmethod
    def reset_instance(cls) -> None:
        with cls._instance_lock:
            cls._instance = None

    def get_logger(self, name: str) -> Logger:
        """Factory: idempotent, and it creates the whole ancestor chain eagerly."""
        if not name or name.strip() != name or ".." in name or name.startswith(".") or name.endswith("."):
            raise LoggingConfigError(f"invalid logger name: {name!r}")
        with self._lock:
            if self._closed:
                raise LoggerShutdownError("log manager is shut down")
            return self._get_locked(name)

    def _get_locked(self, name: str) -> Logger:
        existing = self._loggers.get(name)
        if existing is not None:
            return existing
        logger = Logger(name, self)
        # Build parents first, so a late `get_logger("app")` never has to re-parent
        # an already-created `app.api` (CPython solves this with PlaceHolder nodes).
        logger.parent = self.root if "." not in name else self._get_locked(name.rsplit(".", 1)[0])
        self._loggers[name] = logger
        return logger

    def loggers(self) -> list[Logger]:
        with self._lock:
            return list(self._loggers.values())

    def make_record(self, logger_name: str, level: LogLevel, message: str, context: Mapping[str, object]) -> LogRecord:
        merged = LogContext.current()
        merged.update({key: str(value) for key, value in context.items()})
        return LogRecord(
            id=self._ids.next_id(),
            logger_name=logger_name,
            level=level,
            message=message,
            created=self._clock.now(),
            thread_name=threading.current_thread().name,
            context=merged,
        )

    def on_handler_error(self, logger_name: str, handler: Handler, record: LogRecord, exc: Exception) -> None:
        """Never raise into the caller: an app must not crash because a disk filled up."""
        with self._diagnostics_lock:
            self._errors.append((handler.name, f"{type(exc).__name__}: {exc}"))

    def on_unhandled(self, record: LogRecord) -> None:
        with self._diagnostics_lock:
            self._unhandled += 1

    def errors(self) -> list[tuple[str, str]]:
        with self._diagnostics_lock:
            return list(self._errors)

    def unhandled_count(self) -> int:
        with self._diagnostics_lock:
            return self._unhandled

    def shutdown(self) -> None:
        """Flush every handler exactly once, even when several loggers share one."""
        with self._lock:
            if self._closed:
                return
            self._closed = True
            unique: dict[int, Handler] = {
                id(handler): handler for logger in self._loggers.values() for handler in logger.handlers()
            }
        for handler in unique.values():
            try:
                handler.close()
            except Exception as exc:  # shutdown must reach every handler
                with self._diagnostics_lock:
                    self._errors.append((handler.name, f"close failed: {type(exc).__name__}: {exc}"))

The async wrapper is the piece worth the most points. The queue is bounded, the policy is explicit, drops are counted rather than silent, and flush is a real barrier built on queue.join() rather than a sleep.

code/lld/logging_framework/handlers.py — the async decorator
class AsyncHandler(Handler):
    """Decorator: takes any handler off the request thread onto one worker thread.

    The queue is bounded on purpose. An unbounded queue turns a logging burst
    into an out-of-memory kill; a bounded queue forces you to answer the only
    question that matters -- block the caller, or drop records and count them.
    """

    def __init__(
        self,
        inner: Handler,
        capacity: int = 1024,
        policy: OverflowPolicy = OverflowPolicy.DROP_NEWEST,
        block_timeout: float | None = None,
        on_error: Callable[[Handler, LogRecord, Exception], None] | None = None,
    ) -> None:
        super().__init__(inner.formatter, inner.level, name=f"async:{inner.name}")
        self.inner = inner
        self.policy = policy
        self.block_timeout = block_timeout
        self._queue: queue.Queue[LogRecord | None] = queue.Queue(maxsize=capacity)
        self._on_error = on_error
        self._state_lock = threading.Lock()
        self._state = WorkerState.IDLE
        self._thread: threading.Thread | None = None
        self._drop_lock = threading.Lock()
        self.dropped = 0

    @property
    def state(self) -> WorkerState:
        with self._state_lock:
            return self._state

    def start(self) -> AsyncHandler:
        with self._state_lock:
            if self._state is not WorkerState.IDLE:
                return self
            self._state = WorkerState.RUNNING
            self._thread = threading.Thread(target=self._run, name=f"log-{self.inner.name}", daemon=True)
            self._thread.start()
        return self

    def emit(self, record: LogRecord) -> None:
        if self.policy is OverflowPolicy.BLOCK:
            try:
                self._queue.put(record, timeout=self.block_timeout)
            except queue.Full as exc:
                raise QueueOverflowError(f"{self.name} queue full after {self.block_timeout}s") from exc
            return
        try:
            self._queue.put_nowait(record)
        except queue.Full:
            self._shed(record)

    def _shed(self, record: LogRecord) -> None:
        """Count what was actually lost: the evicted record, the new one, or both.

        Both halves of ``DROP_OLDEST`` can race. If the queue drained between the
        overflow and the eviction there is nothing to evict, and if a producer
        refilled the freed slot the new record cannot take it. Counting the
        outcome rather than assuming one loss keeps ``dropped`` honest, which is
        the whole point of a bounded queue.
        """
        lost = 0
        if self.policy is OverflowPolicy.DROP_OLDEST:
            try:
                self._queue.get_nowait()
                self._queue.task_done()  # keep join() accounting correct
            except queue.Empty:
                pass
            else:
                lost += 1  # the evicted record is gone whatever happens next
            try:
                self._queue.put_nowait(record)
            except queue.Full:
                lost += 1  # a racing producer took the slot we just freed
        else:
            lost += 1  # DROP_NEWEST: the record never enters the queue
        if lost:
            with self._drop_lock:
                self.dropped += lost

    def _run(self) -> None:
        while True:
            record = self._queue.get()
            if record is None:
                self._queue.task_done()
                return
            try:
                self.inner.handle(record)
            except Exception as exc:  # one bad record must not kill the worker
                if self._on_error is not None:
                    self._on_error(self.inner, record, exc)
            finally:
                self._queue.task_done()

    def flush(self) -> None:
        """Deterministic barrier: return once the worker has drained the backlog."""
        if self.state is WorkerState.RUNNING:
            self._queue.join()
        self.inner.flush()

    def close(self) -> None:
        with self._state_lock:
            if self._state is not WorkerState.RUNNING:
                self._state = WorkerState.STOPPED
                self.inner.close()
                return
            self._state = WorkerState.DRAINING
            thread = self._thread
        self._queue.put(None)
        if thread is not None:
            thread.join()
        self.inner.close()
        with self._state_lock:
            self._state = WorkerState.STOPPED

Finally the builder: one validated place to describe a logger, and the replace_handlers step that config reload needs.

code/lld/logging_framework/services.py — the builder
class LoggerConfigBuilder:
    """Fluent configuration. ``apply()`` is the only method that mutates a logger.

    Configuring logging is the textbook Builder case: many optional knobs, an
    invalid half-configured state you must not expose, and a validation step
    that belongs in one place rather than in every caller.
    """

    def __init__(self, manager: LogManager) -> None:
        self._manager = manager
        self._name = ROOT_NAME
        self._level: LogLevel | None = None
        self._propagate: bool | None = None
        self._replace = False
        self._handlers: list[Handler] = []
        self._filters: list[Filter] = []

    def logger(self, name: str) -> LoggerConfigBuilder:
        self._name = name
        return self

    def level(self, level: LogLevel) -> LoggerConfigBuilder:
        if not isinstance(level, LogLevel):
            raise LoggingConfigError(f"level must be a LogLevel, got {level!r}")
        self._level = level
        return self

    def propagate(self, enabled: bool) -> LoggerConfigBuilder:
        self._propagate = enabled
        return self

    def handler(self, handler: Handler) -> LoggerConfigBuilder:
        self._handlers.append(handler)
        return self

    def filter(self, log_filter: Filter) -> LoggerConfigBuilder:
        self._filters.append(log_filter)
        return self

    def replace_handlers(self) -> LoggerConfigBuilder:
        """Drop whatever was attached before applying, for config reload."""
        self._replace = True
        return self

    def apply(self) -> Logger:
        if self._replace and not self._handlers and self._propagate is False:
            raise LoggingConfigError(f"{self._name} would have no handlers and no propagation")
        logger = self._manager.get_logger(self._name)
        if self._replace:
            for handler in logger.handlers():
                handler.close()
            logger.clear_handlers()
        if self._level is not None:
            logger.level = self._level
        if self._propagate is not None:
            logger.propagate = self._propagate
        for handler in self._handlers:
            logger.add_handler(handler)
        for log_filter in self._filters:
            logger.add_filter(log_filter)
        return logger

Running python -m lld.logging_framework.demo shows inheritance, propagation, an ambient correlation id, rotation, isolation and shutdown in one pass:

app.api level=DEBUG (its own)  app.db level=INFO (inherited)
2023-11-14T22:13:20Z INFO     app.api      GET /orders correlation_id=c-42 route=/orders user=u-7
2023-11-14T22:13:21Z DEBUG    app.api      cache miss correlation_id=c-42 key=orders:u-7 user=u-7
2023-11-14T22:13:22Z ERROR    app.api      upstream timeout correlation_id=c-42 upstream=billing user=u-7
rotating file wrote 1 live line(s), 1 rotation(s)
files on disk: /var/log/app.log, /var/log/app.log.1
audit sink (ERROR and above) captured 1: {"correlation_id":"c-42","level":"ERROR","logger":"app.api","msg":"upstream timeout","ts":"2023-11-14T22:13:22Z","upstream":"billing","user":"u-7"}
2023-11-14T22:13:23Z WARNING  app.api      degraded mode
manager errors after a failing handler: [('broken', 'HandlerFailure: disk full')]
async worker state=stopped, dropped=0
unhandled records (no handler anywhere): 0

Concurrency and edge cases

Which lock protects what. There are four kinds, and naming them in this order is the answer:

  1. FileHandler._lock guards one file descriptor and the byte counter beside it. Two threads appending to the same log serialise here; a thread writing to a different file does not wait. An uncontended mutex costs about 17 ns, so this is free relative to any real write — the mistake would be sharing one lock across unrelated files.
  2. Logger._lock (an RLock) guards the handler and filter lists. _emit_local snapshots and iterates outside the lock, so a slow handler never blocks a thread that is merely attaching one. State the consequence: a handler attached mid-dispatch starts receiving from the next record.
  3. LogManager._lock guards the logger registry. It is an RLock because _get_locked recurses to build ancestors.
  4. LogManager._diagnostics_lock guards the failure list and the counters — deliberately separate, so recording a handler failure never contends with get_logger.

The record is immutable, so it needs no lock at all. That is the cheapest concurrency decision on the page.

Queue overflow. The queue is bounded at 1,024 records. Sizing: a stateless app server sustains roughly 1k QPS and a chatty request emits about 5 log lines, so 5k lines/s at 200 B–1 KB per line is 1–5 MB/s, and 1,024 buffered records is under a megabyte of slack. When the sink cannot keep up, exactly one of three things happens and you must name it: BLOCK applies backpressure, DROP_NEWEST sheds the burst, DROP_OLDEST keeps the freshest window. All three are counted — silent loss is the failure mode that costs you the offer.

Handler failure isolation. _emit_local calls handle inside try/except: a failure lands in the manager's diagnostics, the loop continues, and the application call returns normally. The worker thread has the same guard, so one poisonous record cannot kill the drain.

Shutdown. LogManager.shutdown collects handlers into a dict keyed by id(), so one shared by three loggers is closed exactly once. For AsyncHandler, close() enqueues the sentinel behind the backlog and joins the worker; the test asserts 50 queued records all reach disk.

Rotation during a write cannot interleave: _rotate_locked runs inside _write_locked, whose caller already holds the lock. It drops the oldest generation, renames .1 → .2, then reopens.

Other edge cases handled: a disabled call returns None without building a record; a record that reaches no handler anywhere increments unhandled_count instead of vanishing; invalid logger names ("", "app.", "app..api") raise LoggingConfigError; a rate-limited logger reopens its window on the injected clock; LogContext.bind restores the previous mapping even if the block raises.

Common mistake

Putting one global lock around "emit to all handlers", or — worse — an unbounded queue.Queue in front of the file. The global lock makes the slowest sink the speed of the whole application; the unbounded queue converts a downstream stall into an out-of-memory kill, which is a far more expensive outage than dropping debug lines. Bound the queue, name the overflow policy, and count what you drop.

Extensibility and follow-ups

  • Sampling and rate limiting: already a Filter. SamplingFilter(rate=0.01, seed=42) keeps one DEBUG line in a hundred while letting every WARNING through; RateLimitFilter(max_per_window, window_seconds, clock) caps a hot loop. Attach either to a logger or to a single handler; nothing else changes.
  • Remote shipping with batching: RemoteHandler buffers batch_size lines and ships them through an injected Transport. The arithmetic makes the case: a same-datacenter round trip is about 500 µs, so 5k lines/s shipped one at a time would need 5,000 × 500 µs = 2.5 s of blocking per wall-clock second — impossible on one thread. Batch 100 lines and it is 50 round trips/s, about 25 ms of network time per second. Wrap it in AsyncHandler and the request thread pays none of it.
  • Config reload: replace_handlers() closes the old handlers (flushing them) before attaching the new ones, so a SIGHUP reload loses nothing. A file-driven config layer is then just a translation from a mapping into builder calls.
  • Time-based rotation: a sibling of RotatingFileHandler overriding the same _write_locked hook, triggering on clock.now() crossing a boundary instead of on byte count.
  • A new destination (syslog, a database, a pager) is one Handler; a new representation (logfmt, CSV) is one Formatter. That is the Bridge paying for itself, and it is the sentence to say when the interviewer asks "how would you add X?".
  • Multi-process aggregation is where this becomes an HLD question: a per-host agent tailing files, a queue, an indexer, retention tiers, and a sampling budget.

Interview tip

When you draw the handler hierarchy, draw the formatter axis next to it and say "two axes, so I compose instead of multiplying classes". Then, unprompted, name the bounded queue and its overflow policy. Those two sentences separate a candidate who has used a logging library from one who has thought about writing one.

Tests

tests/test_logging_framework.py has 21 cases (17 functions, four of them parameterised). The ones worth walking through are the hierarchy test, the isolation test and the concurrency test.

The hierarchy test asserts both directions of the chain at once — the threshold coming down and the record going up:

code/lld/logging_framework/tests/test_logging_framework.py — hierarchy
def test_record_climbs_the_hierarchy_and_levels_are_inherited(manager: LogManager) -> None:
    root_sink, api_sink = InMemoryHandler(), InMemoryHandler()
    manager.root.add_handler(root_sink)
    LoggerConfigBuilder(manager).logger("app.api").level(LogLevel.DEBUG).handler(api_sink).apply()
    auth = manager.get_logger("app.api.auth")

    assert auth.effective_level() is LogLevel.DEBUG  # inherited from app.api
    assert manager.get_logger("app.db").effective_level() is LogLevel.INFO  # inherited from root

    auth.debug("token refreshed")
    assert len(api_sink.lines()) == 1  # the ancestor's handler ran
    assert len(root_sink.lines()) == 1  # ...and so did the root's
    assert manager.get_logger("app.db").debug("noisy") is None  # below the inherited threshold

Isolation is the test that proves logging is best effort. The broken handler raises, the sibling still writes, the caller gets its record back, and the failure is recorded rather than swallowed:

code/lld/logging_framework/tests/test_logging_framework.py — failure isolation
def test_a_failing_handler_never_stops_its_siblings(manager: LogManager) -> None:
    good = InMemoryHandler()
    LoggerConfigBuilder(manager).logger("app.api").handler(BrokenHandler()).handler(good).apply()

    record = manager.get_logger("app.api").error("upstream timeout")

    assert record is not None  # the caller is not told; logging is best effort
    assert len(good.lines()) == 1  # the sibling handler still wrote
    assert manager.errors() == [("broken", "HandlerFailure: disk full")]

The concurrency test drives 400 records through eight threads into one FileHandler and asserts the invariant that the per-handler lock exists to protect: 400 whole lines, every message present exactly once, none interleaved.

code/lld/logging_framework/tests/test_logging_framework.py — concurrency
def test_concurrent_writers_never_interleave_a_line(manager: LogManager) -> None:
    files = MemoryFileSystem()
    handler = FileHandler("/var/log/app.log", fs=files, formatter=PlainFormatter(show_thread=True))
    manager.root.add_handler(handler)
    log = manager.get_logger("app.api")

    def write(i: int) -> None:
        log.info(f"request-{i:03d}")

    with ThreadPoolExecutor(max_workers=8) as pool:
        list(pool.map(write, range(400)))
    handler.close()

    lines = files.lines("/var/log/app.log")
    assert len(lines) == 400  # one lock per file: no line is split or overwritten
    assert {line.split()[-1] for line in lines} == {f"request-{i:03d}" for i in range(400)}

The rest cover: propagate=False cutting the chain; invalid logger names through parametrize; the builder rejecting a non-level and a dead-end configuration; the worker's IDLE → RUNNING → STOPPED transitions; both drop policies keeping the expected two records out of three; BLOCK timing out and reporting through the manager; rotation producing exactly the expected generations; shutdown draining 50 queued records to disk; ambient context disappearing after its block; the rate-limit window reopening on a FakeClock; the Null Object; and remote batching. No test sleeps — the async ones use flush() or close() as barriers. Run them with uv run pytest code/lld/logging_framework -q.

45-minute pacing

Minutes What to do What to say or write
0–5 Clarify Hierarchy or flat names? Do parent handlers also run? May we block the caller? What happens when a sink fails? Out of scope: parsing, alerting, cross-process aggregation.
5–10 Entities Nouns on the board: Logger, LogManager, LogRecord, Handler, Formatter, Filter. Say "handlers and formatters are two axes" before anyone asks.
10–18 Class diagram Hierarchy first, then hang the handler tree off it, then the formatter axis, then AsyncHandler wrapping a handler. Mark the file lock and the registry lock.
18–34 Code LogLevel with NOTSET → LogRecord (frozen) → Handler.handle/emit → Logger.effective_level and _dispatch → FileHandler with its lock → AsyncHandler with the bounded queue.
34–40 Concurrency Name the four locks and what each prevents. Walk the overflow policy and the drop counter. Explain close() draining through the sentinel.
40–45 Extensions Filters for sampling and rate limiting, batching with the 500 µs arithmetic, config reload, and the hand-off to the observability HLD.