Centralized Logging: ELK Stack, Loki, and Best Practices

# docker-compose.elk.yml
services:
  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:8.11.0
    environment:
      - discovery.type=single-node
      - ES_JAVA_OPTS=-Xms1g -Xmx1g
      - xpack.security.enabled=false
    ports:
      - "9200:9200"
    volumes:
      - esdata:/usr/share/elasticsearch/data

  logstash:
    image: docker.elastic.co/logstash/logstash:8.11.0
    volumes:
      - ./logstash.conf:/usr/share/logstash/pipeline/logstash.conf
    ports:
      - "5000:5000"

  kibana:
    image: docker.elastic.co/kibana/kibana:8.11.0
    ports:
      - "5601:5601"
    environment:
      - ELASTICSEARCH_HOSTS=http://elasticsearch:9200

volumes:
  esdata:

# logstash.conf
input {
  beats {
    port => 5000
  }
}

filter {
  grok {
    match => { "message" => "%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:level} %{GREEDYDATA:message}" }
  }

  if [level] == "ERROR" {
    mutate { add_tag => ["error"] }
  }
}

output {
  elasticsearch {
    hosts => ["elasticsearch:9200"]
    index => "logs-%{+YYYY.MM.dd}"
  }
}

# filebeat.yml
filebeat.inputs:
  - type: log
    enabled: true
    paths:
      - /var/log/*.log
      - /var/log/nginx/access.log

output.logstash:
  hosts: ["logstash:5000"]

# Optional: output directly to Elasticsearch
output.elasticsearch:
  hosts: ["http://elasticsearch:9200"]

# Start Filebeat with custom config
filebeat -e -c filebeat.yml

# Output:
# 2024/01/15 10:30:00.123456 beat is running
# 2024/01/15 10:30:01.234567 Harvester started for file: /var/log/nginx/access.log
# 2024/01/15 10:30:01.345678 Events sent: 47

# docker-compose.loki.yml
services:
  loki:
    image: grafana/loki:2.9.0
    ports:
      - "3100:3100"
    command: -config.file=/etc/loki/local-config.yaml

  promtail:
    image: grafana/promtail:2.9.0
    volumes:
      - /var/log:/var/log
      - ./promtail.yml:/etc/promtail/config.yml
    command: -config.file=/etc/promtail/config.yml

  grafana:
    image: grafana/grafana:latest
    environment:
      - GF_AUTH_ANONYMOUS_ENABLED=true
    ports:
      - "3000:3000"

# promtail.yml
scrape_configs:
  - job_name: nginx
    static_configs:
      - targets: [localhost]
        labels:
          job: nginx
          __path__: /var/log/nginx/*.log

  - job_name: system
    static_configs:
      - targets: [localhost]
        labels:
          job: varlogs
          __path__: /var/log/*.log

{
  "timestamp": "2024-06-20T10:30:00Z",
  "level": "ERROR",
  "service": "user-service",
  "request_id": "req-abc123",
  "message": "Database connection timeout",
  "duration_ms": 5234,
  "user_id": "user-456",
  "stack_trace": "TimeoutError: ..."
}

# structured_logger.py
import logging
import json

class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_entry = {
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
        }
        if hasattr(record, 'extra_data'):
            log_entry.update(record.extra_data)
        return json.dumps(log_entry)

logger = logging.getLogger("myapp")
handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter())
logger.addHandler(handler)
logger.setLevel(logging.INFO)

logger.info("User login", extra={"extra_data": {"user_id": "abc123", "ip": "10.0.0.1"}})

# Output:
# {"timestamp": "2024-06-20 10:30:00,000", "level": "INFO", "logger": "myapp", "message": "User login", "user_id": "abc123", "ip": "10.0.0.1"}

# log_cost_calculator.py
def estimate_log_cost(daily_gb, retention_days, provider="elastic"):
    rates = {
        "elastic": {"storage_gb": 0.15, "ingest_gb": 0.10},
        "loki": {"storage_gb": 0.04, "ingest_gb": 0.03},
        "datadog": {"storage_gb": None, "ingest_gb": 0.10},
    }
    rate = rates[provider]
    total_storage = daily_gb * retention_days
    if rate["storage_gb"]:
        storage_cost = total_storage * rate["storage_gb"]
    else:
        storage_cost = "included"
    ingest_cost = daily_gb * 30 * rate["ingest_gb"]
    return {
        "provider": provider,
        "daily_ingest_gb": daily_gb,
        "retention_days": retention_days,
        "monthly_ingest_cost": round(ingest_cost, 2),
        "total_storage_cost": round(storage_cost, 2) if isinstance(storage_cost, float) else storage_cost,
    }

print("=== Log Cost Comparison ===")
for provider in ["elastic", "loki", "datadog"]:
    cost = estimate_log_cost(100, 30, provider)
    print(f"  {provider:<10} Monthly ingest: ${cost['monthly_ingest_cost']:<8}")

# json_logger.py
import json
import sys
from datetime import datetime

class JsonLogger:
    def __init__(self, service_name, level="INFO"):
        self.service = service_name
        self.level = level
        self.levels = {"DEBUG": 0, "INFO": 1, "WARN": 2, "ERROR": 3}

    def log(self, level, message, **extra):
        if self.levels.get(level, 0) < self.levels.get(self.level, 1):
            return
        entry = {
            "timestamp": datetime.utcnow().isoformat() + "Z",
            "level": level,
            "service": self.service,
            "message": message,
        }
        entry.update(extra)
        print(json.dumps(entry))

logger = JsonLogger("my-api")
logger.log("INFO", "Server started", port=8080, env="production")
logger.log("ERROR", "Connection refused", host="db.internal", retry=3)

# Output:
# {"timestamp": "2024-06-20T10:30:00.000000Z", "level": "INFO", "service": "my-api", "message": "Server started", "port": 8080, "env": "production"}
# {"timestamp": "2024-06-20T10:30:01.000000Z", "level": "ERROR", "service": "my-api", "message": "Connection refused", "host": "db.internal", "retry": 3}