Module 5: Structured Logging for AI Systems

5. JSON Logs with structlog

Next →

Description

JSON logs aren't just a format — they're the interface between your app and any observability system (Datadog, Loki, CloudWatch, ELK). A well-formed JSON log is machine-queryable, indexable by log aggregation systems, and dashboardable with no additional configuration. This capsule implements the complete structlog configuration for production, the essential processors for AI apps, and the querying patterns that turn logs into a real diagnostic system.

JSON Lines: the standard format

Why "JSON Lines" and not pure JSON?

Pure JSON:
[
  {"event": "request_1", "tokens": 450},
  {"event": "request_2", "tokens": 320}
]
Problem: To add an entry, you must read the WHOLE file, parse, append, rewrite.
With millions of logs: impossible.

JSON Lines (JSONL):
{"event": "request_1", "tokens": 450}
{"event": "request_2", "tokens": 320}
Advantages:
→ Append-only: each log is one line, appended at the end
→ Incremental parsing: you can read one line without loading the whole file
→ Easy to grep, sed, awk
→ jq works perfectly
→ All log aggregation systems support it natively
→ Stream-friendly: stdout piped to another process

Complete structlog configuration

# src/logging_config.py
import logging
import os
import sys
import structlog
from typing import Any

def get_log_level() -> int:
    """Reads the log level from an env var or uses INFO by default."""
    level_str = os.getenv("LOG_LEVEL", "INFO").upper()
    return getattr(logging, level_str, logging.INFO)

def configure_structlog(
    env: str = None,
    log_level: int = None,
    json_output: bool = None
) -> None:
    """
    Configures structlog for the specified environment.
    
    Args:
        env: "production", "development", or "testing"
        log_level: Override for the log level
        json_output: Override for the format (True=JSON, False=Console)
    """
    env = env or os.getenv("ENVIRONMENT", "development")
    log_level = log_level or get_log_level()
    
    # Determine whether to use the JSON or Console renderer
    use_json = json_output if json_output is not None else (env == "production")
    
    # ─── Processors to intercept Python's standard logging ──────────────────
    # structlog can capture logs from libraries that use logging.getLogger()
    logging.basicConfig(
        format="%(message)s",
        stream=sys.stdout,
        level=log_level
    )
    
    # ─── Shared processors (dev and prod) ────────────────────────────────────
    shared_processors = [
        # Read request_id (and other fields) from the contextvars context
        structlog.contextvars.merge_contextvars,
        
        # Add level (info, warning, error, etc.)
        structlog.processors.add_log_level,
        
        # Timestamp in ISO 8601 format
        structlog.processors.TimeStamper(fmt="iso"),
        
        # Logger name (useful for identifying the module)
        structlog.stdlib.add_logger_name,
        
        # Render stack traces in exceptions
        structlog.processors.StackInfoRenderer(),
    ]
    
    if use_json:
        # ─── Production: JSON Lines ──────────────────────────────────────────
        processors = shared_processors + [
            # Format the exception as a JSON object (not as text)
            structlog.processors.format_exc_info,
            # Serialize as JSON (one line per event)
            structlog.processors.JSONRenderer()
        ]
    else:
        # ─── Development: readable Console ────────────────────────────────────
        processors = shared_processors + [
            # Colors and human-readable format for the terminal
            structlog.dev.ConsoleRenderer(
                colors=True,
                exception_formatter=structlog.dev.plain_traceback
            )
        ]
    
    structlog.configure(
        processors=processors,
        # Filtering level: logs below log_level are ignored
        wrapper_class=structlog.make_filtering_bound_logger(log_level),
        # context_class: dict is enough for most cases
        context_class=dict,
        # Logger factory: uses print (stdout) by default
        logger_factory=structlog.PrintLoggerFactory(),
        # Cache the configuration for performance
        cache_logger_on_first_use=True,
    )

# Call at the start of the app:
# configure_structlog()

