Observability Basics
10 examples to get you started with Observability - 7 basic and 3 intermediate.
Prerequisites
uv venv && source .venv/bin/activate
uv pip install "structlog>=24.0" "prometheus-client>=0.21" "opentelemetry-sdk>=1.29"- Python 3.14.0 with common observability libraries installed.
Basic Examples
1. Structured Log Line
JSON logs are searchable in Loki, CloudWatch, and ELK.
import json
import logging
logging.basicConfig(level=logging.INFO, format="%(message)s")
log = logging.getLogger("api")
log.info(json.dumps({"event": "order_created", "order_id": "ord_1", "ms": 42}))- One JSON object per line (NDJSON) is the ingestion standard
- Include domain fields (
order_id) not only message strings - Log level still matters for filtering noise
Related: Structured Logging - structlog setup
2. Request Correlation ID
Tie log lines to a single HTTP request.
import uuid
from contextvars import ContextVar
request_id: ContextVar[str] = ContextVar("request_id", default="-")
def set_request_id() -> str:
rid = str(uuid.uuid4())
request_id.set(rid)
return rid- Pass
X-Request-IDfrom gateway or generate at edge - ContextVar propagates ID through async FastAPI calls in same task
- Include
request_idon every log record in that request
Related: Distributed Tracing - trace IDs
3. Counter Metric
Count events for rate and error ratio dashboards.
from prometheus_client import Counter
ORDERS_CREATED = Counter("orders_created_total", "Orders created")
def create_order() -> None:
ORDERS_CREATED.inc()- Counter only increases - use for totals
- Name suffix
_totalis Prometheus convention - Export
/metricsendpoint for scraping
Related: Metrics - histograms and gauges
4. Timer / Latency Histogram
Measure how long operations take.
import time
start = time.perf_counter()
# ... work ...
elapsed_ms = (time.perf_counter() - start) * 1000perf_countermonotonic for durations- Prefer histogram metrics over manual timing in prod
- Log slow thresholds only when exceeding SLO
Related: Performance Monitoring (APM) - latency SLOs
5. Health Endpoint
Load balancers need a cheap OK response.
from fastapi import FastAPI
app = FastAPI()
@app.get("/health/live")
def live():
return {"status": "ok"}- Liveness: process up
- Readiness: can serve traffic (add DB check on separate path)
- Keep liveness fast - no external calls
Related: Health & Readiness - probe design
6. Log Levels
Use levels to separate signal from noise.
import logging
log = logging.getLogger("worker")
log.debug("detail for dev")
log.info("job finished")
log.warning("retrying")
log.error("failed permanently")- INFO default in prod; DEBUG via env flag
- ERROR should trigger alerts when rate spikes
- Do not log secrets at any level
Related: Observability Best Practices - logging rules
7. Exception Logging
Capture stack traces with context.
import logging
log = logging.getLogger("api")
try:
risky()
except Exception:
log.exception("payment failed", extra={"order_id": "1"})log.exceptionincludes traceback automatically- Add business context in
extradict (JSON formatter must include it) - Pair with error tracking for aggregation
Related: Error Tracking - Sentry
Intermediate Examples
8. RED Metrics Sketch
Rate, Errors, Duration per endpoint.
# Rate: requests_total counter
# Errors: requests_failed_total counter
# Duration: request_duration_seconds histogram- RED method quick sanity for request-driven services
- USE method (Utilization, Saturation, Errors) for workers and DB pools
- Pick method matching service type
Related: Metrics - histogram buckets
9. Trace Span Concept
One trace ID, multiple spans across services.
[API span] -> [DB span] -> [SQS publish span]
- Spans have parent/child relationships
- OpenTelemetry standardizes propagation headers
- Sample traces in prod to control cost
Related: Distributed Tracing - OpenTelemetry
10. Three Pillars Together
Incident flow: metric alert → logs with request_id → trace waterfall.
# 1. Alert: error_rate > 1%
# 2. Log query: request_id="abc" AND level=ERROR
# 3. Trace UI: trace_id from log field- Correlate fields (
trace_id,request_id) across pillars - Dashboards link to log search and trace UI
- Runbooks document this flow for on-call
Related: Observability Best Practices - on-call hygiene
Stack versions: This page was written for Python 3.14.0 (stable 3.14, maintenance 3.13), FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+, and uv 0.6+.