Article

日志采集 Filebeat

更新于：2026-07-12

第1章：Filebeat 概述与核心概念

1.1 什么是 Filebeat

概念名称	说明	注意事项
Filebeat	轻量级日志采集器，属于 Elastic Beats 家族，用于将日志文件数据发送到 Logstash 或 Elasticsearch。	不进行复杂处理，适合资源受限环境。
轻量级	基于 Go 编写，无 JVM 依赖，内存和 CPU 占用低。	适用于边缘节点、容器等资源敏感场景。
日志采集器	监控日志文件变化，实时读取新增内容并转发。	支持多行日志（如异常堆栈）合并处理。
Beats 框架组件	Elastic 提供的统一数据采集器框架，Filebeat 是其中最常用的成员。	其他 Beats 包括 Metricbeat、Packetbeat 等。

1.2 Filebeat 的工作原理

概念名称	说明	注意事项
文件监控	Filebeat 启动后扫描指定路径下的日志文件，监控其变化。	使用 inotify（Linux）或 polling 机制。
Harvester	每个打开的日志文件由一个 Harvester 负责逐行读取内容。	文件关闭后 Harvester 退出。
Prospector	管理文件发现，负责启动 Harvester 并跟踪文件状态。	支持 log 和 filestream 两种类型。
数据传输	读取的日志事件经处理器处理后，发送至输出目标（如 ES、Logstash）。	支持背压控制，避免压垮下游服务。
状态记录	使用 registry 文件记录每个文件的读取偏移量，确保重启后不丢数据。	registry 文件默认位于 data/registry。

1.3 Filebeat 的核心组件（Prospector、Harvester、Registry）

组件名称	说明	注意事项
Harvester	单个文件读取器，打开文件并逐行读取内容，发送事件到 spooler。	每个文件一个 Harvester，文件关闭后释放。
Prospector	文件发现器，扫描路径、识别新文件并为每个文件启动 Harvester。	支持 glob 模式匹配文件路径。
Registry	JSON 文件，记录每个日志文件的唯一标识（inode）和当前读取偏移量（offset）。	保证断点续传，避免重复采集。
Spooler	内存缓冲区，临时缓存 Harvester 读取的事件，等待处理器处理。	大小可通过 spool_size 配置。
Publisher	将处理后的事件发送到输出目标，并确认发送成功后更新 registry。	支持 ACK 机制确保至少一次传输。

1.4 Filebeat 与 Logstash、Elasticsearch、Beats 框架的关系

组件名称	说明	注意事项
Beats 框架	Elastic 提供的轻量级数据采集器框架，统一配置与传输协议。	Filebeat 是 Beats 的子项目。
Filebeat	负责日志采集，将数据发送给 Logstash 或直接到 Elasticsearch。	适合简单过滤和转发。
Logstash	数据处理管道，支持复杂解析、过滤、转换（如 Grok、Mutate）。	资源消耗较高，适合中心节点处理。
Elasticsearch	存储和索引数据，提供搜索与分析能力。	Filebeat 可直接写入，无需中间件。
数据流向	Filebeat → (Logstash) → Elasticsearch → Kibana	可根据需求选择是否经过 Logstash。

第2章：安装与基础配置

2.1 支持的平台与安装方式（deb、rpm、tar、docker）

安装方式	语法/命令	用途	代码示例	注意事项
deb 包安装（Debian/Ubuntu）	`sudo dpkg -i filebeat-<version>.deb`	在 Debian 系发行版上安装 Filebeat	`sudo dpkg -i filebeat-8.11.0-amd64.deb`	需提前下载 deb 包，依赖 systemd。
rpm 包安装（RHEL/CentOS）	`sudo rpm -vi filebeat-<version>.rpm`	在 RedHat 系发行版上安装 Filebeat	`sudo rpm -vi filebeat-8.11.0-x86_64.rpm`	使用 yum/dnf 更便于依赖管理。
tar 包解压	`tar -xzf filebeat-<version>-linux-x86_64.tar.gz`	通用安装方式，适用于无包管理的系统	`tar -xzf filebeat-8.11.0-linux-x86_64.tar.gz` `cd filebeat-8.11.0-linux-x86_64`	需手动管理启动脚本和路径。
Docker 安装	`docker pull docker.elastic.co/beats/filebeat:<version>`	在容器环境中运行 Filebeat	`docker pull docker.elastic.co/beats/filebeat:8.11.0`	需挂载日志目录和配置文件。
Docker 运行实例	`docker run -d --name=filebeat` `-v ./filebeat.yml:/usr/share/filebeat/filebeat.yml` `-v /var/log:/var/log` `docker.elastic.co/beats/filebeat:8.11.0`	启动 Filebeat 容器	如上所示	确保容器有权限读取宿主机日志目录。

2.2 目录结构与配置文件说明（filebeat.yml）

项目	说明	注意事项
filebeat.yml	主配置文件，定义 inputs、processors、outputs、path 等	使用 YAML 格式，注意缩进。
modules.d/	存放模块启用/禁用配置文件（如 nginx.yml.disabled）	使用 `filebeat modules enable nginx` 管理。
data/	存储 registry 文件和状态信息	确保目录可写，避免数据丢失。
logs/	Filebeat 自身运行日志输出目录（可选）	启用 `logging.to_files` 后生成。
config/	外部配置文件存放路径（可选）	用于集中管理配置。

filebeat.yml 主要配置项说明：

配置项	说明	示例值	注意事项
filebeat.inputs	定义日志输入源	`- type: log` `paths: /var/log/*.log`	必须至少定义一个输入。
filebeat.modules	启用内置模块	`enabled: true`	模块优先级高于普通 inputs。
output.elasticsearch	配置 ES 输出	`hosts: ["http://localhost:9200"]`	若启用需注释其他 output。
output.logstash	配置 Logstash 输出	`hosts: ["localhost:5044"]`	需 Logstash 配置 beats input。
processors	定义事件处理链	`- drop_event.when.contains.message: "debug"`	在发送前处理事件。
path.data	指定 data 目录路径	`/var/lib/filebeat`	默认为当前目录下的 data。
path.logs	指定日志输出路径	`/var/log/filebeat`	启用日志记录时使用。

2.3 第一个 Filebeat 实例：收集日志并输出到控制台

配置项	语法	用途	代码示例	注意事项
输入类型：log	`type: log`	采集日志文件	`type: log`	新版本推荐使用 filestream。
paths	`paths: ["/path/to/log/*.log"]`	指定日志文件路径	`paths:` `- /tmp/test.log`	支持 glob 模式匹配。
输出到 stdout	`output.console:` `pretty: true` `enable: true`	将事件打印到控制台	`output.console:` `pretty: true`	仅用于调试，生产环境禁用。
禁用其他输出	注释 `output.elasticsearch` 等	避免冲突	`# output.elasticsearch: ...`	确保只有一个 output 启用。
添加字段	`processors:` `- add_fields:` `target: ''` `fields:` `app: myapp`	为事件添加自定义字段	如上所示	可用于标记来源。

完整 filebeat.yml 示例：

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /tmp/test.log

output.console:
  pretty: true
  enable: true

processors:
- add_fields:
    target: ''
    fields:
      source: filebeat-demo

注意事项：

确保 /tmp/test.log 存在并可读。

启动前使用 filebeat test config 验证配置。

控制台输出仅用于测试，生产环境应使用 ES 或 Logstash。

2.4 验证配置与启动服务