Custom processors for AI apps

# src/logging_config.py (continued)
import hashlib
from typing import MutableMapping

def add_app_metadata(
    logger: Any,
    method_name: str,
    event_dict: MutableMapping[str, Any]
) -> MutableMapping[str, Any]:
    """
    Adds app metadata to all logs.
    Useful for identifying the app version in aggregation systems.
    """
    event_dict["app"] = os.getenv("APP_NAME", "production-best-practices")
    event_dict["version"] = os.getenv("APP_VERSION", "0.1.0")
    event_dict["env"] = os.getenv("ENVIRONMENT", "development")
    return event_dict

def sanitize_sensitive_fields(
    logger: Any,
    method_name: str,
    event_dict: MutableMapping[str, Any]
) -> MutableMapping[str, Any]:
    """
    Removes or masks fields that shouldn't appear in logs.
    A safety net to prevent PII or secrets from reaching the logs.
    """
    FIELDS_TO_REMOVE = {"password", "api_key", "secret", "token", "authorization"}
    FIELDS_TO_HASH = {"user_id", "email", "phone"}
    
    for field in list(event_dict.keys()):
        field_lower = field.lower()
        
        if any(sensitive in field_lower for sensitive in FIELDS_TO_REMOVE):
            event_dict[field] = "[REDACTED]"
        
        elif any(pii in field_lower for pii in FIELDS_TO_HASH):
            # Hash for analytics without exposing the real value
            value = str(event_dict[field])
            event_dict[field] = hashlib.sha256(value.encode()).hexdigest()[:12]
    
    return event_dict

def truncate_long_strings(
    logger: Any,
    method_name: str,
    event_dict: MutableMapping[str, Any],
    max_length: int = 500
) -> MutableMapping[str, Any]:
    """
    Truncates long strings to avoid enormous logs.
    In INFO, no string should be longer than 500 chars.
    """
    for key, value in event_dict.items():
        if isinstance(value, str) and len(value) > max_length:
            event_dict[key] = value[:max_length] + f"...[truncated {len(value)} chars]"
    return event_dict

# Configuration with custom processors:
def configure_structlog_with_custom_processors():
    structlog.configure(
        processors=[
            structlog.contextvars.merge_contextvars,
            add_app_metadata,           # Custom: app metadata
            sanitize_sensitive_fields,  # Custom: remove secrets
            truncate_long_strings,      # Custom: truncate long fields
            structlog.processors.add_log_level,
            structlog.processors.TimeStamper(fmt="iso"),
            structlog.stdlib.add_logger_name,
            structlog.processors.StackInfoRenderer(),
            structlog.processors.format_exc_info,
            structlog.processors.JSONRenderer()
        ]
    )

Logging exceptions correctly

import structlog
log = structlog.get_logger()

# ─── Incorrect way ───────────────────────────────────────────────────────────
try:
    response = client.chat.completions.create(...)
except Exception as e:
    log.error("failed", error=str(e))  # Only the message, no stack trace

# ─── Correct way ─────────────────────────────────────────────────────────────

# Option 1: exc_info=True (includes the full stack trace in the log)
try:
    response = client.chat.completions.create(...)
except Exception as e:
    log.error(
        "llm_request_failed",
        error_type=type(e).__name__,
        exc_info=True  # ← structlog includes the traceback in the log
    )

# Option 2: log.exception (shorthand for log.error with exc_info=True)
try:
    response = client.chat.completions.create(...)
except Exception as e:
    log.exception("llm_request_failed", error_type=type(e).__name__)

# With JSONRenderer, the output will be:
# {
#   "event": "llm_request_failed",
#   "error_type": "RateLimitError",
#   "exception": "Traceback (most recent call last):\n  ...",
#   "level": "error",
#   "timestamp": "2024-01-15T10:30:00Z"
# }

Advanced querying with jq

# Install jq:
# macOS: brew install jq
# Ubuntu: sudo apt install jq

