ELK日志分析平台搭建实战：从日志混乱到一目了然

以前查日志：SSH登录服务器，grep、tail、awk轮番上阵，10台服务器查一圈下来半小时过去了。现在：打开Kibana，输入关键词，所有服务器的日志一秒出结果。

一、为什么需要ELK？

先说说我们之前的"原始"日志管理：

1. # 服务器1
2. ssh root@192.168.1.10
3. tail -f /var/log/app/app.log | grep "ERROR"
5. # 服务器2
6. ssh root@192.168.1.11
7. tail -f /var/log/app/app.log | grep "ERROR"
9. # 服务器3...10
10. # 开10个终端窗口...

复制代码

痛点：

• 10台服务器要开10个终端
  
• 日志量大的时候grep到眼花
  
• 想看历史日志？先下载再分析
  
• 跨服务追踪？基本不可能
  
• 老板问"昨天有多少报错"？数到明天也数不完
  

搭了ELK之后：

• 所有服务器日志集中一处
  
• 搜索秒出结果
  
• 自动生成图表
  
• 异常告警自动通知
  

二、ELK是什么？

ELK是三个开源项目的首字母：
组件作用一句话解释Elasticsearch存储+搜索日志数据库，支持全文搜索Logstash收集+处理从各处收集日志，清洗后存入ESKibana可视化Web界面，搜索和做图表现在还经常加个Filebeat，变成EFK或ELFK：

1. 应用日志 → Filebeat（轻量采集）→ Logstash（处理）→ Elasticsearch（存储）→ Kibana（展示）

复制代码

三、快速搭建（Docker Compose）

3.1 服务器要求

ELK比较吃资源，建议配置：
环境CPU内存磁盘测试2核4GB50GB生产8核+16GB+500GB+ SSD3.2 目录结构

1. mkdir -p /data/elk/{elasticsearch,logstash,kibana}
2. cd /data/elk
4. # 目录结构
5. elk/
6. ├── docker-compose.yml
7. ├── elasticsearch/
8. │   └── data/
9. ├── logstash/
10. │   ├── config/
11. │   │   └── logstash.yml
12. │   └── pipeline/
13. │       └── logstash.conf
14. └── kibana/
15.     └── config/
16.         └── kibana.yml

复制代码

3.3 docker-compose.yml

1. version: '3.8'
3. services:
4.   elasticsearch:
5.     image: docker.elastic.co/elasticsearch/elasticsearch:8.11.0
6.     container_name: elasticsearch
7.     environment:
8.       - node.name=es01
9.       - cluster.name=elk-cluster
10.       - discovery.type=single-node
11.       - bootstrap.memory_lock=true
12.       - xpack.security.enabled=false
13.       - "ES_JAVA_OPTS=-Xms2g -Xmx2g"
14.     ulimits:
15.       memlock:
16.         soft: -1
17.         hard: -1
18.     volumes:
19.       - ./elasticsearch/data:/usr/share/elasticsearch/data
20.     ports:
21.       - "9200:9200"
22.     networks:
23.       - elk
25.   logstash:
26.     image: docker.elastic.co/logstash/logstash:8.11.0
27.     container_name: logstash
28.     volumes:
29.       - ./logstash/config/logstash.yml:/usr/share/logstash/config/logstash.yml
30.       - ./logstash/pipeline:/usr/share/logstash/pipeline
31.     ports:
32.       - "5044:5044"   # Beats输入
33.       - "5000:5000"   # TCP输入
34.       - "9600:9600"   # API
35.     environment:
36.       - "LS_JAVA_OPTS=-Xms512m -Xmx512m"
37.     depends_on:
38.       - elasticsearch
39.     networks:
40.       - elk
42.   kibana:
43.     image: docker.elastic.co/kibana/kibana:8.11.0
44.     container_name: kibana
45.     volumes:
46.       - ./kibana/config/kibana.yml:/usr/share/kibana/config/kibana.yml
47.     ports:
48.       - "5601:5601"
49.     environment:
50.       - ELASTICSEARCH_HOSTS=http://elasticsearch:9200
51.     depends_on:
52.       - elasticsearch
53.     networks:
54.       - elk
56. networks:
57.   elk:
58.     driver: bridge

复制代码

3.4 配置文件

logstash/config/logstash.yml：

