如何自定义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 会被自动发现机制读取并转换为对应的输入与解析配置，适合在容器平台做“按服务/命名空间动态采集”。

最新问答

相关标签