命令	语法	用途	代码示例	注意事项
验证配置	`filebeat test config`	检查 filebeat.yml 语法正确性	`filebeat test config`	必须在 Filebeat 目录下执行。
测试输出连接	`filebeat test output`	测试与 ES/Logstash 的连通性	`filebeat test output`	需配置 output 才能测试。
启动（前台）	`./filebeat -e`	前台运行并输出日志到终端	`./filebeat -e`	便于调试，Ctrl+C 退出。
启动（后台）	`sudo service filebeat start`	使用 systemd 启动服务	`sudo systemctl start filebeat`	适用于 deb/rpm 安装方式。
查看状态	`sudo service filebeat status`	检查服务运行状态	`sudo systemctl status filebeat`	确认是否 active (running)。
查看注册状态	`cat data/registry/filebeat`	查看文件读取偏移记录	`cat data/registry/filebeat`	JSON 格式，记录 inode 和 offset。
停止服务	`sudo service filebeat stop`	停止 Filebeat 服务	`sudo systemctl stop filebeat`	停止前会刷新缓冲区。

注意事项：

使用 -e 参数可将日志输出到标准错误，便于观察运行情况。

首次运行时，registry 文件会自动创建。

若采集不到数据，检查文件权限、路径通配符、日志是否更新。

第3章：输入（Inputs）配置详解

3.1 log 输入类型：监控日志文件

方法/参数	语法	用途	代码示例	注意事项
type	`type: log`	指定输入类型为日志文件	`type: log`	已逐步被 filestream 替代。
enabled	`enabled: true/false`	启用或禁用该输入	`enabled: true`	默认为 true。
paths	`paths: ["/path/*.log"]`	指定要监控的日志文件路径	`paths:` `- /var/log/app/*.log`	支持 glob 模式，路径需可读。
exclude_lines	`exclude_lines: ["^DBG", "debug"]`	排除匹配正则的行	`exclude_lines: ["^#"]`	常用于过滤注释或调试日志。
include_lines	`include_lines: ["error", "warn"]`	仅包含匹配正则的行	`include_lines: ["ERROR"]`	与 exclude_lines 可组合使用。
fields	`fields: {app: myapp}`	添加自定义字段到事件中	`fields:` `service: auth-service`	可用于后续过滤或路由。
multiline.pattern	`multiline.pattern: '^\['`	定义多行日志的起始或结束模式	`multiline.pattern: '^[0-9]'`	用于合并 Java 异常堆栈等。
multiline.match	`multiline.match: after\|before`	指定匹配行为：after（前一行）或 before（当前行）	`multiline.match: after`	通常与 pattern 配合使用。
close_eof	`close_eof: true`	文件读取到末尾后关闭	`close_eof: true`	适合一次性日志文件。
scan_frequency	`scan_frequency: 10s`	扫描新文件的频率	`scan_frequency: 5s`	默认 10s，频繁扫描增加 CPU 使用。

完整示例：

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/myapp/*.log
  exclude_lines: ["^DBG"]
  fields:
    service: user-service
  multiline.pattern: '^[0-9]{4}-'
  multiline.match: after

3.2 stdin 输入类型：从标准输入读取数据

方法/参数	语法	用途	代码示例	注意事项
type	`type: stdin`	从标准输入读取数据	`type: stdin`	仅用于测试或管道场景。
enabled	`enabled: true`	启用 stdin 输入	`enabled: true`	默认 true。
stream	`stream: true`	是否持续读取输入流	`stream: true`	设为 false 则读取一行后退出。
add_fields	`fields: {source: cli}`	添加上下文字段	`fields:` `input_type: manual`	便于区分数据来源。

完整示例：

filebeat.inputs:
- type: stdin
  enabled: true
  fields:
    source: console-input

使用方式：

echo "error: login failed" | ./filebeat -e

注意事项：

仅适用于调试或与 shell 脚本集成。

不支持多行处理。

生产环境不推荐使用。

3.3 filestream 输入类型（新版本推荐）

方法/参数	语法	用途	代码示例	注意事项
type	`type: filestream`	新一代日志采集输入类型	`type: filestream`	Elastic 推荐替代 log 类型。
paths	`paths: ["/logs/*.log"]`	指定日志文件路径	`paths:` `- /var/log/nginx/access.log`	支持通配符。
encoding	`encoding: utf-8`	指定文件编码	`encoding: utf-8`	支持 utf-8, utf-16le, ascii 等。
close.on_state_change	`close.on_state_change:` `inactivity: 5m`	基于状态关闭文件	`close.on_state_change:` `inactivity: 3m`	减少文件句柄占用。
prospector.scanner.check_interval	`prospector.scanner.check_interval: 2s`	扫描新文件的间隔	`check_interval: 1s`	默认 10s，可调优。
parsers	`parsers:` `- multiline: ...`	定义解析器（如多行）	`parsers:` `- multiline:` `pattern: '^\['` `match: after`	替代旧版 multiline.* 配置。
fields	`fields: {env: prod}`	添加自定义字段	`fields:` `team: backend`	用于分类和过滤。

完整示例：

filebeat.inputs:
- type: filestream
  paths:
    - /var/log/app/*.log
  encoding: utf-8
  close.on_state_change.inactivity: 3m
  parsers:
    - multiline:
        pattern: '^[0-9]{4}-'
        match: after
  fields:
    service: payment

注意事项：

filestream 是 log 的增强版，性能更好，资源更优。

多行配置使用 parsers 而非顶层 multiline。

支持更多状态管理策略。

3.4 多输入源配置与路径匹配（glob 模式）

方法/参数	语法	用途	代码示例	注意事项
多输入定义	多个 `- type: ...` 块	配置多个独立输入源	`- type: log` `paths: /a/.log` `- type: log` `paths: /b/.log`	每个输入独立运行。
`*`	`*.log`	匹配任意字符（除 `/`）	`paths: /var/log/*.log`	匹配同级 .log 文件。
`**`	`/*/.log`	递归匹配所有子目录	`paths: /var/log/*/.log`	慎用，可能扫描过多文件。
`?`	`app?.log`	匹配单个字符	`app1.log`, `appA.log`	通配单字符。
`[...]`	`[0-9].log`	字符集合匹配	`access[0-9].log`	匹配 access0.log 到 access9.log。
`!`	`!*.tmp`	排除模式	`exclude_files: ['\\.tmp$']`	paths 不支持直接排除，需用 exclude_files。
exclude_files	`exclude_files: ['\\.tmp$']`	排除特定文件	`exclude_files:` `- \\.bak$` `- ~$`	使用正则表达式匹配文件名。

完整示例：

filebeat.inputs:
- type: filestream
  paths:
    - /var/log/app/*.log
    - /var/log/nginx/**/*.log
  exclude_files: ['\\.tmp$', '~$']
  fields:
    source: app-logs

- type: stdin
  enabled: true
  fields:
    source: manual

注意事项：

** 递归扫描可能影响性能，建议限定目录层级。

exclude_files 用于过滤临时文件、备份文件。

多输入时，每个输入可独立配置字段和处理器。

第4章：处理器（Processors）

4.1 Processors 的作用与执行顺序

