feat: support send error-log to kafka brokers #8693

ronething · 2023-01-16T13:13:02Z

Description

support send error-log to kafka brokers.

for producer via lrucache, i use plugin_name .. "#kafka" as key, and metadata.modifiedIndex as version.

        -- reuse producer via lrucache to avoid unbalanced partitions of messages in kafka
        prod, err = lrucache(plugin_name .. "#kafka", metadata.modifiedIndex,
                             create_producer, config.kafka.brokers, broker_config, config.kafka.cluster_name)

Checklist

I have explained the need for this PR and the problem it solves
I have explained the changes or the new features added to this PR
I have added tests corresponding to this change
I have updated the documentation to reflect this change
I have verified that this change is backward compatible (If not, please discuss on the APISIX mailing list first)

spacewander · 2023-01-17T05:54:20Z

apisix/plugins/error-log-logger.lua

@@ -19,6 +19,7 @@ local core = require("apisix.core")
 local errlog = require("ngx.errlog")
 local batch_processor = require("apisix.utils.batch-processor")
 local plugin = require("apisix.plugin")
+local producer = require ("resty.kafka.producer")
 local timers = require("apisix.timers")
 local http = require("resty.http")


Let's group the resty module together.

spacewander · 2023-01-17T05:56:34Z

apisix/plugins/error-log-logger.lua

+
+
+local function send_to_kafka(log_message)
+    core.log.info("sending a batch logs to kafka brokers: ", core.json.encode(config.kafka.brokers))


Can we use delay_encode?

spacewander · 2023-01-17T05:57:14Z

apisix/plugins/error-log-logger.lua

+    if not (metadata and metadata.value and metadata.modifiedIndex) then
+        core.log.info("please set the correct plugin_metadata for ", plugin_name)
+        return
+    else


We can use end here, no need to nest the code

spacewander · 2023-01-17T05:58:46Z

apisix/plugins/error-log-logger.lua

+                             create_producer, config.kafka.brokers, broker_config,
+                             config.kafka.cluster_name)
+        if not prod then
+            return false, "get kafka producer failed " .. err


Suggested change

return false, "get kafka producer failed " .. err

return false, "get kafka producer failed: " .. err

spacewander · 2023-01-17T05:59:30Z

docs/en/latest/plugins/error-log-logger.md

+| kafka.kafka_topic                      | string  | True     |                |                       | Target topic to push the logs for organisation.                                                                                                                                                                                                                                                                                                  |
+| kafka.producer_type                    | string  | False    | async          | ["async", "sync"]     | Message sending mode of the producer.                                                                                                                                                                                                                                                                                                            |
+| kafka.required_acks                    | integer | False    | 1              | [0, 1, -1]            | Number of acknowledgements the leader needs to receive for the producer to consider the request complete. This controls the durability of the sent records. The attribute follows the same configuration as the Kafka `acks` attribute. See [Apache Kafka documentation](https://kafka.apache.org/documentation/#producerconfigs_acks) for more. |
+| kafka.key                              | string  | False    |                |                       | Key used for allocating partitions for messages.                                                                                                                                                                                                                                                                                                 |


cluster_name is not documented?

spacewander · 2023-01-17T06:06:13Z

apisix/plugins/error-log-logger.lua

+            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.encode(log_message[i]))


Can we use delay_encode?

spacewander · 2023-01-17T06:13:37Z

apisix/plugins/error-log-logger.lua

+    core.log.info("sending a batch logs to kafka brokers: ", core.json.encode(config.kafka.brokers))
+
+    local broker_config = {}
+    broker_config["request_timeout"] = config.timeout * 1000


Where is the config from? This function doesn't have an argument called config.

it comes from

apisix/apisix/plugins/error-log-logger.lua

Line 72 in d852953

timeout = {type = "integer", minimum = 1, default = 3},

apisix/apisix/plugins/error-log-logger.lua

Line 110 in d852953

local config = {}

and set by

apisix/apisix/plugins/error-log-logger.lua

Line 305 in d852953

config, err = lrucache(plugin_name, metadata.modifiedIndex, update_filter, metadata.value)

spacewander · 2023-01-17T06:14:58Z

apisix/plugins/error-log-logger.lua

+        return
+    else
+        -- reuse producer via lrucache to avoid unbalanced partitions of messages in kafka
+        prod, err = lrucache(plugin_name .. "#kafka", metadata.modifiedIndex,


Better to use a separate lrucache to cache different data.

spacewander · 2023-01-17T06:17:29Z

t/plugin/error-log-logger-kafka.t

+            core.log.error("this is a error message for test2.")
+        }
+    }
+--- response_body


The response_body in these tests is unnecessary as we don't provide responses at all.

spacewander · 2023-01-17T06:17:38Z

t/plugin/error-log-logger-kafka.t

+            core.log.error("this is a error message for test3.")
+        }
+    }
+--- response_body


ronething · 2023-01-17T07:19:23Z

i will change code and update PR later.

spacewander · 2023-01-28T06:53:00Z

apisix/plugins/error-log-logger.lua

+    local metadata = plugin.plugin_metadata(plugin_name)
+    if not (metadata and metadata.value and metadata.modifiedIndex) then
+        core.log.info("please set the correct plugin_metadata for ", plugin_name)
+        return


Should return a boolean here?

spacewander · 2023-01-28T06:53:15Z

apisix/plugins/error-log-logger.lua

