4. Filebeat实战

Filebeat 是 Elastic 官方提供的轻量级日志采集器，部署在产生日志的服务器上，负责把文本日志读取、加工并投递到 Elasticsearch 或 Logstash。它本身不做复杂计算，核心能力集中在三件事：怎么读（输入与多行合并）、怎么改（字段解析与加工）、怎么送（输出到 ES 或 Logstash）。本节以一个 Java 应用的日志采集为例，把这三件事串成一份可直接参考的完整配置。

一、输入与多行合并：让一条异常堆栈成为一条事件

Java 应用抛异常时，一条日志会跨越多行，例如：

2024-06-11 10:23:41.512 ERROR [order-service,traceId=8f3a] c.e.order.PayHandler - 支付回调处理失败
java.lang.NullPointerException: Cannot invoke "Order.getAmount()" because "order" is null
    at com.example.order.PayHandler.handle(PayHandler.java:88)
    at com.example.order.CallbackController.callback(CallbackController.java:41)
    at java.base/java.lang.Thread.run(Thread.java:833)

如果按行采集，这 5 行会变成 5 条独立事件，堆栈被拆散，检索时无法还原现场。Filebeat 的 multiline 配置就是解决这个问题：它把满足条件的后续行“粘”到上一条事件上。判断规则由 pattern 和 negate 共同决定。

常用写法是“以时间戳开头的行才是新事件”，其余行都归入上一条：

filebeat.inputs:
- type: filestream
  id: order-app-log
  enabled: true
  paths:
    - /data/apps/order-service/logs/*.log
  multiline:
    pattern: '^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}'
    negate: true
    match: after
    max_lines: 500
    timeout: 5s

这里 negate: true 表示“不匹配 pattern 的行”才被合并，match: after 表示合并到上一行之后。max_lines 限制单条事件最多 500 行，防止异常刷屏导致内存膨胀；timeout: 5s 表示超过 5 秒没有新行就强制结束当前事件，避免最后一条日志长期滞留。判断合并是否生效，可以在 Kibana 中查看该条事件的 message 字段是否包含完整堆栈，或观察 log.flags: multiline 标记。

二、字段解析：把 message 拆成可检索的结构化字段

合并后的 message 仍是一整段文本，直接全文检索效率低。Filebeat 提供 dissect、grok、processors 等加工能力。对于格式固定的应用日志，dissect 比 grok 更快，因为它按分隔符切分而不做正则回溯。

processors:
- dissect:
    tokenizer: "%{ts} %{level} [%{service},traceId=%{trace_id}] %{logger} - %{msg}"
    field: "message"
    target_prefix: ""
    ignore_failure: true
- timestamp:
    field: ts
    layouts:
      - '2006-01-02 15:04:05.000'
    timezone: Asia/Shanghai
- drop_fields:
    fields: ["ts", "agent.ephemeral_id", "ecs.version"]
    ignore_missing: true

解析后，level、service、trace_id、logger 都成为独立字段，可以直接做 level: ERROR 过滤或按 trace_id 串联调用链。timestamp 处理器把日志里的时间替换为事件时间，保证排序正确；drop_fields 删除冗余字段以节省存储。若日志格式不固定，可改用 grok：

- grok:
    field: message
    patterns:
      - '%{TIMESTAMP_ISO8601:ts} %{LOGLEVEL:level} \[%{DATA:service},traceId=%{DATA:trace_id}\] %{JAVACLASS:logger} - %{GREEDYDATA:msg}'
    ignore_failure: true

验证解析是否成功，最直接的方式是运行 filebeat test config -c filebeat.yml 检查语法，再用 filebeat -e -c filebeat.yml 前台启动，观察输出中是否出现解析后的字段；若某字段缺失，通常是 tokenizer 分隔符与实际日志不一致。

三、模块配置：用现成模板快速接入常见日志

对于 Nginx、MySQL、Redis、系统日志等标准格式，不必手写解析规则，直接启用 Filebeat 模块即可。模块内置了输入路径、多行规则、字段解析和 Kibana 仪表盘。

filebeat.config.modules:
  path: ${path.config}/modules.d/*.yml
  reload.enabled: true
  reload.period: 10s

然后执行 filebeat modules enable nginx mysql，再编辑 modules.d/nginx.yml 指定日志路径：

- module: nginx
  access:
    enabled: true
    var.paths: ["/var/log/nginx/access.log"]
  error:
    enabled: true
    var.paths: ["/var/log/nginx/error.log"]

模块的价值在于“开箱即用”，但要注意版本匹配：模块模板与 Kibana 版本不一致时，仪表盘可能无法加载。启用后可用 filebeat modules list 确认状态。

四、输出到 Elasticsearch 与 Logstash 的选择

输出目标决定数据链路形态。直连 Elasticsearch 链路短、延迟低，适合日志量中等、无需复杂二次加工的场景；输出到 Logstash 则多一层缓冲与加工能力，适合需要聚合、富化、路由到多个索引的场景。

直连 Elasticsearch 的配置片段：

output.elasticsearch:
  hosts: ["http://es01:9200", "http://es02:9200"]
  username: "filebeat_writer"
  password: "${ES_PWD}"
  index: "order-%{[service]}-%{+yyyy.MM.dd}"
  pipeline: "order-parse-pipeline"

输出到 Logstash 的配置片段：

output.logstash:
  hosts: ["logstash01:5044", "logstash02:5044"]
  loadbalance: true
  worker: 2

对应 Logstash 侧需要 beats 输入：

input {
  beats {
    port => 5044
  }
}
filter {
  if [service] == "order-service" and [level] == "ERROR" {
    mutate { add_tag => ["alert"] }
  }
}
output {
  elasticsearch {
    hosts => ["http://es01:9200"]
    index => "order-%{+yyyy.MM.dd}"
  }
}

选择建议：如果只是把日志集中存储并检索，直连 ES 更简单；如果需要在入库前做字段富化、脱敏、按业务路由到不同索引，或需要削峰缓冲，就走 Logstash。两者可以并存，但同一台 Filebeat 只能启用一个输出，多目标需借助 Logstash 或 Kafka 中转。

五、完整配置与验证闭环

把上述片段合并成一份可运行的 filebeat.yml 骨架：

filebeat.inputs:
- type: filestream
  id: order-app-log
  enabled: true
  paths:
    - /data/apps/order-service/logs/*.log
  multiline:
    pattern: '^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}'
    negate: true
    match: after
    max_lines: 500
    timeout: 5s

processors:
- dissect:
    tokenizer: "%{ts} %{level} [%{service},traceId=%{trace_id}] %{logger} - %{msg}"
    field: "message"
    ignore_failure: true
- timestamp:
    field: ts
    layouts: ['2006-01-02 15:04:05.000']
    timezone: Asia/Shanghai

output.elasticsearch:
  hosts: ["http://es01:9200"]
  index: "order-%{[service]}-%{+yyyy.MM.dd}"

logging.level: info
logging.to_files: true
logging.files:
  path: /var/log/filebeat
  keepfiles: 7

验证按四步走：第一步 filebeat test config -c filebeat.yml 确认语法；第二步 filebeat test output 确认能连上 ES；第三步前台启动 filebeat -e -c filebeat.yml，观察是否有 Harvester started 和 Non-zero metrics；第四步在 Kibana 中按 service: order-service and level: ERROR 检索，检查 trace_id 是否存在、事件时间是否为日志时间、堆栈是否完整。若发现事件被拆散，优先检查 multiline 的 pattern 是否匹配了所有“新事件起始行”；若字段缺失，检查 dissect 分隔符与真实日志是否一致。