Configuration and observability

A service in production runs on a machine you cannot step through with a debugger, with settings someone else chose, and its logs are read by a search engine before they are read by a person. Two habits make that bearable. Configuration is read once, checked, and fails loudly at start-up. And the program describes what it is doing in a form machines can query. The intermediate logging lesson covered loggers, handlers and formatters; this lesson builds on it.

Configuration from the environment

Deployment platforms pass settings as environment variables, so the same build can run in testing and production. Read them in one place, convert them to the right types, and turn the result into an immutable object. Everything else in the program receives that object and never touches os.environ:

from dataclasses import dataclass, field

TRUE = {"1", "true", "yes", "on"}


class ConfigError(Exception):
    """The environment does not describe a runnable service."""


@dataclass(frozen=True)
class Settings:
    port: int
    debug: bool
    database_url: str = field(repr=False)  # may hold a password
    api_key: str = field(repr=False)

The class method does the parsing. It takes the environment as a parameter, so tests pass a plain dictionary:

    @classmethod
    def from_env(cls, env):
        try:
            port = int(env.get("SHOP_PORT", "8000"))
        except ValueError:
            raise ConfigError("SHOP_PORT must be a whole number") from None
        if "SHOP_API_KEY" not in env:
            raise ConfigError("SHOP_API_KEY is required")
        return cls(
            port=port,
            debug=env.get("SHOP_DEBUG", "").lower() in TRUE,
            database_url=env.get("SHOP_DATABASE_URL", "sqlite:///shop.db"),
            api_key=env["SHOP_API_KEY"],
        )

url = "postgresql://shop:s3cret@db.internal/shop"
print(Settings.from_env({"SHOP_API_KEY": "k-91f2", "SHOP_DEBUG": "yes",
                        "SHOP_DATABASE_URL": url}))
for env in ({"SHOP_PORT": "80a", "SHOP_API_KEY": "k"}, {}):
    try:
        Settings.from_env(env)
    except ConfigError as error:
        print("ConfigError:", error)

Settings(port=8000, debug=True)
ConfigError: SHOP_PORT must be a whole number
ConfigError: SHOP_API_KEY is required

field(repr=False) keeps a field out of the representation, so logging the settings at start-up, which is a good habit, cannot leak it. Mark every field that can hold a secret, and remember that URLs often do: postgresql://user:password@host/db puts the password in the middle of an ordinary-looking string. Booleans need care: bool("false") is True, so compare against a set of accepted words. In main, call Settings.from_env(os.environ) once and exit with a clear message on ConfigError. A service that refuses to start with a bad setting is much easier to debug than one that starts and fails on the first request. If you also want a configuration file, read it with tomllib first and let the environment override it, in that order and in that one function.

Logging for a whole application

basicConfig suits scripts. An application with several destinations and per-module levels describes its logging as data and hands it to logging.config.dictConfig. That dictionary can itself come from a TOML or JSON file. First, a formatter that writes one JSON object per record, which log platforms can index field by field:

import json
import logging
import logging.config

# Attribute names every LogRecord has; anything else came from extra=.
STANDARD = set(vars(logging.makeLogRecord({}))) | {"message"}


class JsonFormatter(logging.Formatter):
    def format(self, record):
        entry = {
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
        }
        for key, value in vars(record).items():
            if key not in STANDARD:
                entry[key] = value
        if record.exc_info:
            entry["error"] = repr(record.exc_info[1])
        return json.dumps(entry)

Fields passed with extra= become attributes of the record. Comparing against the attributes of an empty record finds them without a hard-coded list. A real formatter would also add "time": self.formatTime(record); it is left out here so the output can be shown exactly. Now the configuration:

CONFIG = {
    "version": 1,
    "disable_existing_loggers": False,
    "formatters": {"json": {"()": JsonFormatter}},
    "handlers": {
        "stdout": {
            "class": "logging.StreamHandler",
            "stream": "ext://sys.stdout",
            "formatter": "json",
        },
    },
    "loggers": {"shop.db": {"level": "WARNING"}},
    "root": {"level": "INFO", "handlers": ["stdout"]},
}

"()" names a factory for the formatter, ext://sys.stdout refers to an object by import path, and loggers quietens one noisy module without touching its code. Records from every module flow up to the root logger’s single handler:

logging.config.dictConfig(CONFIG)
log = logging.getLogger("shop.orders")
log.info("order placed", extra={"order_id": 42, "total": "19.90"})
logging.getLogger("shop.db").info("query took 3 ms")
try:
    1 / 0
