kafka flusher使用动态topic环境变量使用

[shalousun]

需求

flusher的动态topic表达式支持从环境变量中读取值替换。

涉及组件

kafka、pulsar、elasticsearch等支持动态topic或者动态索引的flusher组件

用户场景

1. 用户在使用ilogtail时在ilogtail容器上绑定系统变量，然后希望获取系统变量来作为动态topic。

- name: _app_
  value: kafka

用户自定义的环境变量可以追加

env:
  - name: ALIYUN_LOG_ENV_TAGS       # add log tags from env
    value: _node_name_|_node_ip_|_app_
# 

这样配置后保证方式环境变量_app_被追加到日志中

描述场景基本为ilogtail sidecar部署模式，用户设置变量通常用于做业务纬度上的区分

2. 通过processor_add_fields处理组件添加

processors:
  - Type: processor_add_fields
    Fields: 
      service: ${env_name}
    IgnoreIfExist: false

通过processor_add_fields processor提取的变量同样也可以添加到日志内容中。

配置使用方式

• %{tag.fieldname} 这个必须保证环境变量的值被采集添加到了日志的tags中，配置如上面的介绍。这种方式当前ilogtail 1.3.0也可以支持
• ${env_name} (对于采用$的表达是直接识别为系统变量获取)，它的好处是不需要把标签数据带入日志存储，例如使用ilogtail去采集kafka、zk等大数据组件日志式，用户可能是基于一类组件建立了一个topic做日志推送，环境变量值仅仅是做topic区分，环境变量值并无存储需求。这种方式的实现只需要读取系统变量，然后从动态表达式里替换掉即可，和fluent-bit、filebeat使用基本一致

[messixukejia]

这是一个使用的说明是吧？ 后面定稿后可以正式提交到文档里。

上文结构是不是可以调衡为两部分：
1、根据日志内容动态topic，通过ALIYUN_LOG_ENV_TAGS、processor_add_fields两种方式添加字段。

2、直接环境变量设置topic，需要打开enable_env_ref_in_config，社区版是可以直接打开的，可以在overwrite_community_edition_flags中修改下，提交个pr
设置：Topic: ${env_name}

[shalousun]

结构区分下会清晰点，后面新增到官方文档里

[q121102302]

请问 enable_env_ref_in_config 这个是在哪里打开呢？ 我使用的版本是1.3.1 没有找到哇

[messixukejia]

connfig文件里配置下就可以了

[q121102302]

是这个里面增加 { "enable_env_ref_in_config":true } 吗？ 我使用后 好像并不行

[shalousun]

我验证的情况看，在ilogtail_config.json中配置是可以的

{
 "enable_env_ref_in_config": true
}

flusher测试配置

flushers:
  - Type: flusher_stdout
    OnlyStdout: true
  - Type: flusher_kafka_v2
    Brokers:
      - 192.168.6.128:9092
    Topic: test-${USER}

我使用一个并不存在Kafka，发送的报错信息：Failed to produce message to topic test-root 看是生效的，设置为false看到的报错信息就是：error:kafka: Failed to produce message to topic test-${USER}:

[shalousun]

pr #746 已经将enable_env_ref_in_config默认开启