如何在Fluent Bit Kafka输出中使用嵌套记录字段作为消息键？

问题场景 #

现有Fluent Bit配置如下：

[INPUT]
    Name tail
    Path /var/log/containers/*.log
    multiline.parser docker, cri
    Tag kube.*
    Mem_Buf_Limit 5MB
    Skip_Long_Lines On

[FILTER]
    Name kubernetes
    Match kube.*
    Merge_Log On
    Keep_Log Off
    K8S-Logging.Parser On
    K8S-Logging.Exclude On

[OUTPUT]
    Name kafka
    Match kube.*
    Brokers <some-broker>
    Topics <some-topic>

发送到Kafka的日志结构包含嵌套的kubernetes对象：

{
   "@timestamp":1670628155.720578,
   "log":"<some-container-log>",
   "stream":"stdout",
   "time":"2022-12-09T23:22:35.720578536Z",
   "kubernetes":{
      "pod_name":"<...>",
      "namespace_name":"<...>",
      "pod_id":"<...>",
      "annotations":{ <...> },
      "host":"<...>",
      "container_name":"<...>",
      "docker_id":"<...>",
      "container_hash":"<...>",
      "container_image":"<...>"
   }
}

尝试用Message_Key_Field引用嵌套字段（如kubernetes.pod_id）时消息键为空，多种写法均无效。

解决方案 #

有两种可靠的方式可以实现引用嵌套字段作为Kafka消息键：

方法1：使用Modify过滤器提取嵌套字段到顶层（兼容所有版本）

添加一个modify过滤器，把需要的嵌套字段提取到日志记录的顶层，之后直接引用顶层字段即可：

[FILTER]
    Name modify
    Match kube.*
    Add pod_id ${kubernetes.pod_id}

然后在Kafka输出配置中引用这个新的顶层字段：

[OUTPUT]
    Name kafka
    Match kube.*
    Brokers <some-broker>
    Topics <some-topic>
    Message_Key_Field pod_id

方法2：直接使用变量引用语法（Fluent Bit 1.9+版本支持）

如果你的Fluent Bit版本是1.9或以上，可以直接在Message_Key_Field中使用$开头的变量引用语法来指定嵌套路径：

[OUTPUT]
    Name kafka
    Match kube.*
    Brokers <some-broker>
    Topics <some-topic>
    Message_Key_Field $kubernetes.pod_id

注意：早期版本的Fluent Bit Kafka输出插件不支持直接引用嵌套字段，所以方法1是通用兼容方案。

内容的提问来源于stack exchange，提问作者kafka-user122