概念	说明	注意事项
Processors	在事件发送前对其进行处理的组件，用于过滤、增强、转换或解析日志事件。	配置在 processors 列表中。
执行顺序	按 processors 列表中的顺序从上到下依次执行。	建议先执行过滤（如 drop_event），再执行添加字段等操作。
条件执行	每个处理器可使用 when 字段添加条件，仅在条件满足时执行。	支持 equals, contains, regexp, or, and, not 等。
作用范围	可在全局、输入级（input-level）或模块级配置 processors。	输入级配置优先级高于全局。
性能影响	处理器链越长，CPU 开销越高。	避免冗余处理器，提升性能。
内置处理器	Filebeat 提供多种开箱即用的处理器，无需额外依赖。	常见：add_fields, drop_event, decode_json_fields 等。

4.2 常用处理器：drop_event、add_fields、decode_json_fields 等

处理器名称	语法	用途	代码示例	注意事项
drop_event	`drop_event:` `when: <condition>`	根据条件丢弃事件，减少传输和存储压力。	`processors:` `- drop_event:` `when:` `contains:` `message: "DEBUG"`	一旦事件被丢弃，后续处理器不再执行。
add_fields	`add_fields:` `target: ''` `fields: {key: value}`	添加静态字段到事件中，用于标记来源、环境等。	`add_fields:` `target: ''` `fields:` `team: backend` `region: cn-north-1`	target 指定字段插入位置，” 表示根对象。
decode_json_fields	`decode_json_fields:` `fields: ["field_name"]` `target: ""` `process_array: false`	将 JSON 字符串字段解析为结构化对象。	`decode_json_fields:` `fields: ["message"]` `target: "json"`	适用于日志内容为 JSON 的场景，避免 Grok 解析。
convert	`convert:` `fields:` `- {field: "field_name", type: "type"}`	转换字段数据类型（如 string → integer）。	`convert:` `fields:` `- {field: "status", type: "integer"}` `- {field: "size", type: "float"}`	支持类型：string, integer, float, bool, auto。
rename	`rename:` `fields:` `- {from: "old", to: "new"}`	重命名字段，避免命名冲突或标准化字段名。	`rename:` `fields:` `- {from: "msg", to: "message"}`	若目标字段已存在，会覆盖。
truncate_fields	`truncate_fields:` `max_bytes: 1024` `fields: ["message"]`	截断过长字段，防止 Elasticsearch 写入失败。	`truncate_fields:` `max_bytes: 512` `fields: ["message"]`	max_bytes 按字节计算，UTF-8 中中文占 3 字节。
dissect	`dissect:` `tokenizer: "pattern"`	使用分隔符模式快速解析日志，比 Grok 更轻量。	`dissect:` `tokenizer: '%{ip} %{user} %{ts} %{method} %{url}'`	适用于格式固定的日志，不支持正则。

完整示例：

processors:
  - add_fields:
      target: ''
      fields:
        env: production
        source: filebeat
  - decode_json_fields:
      fields: ["message"]
      target: "json_payload"
  - convert:
      fields:
        - {field: "json_payload.status", type: "integer"}
  - drop_event:
      when:
        contains:
          message: "healthcheck"

4.3 条件处理器与嵌套处理

方法/参数	语法	用途	代码示例	注意事项
when.equals	`when:` `equals:` `field: value`	字段值完全匹配时执行处理器。	`when:` `equals:` `fields.service: nginx`	区分大小写，适用于精确匹配。
when.contains	`when:` `contains:` `field: substring`	字段值包含指定子串时执行。	`when:` `contains:` `message: "error"`	不支持正则，性能优于 regexp。
when.regexp	`when:` `regexp:` `field: 'pattern'`	使用正则表达式匹配字段值。	`when:` `regexp:` `message: '(fatal\|panic)'`	支持 RE2 语法，避免复杂正则影响性能。
when.or	`when:` `or:` `- condition1` `- condition2`	多个条件满足任一时执行。	`when:` `or:` `- contains.message: error` `- equals.level: FATAL`	可嵌套其他条件。
when.and	`when:` `and: [...]`	所有条件都满足时执行。	`when:` `and:` `- equals.env: prod` `- contains.message: timeout`	类似逻辑与操作。
when.not	`when:` `not: <condition>`	条件不满足时执行。	`when:` `not:` `contains.message: debug`	用于排除特定事件。
嵌套 processors	在处理器中定义 processors 列表	实现”if-then”逻辑块处理。	`when: ...` `processors:` `- add_tag: ...`	嵌套的处理器也按顺序执行。

完整示例：

processors:
  - when:
      and:
        - equals:
            fields.env: production
        - regexp:
            message: 'HTTP/1\.1" 5\d{2}'
    drop_event: {}
  - when:
      contains:
        fields.service: auth
    processors:
      - add_fields:
          fields:
            alert_level: high
      - add_tags:
          tags: ["security"]

注意事项：

条件判断支持嵌套字段，如 json_payload.error_code。

when 条件可与 drop_event、add_fields 等任意处理器组合。

嵌套 processors 适用于复杂条件下的批量处理。

第5章：输出（Outputs）配置

5.1 输出到 Elasticsearch

参数	语法	用途	代码示例	注意事项
hosts	`hosts: ["http://ip:9200"]`	指定 Elasticsearch 节点地址	`hosts:` `- http://192.168.1.10:9200`	支持多个节点实现高可用。
index	`index: "filebeat-%{[agent.version]}-%{+yyyy.MM.dd}"`	自定义索引名称	`index: "app-logs-%{+yyyy.MM.dd}"`	避免使用大写、特殊字符。
username / password	`username: "user"` `password: "pass"`	基本认证	`username: elastic` `password: changeme`	生产环境必须启用认证。
api_key	`api_key: "id:key"`	使用 API Key 认证	`api_key: "AhRk...:U1Jk..."`	更安全，支持细粒度权限。
ssl.enabled	`ssl.enabled: true`	启用 TLS 加密	`ssl:` `enabled: true`	生产环境建议启用。
ssl.certificate_authorities	`ssl.certificate_authorities: ["/path/ca.crt"]`	指定 CA 证书路径	`certificate_authorities:` `- /etc/filebeat/ca.crt`	用于验证 ES 证书。
pipeline	`pipeline: "my-pipeline"`	指定 Ingest Pipeline	`pipeline: "nginx-ingest"`	需提前在 ES 中创建 pipeline。
worker	`worker: 2`	每个主机的并发工作线程数	`worker: 4`	根据主机性能调整。
bulk_max_size	`bulk_max_size: 50`	每次发送的最大事件数	`bulk_max_size: 100`	默认 50，增大可提升吞吐。

完整示例：

output.elasticsearch:
  hosts: ["https://es1:9200", "https://es2:9200"]
  index: "logs-%{+yyyy.MM.dd}"
  username: "filebeat_internal"
  password: "strongpass"
  ssl.enabled: true
  ssl.certificate_authorities: ["/etc/filebeat/ca.crt"]
  pipeline: "log-parse-pipeline"

5.2 输出到 Logstash

参数	语法	用途	代码示例	注意事项
hosts	`hosts: ["ip:port"]`	指定 Logstash 服务器地址	`hosts:` `- localhost:5044`	Logstash 需配置 beats input。
ssl.enabled	`ssl.enabled: true`	启用 TLS 加密	`ssl:` `enabled: true`	建议生产环境启用。
ssl.certificate_authorities	`ssl.certificate_authorities: [...]`	指定 CA 证书	`certificate_authorities:` `- /etc/filebeat/logstash-ca.crt`	确保 Logstash 证书可信。
timeout	`timeout: 30`	连接超时时间（秒）	`timeout: 60`	网络不稳定时可适当调大。
worker	`worker: 2`	并发连接数	`worker: 3`	与 Logstash input 线程匹配。
loadbalance	`loadbalance: true`	多主机时启用负载均衡	`loadbalance: true`	避免单点压力过大。
index	`index: "custom-index"`	指定发送到 Logstash 的索引名	`index: "app-logs"`	Logstash 可再次修改。

