MonkeyCode搞定ELK日志体系：从零搭建生产级日志平台

步调员的一样平常：看日志→猜题目→改代码→看日志。有了ELK，整个过程从"猜"酿成"查"。

ELK是什么？

1. ELK Stack
2. ├── Elasticsearch  → 存储 + 搜索（核心）
3. ├── Logstash     → 收集 + 过滤 + 转发（处理器）
4. ├── Kibana       → 查询 + 可视化（界面）
5. └── Beats        → 日志采集Agent（搬运工）

复制代码

新版本叫 ELK Stack = Elasticsearch + Logstash + Kibana + Beats
团体架构

1. 你的应用（Python/Go/Java）
2.     ↓ 输出JSON日志（stdout或文件）
3. Filebeat（轻量Agent，部署在每台服务器）
4.     ↓ 采集日志，转发
5. Logstash（过滤、解析、丰富）
6.     ↓ 写入
7. Elasticsearch（存储、索引、搜索）
8.     ↓ 查询展示
9. Kibana（可视化、告警、Dashboard）

复制代码

第一步：Docker Compose一键启动

让MonkeyCode天生：

1. # docker-compose.elk.yml
2. version: '3.8'
3. services:
4.   elasticsearch:
5.     image: elasticsearch:8.12.0
6.     environment:
7.       - discovery.type=single-node
8.       - xpack.security.enabled=false
9.       - "ES_JAVA_OPTS=-Xms2g -Xmx2g"
10.     ports: ["9200:9200", "9300:9300"]
11.     volumes: ["es_data:/usr/share/elasticsearch/data"]
12.     healthcheck:
13.       test: ["CMD-SHELL", "curl -f http://localhost:9200 || exit 1"]
14.       interval: 30s
15.       timeout: 10s
16.       retries: 5
17.       
18.   logstash:
19.     image: logstash:8.12.0
20.     ports: ["5044:5044", "5000:5000/udp"]
21.     volumes:
22.       - ./logstash/pipeline:/usr/share/logstash/pipeline:ro
23.       - ./logstash/config:/usr/share/logstash/config:ro
24.     depends_on: [elasticsearch]
25.    
26.   kibana:
27.     image: kibana:8.12.0
28.     ports: ["5601:5601"]
29.     environment:
30.       - ELASTICSEARCH_HOSTS=http://elasticsearch:9200
31.     depends_on: [elasticsearch]
32.    
33.   filebeat:
34.     image: elastic/filebeat:8.12.0
35.     user: root
36.     volumes:
37.       - /var/lib/docker/containers:/var/lib/docker/containers:ro
38.       - /var/run/docker.sock:/var/run/docker.sock:ro
39.       - ./filebeat.yml:/usr/share/filebeat/filebeat.yml:ro
40.       - ./logs:/var/log/app:ro
41.     depends_on: [logstash]
42.     network_mode: "host"
44. volumes:
45.   es_data:

复制代码

启动：

1. docker-compose -f docker-compose.elk.yml up -d

复制代码

访问Kibana：http://localhost:5601
第二步：应用输出布局化日志

关键原则：永久输出JSON格式，不要输出文本日志。

1. # app/logging_config.py
2. import structlog
3. import json
4. import sys
6. def setup_logging():
7.     structlog.configure(
8.         processors=[
9.             # 添加时间戳
10.             structlog.processors.TimeStamper(fmt="iso"),
11.             # 添加日志级别
12.             structlog.processors.add_log_level,
13.             # 添加进程/线程信息
14.             structlog.processors.add_logger_name,
15.             # 异常信息格式化
16.             structlog.processors.format_exc_info,
17.             # 输出为JSON
18.             structlog.dev.JSONRenderer()
19.         ],
20.         logger_factory=structlog.PrintLoggerFactory(file=sys.stdout),
21.         wrapper_class=structlog.stdlib.BoundLogger,
22.     )
24. logger = structlog.get_logger()
26. # 使用
27. async def create_order(req: CreateOrderRequest, user_id: int):
28.     log = logger.bind(user_id=user_id, endpoint="/orders")
29.     log.info("order.create.attempt", amount=req.amount, items=len(req.items))
30.     try:
31.         order = await db.create_order(req, user_id)
32.         log.info("order.create.success", order_id=order.id, amount=order.total_amount)
33.         return order
34.     except Exception as e:
35.         log.error("order.create.failed", error=str(e), exc_info=True)
36.         raise

复制代码

输出到stdout的JSON日志：

1. {"event": "order.create.success", "level": "info", "timestamp": "2024-06-01T10:30:00Z", "user_id": 42, "order_id": 10086, "amount": 299.0}

复制代码

第三步：Filebeat收罗日志

1. # filebeat.yml
2. filebeat.inputs:
3.   - type: log
4.     enabled: true
5.     paths:
6.       - /var/log/app/*.log
7.       - /var/log/app/*.json
8.     json.keys_under_root: true   # 解析JSON，字段提到顶层
9.     json.overwrite_keys: true
10.     tags: ["app", "production"]
11.     fields:
12.       env: production
13.       app: my-service
14.     fields_under_root: true
16.   - type: container                  # Docker容器日志
17.     enabled: true
18.     paths:
19.       - /var/lib/docker/containers/*/*.log
20.     json.keys_under_root: true
22. processors:
23.   - add_docker_metadata: ~
24.   - add_host_metadata: ~
25.   - dissect:
26.       tokenizer: "%{timestamp} %{level} %{message}"
27.       field: "message"
28.       target_prefix: ""
30. # 输出到Logstash（推荐）或直连Elasticsearch
31. output.logstash:
32.   hosts: ["logstash:5044"]
34. # 或者直接输出到Elasticsearch（跳过Logstash，简单场景）
35. # output.elasticsearch:
36. #   hosts: ["elasticsearch:9200"]
37. #   index: "my-app-%{+yyyy.MM.dd}"

复制代码

第四步：Logstash过滤与分析

1. # logstash/pipeline/logstash.conf
2. input {
3.   beats {
4.     port => 5044
5.   }
6. }
8. filter {
9.   # 解析时间戳
10.   if [timestamp] {
11.     date {
12.       match => [ "timestamp", "ISO8601" ]
13.     }
14.   }
15.   
16.   # 解析JSON字段（如果Filebeat没解析）
17.   if [message] =~ /^\{.*\}$/ {
18.     json {
19.       source => "message"
20.       skip_on_invalid_json => true
21.     }
22.   }
23.   
24.   # 添加地理位置（IP地址）
25.   if [remote_ip] {
26.     geoip {
27.       source => "remote_ip"
28.       target => "geoip"
29.     }
30.   }
31.   
32.   # 添加User-Agent解析
33.   if [user_agent] {
34.     useragent {
35.       source => "user_agent"
36.       target => "ua"
37.     }
38.   }
39.   
40.   # 过滤掉健康检查日志
41.   if [endpoint] == "/health" {
42.     drop {}
43.   }
44. }
46. output {
47.   elasticsearch {
48.     hosts => ["elasticsearch:9200"]
49.     index => "logs-%{+YYYY.MM.dd}"
50.     manage_template => false
51.   }
52.   
53.   # 调试用：同时输出到控制台
54.   # stdout { codec => rubydebug }
55. }

复制代码

第五步：在Kibana中查询

启动后，在Kibana中：
1. 创建Index Pattern

进入 Management → Stack Management → Index Patterns，创建 logs-*
2. 常用查询语法（KQL）

[code]# 正确匹配level: error# 暗昧搜索message: "order"# 多条件level: error AND service: payment# 范围查询@timestamp >= "2024-06-01T00:00:00Z" AND @timestamp