Lesson 3: Logging and Tracing

logging properly

print has no levels, no destination control, no structure, and no way to turn off in production. Use the standard library's logging.

# src/recipe_extractor/logging_setup.py
import logging
import sys


def configure_logging(*, level: str = "INFO", json_output: bool = True) -> None:
    """Configure once, at startup, from the composition root."""
    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(JsonFormatter() if json_output else logging.Formatter(
        "%(asctime)s %(levelname)-8s %(name)s %(message)s"
    ))

    root = logging.getLogger()
    root.handlers.clear()
    root.addHandler(handler)
    root.setLevel(level)

    logging.getLogger("httpx").setLevel("WARNING")      # quiet noisy libraries

# In every other module
logger = logging.getLogger(__name__)

Module-scoped loggers using __name__ give you a hierarchy matching your package structure, so you can raise the level for one subsystem without touching the rest. This is why every module in this course has had that line at the top.

Levels, used consistently. DEBUG for detail useful when investigating. INFO for events worth recording in normal operation, such as a request completing. WARNING for something recoverable that happened, such as a retry or a fallback. ERROR for a failure that affected the outcome. CRITICAL for the system being unable to continue.

Two rules that matter more than the definitions. A retry is a warning, not an error, because it recovered, and treating recoveries as errors makes your error rate meaningless. And configure logging once at startup, in the composition root from Module 7, never in library code, because a library calling basicConfig overrides the application's choices.

Structured logging

Prose log lines are unqueryable.

# Unqueryable
logger.info(f"Processed {doc.id} in {ms}ms using {tokens} tokens")

# Queryable
logger.info(
    "document_processed",
    extra={
        "doc_id": doc.id,
        "duration_ms": ms,
        "tokens": tokens,
        "model": model,
        "feature": current_feature.get(),
    },
)

The first requires a regular expression to answer "what is the p95 duration for extraction". The second is a filter and an aggregation.

import json
import logging


class JsonFormatter(logging.Formatter):
    """Emit one JSON object per line, with contextual fields merged in."""

    RESERVED = frozenset(logging.LogRecord("", 0, "", 0, "", None, None).__dict__)

    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "timestamp": self.formatTime(record, "%Y-%m-%dT%H:%M:%S%z"),
            "level": record.levelname,
            "logger": record.name,
            "event": record.getMessage(),
            "request_id": request_id.get(),
            "feature": current_feature.get(),
        }
        for key, value in record.__dict__.items():
            if key not in self.RESERVED and not key.startswith("_"):
                payload[key] = value
        if record.exc_info:
            payload["exception"] = self.formatException(record.exc_info)
        return json.dumps(payload, default=str)

Two conventions worth adopting. The message is an event name, not a sentence, so document_processed rather than "Processed document successfully". Event names are groupable and sentences are not. And fields are consistent across events, so duration_ms everywhere rather than duration in one place and elapsed in another, since inconsistent field names make cross-event queries impossible.

Correlation IDs and contextvars

One user request touches retrieval, several model calls, tool executions, and database queries, possibly concurrently. Without a shared identifier the log lines cannot be reassembled into one story.

Module 10 introduced contextvars for cost attribution. The same mechanism carries a request identifier, and it works across await boundaries where a thread-local would not.

from contextvars import ContextVar
import uuid

request_id: ContextVar[str] = ContextVar("request_id", default="")
trace_context: ContextVar[dict] = ContextVar("trace_context", default={})


@asynccontextmanager
async def request_scope(*, feature: str, user_id: str | None = None):
    """Establish a correlation scope for everything inside."""
    rid = str(uuid.uuid4())
    tokens = [
        request_id.set(rid),
        current_feature.set(feature),
        current_user.set(user_id),
    ]
    started = time.monotonic()
    logger.info("request_started", extra={"feature": feature})
    try:
        yield rid
    except Exception:
        logger.exception("request_failed", extra={"duration_ms": _ms(started)})
        raise
    finally:
        logger.info("request_completed", extra={"duration_ms": _ms(started)})
        for token, var in zip(tokens, (request_id, current_feature, current_user)):
            var.reset(token)

Every log line inside the scope carries the identifier automatically, because the formatter reads the context variable. No function signature changes.

Two things to know about contextvars in async code. A task created with create_task inherits a copy of the current context, so spawned work keeps the identifier, which is what you want. But changes made inside that task do not propagate back out, which is also what you want and occasionally surprises people.

Propagate the identifier across service boundaries by sending it as a header and reading it back on the far side, so a trace spans more than one process.