完整示例：

output.logstash:
  hosts: ["logstash1:5044", "logstash2:5044"]
  loadbalance: true
  worker: 2
  ssl.enabled: true
  ssl.certificate_authorities: ["/etc/filebeat/logstash-ca.crt"]
  index: "filebeat-input"

5.3 输出到 Kafka、Redis、File 等其他目标

输出目标	参数/语法	用途	代码示例	注意事项
Kafka `output.kafka`	`hosts: ["kafka:9092"]` `topic: "logs"`	将日志发送到 Kafka 主题	`output.kafka:` `hosts: ["kafka1:9092"]` `topic: "app-logs"` `partition.round_robin:` `reachable_only: true`	需 Kafka 集群可用，支持 SSL、SASL。
Redis `output.redis`	`hosts: ["redis:6379"]` `key: "filebeat"`	输出到 Redis List 或 Channel	`output.redis:` `hosts: ["redis:6379"]` `key: "logs"` `db: 0`	适用于缓冲，需 Logstash 消费。
File `output.file`	`path: "/tmp"` `filename: "beat.log"`	输出到本地文件（调试用）	`output.file:` `path: /tmp` `filename: filebeat.log` `rotate_every_kb: 10000`	仅用于测试，不支持生产。
Console `output.console`	`pretty: true` `enable: true`	输出到控制台（调试）	`output.console:` `pretty: true`	便于查看结构化事件。

Kafka 完整示例：

output.kafka:
  hosts: ["kafka1:9092", "kafka2:9092"]
  topic: "filebeat-logs"
  partition.round_robin:
    reachable_only: true
  ssl.enabled: true
  ssl.certificate_authorities: ["/etc/filebeat/kafka-ca.crt"]

注意事项：

Kafka 和 Redis 适合作为中间缓冲层。

file 和 console 输出仅用于调试。

所有输出均支持 ssl、timeout、worker 等通用参数。

5.4 多输出配置与负载均衡

方法/参数	语法	用途	代码示例	注意事项
selectors	`selectors: [elasticsearch]`	选择启用的输出（只能启用一个）	`output.selectors:` `- elasticsearch`	Filebeat 不支持多输出同时启用。
条件输出	结合 processors 和 drop_event	实现逻辑上的”多输出路由”	使用不同 Filebeat 实例或 Logstash 路由	原生不支持，需架构设计。
负载均衡（ES/Kafka）	`loadbalance: true`	在多个主机间分发数据	`loadbalance: true` `hosts: [h1,h2,h3]`	适用于 Elasticsearch 和 Logstash。
worker 并发	`worker: N`	提升单输出并发能力	`worker: 4`	根据下游性能调优。
输出优先级	无	Filebeat 只允许一个 output 启用	注释其他 output 配置	配置文件中只能有一个 output.* 活跃。

说明：

Filebeat 不支持同时启用多个输出（如同时输出到 ES 和 Kafka）。

若需多目的地，推荐方案：

输出到 Kafka，再由 Logstash 分发到多个系统。

使用多个 Filebeat 实例，各自配置不同输出。

负载均衡示例（Logstash）：

output.logstash:
  hosts: ["logstash1:5044", "logstash2:5044"]
  loadbalance: true
  worker: 2

第6章：模块化配置（Modules）

6.1 Filebeat 模块的作用与优势

概念	说明	注意事项
模块（Module）	预定义的输入、处理器、Ingest Pipeline 和 Kibana 仪表板配置集合，用于快速采集特定服务日志。	如 nginx、system、mysql、apache 等。
快速部署	无需手动编写复杂的日志路径、解析规则和字段映射。	一键启用，降低配置错误风险。
标准化字段	使用 ECS（Elastic Common Schema）规范字段命名。	便于跨服务日志关联分析。
内置解析	包含 Grok 模式、JSON 解析、多行处理等预设处理器。	自动处理常见日志格式。
Kibana 集成	自动加载对应的可视化仪表板和索引模板。	启用后可在 Kibana 中直接查看。
可覆盖配置	支持在 filebeat.yml 或命令行中覆盖模块默认值。	如修改日志路径、索引名等。

6.2 启用内置模块（如 nginx、system、mysql）

命令/参数	语法	用途	代码示例	注意事项
filebeat modules list	`filebeat modules list`	查看所有可用模块及其状态	`filebeat modules list`	显示 enabled/disabled 状态。
filebeat modules enable	`filebeat modules enable <module>`	启用一个或多个模块	`filebeat modules enable nginx system`	模块配置文件位于 modules.d/。
filebeat modules disable	`filebeat modules disable <module>`	禁用已启用的模块	`filebeat modules disable mysql`	配置文件后缀变为 .yml.disabled。
filebeat setup	`filebeat setup`	加载模块所需的索引模板、Ingest Pipeline 和 Kibana 仪表板	`filebeat setup --dashboards`	需在启用模块后执行。
var.paths	`var.paths: ["/custom/log/*.log"]`	覆盖模块默认日志路径	在 `modules.d/nginx.yml` 中设置： `var.paths:` `- /data/logs/nginx/*.log`	每个模块支持特定变量。
var.reload.enabled	`var.reload.enabled: true`	启用模块配置热重载	`var.reload.enabled: true` `var.reload.period: 10s`	避免重启 Filebeat。

完整操作流程示例（启用 Nginx 模块）：

# 1. 查看模块状态
filebeat modules list

# 2. 启用 nginx 和 system 模块
filebeat modules enable nginx system

# 3. （可选）修改 modules.d/nginx.yml 中的 paths
# var.paths: ["/var/log/nginx/access.log"]

# 4. 加载模板和仪表板
filebeat setup

# 5. 启动 Filebeat
./filebeat -e

注意事项：

filebeat setup 需要能访问 Elasticsearch 和 Kibana。

模块默认路径可能因操作系统而异，需根据实际日志位置调整。

可通过 filebeat help modules 查看命令帮助。

6.3 自定义模块开发与部署

步骤/参数	语法/说明	用途	代码示例	注意事项
filebeat modules create	`filebeat modules create <module_name>`	创建自定义模块骨架	`filebeat modules create myapp`	生成 modules.d/myapp.yml 和 module/myapp/ 目录。
module/myapp/log/config	YAML 配置文件	定义输入、处理器等	见下方示例	支持多变体（如 access、error）。
ingest-pipeline.yml	Pipeline 定义文件	定义 ES Ingest Pipeline	`processor:` `- grok:` `patterns: ["%{COMMONAPACHELOG}"]`	用于结构化解析。
fields.yml	字段定义文件	声明自定义字段及其类型	`- name: app_id` `type: keyword`	用于生成索引模板。
dashboard	JSON 仪表板文件	导入 Kibana 可视化	通过 Kibana 导出后放入 dashboard/	支持 .json 格式。
filebeat setup	`filebeat setup`	部署自定义模块	启用模块后运行 setup	自动加载 pipeline、模板和仪表板。

自定义模块配置示例（module/myapp/log/config.yml）：

- type: log
  paths: ${paths}
  fields:
    app: myapp
  processors:
    - dissect:
        tokenizer: '%{timestamp} %{level} %{msg}'

module/myapp/_meta/fields.yml 示例：

- name: app_id
  type: keyword
  description: "Application instance ID"

