WEBKT

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

467 0 0 0

作为一名 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

  1. 使用变量和静态值组合：
    [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，让你的日志管理工作更上一层楼！

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

技术老兵 Fluent BitKubernetes日志处理

评论点评