feat: support send error-log to kafka brokers by ronething · Pull Request #8693 · apache/apisix

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

 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

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

+    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

+                             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

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

+            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

+    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

+        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

+            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

+            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

+    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

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

+    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

+                            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

+        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

+    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



		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)

Uh oh!

Conversation

ronething commented Jan 16, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Checklist

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ronething Jan 17, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ronething commented Jan 17, 2023

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ronething Jan 28, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ronething commented Jan 30, 2023

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

ronething commented Jan 16, 2023 •

edited

Loading

ronething Jan 17, 2023 •

edited

Loading

ronething Jan 28, 2023 •

edited

Loading