注意事项：

自定义模块需符合 Filebeat 模块目录结构。

filebeat setup 会自动处理 fields.yml 生成索引模板。

建议先在测试环境验证模块功能。

第7章：高级配置与性能调优

7.1 读取大文件与性能相关参数（close_*、scan_frequency）

参数	语法	用途	代码示例	注意事项
close_eof	`close_eof: true`	文件读到末尾后关闭	`close_eof: true`	适合一次性写入的日志文件。
close_inactive	`close_inactive: 5m`	文件在指定时间内无变化则关闭	`close_inactive: 10m`	减少文件句柄占用，推荐设置。
close_removed	`close_removed: true`	文件被删除后关闭 Harvester	`close_removed: true`	默认启用，避免 inode 复用问题。
close_renamed	`close_renamed: true`	文件被重命名后关闭	`close_renamed: true`	配合日志轮转使用。
close_timeout	`close_timeout: 30m`	文件打开最长持续时间	`close_timeout: 1h`	即使有新内容也强制关闭。
scan_frequency	`scan_frequency: 10s`	Prospector 扫描新文件的间隔	`scan_frequency: 5s`	频繁扫描增加 CPU，可调大。
tail_files	`tail_files: true`	从文件末尾开始读取（而非开头）	`tail_files: true`	避免首次采集全量日志。

完整示例：