1. http.host: "0.0.0.0"
2. xpack.monitoring.elasticsearch.hosts: ["http://elasticsearch:9200"]

复制代码

logstash/pipeline/logstash.conf：

1. input {
2.   # 接收Filebeat发来的日志
3.   beats {
4.     port => 5044
5.   }
6.   
7.   # 也可以接收TCP发来的日志
8.   tcp {
9.     port => 5000
10.     codec => json
11.   }
12. }
14. filter {
15.   # 解析JSON格式日志
16.   if [message] =~ /^\{.*\}$/ {
17.     json {
18.       source => "message"
19.     }
20.   }
21.   
22.   # 解析时间戳
23.   date {
24.     match => ["timestamp", "yyyy-MM-dd HH:mm:ss.SSS", "ISO8601"]
25.     target => "@timestamp"
26.   }
27.   
28.   # 添加自定义字段
29.   mutate {
30.     add_field => { "env" => "production" }
31.   }
32. }
34. output {
35.   elasticsearch {
36.     hosts => ["elasticsearch:9200"]
37.     index => "app-logs-%{+YYYY.MM.dd}"
38.   }
39.   
40.   # 调试时可以输出到控制台
41.   # stdout { codec => rubydebug }
42. }

复制代码

kibana/config/kibana.yml：

1. server.host: "0.0.0.0"
2. server.name: kibana
3. elasticsearch.hosts: ["http://elasticsearch:9200"]
4. i18n.locale: "zh-CN"

复制代码

3.5 启动

1. # 设置ES数据目录权限
2. chown -R 1000:1000 elasticsearch/data
4. # 调整系统参数（ES需要）
5. sysctl -w vm.max_map_count=262144
6. echo "vm.max_map_count=262144" >> /etc/sysctl.conf
8. # 启动
9. docker-compose up -d
11. # 查看状态
12. docker-compose ps
14. # 查看日志
15. docker-compose logs -f

复制代码

3.6 验证

1. # ES健康检查
2. curl http://localhost:9200/_cluster/health?pretty
4. # Kibana访问
5. # 浏览器打开 http://你的IP:5601

复制代码

四、配置日志收集

4.1 安装Filebeat

在需要收集日志的服务器上安装：

1. # CentOS
2. rpm -ivh https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-8.11.0-x86_64.rpm
4. # Ubuntu
5. wget https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-8.11.0-amd64.deb
6. dpkg -i filebeat-8.11.0-amd64.deb
8. # Docker方式
9. docker pull docker.elastic.co/beats/filebeat:8.11.0

复制代码

4.2 配置Filebeat

/etc/filebeat/filebeat.yml：

1. filebeat.inputs:
2.   # 收集应用日志
3.   - type: log
4.     enabled: true
5.     paths:
6.       - /var/log/app/*.log
7.     fields:
8.       app: my-app
9.       env: prod
10.     fields_under_root: true
11.    
12.     # 多行日志合并（Java堆栈）
13.     multiline.pattern: '^\d{4}-\d{2}-\d{2}'
14.     multiline.negate: true
15.     multiline.match: after
16.   
17.   # 收集Nginx日志
18.   - type: log
19.     enabled: true
20.     paths:
21.       - /var/log/nginx/access.log
22.     fields:
23.       app: nginx
24.       type: access
25.     fields_under_root: true
27.   # 收集系统日志
28.   - type: log
29.     enabled: true
30.     paths:
31.       - /var/log/messages
32.       - /var/log/secure
33.     fields:
34.       app: system
35.     fields_under_root: true
37. # 输出到Logstash
38. output.logstash:
39.   hosts: ["ELK服务器IP:5044"]
41. # 如果直接输出到ES
42. # output.elasticsearch:
43. #   hosts: ["ELK服务器IP:9200"]
44. #   index: "filebeat-%{+yyyy.MM.dd}"
46. # 日志处理器
47. processors:
48.   - add_host_metadata: ~
49.   - add_cloud_metadata: ~

复制代码

4.3 启动Filebeat

1. # 测试配置
2. filebeat test config
4. # 启动
5. systemctl enable filebeat
6. systemctl start filebeat
8. # 查看状态
9. systemctl status filebeat

复制代码

五、Kibana使用指南

5.1 创建索引模式

• 打开Kibana：http://IP:5601
  
• 进入 Stack Management → Index Patterns
  
• 点击 Create index pattern
  
• 输入 app-logs-*
  
• 选择时间字段 @timestamp
  

5.2 日志搜索（Discover）

进入 Discover：

1. # 搜索包含ERROR的日志
2. message: ERROR
4. # 搜索特定应用
5. app: my-app AND level: ERROR
7. # 时间范围内的错误
8. level: ERROR AND @timestamp >= "2024-01-01"
10. # 组合查询
11. (level: ERROR OR level: WARN) AND app: my-app AND NOT message: "expected error"

复制代码

5.3 常用搜索技巧

需求查询语法包含关键词message: "OutOfMemory"排除关键词NOT message: "health check"某个字段等于level: ERROR某个字段存在_exists_: traceId正则匹配message: /.*Exception.*/范围查询response_time: [100 TO 500]通配符message: *timeout*5.4 创建可视化图表