[IMAGE PROMPT M11-4
Purpose: Show how one correlation identifier ties together concurrent and nested operations from a single request into one trace.
Visual type: Trace tree with a timeline.
Prompt: A clean educational trace diagram with a horizontal time axis. At the top, one wide bar spans the full width labelled "request_scope, request_id abc123". Beneath it, indented child bars are drawn at their time positions: a bar labelled "embed query" occupying an early narrow slot, then three bars stacked vertically at overlapping positions labelled "vector search", "keyword search", and "load history", bracketed together and annotated "concurrent, same request_id", then a bar labelled "rerank", then a wider bar labelled "generate (stream)" containing two internal markers labelled "first token" and "usage". Beneath that, one further indented bar under generate labelled "tool: search_recipes". Every bar carries a small tag reading "abc123". A note on the right reads "one identifier, every line, across await boundaries and concurrent tasks".
Required elements: A parent span spanning the full width, correctly nested and time-positioned child spans, three genuinely overlapping concurrent spans with a bracket, a nested tool span under generate, the shared identifier tag on every bar, the note.
Style: Clean educational illustration, professional, uncluttered, high contrast, flat vector, indentation showing nesting depth.
Layout: Horizontal time axis, spans stacked vertically with indentation indicating parent and child relationships.
Text labels: "request_scope, request_id abc123", "embed query", "vector search", "keyword search", "load history", "concurrent, same request_id", "rerank", "generate (stream)", "first token", "usage", "tool: search_recipes", "abc123", "one identifier, every line, across await boundaries and concurrent tasks".
Aspect ratio: 16:9
Accessibility: Show nesting through indentation and label every span in text rather than relying on colour.
Avoid: Vendor observability tool screenshots, decorative icons, tiny text, watermarks.
Alt text: Trace diagram showing one request span containing an embedding call, three concurrent retrieval operations, a rerank, and a streamed generation containing a nested tool call, all tagged with the same request identifier.
END IMAGE PROMPT]

What to log around a model call, and what must never be logged

async def traced_generate(messages: list[Message], *, model: str) -> Completion:
    started = time.monotonic()
    count = counter.count_request(messages)

    logger.info(
        "model_call_started",
        extra={
            "model": model,
            "message_count": len(messages),
            "input_tokens_estimated": count.total,
            "prompt_version": prompts.CURRENT_VERSION,
        },
    )

    try:
        completion = await provider.generate(messages, model=model)
    except ProviderError as exc:
        logger.warning(
            "model_call_failed",
            extra={"model": model, "error_type": type(exc).__name__,
                   "duration_ms": _ms(started)},
        )
        raise

    logger.info(
        "model_call_completed",
        extra={
            "model": model,
            "duration_ms": _ms(started),
            "input_tokens": completion.usage.input_tokens,
            "cached_input_tokens": completion.usage.cached_input_tokens,
            "output_tokens": completion.usage.output_tokens,
            "reasoning_tokens": completion.usage.reasoning_tokens,
            "finish_reason": completion.finish_reason,
            "cost_usd": str(pricing.cost(completion.usage, model=model, when=date.today())),
        },
    )
    return completion

Everything there is metadata: counts, durations, identifiers, and outcomes. That is deliberate.

What must never be logged. Full prompts and completions, because they contain user content and, in a retrieval system, whatever your corpus holds. API keys and tokens, including inside a repr, which is why Module 7 excluded them. Personal data of any kind. Base64 image payloads, which Module 4 noted will fill a disk. And retrieved chunk text, which is user or customer content by another name.

Log identifiers instead of content. Chunk identifiers rather than chunk text, a prompt version rather than the prompt, a content hash rather than the document. Every one of those is enough to reconstruct the situation from your own storage, without copying sensitive content into a logging system that has different access controls and a different retention period.

This is the single most consequential paragraph in the lesson, because logging content is convenient, it works, and it creates a compliance problem that is discovered much later and is expensive to unwind.

If you must capture content for debugging, do it deliberately: a separate store, explicit opt-in, short retention, access controls, and never the default path.

Redaction and retention as code

Policy that lives in a document is not enforced. Policy in a filter is.

REDACTION_PATTERNS = [
    (re.compile(r"\b[\w.+-]+@[\w-]+\.[\w.]+\b"), "[EMAIL]"),
    (re.compile(r"\bsk-[A-Za-z0-9]{16,}\b"), "[API_KEY]"),
    (re.compile(r"\b\d{13,19}\b"), "[CARD_NUMBER]"),
]

SENSITIVE_FIELDS = frozenset({"api_key", "password", "token", "authorization",
                              "prompt", "completion", "content", "chunk_text"})


class RedactingFilter(logging.Filter):
    """Strip sensitive fields and patterns before a record is emitted."""

    def filter(self, record: logging.LogRecord) -> bool:
        for key in list(record.__dict__):
            if key.lower() in SENSITIVE_FIELDS:
                record.__dict__[key] = "[REDACTED]"
        record.msg = self._scrub(str(record.msg))
        return True

    @staticmethod
    def _scrub(text: str) -> str:
        for pattern, replacement in REDACTION_PATTERNS:
            text = pattern.sub(replacement, text)
        return text

Redaction is a safety net, not a strategy. Pattern matching misses things, and a filter aggressive enough to catch everything mangles legitimate content. The real defence is not putting sensitive data into a log line in the first place, and the filter catches the mistakes.

Retention as code means the expiry is configured rather than intended: a retention period set on the log store, a shorter one for anything containing user content, and a documented reason for each. Module 7's point about TTLs applies here, since a log store without expiry is the same unbounded growth problem with a compliance dimension.