---
--- Generated by UUSEC(https://www.uusec.com/)
--- Created by Safe3.
--- DateTime: 2022/9/21 20:37
---
local producer = require("resty.kafka.producer")
local log = require("waf.log")

local _M = {
    version = 0.1,
    name = "kafka-logger"
}

local function kafkaLog(_, brokerList, info)
    local kp = producer:new(brokerList, { producer_type = "async" })
    local key = "key"
    local message = log.encodeJson(info)
    local ok, err = kp:send("waf-log", key, message)
    if not ok then
        log.errLog(_M.name, " send err: ", err)
    end
end

function _M.log_post_filter(waf)
    local brokerList = {
        {
            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 = log.utf8(waf.msg), req = log.utf8(log.getReq()), country = country, province = province, city = city, create_at = ngx.localtime() }
        log.broker(kafkaLog, brokerList, info)
    end
end

return _M