如何自定义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 从文件尾部开始读取（首次导入慎用，避免丢首条）。
• 多行日志（如 Java 堆栈）
  • 使用 multiline 将多行合并为一条事件，常见策略是“以非行首模式匹配作为上一行延续”，并设置超时以结束聚合。
• JSON 日志
  • 行内 JSON：使用 json.keys_under_root / json.overwrite_keys / json.add_error_key / json.message_key 将 JSON 解析到顶层，并可配合行过滤与多行策略使用。
• 示例（组合多行 + JSON + 字段增强）
  • 说明：该示例演示如何采集 /var/log/myapp/*.log，将包含 timestamp 字段的 JSON 消息解析并将时间赋给 @timestamp，同时添加业务字段与标签。
  • 配置片段：

    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：将指定字段解析为 @timestamp（注意时区与布局匹配）。
  • fingerprint：生成事件指纹，常用于去重或作为文档 ID（写入 ES 时设置 metadata._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 / 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 会被自动发现机制读取并转换为对应的输入与解析配置，适合在容器平台做“按服务/命名空间动态采集”。