Fluentd日志过滤与结构化处理的核心是精准配置filter插件链：先用grep过滤ERROR/WARN级别并排除DEBUG/TRACE日志，再用record_transformer标准化level字段、注入env/prod等元数据，并通过本地验证和灰度上线确保可靠性。
部署 Fluentd 日志过滤与结构化处理，核心在于配置好 filter 插件链，并确保它能准确接入上游日志源、输出到下游目标。这不是装完就跑的事，关键在规则写得准、逻辑理得清、测试验得实。
明确日志源与目标格式
结构化处理的前提是知道原始日志长什么样、最终要变成什么样。比如 TIS 集群中 Flink 作业日志：
- 原始格式示例：
2026-06-12 14:22:38 [ERROR] TaskManager failed to register: timeout - 目标结构：提取
time（转为 ISO 时间戳）、level（大写标准化）、message（保留原文），并自动添加service: flink和env: prod字段
不先定义这个映射关系，后续所有过滤和转换都容易偏离业务需求。
用 grep 过滤无效日志
这是第一道轻量级防线，建议放在 filter 流水线最前端，减少后续处理压力：
- 只保留 ERROR/WARN 级别：
<filter tis.flink.**><br> @type grep<br> <regexp><br> key level<br> pattern /^(ERROR|WARN)$/br> </regexp><br></filter>
- 排除调试类日志（如含
DEBUG或TRACE）：<exclude><br> key message<br> pattern /DEBUG|TRACE/<br></exclude>
注意：grep 支持 <or> 和 <and> 嵌套，但多条件叠加时建议拆成多个 filter 块，更易定位问题。
用 record_transformer 做字段增强与标准化
完成初步筛选后，用 record_transformer 统一注入元数据、重命名字段、清理空值：
- 添加固定字段：
<filter tis.**><br> @type record_transformer<br> enable_ruby true<br> <record><br> env prod<br> cluster tis-prod<br> </record><br></filter>
- 将 level 转为小写并重命名为
log_level：<record><br> log_level ${record["level"]&.downcase}<br></record> - 删除无意义字段（如原始日志里的冗余前缀）：
remove_keys level, host
ruby 表达式支持简单逻辑，但避免复杂计算；大量转换建议用 parser 插件配合 regexp 提前解析。
验证与上线节奏
别跳过本地验证环节，尤其对正则表达式和字段引用：
- 用
fluentd -c td-agent.conf -v检查配置语法 - 启动时加
--dry-run（部分版本支持）或临时把 output 改为stdout，观察实际输出是否符合预期 - 上线分两步：先灰度 1–2 个节点，确认 CPU/内存无明显上涨、日志无丢失；再批量滚动更新
- 生产环境务必开启 buffer（如
buf_file），防止网络抖动导致日志堆积或丢失
结构化不是一步到位的工程，而是随日志格式演进持续迭代的过程。每次新增服务或调整日志格式，都应同步更新对应 filter 规则。
