SkyWalking集成Kafka实现日志异步采集经验总结
SkyWalking日志异步采集架构
【重点知识】
1、kafka-reporter-plugin-x.x.x.jar包放plugins目录后必走kafka（kafka没有正确配置就会报错）
2、异步不开启数据压缩，数据量较大，pod多、业务大时容易造成网络拥堵
3、SW_AGENT_INSTANCE_NAME默认：UUID@hostname，建议调整为：nodeip@podip
4、SkyWalking OAP配置启用kafka并配置好后，OAP启动时会自动创建Topic
5、需要开启数据压缩，否则容易造成网络拥堵
6、官方说：使用Kafka时skywalking.collector.backend_service参数依然要配置
环境及帮助文件
SkyWalking-OAP官方下载地址
SkyWalking-OAP帮助文档 -- 配置说明
SkyWalking-Agent官方下载地址
SkyWalking-Agent帮助文档 -- 配置说明
Agent is available for JDK 8 - 21
Oracle JDK8u202(Oracle JDK8最后一个非商业版本) 下载地址：Oracle JDK8u202
skywalking-OAP服务
启用kafka配置
编辑oap配置文件 ${skywalking-apm}/config/application.yml配置
# line 323 apache-skywalking-apm-9.6.0
kafka-fetcher:
selector: ${SW_KAFKA_FETCHER:default}
default:
# 多个使用逗到分隔
bootstrapServers: ${SW_KAFKA_FETCHER_SERVERS:localhost:9092}
namespace: ${SW_NAMESPACE:""}
# 默认3，几个消费者就配几
partitions: ${SW_KAFKA_FETCHER_PARTITIONS:3}
# 不能超过kafka集群机器的数量，默认2
replicationFactor: ${SW_KAFKA_FETCHER_PARTITIONS_FACTOR:2}
enableNativeProtoLog: ${SW_KAFKA_FETCHER_ENABLE_NATIVE_PROTO_LOG:true}
enableNativeJsonLog: ${SW_KAFKA_FETCHER_ENABLE_NATIVE_JSON_LOG:true}
consumers: ${SW_KAFKA_FETCHER_CONSUMERS:1}
kafkaHandlerThreadPoolSize: ${SW_KAFKA_HANDLER_THREAD_POOL_SIZE:-1}
kafkaHandlerThreadPoolQueueSize: ${SW_KAFKA_HANDLER_THREAD_POOL_QUEUE_SIZE:-1}
...skywalking-agent服务
经验
-》 kafka-reporter-plugin-x.x.x.jar 这个包放plugins目录后必走kafka，如果kafka没有正确配置就会报错
-》 不压缩的情况下，一次请求约1KB日志
-》SW_AGENT_INSTANCE_NAME 默认：UUID@hostname，建议：host_ip@pod_ip
-》使用Kafka时skywalking.collector.backend_service参数依然要配置（官方说的）
开启Kafka
- 将${skywalking-agent}/optional-reporter-plugins/kafka-reporter-plugin-x.x.x.jar 拷贝到 ${skywalking-agent}/plugins
开启消息数据压缩
# optional-reporter-plugins下面任选一种（lz4、snappy、zstd-jni）压缩实现包即可
- 将${skywalking-agent}/optional-reporter-plugins/zstd-jni-x.x.x.jar 拷贝到 ${skywalking-agent}/plugins
Springboot应用增加jvm参数
应用启动参数,多个kafka地址使用逗号分割
# myApp 改为你的包名
java \
-javaagent:/usr/local/agent/skywalking-agent.jar \
-Dskywalking.agent.service_name=myApp \
-Dskywalking.collector.backend_service=aigoServer:11800 \
-Dskywalking.plugin.kafka.bootstrap_servers=10.81.**.14:9092 \
-Xmx1g \
myApp.jarTomcat应用增加jvm参数
应用启动参数,多个kafka地址使用逗号分割
# myApp 改为实际应用名
JAVA_OPTS="$JAVA_OPTS --javaagent:/usr/local/agent/skywalking-agent.jar \
-Dskywalking.agent.service_name=myApp \
-Dskywalking.collector.backend_service=aigoServer:11800 \
-Dskywalking.plugin.kafka.bootstrap_servers=10.81.**.14:9092"说明：
使用Kafka时skywalking.collector.backend_service参数依然要配置（官方说的）
附件
附件一：lz4压缩算法
LZ4压缩算法详解：原理、实现与Java应用-CSDN博客
附件二：snappy压缩工具
压缩速度快
GitHub - google/snappy: A fast compressor/decompressor
附件三：Zstd压缩算法
压缩比高，CPU够用此算法
附件四：Kafka可视化管理工具
Kafka Map -- 通过配置kafka地址实现管理
kafka CMAK -- 通过配置zookeeper地址实现管理
等等
附件五：客户端配置示例
# 客户端参数两种key形式
# 客户端参数两种key形式
-Dskywalking.agent.service_name=smet-rpc-service
-javaagent:/app/saashealth/skywalking-agent/skywalking-agent-9.2.0/skywalking-agent.jar
-Dskywalking.plugin.kafka.bootstrap_servers=10.130.***.92:19092
-Dskywalking.collector.backend_service=10.130.***.92:11801# 大写变量形式
-DSW_AGENT_NAME=smet-rpc-service \
-javaagent:/app/saashealth/skywalking-agent/skywalking-agent-9.2.0/skywalking-agent.jar \
-DSW_KAFKA_BOOTSTRAP_SERVERS=10.130.***.92:19092 \
-DSW_AGENT_COLLECTOR_BACKEND_SERVICES=10.130.***.92:11801 \