except ZeroDivisionError:
    log.exception("refund failed", extra={"order_id": 42})

{"level": "INFO", "logger": "shop.orders", "message": "order placed", "order_id": 42, "total": "19.90"}
{"level": "ERROR", "logger": "shop.orders", "message": "refund failed", "order_id": 42, "error": "ZeroDivisionError('division by zero')"}

The shop.db message was below that logger’s WARNING level and was dropped. Now a search for order_id:42 finds both lines, with no regular expression over free text. Services usually log to stdout and let the platform collect the stream, rather than managing files themselves.

A request ID in every line

When fifty requests are handled at once, their log lines interleave. Tag each line with the request it belongs to. That needs a value that is “global” to one request but different for the request next to it, which is what contextvars provides.

Context variables

A ContextVar is read with get() and changed with set(). set() returns a token, and reset(token) puts back whatever value was there before, the way a finally block undoes a change. Each asyncio task runs in its own copy of the context, so a value set in one task is invisible to the others. Here two requests run as two tasks, and each pauses at an await, as a real handler does when it waits for a database:

import asyncio
import contextvars

current = "-"  # a plain global, shared by every task
request_id = contextvars.ContextVar("request_id", default="-")


async def handle(rid):
    global current
    current = rid
    token = request_id.set(rid)
    await asyncio.sleep(0)  # the other request runs here
    print(f"{rid}: global says {current}, ContextVar says {request_id.get()}")
    request_id.reset(token)


async def main():
    async with asyncio.TaskGroup() as tg:
        tg.create_task(handle("req-1"))
        tg.create_task(handle("req-2"))
    print("after both:", request_id.get())


asyncio.run(main())

req-1: global says req-2, ContextVar says req-1
req-2: global says req-2, ContextVar says req-2
after both: -

While req-1 waited, req-2 overwrote the global, so req-1 would log under the wrong ID. The ContextVar kept both values apart, and main never saw either. The output is the same on every run, because sleep(0) hands control to the other task at a fixed point. (asyncio fundamentals covers tasks and the event loop.)

Threads differ. On the default build, a new threading.Thread starts with an empty context, so it sees each variable’s default. On the free-threaded build, the thread_inherit_context flag is on, and a thread starts with a copy of the context of the code that called start(). Pass the ID explicitly when you hand work to a thread (threads). Now the logging part: a filter copies the variable onto every record:

import contextvars
import logging
import sys

request_id = contextvars.ContextVar("request_id", default="-")


class RequestIdFilter(logging.Filter):
    def filter(self, record):
        record.request_id = request_id.get()
        return True


handler = logging.StreamHandler(sys.stdout)
handler.addFilter(RequestIdFilter())
handler.setFormatter(
    logging.Formatter("%(request_id)s %(name)s: %(message)s"))
logging.basicConfig(level=logging.INFO, handlers=[handler])
log = logging.getLogger("shop.web")

def handle(rid, path):
    token = request_id.set(rid)
    try:
        log.info("GET %s", path)
        log.info("200 OK")
    finally:
        request_id.reset(token)


handle("req-7f3a", "/cart")
handle("req-91bc", "/checkout")
log.info("idle")

req-7f3a shop.web: GET /cart
req-7f3a shop.web: 200 OK
req-91bc shop.web: GET /checkout
req-91bc shop.web: 200 OK
- shop.web: idle

None of the code that logs had to pass the ID along. handle uses the token to restore the old value, which matters when one thread serves many requests in turn.

Following one request across services

A request ID answers “which lines belong together?” Distributed tracing also answers “where did the time go?”, across every service a request touches. A trace is the whole journey and has one trace ID. A span is one step in it: a name, a start and an end, its own ID, and the ID of its parent span. The trace ID and the caller’s span ID travel between services in a traceparent header, whose format is the W3C Trace Context standard. Here is the idea with the standard library only, using a context variable for the current span:

import contextlib
import contextvars
import itertools
import time

current_span = contextvars.ContextVar("current_span", default=None)
span_ids = itertools.count(1)  # real tracers use random 64-bit ids


@contextlib.contextmanager
def span(name, trace_id=None, parent_id=None):
    """Record one step: its trace, its parent, and how long it took."""
    parent = current_span.get()
    if parent is not None:
        trace_id, parent_id = parent["trace"], parent["id"]
    record = {"trace": trace_id, "id": next(span_ids), "parent": parent_id,
              "name": name}
    token = current_span.set(record)
    start = time.perf_counter()
    try:
        yield record
    finally:
        record["ms"] = (time.perf_counter() - start) * 1000
        current_span.reset(token)
        print(f"trace={record['trace']} span={record['id']} "
              f"parent={record['parent']} {name}")

