4.8 集中式日志管理方案

预计阅读时间：14 分钟

学习目标

学完本章后，你将能够：

• 理解集中式日志管理的架构和必要性
• 配置 rsyslog 实现多服务器日志集中收集
• 搭建 ELK（Elasticsearch + Logstash + Kibana）日志平台
• 使用 Filebeat 采集日志并发送到 Logstash/Elasticsearch
• 编写结构化日志（JSON 格式）并遵循日志规范

核心知识

• 集中式日志管理——将分散在多台服务器上的日志统一收集、存储、索引和查询的体系
• rsyslog——Linux 默认的日志采集和转发服务，支持 TCP/UDP 远程发送
• ELK Stack——Elasticsearch（存储+搜索）+ Logstash（加工处理）+ Kibana（可视化）的日志方案
• Filebeat——轻量级日志采集 Agent，资源占用低，支持将日志发送到 ES 或 Logstash
• Loki——Grafana 出品的轻量日志系统，使用标签索引而非全文索引，更省资源
• 结构化日志（Structured Logging）——使用 JSON 等格式输出日志，便于程序解析和搜索
• 日志轮转（Log Rotation）——按大小或时间拆分日志文件，防止磁盘写满
• 日志等级——DEBUG < INFO < WARN < ERROR < FATAL，用于控制日志输出粒度

知识关联

• 前置知识：2.7：日志与故障排查 日志与故障排查（本地日志基础）、1.9：重定向与管道 重定向与管道（理解日志输出机制）
• 后续影响：集中日志是 3.12：系统监控与告警 系统监控方案和运维可观测性的基础数据源
• 进阶阅读：6.9：OpenTelemetry 与可观测性 OpenTelemetry（分布式追踪与度量）、5.12：eBPF 基础 eBPF（内核级观测）

原理讲解

为什么需要集中式日志

当只有一两台服务器时，SSH 登录到机器上用 tail -f /var/log/syslog 就能解决问题。但当服务器数量增长到 10+ 台，每个问题都需要"登录多台机器查日志"，效率极低。集中式日志的核心价值在于：一个搜索框查所有服务器。此外，集中存储还能解决安全问题——攻击者清理本地日志后，远程日志服务器上的记录仍然存在。

ELK vs Loki 选型

ELK（Elasticsearch + Logstash + Kibana）是传统选择，功能全面但资源消耗较大。Logstash 需要 JVM，Elasticsearch 对内存需求高。Loki + Grafana 是新兴的轻量方案，Loki 不建全文索引，而是复用 Prometheus 的标签体系，存储效率更高。选型参考：已有 Prometheus/Grafana 监控体系 → Loki；需要复杂全文检索和聚合分析 → ELK；仅基础需求 → rsyslog + grep。

结构化日志的意义

非结构化日志如 2024-01-15 10:30:00 ERROR: connection timeout to db 只能用 grep 或全文搜索。结构化日志（JSON）如 {"timestamp":"2024-01-15T10:30:00Z","level":"ERROR","service":"api","message":"connection timeout","db_host":"db01"} 可以在 ELK 中按字段过滤和聚合，例如：统计每个数据库的连接超时次数、按服务聚合错误率等。

日志流水线模型：采集→传输→处理→存储→查询

任何集中式日志方案都遵循同一条流水线：采集（Agent 读文件/收端口）→ 传输（网络发送，需考虑压缩与可靠性）→ 处理（解析字段、补时区、脱敏、丢弃无用日志）→ 存储（索引或标签化存储，含保留策略）→ 查询/告警（搜索、聚合、可视化）。理解这条流水线后，任何组件都能对号入座：Filebeat/Promtail 是采集器，Kafka/Redis 可作缓冲队列（传输层），Logstash/Fluentd 是处理器，Elasticsearch/Loki 是存储层，Kibana/Grafana 是查询界面。排查日志链路问题时，从采集端到查询端逐段验证：Agent 进程活着吗 → 输出连通吗 → 处理器有报错吗 → 存储里有数据吗 → 界面索引对不对。大多数"日志丢了"的问题都出在采集和传输段。

