通过Kafka-Logger插件收集流量进行漏洞扫描

方案

可以通过APISIX kafka-logger 插件将日志作为 JSON 对象推送到 Apache Kafka 集群中，消费Kafka里的数据格式化后添加到MySQL。

方案详情

1、登录APISIX，启用内置的kafka-logger 插件：

2、配置插件：

配置参考：https://apisix.apache.org/zh/docs/apisix/plugins/kafka-logger/

关键字段解释：
batch_max_size：数据首先写入缓冲区。当缓冲区超过 batch_max_size 或 buffer_duration 设置的值时，则会将数据发送到 Kafka 服务器并刷新缓冲区。设置为1时，每次写入到kafka一条流量数据。
brokers：需要推送的 Kafka 的 broker 列表。
cluster_name：Kafka 集群的名称，当有两个及以上 Kafka 集群时使用。只有当 producer_type 设为 async 模式时才可以使用该属性。
include_req_body：当设置为 true 时，包含请求体。注意：如果请求体无法完全存放在内存中，由于 NGINX 的限制，APISIX 无法将它记录下来。
include_resp_body：当设置为 true 时，包含响应体。
producer_batch_size：对应 lua-resty-kafka 中的 batch_size 参数，单位为字节。
producer_type：生产者发送消息的模式。
以下是本地配置，满足现有需求:

{"_meta": {"disable": true},"batch_max_size": 1,"brokers": [{"host": "10.200.0.228","port": 9092}],"buffer_duration": 60,"cluster_name": 1,"inactive_timeout": 5,"include_req_body": true,"include_resp_body": true,"kafka_topic": "secapisixtest","max_retry_count": 0,"meta_format": "default","name": "kafka logger","producer_batch_num": 1,"producer_batch_size": 1048576,"producer_max_buffering": 50000,"producer_time_linger": 1,"producer_type": "async","required_acks": 1,"retry_delay": 1,"timeout": 3
}

3、点击启用并提交：

4、自此，就已经完成了对APISIX Kafka-Logger插件的配置，Kafka中也有了预期的数据：

5、后续可以消费Kafka中的数据入新的MySQL库，也可以用现有MySQL，在00:00 - 8:00 限制并发写MySQL，不影响工作时间使用。

稳定性测试

场景一、当Kafka挂掉时，是否影响测试环境稳定性：
1、当开启 kafka-logger 插件，并配置正确Kafka地址时，请求接口可正常请求，延迟在7 millis左右：


2、给kafka-logger 插件配置一个不存在的kafka地址，请求接口可正常请求，延迟在7 millis左右，符合预期：

回退方案

理论 Kafka-Logger插件作为APISIX原生插件不会影响到测试环境，若出现问题，关闭插件即可（大约占用1分钟）：