场景：统计每小时的错误数量

• 进入 Visualize → Create visualization
  
• 选择 Lens 或 Aggregation based → Line
  
• 配置：
  
  
  
  • Y轴：Count
    
  • X轴：@timestamp（Date Histogram，间隔1小时）
    
  • 过滤：level: ERROR
    
  
  

场景：错误类型分布饼图

• 选择 Pie Chart
  
• Slice by：Terms → error_type
  
• Size by：Count
  

5.5 创建Dashboard

把多个可视化组合成仪表盘：

• 进入 Dashboard → Create dashboard
  
• 添加已创建的可视化
  
• 调整布局
  
• 保存
  

推荐的Dashboard组成：

• 日志总量趋势图
  
• 错误数量趋势图
  
• 错误类型分布
  
• 响应时间分布
  
• 各服务日志量对比
  
• 最近错误列表
  

六、Logstash高级配置

6.1 解析Nginx日志

1. input {
2.   beats {
3.     port => 5044
4.   }
5. }
7. filter {
8.   if [fields][type] == "nginx-access" {
9.     grok {
10.       match => { "message" => '%{IPORHOST:client_ip} - %{DATA:user} \[%{HTTPDATE:timestamp}\] "%{WORD:method} %{URIPATHPARAM:request} HTTP/%{NUMBER:http_version}" %{NUMBER:status} %{NUMBER:bytes} "%{DATA:referrer}" "%{DATA:user_agent}"' }
11.     }
12.    
13.     date {
14.       match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
15.       target => "@timestamp"
16.     }
17.    
18.     geoip {
19.       source => "client_ip"
20.       target => "geoip"
21.     }
22.    
23.     useragent {
24.       source => "user_agent"
25.       target => "ua"
26.     }
27.    
28.     mutate {
29.       convert => {
30.         "status" => "integer"
31.         "bytes" => "integer"
32.       }
33.       remove_field => ["message", "timestamp"]
34.     }
35.   }
36. }
38. output {
39.   if [fields][type] == "nginx-access" {
40.     elasticsearch {
41.       hosts => ["elasticsearch:9200"]
42.       index => "nginx-access-%{+YYYY.MM.dd}"
43.     }
44.   }
45. }

复制代码

6.2 解析Java日志

1. filter {
2.   if [fields][app] == "java-app" {
3.     # 解析标准格式：2024-01-01 12:00:00.123 [thread] LEVEL class - message
4.     grok {
5.       match => { "message" => "%{TIMESTAMP_ISO8601:timestamp} \[%{DATA:thread}\] %{LOGLEVEL:level} %{DATA:class} - %{GREEDYDATA:log_message}" }
6.     }
7.    
8.     date {
9.       match => ["timestamp", "yyyy-MM-dd HH:mm:ss.SSS"]
10.       target => "@timestamp"
11.     }
12.    
13.     # 提取异常类型
14.     if [level] == "ERROR" {
15.       grok {
16.         match => { "log_message" => "%{DATA:exception_class}:%{GREEDYDATA:exception_message}" }
17.         tag_on_failure => []
18.       }
19.     }
20.   }
21. }

复制代码

6.3 日志脱敏

