Documentation

Logging Standards

Structured JSON logging, log levels, request context propagation, and what not to log.

Show:

Status: Extracted
Purpose: Structured logging, log levels, log organization, and logging best practices.


Logging Principles

Centralized Logging

Pattern: Centralized logging through infrastructure

Benefits:

  • Consistent logging format
  • Centralized log management
  • Easy log aggregation
  • Request context propagation

Integration Points:

  • API layer: Request/response logging
  • Application layer: Use case logging
  • Domain layer: Business event logging
  • Infrastructure layer: Technical event logging

Structured Logging

Structured JSON Format

Standard: Use structured JSON logging format

Benefits:

  • Machine-readable logs
  • Easy log querying and filtering
  • Consistent log structure
  • Better log analysis

Pattern:

logger.info(
    "account.created",
    extra={
        "request_id": request_id,
        "account_id": account.id,
        "phone": account.phone,
        "path": "/api/v1/accounts",
        "method": "POST",
    },
)

Log Levels

Log Level Usage

DEBUG: Detailed diagnostic information

  • Development debugging
  • Detailed execution flow
  • Variable values

INFO: General informational messages

  • Request start/end
  • Successful operations
  • Important state changes

WARNING: Warning messages

  • Expected error cases
  • Deprecated feature usage
  • Performance warnings

ERROR: Error messages

  • Unexpected errors
  • Exception occurrences
  • Failed operations

Request Logging

Request Start/End Logging

Pattern:

# Log request start
logger.info(
    "request.start",
    extra={
        "request_id": request_id,
        "path": path,
        "method": method,
    },
)
 
# Log request end
logger.info(
    "request.end",
    extra={
        "request_id": request_id,
        "path": path,
        "method": method,
        "status_code": status_code,
        "duration_ms": duration_ms,
    },
)

Middleware: RequestLoggingMiddleware handles request logging automatically


Logging Best Practices

What to Log

✅ Log:

  • Request/response metadata
  • Important business events
  • Error conditions
  • Performance metrics
  • State changes

❌ Don't Log:

  • Passwords, tokens, secrets
  • Full request/response bodies (unless necessary)
  • Excessive debug information in production
  • Personal identifiable information (PII) in plain text

Sensitive Data Handling

Pattern:

  • Don't log passwords, tokens, secrets
  • Mask sensitive data in logs
  • Use structured logging for safe data
  • Encrypt sensitive log data if required

Related Documents

  • Request Context Patterns - Request context propagation
  • Metrics Patterns - Metrics collection
  • Alerting Strategies - Alert configuration

Reference implementation: services/account-api/api/middleware/logging.py

Framework implementations

Abstract reference: 14-observability/

Logging in our FastAPI services is structured (JSON), contextual (includes request ID, actor, service metadata), and consistent (every log event uses the same shape).


Setup

Logging configuration (infrastructure/logging/config.py)

import json
import logging
import time
from typing import Any
 
 
class JsonFormatter(logging.Formatter):
    """Formats log records as single-line JSON objects."""
 
    def __init__(self, service_name: str, environment: str):
        super().__init__()
        self._service = service_name
        self._environment = environment
 
    def format(self, record: logging.LogRecord) -> str:
        log_data: dict[str, Any] = {
            "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(record.created)),
            "level": record.levelname,
            "message": record.getMessage(),
            "logger": record.name,
            "service": self._service,
            "environment": self._environment,
        }
        # Merge any extra={...} fields passed by the caller
        for key, value in record.__dict__.items():
            if key not in logging.LogRecord.__dict__ and not key.startswith("_"):
                log_data[key] = value
 
        if record.exc_info:
            log_data["exception"] = self.formatException(record.exc_info)
 
        return json.dumps(log_data, default=str)
 
 
def configure_logging(service_name: str, environment: str, level: str = "INFO") -> None:
    """Call once at application startup (api/main.py or api/lifespan.py)."""
    formatter = JsonFormatter(service_name, environment)
    handler = logging.StreamHandler()
    handler.setFormatter(formatter)
 
    root = logging.getLogger()
    root.setLevel(level)
    root.handlers = [handler]
 
    # Suppress SQLAlchemy's default verbose logging
    logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING)

Logging service interface (application/logging/service.py)

from abc import ABC, abstractmethod
from typing import Any
 
 
class LoggingService(ABC):
    @abstractmethod
    def info(self, message: str, extra: dict[str, Any] | None = None) -> None: ...
 
    @abstractmethod
    def warning(self, message: str, extra: dict[str, Any] | None = None) -> None: ...
 
    @abstractmethod
    def error(self, message: str, extra: dict[str, Any] | None = None, exc_info: bool = False) -> None: ...

Structured adapter (infrastructure/logging/adapters/structured.py)

import logging
from application.logging.service import LoggingService
 
 
class StructuredLoggingService(LoggingService):
    def __init__(self, logger_name: str):
        self._logger = logging.getLogger(logger_name)
 
    def info(self, message, extra=None):
        self._logger.info(message, extra=extra or {})
 
    def warning(self, message, extra=None):
        self._logger.warning(message, extra=extra or {})
 
    def error(self, message, extra=None, exc_info=False):
        self._logger.error(message, extra=extra or {}, exc_info=exc_info)

Traceability — Request IDs

Every incoming request gets a UUID assigned by RequestContextMiddleware (see 04-middleware-stack.md). This ID is:

  • Added to request.state.ctx["request_id"]
  • Included in every log line emitted during that request via extra={"request_id": ...}
  • Returned in the X-Request-ID response header

This allows all log lines for a single request to be found by filtering on request_id in any log aggregation tool (CloudWatch Logs Insights, Datadog, etc.).


Log Event Conventions

Use dot-separated event names that describe what happened, not free-form messages:

# Good — event names describe the action and outcome
logger.info("account.created", extra={"account_id": str(account.id), "request_id": request_id})
logger.warning("account.not_found", extra={"account_id": str(account_id), "request_id": request_id})
logger.error("message.processing_failed", extra={
    "message_id": str(message_id),
    "request_id": request_id,
    "error_code": "PROCESSING_FAILED",
    "is_retryable": True,
    "exception_type": type(exc).__name__,
}, exc_info=True)
 
# Bad — unstructured free text
logger.info(f"Account {account_id} was created successfully")

Standard extra fields:

FieldTypeWhen to include
request_idstr (UUID)Always (when in a request context)
actor_idstrWhen the action is by an authenticated user
<entity>_idstrThe primary entity involved (e.g. account_id, message_id)
error_codestrOn warnings and errors
is_retryableboolOn errors — helps consumers decide whether to retry
exception_typestrOn errors — type(exc).__name__
duration_msfloatOn performance-sensitive operations

Where Logging Happens

LocationWhat is logged
RequestLoggingMiddlewarerequest.start and request.end (with status code and duration)
error_handler.pyAll exception details at appropriate log level
Application handlersDomain events (entity created, updated, deleted)
Infrastructure adaptersAdapter-level warnings (e.g. CloudWatch circuit-breaker open)

Route handlers themselves should not log directly — that is the middleware's job. Handlers emit domain events, middleware logs the request lifecycle.


Startup

Call configure_logging() at the top of api/main.py or in api/lifespan.py before the app starts handling requests:

# api/main.py
from infrastructure.logging.config import configure_logging
from config.settings import get_settings
 
settings = get_settings()
configure_logging(
    service_name=settings.service_name,
    environment=settings.environment,
    level="DEBUG" if settings.environment == "development" else "INFO",
)