Three patterns in one small library: a chain that filters and cleans each log line, a bus that fans it out to whoever subscribed, and formatters you swap like plugs. Plus the one rule every logging library lives by: never slow down the code that's logging.
"Design a logger" sounds like a warm-up, and that's why it's asked: in 90 minutes it shows whether you can use design patterns for a reason. The interviewer will usually name them ("use Chain of Responsibility for the log levels", or "make the outputs pluggable"), and then watch whether each pattern solves a real problem in your code or is bolted on to tick a box. It's also a quiet test of production instincts: thread safety, a broken sink, card numbers in log lines, and what happens when the disk is slow.
This lesson builds a compact library in the shape of real ones (Python's logging, Java's Logback): loggers, records, a processing chain, a pub-sub bus, sinks and formatters, and an async sink, then uses it the way a payments service would.
A logging library with three patterns, each with a job: a Chain of Responsibility (level filter, then redaction of card numbers and UPI IDs) decides whether and how a record continues, a pub-sub bus delivers it to sinks subscribed by level and topic, and a Strategy formats it as plain text or JSON. A search error skips the payments alerts, a broken sink is caught and counted, and an async sink takes 4,000 lines from 8 threads without blocking them.
How the round runs#
"Design a logging library for our services. Levels DEBUG to ERROR. Logs can go to several places at once: the console, a file, an alerting system for errors from payments. Each destination can choose its format. We must never log a card number. And logging must not slow down request handling. 90 minutes, working code."
- 0–10 min — clarify: levels, destinations, per-destination filters, formats, redaction, sync or async
- 10–25 min — the record, the logger, and where each pattern goes, agreed before code
- 25–45 min — formatters (Strategy) and sinks, then the bus (Observer)
- 45–60 min — the processor chain (Chain of Responsibility): level filter, redactor, publisher
- 60–75 min — the async sink, and a broken sink that must not break the others
- 75–90 min — a demo, then follow-ups: sampling, hierarchy, MDC, rotation
Requirements and clarifying questions#
| You ask | They answer | What it changes |
|---|---|---|
| Which levels? | DEBUG, INFO, WARN, ERROR, ordered. | An IntEnum, so levels compare with >= |
| One global level, or per destination? | Both: a global minimum, and each destination its own. | A filter at the head of the chain, and a min_level per subscription |
| Can a destination take only some loggers? | Yes: alerting wants only payments.* errors. |
Topic subscriptions by logger-name prefix |
| Formats? | Plain text for the console, JSON for machines. | A Formatter strategy per sink |
| Card numbers, UPI IDs? | Masked before anything leaves the process. | A redaction step that every record passes through |
| What if a destination is slow or fails? | Requests must not wait; other destinations must still work. | An async sink with a bounded queue; the bus isolates failures |
Three patterns, three jobs. Which goes where?
You have Chain of Responsibility, Observer and Strategy. Match each to one of: "decide whether and how a record continues", "deliver one record to many destinations", "turn a record into text".
One strong answer
- Chain of Responsibility: whether and how a record continues. Each link (level filter, redactor, sampler) may pass the record on, change it, or stop it, and links can be added or reordered without the others knowing.
- Observer (pub-sub): deliver one record to many destinations. Sinks subscribe to the bus with a level and a topic; the logger has no idea how many there are.
- Strategy: turn a record into text. Each sink holds a formatter; plain or JSON is a constructor argument.
The classes#
%% title: Logging framework: the classes
classDiagram
class Logger {
+name str
+debug(msg, ctx)
+info(msg, ctx)
+warn(msg, ctx)
+error(msg, ctx)
}
class LogRecord {
+level Level
+logger str
+message str
+context dict
}
class Processor {
<<abstract>>
+next Processor
+then(next) Processor
+handle(record)
+process(record) LogRecord
}
class LogBus {
+subscribe(sink, min_level, prefix) Subscription
+unsubscribe(sub)
+publish(record)
}
class Sink {
<<abstract>>
+formatter Formatter
+write(record)
}
class Formatter {
<<interface>>
+format(record) str
}
class AsyncSink {
-queue Queue
+dropped int
}
Logger --> Processor
Logger ..> LogRecord
Processor <|-- LevelFilter
Processor <|-- Redactor
Processor <|-- Publish
Publish --> LogBus
LogBus "1" o-- "many" Sink
Sink --> Formatter
Formatter <|.. PlainFormatter
Formatter <|.. JsonFormatter
Sink <|-- ConsoleSink
Sink <|-- MemorySink
Sink <|-- AsyncSink
AsyncSink --> SinkA record's journey: Logger.error() builds a LogRecord and hands it to the head of the chain. LevelFilter drops it if it's below the global level. Redactor returns a cleaned copy. Publish hands it to the LogBus, which delivers it to every subscription whose level and topic match. Each sink formats it with its own formatter and writes it. The AsyncSink wraps any other sink and does that writing on its own thread.
The interfaces#
The record and the formatting strategy. Level is an IntEnum so record.level >= Level.WARN just works:
# logkit.py, part 1: the record, and formatting as a strategy
from __future__ import annotations
import json
import queue
import re
import threading
from abc import ABC, abstractmethod
from dataclasses import dataclass, field, replace
from datetime import datetime, timezone
from enum import IntEnum
class Level(IntEnum):
DEBUG = 10
INFO = 20
WARN = 30
ERROR = 40
@dataclass(frozen=True)
class LogRecord:
level: Level
logger: str
message: str
ts: datetime
context: dict[str, object] = field(default_factory=dict)
class Formatter(ABC):
@abstractmethod
def format(self, r: LogRecord) -> str: ...
class PlainFormatter(Formatter):
def format(self, r: LogRecord) -> str:
ctx = " ".join(f"{k}={v}" for k, v in r.context.items())
return f"{r.ts:%H:%M:%S} {r.level.name:<5} {r.logger}: {r.message} {ctx}".rstrip()
class JsonFormatter(Formatter):
def format(self, r: LogRecord) -> str:
return json.dumps({"ts": r.ts.isoformat(), "level": r.level.name, "logger": r.logger, "msg": r.message, **r.context}, default=str)Sinks, the async decorator, and the bus. The bus is a small pub-sub: a subscription is a sink plus a minimum level plus a topic prefix ("payments" matches payments and payments.refunds, not paymentsx).
# part 2: sinks are the subscribers; the bus is the publisher
class Sink(ABC):
def __init__(self, formatter: Formatter) -> None:
self.formatter = formatter
@abstractmethod
def write(self, r: LogRecord) -> None: ...
class ConsoleSink(Sink):
def write(self, r: LogRecord) -> None:
print(self.formatter.format(r))
class MemorySink(Sink):
def __init__(self, formatter: Formatter) -> None:
super().__init__(formatter)
self.lines: list[str] = []
self._lock = threading.Lock()
def write(self, r: LogRecord) -> None:
line = self.formatter.format(r)
with self._lock:
self.lines.append(line)
class AsyncSink(Sink):
"""Decorates a slow sink: callers enqueue and return; one worker thread writes. Full queue -> drop and count."""
def __init__(self, inner: Sink, maxsize: int = 10_000) -> None:
super().__init__(inner.formatter)
self.inner, self.dropped = inner, 0
self._q: queue.Queue[LogRecord | None] = queue.Queue(maxsize)
self._worker = threading.Thread(target=self._run, daemon=True)
self._worker.start()
def write(self, r: LogRecord) -> None:
try:
self._q.put_nowait(r)
except queue.Full:
self.dropped += 1 # never block the request thread for a log line
def _run(self) -> None:
while (r := self._q.get()) is not None:
self.inner.write(r)
def close(self) -> None:
self._q.put(None) # everything queued before this is written first
self._worker.join()
@dataclass
class Subscription:
sink: Sink
min_level: Level
prefix: str # topic: "payments" matches "payments" and "payments.refunds"
class LogBus:
def __init__(self) -> None:
self._subs: list[Subscription] = []
self.sink_errors = 0
def subscribe(self, sink: Sink, min_level: Level = Level.DEBUG, prefix: str = "") -> Subscription:
s = Subscription(sink, min_level, prefix)
self._subs = [*self._subs, s] # copy on write: publishers iterate a stable list
return s
def unsubscribe(self, s: Subscription) -> None:
self._subs = [x for x in self._subs if x is not s]
def publish(self, r: LogRecord) -> None:
for s in self._subs:
if r.level >= s.min_level and (r.logger == s.prefix or r.logger.startswith(s.prefix + ".") or not s.prefix):
try:
s.sink.write(r)
except Exception:
self.sink_errors += 1 # one broken sink must not silence the othersThe patterns, and why#
| Pattern | Where | Why it earns its place |
|---|---|---|
| Chain of Responsibility | Processor → LevelFilter → Redactor → Publish |
Each step is independent and may stop the record. Adding a sampler ("keep 1 in 100 DEBUG lines") is a new link, inserted with then(). |
| Observer / pub-sub | LogBus with Subscriptions |
The logger publishes once; any number of sinks receive. Adding an alerting sink touches no logger. |
| Strategy | Formatter |
The same record becomes a console line or a JSON document, per sink. |
| Decorator | AsyncSink(inner) |
Asynchrony is added around any sink without changing it, and is itself a Sink. |
| Registry | LogManager.get(name) |
One logger per name, so get("search") twice returns the same object, cheaply. |
Concurrency and edge cases#
Logging is called from every thread in the service, so every shared structure must be safe:
- The bus's subscriber list is copy-on-write:
subscribebuilds a new list and swaps the reference. A thread in the middle ofpublishkeeps iterating the old list safely; nobody takes a lock on the hot path. - Sinks that keep state (a file, a list in memory) take their own lock around the write.
- The
AsyncSinkputs records on a bounded queue and returns at once; one worker thread writes them. When the queue is full it drops and counts rather than blocking the request: losing a log line is better than an outage caused by logging.close()puts a sentinel on the queue and joins the worker, so everything logged before shutdown is written.
- A sink throws (disk full): catch it in the bus, count it, carry on delivering to the other sinks; never let it reach the caller
- Secrets in the context, not just the message: redact both
- A logger name that is a prefix of another (
payvspayments): match whole segments, not rawstartswith - Expensive messages at a filtered level:
log.debug(f"{huge_dump()}")builds the string even when DEBUG is off; accept a callable or check the level first - Shutdown: flush async sinks, or the last seconds before a crash, the ones you need most, are lost
- Logging inside a sink: a sink that logs its own errors through the same bus can loop forever; sinks report problems through a counter
The code#
The chain: every link implements process, which returns the record to pass on (possibly changed) or None to stop.
# part 3: the processing chain: each link may change the record, stop it, or pass it on
class Processor(ABC):
def __init__(self) -> None:
self.next: Processor | None = None
def then(self, nxt: Processor) -> Processor:
self.next = nxt
return nxt
def handle(self, r: LogRecord) -> None:
out = self.process(r)
if out is not None and self.next is not None:
self.next.handle(out)
@abstractmethod
def process(self, r: LogRecord) -> LogRecord | None: ...
class LevelFilter(Processor):
def __init__(self, min_level: Level) -> None:
super().__init__()
self.min_level = min_level
def process(self, r: LogRecord) -> LogRecord | None:
return r if r.level >= self.min_level else None
class Redactor(Processor):
CARD = re.compile(r"\b(\d{4})\d{4,8}(\d{4})\b")
VPA = re.compile(r"\b([\w.-])[\w.-]*@([a-z]+)\b")
def _clean(self, s: str) -> str:
return self.VPA.sub(r"\1***@\2", self.CARD.sub(r"\1********\2", s))
def process(self, r: LogRecord) -> LogRecord | None:
ctx = {k: self._clean(v) if isinstance(v, str) else v for k, v in r.context.items()}
return replace(r, message=self._clean(r.message), context=ctx)
class Publish(Processor):
def __init__(self, bus: LogBus) -> None:
super().__init__()
self.bus = bus
def process(self, r: LogRecord) -> LogRecord | None:
self.bus.publish(r)
return rThe logger and its registry. LogManager builds the default chain once: filter, redact, publish.
# part 4: the logger callers see, and a registry so each name gets one logger
class Logger:
def __init__(self, name: str, chain: Processor) -> None:
self.name, self.chain = name, chain
def log(self, level: Level, msg: str, **ctx: object) -> None:
self.chain.handle(LogRecord(level, self.name, msg, datetime.now(timezone.utc), ctx))
def debug(self, msg: str, **ctx: object) -> None:
self.log(Level.DEBUG, msg, **ctx)
def info(self, msg: str, **ctx: object) -> None:
self.log(Level.INFO, msg, **ctx)
def warn(self, msg: str, **ctx: object) -> None:
self.log(Level.WARN, msg, **ctx)
def error(self, msg: str, **ctx: object) -> None:
self.log(Level.ERROR, msg, **ctx)
class LogManager:
def __init__(self, min_level: Level = Level.INFO) -> None:
self.bus = LogBus()
self.head = LevelFilter(min_level)
self.head.then(Redactor()).then(Publish(self.bus))
self._loggers: dict[str, Logger] = {}
self._lock = threading.Lock()
def get(self, name: str) -> Logger:
with self._lock:
return self._loggers.setdefault(name, Logger(name, self.head))A payments service using it:
# part 5: wire it up and use it
if __name__ == "__main__":
logs = LogManager(min_level=Level.INFO)
console = ConsoleSink(PlainFormatter())
alerts = MemorySink(JsonFormatter()) # stands in for a pager or an alerting webhook
console_sub = logs.bus.subscribe(console, Level.INFO)
logs.bus.subscribe(alerts, Level.ERROR, prefix="payments")
pay, search = logs.get("payments.refunds"), logs.get("search")
assert logs.get("search") is search
search.debug("cache warm") # dropped by the first link: below INFO
search.info("query served", q="biryani", ms=38)
pay.error("refund failed for card 4111111111111111", vpa="riya.demo@upi", amount=499)
search.error("index shard down", shard=3) # an error, but not on the payments topic
print("alerts:", alerts.lines)
class Broken(Sink):
def write(self, r: LogRecord) -> None:
raise OSError("disk full")
logs.bus.subscribe(Broken(PlainFormatter()))
search.warn("slow query", ms=900) # still reaches the console
print("sink errors:", logs.bus.sink_errors)
store = MemorySink(PlainFormatter())
slow = AsyncSink(store, maxsize=50_000)
logs.bus.subscribe(slow, Level.INFO)
logs.bus.unsubscribe(console_sub) # keep the demo output short
def burst() -> None:
for i in range(500):
search.info("tick", i=i)
workers = [threading.Thread(target=burst) for _ in range(8)]
for w in workers:
w.start()
for w in workers:
w.join()
slow.close()
print("async sink wrote", len(store.lines), "lines, dropped", slow.dropped) # 4000, 0Your timestamps will differ:
19:06:50 INFO search: query served q=biryani ms=38
19:06:50 ERROR payments.refunds: refund failed for card 4111********1111 vpa=r***@upi amount=499
19:06:50 ERROR search: index shard down shard=3
alerts: ['{"ts": "2026-10-24T19:06:50.579193+00:00", "level": "ERROR", "logger": "payments.refunds", "msg": "refund failed for card 4111********1111", "vpa": "r***@upi", "amount": 499}']
19:06:50 WARN search: slow query ms=900
sink errors: 1
async sink wrote 4000 lines, dropped 0The DEBUG line never appears. The card number and the UPI ID are masked in both the console and the JSON. The alerting sink received the payments error and not the search error, which is the topic subscription at work. The broken sink failed once, was counted, and the console still got the WARN line. And 8 threads × 500 lines all reached the async sink.
Extending it: the follow-ups#
- "Only keep 1% of DEBUG logs in production." A
Samplerprocessor, linked in after the level filter. Use a counter (or a hash of a request id, to keep whole requests together), notrandom(), so it's testable. - "Add request ids to every line automatically." A context processor that reads a
contextvars.ContextVarset by the web framework per request and addsrequest_idto the record. This is what Java calls the MDC. - "Per-module levels: DEBUG for
payments, WARN for everything else." Hierarchical loggers: each logger has an optional level and inherits its parent's (payments.refunds→payments→ root). The filter link asks the logger's effective level. - "Rotate log files at 100 MB." A
RotatingFileSinkthat checks the size before each write and renames files. Wrapped inAsyncSink, rotation never blocks a request. - "Ship logs to a central system." A sink that batches records and sends them over the network, wrapped in
AsyncSink, with a retry on the batch. The rest of the library is unchanged.
Common mistakes#
- One
Loggerclass that formats, filters and writes to a hard-coded file - Patterns as decoration: a chain where every link just calls the next, or an observer with one hard-wired subscriber
- Synchronous writes to a slow sink on the request thread
- A blocking
put()on a full queue, which turns a slow disk into a stuck service - No isolation between sinks, so one exception stops every other destination
- Redacting only the message and leaking the card number in the context fields
Logging framework
Q1/4 Which pattern lets a new "sample 1 in 100 DEBUG lines" step be added without touching the others?
Answer & explanation
Answer: B
Sampling decides whether a record continues, which is exactly what a chain link does. It slots in after the level filter.
Q2/4 The async sink's queue is full. What should write() do?
Answer & explanation
Answer: C
Logging must never take the service down. A counted drop is visible in metrics; a blocked request thread is an outage.
Q3/4 An alerting sink subscribed with prefix "payments" and level ERROR. Which record does it receive?
Answer & explanation
Answer: B
It needs both: level at least ERROR, and a logger name equal to "payments" or under "payments.". "paymentsx" is a different topic.
Q4/4 Why is the bus's subscriber list copied on every subscribe instead of being locked?
Answer & explanation
Answer: C
Subscriptions change rarely; publishing happens on every log call. Copy-on-write makes the rare path pay and keeps the hot path lock-free.
New lessons by email
New lessons, side quests and case files as they ship, at most once a week. No spam, unsubscribe anytime.
Discussion · …
Public: everyone reading this lesson sees comments. Be kind and keep it on the material.