为什么 ELK 需要 Logstash 而 Loki 不需要

ELK 架构中 Logstash 是必需的中间层，而 Loki + Promtail 架构中没有对应的重型处理器。根本原因在于索引策略的不同：Elasticsearch 对每条日志的所有字段建立全文索引，因此在存储前必须先解析出结构化字段（通过 Logstash 的 grok/mutate 过滤器）。Loki 则只对日志的标签（label）建立索引，日志正文保持原样存储，查询时再用 LogQL 正则过滤。这意味着 Loki 的"处理"发生在查询时而非采集时，代价是查询性能略低，但采集链路大幅简化——Promtail 只需要做标签分配和基本的 Pipeline 处理（解析时间戳、添加标签），不需要复杂的字段解析。这是写时索引 vs 读时解析的经典权衡。

为什么推荐结构化日志而不是纯文本

纯文本日志如 2024-01-15 10:30 ERROR: connection timeout to db 的问题在于：解析不确定性——不同开发者、不同模块的日志格式不统一，grep 正则需要针对每种格式单独编写；无法聚合——想统计"每种错误类型的数量"需要用复杂的正则分组，效率极低；字段丢失——时间戳、级别、服务名、请求 ID 等关键信息混在文本中，无法被程序直接使用。结构化日志（JSON 格式）将每个字段显式键值化，ELK/Loki 可自动解析为独立字段，支持按字段过滤、聚合、告警。最佳实践：应用层直接输出 JSON（Python structlog / Node.js pino），而非在 Logstash 中用 grok 事后解析——后者既消耗 Logstash 算力，又容易因格式变化导致解析失败。

示例代码

1. rsyslog 集中收集

# 方案 A：日志服务器（接收端）配置
# /etc/rsyslog.conf
sudo tee -a /etc/rsyslog.conf << 'EOF'
# 启用 TCP 模块接收远程日志
module(load="imtcp")
input(type="imtcp" port="514")
EOF

sudo systemctl restart rsyslog
sudo ufw allow 514/tcp

# 方案 B：客户端（发送端）配置
# /etc/rsyslog.d/50-remote.conf
echo '*.* @@logserver.example.com:514' | sudo tee /etc/rsyslog.d/50-remote.conf
sudo systemctl restart rsyslog

# 验证远程日志收到
sudo tail -f /var/log/syslog | grep "client-hostname"

2. Filebeat 安装与配置

# 安装 Filebeat（每台需要采集日志的服务器）
curl -L -O https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-8.15.0-amd64.deb
sudo dpkg -i filebeat-8.15.0-amd64.deb

