kafka flusher support dynamic topic and log fields extension

[shalousun]

设计实现目标

• kafka flusher刷入kafka的日志数据存储扁平化，方便ilogtail用户接入第三方开源日志生态系统，例如：通过logstash消费写入ES的场景，当前kafka flusher写入kafka的数据尤其是json日志处理成期望结构写入es比较难。
• kafka flusher提供动态的topic写入支持能力。支持类似test_%{kubernetes['labels']['name']}这样的动态topic配置能力。
• kafka flusher针对文件采集和容器标准容器采集的基准字段刷入kafka存储保持风格统一，例如：文件采集字段名__tag__:_container_name_ ，容器采集字段名__container_name_ ，

设计改进措施

• 数据扁平化处理措施：将contents中的key value接口平展为一级字段。

例如刷入kafka数据样例：

{
	"Time": 1659110162,
	"Contents": [{
		"Key": "content",
		"Value": "{\"date\":\"2022-07-29 23:56:02.131\"}"
	}]
}

扁平化处理后变为下面格式就非常方便提取content字段中真实的日志了（ps: 样例参考stdout插件，这是配置json插件 ExpandDepth=0的效果）：

{ 
  "Time": 1659110162,"content": "{\"date\":\"2022-07-29 23:56:02.131\"}"
}

• kafka flusher动态topic改进措施：TopicFormat这个当前基于文件路径或者自定义topic的能力在kafka flusher层面利用起来；同时针对kubernetes云原生场景提供pod label作为动态topic过滤的一种维度，这和filebeat和fluent bit比较类似。
• 针对两个采集类型字段不统一的问题，是否从原始数据源端统一化，还是各flusher去处理，需要进一步讨论。

参数说明

参数	类型	是否必选	说明
Type	String	是	插件类型
Brokers	String数组	是	kafka Brokers
Version	String	是	kafka版本号 ，0.8.2.0到2.6.x
Topic	String	是	Kafka Topic，可支持动态topic表达式，例如：topicprefix_%{field_name}
SASLMechanism	String	否	SASL mechanism ，选项(待定完善)：PLAIN，SCRAM-SHA-512，SCRAM-SHA-256，OAUTHBEARER（filebeat仅支持前三种）
SASLUsername	String	否	SASL用户名
SASLPassword	String	否	SASL密码
PartitionerType	String	否	Partitioner类型。取值：roundrobin、hash、random。默认为：random。
HashKeys	String数组	否	PartitionerType为hash时，需指定HashKeys。
HashOnce	Boolean	否
ClientID	String	否	写入Kafka的Client ID，默认取值：LogtailPlugin。
QequiredAcks	int	否	ACK的可靠等级.0=无响应,1=等待本地消息,-1=等待所有副本提交.默认1，
Compression	String	否	压缩算法，可选值：none, snappy，lz4和gzip，默认值gzip
CompressionLevel	Int	否	压缩级别，可选值：1~9，默认值：4,设置为0则禁用Compression
MaxMessageBytes	Int	否	一个批次提交的大小限制，配置和message.max.bytes对应，默认值：1000000
MaxRetries	Int	否	提交失败重试次数，最大3次，默认值：3
BulkMaxSize	Int	否	单次请求提交事件数，默认2048
BulkFlushFrequency	Int	否	发送批量 Kafka 请求之前等待的时间,0标识没有时延，默认值0
Timeout	Int	否	超时前等待 Kafka 代理响应的秒数,默认值为 30（秒）
FlusherMode	Int	否	数据刷入模式：0表示数据扁平处理，不保留__content__内容；1选项和0选项数据处理相同，保留__content__原始日志；2表示直接ilogtail原始数据写入topic，社区选择配置为2的概率非常低。参考#172 方式，不过面向社区用户场景更倾向默认数据扁平化。

关于动态topic也可能存在另外一种可探讨的方式，增加一个：TopicKey去匹配动态topic配置的维度字段。但是也这种情况也会出现另一个可选字段：TopicPrefix 即topic前缀，肯定会有要求前缀的需求出现。

因此Topic支持静态和动态配置项更少，实现是支持既定的规则表达式解析即可。

样例

kafka flusher配置样例

apiVersion: v1
kind: ConfigMap
metadata:
  name: ilogtail-user-cm
  namespace: ilogtail
data:
  java.yaml: |
    enable: true
    inputs:
      - Type: service_docker_stdout
        Stderr: false
        Stdout: true                # only collect stdout
        IncludeK8sLabel:
          app: springboot-tomcat # choose containers with this label
    flushers:
      - Type: flusher_kafka
        Brokers:
          - localhost:9092
        Topic: kubernetes_springboot-docker
        FlusherMode: 1

kafka flusher优化后写入kafka的数据样例(如果添加json插件处理，content中的json字段还能再平铺到成一级字段)

