如何自定义Filebeat的日志处理规则

自定义 Filebeat 日志处理规则的实用指南

先说说最基本的配置结构和生效方式。

默认情况下，配置文件位于 /etc/filebeat/filebeat.yml（rpm/deb 安装方式），旁边还有一个完整的参考示例 /etc/filebeat/filebeat.reference.yml，可以随时拿来对照。建议把自定义规则拆成三个部分来维护：inputs 负责输入，processors 负责处理，output 负责输出。如果用到模块，modules.d 目录下统一管理。改完配置之后，别急着重启，先跑一遍语法校验和连通性测试，确保万无一失再上线。

常用操作其实就那么几条：

• 语法和配置测试：filebeat test config -e
• 输出连通性测试：filebeat test output
• 服务管理（systemd）：sudo systemctl start|enable|status filebeat
• 看运行日志：journalctl -u filebeat -f
• 模块管理：在 modules.d 目录执行 filebeat modules enable nginx mysql，或者用 filebeat modules list 查看已启用的模块

输入与多行 JSON 配置

基础的文件输入配置，核心参数就那么几个：paths 指定采集路径，tags 打业务标签，fields 和 fields_under_root 控制自定义字段是否提升到顶层，encoding 处理编码问题，ignore_older 忽略旧文件，tail_files 从文件尾部开始读取。不过 tail_files 首次导入时要慎用，否则容易丢掉第一条日志。

那多行日志怎么处理呢？比如 Ja va 的堆栈信息，一行根本说不清。用 multiline 配置就能把多行合并成一条事件。常见的策略是“以非行首模式匹配的行作为上一行的延续”，再设置一个超时时间，确保聚合不会无限等待下去。

JSON 日志的处理也不复杂。行内 JSON 可以用 json.keys_under_root、json.overwrite_keys、json.add_error_key、json.message_key 这几个参数，把 JSON 解析到顶层字段，而且还能配合行过滤和多行策略一起使用。

举个例子，看看多行、JSON 和字段增强怎么组合在一起：

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/myapp/*.log
  tags: ["myapp", "json"]
  fields:
    app_id: myapp-prod
    env: prod
  fields_under_root: true
  multiline:
    pattern: '^\\[?\\d{4}-\\d{2}-\\d{2}'  # 以日期或 [ 开头的行作为新事件起点
    negate: true
    match: after
    timeout: 5s
  json.keys_under_root: true
  json.overwrite_keys: true
  json.add_error_key: true
  json.message_key: message

processors:
  - timestamp:
      source: datetime
      target: "@timestamp"
      layouts:
        - "2006-01-02T15:04:05.999999999Z07:00"  # 注意这是 Go 时间格式，必须和日志中的时间格式一致

output.elasticsearch:
  hosts: ["http://localhost:9200"]
  index: "filebeat-%{[agent.version]}-%{+yyyy.MM.dd}"

有几个细节需要注意：如果 JSON 顶层没有 message 字段，调整 json.message_key 或者直接移除这个选项；多行和 JSON 解析的先后顺序由配置顺序决定；如果还不够，后面还可以用 processors 继续清洗和增强。

处理器与条件编排

常用处理器其实就那么几个：add_fields、rename、drop_fields、convert 负责字段的增删改和类型转换；decode_json_fields 可以把字符串字段解码成结构化 JSON；timestamp 用来解析时间字段，注意时区要和日志格式匹配；fingerprint 生成事件指纹，常用于去重或者作为文档 ID；add_tags 和 drop_event 则基于条件来打标签或丢弃事件。

条件处理这块，用 when 关键字做判断，支持 equals、contains、regexp 等方式。比如“按日志级别分流处理”，就是典型的应用场景。

来看一个条件打标加去重的例子：

processors:
  - add_tags:
      tags: ["error"]
      when:
        equals:
          log.level: "ERROR"
  - fingerprint:
      fields: ["message", "host.name"]
      target_field: "@metadata._id"
      method: "sha256"

处理器的编排顺序建议遵循一个原则：先解析和标准化（timestamp/json），再丰富和路由（add_fields/tags），最后清理和去重（drop_fields/fingerprint）。这样能减少重复计算，也避免歧义。

输出路由与索引管理

输出到 Elasticsearch 是最直接的方式。通过 index 模板控制索引名，如果需要更复杂的解析和转换，可以在 ES 端配置 pipeline（摄取节点管道），或者在 Filebeat 内部用处理器做轻量清洗。

output.elasticsearch:
  hosts: ["http://es:9200"]
  index: "myapp-%{+yyyy.MM.dd}"
  pipeline: "myapp-ingest-pipeline"

输出到 Kafka 则适合高吞吐、多下游的场景，起到缓冲和解耦的作用。

output.kafka:
  hosts: ["kafka1:9092", "kafka2:9092"]
  topic: "logs-myapp"
  required_acks: 1
  compression: gzip
  partition.round_robin.reachable_only: false

索引和模板方面，用 setup.template.name 和 setup.template.pattern 自定义索引模板。如果用了 ILM（索引生命周期管理），可以结合 setup.ilm.enabled 和 rollover_alias 等参数。如果完全自定义索引名和模板，通常把 setup.template.enabled 设为 false，自己管理模板和 ILM。

Kubernetes 环境的提示式自动发现

在 Kubernetes 或 Docker 环境中，通过 annotations 或 labels 前缀 co.elastic.logs/ 就能动态下发采集规则，完全不用改 Filebeat 主配置。

常用的 hints 包括：

• co.elastic.logs/enabled：控制是否采集该容器日志，设为 "false" 就直接忽略
• co.elastic.logs/module 和 fileset.stdout / fileset.stderr：指定模块和标准输出/错误输出文件集
• co.elastic.logs/json.*：JSON 解析选项，比如 message_key、add_error_key
• co.elastic.logs/include_lines / exclude_lines：按正则包含或排除行
• co.elastic.logs/multiline.*：多行合并策略

举个例子，把容器日志按 JSON 解析并指定 message 键：

apiVersion: apps/v1
kind: Deployment
metadata:
  name: myapp
spec:
  template:
    metadata:
      annotations:
        co.elastic.logs/module: nginx
        co.elastic.logs/fileset.stdout: access
        co.elastic.logs/fileset.stderr: error
        co.elastic.logs/json.message_key: "log"
        co.elastic.logs/json.add_error_key: "true"

这些 hints 会被自动发现机制读取，并转换成对应的输入和解析配置。在容器平台上，按服务或命名空间动态采集，这种方式非常灵活，值得推荐。