+
+local function send_to_kafka(log_message)
+    core.log.info("sending a batch logs to kafka brokers: ",
+                   core.json.delay_encode(config.kafka.brokers))


Suggested change

core.json.delay_encode(config.kafka.brokers))

core.json.delay_encode(config.kafka.brokers))

spacewander · 2023-01-28T06:54:49Z

apisix/plugins/error-log-logger.lua

+    end
+
+    -- reuse producer via kafka_prod_lrucache to avoid unbalanced partitions of messages in kafka
+    local prod, err = kafka_prod_lrucache(plugin_name .. "#kafka", metadata.modifiedIndex,


We don't need to add "#kafka" suffix as this cache is individual

spacewander · 2023-01-28T06:55:05Z

apisix/plugins/error-log-logger.lua

+                            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.delay_encode(config.kafka.brokers)


delay_encode is only for log

spacewander · 2023-01-28T06:55:17Z

apisix/plugins/error-log-logger.lua

+        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)


Suggested change

prod.client.broker_list[1].port)

prod.client.broker_list[1].port)

spacewander · 2023-01-28T07:01:18Z

apisix/plugins/error-log-logger.lua

+    broker_config["producer_type"] = config.kafka.producer_type
+    broker_config["required_acks"] = config.kafka.required_acks
+
+    local metadata = plugin.plugin_metadata(plugin_name)


It seems using the config passed from process here might create a race:

Consider we have c1(config) and m1 (modifiedIndex) in process, and c2/m2 in send. It looks like we might use m2 as key and c1 as value in the cache below.

@spacewander

It looks like we might use m2 as key and c1 as value in the cache below.

Do you mean that we need to clone config before we use it like below? then we can use m2 and c2 in send.

local config = core.table.clone(config)

A clone of c1 still has a c1's value.
Maybe we can get the c2 in send like what we have done in process.

ok, i will change the code and update the PR later.

@spacewander Could you please take a look, thanks.

ronething · 2023-01-30T06:01:37Z

@soulbird Could you please take a look, thanks.

* upstream/master: feat(elasticsearch-logger): support multi elasticsearch endpoints (apache#8604) chore: use operator # instead of string.len (apache#8751) chore: hi 2023 (apache#8748) refactor(admin): stream_routes/upstreams/protos/services/global_rules/consumer_groups/plugin_configs (apache#8661) feat: support send error-log to kafka brokers (apache#8693) chore: upgrade `casbin` to `1.41.5` (apache#8744)

ronething added 3 commits January 16, 2023 21:09

feat: support send error-log to kafka brokers

1b378d6

fix: ci

202a3a2

docs: add error log to kafka

afcdcfa

ronething marked this pull request as ready for review January 17, 2023 03:59

spacewander requested changes Jan 17, 2023

View reviewed changes

ronething added 2 commits January 17, 2023 15:38

fix: review

dff0e68

fix: ci

4e35ef8

ronething requested a review from spacewander January 17, 2023 08:47

chore: test case add producer type

833467e

spacewander reviewed Jan 28, 2023

View reviewed changes

fix: review

6357cd1

ronething requested a review from spacewander January 28, 2023 09:58

fix: review

e663ae9

spacewander approved these changes Jan 30, 2023

View reviewed changes

soulbird approved these changes Jan 30, 2023

View reviewed changes

spacewander merged commit a5dc4c3 into apache:master Jan 31, 2023

ronething deleted the feat/error_logger_support_kafka branch January 31, 2023 01:37

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: support send error-log to kafka brokers #8693

feat: support send error-log to kafka brokers #8693

ronething commented Jan 16, 2023 •

edited

Loading

spacewander Jan 17, 2023

spacewander Jan 17, 2023

spacewander Jan 17, 2023

spacewander Jan 17, 2023

spacewander Jan 17, 2023

spacewander Jan 17, 2023

spacewander Jan 17, 2023

ronething Jan 17, 2023 •

edited

Loading

spacewander Jan 17, 2023

spacewander Jan 17, 2023

spacewander Jan 17, 2023

ronething commented Jan 17, 2023

spacewander Jan 28, 2023

spacewander Jan 28, 2023

spacewander Jan 28, 2023

spacewander Jan 28, 2023

spacewander Jan 28, 2023

spacewander Jan 28, 2023

ronething Jan 28, 2023 •

edited

Loading

spacewander Jan 29, 2023

ronething Jan 29, 2023

ronething Jan 29, 2023

ronething commented Jan 30, 2023



		local function send_to_kafka(log_message)
		core.log.info("sending a batch logs to kafka brokers: ", core.json.encode(config.kafka.brokers))

	return false, "get kafka producer failed " .. err
	return false, "get kafka producer failed: " .. err

	core.json.delay_encode(config.kafka.brokers))
	core.json.delay_encode(config.kafka.brokers))

	prod.client.broker_list[1].port)
	prod.client.broker_list[1].port)

feat: support send error-log to kafka brokers #8693

feat: support send error-log to kafka brokers #8693

Conversation

ronething commented Jan 16, 2023 • edited Loading

Description

Checklist

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ronething Jan 17, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ronething commented Jan 17, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ronething Jan 28, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ronething commented Jan 30, 2023

ronething commented Jan 16, 2023 •

edited

Loading

ronething Jan 17, 2023 •

edited

Loading

ronething Jan 28, 2023 •

edited

Loading