elk-stack

elk-stack Best Practices

The ELK (Elasticsearch-Logstash-Kibana) stack is our standard for centralized logging and observability. Adhering to these guidelines ensures our logs are consistent, actionable, and efficient, enabling rapid troubleshooting and deep insights.

Code Organization and Structure

1. Standardize on Structured Logging

Always emit logs as structured JSON to stdout. This is the only acceptable method for application logging. Avoid writing to local files.

Rationale: stdout is the standard stream for containerized applications, easily captured by Elastic Agent or Filebeat. Structured JSON ensures logs are machine-readable and parsable without complex regex, making them immediately queryable in Elasticsearch.

✅ GOOD: Python with structlog

# app/logging_config.py
import sys
import structlog
import os

def configure_logging():
    # Define canonical fields and processors
    shared_processors = [
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
        structlog.processors.merge_extra_context,
        # Enforce canonical fields: service_name, trace_id, span_id
        lambda logger, method_name, event_dict: event_dict.update(
            service_name=os.getenv("SERVICE_NAME", "unknown-service"),
            trace_id=os.getenv("X_B3_TRACEID", "no-trace-id"), # Example for B3 propagation
            span_id=os.getenv("X_B3_SPANID", "no-span-id"),
        ),
    ]

    if os.getenv("APP_ENV", "development") == "production":
        # Production: JSON output for log aggregators
        processors = shared_processors + [
            structlog.processors.dict_tracebacks, # Structured tracebacks
            structlog.processors.JSONRenderer(),
        ]
    else:
        # Development: Pretty printing for local readability
        processors = shared_processors + [
            structlog.dev.ConsoleRenderer(),
        ]

    structlog.configure(
        processors=processors,
        logger_factory=structlog.stdlib.LoggerFactory(),
        wrapper_class=structlog.stdlib.BoundLogger,
        cache_logger_on_first_use=True,
    )

    # Optionally, redirect standard Python logging to structlog
    # import logging
    # logging.basicConfig(handlers=[structlog.stdlib.ProcessorFormatter.wrap_for_formatter], level=os.getenv("LOG_LEVEL", "INFO").upper())
    # structlog.stdlib.ProcessorFormatter.remove_processors_from_logger(logging.getLogger())

# In your application entry point:
# from app.logging_config import configure_logging
# configure_logging()
# log = structlog.get_logger(__name__)