This is an automated email from the ASF dual-hosted git repository.
ashishtiwari pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/apisix.git
The following commit(s) were added to refs/heads/master by this push:
new 7d5aeafdb feat: add max pending entries option to batch-processor
(#12338)
7d5aeafdb is described below
commit 7d5aeafdb442151667be95ad4ebc37cd3b1fa6bb
Author: Ashish Tiwari <[email protected]>
AuthorDate: Mon Jun 23 11:10:17 2025 +0530
feat: add max pending entries option to batch-processor (#12338)
---
apisix/plugins/kafka-logger.lua | 16 ++++-
apisix/utils/batch-processor-manager.lua | 35 ++++++++-
apisix/utils/batch-processor.lua | 6 +-
t/plugin/kafka-logger3.t | 120 +++++++++++++++++++++++++++++++
4 files changed, 170 insertions(+), 7 deletions(-)
diff --git a/apisix/plugins/kafka-logger.lua b/apisix/plugins/kafka-logger.lua
index adeec2921..75510f52c 100644
--- a/apisix/plugins/kafka-logger.lua
+++ b/apisix/plugins/kafka-logger.lua
@@ -19,6 +19,7 @@ local core = require("apisix.core")
local log_util = require("apisix.utils.log-util")
local producer = require ("resty.kafka.producer")
local bp_manager_mod = require("apisix.utils.batch-processor-manager")
+local plugin = require("apisix.plugin")
local math = math
local pairs = pairs
@@ -140,7 +141,12 @@ local metadata_schema = {
properties = {
log_format = {
type = "object"
- }
+ },
+ max_pending_entries = {
+ type = "integer",
+ description = "maximum number of pending entries in the batch
processor",
+ minimum = 1,
+ },
},
}
@@ -246,6 +252,9 @@ end
function _M.log(conf, ctx)
+ local metadata = plugin.plugin_metadata(plugin_name)
+ local max_pending_entries = metadata and metadata.value and
+ metadata.value.max_pending_entries or nil
local entry
if conf.meta_format == "origin" then
entry = log_util.get_req_original(ctx, conf)
@@ -255,7 +264,7 @@ function _M.log(conf, ctx)
entry = log_util.get_log_entry(plugin_name, conf, ctx)
end
- if batch_processor_manager:add_entry(conf, entry) then
+ if batch_processor_manager:add_entry(conf, entry, max_pending_entries) then
return
end
@@ -307,10 +316,11 @@ function _M.log(conf, ctx)
end
core.log.info("send data to kafka: ", data)
+
return send_kafka_data(conf, data, prod)
end
- batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, func)
+ batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, func,
max_pending_entries)
end
diff --git a/apisix/utils/batch-processor-manager.lua
b/apisix/utils/batch-processor-manager.lua
index a8122773d..4e97bd617 100644
--- a/apisix/utils/batch-processor-manager.lua
+++ b/apisix/utils/batch-processor-manager.lua
@@ -29,6 +29,7 @@ function _M.new(name)
return setmetatable({
stale_timer_running = false,
buffers = {},
+ total_pushed_entries = 0,
name = name,
}, mt)
end
@@ -85,7 +86,25 @@ do
end
-function _M:add_entry(conf, entry)
+local function total_processed_entries(self)
+ local processed_entries = 0
+ for _, log_buffer in pairs(self.buffers) do
+ processed_entries = processed_entries + log_buffer.processed_entries
+ end
+ return processed_entries
+end
+
+function _M:add_entry(conf, entry, max_pending_entries)
+ if max_pending_entries then
+ local total_processed_entries_count = total_processed_entries(self)
+ if self.total_pushed_entries - total_processed_entries_count >
max_pending_entries then
+ core.log.error("max pending entries limit exceeded. discarding
entry.",
+ " total_pushed_entries: ",
self.total_pushed_entries,
+ " total_processed_entries: ",
total_processed_entries_count,
+ " max_pending_entries: ", max_pending_entries)
+ return
+ end
+ end
check_stale(self)
local log_buffer = self.buffers[conf]
@@ -94,11 +113,22 @@ function _M:add_entry(conf, entry)
end
log_buffer:push(entry)
+ self.total_pushed_entries = self.total_pushed_entries + 1
return true
end
-function _M:add_entry_to_new_processor(conf, entry, ctx, func)
+function _M:add_entry_to_new_processor(conf, entry, ctx, func,
max_pending_entries)
+ if max_pending_entries then
+ local total_processed_entries_count = total_processed_entries(self)
+ if self.total_pushed_entries - total_processed_entries_count >
max_pending_entries then
+ core.log.error("max pending entries limit exceeded. discarding
entry.",
+ " total_pushed_entries: ",
self.total_pushed_entries,
+ " total_processed_entries: ",
total_processed_entries_count,
+ " max_pending_entries: ", max_pending_entries)
+ return
+ end
+ end
check_stale(self)
local config = {
@@ -120,6 +150,7 @@ function _M:add_entry_to_new_processor(conf, entry, ctx,
func)
log_buffer:push(entry)
self.buffers[conf] = log_buffer
+ self.total_pushed_entries = self.total_pushed_entries + 1
return true
end
diff --git a/apisix/utils/batch-processor.lua b/apisix/utils/batch-processor.lua
index cd831f2ad..eabee4f9e 100644
--- a/apisix/utils/batch-processor.lua
+++ b/apisix/utils/batch-processor.lua
@@ -91,6 +91,7 @@ function execute_func(premature, self, batch)
core.log.error("Batch Processor[", self.name, "] failed to process
entries [",
#batch.entries + 1 - first_fail, "/",
#batch.entries ,"]: ", err)
batch.entries = slice_batch(batch.entries, first_fail)
+ self.processed_entries = self.processed_entries + first_fail - 1
else
core.log.error("Batch Processor[", self.name,
"] failed to process entries: ", err)
@@ -101,13 +102,14 @@ function execute_func(premature, self, batch)
schedule_func_exec(self, self.retry_delay,
batch)
else
+ self.processed_entries = self.processed_entries + #batch.entries
core.log.error("Batch Processor[", self.name,"] exceeded ",
"the max_retry_count[", batch.retry_count,
"] dropping the entries")
end
return
end
-
+ self.processed_entries = self.processed_entries + #batch.entries
core.log.debug("Batch Processor[", self.name,
"] successfully processed the entries")
end
@@ -170,12 +172,12 @@ function batch_processor:new(func, config)
last_entry_t = 0,
route_id = config.route_id,
server_addr = config.server_addr,
+ processed_entries = 0
}
return setmetatable(processor, batch_processor_mt)
end
-
function batch_processor:push(entry)
-- if the batch size is one then immediately send for processing
if self.batch_max_size == 1 then
diff --git a/t/plugin/kafka-logger3.t b/t/plugin/kafka-logger3.t
new file mode 100644
index 000000000..82c8248c6
--- /dev/null
+++ b/t/plugin/kafka-logger3.t
@@ -0,0 +1,120 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+use t::APISIX 'no_plan';
+
+repeat_each(1);
+no_long_string();
+no_root_location();
+
+add_block_preprocessor(sub {
+ my ($block) = @_;
+
+ if (!$block->request) {
+ $block->set_value("request", "GET /t");
+ }
+});
+
+run_tests;
+
+__DATA__
+
+=== TEST 1: should drop entries
+--- extra_yaml_config
+plugins:
+ - kafka-logger
+--- config
+location /t {
+ content_by_lua_block {
+ local http = require "resty.http"
+ local httpc = http.new()
+ local data = {
+ {
+ input = {
+ plugins = {
+ ["kafka-logger"] = {
+ broker_list = {
+ ["127.0.0.1"] = 1234
+ },
+ kafka_topic = "test2",
+ producer_type = "sync",
+ timeout = 1,
+ batch_max_size = 1,
+ required_acks = 1,
+ meta_format = "origin",
+ max_retry_count = 10,
+ }
+ },
+ upstream = {
+ nodes = {
+ ["127.0.0.1:1980"] = 1
+ },
+ type = "roundrobin"
+ },
+ uri = "/hello",
+ },
+ },
+ }
+
+ local t = require("lib.test_admin").test
+
+ -- Set plugin metadata
+ local metadata = {
+ log_format = {
+ host = "$host",
+ ["@timestamp"] = "$time_iso8601",
+ client_ip = "$remote_addr"
+ },
+ max_pending_entries = 1
+ }
+
+ local code, body = t('/apisix/admin/plugin_metadata/kafka-logger',
ngx.HTTP_PUT, metadata)
+ if code >= 300 then
+ ngx.status = code
+ ngx.say(body)
+ return
+ end
+
+ -- Create route
+ local code, body = t('/apisix/admin/routes/1', ngx.HTTP_PUT,
data[1].input)
+ if code >= 300 then
+ ngx.status = code
+ ngx.say(body)
+ return
+ end
+
+ local uri = "http://127.0.0.1:"; .. ngx.var.server_port .. "/hello"
+ httpc:request_uri(uri, {
+ method = "GET",
+ keepalive_timeout = 1,
+ keepalive_pool = 1,
+ })
+ httpc:request_uri(uri, {
+ method = "GET",
+ keepalive_timeout = 1,
+ keepalive_pool = 1,
+ })
+ httpc:request_uri(uri, {
+ method = "GET",
+ keepalive_timeout = 1,
+ keepalive_pool = 1,
+ })
+ ngx.sleep(2)
+ }
+}
+--- error_log
+max pending entries limit exceeded. discarding entry
+--- timeout: 5
