3. Fluentd 核心组件：Input、Parser、Filter、Buffer、Output 插件机制与配置示例

Fluentd 采用“插件化”架构，将日志处理链路拆解为五个核心阶段：Input（输入）、Parser（解析）、Filter（过滤）、Buffer（缓冲）和 Output（输出）。每个阶段对应一类插件，通过配置文件组合，形成从采集到存储的完整管道。理解这五类插件的工作原理与配置方式，是掌握 Fluentd 的关键。

3.1 Input 插件：数据采集的入口

Input 插件负责从各种数据源拉取或接收日志事件。Fluentd 内置了丰富的 Input 插件，覆盖文件、网络、系统日志等场景。最常用的包括：

• in_tail：跟踪文本文件（如 /var/log/nginx/access.log），支持 pos_file 记录读取位置，实现断点续传。
• in_http：启动 HTTP 服务，接收 POST 请求中的日志数据。
• in_forward：监听 TCP/UDP 端口，接收其他 Fluentd 实例转发的事件。
• in_syslog：接收 Syslog 协议（RFC 5424/3164）的日志。

<source>
  @type tail
  path /var/log/nginx/access.log
  pos_file /var/log/td-agent/nginx-access.pos
  tag nginx.access
  <parse>
    @type nginx
  </parse>
</source>

3.2 Parser 插件：非结构化到结构化

Parser 插件将非结构化的原始日志（如文本行）解析为 JSON 格式的事件记录。它通常内嵌在 Input 或 Output 的 <parse> 块中。常见 Parser 类型：

3.3 Filter 插件：事件加工与路由

Filter 插件在 Parser 之后、Buffer 之前执行，用于对事件进行修改、过滤、丰富或路由。Filter 可以串联多个，形成处理链。常用 Filter：

• grep：基于正则表达式或条件过滤事件，仅保留匹配的记录。
• record_transformer：修改或添加事件字段，如添加 @timestamp、hostname。
• geoip：根据 IP 地址添加地理位置信息。
• stdout：将事件输出到标准输出，用于调试。

<filter nginx.access>
  @type grep
  <regexp>
    key status
    pattern ^5\d{2}$
  </regexp>
</filter>

<filter nginx.access>
  @type record_transformer
  <record>
    @timestamp ${time}
    severity ERROR
  </record>
</filter>

3.4 Buffer 插件：可靠性与背压控制

Buffer 是 Output 插件的内部组件，用于暂存待发送的事件数据。它解决了网络抖动、目标端不可用时的数据丢失问题，并提供背压机制。Buffer 配置在 Output 的 <buffer> 块中。

<match nginx.access>
  @type elasticsearch
  host localhost
  port 9200
  <buffer>
    @type file
    path /var/log/td-agent/buffer/elasticsearch
    chunk_limit_size 8MB
    flush_interval 5s
    retry_max_times 10
  </buffer>
</match>

3.5 Output 插件：数据路由与存储

Output 插件是管道的终点，负责将缓冲的事件发送到目标系统。Fluentd 支持超过 500 种 Output 插件，覆盖数据库、消息队列、对象存储、监控系统等。常用 Output：

• elasticsearch：写入 Elasticsearch 集群，支持索引模板、批量写入。
• kafka2：发送到 Apache Kafka 主题，支持分区键和压缩。
• s3：上传到 AWS S3 或兼容对象存储，支持压缩和加密。
• forward：转发到另一个 Fluentd 实例，用于构建多层架构。

<match nginx.access>
  @type kafka2
  brokers kafka1:9092,kafka2:9092
  topic nginx-logs
  <buffer>
    @type file
    path /var/log/td-agent/buffer/kafka
    flush_interval 10s
  </buffer>
</match>

3.6 插件机制总结