Struktureret JSON-logging i Python — guide til produktion

Strukturerede logs gør fejlfinding i produktion mulig. Her er opsætningen med structlog til JSON-output klar til Grafana Loki.

---

structlog opsætning

# server/src/core/logging.py

import logging
import sys
import structlog
from structlog.stdlib import BoundLogger

def configure_logging(log_level: str = "INFO", json_output: bool = True) -> None:
    """
    Konfigurér structlog med stdlib logging backend.
    json_output=True i produktion, False i udvikling (pæn konsoloutput).
    """
    shared_processors = [
        structlog.contextvars.merge_contextvars,
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.StackInfoRenderer(),
    ]

    if json_output:
        renderer = structlog.processors.JSONRenderer()
    else:
        renderer = structlog.dev.ConsoleRenderer(colors=True)

    structlog.configure(
        processors=shared_processors + [
            structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
        ],
        wrapper_class=structlog.stdlib.BoundLogger,
        context_class=dict,
        logger_factory=structlog.stdlib.LoggerFactory(),
        cache_logger_on_first_use=True,
    )

    formatter = structlog.stdlib.ProcessorFormatter(
        processor=renderer,
        foreign_pre_chain=shared_processors,
    )

    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(formatter)

    root_logger = logging.getLogger()
    root_logger.addHandler(handler)
    root_logger.setLevel(log_level)

def get_logger(name: str) -> BoundLogger:
    return structlog.get_logger(name)

---

FastAPI: request-kontekst

# server/src/middleware/logging.py

import uuid
import time
import structlog
from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware

logger = structlog.get_logger("http")

class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next) -> Response:
        request_id = str(uuid.uuid4())
        start_time = time.perf_counter()

        # Bind request-kontekst til alle log-kald i denne request
        structlog.contextvars.bind_contextvars(
            request_id=request_id,
            method=request.method,
            path=request.url.path,
            client_ip=request.client.host if request.client else None,
        )

        try:
            response = await call_next(request)
            duration_ms = (time.perf_counter() - start_time) * 1000

            logger.info(
                "request_completed",
                status_code=response.status_code,
                duration_ms=round(duration_ms, 2),
            )
            response.headers["X-Request-ID"] = request_id
            return response

        except Exception as exc:
            logger.exception("request_failed", exc_info=exc)
            raise

        finally:
            structlog.contextvars.clear_contextvars()

# Tilføj til FastAPI app:
# app.add_middleware(LoggingMiddleware)

---

Log output eksempler

// Produktion JSON output (Loki-kompatibelt):
{
  "event": "request_completed",
  "logger": "http",
  "level": "info",
  "timestamp": "2026-05-24T06:00:01.234Z",
  "request_id": "550e8400-e29b-41d4-a716-446655440000",
  "method": "GET",
  "path": "/api/v1/portfolio/kpi",
  "client_ip": "185.234.56.78",
  "status_code": 200,
  "duration_ms": 4.23
}

// Fejl-log med stack trace:
{
  "event": "database_error",
  "logger": "server.properties",
  "level": "error",
  "timestamp": "2026-05-24T06:00:02.567Z",
  "request_id": "660e8400-e29b-41d4-a716-446655440000",
  "tenant_id": "550e8400-e29b-41d4-a716-446655440000",
  "property_id": "770e8400-e29b-41d4-a716-446655440000",
  "exception": "asyncpg.exceptions.ConnectionDoesNotExistError",
  "exc_info": "Traceback (most recent call last):\n  ..."
}

---

Kontekst-binding i services

# server/src/properties/service.py

import structlog

logger = structlog.get_logger("properties")

async def get_property(
    session: AsyncSession,
    property_id: UUID,
    tenant_id: UUID,
) -> Property:
    log = logger.bind(
        property_id=str(property_id),
        tenant_id=str(tenant_id),
    )

    log.debug("fetching_property")
    result = await session.execute(
        select(Property).where(
            Property.id == property_id,
            Property.tenant_id == tenant_id,
            Property.deleted_at.is_(None),
        )
    )
    prop = result.scalar_one_or_none()

    if not prop:
        log.warning("property_not_found")
        raise HTTPException(status_code=404, detail="Ejendom ikke fundet")

    log.debug("property_fetched", name=prop.name)
    return prop

async def generate_settlement(session: AsyncSession, property_id: UUID) -> Settlement:
    log = logger.bind(property_id=str(property_id))
    log.info("settlement_generation_started")
    
    try:
        result = await _calculate_distribution(session, property_id)
        log.info("settlement_generated", lines=len(result.lines), total_kwh=result.total_kwh)
        return result
    except Exception as e:
        log.exception("settlement_generation_failed")
        raise

---

Celery worker logging

# server/src/workers/logging_config.py

from celery.signals import setup_logging, task_prerun, task_postrun, task_failure
import structlog

@setup_logging.connect
def configure_celery_logging(**kwargs):
    """Celery bruger sit eget logging-system — override med structlog."""
    configure_logging(log_level="INFO", json_output=True)

@task_prerun.connect
def task_started(task_id, task, args, kwargs, **_):
    structlog.contextvars.bind_contextvars(
        task_id=task_id,
        task_name=task.name,
    )
    structlog.get_logger("celery").info("task_started")

@task_postrun.connect
def task_finished(task_id, task, retval, state, **_):
    structlog.get_logger("celery").info("task_completed", state=state)
    structlog.contextvars.clear_contextvars()

@task_failure.connect
def task_failed(task_id, exception, traceback, **_):
    structlog.get_logger("celery").error(
        "task_failed",
        exception=str(exception),
        exc_info=True,
    )

---

Grafana Loki integration

# /etc/promtail/config.yml
# Promtail sender logs til Loki

server:
  http_listen_port: 9080

clients:
  - url: http://loki:3100/loki/api/v1/push

scrape_configs:
  - job_name: mbus-server
    static_configs:
      - targets: [localhost]
        labels:
          job: mbus-server
          env: production
          __path__: /var/log/mbus-server/*.log

    pipeline_stages:
      - json:
          expressions:
            level: level
            request_id: request_id
            tenant_id: tenant_id
            path: path
            duration_ms: duration_ms
      - labels:
          level:
          path:
      - timestamp:
          source: timestamp
          format: RFC3339Nano

# Nyttige LogQL queries i Grafana:

# Alle fejl seneste time:
{job="mbus-server"} | json | level="error"

# Slow requests (over 500ms):
{job="mbus-server"} | json | duration_ms > 500

# Specifik tenant fejl:
{job="mbus-server"} | json | tenant_id="550e8400-..." | level!="info"

# AES-dekrypteringsfejl fra gateway:
{job="mbus-gateway"} |= "DEC_ERR"

---

Log-rotation (Docker)

# docker-compose.yml

services:
  server:
    logging:
      driver: "json-file"
      options:
        max-size: "100m"
        max-file: "5"
        labels: "service,env"
    labels:
      service: "mbus-server"
      env: "production"

---

Konklusion

structlog med JSON-output giver Loki-kompatible strukturerede logs. Middleware binder request_id til alle log-linjer i en request — fejldiagnose i produktion reduceres fra timer til minutter. Celery-tasks binder task_id og task_name automatisk via signals. Log-rotation i Docker med max-size: 100m forhindrer disk-overflow.

Se Grafana Loki monitoring guide eller Docker Compose produktion guide.