Fluent Bit 过滤器深度解析：grep、record_modifier 和 Lua 脚本实战

作为一名 Kubernetes 开发者或运维人员，你肯定对 Fluent Bit 不陌生。它是一个高性能、轻量级的日志收集和处理工具，广泛应用于容器化环境中。Fluent Bit 的强大之处在于其丰富的插件系统，其中 Filter 插件更是赋予了它灵活处理日志的能力。今天，咱们就来深入聊聊 Fluent Bit 的 Filter 插件，特别是 grep、record_modifier 和 lua 这三个常用的过滤器，并通过实战案例让你彻底掌握它们的高级用法。

为什么要用 Filter 插件？

在 Kubernetes 集群中，日志数据来源多样、格式不一。直接将原始日志一股脑地推送到后端存储（如 Elasticsearch、Kafka）不仅会造成资源浪费，还会给后续的日志分析带来困难。Filter 插件的作用就在于对原始日志进行预处理，实现以下目标：

• 数据清洗： 过滤掉无用的日志、敏感信息，保留关键信息。
• 格式转换： 将非结构化日志转换为结构化数据（如 JSON），方便查询和分析。
• 字段增强： 添加额外的字段，如 Kubernetes 元数据（Pod 名称、Namespace 等），丰富日志上下文。
• 自定义处理： 通过脚本实现复杂的日志处理逻辑。

认识 Filter 插件

Fluent Bit 提供了多种 Filter 插件，每个插件都有特定的功能。咱们今天重点关注以下三个：

• grep 过滤器： 基于正则表达式匹配，过滤符合或不符合条件的日志记录。
• record_modifier 过滤器： 修改或添加日志记录中的字段。
• lua 过滤器： 使用 Lua 脚本编写自定义的过滤逻辑。

在 Fluent Bit 的配置文件中，Filter 插件通常定义在 [FILTER] 部分。一个典型的配置示例如下：

[FILTER]
    Name  grep
    Match  kube.*
    Regex  log  error
 
[FILTER]
    Name  record_modifier
    Match  *
    Record  hostname  ${HOSTNAME}
 
[FILTER]
    Name          lua
    Match         app.*
    Script        my_script.lua
    Call          my_filter_function

在这个例子中：

• 第一个 Filter 使用 grep 插件，匹配所有 kube.* 的 Tag，并筛选出 log 字段中包含 error 的日志记录。
• 第二个 Filter 使用 record_modifier 插件，匹配所有 Tag，并添加一个名为 hostname 的字段，值为当前主机名。
• 第三个 Filter 使用 lua 插件, 匹配app.*的tag，并调用my_script.lua中的 my_filter_function函数。

接下来，咱们就来详细剖析这三个 Filter 插件。

grep 过滤器：精准筛选日志

grep 过滤器是最常用的 Filter 插件之一，它通过正则表达式匹配来筛选日志。grep 过滤器支持两种匹配模式：

• Regex： 匹配符合正则表达式的日志记录。
• Exclude： 排除符合正则表达式的日志记录。

常用配置选项

• Name： 必须设置为 grep。
• Match： 指定要匹配的 Tag，支持通配符（* 和 ?）。
• Regex： 指定一个正则表达式，用于匹配日志记录中的某个字段。格式为 field_name regex。
• Exclude： 指定一个正则表达式，用于排除日志记录中的某个字段。格式为 field_name regex。

实战案例

1. 筛选出包含特定错误码的日志：

   [FILTER]
       Name  grep
       Match  *
       Regex  log  ERROR_CODE=500

   这个配置会筛选出所有 log 字段中包含 ERROR_CODE=500 的日志记录。

2. 排除调试日志：

   [FILTER]
       Name  grep
       Match  *
       Exclude  log  DEBUG

   这个配置会排除所有 log 字段中包含 DEBUG 的日志记录。

3. 匹配多个字段：
   你可以使用多个Regex或者Exclude 来实现更复杂的匹配。

    [FILTER]
        Name grep
        Match *
        Regex log error
        Regex level FATAL

   这个配置会筛选出log字段中包含error并且level字段包含FATAL的日志。

record_modifier 过滤器：修改和增强日志记录

record_modifier 过滤器用于修改或添加日志记录中的字段。它可以帮助你标准化日志格式，添加上下文信息。

常用配置选项

• Name： 必须设置为 record_modifier。
• Match： 指定要匹配的 Tag，支持通配符。
• Record： 添加或修改字段。格式为 field_name field_value。field_value 可以是静态值，也可以使用变量。
• Remove_key: 移除某个key，如果不想在日志中保留某个字段。格式为Remove_key field_name