{
	"content": "{\"date\":\"2022-07-29 23:56:02.132\",\"appname\":\"springboot-docker\",\"thread\":\"http-nio-8080-exec-8\",\"level\":\"ERROR\",\"class\":\"com.benchmark.springboot.controller.LogController\",\"msg\":\"error log\"}",
	"_time_": "2022-07-29T15:56:02.134196156Z",
	"_source_": "stdout",
	"_image_name_": "registry.cn-shanghai.aliyuncs.com/shalousun/springboot:2.6.6-tomcat",
	"_container_name_": "springboot-tomcat",
	"_pod_name_": "springboot-tomcat-deployment-7d477c67d8-dctld",
	"_namespace_": "default",
	"_pod_uid_": "4e219086-a1b5-473a-87ea-10db2945acd3",
	"_container_ip_": "172.30.166.167",
	"__time__": "1659110162"
}

[askuy]

我觉得可以参考下 fluent-bit 采集方式，他们很好的兼容了kafka、es、clickhouse

1 fluent-bit的默认采集数据结构

• @timestamp字段：string or float，用于记录采集日志的时间
• message字段：string，用于记录用户写入的自定义日志的完整内容

这样的好处是，确保有个系统时间，在最外层，能够方便的写入es，clickhouse种，因为完整的log里是有用户记录的，不确定是否会有时间字段。

2 可以修改字段

例如 @timestamp 可以修改为别的字段适配用户的业务系统

3 可以追加字段

[FILTER]
    Name            modify
    Match           *
    Add             _cluster_ ${CLUSTER_NAME}
    Add             _log_agent_ ${HOSTNAME}
    Add             _node_ip_ ${NODE_IP}

[shalousun]

@timestamp 对于es生态接入确实是需要这个字段，属于默认的时间检索字段，kibana使用时没有这个字段好像也不能从es导入索引。message和content性质一样

[messixukejia]

message/content 这一层级字段的必要性，可以进一步讨论下。还是直接一级平铺？

[messixukejia]

关于Kafka Flusher的配置项，丰富度、层次是不是也需要考虑下。有些功能应该可以先不实现，不过扩展性最好需要先考虑下。

ilogtail跟Beats用的都是github.com/Shopify/sarama，是不是可以借鉴下Elastic Beats，适当扩展下Fluentd、Fluent-bit的配置项？

1、Elastic Beats：
代码：https://github.com/elastic/beats/blob/main/libbeat/outputs/kafka/config.go
文档：https://www.elastic.co/guide/en/beats/filebeat/current/kafka-output.html

2、Fluentd
代码：https://github.com/fluent/fluent-plugin-kafka/blob/master/lib/fluent/plugin/out_kafka2.rb
文档：https://docs.fluentd.org/output/kafka

3、Fluent-bit
文档：https://docs.fluentbit.io/manual/pipeline/outputs/kafka
代码：https://github.com/fluent/fluent-bit/blob/master/plugins/out_kafka/kafka_config.c

[shalousun]

看了下beats的实现，使用github.com/Shopify/sarama确实可以借鉴Beats，github.com/Shopify/sarama很多kafka关键参数配置传过去都自己处理了，实现时也简洁很多

[messixukejia]

格式不一致问题探讨

kafka flusher针对文件采集和容器标准容器采集的基准字段刷入kafka存储保持风格统一，例如：文件采集字段名__tag__:_container_name_ ，容器采集字段名__container_name_ ，
针对两个采集类型字段不统一的问题，是否从原始数据源端统一化，还是各flusher去处理，需要进一步讨论。

回答这个问题，首先要明确一点：ilogtail内部数据流是按照SLS的PB结构来组织的，LogGroup是多条日志组成的，其中日志公共部分存在LogTags中。一条完整的日志应该是由__topic__, LogTags, Logs共同组成。

方案的探讨：目前确实不同场景格式不太统一，鉴于历史兼容性原因无法从数据源进行统一。目前比较合适的实现机制是增加一个通用转换层，可以根据第三方存储系统提供格式转换功能。ES Flusher默认使用ES友好格式，CK FLusher默认使用CK友好格式，Kafka Flusher视背后的存储系统可以自定义选择。

[messixukejia]

关于topic：
1、继承文本日志采集时配置的topic，即TopicFormat，目前支持3种，default、文件路径正则表达式、静态自定义。
2、从日志字段中提取，可以加前缀。
3、2的扩展：可以从K8s属性（Pod、容器名等字段）获取topic规则
4、2的扩展：K8s采集支持通过ExternalEnvTag、ExternalK8sLabelTag参数实现容ENV、Label相关属性，这样也可以做到容器标签的topic命名。

Filebeat的情况，支持条件判断 + 字段取值

output.kafka:
  hosts: ["localhost:9092"]
  topic: "logs-%{[agent.version]}"
  topics:
    - topic: "critical-%{[agent.version]}"
      when.contains:
        message: "CRITICAL"
    - topic: "error-%{[agent.version]}"
      when.contains:
        message: "ERR"

[lx668]

关于ilogtail采集数据，希望保留kafka属性原有字段offset。 再做日志实时查询的时候，需要用到该字段