Design a Logger with Independent Handlers
RipplingSoftware Engineer · Technical ScreenLast updated
Design an extensible, thread-safe in-process logging framework that supports multiple isolated handlers, immutable log records, backpressure handling, and dynamic handler registration.
Problem Statement
Design an in-process logging library where a central Logger broadcasts log records to multiple independent LogHandler instances (e.g., console, rotating file, or remote telemetry).
Your design must satisfy the following core requirements:
- Immutable Log Records: A
LogRecordcreated by the logger must be immutable so that one handler cannot alter data seen by another handler. - Isolated Failures: If a handler encounters an unhandled exception or network failure during emit, it must not disrupt the calling application or prevent other handlers from receiving the log record.
- Safe Runtime Reconfiguration: Handlers can be registered or deregistered at runtime without causing race conditions or deadlocks with active logging calls.
- Bounded Backpressure: Asynchronous handlers should buffer records up to a configurable capacity. When the buffer is full, the handler must execute a defined overflow strategy (e.g., drop newest, drop oldest, or block with timeout).
Interface
from enum import Enum
from typing import Any, Dict, Optional
class LogLevel(Enum):
DEBUG = 10
INFO = 20
WARN = 30
ERROR = 40
class OverflowPolicy(Enum):
DROP_NEWEST = "DROP_NEWEST"
DROP_OLDEST = "DROP_OLDEST"
BLOCK = "BLOCK"
class LogRecord:
# Represents an immutable log event
def __init__(self, level: LogLevel, message: str, timestamp: float, metadata: Optional[Dict[str, Any]] = None):
...
class LogHandler:
def __init__(self, min_level: LogLevel = LogLevel.INFO):
...
def emit(self, record: LogRecord) -> None:
"""Process the log record. Must handle errors internally."""
raise NotImplementedError
def close(self) -> None:
"""Flush buffers and release resources."""
pass
class Logger:
def __init__(self, min_level: LogLevel = LogLevel.INFO):
...
def add_handler(self, name: str, handler: LogHandler) -> None:
...
def remove_handler(self, name: str) -> Optional[LogHandler]:
...
def log(self, level: LogLevel, message: str, **kwargs: Any) -> None:
...
Example Usage
logger = Logger(min_level=LogLevel.DEBUG)
# Console handler prints to stdout
console = ConsoleHandler(min_level=LogLevel.INFO)
# Faulty handler always throws an exception
faulty = FaultyNetworkHandler(min_level=LogLevel.DEBUG)
logger.add_handler("console", console)
logger.add_handler("faulty", faulty)
# Should not throw an exception despite faulty handler failing
logger.log(LogLevel.INFO, "User signed in", user_id=42)
# Dynamically remove faulty handler
logger.remove_handler("faulty")
Edge Cases & Considerations
- Concurrent Mutation: A thread removing a handler while another thread is iterating through handlers to dispatch logs.
- Metadata Mutation: Passing a mutable dictionary into
kwargsshould not allow external callers or downstream handlers to mutate the record's payload. - Slow Handlers: Ensure slow handlers (e.g., remote HTTP POST) do not block fast handlers (e.g., console output).
Hints
Hint 1
Use a shallow or deep copy of kwargs/metadata and consider Python's `dataclasses(frozen=True)` or `NamedTuple` to enforce record immutability.
Hint 2
For thread-safe handler dispatch and runtime modifications, consider copy-on-write semantics or a read-write lock so reads don't block other reads.
Hint 3
For asynchronous handlers, decouple emission using a bounded `queue.Queue` with a background worker thread and implement drop policies using `put_nowait()` or `get_nowait()`.
Follow-up questions
- How would you handle graceful shutdown so buffered logs in async handlers are not lost when the application terminates?
- If an async handler drops records due to backpressure, how would you alert monitoring systems without triggering infinite logging recursion?
- How would you implement structured filtering and sampled logging (e.g., log only 1% of DEBUG logs) without changing handler implementations?
Solution walkthrough is for members
Get the full walkthrough with reference code and complexity analysis, plus Kitty Coach at 300 messages a day. Hints, the code runner and community answers stay free.