# ─── Basic queries ───────────────────────────────────────────────────────────

# See all unique events
jq -r '.event' logs.json | sort | uniq -c | sort -rn

# Filter by level
jq 'select(.level == "error")' logs.json

# Filter by event
jq 'select(.event == "llm_request_completed")' logs.json

# ─── Cost queries ───────────────────────────────────────────────────────────

# Total cost
jq -s '[.[].total_cost_usd // 0] | add' logs.json

# Most expensive requests (top 10)
jq -s 'map(select(.total_cost_usd != null)) | sort_by(-.total_cost_usd) | .[0:10] | .[] | {request_id, total_cost_usd, total_tokens, model}' logs.json

# Average cost by model
jq -s '
  map(select(.event == "llm_request_completed"))
  | group_by(.model)
  | map({
      model: .[0].model,
      avg_cost: ([.[].total_cost_usd] | add / length),
      count: length
    })
' logs.json

# ─── Performance queries ────────────────────────────────────────────────────

# Slow requests (>5 seconds)
jq 'select(.duration_ms > 5000)' logs.json | jq -r '.request_id'

# Average latency
jq -s '[map(select(.duration_ms != null)) | .[].duration_ms] | add / length' logs.json

# ─── Guardrail queries ──────────────────────────────────────────────────────

# How many guardrails activated
jq 'select(.event == "guardrail_activated")' logs.json | wc -l

# By guardrail type
jq 'select(.event == "guardrail_activated") | .guardrail_type' logs.json | sort | uniq -c

# ─── Error queries ──────────────────────────────────────────────────────────

# All errors with their request_id
jq 'select(.level == "error") | {timestamp, request_id, event, error_type}' logs.json

# ─── Time-based queries ─────────────────────────────────────────────────────

# Logs from the last hour (timestamp is ISO 8601)
jq 'select(.timestamp > "2024-01-15T09:00:00Z")' logs.json

# ─── Tracing: the story of a request ────────────────────────────────────────
jq 'select(.request_id == "a1b2c3d4")' logs.json | jq -s 'sort_by(.timestamp) | .[]'

Python script for log analysis

# scripts/log_analyzer.py
# Alternative to jq when you need more complex logic

import json
import sys
from pathlib import Path
from collections import defaultdict

def read_jsonl(path: str) -> list:
    """Reads a JSON Lines file."""
    logs = []
    with open(path) as f:
        for line in f:
            line = line.strip()
            if line:
                try:
                    logs.append(json.loads(line))
                except json.JSONDecodeError:
                    pass
    return logs

