Log Aggregation Best Practices: Structure, Ship, and Survive at Scale

import json
import logging
import time
import uuid
from datetime import datetime, timezone

# ─── Custom JSON formatter ────────────────────────────────────────────────────
# This replaces the default plain-text log format with a structured JSON object.
# Every log line becomes a machine-parseable record with consistent field names.
class JsonFormatter(logging.Formatter):
    def __init__(self, service_name: str):
        super().__init__()
        self.service_name = service_name

    def format(self, record: logging.LogRecord) -> str:
        # Build the base log payload — these fields appear on EVERY log line.
        # This is your organisation-wide schema. Everyone uses the same keys.
        log_payload = {
            "timestamp": datetime.now(timezone.utc).isoformat(),  # Always UTC — never local time
            "level": record.levelname.lower(),                     # 'info', 'error', 'warning'
            "service": self.service_name,                          # Which service produced this
            "message": record.getMessage(),                        # The human-readable description
            "logger": record.name,                                 # Python logger name (e.g. 'payments.processor')
            "line": f"{record.filename}:{record.lineno}",          # Exact source location for fast debugging
        }

        # Merge any extra context fields the caller passed in.
        # This is where structured data lives: user_id, order_id, duration_ms, etc.
        if hasattr(record, 'extra_fields'):
            log_payload.update(record.extra_fields)

        # If an exception is attached, include the full traceback as a string field
        # rather than letting it spill onto multiple lines and break parsing.
        if record.exc_info:
            log_payload['exception'] = self.formatException(record.exc_info)

        return json.dumps(log_payload)  # One JSON object per line (NDJSON format)


# ─── Logger factory ───────────────────────────────────────────────────────────
def create_logger(service_name: str) -> logging.Logger:
    logger = logging.getLogger(service_name)
    logger.setLevel(logging.DEBUG)

    handler = logging.StreamHandler()  # In containers, always log to stdout — never to a file
    handler.setFormatter(JsonFormatter(service_name))
    logger.addHandler(handler)
    return logger


# ─── Context-aware logging helper ─────────────────────────────────────────────
# This wrapper adds structured fields to a log call without cluttering call sites.
class ContextLogger:
    def __init__(self, logger: logging.Logger, context: dict):
        self._logger = logger
        self._context = context  # Base context fields attached to every log from this instance

    def info(self, message: str, **extra):
        self._emit(logging.INFO, message, extra)

    def error(self, message: str, exc_info=False, **extra):
        self._emit(logging.ERROR, message, extra, exc_info=exc_info)

    def warning(self, message: str, **extra):
        self._emit(logging.WARNING, message, extra)

    def _emit(self, level: int, message: str, extra: dict, exc_info=False):
        # Merge base context with call-site extra fields.
        # Call-site values win on conflict — more specific context overrides general.
        merged = {**self._context, **extra}
        record = self._logger.makeRecord(
            self._logger.name, level, "", 0, message, [], None
        )
        record.extra_fields = merged
        if exc_info:
            import sys
            record.exc_info = sys.exc_info()
        self._logger.handle(record)


# ─── Simulated payment processing function ────────────────────────────────────
def process_payment(order_id: str, user_id: int, amount_cents: int):
    base_logger = create_logger("payments-service")

    # Attach a trace_id at the start of the request.
    # This single ID lets you find every log line for this one transaction
    # across ALL services — frontend, auth, payments, database — in one query.
    trace_id = str(uuid.uuid4())
    log = ContextLogger(base_logger, {
        "trace_id": trace_id,
        "order_id": order_id,
        "user_id": user_id,
    })

    log.info("payment_processing_started", amount_cents=amount_cents)

    start_time = time.monotonic()

    try:
        # Simulate payment gateway call
        time.sleep(0.042)  # Pretend network latency
        if amount_cents > 100_000:
            raise ValueError("Amount exceeds single-transaction limit")

        duration_ms = round((time.monotonic() - start_time) * 1000, 2)

        # duration_ms is logged as a NUMBER, not a string.
        # This matters — your aggregator can now compute p99 latency with a simple query.
        log.info("payment_succeeded",
                 amount_cents=amount_cents,
                 duration_ms=duration_ms,
                 gateway="stripe")

    except ValueError as exc:
        duration_ms = round((time.monotonic() - start_time) * 1000, 2)
        log.error("payment_failed",
                  exc_info=True,
                  amount_cents=amount_cents,
                  duration_ms=duration_ms,
                  failure_reason="limit_exceeded")
        raise


