ELK Stack日志管理实战指南：从部署到可视化全流程

version: '3.8'
services:
  # Elasticsearch：存储与检索
  elasticsearch:
    image: elasticsearch:8.13.0
    environment:
      - discovery.type=single-node  # 单节点模式（测试用，生产建议3节点）
      - ES_JAVA_OPTS=-Xms2g -Xmx2g  # JVM内存（根据服务器配置调整，至少1G）
      - xpack.security.enabled=false  # 关闭安全验证（测试用，生产要开）
    ports:
      - "9200:9200"  # ES HTTP端口
    volumes:
      - es_data:/usr/share/elasticsearch/data  # 持久化数据
  # Logstash：过滤与转发
  logstash:
    image: logstash:8.13.0
    volumes:
      - ./logstash/pipeline:/usr/share/logstash/pipeline  # 挂载pipeline配置
      - ./logstash/config:/usr/share/logstash/config      # 挂载logstash.yml
    ports:
      - "5044:5044"  # 接收Filebeat的端口
    depends_on:
      - elasticsearch
  # Kibana：可视化
  kibana:
    image: kibana:8.13.0
    ports:
      - "5601:5601"  # Kibana Web端口
    environment:
      - ELASTICSEARCH_HOSTS=http://elasticsearch:9200  # 连接ES
    depends_on:
      - elasticsearch

volumes:
  es_data:
    driver: local  # 本地卷，持久化ES数据

# Filebeat输入：采集Nginx日志
filebeat.inputs:
- type: log  # 日志类型输入
  enabled: true
  paths:
    - /var/log/nginx/access.log  # 你的Nginx访问日志路径（要确保Filebeat有权限读）
  fields:  # 自定义字段，方便后续过滤
    log_source: nginx_access
  fields_under_root: true  # 把自定义字段放到根级别（不用嵌套在fields里）

# 输出到Logstash（别直接发ES！要先清洗）
output.logstash:
  hosts: ["localhost:5044"]  # Logstash的地址+端口

# 连接Kibana（自动导入索引模式）
setup.kibana:
  host: "http://localhost:5601"

./filebeat -e -c filebeat.yml

# 输入：接收Filebeat的日志
input {
  beats {
    port => 5044
  }
}

# 过滤：只处理Nginx访问日志
filter {
  if [log_source] == "nginx_access" {  # 匹配Filebeat的自定义字段
    # 1. 用grok解析Nginx日志（预定义模式%{NGINX_ACCESS_LOG}）
    grok {
      match => { "message" => "%{NGINX_ACCESS_LOG}" }
      remove_field => ["message"]  # 解析完删除原始message字段，省空间
    }
    # 2. 校正时间戳（用日志里的时间代替Logstash接收时间）
    date {
      match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
      target => "@timestamp"  # 覆盖默认的@timestamp
    }
    # 3. 解析User-Agent（比如Chrome、iOS）
    useragent {
      source => "agent"
      target => "user_agent"
    }
    # 4. 解析IP地理信息（比如中国、北京）
    geoip {
      source => "clientip"
      target => "geoip"
    }
  }
}

# 输出：把清洗后的日志送进Elasticsearch
output {
  elasticsearch {
    hosts => ["http://elasticsearch:9200"]
    index => "nginx_access-%{+YYYY.MM.dd}"  # 按天生成索引（比如nginx_access-2024.05.10）
  }
}

PUT _index_template/nginx_access_template
{
  "index_patterns": ["nginx_access-*"],  # 匹配所有nginx_access开头的索引
  "template": {
    "settings": {
      "number_of_shards": 1,  # 分片数（单节点用1，多节点建议3）
      "number_of_replicas": 0  # 副本数（测试用0，生产用1）
    },
    "mappings": {
      "properties": {
        "clientip": { "type": "ip" },  # IP类型（支持IP范围查询，比如192.168.0.0/16）
        "response": { "type": "integer" },  # 响应状态码（整数类型，节省空间）
        "bytes": { "type": "long" },  # 发送字节数（长整型）
        "geoip": {
          "properties": {
            "country_name": { "type": "keyword" },  # 国家名称（不分词，适合聚合）
            "city_name": { "type": "keyword" }
          }
        }
      }
    }
  }
}

PUT _ilm/policy/nginx_access_policy
{
  "policy": {
    "phases": {
      "hot": {  # 热阶段：最近7天的日志（频繁查询）
        "actions": {
          "rollover": { "max_size": "50GB", "max_age": "7d" }  # 达到50GB或7天就滚动新索引
        }
      },
      "warm": {  # 温阶段：7-30天的日志（偶尔查询）
        "min_age": "7d",
        "actions": {
          "allocate": { "require": { "box_type": "warm" } }  # 移到性能较低的温节点
        }
      },
      "cold": {  # 冷阶段：30-90天的日志（很少查询）
        "min_age": "30d",
        "actions": {
          "freeze": {}  # 冻结索引（减少内存占用）
        }
      },
      "delete": {  # 删除阶段：超过90天的日志（不用再查）
        "min_age": "90d",
        "actions": { "delete": {} }
      }
    }
  }
}