def print_report(logs: list):
    """Generates a readable report from the logs."""
    
    # Separate by log type
    llm_logs = [l for l in logs if l.get("event") == "llm_request_completed"]
    error_logs = [l for l in logs if l.get("level") == "error"]
    guardrail_logs = [l for l in logs if l.get("event") == "guardrail_activated"]
    
    print("=" * 60)
    print("AI SYSTEM LOG ANALYSIS REPORT")
    print("=" * 60)
    
    # Summary
    print(f"\n📊 SUMMARY")
    print(f"  Total LLM requests:  {len(llm_logs)}")
    print(f"  Total errors:        {len(error_logs)}")
    print(f"  Guardrail activations: {len(guardrail_logs)}")
    
    # Cost summary
    if llm_logs:
        total_cost = sum(l.get("total_cost_usd", 0) for l in llm_logs)
        avg_cost = total_cost / len(llm_logs)
        max_cost_log = max(llm_logs, key=lambda l: l.get("total_cost_usd", 0))
        
        print(f"\n💰 COST SUMMARY")
        print(f"  Total cost:          ${total_cost:.6f}")
        print(f"  Avg cost/request:    ${avg_cost:.8f}")
        print(f"  Most expensive:      ${max_cost_log.get('total_cost_usd', 0):.6f}")
        print(f"    → request_id: {max_cost_log.get('request_id', 'unknown')}")
        print(f"    → tokens: {max_cost_log.get('total_tokens', 0)}")
        print(f"    → model: {max_cost_log.get('model', 'unknown')}")
    
    # Performance summary
    if llm_logs:
        durations = [l.get("duration_ms", 0) for l in llm_logs]
        avg_duration = sum(durations) / len(durations)
        slow_requests = [l for l in llm_logs if l.get("duration_ms", 0) > 5000]
        
        print(f"\n⚡ PERFORMANCE")
        print(f"  Avg duration:        {avg_duration:.0f}ms")
        print(f"  Slow (>5s) requests: {len(slow_requests)}")
    
    # Errors
    if error_logs:
        print(f"\n❌ ERRORS ({len(error_logs)} total)")
        error_types = defaultdict(int)
        for l in error_logs:
            error_types[l.get("error_type", "unknown")] += 1
        for error_type, count in sorted(error_types.items(), key=lambda x: -x[1]):
            print(f"  {error_type}: {count}")
    
    # Guardrails
    if guardrail_logs:
        print(f"\n🛡️  GUARDRAILS ({len(guardrail_logs)} activations)")
        guard_types = defaultdict(int)
        for l in guardrail_logs:
            guard_types[l.get("guardrail_type", "unknown")] += 1
        for guard_type, count in sorted(guard_types.items(), key=lambda x: -x[1]):
            print(f"  {guard_type}: {count}")
    
    print("\n" + "=" * 60)

if __name__ == "__main__":
    log_file = sys.argv[1] if len(sys.argv) > 1 else "logs/app.json"
    logs = read_jsonl(log_file)
    print_report(logs)

Log rotation configuration

# To prevent log files from growing indefinitely:

import logging
from logging.handlers import TimedRotatingFileHandler
import structlog

def configure_file_logging(log_dir: str = "logs"):
    """
    Configures logging to a file with daily rotation.
    Retains logs from the last 30 days.
    """
    import os
    os.makedirs(log_dir, exist_ok=True)
    
    # Handler for a file with daily rotation
    file_handler = TimedRotatingFileHandler(
        filename=f"{log_dir}/app.json",
        when="midnight",        # Rotate at midnight
        interval=1,             # Every 1 day
        backupCount=30,         # Retain 30 days
        encoding="utf-8"
    )
    file_handler.setLevel(logging.INFO)
    
    # Also log to stdout (for systems like Docker/Kubernetes)
    # that collect logs from the container's stdout
    stdout_handler = logging.StreamHandler(sys.stdout)
    stdout_handler.setLevel(logging.INFO)
    
    logging.basicConfig(
        handlers=[file_handler, stdout_handler],
        format="%(message)s",
        level=logging.INFO
    )
    
    structlog.configure(
        processors=[
            structlog.contextvars.merge_contextvars,
            structlog.processors.add_log_level,
            structlog.processors.TimeStamper(fmt="iso"),
            structlog.processors.JSONRenderer()
        ],
        logger_factory=structlog.stdlib.LoggerFactory(),
    )

Testing the logging system

# tests/unit/test_json_logs.py
import json
import io
import pytest
import structlog

@pytest.fixture
def json_log_capture():
    """Captures logs as JSON for integrity tests."""
    output = io.StringIO()
    
    structlog.configure(
        processors=[
            structlog.contextvars.merge_contextvars,
            structlog.processors.add_log_level,
            structlog.processors.TimeStamper(fmt="iso"),
            structlog.processors.JSONRenderer()
        ],
        logger_factory=structlog.PrintLoggerFactory(file=output),
        cache_logger_on_first_use=False,  # Important: don't cache in tests
    )
    yield output
    structlog.reset_defaults()

def parse_logs(output: io.StringIO) -> list:
    """Parses JSON Lines logs into a list of dicts."""
    return [
        json.loads(line)
        for line in output.getvalue().strip().split("\n")
        if line.strip()
    ]