# ─── Run the simulation ───────────────────────────────────────────────────────
if __name__ == "__main__":
    print("--- Successful payment ---")
    process_payment(order_id="ORD-9921", user_id=4821, amount_cents=4999)

    print("\n--- Failed payment ---")
    try:
        process_payment(order_id="ORD-9922", user_id=4821, amount_cents=150_000)
    except ValueError:
        pass  # Error is logged inside process_payment; we just suppress the re-raise here

# Fluent Bit configuration for a Kubernetes sidecar log shipping agent.
# This config implements the reliable pipeline pattern:
#   Container stdout → Fluent Bit tail → Disk buffer → Loki (with retry)
#
# Deploy this as a DaemonSet so every node has exactly one agent.
# The agent tails /var/log/containers/* which is where Kubernetes writes
# all container stdout/stderr on the host.

[SERVICE]
    # How often Fluent Bit flushes buffered records to the output plugin (seconds).
    # Lower = lower latency, higher CPU. 5s is a sensible production default.
    Flush         5

    # Grace period on shutdown — gives the agent time to flush pending messages
    # before the process exits. Critical for log completeness during rolling deploys.
    Grace         30

    # Enable the built-in HTTP server for health checks and metrics.
    # Your orchestrator should probe /api/v1/health before marking the pod ready.
    HTTP_Server   On
    HTTP_Listen   0.0.0.0
    HTTP_Port     2020

    # Storage type 'filesystem' means Fluent Bit buffers to disk, not just RAM.
    # If the output plugin (Loki) is unreachable, messages queue on disk instead
    # of being dropped. This is your primary reliability mechanism.
    storage.type  filesystem
    storage.path  /var/log/flb-storage/
    storage.sync  normal

    # Maximum disk space the buffer can consume. Size this generously —
    # you'd rather fill disk temporarily than lose log data during an incident.
    storage.max_chunks_up  128

