Observability — Logging, Metrics, and the Run Log

You can't fix what you can't see

The transition from "my pipeline runs" to "my pipeline is observable" is the same as the transition from a script to a system. Observability has three layers, and each one solves a different class of incident:

Logs — what happened, in order, in human-readable form. Used for debugging.
Metrics — counts, durations, rates emitted at every stage. Used for trends and alerting.
The run log — a per-run record (start time, end time, success/failure, row counts in/out per stage, errors). Used for forensics and SLA tracking.

Structured logs are non-negotiable

Plain-text logs ('started extract') are fine for humans staring at a terminal. Production-grade logs are structured (key=value or JSON), so a log aggregator can index them, alert on patterns, and let you query across runs. The Python logging module gets you there with two lines of config and a habit of including the values that matter.

Code

Structured logging configuration that scales·python

import json
import logging
from datetime import datetime

class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload = {
            'ts': datetime.utcnow().isoformat() + 'Z',
            'level': record.levelname,
            'name': record.name,
            'msg': record.getMessage(),
        }
        # any extras passed via logger.info(..., extra={'kv': {...}})
        if hasattr(record, 'kv'):
            payload.update(record.kv)
        return json.dumps(payload, ensure_ascii=False)

handler = logging.StreamHandler()
handler.setFormatter(JsonFormatter())
logging.basicConfig(level=logging.INFO, handlers=[handler])
log = logging.getLogger('orders_pipeline')

log.info('extract complete', extra={'kv': {
    'stage': 'extract', 'rows': 12_345, 'ms': 1834,
}})

A run-log record — the artifact your dashboard queries·python

import time
from dataclasses import dataclass, field

@dataclass
class RunLog:
    pipeline: str
    window: str
    started_at: float = field(default_factory=time.time)
    stages: dict = field(default_factory=dict)
    success: bool = False
    error: str | None = None
    finished_at: float | None = None

    def stage(self, name: str, **kv) -> None:
        self.stages[name] = {'ms': int(kv.pop('ms', 0)), **kv}

    def to_row(self) -> dict:
        return {
            'pipeline':    self.pipeline,
            'window':      self.window,
            'started_at':  self.started_at,
            'finished_at': self.finished_at,
            'duration_s':  (self.finished_at or time.time()) - self.started_at,
            'success':     self.success,
            'error':       self.error,
            **{f'stage_{k}': v for k, v in self.stages.items()},
        }

Exercise

Add structured JSON logging and a per-run run-log record to one of your pipelines. After the next run, write the run-log row to a Parquet file or DuckDB table. Plot duration over time. The plot is the answer to "is this pipeline getting slower?" — a question you can't answer without observability.