Fluentbit S3配置：如何实现与FluentD一致的日志路径？

正在将K8S日志从FluentD迁移至FluentBit转存到S3，目前在tag重写和日志写入S3正确路径的配置上遇到问题。

FluentD参考配置（可生成正确路径） #

<record>
#               environment ${record["kubernetes"]["namespace_name"]}
#               pod ${record["kubernetes"]["pod_name"]}
#               podid ${record["kubernetes"]["pod_id"]}
#               container ${record["kubernetes"]["container_name"]}
#             </record>
#           </filter>
#           <match **>
#             @type s3
#             s3_bucket logs.bucket.us-east-1.domain.com
#             s3_region us-east-1
#             s3_object_key_format %Y/%m/%d/${environment}/${container}/${container}-${environment}-%Y%m%d-%H%M-${podid}-%{index}.%{file_extension}
#             store_as text

当前FluentBit配置（路径错误） #

[INPUT]
    Name              tail
    Tag               s3logs.*
    Path              /var/log/containers/*.log
    parser            cri
    multiline.parser  cri         
    Mem_Buf_Limit     5MB
    Skip_Long_Lines   On
    Skip_Empty_Lines  On
    Refresh_Interval  10
[FILTER]
    Name                kubernetes
    Match               s3logs.*
    Merge_Log           On
    K8S-Logging.Parser  On
    K8S-Logging.Exclude On
    Keep_Log            Off
    Labels              Off   
    Annotations         Off         
[FILTER]
    Name    record_modifier
    Match   s3logs.*
    Record  cluster_name ${CLUSTER}
[FILTER]
    name    lua
    alias   set_std_keys
    match   s3logs.*
    script  /fluent-bit/scripts/s3_path.lua
    call    set_std_keys
[FILTER]
    name rewrite_tag
    match s3logs.*
    rule $log ^.*$ s3.$namespace_name.$app_name.$container_name.$pod_id true               

outputs: |
    [OUTPUT]
        Name s3
        Match s3logs.*
        bucket                       logs.bucket.us-east-1.domain.com
        region                       us-east-1
        s3_key_format                /%Y/%m/%d/$TAG[1]/$TAG[2]/$TAG[3]/$TAG[3]-$TAG[1]-%Y%m%d-%H%M-${podid}.txt
        store_dir                    /var/log/fluentbit-s3-buffers
        total_file_size              256MB
        upload_timeout               2m
        use_put_object               On
        preserve_data_ordering       On

Lua元数据提取脚本 #

function set_std_keys(tag, timestamp, record)

    -- Pull up cluster
    if (record["cluster_name"] ~= nil) then
        record["cluster_name"] = record["cluster_name"]
    else
        record["cluster_name"] = "mycluster"
    end

    if (record["kubernetes"] ~= nil) then
        kube = record["kubernetes"]

        -- Pull up namespace
        if (kube["namespace_name"] ~= nil and string.len(kube["namespace_name"]) > 0) then
            record["namespace_name"] = kube["namespace_name"]
        else
            record["namespace_name"] = "default"
        end

        -- Pull up container name
        if (kube["container_name"] ~= nil and string.len(kube["container_name"]) > 0) then
            record["container_name"] = kube["container_name"]
        end

        -- Pull up pod id
        if (kube["pod_id"] ~= nil and string.len(kube["pod_id"]) > 0) then
            record["pod_id"] = kube["pod_id"]
        end

        -- Pull up app name (Deployment, StateFuleSets, DaemonSet, Job, CronJob etc)
        if (kube["labels"] ~= nil) then
            labels = kube["labels"]

            if (labels["app"] ~= nil and string.len(labels["app"]) > 0) then
                record["app_name"] = labels["app"]
            elseif (labels["app.kubernetes.io/instance"] ~= nil and string.len(labels["app.kubernetes.io/instance"]) > 0) then
                record["app_name"] = labels["app.kubernetes.io/instance"]
            elseif (labels["k8s-app"] ~= nil and string.len(labels["k8s-app"]) > 0) then
                record["app_name"] = labels["k8s-app"]
            elseif (labels["name"] ~= nil and string.len(labels["name"]) > 0) then
                record["app_name"] = labels["name"]
            end
        else
            record["app_name"] = record["app_name"]
        end
    end

  return 2, timestamp, record
end                    

路径对比 #

• 正确路径（FluentD生成）：2023/10/14/namespace/container_name/container-name-namespace_name-2023-10-14-UUID.txt
• 错误路径（FluentBit生成）：2023/10/14/var/log/containers/containers-var-20231014-0759-.log-object00N1PX3n

问题根源在于Tag重写规则错误、S3输出未匹配重写后的Tag，以及K8S元数据提取不完整，以下是具体修改步骤：

1. 修复Kubernetes过滤器配置 #

要提取Pod的labels，必须开启Labels On，否则Lua脚本无法获取labels字段，导致app_name为空：

[FILTER]
    Name                kubernetes
    Match               s3logs.*
    Merge_Log           On
    K8S-Logging.Parser  On
    K8S-Logging.Exclude On
    Keep_Log            Off
    Labels              On   # 改为On，确保能获取labels
    Annotations         Off         

2. 修正Rewrite_Tag过滤器规则 #

当前规则依赖$log匹配，且未正确引用record字段，导致Tag重写失败，实际仍使用原始的s3logs./var/log/containers/xxx.log Tag，这是错误路径的来源。修改为：

[FILTER]
    Name rewrite_tag
    Match s3logs.*
    # 匹配所有s3logs开头的Tag，用record中的字段重写为s3.namespace.app.container.podid格式
    Rule $tag ^s3logs\..*$ s3.$record:namespace_name.$record:app_name.$record:container_name.$record:pod_id true

注意：这个过滤器必须放在Lua过滤器之后，确保Lua已将元数据提取到record中。

3. 更新Lua脚本处理空值 #

避免app_name为空导致Tag出现空段，给app_name添加默认值：

-- 原Lua脚本中app_name提取的else分支修改为：
else
    record["app_name"] = "unknown-app"
end

4. 修正S3输出的Match规则 #

重写后的Tag以s3.开头，因此OUTPUT的Match要改为s3.*，否则不会匹配到重写后的日志流：

[OUTPUT]
    Name s3
    Match s3.*  # 改为匹配重写后的Tag
    bucket                       logs.bucket.us-east-1.domain.com
    region                       us-east-1
    # 调整路径格式，与FluentD一致：年/月/日/namespace/container/container-namespace-时间-podid.txt
    s3_key_format                /%Y/%m/%d/$TAG[1]/$TAG[3]/$TAG[3]-$TAG[1]-%Y%m%d-%H%M-$TAG[4].txt
    store_dir                    /var/log/fluentbit-s3-buffers
    total_file_size              256MB
    upload_timeout               2m
    use_put_object               On
    preserve_data_ordering       On

解释：重写后的Tag格式为s3.<namespace>.<app>.<container>.<podid>，所以$TAG[1]是namespace，$TAG[3]是container，$TAG[4]是podid。

验证修改 #

重启FluentBit后，日志会被重写为正确的Tag，S3输出会根据新Tag生成与FluentD一致的路径格式。

内容的提问来源于stack exchange，提问作者Doug