The service reads the incoming header and opens its first span as a child of the caller’s (the IDs are shortened to 8 characters for printing):

# The caller sent a W3C traceparent header: version-trace-parent-flags.
header = "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"
_version, trace_id, caller_span, _flags = header.split("-")

with span("POST /checkout", trace_id[:8], caller_span[:8]):
    with span("load cart"):
        pass
    with span("charge card"):
        pass

trace=4bf92f35 span=2 parent=1 load cart
trace=4bf92f35 span=3 parent=1 charge card
trace=4bf92f35 span=1 parent=00f067aa POST /checkout

Spans finish inside out, and each knows its parent, so a tracing system can rebuild the tree and draw the timings. Do not ship this sketch. The third-party OpenTelemetry packages do the same job properly: with opentelemetry-api, trace.get_tracer(__name__).start_as_current_span("load cart") replaces span(), and opentelemetry.propagate.extract(headers) reads traceparent. The API alone records nothing; the opentelemetry-sdk package and an exporter send the spans to a tracing system.

Health checks

A load balancer or orchestrator asks a service “are you able to serve?” through a health endpoint. Report each dependency separately, so the answer says what is wrong. The last exercise builds one, with a database check that uses the standard library’s sqlite3 module.

Stopping cleanly

Containers and service managers stop a process by sending it SIGTERM, and send SIGKILL if it has not exited after a grace period. Python installs no handler for SIGTERM, so by default the process dies on the spot: no finally blocks, no atexit functions, no flushed log buffers, and requests in flight are cut off. An asyncio service registers a handler with loop.add_signal_handler(), which runs it safely inside the event loop. This script starts a small service twice and terminates it, first without a handler, then with one:

import signal
import subprocess
import sys

SERVICE = """
import asyncio, signal, sys

async def main(handle_sigterm):
    stop = asyncio.Event()
    if handle_sigterm:
        asyncio.get_running_loop().add_signal_handler(signal.SIGTERM, stop.set)
    print("ready", flush=True)
    try:
        await stop.wait()  # stands in for serving requests
        print("SIGTERM: finishing open requests")
    finally:
        print("cleanup ran")

asyncio.run(main(sys.argv[1] == "yes"))
"""

for handle in ("no", "yes"):
    child = subprocess.Popen([sys.executable, "-c", SERVICE, handle],
                             stdout=subprocess.PIPE, text=True)
    child.stdout.readline()  # wait for "ready"
    child.send_signal(signal.SIGTERM)
    rest = child.stdout.read().splitlines()
    print(f"handler {handle}: exit {child.wait()}, printed {rest}")

handler no: exit -15, printed []
handler yes: exit 0, printed ['SIGTERM: finishing open requests', 'cleanup ran']

A negative return code means “killed by that signal”: 15 is SIGTERM. Without the handler, not even the finally block ran. With it, the event is set, main returns normally, and everything after it (closing the server, flushing logs) happens. A real handler also makes the readiness check fail first, so the load balancer stops sending new requests, and finishes the open ones within the grace period. add_signal_handler() works only on Unix; elsewhere, and in programs without an event loop, use signal.signal(signal.SIGTERM, handler), whose handler runs in the main thread between bytecodes. The threads lesson shows how a threaded service stops its workers with an Event.

When the process crashes or hangs

A segmentation fault in an extension module kills the process without a Python traceback, and a deadlock produces no output at all. faulthandler covers both. Enable it at start-up with faulthandler.enable(), python -X faulthandler or PYTHONFAULTHANDLER=1, and a fatal signal prints every thread’s Python stack (and, from 3.14, the C stack too). For hangs, dump_traceback_later acts as a watchdog:

import subprocess
import sys

# A child process that hangs. faulthandler dumps its stack after 0.5 s.
CHILD = """
import faulthandler
import time

faulthandler.dump_traceback_later(0.5, exit=True)


def wait_for_lock():
    time.sleep(30)


wait_for_lock()
"""

result = subprocess.run([sys.executable, "-c", CHILD],
                        capture_output=True, text=True)
print("exit code:", result.returncode)
for line in result.stderr.splitlines():
    if line.startswith(("Timeout", "  File")):
        print(line)