实战案例

1. 添加 Kubernetes 元数据：

   [FILTER]
       Name  record_modifier
       Match  kube.*
       Record  pod_name  ${K8S_POD_NAME}
       Record  namespace  ${K8S_NAMESPACE_NAME}

   这个配置会给所有 kube.* 的 Tag 添加 pod_name 和 namespace 字段，值分别为 Kubernetes Pod 名称和 Namespace 名称。这里使用了 Fluent Bit 内置的 Kubernetes 元数据变量。

2. 修改时间戳格式：

   [FILTER]
       Name  record_modifier
       Match  *
       Record  timestamp  $timestamp
       Remove_key @timestamp

将@timestamp重命名为timestamp

3. 使用变量和静态值组合：

   [FILTER]
        Name record_modifier
        Match *
        Record my_field My Application - ${HOSTNAME}

添加字段my_field, 值为My Application - 加上当前hostname。

lua 过滤器：自定义日志处理逻辑

当内置的 Filter 插件无法满足你的需求时，lua 过滤器就派上用场了。它允许你使用 Lua 脚本编写自定义的过滤逻辑，实现高度灵活的日志处理。

常用配置选项

• Name： 必须设置为 lua。
• Match： 指定要匹配的 Tag，支持通配符。
• Script： 指定 Lua 脚本文件的路径。
• Call： 指定 Lua 脚本中要调用的函数名。该函数必须接收三个参数：tag、timestamp 和 record，并返回一个包含修改后的记录的表。
• Time_As_Table: (Boolean, optional, default: false) 如果设置为 true，则时间戳将作为 Lua 表传递给 Lua 脚本, 否则作为浮点数传递。

实战案例

1. 解析 JSON 格式的日志字段：

   假设你的日志中有一个名为 data 的字段，它包含 JSON 格式的数据。你可以使用 Lua 脚本将其解析为独立的字段。

   my_script.lua：

   function my_filter_function(tag, timestamp, record)
     if record.data then
       local data = cjson.decode(record.data)
       if data then
         for k, v in pairs(data) do
           record[k] = v
         end
         record.data = nil -- 移除原始的 data 字段
       end
     end
     return 1, timestamp, record
   end

   Fluent Bit 配置：

   [FILTER]
       Name   lua
       Match  *
       Script my_script.lua
       Call   my_filter_function

   这个配置会将 data 字段中的 JSON 数据解析为独立的字段，并移除原始的 data 字段。
   返回值必须是code, timestamp, record

   • code: 0 代表修改成功， 1代表无变化， 2代表出错。

2. 更复杂的逻辑判断和数据转换：

    function complex_logic(tag, timestamp, record)
        local new_record = {}
        if record.level == "error" then
           new_record.error_code = tonumber(string.match(record.message, "Error Code: (%d+)"))
           if new_record.error_code and new_record.error_code > 500 then
                new_record.severity = "critical"
           elseif new_record.error_code then
                new_record.severity = "major"
           end
           new_record.original_message = record.message
           return 0, timestamp, new_record
        end
       return 1, timestamp, record
    end

   这个例子中，如果日志等级为error, 则从message中提取error code, 并根据error code 设置severity。

组合使用 Filter 插件

在实际应用中，你通常需要组合使用多个 Filter 插件来实现复杂的日志处理流程。例如：

[FILTER]
    Name  grep
    Match  kube.*
    Exclude  log  DEBUG
 
[FILTER]
    Name  record_modifier
    Match  kube.*
    Record  pod_name  ${K8S_POD_NAME}
    Record  namespace  ${K8S_NAMESPACE_NAME}
 
[FILTER]
    Name          lua
    Match         kube.*
    Script        parse_json.lua
    Call          parse_json_data

这个配置首先使用 grep 过滤器排除调试日志，然后使用 record_modifier 过滤器添加 Kubernetes 元数据，最后使用 lua 过滤器解析 JSON 格式的日志字段。通过合理的组合，你可以构建出强大而灵活的日志处理流水线。

总结

Fluent Bit 的 Filter 插件为日志处理提供了强大的工具。grep、record_modifier 和 lua 这三个过滤器各具特色，grep 负责筛选，record_modifier 负责修改和增强，lua 则提供了最大的灵活性。掌握这三个过滤器的高级用法，并结合实际需求灵活组合，你就能轻松应对 Kubernetes 环境中的各种日志处理挑战。希望这篇文章能帮助你更好地理解和使用 Fluent Bit，让你的日志管理工作更上一层楼！

如果你在使用过程中遇到任何问题，或者有更好的实践经验，欢迎留言交流，咱们一起探讨！