filebeat.inputs:
- type: filestream
  paths:
    - /var/log/app/*.log
  close:
    inactive: 5m
    removed: true
    renamed: true
    eof: true
    timeout: 30m
  scan_frequency: 10s
  tail_files: true

注意事项：

close_inactive 是关键性能参数，建议设置为 5-10 分钟。

tail_files: true 可避免重启时重读历史日志。

对于高频写入的大文件，适当调大 close_timeout。

7.2 内存与队列管理（queue、mem pool）

参数	语法	用途	代码示例	注意事项
queue.mem.events	`queue.mem.events: 4096`	内存队列最大事件数	`queue.mem.events: 8192`	默认 4096，增大可缓冲突发流量。
queue.mem.flush.min_events	`flush.min_events: 2048`	触发 flush 的最小事件数	`flush.min_events: 1024`	结合 flush.timeout 使用。
queue.mem.flush.timeout	`flush.timeout: 5s`	最大等待时间后强制 flush	`flush.timeout: 1s`	降低延迟，但增加 I/O。
max_procs	`max_procs: 2`	限制 Go 运行时使用的 CPU 核心数	`max_procs: 1`	在多实例部署时控制资源占用。
path.data	`path.data: /opt/filebeat/data`	指定 data 目录位置	`path.data: /ssd/filebeat/data`	建议使用 SSD 提升 registry 性能。
filebeat.spool_size	`filebeat.spool_size: 2048`	Harvester 到 spooler 的缓冲区大小	`spool_size: 4096`	旧版本参数，新版本建议用 queue。

队列配置示例（内存队列）：

queue.mem:
  events: 8192
  flush.min_events: 1024
  flush.timeout: 5s

注意事项：

内存队列（mem）是默认队列类型，简单高效。

大流量场景可考虑 disk 队列（需额外配置）。

flush.timeout 设置过短可能导致频繁 I/O。

7.3 TLS/SSL 与身份认证配置

参数	语法	用途	代码示例	注意事项
ssl.enabled	`ssl.enabled: true`	启用 TLS 加密	`ssl.enabled: true`	所有输出（ES、Logstash、Kafka）均支持。
ssl.certificate_authorities	`ssl.certificate_authorities: ["/path/ca.crt"]`	指定 CA 证书路径	`certificate_authorities:` `- /etc/filebeat/ca.crt`	用于验证服务端证书。
ssl.certificate / key	`ssl.certificate: "/client.crt"` `ssl.key: "/client.key"`	启用客户端证书认证	`certificate: /certs/client.crt` `key: /certs/client.key`	双向认证场景使用。
ssl.verification_mode	`ssl.verification_mode: full`	证书验证模式	可选：full, strict, none	生产环境禁用 none。
username / password	`username: "user"` `password: "pass"`	基本认证	用于 ES、Logstash 等	密码建议通过 secrets 或环境变量管理。
api_key	`api_key: "id:key"`	使用 API Key 认证	`api_key: "AhRk...:U1Jk..."`	更安全，支持细粒度权限。

ES 输出 TLS 示例：

output.elasticsearch:
  hosts: ["https://es1:9200"]
  username: filebeat
  password: secret
  ssl.enabled: true
  ssl.verification_mode: full
  ssl.certificate_authorities: ["/etc/filebeat/ca.crt"]

注意事项：

生产环境必须启用 TLS 和认证。

避免在配置文件中明文存储密码，可使用 keystore。

客户端证书需被服务端 CA 签名。

7.4 故障恢复与断点续传机制

机制	说明	注意事项
Registry 文件	Filebeat 使用 data/registry 文件记录每个日志文件的 source（路径）、offset（偏移量）、File State OS（inode、device）	唯一标识文件，确保重启后从断点继续读取。
原子写入	registry 更新与事件发送使用 ACK 机制，确保至少一次传输	发送成功后才更新 offset。
文件重命名处理	当日志轮转（如 logrotate）发生时，Filebeat 检测到文件名变化或 inode 变化，关闭旧文件，打开新文件	需配置 `close_renamed: true`。
inode 复用防护	结合 device ID 和 inode 识别文件，避免不同文件 inode 复用导致错乱	依赖操作系统文件系统特性。
重复事件防护	在网络中断后重连，可能重发部分事件（at-least-once）	下游系统需支持幂等处理。
手动重置	删除 data/registry 文件可强制重新采集所有日志	仅用于调试或数据修复。

注意事项：

确保 data/ 目录持久化，避免 registry 丢失。

不要手动修改 registry 文件。

若文件被删除后重建（相同路径），Filebeat 会当作新文件处理，可能重复采集。

第8章：日志解析与结构化（Ingest Pipeline 集成）

8.1 使用 Elasticsearch Ingest Pipeline 预处理数据

概念/参数	语法	用途	代码示例	注意事项
Ingest Pipeline	一组处理器（Processors）的集合，在文档索引前执行转换操作。	在 ES 中定义并命名 pipeline。	`PUT _ingest/pipeline/nginx-pipeline`	可在 Kibana Dev Tools 中创建。
pipeline 参数	`pipeline: "pipeline-name"`	指定 Filebeat 发送事件时使用的 pipeline。	`output.elasticsearch:` `pipeline: "app-pipeline"`	需确保 pipeline 已存在。
常用处理器	grok, date, rename, remove, set 等	实现日志解析、字段转换、时间提取等。	`processors:` `- grok: {patterns: ["%{IP:client}"]}` `- date: {field: "ts", formats: ["ISO8601"]}`	支持 30+ 内置处理器。
条件执行	`if condition { ... }`	根据条件决定是否执行某处理器。	`if ctx.message.contains("error") {` `set: {field: "level", value: "ERROR"}` `}`	使用 Painless 脚本语法。
失败处理	`on_failure`	定义处理器执行失败后的处理逻辑。	`on_failure:` `- set: {field: "error", value: "parse_failed"}`	避免整个 pipeline 中断。
查看 pipeline	`GET _ingest/pipeline/<name>`	检查 pipeline 配置是否正确	`GET _ingest/pipeline/nginx-ingest`	用于调试和验证。

完整 pipeline 示例（解析 Nginx 日志）：

PUT _ingest/pipeline/nginx-ingest
{
  "description": "Parse Nginx access logs",
  "processors": [
    {
      "grok": {
        "field": "message",
        "patterns": ["%{IPORHOST:clientip} %{USER:ident} %{USER:auth} \\[%{HTTPDATE:timestamp}\\] \"%{WORD:method} %{DATA:url} HTTP/%{NUMBER:httpversion}\" %{NUMBER:response} (?:%{NUMBER:bytes}|-)"]
      },
      "on_failure": [
        { "set": { "field": "error", "value": "grok_failed" } }
      ]
    },
    {
      "date": {
        "field": "timestamp",
        "formats": ["dd/MMM/yyyy:HH:mm:ss Z"],
        "target_field": "@timestamp"
      }
    },
    {
      "remove": {
        "field": "timestamp"
      }
    }
  ]
}

注意事项：

pipeline 必须在 Filebeat 启用前创建。

错误处理可防止解析失败导致数据丢失。

复杂 pipeline 可能增加 ES 节点 CPU 负载。

8.2 配合 Grok 模式解析非结构化日志

概念/参数	语法	用途	代码示例	注意事项
Grok 模式	`%{SYNTAX:semantic}`	匹配文本并提取命名字段	`%{IP:client_ip} %{WORD:method}`	SYNTAX 是预定义模式名。
常用模式	IP, WORD, DATA, NUMBER, TIMESTAMP_ISO8601 等	快速构建解析规则	`patterns: ["%{IP:src} %{NUMBER:port}"]`	可组合使用。
自定义模式	在 pipeline 中定义 pattern_definitions	扩展 Grok 模式库	`pattern_definitions: {` `"MYAPP_LOG": "%{TIMESTAMP_ISO8601:ts} %{LOGLEVEL:level} %{GREEDYDATA:msg}"` `}`	避免重复编写复杂模式。
多模式匹配	`patterns: ["p1", "p2"]`	尝试多个模式直到匹配成功	`patterns: [` `"%{COMMONAPACHELOG}",` `"%{COMBINEDAPACHELOG}"` `]`	提高解析容错性。
性能优化	避免过度使用 GREEDYDATA	减少回溯，提升性能	用 DATA 替代 GREEDYDATA 若长度可控	GREEDYDATA 可能导致性能瓶颈。
调试工具	Grok Debugger（Kibana）	测试 Grok 模式是否匹配	在 Kibana > Dev Tools > Grok Debugger 中测试	推荐开发阶段使用。

Grok 示例（结合 pipeline）：

{
  "grok": {
    "field": "message",
    "patterns": [
      "%{IPORHOST:clientip} \\[%{HTTPDATE:timestamp}\\] %{QS:referrer} %{QS:agent}"
    ],
    "pattern_definitions": {
      "HTTPDATE": "%{MONTHNUM}[/-]%{MONTHDAY}[/-]%{YEAR}:%{TIME} %{ISO8601_TIMEZONE}"
    }
  }
}

注意事项：

Grok 是正则的封装，性能低于 dissect。

对于固定分隔符日志，优先使用 dissect 或 csv 处理器。

复杂 Grok 模式建议在测试环境充分验证。

8.3 Filebeat 与 Pipeline 的联动配置

配置项	语法	用途	代码示例	注意事项
output.elasticsearch.pipeline	`pipeline: "my-pipeline"`	指定全局 pipeline	`output.elasticsearch:` `hosts: ["es:9200"]` `pipeline: "log-parse"`	所有事件走同一 pipeline。
按模块指定 pipeline	在模块配置中设置 var.pipeline	不同模块使用不同 pipeline	在 `modules.d/nginx.yml` 中： `var.pipeline: "nginx-ingest"`	更灵活的路由方式。
条件路由（间接）	结合 fields 和 pipeline 的 if 条件	实现逻辑上的多 pipeline 路由	在 pipeline 中： `if ctx.fields.service == 'nginx'`	Filebeat 本身不支持动态 pipeline。
pipeline 版本管理	使用不同 pipeline 名称（如 app-v1, app-v2）	支持灰度发布和回滚	`pipeline: "app-v2"`	避免直接修改生产 pipeline。
验证 pipeline 存在	`filebeat setup` 自动加载	确保 pipeline 已注册	`filebeat setup --pipelines`	建议在部署流程中包含。
错误处理	在 pipeline 中设置 on_failure	记录解析失败事件	`on_failure:` `- set: {field: "_ingest.failure", value: "parse_error"}`	便于后续排查。

模块级 pipeline 配置示例：

# modules.d/myapp.yml
- module: myapp
  log:
    enabled: true
    var.paths: ["/logs/myapp/*.log"]
    var.pipeline: "myapp-ingest-pipeline"  # 指定专用 pipeline

注意事项：

Filebeat 不支持为单个事件动态指定 pipeline。

若需复杂路由，建议输出到 Kafka，由 Logstash 根据内容路由到不同 pipeline。

filebeat setup 会自动加载模块自带的 pipeline。

第9章：监控与运维

9.1 启用 Filebeat 自身监控（metrics、logs）

配置项	语法	用途	代码示例	注意事项
monitoring.enabled	`monitoring.enabled: true`	启用内部指标采集	`monitoring.enabled: true`	可输出到 ES 或 Logstash。
monitoring.elasticsearch	`monitoring.elasticsearch: true`	将监控数据发送到 ES	`monitoring.elasticsearch:` `hosts: ["es:9200"]`	需配置认证。
monitoring.log_level	`monitoring.log_level: info`	设置日志级别	可选：error, warning, info, debug	debug 产生大量日志。
logging.to_files	`logging.to_files: true`	启用日志文件输出	`logging.to_files: true` `logging.files:` `path: /var/log/filebeat`	便于长期审计。
logging.metrics.enabled	`logging.metrics.enabled: true`	启用内部指标日志	`logging.metrics.enabled: true` `logging.metrics.period: 30s`	每 30 秒输出一次指标。
xpack.monitoring	`xpack.monitoring.enabled: true`	（旧版本）启用 X-Pack 监控	`xpack.monitoring:` `enabled: true` `elasticsearch: ...`	新版本推荐使用 monitoring.*。

完整监控配置示例：

monitoring.enabled: true
monitoring.elasticsearch:
  hosts: ["https://es1:9200"]
  username: "beats_system"
  password: "secret"
  ssl.certificate_authorities: ["/etc/filebeat/ca.crt"]

logging.to_files: true
logging.files:
  path: /var/log/filebeat
  name: filebeat.log
  keepfiles: 7
  permissions: 0644

logging.metrics.enabled: true
logging.metrics.period: 30s

注意事项：

监控数据包含事件数、CPU、内存、harvester 状态等。

建议将监控数据发送到独立的监控集群。

开启 debug 日志仅用于问题排查。

9.2 使用 Kibana 查看 Filebeat 状态

功能	位置	说明	注意事项
Beats 状态面板	Stack Monitoring > Beats	查看 Filebeat 实例列表、版本、状态、事件吞吐率	需启用 xpack.monitoring 或 monitoring。
指标图表	Beats 面板内	显示 events.total, libbeat.pipeline.events.published, system.cpu.total.pct 等	实时监控性能。
日志查看	Discover	搜索 `agent.type: filebeat` 查看自身日志	用于排查错误信息。
模块仪表板	Dashboards	查看已启用模块的可视化数据（如 Nginx 请求量、状态码）	需执行 `filebeat setup --dashboards`。
数据流管理	Stack Management > Data Streams	查看 Filebeat 生成的索引和数据流	确认数据是否正常写入。
Ingest Node 状态	Stack Monitoring > Ingest	查看 pipeline 处理速率和失败数	定位解析瓶颈。

操作建议：

定期检查 Beats 状态，确保所有实例在线。

使用 Discover 搜索 error 或 failed 关键词排查问题。

通过 Dashboards 验证日志解析是否正确。

9.3 常见问题排查（文件不采集、偏移丢失、性能瓶颈）

问题现象	可能原因	排查方法	解决方案	注意事项
文件未被采集	路径配置错误、权限不足、文件未匹配 glob	1. 检查 paths 是否正确 2. `ls -l` 确认文件可读 3. 使用 `filebeat test config` 验证配置	修正路径、赋权 `chmod 644`	避免使用 root 外用户运行。
重复采集/偏移丢失	data/registry 文件被删除或损坏	检查 data/registry 是否存在且可读	恢复备份或接受重采	不要手动删除 registry。
CPU 使用率高	扫描频率过高、Grok 模式复杂、多行处理	1. 检查 scan_frequency 2. 优化 Grok 模式 3. 减少 filestream 输入数	调大 scan_frequency，改用 dissect	使用 top 或 htop 监控。
内存占用过高	事件积压、队列过大、大量文件打开	查看 queue.mem.events 设置和 registry 条目数	减小 queue.mem.events，优化 close_* 参数	建议使用 SSD 存储 data/ 目录。
输出失败（ES/Kafka）	网络不通、认证失败、TLS 证书错误	1. ping/telnet 测试连通性 2. 检查用户名/密码 3. 验证证书有效期	修复网络、更新证书、检查配置	使用 `filebeat test output` 测试。
日志解析失败	Grok 模式不匹配、字段名错误	1. 查看 pipeline on_failure 2. 使用 Grok Debugger 测试	调整模式、添加 on_failure 处理	在测试环境充分验证。
Harvester 卡住	大文件写入慢、inode 复用	查看 filebeat.status 指标	优化 close_timeout，避免频繁轮转	监控 filebeat.harvester.running 指标。

常用诊断命令：

# 测试配置文件
filebeat test config

# 测试输出连接
filebeat test output

# 启用 debug 日志
./filebeat -e -d "*"

# 查看当前状态
curl http://localhost:5066/stats

注意事项：

生产环境避免长期开启 debug 日志。

建立定期巡检机制，关注关键指标。

对于关键系统，建议部署高可用 Filebeat 架构。

第10章：安全与权限管理

10.1 配置安全输出（用户名/密码、API Key、证书）

认证方式	语法	用途	代码示例	注意事项
用户名/密码（基本认证）	`username: "user"` `password: "pass"`	用于 Elasticsearch、Logstash、Kafka 等支持基本认证的服务	`output.elasticsearch:` `hosts: ["https://es:9200"]` `username: "filebeat_internal"` `password: "strong_password"`	密码不应明文存储，建议使用 keystore。
API Key	`api_key: "id:key"`	更安全的认证方式，支持细粒度权限和自动轮换	`api_key: "V1RfRk9vYmFyOmFhYmJjY2NkZA=="`	需在 Elasticsearch 中预先创建 API Key。
客户端证书（mTLS）	`ssl.certificate: "/client.crt"` `ssl.key: "/client.key"`	双向 TLS 认证，增强安全性	`ssl:` `enabled: true` `certificate: "/etc/filebeat/client.crt"` `key: "/etc/filebeat/client.key"`	证书需被服务端 CA 签名，适用于高安全要求环境。
Keystore 管理敏感信息	`filebeat keystore create` `filebeat keystore add ES_PASSWORD`	将密码、API Key 等存入加密 keystore	`filebeat keystore add ES_PASSWORD` `filebeat keystore list`	在配置中使用 `${ES_PASSWORD}`
			避免配置文件泄露导致凭据暴露。

完整安全输出示例（Elasticsearch）：

output.elasticsearch:
  hosts: ["https://es1:9200", "https://es2:9200"]
  api_key: "V1RfRk9vYmFyOmFhYmJjY2NkZA=="
  ssl.enabled: true
  ssl.certificate_authorities: ["/etc/filebeat/ca.crt"]
  ssl.verification_mode: full

注意事项：

生产环境必须启用认证，禁用匿名访问。

API Key 优于用户名/密码，支持更细粒度权限。

使用 filebeat keystore 管理敏感字段。

10.2 文件权限与采集用户建议

安全实践	说明	建议配置	注意事项
使用专用用户运行	避免使用 root，降低权限滥用风险	创建 filebeat 用户： `useradd -r -s /sbin/nologin filebeat`	确保该用户仅拥有必要权限。
日志文件权限	Filebeat 用户需对日志文件有读权限	`chmod 644 /var/log/app/*.log` `chown app:filebeat /var/log/app/`	避免 `chmod 777`。
配置文件权限	防止配置泄露（含密码、路径）	`chmod 600 /etc/filebeat/filebeat.yml` `chown root:filebeat /etc/filebeat/filebeat.yml`	仅 root 和 filebeat 组可读。
data 目录权限	registry 文件包含文件偏移信息	`chown filebeat:filebeat /var/lib/filebeat`	确保 Filebeat 用户可读写。
最小权限原则	仅授予必要目录的读取权限	在 filebeat.yml 中精确指定 paths	避免使用 `/var/log/*` 全局路径。

权限配置示例：

# 创建专用用户
useradd -r -s /sbin/nologin filebeat

# 设置日志目录权限
chown -R app:filebeat /var/log/myapp
chmod 750 /var/log/myapp
chmod 644 /var/log/myapp/*.log

# 设置 Filebeat 配置权限
chown root:filebeat /etc/filebeat/filebeat.yml
chmod 640 /etc/filebeat/filebeat.yml

10.3 启用加密通信（TLS/SSL）

参数	语法	用途	代码示例	注意事项
ssl.enabled	`ssl.enabled: true`	启用 TLS 加密通信	`ssl.enabled: true`	所有输出均支持。
ssl.certificate_authorities	`ssl.certificate_authorities: ["/path/ca.crt"]`	指定 CA 证书，用于验证服务端身份	`certificate_authorities: ["/etc/filebeat/ca.crt"]`	必须信任服务端证书颁发者。
ssl.certificate / key	`ssl.certificate: "client.crt"` `ssl.key: "client.key"`	启用客户端证书认证（mTLS）	配合服务端配置使用	证书需由服务端信任的 CA 签发。
ssl.verification_mode	`ssl.verification_mode: full`	控制证书验证严格程度	可选： `full`（验证主机名+证书） `certificate`（仅验证证书） `none`（不验证，禁止生产使用）	生产环境必须使用 full。
ssl.supported_protocols	`ssl.supported_protocols: [TLSv1.2, TLSv1.3]`	指定支持的 TLS 协议版本	禁用 TLSv1.0/1.1	符合安全合规要求。
ssl.cipher_suites	`ssl.cipher_suites: ["TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256"]`	指定加密套件	使用现代、安全的套件	避免使用弱加密算法。

TLS 完整配置示例：

output.elasticsearch:
  hosts: ["https://es1:9200"]
  ssl.enabled: true
  ssl.verification_mode: full
  ssl.supported_protocols: [TLSv1.2, TLSv1.3]
  ssl.certificate_authorities: ["/etc/filebeat/ca.crt"]
  ssl.certificate: "/etc/filebeat/client.crt"
  ssl.key: "/etc/filebeat/client.key"

注意事项：

禁止在生产环境使用 ssl.verification_mode: none。

定期更新 CA 证书和客户端证书。

使用工具（如 openssl s_client -connect es:9200）测试 TLS 连接。

第11章：实战案例

11.1 收集 Nginx 日志并导入 Elasticsearch

步骤	操作	说明	注意事项
1. 启用 Nginx 模块	`filebeat modules enable nginx`	自动配置输入、处理器、pipeline	确保日志路径正确（默认 `/var/log/nginx/*.log`）。
2. 修改日志路径（可选）	编辑 `modules.d/nginx.yml`	自定义 `var.paths`	`var.paths:` `- /data/logs/nginx/*.log`
3. 加载 ES 资源	`filebeat setup`	创建索引模板、Ingest Pipeline、Kibana 仪表板	需能访问 Elasticsearch 和 Kibana。
4. 配置输出	编辑 `filebeat.yml`	指定 ES 地址、认证	`output.elasticsearch:` `hosts: ["https://es:9200"]` `username: "filebeat"` `password: "${ES_PASS}"`
5. 启动 Filebeat	`./filebeat -e`	启动并查看日志	使用 `systemctl start filebeat` 用于生产。

验证：

在 Kibana Discover 中搜索 fileset.module: nginx。
查看 Dashboards 中 “Nginx” 仪表板是否正常显示数据。

11.2 多行日志处理（如 Java 异常堆栈）

配置项	语法	用途	代码示例	注意事项
`multiline.type`	`multiline.type: pattern`	指定多行匹配方式	`multiline.type: pattern`	filestream 输入使用 `multiline.*`。
`multiline.pattern`	`multiline.pattern: '^\s'`	正则匹配续行开头	`multiline.pattern: '^\s+at '`	Java 堆栈通常以 `at` 开头。
`multiline.match`	`multiline.match: after`	如何处理匹配行	`after`（附加到上一行）或 `before`（合并到下一行）	Java 日志用 `after`。
`multiline.negate`	`multiline.negate: false`	是否取反匹配	`negate: true` 表示”不匹配该模式的行是续行”	通常为 `false`。
`close_eof`	`close_eof: false`	避免文件末尾过早关闭	`close_eof: false`	确保完整读取多行块。

Java 日志多行配置示例：

- type: filestream
  paths:
    - /var/log/app/application.log
  multiline:
    type: pattern
    pattern: '^\s*at '
    match: after
    negate: false
  close:
    eof: false
    inactive: 5m

日志示例：

2025-09-30 10:00:00 ERROR [main] com.example.App - Exception occurred
at com.example.Service.process(Service.java:45)
at com.example.Controller.handle(Controller.java:23)

→ 将被合并为一条事件。

注意事项：

多行处理对性能有轻微影响。

确保 pattern 准确，避免错误合并。

11.3 Docker 环境下的 Filebeat 部署

部署方式	配置要点	代码示例（`docker-compose.yml`）	注意事项
共享日志卷	将宿主机日志目录挂载到容器	`volumes:` `- /var/log/app:/logs:ro`	使用 `:ro` 只读挂载。
配置文件挂载	挂载自定义 `filebeat.yml`	`volumes:` `- ./filebeat.docker.yml:/usr/share/filebeat/filebeat.yml:ro`	确保配置适用于容器环境。
data 目录持久化	挂载 `data/` 目录防止 registry 丢失	`volumes:` `- filebeat_data:/usr/share/filebeat/data`	使用命名卷或宿主机路径。
用户与权限	确保容器内用户有读权限	在 Dockerfile 中 `USER filebeat` 或运行时 `--user`	避免权限错误。

完整 docker-compose 示例：

version: '3'
services:
  filebeat:
    image: docker.elastic.co/beats/filebeat:8.11.0
    user: root
    volumes:
      - /var/log/app:/logs:ro
      - ./filebeat.docker.yml:/usr/share/filebeat/filebeat.yml:ro
      - filebeat_data:/usr/share/filebeat/data
    environment:
      - ES_HOST=https://elasticsearch:9200
      - ES_USER=filebeat_internal
      - ES_PASS=${ES_PASS}
    environment_file:
      - ./filebeat.env
    depends_on:
      - elasticsearch

volumes:
  filebeat_data:

filebeat.docker.yml 示例：

filebeat.inputs:
- type: filestream
  paths:
    - /logs/*.log

output.elasticsearch:
  hosts: ["${ES_HOST}"]
  username: "${ES_USER}"
  password: "${ES_PASS}"

注意事项：

使用环境变量注入敏感信息。

确保宿主机日志路径正确挂载。

11.4 Kubernetes 中的 Filebeat DaemonSet 配置

配置要点	说明	YAML 片段	注意事项
DaemonSet	确保每个节点运行一个 Filebeat 实例	`kind: DaemonSet`	用于采集节点级日志。
挂载日志目录	采集容器日志（存储在 `/var/log/containers`）	`volumeMounts:` `- name: varlog` `mountPath: /var/log` `readOnly: true`	Kubernetes 日志位于 `/var/log/pods` 和 `/var/log/containers`。
挂载 data 目录	持久化 registry	使用 `hostPath` 或 `emptyDir`	推荐 `hostPath` 防止重启丢 offset。
RBAC 授权	获取元数据（Pod 名、Label）	创建 ServiceAccount、ClusterRole、ClusterRoleBinding	需要 get、list、watch 权限。
自动发现	动态配置输入（基于 Pod Label）	使用 `filebeat.autodiscover`	实现日志采集自动化。

完整 DaemonSet 示例（精简）：

apiVersion: apps/v1
kind: DaemonSet
metadata:
  name: filebeat
spec:
  selector:
    matchLabels:
      app: filebeat
  template:
    metadata:
      labels:
        app: filebeat
    spec:
      serviceAccountName: filebeat
      containers:
      - name: filebeat
        image: docker.elastic.co/beats/filebeat:8.11.0
        args: ["-c", "/etc/filebeat.yml", "-e"]
        volumeMounts:
        - name: config
          mountPath: /etc/filebeat.yml
          subPath: filebeat.yml
        - name: varlog
          mountPath: /var/log
          readOnly: true
        - name: varlib
          mountPath: /var/lib/filebeat
        - name: dockersock
          mountPath: /var/run/docker.sock
          readOnly: true
      volumes:
      - name: config
        configMap:
          name: filebeat-config
      - name: varlog
        hostPath:
          path: /var/log
      - name: varlib
        hostPath:
          path: /var/lib/filebeat
      - name: dockersock
        hostPath:
          path: /var/run/docker.sock

filebeat.yml（K8s 自动发现）：

filebeat.autodiscover:
  providers:
    - type: kubernetes
      node: ${NODE_NAME}
      hints.enabled: true
      hints.default_config:
        type: container
        paths:
          - /var/log/containers/*${data.kubernetes.pod.name}*.log

processors:
  - add_kubernetes_metadata:
      host: ${NODE_NAME}
      matchers:
        - logs_path: /var/log/containers/

output.elasticsearch:
  hosts: ["https://elasticsearch:9200"]
  username: "filebeat"
  password: "${ES_PASS}"

注意事项：

必须配置 RBAC，否则无法获取 Pod 元数据。

add_kubernetes_metadata 处理器自动添加 kubernetes.* 字段。

使用 hints.enabled: true 可通过 Pod Annotation 配置 Filebeat 行为。