exit code: 1
Timeout (0:00:00.500000)!
  File "<string>", line 9 in wait_for_lock
  File "<string>", line 12 in <module>

The dump names the function it was stuck in. In a real service, call dump_traceback_later(timeout) before a step that should finish quickly and cancel_dump_traceback_later() after it. faulthandler writes directly to a file descriptor without allocating memory, which is why it still works when the interpreter is broken. To inspect a live process instead, attach with python -m pdb -p PID (3.14).

When not to build this yourself

Do not configure logging in a library: emit records and let the application decide, as the intermediate lesson explained. Do not ship JSON logs from a command-line tool that a person reads; plain lines are kinder. And do not grow your own metrics, tracing or log-shipping system inside the application. Emit structured events and let the platform’s collector and a standard library such as OpenTelemetry move them. Your job is to make the events accurate.

Common pitfalls

• Leaving disable_existing_loggers at its default of True. Loggers created before dictConfig ran are silenced. The first exercise shows it.
• Reading os.environ all over the code. Settings become impossible to list, test or validate.
• Logging a secret through extra=. Structured logs are copied everywhere; redact before logging.
• Health checks that only return “ok”. A check that cannot fail tells the balancer nothing.
• Setting a ContextVar without resetting it. In a thread that serves request after request, the next request starts with the last one’s ID. Keep the token and reset in finally.

import logging
import logging.config

early = logging.getLogger("shop.early")  # created before configuration

logging.config.dictConfig({
    "version": 1,
    "handlers": {
        "out": {"class": "logging.StreamHandler",
                "stream": "ext://sys.stdout"},
    },
    "root": {"level": "INFO", "handlers": ["out"]},
})

late = logging.getLogger("shop.late")
early.warning("A from shop.early")
late.warning("B from shop.late")
logging.getLogger("shop").warning("C from shop")

B from shop.late
C from shop

4
8
ConfigError: SHOP_WORKERS must be 1 to 32, not 0
ConfigError: SHOP_WORKERS must be a whole number, not 'many'

class ConfigError(Exception):
    pass


def read_workers(env):
    text = env.get("SHOP_WORKERS", "4")
    try:
        workers = int(text)
    except ValueError:
        raise ConfigError(f"SHOP_WORKERS must be a whole number, "
                          f"not {text!r}") from None
    if not 1 <= workers <= 32:
        raise ConfigError(f"SHOP_WORKERS must be 1 to 32, not {workers}")
    return workers


for env in ({}, {"SHOP_WORKERS": "8"}, {"SHOP_WORKERS": "0"},
            {"SHOP_WORKERS": "many"}):
    try:
        print(read_workers(env))
    except ConfigError as error:
        print("ConfigError:", error)

4
8
ConfigError: SHOP_WORKERS must be 1 to 32, not 0
ConfigError: SHOP_WORKERS must be a whole number, not 'many'

{"status": "ok", "checks": {"database": "ok"}}
{"status": "degraded", "checks": {"database": "ok", "disk": "failed: only 2% free"}}

import json
import sqlite3


def health(checks):
    """Run each check; any exception marks that check as failed."""
    results = {}
    for name, check in checks.items():
        try:
            check()
            results[name] = "ok"
        except Exception as error:
            results[name] = f"failed: {error}"
    healthy = all(result == "ok" for result in results.values())
    return {"status": "ok" if healthy else "degraded", "checks": results}

db = sqlite3.connect(":memory:")


def disk_space():
    raise OSError("only 2% free")


checks = {"database": lambda: db.execute("SELECT 1")}
print(json.dumps(health(checks)))
checks["disk"] = disk_space
print(json.dumps(health(checks)))

{"status": "ok", "checks": {"database": "ok"}}
{"status": "degraded", "checks": {"database": "ok", "disk": "failed: only 2% free"}}

Summary

• Parse the environment once into a frozen, typed settings object; fail at start-up on bad values; hide secrets from repr.
• dictConfig describes an application’s logging as data. Set disable_existing_loggers: False.
• Structured logs are one JSON object per line, with extra= fields; a ContextVar filter adds a request ID to every record. set() returns a token for reset(), and each asyncio task has its own copy.
• A trace is a tree of spans that shares one trace ID, passed between services in a traceparent header. Use OpenTelemetry rather than your own.
• Health checks report each dependency. Handle SIGTERM (loop.add_signal_handler() or signal.signal()), or the process dies without cleaning up. faulthandler turns crashes and hangs into Python stacks.
• Next: architecture for large codebases.