# ─── INPUT: Tail all Kubernetes container log files ───────────────────────────
[INPUT]
    Name              tail
    Tag               kube.*

    # This glob matches every container log file Kubernetes writes on this node.
    Path              /var/log/containers/*.log

    # Kubernetes container logs are JSON lines written by the container runtime.
    # This parser extracts the actual log text and metadata from that wrapper.
    Parser            docker

    # db.lmdb tracks which byte offset has been read in each file.
    # On Fluent Bit restart, it resumes from where it left off — no duplicate
    # shipping and no gap, even across agent restarts or node reboots.
    DB                /var/log/flb_kube.db

    # Skip lines older than 5 minutes on startup.
    # Prevents flooding the aggregator with historical data after a long outage.
    Skip_Long_Lines   On
    Refresh_Interval  10

# ─── FILTER: Enrich logs with Kubernetes pod/namespace/container metadata ─────
[FILTER]
    Name                kubernetes
    Match               kube.*

    # This calls the Kubernetes API to look up pod metadata for each log line
    # and merges it in. Result: every log record now has namespace, pod_name,
    # container_name, and any labels/annotations you want to query on.
    Kube_URL            https://kubernetes.default.svc:443
    Kube_CA_File        /var/run/secrets/kubernetes.io/serviceaccount/ca.crt
    Kube_Token_File     /var/run/secrets/kubernetes.io/serviceaccount/token
    Merge_Log           On   # Merge the nested JSON log string into the top-level record
    Keep_Log            Off  # Drop the raw 'log' string field after merging to save space
    K8S-Logging.Parser  On   # Honour pod annotation 'fluentbit.io/parser' for custom parsers

# ─── FILTER: Drop noisy health-check logs before they hit storage ──────────────
[FILTER]
    Name   grep
    Match  kube.*
    # Kubernetes liveness probes generate hundreds of GET /healthz lines per hour.
    # They are never useful for debugging. Drop them here, before the buffer,
    # to save storage costs and reduce query noise. Adjust the regex to your paths.
    Exclude  log  GET /healthz|GET /readyz|GET /metrics

# ─── OUTPUT: Ship to Grafana Loki with disk-backed retry ──────────────────────
[OUTPUT]
    Name            loki
    Match           kube.*
    Host            loki.monitoring.svc.cluster.local
    Port            3100

    # Labels become the high-cardinality index in Loki — choose wisely.
    # namespace and app give you fast per-service filtering.
    # Do NOT use pod_name or container_name as labels — too high cardinality,
    # will degrade Loki performance severely (see Gotchas section).
    Labels          job=kubernetes, namespace=$kubernetes['namespace_name'], app=$kubernetes['labels']['app']

    # Loki expects log lines as strings. This serialises the full enriched record
    # back to JSON, so structured fields are preserved inside the log line value.
    Line_Format     json

    # Retry failed sends up to 5 times with exponential backoff.
    # Combined with the filesystem storage above, this means messages survive
    # a Loki restart of up to (buffer_size / throughput) duration.
    Retry_Limit     5

# Loki alerting rules using LogQL — Loki's query language.
# These rules run continuously inside Loki's ruler component and fire
# alerts into your Alertmanager when conditions are met.
#
# The philosophy here is: alert on RATES and ABSENCE, not individual errors.
# One error log is noise. 50 error logs per minute is an incident.

groups:
  - name: payments-service-alerts
    interval: 1m  # Evaluate every rule in this group every 60 seconds
    rules:

      # ── Rule 1: Error RATE alert ─────────────────────────────────────────────
      # This fires when the payments service logs more than 10 errors per minute.
      # Using rate() means the alert auto-scales — a brief spike doesn't wake
      # someone up, but a sustained error rate does.
      - alert: PaymentsServiceHighErrorRate
        # LogQL explanation:
        #   {app="payments-service"}  — select only logs from this app
        #   |= `"level":"error"`      — filter to lines containing this string
        #   [1m]                      — sliding 1-minute window
        #   > 10                      — fire if rate exceeds 10 per minute
        expr: |
          sum(rate({app="payments-service", namespace="production"} |= `"level":"error"` [1m])) > 10
        for: 2m  # Must be true for 2 consecutive minutes before firing (reduces flapping)
        labels:
          severity: critical
          team: payments
        annotations:
          summary: "Payments service error rate is {{ $value | humanize }} errors/min"
          # The runbook URL is MANDATORY — on-call engineers should never need to
          # think about what to do when an alert fires. The runbook tells them.
          runbook_url: "https://wiki.internal/runbooks/payments-high-error-rate"
          # This LogQL link drops the on-call engineer directly into the relevant
          # logs in Grafana — zero time wasted navigating dashboards.
          grafana_explore_url: "https://grafana.internal/explore?orgId=1&left=[...]"

      # ── Rule 2: Absence of expected events alert ──────────────────────────────
      # This is the alert most teams forget to write — and it catches the scariest
      # failure mode: the service is running, returning 200s, but silently not
      # processing any actual business events (e.g. stuck queue consumer).
      - alert: PaymentsNoSuccessfulTransactions
        # If there are zero payment_succeeded log events in the last 5 minutes
        # during business hours, something is very wrong.
        expr: |
          sum(rate({app="payments-service", namespace="production"} |= `"payment_succeeded"` [5m])) == 0
        for: 5m
        labels:
          severity: critical
          team: payments
        annotations:
          summary: "No successful payments processed in the last 5 minutes"
          runbook_url: "https://wiki.internal/runbooks/payments-no-transactions"

      # ── Rule 3: Spike in a specific error type ────────────────────────────────
      # Alert when a specific failure_reason appears more than 5 times per minute.
      # This catches targeted issues (e.g. card network outage) before they become
      # the dominant error type in the broader error-rate alert above.
      - alert: PaymentsGatewayTimeouts
        expr: |
          sum(
            rate(
              {app="payments-service", namespace="production"}
                | json                            # Parse the JSON log line into fields
                | failure_reason = "gateway_timeout"  # Filter on the structured field
              [2m]
            )
          ) > 5
        for: 3m
        labels:
          severity: warning
          team: payments
        annotations:
          summary: "Gateway timeout rate: {{ $value | humanize }}/min — possible Stripe outage"
          runbook_url: "https://wiki.internal/runbooks/payments-gateway-timeout"