Update and rename test.lua to kafka-logger.lua

plugins/kafka-logger.lua

@@ -0,0 +1,60 @@
+---
+--- Generated by UUSEC(https://www.uusec.com/)
+--- Created by Safe3.
+--- DateTime: 2022/9/21 20:37
+---
+local cjson = require("cjson.safe")
+local producer = require("resty.kafka.producer")
+local log = require("waf.log")
+
+local _M = {
+    version = 0.1,
+    name = "kafka-logger"
+}
+
+--[[
+function _M.req_filter(waf)
+
+end
+
+function _M.resp_header_filter(waf)
+
+end
+
+function _M.resp_body_filter(waf)
+
+end
+--]]
+
+local function kafka_log(_, broker_list, info)
+    local kp = producer:new(broker_list, { producer_type = "async" })
+    local json = cjson.new()
+
+    local key = "key"
+    local message = json.encode(info)
+    local ok, err = kp:send("waf-log", key, message)
+    if not ok then
+        log.errorlog(_M.name, " send err: ", err)
+    end
+end
+
+function _M.log(waf)
+    local broker_list = {
+        {
+            host = "127.0.0.1",
+            port = 9092,
+
+            sasl_config = {
+                mechanism = "PLAIN",
+                user = "USERNAME",
+                password = "PASSWORD",
+            },
+        },
+    }
+
+    if waf.msg then
+        local country, province, city = log.ip2loc(waf.ip)
+        local info = { rule_id = waf.rule_id, ip = waf.ip, host = waf.host, url = waf.reqUri, data = waf.msg, country = country, province = province, city = city }
+        log.broker(kafka_log, broker_list, info)
+    end
+end