Structured Logging & Log Aggregation with ELK Stack

2026-03-21 10:15:23 Error: Database connection refused

# Server 2: /var/log/app.log (you don't see this for 15 minutes)
2026-03-21 10:15:22 Error: Database connection refused

# Server 3: Combined, these tell a story, but:
# - They're on 3 different machines
# - You can't search them together
# - Container restart and logs are gone
# - You have no context (which user? which request?)

1
services:
2
  elasticsearch:
3
    image: docker.elastic.co/elasticsearch/elasticsearch:8.11.0
4
    container_name: elasticsearch
5
    environment:
6
      discovery.type: single-node
7
      xpack.security.enabled: false
8
      xpack.security.transport.ssl.enabled: false
9
    ports:
10
      - '9200:9200'
11
    volumes:
12
      - elasticsearch-data:/usr/share/elasticsearch/data
13

14
  kibana:
15
    image: docker.elastic.co/kibana/kibana:8.11.0
16
    container_name: kibana
17
    ports:
18
      - '5601:5601'
19
    environment:
20
      ELASTICSEARCH_HOSTS: http://elasticsearch:9200
21
    depends_on:
22
      - elasticsearch
23

24
  logstash:
25
    image: docker.elastic.co/logstash/logstash:8.11.0
26
    container_name: logstash
27
    volumes:
28
      - ./logstash.conf:/usr/share/logstash/pipeline/logstash.conf
29
    ports:
30
      - '5000:5000'
31
    environment:
32
      discovery.seed_hosts: elasticsearch
33
      LS_JAVA_OPTS: '-Xmx256m -Xms256m'
34
    depends_on:
35
      - elasticsearch
36

37
volumes:
38
  elasticsearch-data:

1
import json
2
import logging
3
from pythonjsonlogger import jsonlogger
4

5
# Configure JSON logging
6
logHandler = logging.StreamHandler()
7
formatter = jsonlogger.JsonFormatter()
8
logHandler.setFormatter(formatter)
9
logger = logging.getLogger()
10
logger.addHandler(logHandler)
11
logger.setLevel(logging.INFO)
12

13
# Use logging with context
14
logger.info("User login", extra={
15
    "user_id": 42,
16
    "request_id": "req-abc-123",
17
    "service": "user-api",
18
    "ip_address": "192.168.1.1"
19
})
20

21
logger.error("Database connection failed", extra={
22
    "user_id": 42,
23
    "request_id": "req-abc-123",
24
    "service": "user-api",
25
    "error_code": "DB_CONN_REFUSED",
26
    "retry_count": 3
27
})

1
import winston from 'winston';
2

3
const logger = winston.createLogger({
4
  format: winston.format.json(),
5
  defaultMeta: {service: 'api-gateway'},
6
  transports: [new winston.transports.Console()],
7
});
8

9
// Log with context
10
logger.info('User authenticated', {
11
  user_id: 42,
12
  request_id: 'req-abc-123',
13
  ip_address: '192.168.1.1',
14
});
15

16
logger.error('Database connection failed', {
17
  user_id: 42,
18
  request_id: 'req-abc-123',
19
  error_code: 'DB_CONN_REFUSED',
20
  retry_count: 3,
21
});

1
input {
2
  tcp {
3
    port => 5000
4
    codec => json
5
  }
6

7
  # Read from files
8
  file {
9
    path => "/var/log/app/*.log"
10
    codec => json
11
  }
12
}
13

14
filter {
15
  # Parse and enrich logs
16
  if [service] == "api-gateway" {
17
    mutate {
18
      add_field => { "service_tier" => "frontend" }
19
    }
20
  }
21

22
  # Extract request ID from logs for correlation
23
  grok {
24
    match => { "message" => "request_id=%{NOTSPACE:request_id}" }
25
  }
26

27
  # Add timestamp if missing
28
  date {
29
    match => [ "timestamp", "ISO8601" ]
30
    target => "@timestamp"
31
  }
32
}
33

34
output {
35
  elasticsearch {
36
    hosts => ["elasticsearch:9200"]
37
    index => "logs-%{+YYYY.MM.dd}"
38
  }
39

40
  # Also output to stdout for debugging
41
  stdout {
42
    codec => rubydebug
43
  }
44
}

1
input {
2
  tcp {
3
    port => 5000
4
    codec => json
5
  }
6
}
7

8
filter {
9
  # Normalize service names
10
  translate {
11
    field => "service"
12
    destination => "service_normalized"
13
    dictionary => {
14
      "user-api" => "user-service"
15
      "user_api" => "user-service"
16
      "users" => "user-service"
17
    }
18
  }
19

20
  # Add environment if not present
21
  if ![environment] {
22
    mutate {
23
      add_field => { "environment" => "production" }
24
    }
25
  }
26

27
  # Parse error stack traces
28
  if [level] == "ERROR" and [stack_trace] {
29
    mutate {
30
      split => { "stack_trace" => "\n" }
31
    }
32
  }
33
}
34

35
output {
36
  elasticsearch {
37
    hosts => ["elasticsearch:9200"]
38
    index => "logs-%{environment}-%{+YYYY.MM.dd}"
39
  }
40
}

1
from fastapi import Request
2
import uuid
3
import logging
4

5
logger = logging.getLogger(__name__)
6

7
async def add_request_id(request: Request, call_next):
8
    # Generate or extract request ID
9
    request_id = request.headers.get("X-Request-ID") or str(uuid.uuid4())
10

11
    # Store in request state
12
    request.state.request_id = request_id
13

14
    # Log with correlation
15
    logger.info("Request started", extra={
16
        "request_id": request_id,
17
        "method": request.method,
18
        "path": request.url.path
19
    })
20

21
    response = await call_next(request)
22

23
    # Add to response headers for client
24
    response.headers["X-Request-ID"] = request_id
25

26
    return response