def test_each_log_is_valid_json(json_log_capture):
    """Each log must be valid JSON."""
    log = structlog.get_logger()
    log.info("event_1", field_a="value_a")
    log.warning("event_2", field_b=42)
    log.error("event_3", field_c=True)
    
    # Must not raise JSONDecodeError
    logs = parse_logs(json_log_capture)
    assert len(logs) == 3

def test_log_has_required_fields(json_log_capture):
    """Each log must have event, level, and timestamp."""
    log = structlog.get_logger()
    log.info("test_event", some_field="some_value")
    
    logs = parse_logs(json_log_capture)
    assert len(logs) == 1
    
    entry = logs[0]
    assert "event" in entry
    assert "level" in entry
    assert "timestamp" in entry

def test_context_vars_appear_in_all_logs(json_log_capture):
    """The context fields must appear in all logs."""
    structlog.contextvars.bind_contextvars(request_id="test-req-99")
    
    log = structlog.get_logger()
    log.info("event_a")
    log.info("event_b")
    log.warning("event_c")
    
    structlog.contextvars.clear_contextvars()
    
    logs = parse_logs(json_log_capture)
    for entry in logs:
        assert entry.get("request_id") == "test-req-99", \
            f"request_id missing from: {entry}"

def test_exception_logged_as_json(json_log_capture):
    """Exceptions must appear as a JSON field, not as loose text."""
    log = structlog.get_logger()
    
    try:
        raise ValueError("test error")
    except ValueError:
        log.exception("something_failed")
    
    logs = parse_logs(json_log_capture)
    assert len(logs) == 1
    entry = logs[0]
    
    # The exception must be a field in the JSON
    assert "exception" in entry or "exc_info" in entry
    # It must not appear in the event field
    assert "Traceback" not in entry.get("event", "")

Exercises

Exercise 1: Configure structlog for production

Write the complete structlog configuration for a production app that:

  1. Uses JSON Lines
  2. Includes the request_id from the context
  3. Adds level and timestamp
  4. Redacts fields with "password" or "secret" in the name
See solution 
import structlog, os, hashlib

def redact_secrets(logger, method_name, event_dict):
    for key in list(event_dict.keys()):
        if any(s in key.lower() for s in ["password", "secret", "api_key"]):
            event_dict[key] = "[REDACTED]"
    return event_dict

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        redact_secrets,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.format_exc_info,
        structlog.processors.JSONRenderer()
    ]
)

Exercise 2: jq one-liners

Write the jq commands to:

  1. Count how many requests completed successfully today
  2. List the request_ids of all errors
  3. Find the most used model
See solution 
# 1. Requests completed today (assuming date 2024-01-15)
jq 'select(.event == "request_completed" and (.timestamp | startswith("2024-01-15")))' logs.json | wc -l

# 2. request_ids of errors
jq -r 'select(.level == "error") | .request_id' logs.json

# 3. Most used model
jq -r 'select(.event == "llm_request_completed") | .model' logs.json | sort | uniq -c | sort -rn | head -1

Summary

  • JSON Lines: one line per event, append-only, stream-friendly — the standard format
  • structlog + JSONRenderer for production, ConsoleRenderer for development
  • Processors in order: contextvars → add_log_level → timestamp → (custom) → JSONRenderer
  • Custom processors: sanitize secrets, truncate long strings, add app metadata
  • jq is enough for ad-hoc queries; a Python script for recurring analyses
  • Log rotation to avoid unlimited files — 30 days of retention as a starting point

Additional resources

  1. structlog Documentation — Complete documentation with examples
  2. jq Manual — Complete jq reference
  3. JSON Lines format — The format specification
  4. Loki — Log aggregation system that indexes JSON logs natively
  5. Structured Logging in Python (Real Python) — Complete tutorial
← Previous4. Token and Cost TrackingNext →6. Debugging Non-Deterministic Systems