feat: support send error-log to kafka brokers #8693
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -21,6 +21,7 @@ local batch_processor = require("apisix.utils.batch-processor") | |
| local plugin = require("apisix.plugin") | ||
| local timers = require("apisix.timers") | ||
| local http = require("resty.http") | ||
| local producer = require("resty.kafka.producer") | ||
| local plugin_name = "error-log-logger" | ||
| local table = core.table | ||
| local schema_def = core.schema | ||
|
|
@@ -32,6 +33,9 @@ local string = require("string") | |
| local lrucache = core.lrucache.new({ | ||
| ttl = 300, count = 32 | ||
| }) | ||
| local kafka_prod_lrucache = core.lrucache.new({ | ||
| ttl = 300, count = 32 | ||
| }) | ||
|
|
||
|
|
||
| local metadata_schema = { | ||
|
|
@@ -66,6 +70,62 @@ local metadata_schema = { | |
| }, | ||
| required = {"endpoint_addr", "user", "password", "database", "logtable"} | ||
| }, | ||
| kafka = { | ||
| type = "object", | ||
| properties = { | ||
| brokers = { | ||
| type = "array", | ||
| minItems = 1, | ||
| items = { | ||
| type = "object", | ||
| properties = { | ||
| host = { | ||
| type = "string", | ||
| description = "the host of kafka broker", | ||
| }, | ||
| port = { | ||
| type = "integer", | ||
| minimum = 1, | ||
| maximum = 65535, | ||
| description = "the port of kafka broker", | ||
| }, | ||
| sasl_config = { | ||
| type = "object", | ||
| description = "sasl config", | ||
| properties = { | ||
| mechanism = { | ||
| type = "string", | ||
| default = "PLAIN", | ||
| enum = {"PLAIN"}, | ||
| }, | ||
| user = { type = "string", description = "user" }, | ||
| password = { type = "string", description = "password" }, | ||
| }, | ||
| required = {"user", "password"}, | ||
| }, | ||
| }, | ||
| required = {"host", "port"}, | ||
| }, | ||
| uniqueItems = true, | ||
| }, | ||
| kafka_topic = {type = "string"}, | ||
| producer_type = { | ||
| type = "string", | ||
| default = "async", | ||
| enum = {"async", "sync"}, | ||
| }, | ||
| required_acks = { | ||
| type = "integer", | ||
| default = 1, | ||
| enum = { 0, 1, -1 }, | ||
| }, | ||
| key = {type = "string"}, | ||
| -- in lua-resty-kafka, cluster_name is defined as number | ||
| -- see https://github.com/doujiang24/lua-resty-kafka#new-1 | ||
| cluster_name = {type = "integer", minimum = 1, default = 1}, | ||
| }, | ||
| required = {"brokers", "kafka_topic"}, | ||
| }, | ||
| name = {type = "string", default = plugin_name}, | ||
| level = {type = "string", default = "WARN", enum = {"STDERR", "EMERG", "ALERT", "CRIT", | ||
| "ERR", "ERROR", "WARN", "NOTICE", "INFO", "DEBUG"}}, | ||
|
|
@@ -81,6 +141,7 @@ local metadata_schema = { | |
| {required = {"skywalking"}}, | ||
| {required = {"tcp"}}, | ||
| {required = {"clickhouse"}}, | ||
| {required = {"kafka"}}, | ||
| -- for compatible with old schema | ||
| {required = {"host", "port"}} | ||
| }, | ||
|
|
@@ -285,11 +346,63 @@ local function update_filter(value) | |
| end | ||
|
|
||
|
|
||
| local function create_producer(broker_list, broker_config, cluster_name) | ||
| core.log.info("create new kafka producer instance") | ||
| return producer:new(broker_list, broker_config, cluster_name) | ||
| end | ||
|
|
||
|
|
||
| local function send_to_kafka(log_message) | ||
| -- avoid race of the global config | ||
| local metadata = plugin.plugin_metadata(plugin_name) | ||
| if not (metadata and metadata.value and metadata.modifiedIndex) then | ||
| return false, "please set the correct plugin_metadata for " .. plugin_name | ||
| end | ||
| local config, err = lrucache(plugin_name, metadata.modifiedIndex, update_filter, metadata.value) | ||
| if not config then | ||
| return false, "get config failed: " .. err | ||
| end | ||
|
|
||
| core.log.info("sending a batch logs to kafka brokers: ", | ||
| core.json.delay_encode(config.kafka.brokers)) | ||
|
|
||
| local broker_config = {} | ||
| broker_config["request_timeout"] = config.timeout * 1000 | ||
|
There was a problem hiding this comment. Where is the config from? This function doesn't have an argument called config. There was a problem hiding this comment. it comes from apisix/apisix/plugins/error-log-logger.lua Line 110 in d852953 and set by apisix/apisix/plugins/error-log-logger.lua Line 305 in d852953 |
||
| broker_config["producer_type"] = config.kafka.producer_type | ||
| broker_config["required_acks"] = config.kafka.required_acks | ||
|
|
||
| -- reuse producer via kafka_prod_lrucache to avoid unbalanced partitions of messages in kafka | ||
| local prod, err = kafka_prod_lrucache(plugin_name, metadata.modifiedIndex, | ||
| create_producer, config.kafka.brokers, broker_config, | ||
| config.kafka.cluster_name) | ||
| if not prod then | ||
| return false, "get kafka producer failed: " .. err | ||
| end | ||
| core.log.info("kafka cluster name ", config.kafka.cluster_name, ", broker_list[1] port ", | ||
| prod.client.broker_list[1].port) | ||
|
|
||
| local ok | ||
| for i = 1, #log_message, 2 do | ||
| ok, err = prod:send(config.kafka.kafka_topic, | ||
| config.kafka.key, core.json.encode(log_message[i])) | ||
| if not ok then | ||
| return false, "failed to send data to Kafka topic: " .. err .. | ||
| ", brokers: " .. core.json.encode(config.kafka.brokers) | ||
| end | ||
| core.log.info("send data to kafka: ", core.json.delay_encode(log_message[i])) | ||
| end | ||
|
|
||
| return true | ||
| end | ||
|
|
||
|
|
||
| local function send(data) | ||
| if config.skywalking then | ||
| return send_to_skywalking(data) | ||
| elseif config.clickhouse then | ||
| return send_to_clickhouse(data) | ||
| elseif config.kafka then | ||
| return send_to_kafka(data) | ||
| end | ||
| return send_to_tcp_server(data) | ||
| end | ||
|
|
@@ -307,7 +420,7 @@ local function process() | |
| core.log.warn("set log filter failed for ", err) | ||
| return | ||
| end | ||
| if not (config.tcp or config.skywalking or config.clickhouse or config.kafka) then | ||
| config.tcp = { | ||
| host = config.host, | ||
| port = config.port, | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Let's group the resty module together.