ELK Stack — Elasticsearch, Logstash, and Kibana Centralized Logging Guide

# docker-compose.yml
version: '3.8'
 
services:
  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:8.11.0
    container_name: elasticsearch
    environment:
      - discovery.type=single-node
      - ES_JAVA_OPTS=-Xms2g -Xmx2g
      - xpack.security.enabled=true
      - ELASTIC_PASSWORD=change-me-secure-password
    ports:
      - "9200:9200"
    volumes:
      - es_data:/usr/share/elasticsearch/data
    ulimits:
      memlock:
        soft: -1
        hard: -1
    deploy:
      resources:
        limits:
          memory: 4g
    healthcheck:
      test: ["CMD-SHELL", "curl -s -u elastic:change-me-secure-password http://localhost:9200/_cluster/health | grep -v red"]
      interval: 30s
 
  logstash:
    image: docker.elastic.co/logstash/logstash:8.11.0
    container_name: logstash
    ports:
      - "5044:5044"  # Beats input
      - "5000:5000"  # TCP input
    volumes:
      - ./logstash/pipeline:/usr/share/logstash/pipeline
      - ./logstash/config/logstash.yml:/usr/share/logstash/config/logstash.yml
    environment:
      - LS_JAVA_OPTS=-Xms1g -Xmx1g
    depends_on:
      - elasticsearch
 
  kibana:
    image: docker.elastic.co/kibana/kibana:8.11.0
    container_name: kibana
    ports:
      - "5601:5601"
    environment:
      - ELASTICSEARCH_HOSTS=http://elasticsearch:9200
      - ELASTICSEARCH_USERNAME=kibana_system
      - ELASTICSEARCH_PASSWORD=change-me-secure-password
    depends_on:
      - elasticsearch
 
volumes:
  es_data:

# logstash/pipeline/main.conf
 
input {
  # Receive logs from Filebeat
  beats {
    port => 5044
  }
 
  # Receive JSON logs via TCP
  tcp {
    port => 5000
    codec => json_lines
  }
}
 
filter {
  # Parse JSON application logs
  if [fields][log_type] == "application" {
    json {
      source => "message"
      target => "app"
    }
 
    # Extract structured fields
    mutate {
      rename => { "[app][level]" => "log_level" }
      rename => { "[app][service]" => "service_name" }
      rename => { "[app][requestId]" => "trace_id" }
    }
  }
 
  # Parse nginx access logs
  if [fields][log_type] == "nginx" {
    grok {
      match => {
        "message" => '%{IPORHOST:client_ip} - %{USER:ident} \[%{HTTPDATE:timestamp}\] "%{WORD:method} %{DATA:path} HTTP/%{NUMBER:http_version}" %{NUMBER:response_code:int} %{NUMBER:bytes:int}'
      }
    }
    date {
      match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
      target => "@timestamp"
    }
  }
 
  # Geolocation enrichment
  if [client_ip] {
    geoip {
      source => "client_ip"
      target => "geoip"
    }
  }
 
  # Drop health check logs
  if [path] == "/health" {
    drop {}
  }
 
  # Tag high-severity events
  if [response_code] >= 500 {
    mutate {
      add_tag => ["server_error"]
    }
  }
}
 
output {
  elasticsearch {
    hosts => ["elasticsearch:9200"]
    user => "elastic"
    password => "${ELASTIC_PASSWORD}"
    index => "logs-%{[fields][environment]}-%{+YYYY.MM.dd}"
    action => "create"
  }
 
  # Debug output (disable in production)
  # stdout { codec => rubydebug }
}

# filebeat.yml — runs on each application host
filebeat.inputs:
  - type: log
    enabled: true
    paths:
      - /var/log/my-app/*.log
    fields:
      log_type: application
      environment: production
      service: my-api
    fields_under_root: false
    json.keys_under_root: true
    json.add_error_key: true
    multiline.pattern: '^\d{4}-\d{2}-\d{2}'
    multiline.negate: true
    multiline.match: after
 
  - type: log
    enabled: true
    paths:
      - /var/log/nginx/access.log
    fields:
      log_type: nginx
      environment: production
 
output.logstash:
  hosts: ["logstash.example.com:5044"]
  ssl.enabled: true
 
processors:
  - add_host_metadata:
      when.not.contains.tags: forwarded
  - add_docker_metadata: ~

# Create ILM policy — hot/warm/cold/delete phases
curl -X PUT "localhost:9200/_ilm/policy/logs-policy" \
  -u elastic:password \
  -H "Content-Type: application/json" \
  -d '{
    "policy": {
      "phases": {
        "hot": {
          "min_age": "0ms",
          "actions": {
            "rollover": {
              "max_primary_shard_size": "50gb",
              "max_age": "1d"
            }
          }
        },
        "warm": {
          "min_age": "7d",
          "actions": {
            "shrink": {"number_of_shards": 1},
            "forcemerge": {"max_num_segments": 1}
          }
        },
        "cold": {
          "min_age": "30d",
          "actions": {
            "freeze": {}
          }
        },
        "delete": {
          "min_age": "90d",
          "actions": {
            "delete": {}
          }
        }
      }
    }
  }'

# Full-text search across all logs
curl -X GET "localhost:9200/logs-*/_search" \
  -H "Content-Type: application/json" \
  -d '{
    "query": {
      "bool": {
        "must": [
          {"match": {"message": "database connection failed"}},
          {"range": {"@timestamp": {"gte": "now-1h"}}}
        ],
        "filter": [
          {"term": {"service_name": "my-api"}},
          {"terms": {"log_level": ["error", "fatal"]}}
        ]
      }
    },
    "sort": [{"@timestamp": {"order": "desc"}}],
    "size": 50
  }'
 
# Aggregate error count by service
curl -X GET "localhost:9200/logs-*/_search" \
  -H "Content-Type: application/json" \
  -d '{
    "aggs": {
      "errors_by_service": {
        "terms": {"field": "service_name.keyword"},
        "aggs": {
          "error_count": {"filter": {"term": {"log_level": "error"}}}
        }
      }
    },
    "size": 0
  }'