# 配置 filebeat.yml
sudo tee /etc/filebeat/filebeat.yml << 'EOF'
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/nginx/*.log
    - /var/log/syslog
  json.keys_under_root: true
  json.overwrite_keys: true

output.elasticsearch:
  hosts: ["http://elasticsearch-server:9200"]
  username: "filebeat"
  password: "your_password"
EOF

# 启动 Filebeat
sudo systemctl enable --now filebeat

# 验证连接状态
sudo filebeat test output
# 输出: elasticsearch: http://elasticsearch-server:9200... OK

3. 搭建 ELK 栈（Docker Compose）

# docker-compose.elk.yml（新版 Docker Compose V2 已废弃 version 字段）
services:
  elasticsearch:
    image: elasticsearch:8.12.0
    environment:
      - discovery.type=single-node
      - "ES_JAVA_OPTS=-Xms1g -Xmx1g"
      - xpack.security.enabled=false
    volumes:
      - es_data:/usr/share/elasticsearch/data
    ports:
      - "9200:9200"

  logstash:
    image: logstash:8.12.0
    volumes:
      - ./logstash.conf:/usr/share/logstash/pipeline/logstash.conf
    ports:
      - "5044:5044"
    depends_on:
      - elasticsearch

  kibana:
    image: kibana:8.12.0
    ports:
      - "5601:5601"
    environment:
      - ELASTICSEARCH_HOSTS=http://elasticsearch:9200
    depends_on:
      - elasticsearch

volumes:
  es_data:

# logstash.conf（Logstash 处理流水线）
input {
  beats { port => 5044 }
}
filter {
  grok {
    match => { "message" => "%{COMBINEDAPACHELOG}" }
  }
  date {
    match => ["timestamp", "ISO8601"]
  }
}
output {
  elasticsearch {
    hosts => ["http://elasticsearch:9200"]
    index => "logs-%{+YYYY.MM.dd}"
  }
}

# 启动
docker compose -f docker-compose.elk.yml up -d
# 验证 ES 可用
curl http://localhost:9200
# 访问 Kibana: http://localhost:5601

4. Loki + Grafana 轻量方案

# docker-compose.loki.yml（新版 Docker Compose V2 已废弃 version 字段）
services:
  loki:
    image: grafana/loki:3.0.0
    ports:
      - "3100:3100"
    command: -config.file=/etc/loki/local-config.yaml
    volumes:
      - loki_data:/loki

  promtail:
    image: grafana/promtail:2.9.0
    volumes:
      - /var/log:/var/log
      - ./promtail-config.yaml:/etc/promtail/config.yml
    command: -config.file=/etc/promtail/config.yml

  grafana:
    image: grafana/grafana:latest
    ports:
      - "3000:3000"
    environment:
      - GF_INSTALL_PLUGINS=grafana-lokiexplore-app
    volumes:
      - grafana_data:/var/lib/grafana

volumes:
  loki_data:
  grafana_data:

# promtail-config.yaml
# scrape_configs:
#   - job_name: system
#     static_configs:
#       - targets: [localhost]
#         labels:
#           job: varlogs
#           __path__: /var/log/*log

# 启动
docker compose -f docker-compose.loki.yml up -d
# 访问 Grafana: http://localhost:3000 → Explore → 选择 Loki 数据源

5. 应用结构化日志输出示例

# Node.js 结构化日志示例
# npm install pino
const pino = require('pino');
const logger = pino({
  level: process.env.LOG_LEVEL || 'info',
  formatters: {
    level: (label) => ({ level: label }),
  },
  timestamp: pino.stdTimeFunctions.isoTime,
});

logger.info({ userId: 123, action: 'login' }, '用户登录');
logger.error({ err, db_host: 'db01' }, '数据库连接失败');

# 输出:
# {"level":"info","time":"2024-01-15T10:30:00.000Z","userId":123,"action":"login","msg":"用户登录"}
# {"level":"error","time":"2024-01-15T10:30:01.000Z","err":{"message":"connect ECONNREFUSED"},"db_host":"db01","msg":"数据库连接失败"}

# Python 结构化日志示例
# pip install structlog
import structlog
logger = structlog.get_logger()
logger.info("user_login", user_id=123, action="login")
logger.error("db_connection_failed", db_host="db01")

6. 日志轮转配置

# logrotate 配置文件 /etc/logrotate.d/custom-app
/var/log/myapp/*.log {
    daily
    rotate 30
    maxsize 100M
    compress
    delaycompress
    missingok
    notifempty
    copytruncate
    postrotate
        systemctl reload myapp > /dev/null 2>&1 || true
    endscript
}

# 手动触发生效
sudo logrotate -f /etc/logrotate.d/custom-app

7. Logstash grok 解析实战

# 目标: 将 Nginx access log 拆成可过滤字段
# 原始行: 10.0.0.8 - - [15/Jan/2026:10:30:05 +0800] "GET /api/users?id=1 HTTP/1.1" 200 1024 "Mozilla/5.0" "-"

input { beats { port => 5044 } }

filter {
  grok {
    match => { "message" => "%{IPORHOST:client_ip} - - \[%{HTTPDATE:ts}\] \"%{WORD:method} %{URIPATHPARAM:uri} HTTP/%{NUMBER:http_version}\" %{NUMBER:status:int} %{NUMBER:bytes:int} \"%{DATA:referer}\" \"%{DATA:ua}\"" }
  }
  date {
    match => ["ts", "dd/MMM/yyyy:HH:mm:ss Z"]
    target => "@timestamp"
  }
  # 按状态码打业务标签
  if [status] >= 500 {
    mutate { add_field => { "severity" => "error" } }
  }
  # 丢弃静态资源日志（可减少 50%-70% 存储）
  if [uri] =~ "^/(css|js|img)/" {
    drop { }
  }
  # 移除原始字段，降低存储成本
  mutate { remove_field => ["message", "ts", "referer", "ua"] }
}

output { elasticsearch { hosts => ["http://elasticsearch:9200"] index => "nginx-%{+YYYY.MM.dd}" } }

# 常用 grok 模式: %{IPORHOST} %{HTTPDATE} %{WORD} %{NUMBER} %{URIPATHPARAM} %{DATA}
# 调试: Kibana → Dev Tools → Grok Debugger 粘贴原始行实时验证表达式

8. Loki LogQL 查询语法

# LogQL = 标签选择器 + 过滤表达式 + 聚合函数

# 基础选择与过滤
{job="varlogs"}                            # 按标签选日志流
{job="nginx"} |= "ERROR"                   # 包含子串
{job="nginx"} != "healthcheck"             # 排除子串
{job="nginx"} |~ "5\\d\\d"                 # 正则（匹配 5xx 状态码）

# 时间范围
{job="nginx"} |= "500" [5m]                # 最近 5 分钟

# 解析 JSON 字段（对应本章结构化日志实践）
{job="app"} | json level, message, db_host="db_host"
{job="app"} | json | level="error"         # 链式: 先解析再过滤

# 速率统计: ERROR 日志每秒条数
sum(rate({job="app"} | json | level="error" [5m]))

# 按 path 分组统计 404 数量
sum by (path) (count_over_time({job="nginx"} | json path, status | status="404" [5m]))

# 日志量排名（哪些服务日志最大）
topk(10, sum by (job) (rate({job=~".+"} [5m])))

# 告警规则（与 Prometheus 共用 Alertmanager 路由）
# groups:
#   - name: log-alerts
#     rules:
#       - alert: ErrorLogsHigh
#         expr: sum(rate({job="app"} | json | level="error" [5m])) > 5
#         for: 10m
#         annotations:
#           summary: "应用错误日志超过 5 条/秒"

9. 日志脱敏与敏感信息保护

# 原则: 密码/手机号/身份证/信用卡/token 要么不打日志，要么打码
# 方案一: Logstash 脱敏（gsub 正则替换）
filter {
  mutate {
    gsub => [
      "message", "(password|passwd|pwd)[=:][^\\s,;>]+", "\\1[REDACTED]",
      "message", "[0-9]{11}", "[PHONE]"
    ]
  }
}

# 方案二: Filebeat 采集端脱敏（更早处理，节省传输与存储）
processors:
  - drop_fields:
      fields: ["user.password", "credit_card"]
  - fingerprint:
      fields: ["email"]
      target_field: "email_hash"        # 只保留可关联哈希，不存明文
  - redact:
      fields:
        - user.password
        - token
      pattern: "[A-Za-z0-9]{16,}"       # 疑似 token 全部打码

# 方案三: 应用侧规范（最可靠 —— 源头不产生敏感数据）
# 日志模板只允许白名单字段；错误对象序列化时排除敏感 key
# Python structlog: logger.info("login", user_id=123)  —— 绝不打印 password

# 自检: 扫描现有日志中的敏感格式
sudo grep -rE 'password[=:]|BEGIN (RSA )?PRIVATE KEY|access_token|card_number' /var/log/ | head

10. Elasticsearch ILM 索引生命周期管理

# 场景: 每日约 20GB 日志，要求 30 天在线可查 + 90 天冷存
# 策略: 7 天 hot（热索引）→ 23 天 warm（压缩）→ 60 天 cold（可搜索）→ delete

# 创建 ILM 策略
PUT _ilm/policy/logs-policy
{
  "policy": {
    "phases": {
      "hot":  { "actions": { "rollover": { "max_size": "50gb", "max_age": "1d" } } },
      "warm": { "min_age": "7d",  "actions": { "forcemerge": { "max_num_segments": 1 } } },
      "cold": { "min_age": "30d", "actions": { "allocate": { "number_of_replicas": 0 } } },  # ES 8.x 已移除 freeze 动作，冷阶段可缩副本或转 searchable snapshot
      "delete": { "min_age": "90d", "actions": { "delete": {} } }
    }
  }
}

# 索引模板: 新索引自动套用策略与别名
PUT _index_template/logs-template
{
  "index_patterns": ["logs-*"],
  "template": {
    "settings": {
      "number_of_shards": 3,
      "number_of_replicas": 1,
      "index.lifecycle.name": "logs-policy",
      "index.lifecycle.rollover_alias": "logs"
    }
  }
}

# 查看 ILM 执行状态
GET _ilm/status
GET logs-*/_ilm/explain