1. filter {
2.   # 脱敏手机号
3.   mutate {
4.     gsub => [
5.       "message", "1[3-9]\d{9}", "***phone***"
6.     ]
7.   }
8.   
9.   # 脱敏身份证
10.   mutate {
11.     gsub => [
12.       "message", "\d{17}[\dXx]", "***idcard***"
13.     ]
14.   }
15.   
16.   # 脱敏银行卡
17.   mutate {
18.     gsub => [
19.       "message", "\d{16,19}", "***bankcard***"
20.     ]
21.   }
22. }

复制代码

七、告警配置

7.1 使用Elastalert

1. # 安装
2. pip install elastalert2
4. # 配置 config.yaml
5. rules_folder: /opt/elastalert/rules
6. es_host: localhost
7. es_port: 9200

复制代码

告警规则 rules/error_alert.yaml：

1. name: "应用错误告警"
2. type: frequency
3. index: app-logs-*
5. # 5分钟内超过10次ERROR就告警
6. num_events: 10
7. timeframe:
8.   minutes: 5
10. filter:
11.   - term:
12.       level: ERROR
14. alert:
15.   - dingtalk
17. dingtalk_webhook: "https://oapi.dingtalk.com/robot/send?access_token=xxx"
18. dingtalk_msgtype: "markdown"

复制代码

7.2 使用Kibana自带告警（需订阅）

进入 Stack Management → Rules and Connectors
八、异地服务器日志收集

公司有多个机房，服务器分布在不同网络，怎么把日志集中到一个ELK？
方案1：公网暴露端口（不推荐）

把Logstash的5044端口暴露到公网，安全风险大。
方案2：VPN

在各机房部署VPN客户端，打通到ELK服务器的网络。
缺点：

• 配置复杂
  
• 经常断连
  
• 影响日志实时性
  

方案3：SD-WAN组网（我在用的）

用星空组网把所有服务器组到一个虚拟网络：

1. # 在ELK服务器
2. curl -sSL https://down.starvpn.cn/linux.sh | bash
3. xkcli login your_token && xkcli up
4. # 虚拟IP: 192.168.188.10
6. # 在各机房的服务器
7. # 同样操作，获得虚拟IP

复制代码

Filebeat配置直接用虚拟IP：

1. output.logstash:
2.   hosts: ["192.168.188.10:5044"]

复制代码

效果：

• 配置简单，5分钟搞定
  
• 延迟低（P2P直连）
  
• 稳定，用了3个月没断过
  
• 安全，流量加密
  

九、性能优化

9.1 ES优化

索引生命周期管理（ILM）：

1. PUT _ilm/policy/logs-policy
2. {
3.   "policy": {
4.     "phases": {
5.       "hot": {
6.         "actions": {
7.           "rollover": {
8.             "max_size": "50GB",
9.             "max_age": "1d"
10.           }
11.         }
12.       },
13.       "warm": {
14.         "min_age": "7d",
15.         "actions": {
16.           "shrink": { "number_of_shards": 1 },
17.           "forcemerge": { "max_num_segments": 1 }
18.         }
19.       },
20.       "delete": {
21.         "min_age": "30d",
22.         "actions": { "delete": {} }
23.       }
24.     }
25.   }
26. }

复制代码

9.2 Logstash优化

1. # logstash.yml
2. pipeline.workers: 4          # CPU核心数
3. pipeline.batch.size: 1000    # 批处理大小
4. pipeline.batch.delay: 50     # 批处理等待时间(ms)

复制代码

9.3 Filebeat优化

1. # 批量发送output.logstash:
2.   hosts: ["192.168.188.10:5044"]  bulk_max_size: 2048  # 队列设置queue.mem:  events: 4096  flush.min_events: 2048  flush.timeout: 1s

复制代码

十、效果对比

指标传统方式ELK查日志耗时10-30分钟几秒跨服务器查询手动逐台查一次搜索全部历史日志需要下载直接查统计分析写脚本点几下告警无自动实际收益：

• 排查问题效率提升10倍
  
• 终于能回答"昨天有多少报错"了
  
• 定位问题从"猜"变成"查"
  

总结

ELK搭建不难，难的是：

• 日志格式要规范（JSON最好）
  
• 字段要统一（各应用约定好）
  
• 索引要管理（不然磁盘爆炸）
  
• 网络要打通（异地服务器用组网）
  

建议先从单个应用开始，跑顺了再推广到其他应用。
有问题评论区交流~
[code][/code]
来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！