通过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分钟）：