# 成本估算: 原始 20GB/天 × 副本 1 = 40GB/天，90 天 ≈ 3.6TB
# 结合 grok 丢弃静态资源 + 字段裁剪 + forcemerge 压缩，实际可降到 30%-40%

常见错误

最佳实践

练习题

1. （概念）为什么说"轮转配置和集中存储"是日志管理的两个基本要求？
2. （概念）ELK 中 Elasticsearch、Logstash、Kibana 各自负责什么？Filebeat 和 Logstash 有什么区别？
3. （实操）在本地用 Docker Compose 启动 ELK 栈（参考示例代码），启动后通过 curl 向 Elasticsearch 的 test-index 写入一条 JSON 文档，然后在 Kibana 中创建索引模式并搜索到该文档。
4. （实操）编写一个 Shell 脚本，生成包含以下字段的结构化日志到 /var/log/myapp.log：timestamp、level、service、message。配置 logrotate 每日轮转，保留 30 天，压缩旧文件。验证轮转生效。
5. （🔍 挑战）搭建 ELK + Filebeat 体系：启动 ELK Docker Compose；在另一台（或同一台另开终端）安装 Filebeat，配置采集 Nginx access log（/var/log/nginx/access.log）；用 ab 或 wrk 生成一些 HTTP 请求产生日志；在 Kibana 中搜索并筛选出状态码为 404 的请求，创建一个聚合图表展示 404 比例变化趋势。

学习检查点

学完本章后，请检验自己是否掌握以下内容：

本章总结

集中式日志管理将分散的服务器日志汇聚到统一平台，是运维可观测性的基石。三个主流方案各有定位：rsyslog 轻量无依赖，适合传统架构和基本需求；ELK（Elasticsearch + Logstash + Kibana）功能全面但资源消耗高，适合需要复杂搜索和聚合的场景；Loki + Grafana 轻量高效，与 Prometheus 监控生态无缝集成。无论选择哪种方案，应用输出结构化的 JSON 日志是最重要的一步——结构化数据才是真正可被程序高效利用的数据。

速查表

学习路径建议

• 学完本章后建议阅读 6.9：OpenTelemetry 与可观测性 OpenTelemetry（日志+指标+追踪三合一的可观测性标准）
• 进阶可学习 Elasticsearch 集群管理和性能调优
• 日志告警场景可参考 3.12：系统监控与告警 系统监控（Prometheus Alertmanager）

延伸阅读