This is an automated email from the ASF dual-hosted git repository.
nic-6443 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 54bbdc46ab fix(batch-processor): bound the pending entries backlog by
default (#13826)
54bbdc46ab is described below
commit 54bbdc46ab2c5d2dd7754ced4c5c31e76ed9a02c
Author: Nic <[email protected]>
AuthorDate: Mon Aug 17 15:03:41 2026 +0800
fix(batch-processor): bound the pending entries backlog by default (#13826)
---
apisix/plugins/clickhouse-logger.lua | 15 +-
apisix/plugins/datadog.lua | 2 +-
apisix/plugins/elasticsearch-logger.lua | 16 +-
apisix/plugins/google-cloud-logging.lua | 16 +-
apisix/plugins/http-logger.lua | 17 +-
apisix/plugins/kafka-logger.lua | 17 +-
apisix/plugins/lago.lua | 13 +-
apisix/plugins/loggly.lua | 2 +-
apisix/plugins/loki-logger.lua | 17 +-
apisix/plugins/rocketmq-logger.lua | 17 +-
apisix/plugins/skywalking-logger.lua | 17 +-
apisix/plugins/sls-logger.lua | 2 +-
apisix/plugins/splunk-hec-logging.lua | 16 +-
apisix/plugins/syslog.lua | 4 +-
apisix/plugins/syslog/init.lua | 3 +-
apisix/plugins/tcp-logger.lua | 17 +-
apisix/plugins/tencent-cloud-cls.lua | 16 +-
apisix/plugins/udp-logger.lua | 17 +-
apisix/stream/plugins/syslog.lua | 4 +-
apisix/utils/batch-processor-manager.lua | 112 ++++++--
docs/en/latest/batch-processor.md | 33 ++-
docs/en/latest/plugins/clickhouse-logger.md | 2 +-
docs/en/latest/plugins/datadog.md | 1 +
docs/en/latest/plugins/elasticsearch-logger.md | 2 +-
docs/en/latest/plugins/google-cloud-logging.md | 2 +-
docs/en/latest/plugins/http-logger.md | 11 +-
docs/en/latest/plugins/kafka-logger.md | 3 +-
docs/en/latest/plugins/lago.md | 6 +
docs/en/latest/plugins/loggly.md | 1 +
docs/en/latest/plugins/loki-logger.md | 2 +-
docs/en/latest/plugins/rocketmq-logger.md | 11 +-
docs/en/latest/plugins/skywalking-logger.md | 2 +-
docs/en/latest/plugins/sls-logger.md | 1 +
docs/en/latest/plugins/splunk-hec-logging.md | 2 +-
docs/en/latest/plugins/syslog.md | 9 +-
docs/en/latest/plugins/tcp-logger.md | 3 +-
docs/en/latest/plugins/tencent-cloud-cls.md | 3 +-
docs/en/latest/plugins/udp-logger.md | 3 +-
docs/zh/latest/batch-processor.md | 33 ++-
docs/zh/latest/plugins/clickhouse-logger.md | 2 +-
docs/zh/latest/plugins/datadog.md | 1 +
docs/zh/latest/plugins/elasticsearch-logger.md | 2 +-
docs/zh/latest/plugins/google-cloud-logging.md | 2 +-
docs/zh/latest/plugins/http-logger.md | 9 +-
docs/zh/latest/plugins/kafka-logger.md | 3 +-
docs/zh/latest/plugins/loggly.md | 1 +
docs/zh/latest/plugins/loki-logger.md | 2 +-
docs/zh/latest/plugins/rocketmq-logger.md | 9 +-
docs/zh/latest/plugins/skywalking-logger.md | 2 +-
docs/zh/latest/plugins/sls-logger.md | 1 +
docs/zh/latest/plugins/splunk-hec-logging.md | 2 +-
docs/zh/latest/plugins/syslog.md | 7 +-
docs/zh/latest/plugins/tcp-logger.md | 3 +-
docs/zh/latest/plugins/tencent-cloud-cls.md | 3 +-
docs/zh/latest/plugins/udp-logger.md | 3 +-
t/plugin/ai-proxy-kafka-log.t | 8 +-
t/plugin/kafka-logger-large-body.t | 8 +-
t/plugin/kafka-logger-log-format.t | 8 +-
t/plugin/kafka-logger.t | 8 +-
t/plugin/kafka-logger2.t | 8 +-
t/plugin/kafka-logger4.t | 8 +-
t/plugin/rocketmq-logger-log-format.t | 8 +-
t/plugin/rocketmq-logger.t | 8 +-
t/plugin/rocketmq-logger2.t | 8 +-
t/plugin/sls-logger.t | 4 +-
t/plugin/syslog.t | 8 +-
t/plugin/tcp-logger.t | 4 +-
t/plugin/udp-logger.t | 4 +-
t/utils/batch-processor-manager.t | 374 +++++++++++++++++++++++++
69 files changed, 698 insertions(+), 290 deletions(-)
diff --git a/apisix/plugins/clickhouse-logger.lua
b/apisix/plugins/clickhouse-logger.lua
index fd44e32fc1..30a8b0d354 100644
--- a/apisix/plugins/clickhouse-logger.lua
+++ b/apisix/plugins/clickhouse-logger.lua
@@ -17,7 +17,6 @@
local bp_manager_mod = require("apisix.utils.batch-processor-manager")
local log_util = require("apisix.utils.log-util")
-local plugin = require("apisix.plugin")
local core = require("apisix.core")
local http = require("resty.http")
local url = require("net.url")
@@ -79,11 +78,6 @@ local metadata_schema = {
log_format = {
type = "object"
},
- max_pending_entries = {
- type = "integer",
- description = "maximum number of pending entries in the batch
processor",
- minimum = 1,
- },
},
}
@@ -93,7 +87,7 @@ local _M = {
priority = 398,
name = plugin_name,
schema = batch_processor_manager:wrap_schema(schema),
- metadata_schema = metadata_schema,
+ metadata_schema =
batch_processor_manager:wrap_metadata_schema(metadata_schema),
}
@@ -189,12 +183,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 = log_util.get_log_entry(plugin_name, conf, ctx)
- if batch_processor_manager:add_entry(conf, entry, max_pending_entries) then
+ if batch_processor_manager:add_entry(conf, entry) then
return
end
@@ -219,7 +210,7 @@ function _M.log(conf, ctx)
return send_http_data(conf, data)
end
- batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, func,
max_pending_entries)
+ batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, func)
end
diff --git a/apisix/plugins/datadog.lua b/apisix/plugins/datadog.lua
index 05c21c6d54..66e5e8ae18 100644
--- a/apisix/plugins/datadog.lua
+++ b/apisix/plugins/datadog.lua
@@ -90,7 +90,7 @@ local _M = {
priority = 495,
name = plugin_name,
schema = batch_processor_manager:wrap_schema(schema),
- metadata_schema = metadata_schema,
+ metadata_schema =
batch_processor_manager:wrap_metadata_schema(metadata_schema),
}
diff --git a/apisix/plugins/elasticsearch-logger.lua
b/apisix/plugins/elasticsearch-logger.lua
index 2d1dc77433..0a787040b0 100644
--- a/apisix/plugins/elasticsearch-logger.lua
+++ b/apisix/plugins/elasticsearch-logger.lua
@@ -18,7 +18,6 @@ local core = require("apisix.core")
local http = require("resty.http")
local log_util = require("apisix.utils.log-util")
local bp_manager_mod = require("apisix.utils.batch-processor-manager")
-local plugin = require("apisix.plugin")
local ngx = ngx
local ngx_re = ngx.re
local str_format = core.string.format
@@ -128,11 +127,6 @@ local metadata_schema = {
log_format = {
type = "object"
},
- max_pending_entries = {
- type = "integer",
- description = "maximum number of pending entries in the batch
processor",
- minimum = 1,
- },
},
}
@@ -142,7 +136,7 @@ local _M = {
priority = 413,
name = plugin_name,
schema = batch_processor_manager:wrap_schema(schema),
- metadata_schema = metadata_schema,
+ metadata_schema =
batch_processor_manager:wrap_metadata_schema(metadata_schema),
}
@@ -340,12 +334,9 @@ end
function _M.log(conf, ctx)
local index = resolve_index_vars(conf.field.index, ctx.var)
- 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 = get_logger_entry(conf, ctx, index)
- if batch_processor_manager:add_entry(conf, entry, max_pending_entries) then
+ if batch_processor_manager:add_entry(conf, entry) then
return
end
@@ -353,8 +344,7 @@ function _M.log(conf, ctx)
return send_to_elasticsearch(conf, entries)
end
- batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx,
- process,
max_pending_entries)
+ batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx,
process)
end
_M._resolve_index_vars = resolve_index_vars
diff --git a/apisix/plugins/google-cloud-logging.lua
b/apisix/plugins/google-cloud-logging.lua
index 6133293d25..1c5af90a06 100644
--- a/apisix/plugins/google-cloud-logging.lua
+++ b/apisix/plugins/google-cloud-logging.lua
@@ -16,7 +16,6 @@
--
local core = require("apisix.core")
-local plugin = require("apisix.plugin")
local tostring = tostring
local pairs = pairs
local http = require("resty.http")
@@ -117,11 +116,6 @@ local metadata_schema = {
log_format = {
type = "object"
},
- max_pending_entries = {
- type = "integer",
- description = "maximum number of pending entries in the batch
processor",
- minimum = 1,
- },
},
}
@@ -245,7 +239,7 @@ local _M = {
version = 0.1,
priority = 407,
name = plugin_name,
- metadata_schema = metadata_schema,
+ metadata_schema =
batch_processor_manager:wrap_metadata_schema(metadata_schema),
schema = batch_processor_manager:wrap_schema(schema),
}
@@ -260,9 +254,6 @@ 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 oauth, err = core.lrucache.plugin_ctx(lrucache, ctx, nil,
create_oauth_object, conf)
if not oauth then
@@ -272,7 +263,7 @@ function _M.log(conf, ctx)
local entry = get_logger_entry(conf, ctx, oauth)
- if batch_processor_manager:add_entry(conf, entry, max_pending_entries) then
+ if batch_processor_manager:add_entry(conf, entry) then
return
end
@@ -280,8 +271,7 @@ function _M.log(conf, ctx)
return send_to_google(oauth, entries)
end
- batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx,
- process,
max_pending_entries)
+ batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx,
process)
end
diff --git a/apisix/plugins/http-logger.lua b/apisix/plugins/http-logger.lua
index a57eb35648..d7b0d4809c 100644
--- a/apisix/plugins/http-logger.lua
+++ b/apisix/plugins/http-logger.lua
@@ -15,7 +15,6 @@
-- limitations under the License.
--
local bp_manager_mod = require("apisix.utils.batch-processor-manager")
-local plugin = require("apisix.plugin")
local log_util = require("apisix.utils.log-util")
local core = require("apisix.core")
local http = require("resty.http")
@@ -25,7 +24,7 @@ local tostring = tostring
local ipairs = ipairs
local plugin_name = "http-logger"
-local batch_processor_manager = bp_manager_mod.new("http logger")
+local batch_processor_manager = bp_manager_mod.new("http logger", plugin_name)
local schema = {
type = "object",
@@ -71,11 +70,6 @@ local metadata_schema = {
log_format = {
type = "object"
},
- max_pending_entries = {
- type = "integer",
- description = "maximum number of pending entries in the batch
processor",
- minimum = 1,
- },
},
}
@@ -85,7 +79,7 @@ local _M = {
priority = 410,
name = plugin_name,
schema = batch_processor_manager:wrap_schema(schema),
- metadata_schema = metadata_schema,
+ metadata_schema =
batch_processor_manager:wrap_metadata_schema(metadata_schema),
}
@@ -183,16 +177,13 @@ 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 = log_util.get_log_entry(plugin_name, conf, ctx)
if not entry.route_id then
entry.route_id = "no-matched"
end
- if batch_processor_manager:add_entry(conf, entry, max_pending_entries) then
+ if batch_processor_manager:add_entry(conf, entry) then
return
end
@@ -234,7 +225,7 @@ function _M.log(conf, ctx)
return send_http_data(conf, data)
end
- batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, func,
max_pending_entries)
+ batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, func)
end
diff --git a/apisix/plugins/kafka-logger.lua b/apisix/plugins/kafka-logger.lua
index b9b254e2fa..012f842127 100644
--- a/apisix/plugins/kafka-logger.lua
+++ b/apisix/plugins/kafka-logger.lua
@@ -18,14 +18,13 @@ 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
local type = type
local plugin_name = "kafka-logger"
-local batch_processor_manager = bp_manager_mod.new("kafka logger")
+local batch_processor_manager = bp_manager_mod.new("kafka logger", plugin_name)
local lrucache = core.lrucache.new({
type = "plugin",
@@ -163,11 +162,6 @@ local metadata_schema = {
log_format = {
type = "object"
},
- max_pending_entries = {
- type = "integer",
- description = "maximum number of pending entries in the batch
processor",
- minimum = 1,
- },
},
}
@@ -176,7 +170,7 @@ local _M = {
priority = 403,
name = plugin_name,
schema = batch_processor_manager:wrap_schema(schema),
- metadata_schema = metadata_schema,
+ metadata_schema =
batch_processor_manager:wrap_metadata_schema(metadata_schema),
}
@@ -252,9 +246,6 @@ 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)
@@ -264,7 +255,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, max_pending_entries) then
+ if batch_processor_manager:add_entry(conf, entry) then
return
end
@@ -323,7 +314,7 @@ function _M.log(conf, ctx)
return send_kafka_data(conf, data, prod)
end
- batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, func,
max_pending_entries)
+ batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, func)
end
diff --git a/apisix/plugins/lago.lua b/apisix/plugins/lago.lua
index 3c5b1f1664..71e83559e3 100644
--- a/apisix/plugins/lago.lua
+++ b/apisix/plugins/lago.lua
@@ -25,7 +25,7 @@ local core = require("apisix.core")
local str_format = core.string.format
local plugin_name = "lago"
-local batch_processor_manager = bp_manager_mod.new("lago logger")
+local batch_processor_manager = bp_manager_mod.new("lago logger", plugin_name)
local schema = {
type = "object",
@@ -120,15 +120,26 @@ schema = batch_processor_manager:wrap_schema(schema)
schema.properties.batch_max_size.default = 100
+local metadata_schema = batch_processor_manager:wrap_metadata_schema({
+ type = "object",
+ properties = {},
+})
+
+
local _M = {
version = 0.1,
priority = 415,
name = plugin_name,
schema = schema,
+ metadata_schema = metadata_schema,
}
function _M.check_schema(conf, schema_type)
+ if schema_type == core.schema.TYPE_METADATA then
+ return core.schema.check(metadata_schema, conf)
+ end
+
local check = {"endpoint_addrs"}
core.utils.check_https(check, conf, plugin_name)
core.utils.check_tls_bool({"ssl_verify"}, conf, plugin_name)
diff --git a/apisix/plugins/loggly.lua b/apisix/plugins/loggly.lua
index a0bf9cfac7..ad57438e61 100644
--- a/apisix/plugins/loggly.lua
+++ b/apisix/plugins/loggly.lua
@@ -159,7 +159,7 @@ local _M = {
priority = 411,
name = plugin_name,
schema = batch_processor_manager:wrap_schema(schema),
- metadata_schema = metadata_schema
+ metadata_schema =
batch_processor_manager:wrap_metadata_schema(metadata_schema)
}
diff --git a/apisix/plugins/loki-logger.lua b/apisix/plugins/loki-logger.lua
index c7eb744f49..9f0828970c 100644
--- a/apisix/plugins/loki-logger.lua
+++ b/apisix/plugins/loki-logger.lua
@@ -17,7 +17,6 @@
local bp_manager_mod = require("apisix.utils.batch-processor-manager")
local log_util = require("apisix.utils.log-util")
local core = require("apisix.core")
-local plugin = require("apisix.plugin")
local http = require("resty.http")
local new_tab = require("table.new")
@@ -32,7 +31,7 @@ local ngx = ngx
local str_format = core.string.format
local plugin_name = "loki-logger"
-local batch_processor_manager = bp_manager_mod.new("loki logger")
+local batch_processor_manager = bp_manager_mod.new("loki logger", plugin_name)
local schema = {
type = "object",
@@ -125,11 +124,6 @@ local metadata_schema = {
log_format_extra = {
type = "object"
},
- max_pending_entries = {
- type = "integer",
- description = "maximum number of pending entries in the batch
processor",
- minimum = 1,
- },
},
}
@@ -139,7 +133,7 @@ local _M = {
priority = 414,
name = plugin_name,
schema = batch_processor_manager:wrap_schema(schema),
- metadata_schema = metadata_schema,
+ metadata_schema =
batch_processor_manager:wrap_metadata_schema(metadata_schema),
}
@@ -228,9 +222,6 @@ 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 = log_util.get_log_entry(plugin_name, conf, ctx)
if not entry.route_id then
@@ -258,7 +249,7 @@ function _M.log(conf, ctx)
end
entry.loki_labels = labels
- if batch_processor_manager:add_entry(conf, entry, max_pending_entries) then
+ if batch_processor_manager:add_entry(conf, entry) then
return
end
@@ -305,7 +296,7 @@ function _M.log(conf, ctx)
return send_http_data(conf, data)
end
- batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, func,
max_pending_entries)
+ batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, func)
end
diff --git a/apisix/plugins/rocketmq-logger.lua
b/apisix/plugins/rocketmq-logger.lua
index 9059e9ee23..a394297d44 100644
--- a/apisix/plugins/rocketmq-logger.lua
+++ b/apisix/plugins/rocketmq-logger.lua
@@ -15,7 +15,6 @@
-- limitations under the License.
--
local core = require("apisix.core")
-local plugin = require("apisix.plugin")
local log_util = require("apisix.utils.log-util")
local producer = require ("resty.rocketmq.producer")
local acl_rpchook = require("resty.rocketmq.acl_rpchook")
@@ -23,7 +22,7 @@ local bp_manager_mod =
require("apisix.utils.batch-processor-manager")
local type = type
local plugin_name = "rocketmq-logger"
-local batch_processor_manager = bp_manager_mod.new("rocketmq logger")
+local batch_processor_manager = bp_manager_mod.new("rocketmq logger",
plugin_name)
local lrucache = core.lrucache.new({
type = "plugin",
@@ -85,11 +84,6 @@ local metadata_schema = {
log_format = {
type = "object"
},
- max_pending_entries = {
- type = "integer",
- description = "maximum number of pending entries in the batch
processor",
- minimum = 1,
- },
},
}
@@ -98,7 +92,7 @@ local _M = {
priority = 402,
name = plugin_name,
schema = batch_processor_manager:wrap_schema(schema),
- metadata_schema = metadata_schema,
+ metadata_schema =
batch_processor_manager:wrap_metadata_schema(metadata_schema),
}
@@ -153,9 +147,6 @@ 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)
@@ -163,7 +154,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, max_pending_entries) then
+ if batch_processor_manager:add_entry(conf, entry) then
return
end
@@ -201,7 +192,7 @@ function _M.log(conf, ctx)
return send_rocketmq_data(conf, data, prod)
end
- batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, func,
max_pending_entries)
+ batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, func)
end
diff --git a/apisix/plugins/skywalking-logger.lua
b/apisix/plugins/skywalking-logger.lua
index 712dca9456..67d669f045 100644
--- a/apisix/plugins/skywalking-logger.lua
+++ b/apisix/plugins/skywalking-logger.lua
@@ -15,7 +15,6 @@
-- limitations under the License.
--
local bp_manager_mod = require("apisix.utils.batch-processor-manager")
-local plugin = require("apisix.plugin")
local log_util = require("apisix.utils.log-util")
local core = require("apisix.core")
local http = require("resty.http")
@@ -29,7 +28,7 @@ local tostring = tostring
local tonumber = tonumber
local plugin_name = "skywalking-logger"
-local batch_processor_manager = bp_manager_mod.new("skywalking logger")
+local batch_processor_manager = bp_manager_mod.new("skywalking logger",
plugin_name)
local schema = {
type = "object",
properties = {
@@ -71,11 +70,6 @@ local metadata_schema = {
log_format = {
type = "object"
},
- max_pending_entries = {
- type = "integer",
- description = "maximum number of pending entries in the batch
processor",
- minimum = 1,
- },
},
}
@@ -85,7 +79,7 @@ local _M = {
priority = 408,
name = plugin_name,
schema = batch_processor_manager:wrap_schema(schema),
- metadata_schema = metadata_schema,
+ metadata_schema =
batch_processor_manager:wrap_metadata_schema(metadata_schema),
}
@@ -153,9 +147,6 @@ 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 log_body = log_util.get_log_entry(plugin_name, conf, ctx)
local trace_context
local sw_header = ngx.req.get_headers()["sw8"]
@@ -190,7 +181,7 @@ function _M.log(conf, ctx)
endpoint = ctx.var.uri,
}
- if batch_processor_manager:add_entry(conf, entry, max_pending_entries) then
+ if batch_processor_manager:add_entry(conf, entry) then
return
end
@@ -204,7 +195,7 @@ function _M.log(conf, ctx)
return send_http_data(conf, data)
end
- batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, func,
max_pending_entries)
+ batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, func)
end
diff --git a/apisix/plugins/sls-logger.lua b/apisix/plugins/sls-logger.lua
index 65918f2bd8..d2fb1cc852 100644
--- a/apisix/plugins/sls-logger.lua
+++ b/apisix/plugins/sls-logger.lua
@@ -82,7 +82,7 @@ local _M = {
priority = 406,
name = plugin_name,
schema = batch_processor_manager:wrap_schema(schema),
- metadata_schema = metadata_schema,
+ metadata_schema =
batch_processor_manager:wrap_metadata_schema(metadata_schema),
}
function _M.check_schema(conf,schema_type)
diff --git a/apisix/plugins/splunk-hec-logging.lua
b/apisix/plugins/splunk-hec-logging.lua
index ae2f73a156..5817f8a46c 100644
--- a/apisix/plugins/splunk-hec-logging.lua
+++ b/apisix/plugins/splunk-hec-logging.lua
@@ -21,7 +21,6 @@ local ngx_now = ngx.now
local http = require("resty.http")
local log_util = require("apisix.utils.log-util")
local bp_manager_mod = require("apisix.utils.batch-processor-manager")
-local plugin = require("apisix.plugin")
local table_insert = core.table.insert
local table_concat = core.table.concat
local ipairs = ipairs
@@ -83,11 +82,6 @@ local metadata_schema = {
log_format = {
type = "object"
},
- max_pending_entries = {
- type = "integer",
- description = "maximum number of pending entries in the batch
processor",
- minimum = 1,
- }
},
}
@@ -95,7 +89,7 @@ local _M = {
version = 0.1,
priority = 409,
name = plugin_name,
- metadata_schema = metadata_schema,
+ metadata_schema =
batch_processor_manager:wrap_metadata_schema(metadata_schema),
schema = batch_processor_manager:wrap_schema(schema),
}
@@ -189,12 +183,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 = get_logger_entry(conf, ctx)
- if batch_processor_manager:add_entry(conf, entry, max_pending_entries) then
+ if batch_processor_manager:add_entry(conf, entry) then
return
end
@@ -202,8 +193,7 @@ function _M.log(conf, ctx)
return send_to_splunk(conf, entries)
end
- batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx,
- process,
max_pending_entries)
+ batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx,
process)
end
diff --git a/apisix/plugins/syslog.lua b/apisix/plugins/syslog.lua
index 078d2264b8..60124483fb 100644
--- a/apisix/plugins/syslog.lua
+++ b/apisix/plugins/syslog.lua
@@ -20,7 +20,7 @@ local bp_manager_mod =
require("apisix.utils.batch-processor-manager")
local syslog = require("apisix.plugins.syslog.init")
local plugin_name = "syslog"
-local batch_processor_manager = bp_manager_mod.new("sys logger")
+local batch_processor_manager = bp_manager_mod.new("sys logger", plugin_name)
local schema = {
type = "object",
properties = {
@@ -76,7 +76,7 @@ local _M = {
priority = 401,
name = plugin_name,
schema = schema,
- metadata_schema = metadata_schema,
+ metadata_schema =
batch_processor_manager:wrap_metadata_schema(metadata_schema),
flush_syslog = syslog.flush_syslog,
}
diff --git a/apisix/plugins/syslog/init.lua b/apisix/plugins/syslog/init.lua
index 6ca07e7d09..15f1eb2b68 100644
--- a/apisix/plugins/syslog/init.lua
+++ b/apisix/plugins/syslog/init.lua
@@ -23,7 +23,8 @@ local ipairs = ipairs
local table_insert = core.table.insert
local table_concat = core.table.concat
-local batch_processor_manager = bp_manager_mod.new("sys logger")
+-- shared by the http and the stream syslog plugin, both named "syslog"
+local batch_processor_manager = bp_manager_mod.new("sys logger", "syslog")
local lrucache = core.lrucache.new({
ttl = 300, count = 512, serial_creating = true,
diff --git a/apisix/plugins/tcp-logger.lua b/apisix/plugins/tcp-logger.lua
index a52e35b6a9..da64de4549 100644
--- a/apisix/plugins/tcp-logger.lua
+++ b/apisix/plugins/tcp-logger.lua
@@ -16,7 +16,6 @@
--
local core = require("apisix.core")
local log_util = require("apisix.utils.log-util")
-local plugin = require("apisix.plugin")
local bp_manager_mod = require("apisix.utils.batch-processor-manager")
local plugin_name = "tcp-logger"
local tostring = tostring
@@ -24,7 +23,7 @@ local ngx = ngx
local tcp = ngx.socket.tcp
-local batch_processor_manager = bp_manager_mod.new("tcp logger")
+local batch_processor_manager = bp_manager_mod.new("tcp logger", plugin_name)
local schema = {
type = "object",
properties = {
@@ -66,11 +65,6 @@ local metadata_schema = {
log_format = {
type = "object"
},
- max_pending_entries = {
- type = "integer",
- description = "maximum number of pending entries in the batch
processor",
- minimum = 1,
- },
},
}
@@ -78,7 +72,7 @@ local _M = {
version = 0.1,
priority = 405,
name = plugin_name,
- metadata_schema = metadata_schema,
+ metadata_schema =
batch_processor_manager:wrap_metadata_schema(metadata_schema),
schema = batch_processor_manager:wrap_schema(schema),
}
@@ -146,12 +140,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 = log_util.get_log_entry(plugin_name, conf, ctx)
- if batch_processor_manager:add_entry(conf, entry, max_pending_entries) then
+ if batch_processor_manager:add_entry(conf, entry) then
return
end
@@ -171,7 +162,7 @@ function _M.log(conf, ctx)
return send_tcp_data(conf, data)
end
- batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, func,
max_pending_entries)
+ batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, func)
end
diff --git a/apisix/plugins/tencent-cloud-cls.lua
b/apisix/plugins/tencent-cloud-cls.lua
index 9e176b3700..18f0cd434f 100644
--- a/apisix/plugins/tencent-cloud-cls.lua
+++ b/apisix/plugins/tencent-cloud-cls.lua
@@ -16,7 +16,6 @@
--
local core = require("apisix.core")
-local plugin = require("apisix.plugin")
local log_util = require("apisix.utils.log-util")
local bp_manager_mod = require("apisix.utils.batch-processor-manager")
local cls_sdk = require("apisix.plugins.tencent-cloud-cls.cls-sdk")
@@ -77,11 +76,6 @@ local metadata_schema = {
log_format = {
type = "object"
},
- max_pending_entries = {
- type = "integer",
- description = "maximum number of pending entries in the batch
processor",
- minimum = 1,
- },
},
}
@@ -91,7 +85,7 @@ local _M = {
priority = 397,
name = plugin_name,
schema = batch_processor_manager:wrap_schema(schema),
- metadata_schema = metadata_schema,
+ metadata_schema =
batch_processor_manager:wrap_metadata_schema(metadata_schema),
}
@@ -129,9 +123,6 @@ 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
-- sample if set
if not ctx.cls_sample then
core.log.debug("cls not sampled, skip log")
@@ -146,7 +137,7 @@ function _M.log(conf, ctx)
end
end
- if batch_processor_manager:add_entry(conf, entry, max_pending_entries) then
+ if batch_processor_manager:add_entry(conf, entry) then
return
end
@@ -162,8 +153,7 @@ function _M.log(conf, ctx)
return sdk:send_to_cls(entries)
end
- batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx,
- process,
max_pending_entries)
+ batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx,
process)
end
diff --git a/apisix/plugins/udp-logger.lua b/apisix/plugins/udp-logger.lua
index ae8b626942..79fe691f88 100644
--- a/apisix/plugins/udp-logger.lua
+++ b/apisix/plugins/udp-logger.lua
@@ -16,7 +16,6 @@
--
local core = require("apisix.core")
local log_util = require("apisix.utils.log-util")
-local plugin = require("apisix.plugin")
local bp_manager_mod = require("apisix.utils.batch-processor-manager")
local plugin_name = "udp-logger"
local tostring = tostring
@@ -24,7 +23,7 @@ local ngx = ngx
local udp = ngx.socket.udp
-local batch_processor_manager = bp_manager_mod.new("udp logger")
+local batch_processor_manager = bp_manager_mod.new("udp logger", plugin_name)
local schema = {
type = "object",
properties = {
@@ -64,11 +63,6 @@ local metadata_schema = {
log_format = {
type = "object"
},
- max_pending_entries = {
- type = "integer",
- description = "maximum number of pending entries in the batch
processor",
- minimum = 1,
- },
},
}
@@ -76,7 +70,7 @@ local _M = {
version = 0.1,
priority = 400,
name = plugin_name,
- metadata_schema = metadata_schema,
+ metadata_schema =
batch_processor_manager:wrap_metadata_schema(metadata_schema),
schema = batch_processor_manager:wrap_schema(schema),
}
@@ -131,12 +125,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 = log_util.get_log_entry(plugin_name, conf, ctx)
- if batch_processor_manager:add_entry(conf, entry, max_pending_entries) then
+ if batch_processor_manager:add_entry(conf, entry) then
return
end
@@ -156,7 +147,7 @@ function _M.log(conf, ctx)
return send_udp_data(conf, data)
end
- batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, func,
max_pending_entries)
+ batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, func)
end
return _M
diff --git a/apisix/stream/plugins/syslog.lua b/apisix/stream/plugins/syslog.lua
index 5a44ce4257..da67d8c6a7 100644
--- a/apisix/stream/plugins/syslog.lua
+++ b/apisix/stream/plugins/syslog.lua
@@ -21,7 +21,7 @@ local bp_manager_mod =
require("apisix.utils.batch-processor-manager")
local syslog = require("apisix.plugins.syslog.init")
local plugin_name = "syslog"
-local batch_processor_manager = bp_manager_mod.new("stream sys logger")
+local batch_processor_manager = bp_manager_mod.new("stream sys logger",
plugin_name)
local schema = {
type = "object",
properties = {
@@ -54,7 +54,7 @@ local _M = {
priority = 401,
name = plugin_name,
schema = schema,
- metadata_schema = metadata_schema,
+ metadata_schema =
batch_processor_manager:wrap_metadata_schema(metadata_schema),
flush_syslog = syslog.flush_syslog,
}
diff --git a/apisix/utils/batch-processor-manager.lua
b/apisix/utils/batch-processor-manager.lua
index 5ff594c056..990fd73808 100644
--- a/apisix/utils/batch-processor-manager.lua
+++ b/apisix/utils/batch-processor-manager.lua
@@ -18,21 +18,39 @@ local core = require("apisix.core")
local plugin = require("apisix.plugin")
local batch_processor = require("apisix.utils.batch-processor")
local timer_at = ngx.timer.at
+local now = ngx.now
local pairs = pairs
local setmetatable = setmetatable
+-- A pending entry keeps a whole log payload alive, including the request and
+-- response bodies when body logging is on, so a log server that is slow or
+-- unreachable turns the backlog into unbounded worker memory. Cap it by
default;
+-- the batch-processor documentation records what the cap costs per body size.
+local DEFAULT_MAX_PENDING_ENTRIES = 8192
+-- Logging one line per discarded entry would itself become a flood during the
very
+-- outage that causes the discards, so report a summary at most this often, in
seconds.
+local DISCARD_LOG_INTERVAL = 1
+
+
local _M = {}
local mt = { __index = _M }
-function _M.new(name)
+-- `name` labels the batch processor in logs and metrics and is not
necessarily the
+-- plugin's name; `plugin_name` names the plugin whose metadata carries
+-- `max_pending_entries`, and defaults to `name`.
+function _M.new(name, plugin_name)
return setmetatable({
stale_timer_running = false,
buffers = {},
total_pushed_entries = 0,
total_stale_processed_entries = 0,
+ processed_entries_snapshot = 0,
+ discarded_entries = 0,
+ last_discard_log_time = 0,
name = name,
+ plugin_name = plugin_name or name,
}, mt)
end
@@ -52,6 +70,20 @@ function _M:wrap_schema(schema)
end
+-- Every batch-processor based logger exposes the same backlog limit, so
declare it
+-- here instead of repeating it in each plugin's metadata schema.
+function _M:wrap_metadata_schema(schema)
+ schema.properties.max_pending_entries = {
+ type = "integer",
+ minimum = 1,
+ default = DEFAULT_MAX_PENDING_ENTRIES,
+ description = "maximum number of entries waiting to be processed; new "
+ .. "entries are discarded while the backlog exceeds it",
+ }
+ return schema
+end
+
+
-- remove stale objects from the memory after timer expires
local function remove_stale_objects(premature, self)
if premature then
@@ -98,16 +130,58 @@ local function total_processed_entries(self)
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
+
+local function max_pending_entries(self)
+ local metadata = plugin.plugin_metadata(self.plugin_name)
+ return metadata and metadata.value and metadata.value.max_pending_entries
+ or DEFAULT_MAX_PENDING_ENTRIES
+end
+
+
+-- The processed count only ever grows, so the last one we read is a lower
bound on
+-- the current one, and `pushed - snapshot` is therefore an upper bound on the
+-- backlog. While that bound is under the limit the backlog certainly is too,
which
+-- keeps the common path off the per-buffer walk: the walk only happens once
the
+-- bound catches up with the limit, roughly once every `max_pending_entries`
entries.
+local function backlog_is_full(self, limit)
+ if self.total_pushed_entries - self.processed_entries_snapshot < limit then
+ return false
+ end
+
+ self.processed_entries_snapshot = total_processed_entries(self)
+ return self.total_pushed_entries - self.processed_entries_snapshot >= limit
+end
+
+
+local function report_discard(self, limit)
+ self.discarded_entries = self.discarded_entries + 1
+
+ local time = now()
+ if time - self.last_discard_log_time < DISCARD_LOG_INTERVAL then
+ return
+ end
+
+ core.log.error("max pending entries limit exceeded. discarding entry.",
+ " total_pushed_entries: ", self.total_pushed_entries,
+ " total_processed_entries: ",
self.processed_entries_snapshot,
+ " max_pending_entries: ", limit,
+ " discarded_entries: ", self.discarded_entries)
+ self.last_discard_log_time = time
+ self.discarded_entries = 0
+end
+
+
+-- Returns true once the entry needs nothing further from the caller, which
covers
+-- both pushing it to an existing processor and discarding it over the backlog
+-- limit; false means there is no processor yet and the caller should build
one.
+-- Reporting a discard as handled matters on the path where it happens: the
caller
+-- would otherwise build the batch-processing closure and take the backlog
check a
+-- second time, for every request, throughout the outage that caused the
discards.
+function _M:add_entry(conf, entry)
+ local limit = max_pending_entries(self)
+ if backlog_is_full(self, limit) then
+ report_discard(self, limit)
+ return true
end
check_stale(self)
@@ -122,16 +196,12 @@ function _M:add_entry(conf, entry, max_pending_entries)
end
-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
+function _M:add_entry_to_new_processor(conf, entry, ctx, func)
+ -- add_entry() reports a discard as handled, so the usual caller never
arrives
+ -- here over the limit; this keeps a caller that skips add_entry()
bounded, and
+ -- deliberately does not count, to leave the discard total to add_entry().
+ if backlog_is_full(self, max_pending_entries(self)) then
+ return
end
check_stale(self)
diff --git a/docs/en/latest/batch-processor.md
b/docs/en/latest/batch-processor.md
index 48c1a9d39a..01be6ef42c 100644
--- a/docs/en/latest/batch-processor.md
+++ b/docs/en/latest/batch-processor.md
@@ -39,6 +39,29 @@ or when the buffer duration exceeds.
| max_retry_count | integer | optional | 0 | [0,...] | Maximum
number of retries before removing the entry from the processing pipeline when
an error occurs. |
| retry_delay | integer | optional | 1 | [0,...] | Number of
seconds the process execution should be delayed if the execution fails. |
+## Limiting the backlog
+
+Entries that have been buffered but not yet delivered are held in the worker's
memory. When the log server is slow or unreachable, entries arrive faster than
they leave and the backlog — along with the worker's memory — grows with the
request rate.
+
+Every logger built on the batch processor therefore accepts a
`max_pending_entries` limit through its [plugin
metadata](./terminology/plugin-metadata.md), which defaults to `8192`. While
the backlog exceeds the limit, new entries are discarded and a summary is
written to the error log at most once per second:
+
+```text
+max pending entries limit exceeded. discarding entry. total_pushed_entries:
12289 total_processed_entries: 4096 max_pending_entries: 8192
discarded_entries: 3172
+```
+
+The limit counts entries, so what it costs in memory depends on how large each
entry is — above all on whether `include_req_body` and `include_resp_body` are
enabled and how large those bodies are. The figures below are the peak growth
in a worker's resident memory while its log server accepted connections but
never answered, measured with `http-logger`, both bodies logged, and otherwise
stock batch processor settings:
+
+| Body logged per request | Peak worker memory at the default limit |
+|-------------------------|-----------------------------------------|
+| bodies not logged | ~40 MB |
+| 1 KB request + 1 KB response | ~100 MB |
+| 4 KB request + 4 KB response | ~250 MB |
+| 16 KB request + 16 KB response | ~840 MB |
+
+The cost grows roughly in proportion to the body size, so lower
`max_pending_entries` if you log bodies larger than a few KB. The figures are
higher than the entries alone would account for because batches already handed
to the sender hold both their entries and the serialized payload built from
them.
+
+The limit only comes into play when delivery falls behind. With a log server
that keeps up, the backlog stays close to `batch_max_size` — under 1000 entries
at 3000 requests per second in the same setup — so the default leaves about
eight times the room healthy operation needs. Raising `batch_max_size` raises
the healthy backlog with it, so raise `max_pending_entries` too if you do.
+
The following code shows an example of how to use batch processor in your
plugin:
```lua
@@ -46,12 +69,17 @@ local bp_manager_mod =
require("apisix.utils.batch-processor-manager")
...
local plugin_name = "xxx-logger"
-local batch_processor_manager = bp_manager_mod.new(plugin_name)
+-- the second argument names the plugin whose metadata carries
max_pending_entries,
+-- and is only needed when the batch processor's own name differs from it
+local batch_processor_manager = bp_manager_mod.new("xxx logger", plugin_name)
local schema = {...}
+local metadata_schema = {...}
local _M = {
...
name = plugin_name,
schema = batch_processor_manager:wrap_schema(schema),
+ -- adds max_pending_entries to the plugin's metadata schema
+ metadata_schema =
batch_processor_manager:wrap_metadata_schema(metadata_schema),
}
...
@@ -60,6 +88,9 @@ local _M = {
function _M.log(conf, ctx)
local entry = {...} -- data to log
+ -- a true return means the entry needs nothing further from you: it was
either
+ -- pushed to an existing processor, or discarded because the backlog is
over
+ -- max_pending_entries
if batch_processor_manager:add_entry(conf, entry) then
return
end
diff --git a/docs/en/latest/plugins/clickhouse-logger.md
b/docs/en/latest/plugins/clickhouse-logger.md
index 5b530bc38f..ef95a7ee40 100644
--- a/docs/en/latest/plugins/clickhouse-logger.md
+++ b/docs/en/latest/plugins/clickhouse-logger.md
@@ -73,7 +73,7 @@ This Plugin supports using batch processors to aggregate and
process entries (lo
|--------------------|---------|----------|---------|--------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| log_format | object | False | | | Custom
log format using key-value pairs in JSON format. Values can reference [APISIX
variables](../apisix-variable.md) or [NGINX
variables](https://nginx.org/en/docs/http/ngx_http_core_module.html) by
prefixing with `$`. This configuration is global and applies to all Routes and
Services that use the `clickhouse-logger` Plugin. |
| log_format_extra | object | False | | |
Extra log fields **added on top of** the default log entry, keeping every
default field instead of replacing them (unlike `log_format`). Same value
syntax as `log_format`. Ignored when `log_format` is set. |
-| max_pending_entries | integer | False | | >= 1 | Maximum
number of unprocessed entries allowed in the batch processor. When this limit
is reached, new entries will be dropped until the backlog is reduced.
|
+| max_pending_entries | integer | False | 8192 | >= 1 | Maximum number of
entries waiting to be processed. New entries are discarded while the backlog
exceeds this, which stops a slow or unreachable log server from growing the
worker's memory without bound. See [Batch
Processor](../batch-processor.md#limiting-the-backlog) for the memory a backlog
of this size costs. |
## Examples
diff --git a/docs/en/latest/plugins/datadog.md
b/docs/en/latest/plugins/datadog.md
index e5bd330e5a..c282c92fff 100644
--- a/docs/en/latest/plugins/datadog.md
+++ b/docs/en/latest/plugins/datadog.md
@@ -63,6 +63,7 @@ You can configure the Plugin through Plugin metadata.
| port | integer | False | 8125 | DogStatsD server
port.
|
| namespace | string | False | "apisix" | Prefix for all
custom metrics sent by the APISIX agent. Useful for finding entities for
metrics graph. For example, `apisix.request.counter`. |
| constant_tags | array | False | [ "source:apisix" ] | Static tags to
embed into generated metrics. Useful for grouping metrics over certain signals.
See [defining
tags](https://docs.datadoghq.com/getting_started/tagging/#defining-tags) for
more. |
+| max_pending_entries | integer | False | 8192 | Maximum number of entries
waiting to be processed. New entries are discarded while the backlog exceeds
this, which stops a slow or unreachable log server from growing the worker's
memory without bound. See [Batch
Processor](../batch-processor.md#limiting-the-backlog) for the memory a backlog
of this size costs. |
## Metrics
diff --git a/docs/en/latest/plugins/elasticsearch-logger.md
b/docs/en/latest/plugins/elasticsearch-logger.md
index a6c15f14af..d4ea6c4fb8 100644
--- a/docs/en/latest/plugins/elasticsearch-logger.md
+++ b/docs/en/latest/plugins/elasticsearch-logger.md
@@ -67,7 +67,7 @@ This Plugin supports using batch processors to aggregate and
process entries (lo
|------|------|----------|---------|-------------|
| log_format | object | False | | Log format declared as key-value pairs in
JSON. Values support strings and nested objects (up to five levels deep; deeper
fields are truncated). Within strings, [APISIX](../apisix-variable.md) or
[NGINX](http://nginx.org/en/docs/varindex.html) variables can be referenced by
prefixing with `$`. |
| log_format_extra | object | False | | Extra log fields **added on top of**
the default log entry, keeping every default field instead of replacing them
(unlike `log_format`). Same value syntax as `log_format`. Ignored when
`log_format` is set. |
-| max_pending_entries | integer | False | | Maximum number of pending entries
that can be buffered in batch processor before it starts dropping them. |
+| max_pending_entries | integer | False | 8192 | Maximum number of entries
waiting to be processed. New entries are discarded while the backlog exceeds
this, which stops a slow or unreachable log server from growing the worker's
memory without bound. See [Batch
Processor](../batch-processor.md#limiting-the-backlog) for the memory a backlog
of this size costs. |
## Examples
diff --git a/docs/en/latest/plugins/google-cloud-logging.md
b/docs/en/latest/plugins/google-cloud-logging.md
index 0e1b473649..0518a853fa 100644
--- a/docs/en/latest/plugins/google-cloud-logging.md
+++ b/docs/en/latest/plugins/google-cloud-logging.md
@@ -68,7 +68,7 @@ This Plugin supports using batch processors to aggregate and
process entries (lo
|--------------------|---------|----------|---------|--------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| log_format | object | False | | | Custom
log format using key-value pairs in JSON format. Values can reference [APISIX
variables](../apisix-variable.md) or [NGINX
variables](https://nginx.org/en/docs/http/ngx_http_core_module.html) by
prefixing with `$`. This configuration is global and applies to all Routes and
Services that use the `google-cloud-logging` Plugin. |
| log_format_extra | object | False | | |
Extra log fields **added on top of** the default log entry, keeping every
default field instead of replacing them (unlike `log_format`). Same value
syntax as `log_format`. Ignored when `log_format` is set. |
-| max_pending_entries | integer | False | | >= 1 | Maximum
number of unprocessed entries allowed in the batch processor. When this limit
is reached, new entries will be dropped until the backlog is reduced.
|
+| max_pending_entries | integer | False | 8192 | >= 1 | Maximum number of
entries waiting to be processed. New entries are discarded while the backlog
exceeds this, which stops a slow or unreachable log server from growing the
worker's memory without bound. See [Batch
Processor](../batch-processor.md#limiting-the-backlog) for the memory a backlog
of this size costs. |
## Examples
diff --git a/docs/en/latest/plugins/http-logger.md
b/docs/en/latest/plugins/http-logger.md
index c511f3f2f7..4f26a3dcde 100644
--- a/docs/en/latest/plugins/http-logger.md
+++ b/docs/en/latest/plugins/http-logger.md
@@ -63,12 +63,11 @@ This Plugin supports using batch processors to aggregate
and process entries (lo
You can also set the format of the logs by configuring the Plugin metadata.
The following configurations are available:
-| Name | Type | Required | Description
|
-|---------------------|---------|----------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| log_format | object | False | Custom log format using key-value
pairs in JSON format. Values can reference [NGINX
variables](https://nginx.org/en/docs/http/ngx_http_core_module.html).
|
-| log_format_extra | object | False | Extra log fields **added on
top of** the default log entry, keeping every default field instead of
replacing them (unlike `log_format`). Same value syntax as `log_format`.
Ignored when `log_format` is set. |
-| max_pending_entries | integer | False | Maximum number of unprocessed
entries allowed in the batch processor. When this limit is reached, new entries
will be dropped until the backlog is reduced. Available in APISIX from version
3.15.0. |
-
+| Name | Type | Required | Default | Description
|
+|---------------------|---------|----------|---------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| log_format | object | False | | Custom log format using
key-value pairs in JSON format. Values can reference [NGINX
variables](https://nginx.org/en/docs/http/ngx_http_core_module.html).
|
+| log_format_extra | object | False | | Extra log fields
**added on top of** the default log entry, keeping every default field instead
of replacing them (unlike `log_format`). Same value syntax as `log_format`.
Ignored when `log_format` is set. |
+| max_pending_entries | integer | False | 8192 | Maximum number of entries
waiting to be processed. New entries are discarded while the backlog exceeds
this, which stops a slow or unreachable log server from growing the worker's
memory without bound. See [Batch
Processor](../batch-processor.md#limiting-the-backlog) for the memory a backlog
of this size costs. |
:::info IMPORTANT
Configuring the Plugin metadata is global in scope. This means that it will
take effect on all Routes and Services which use the `http-logger` Plugin.
diff --git a/docs/en/latest/plugins/kafka-logger.md
b/docs/en/latest/plugins/kafka-logger.md
index 22ecb3c3fb..61081bca67 100644
--- a/docs/en/latest/plugins/kafka-logger.md
+++ b/docs/en/latest/plugins/kafka-logger.md
@@ -148,8 +148,7 @@ You can also set the format of the logs by configuring the
Plugin metadata. The
| ------------------- | ------- | -------- | ------- |
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
|
| log_format | object | False | | Log format declared as
key-value pairs in JSON. Values support strings and nested objects (up to five
levels deep; deeper fields are truncated). Within strings,
[APISIX](../apisix-variable.md) or
[NGINX](http://nginx.org/en/docs/varindex.html) variables can be referenced by
prefixing with `$`. |
| log_format_extra | object | False | | Extra log fields
**added on top of** the default log entry, keeping every default field instead
of replacing them (unlike `log_format`). Same value syntax as `log_format`.
Ignored when `log_format` is set. |
-| max_pending_entries | integer | False | | Maximum number of
pending entries that can be buffered in the batch processor before it starts
dropping them.
|
-
+| max_pending_entries | integer | False | 8192 | Maximum number of entries
waiting to be processed. New entries are discarded while the backlog exceeds
this, which stops a slow or unreachable log server from growing the worker's
memory without bound. See [Batch
Processor](../batch-processor.md#limiting-the-backlog) for the memory a backlog
of this size costs. |
:::info IMPORTANT
Configuring the Plugin metadata is global in scope. This means that it will
take effect on all Routes and Services which use the `kafka-logger` Plugin.
diff --git a/docs/en/latest/plugins/lago.md b/docs/en/latest/plugins/lago.md
index c2b445f295..dd503ed55c 100644
--- a/docs/en/latest/plugins/lago.md
+++ b/docs/en/latest/plugins/lago.md
@@ -66,6 +66,12 @@ When enabled, the plugin will collect information from the
request context (e.g.
This Plugin supports using batch processors to aggregate and process events in
a batch. This avoids the need for frequently submitting the data. The batch
processor submits data every `5` seconds or when the data in the queue reaches
`1000`. See [Batch Processor](../batch-processor.md#configuration) for more
information or setting your custom configuration.
+## Plugin Metadata
+
+| Name | Type | Required | Default | Description |
+|------|------|----------|---------|-------------|
+| max_pending_entries | integer | False | 8192 | Maximum number of entries
waiting to be processed. New entries are discarded while the backlog exceeds
this, which stops a slow or unreachable Lago service from growing the worker's
memory without bound. See [Batch
Processor](../batch-processor.md#limiting-the-backlog) for the memory a backlog
of this size costs. |
+
## Examples
The examples below demonstrate how you can configure `lago` Plugin for typical
scenario.
diff --git a/docs/en/latest/plugins/loggly.md b/docs/en/latest/plugins/loggly.md
index e672f9fcba..60e46371e8 100644
--- a/docs/en/latest/plugins/loggly.md
+++ b/docs/en/latest/plugins/loggly.md
@@ -74,6 +74,7 @@ You can also configure the Plugin through Plugin metadata.
The following configu
| protocol | string | False | "syslog" | [ "syslog" ,
"http", "https" ] | Protocol in which the logs are sent to Loggly.
|
| log_format | object | False | nil |
| Log format declared as key-value pairs in JSON. Values support
strings and nested objects (up to five levels deep; deeper fields are
truncated). Within strings, [APISIX](../apisix-variable.md) or
[NGINX](http://nginx.org/en/docs/varindex.html) variables can be referenced by
prefixing with `$`. |
| log_format_extra | object | False | |
| Extra log fields **added on top of** the default log entry,
keeping every default field instead of replacing them (unlike `log_format`).
Same value syntax as `log_format`. Ignored when `log_format` is set. |
+| max_pending_entries | integer | False | 8192 | >= 1 | Maximum number of
entries waiting to be processed. New entries are discarded while the backlog
exceeds this, which stops a slow or unreachable log server from growing the
worker's memory without bound. See [Batch
Processor](../batch-processor.md#limiting-the-backlog) for the memory a backlog
of this size costs. |
We support
[Syslog](https://documentation.solarwinds.com/en/success_center/loggly/content/admin/streaming-syslog-without-using-files.htm),
[HTTP/S](https://documentation.solarwinds.com/en/success_center/loggly/content/admin/http-bulk-endpoint.htm)
(bulk endpoint) protocols to send log events to Loggly. By default, in APISIX
side, the protocol is set to "syslog". It lets you send RFC5424 compliant
syslog events with some fine-grained control (log severity mapping based on
upstream HTTP re [...]
diff --git a/docs/en/latest/plugins/loki-logger.md
b/docs/en/latest/plugins/loki-logger.md
index 6b7bd6c93c..43e052a1b6 100644
--- a/docs/en/latest/plugins/loki-logger.md
+++ b/docs/en/latest/plugins/loki-logger.md
@@ -72,7 +72,7 @@ You can also configure log format on a global scale using the
[Plugin Metadata](
|------|------|----------|---------|--------------|-------------|
| log_format | object | False | | | Custom log format as key-value pairs in
JSON. Setting this **replaces** the default log entry with a flat custom
format. Values support strings and nested objects (up to five levels deep;
deeper fields are truncated). Within strings, [APISIX
variables](../apisix-variable.md) and [NGINX
variables](http://nginx.org/en/docs/varindex.html) can be referenced by
prefixing with `$`. |
| log_format_extra | object | False | | | Extra log fields **added on top
of** the default log entry, keeping every default field instead of replacing
them (unlike `log_format`). Same value syntax as `log_format`. Ignored when
`log_format` is set. |
-| max_pending_entries | integer | False | | | Maximum number of pending
entries that can be buffered in batch processor before it starts dropping them.
|
+| max_pending_entries | integer | False | 8192 | >= 1 | Maximum number of
entries waiting to be processed. New entries are discarded while the backlog
exceeds this, which stops a slow or unreachable log server from growing the
worker's memory without bound. See [Batch
Processor](../batch-processor.md#limiting-the-backlog) for the memory a backlog
of this size costs. |
## Examples
diff --git a/docs/en/latest/plugins/rocketmq-logger.md
b/docs/en/latest/plugins/rocketmq-logger.md
index 2b3d2a9271..02aa93d341 100644
--- a/docs/en/latest/plugins/rocketmq-logger.md
+++ b/docs/en/latest/plugins/rocketmq-logger.md
@@ -68,12 +68,11 @@ NOTE: `encrypt_fields = {"secret_key"}` is also defined in
the schema, which mea
You can also set the format of the logs by configuring the Plugin metadata.
The following configurations are available:
-| Name | Type | Required | Description
|
-|---------------------|---------|----------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| log_format | object | False | Custom log format using key-value
pairs in JSON format. Values can reference [NGINX
variables](https://nginx.org/en/docs/http/ngx_http_core_module.html).
|
-| log_format_extra | object | False | Extra log fields **added on
top of** the default log entry, keeping every default field instead of
replacing them (unlike `log_format`). Same value syntax as `log_format`.
Ignored when `log_format` is set. |
-| max_pending_entries | integer | False | Maximum number of unprocessed
entries allowed in the batch processor. When this limit is reached, new entries
will be dropped until the backlog is reduced. Available in APISIX from version
3.15.0. |
-
+| Name | Type | Required | Default | Description
|
+|---------------------|---------|----------|---------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| log_format | object | False | | Custom log format using
key-value pairs in JSON format. Values can reference [NGINX
variables](https://nginx.org/en/docs/http/ngx_http_core_module.html).
|
+| log_format_extra | object | False | | Extra log fields
**added on top of** the default log entry, keeping every default field instead
of replacing them (unlike `log_format`). Same value syntax as `log_format`.
Ignored when `log_format` is set. |
+| max_pending_entries | integer | False | 8192 | Maximum number of entries
waiting to be processed. New entries are discarded while the backlog exceeds
this, which stops a slow or unreachable log server from growing the worker's
memory without bound. See [Batch
Processor](../batch-processor.md#limiting-the-backlog) for the memory a backlog
of this size costs. |
:::info IMPORTANT
Configuring the Plugin metadata is global in scope. This means that it will
take effect on all Routes and Services which use the `rocketmq-logger` Plugin.
diff --git a/docs/en/latest/plugins/skywalking-logger.md
b/docs/en/latest/plugins/skywalking-logger.md
index d61589e7fa..f527bdf854 100644
--- a/docs/en/latest/plugins/skywalking-logger.md
+++ b/docs/en/latest/plugins/skywalking-logger.md
@@ -66,7 +66,7 @@ You can also set the format of the logs by configuring the
Plugin metadata. The
| ---------- | ------ | -------- |
----------------------------------------------------------------------------- |
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
|
| log_format | object | False | | Custom log format as key-value pairs in
JSON. Values support strings and nested objects (up to five levels deep; deeper
fields are truncated). Within strings, [APISIX](../apisix-variable.md) or
[NGINX variables](http://nginx.org/en/docs/varindex.html) can be referenced by
prefixing with `$`. |
| log_format_extra | object | False | | Extra log fields **added on top
of** the default log entry, keeping every default field instead of replacing
them (unlike `log_format`). Same value syntax as `log_format`. Ignored when
`log_format` is set. |
-| max_pending_entries | integer | False | | Maximum number of pending entries
that can be buffered in batch processor before it starts dropping them. |
+| max_pending_entries | integer | False | 8192 | Maximum number of entries
waiting to be processed. New entries are discarded while the backlog exceeds
this, which stops a slow or unreachable log server from growing the worker's
memory without bound. See [Batch
Processor](../batch-processor.md#limiting-the-backlog) for the memory a backlog
of this size costs. |
## Examples
diff --git a/docs/en/latest/plugins/sls-logger.md
b/docs/en/latest/plugins/sls-logger.md
index cd53f5efce..1ccfadc157 100644
--- a/docs/en/latest/plugins/sls-logger.md
+++ b/docs/en/latest/plugins/sls-logger.md
@@ -94,6 +94,7 @@ You can also set the format of the logs by configuring the
Plugin metadata. The
| ---------- | ------ | -------- |
----------------------------------------------------------------------------- |
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
|
| log_format | object | False | | Log format declared as key-value pairs
in JSON. Values support strings and nested objects (up to five levels deep;
deeper fields are truncated). Within strings, [APISIX](../apisix-variable.md)
or [NGINX](http://nginx.org/en/docs/varindex.html) variables can be referenced
by prefixing with `$`. |
| log_format_extra | object | False | | Extra log fields **added on top
of** the default log entry, keeping every default field instead of replacing
them (unlike `log_format`). Same value syntax as `log_format`. Ignored when
`log_format` is set. |
+| max_pending_entries | integer | False | 8192 | Maximum number of entries
waiting to be processed. New entries are discarded while the backlog exceeds
this, which stops a slow or unreachable log server from growing the worker's
memory without bound. See [Batch
Processor](../batch-processor.md#limiting-the-backlog) for the memory a backlog
of this size costs. |
:::info IMPORTANT
diff --git a/docs/en/latest/plugins/splunk-hec-logging.md
b/docs/en/latest/plugins/splunk-hec-logging.md
index 6ebab57ce6..df3c5e2c4e 100644
--- a/docs/en/latest/plugins/splunk-hec-logging.md
+++ b/docs/en/latest/plugins/splunk-hec-logging.md
@@ -64,7 +64,7 @@ This Plugin supports using batch processors to aggregate and
process entries (lo
|--------------------|---------|----------|---------|--------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| log_format | object | False | | | Custom
log format using key-value pairs in JSON format. Values can reference [APISIX
variables](../apisix-variable.md) or [NGINX
variables](https://nginx.org/en/docs/http/ngx_http_core_module.html) by
prefixing with `$`. This configuration is global and applies to all Routes and
Services that use the `splunk-hec-logging` Plugin. |
| log_format_extra | object | False | | |
Extra log fields **added on top of** the default log entry, keeping every
default field instead of replacing them (unlike `log_format`). Same value
syntax as `log_format`. Ignored when `log_format` is set. |
-| max_pending_entries | integer | False | | >= 1 | Maximum
number of unprocessed entries allowed in the batch processor. When this limit
is reached, new entries will be dropped until the backlog is reduced.
|
+| max_pending_entries | integer | False | 8192 | >= 1 | Maximum number of
entries waiting to be processed. New entries are discarded while the backlog
exceeds this, which stops a slow or unreachable log server from growing the
worker's memory without bound. See [Batch
Processor](../batch-processor.md#limiting-the-backlog) for the memory a backlog
of this size costs. |
## Examples
diff --git a/docs/en/latest/plugins/syslog.md b/docs/en/latest/plugins/syslog.md
index 787c695d13..b7e76a7030 100644
--- a/docs/en/latest/plugins/syslog.md
+++ b/docs/en/latest/plugins/syslog.md
@@ -71,10 +71,11 @@ This Plugin supports using batch processors to aggregate
and process entries (lo
You can also set the format of the logs by configuring the Plugin metadata.
The following configurations are available:
-| Name | Type | Required | Description
|
-|------------|--------|----------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| log_format | object | False | Custom log format using key-value pairs in
JSON format. Values can reference [NGINX
variables](https://nginx.org/en/docs/http/ngx_http_core_module.html).
|
-| log_format_extra | object | False | Extra log fields **added on top of**
the default log entry, keeping every default field instead of replacing them
(unlike `log_format`). Same value syntax as `log_format`. Ignored when
`log_format` is set. |
+| Name | Type | Required | Default | Description
|
+|------------|--------|----------|---------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| log_format | object | False | | Custom log format using key-value
pairs in JSON format. Values can reference [NGINX
variables](https://nginx.org/en/docs/http/ngx_http_core_module.html).
|
+| log_format_extra | object | False | | Extra log fields **added on
top of** the default log entry, keeping every default field instead of
replacing them (unlike `log_format`). Same value syntax as `log_format`.
Ignored when `log_format` is set. |
+| max_pending_entries | integer | False | 8192 | Maximum number of entries
waiting to be processed. New entries are discarded while the backlog exceeds
this, which stops a slow or unreachable log server from growing the worker's
memory without bound. See [Batch
Processor](../batch-processor.md#limiting-the-backlog) for the memory a backlog
of this size costs. |
:::info IMPORTANT
diff --git a/docs/en/latest/plugins/tcp-logger.md
b/docs/en/latest/plugins/tcp-logger.md
index ac284d4143..cb29b3ac3f 100644
--- a/docs/en/latest/plugins/tcp-logger.md
+++ b/docs/en/latest/plugins/tcp-logger.md
@@ -104,8 +104,7 @@ You can also set the format of the logs by configuring the
Plugin metadata. The
| ---------- | ------ | -------- |
----------------------------------------------------------------------------- |
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
|
| log_format | object | False | | Log format declared as key-value pairs
in JSON. Values support strings and nested objects (up to five levels deep;
deeper fields are truncated). Within strings, [APISIX](../apisix-variable.md)
or [NGINX](http://nginx.org/en/docs/varindex.html) variables can be referenced
by prefixing with `$`. |
| log_format_extra | object | False | | Extra log fields **added on top
of** the default log entry, keeping every default field instead of replacing
them (unlike `log_format`). Same value syntax as `log_format`. Ignored when
`log_format` is set. |
-| max_pending_entries | integer | False | | Maximum number of pending entries
that can be buffered in batch processor before it starts dropping them. |
-
+| max_pending_entries | integer | False | 8192 | Maximum number of entries
waiting to be processed. New entries are discarded while the backlog exceeds
this, which stops a slow or unreachable log server from growing the worker's
memory without bound. See [Batch
Processor](../batch-processor.md#limiting-the-backlog) for the memory a backlog
of this size costs. |
:::info IMPORTANT
Configuring the Plugin metadata is global in scope. This means that it will
take effect on all Routes and Services which use the `tcp-logger` Plugin.
diff --git a/docs/en/latest/plugins/tencent-cloud-cls.md
b/docs/en/latest/plugins/tencent-cloud-cls.md
index 980072d3a3..6b02a4cf7c 100644
--- a/docs/en/latest/plugins/tencent-cloud-cls.md
+++ b/docs/en/latest/plugins/tencent-cloud-cls.md
@@ -104,8 +104,7 @@ You can also set the format of the logs by configuring the
Plugin metadata. The
| ---------- | ------ | -------- |
----------------------------------------------------------------------------- |
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
|
| log_format | object | False | | Log format declared as key-value pairs
in JSON. Values support strings and nested objects (up to five levels deep;
deeper fields are truncated). Within strings, [APISIX](../apisix-variable.md)
or [NGINX](http://nginx.org/en/docs/varindex.html) variables can be referenced
by prefixing with `$`. |
| log_format_extra | object | False | | Extra log fields **added on top
of** the default log entry, keeping every default field instead of replacing
them (unlike `log_format`). Same value syntax as `log_format`. Ignored when
`log_format` is set. |
-| max_pending_entries | integer | False | | Maximum number of pending entries
that can be buffered in batch processor before it starts dropping them. |
-
+| max_pending_entries | integer | False | 8192 | Maximum number of entries
waiting to be processed. New entries are discarded while the backlog exceeds
this, which stops a slow or unreachable log server from growing the worker's
memory without bound. See [Batch
Processor](../batch-processor.md#limiting-the-backlog) for the memory a backlog
of this size costs. |
:::info IMPORTANT
Configuring the Plugin metadata is global in scope. This means that it will
take effect on all Routes and Services which use the `tencent-cloud-cls` Plugin.
diff --git a/docs/en/latest/plugins/udp-logger.md
b/docs/en/latest/plugins/udp-logger.md
index 9eca52e934..02c2b3c37f 100644
--- a/docs/en/latest/plugins/udp-logger.md
+++ b/docs/en/latest/plugins/udp-logger.md
@@ -102,8 +102,7 @@ You can also set the format of the logs by configuring the
Plugin metadata. The
| ---------- | ------ | -------- |
----------------------------------------------------------------------------- |
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
|
| log_format | object | False | | Log format declared as key-value pairs
in JSON. Values support strings and nested objects (up to five levels deep;
deeper fields are truncated). Within strings, [APISIX](../apisix-variable.md)
or [NGINX](http://nginx.org/en/docs/varindex.html) variables can be referenced
by prefixing with `$`. |
| log_format_extra | object | False | | Extra log fields **added on top
of** the default log entry, keeping every default field instead of replacing
them (unlike `log_format`). Same value syntax as `log_format`. Ignored when
`log_format` is set. |
-| max_pending_entries | integer | False | | Maximum number of pending entries
that can be buffered in batch processor before it starts dropping them. |
-
+| max_pending_entries | integer | False | 8192 | Maximum number of entries
waiting to be processed. New entries are discarded while the backlog exceeds
this, which stops a slow or unreachable log server from growing the worker's
memory without bound. See [Batch
Processor](../batch-processor.md#limiting-the-backlog) for the memory a backlog
of this size costs. |
:::info IMPORTANT
Configuring the Plugin metadata is global in scope. This means that it will
take effect on all Routes and Services which use the `udp-logger` Plugin.
diff --git a/docs/zh/latest/batch-processor.md
b/docs/zh/latest/batch-processor.md
index b8f0290ffc..88ce01832c 100644
--- a/docs/zh/latest/batch-processor.md
+++ b/docs/zh/latest/batch-processor.md
@@ -36,6 +36,30 @@ title: 批处理器
| buffer_duration | integer | 可选 | 60 | [1,...] |
必须先处理批次中最旧条目的最长期限（以秒为单位）。 |
| max_retry_count | integer | 可选 | 0 | [0,...] | 从处理管道中移除之前的最大重试次数。
|
| retry_delay | integer | 可选 | 1 | [0,...] | 如果执行失败，则应延迟执行流程的秒数。
|
+
+## 限制积压条目数
+
+已缓冲但尚未发送成功的条目保存在 worker 内存中。当日志服务变慢或不可达时，条目进入的速度快于离开的速度，积压量以及 worker
内存会随请求速率不断增长。
+
+因此所有基于批处理器的日志插件都通过[插件元数据](./terminology/plugin-metadata.md)提供
`max_pending_entries` 上限，默认值为 `8192`。积压超过该上限期间新条目会被丢弃，并且每秒最多向错误日志输出一条汇总信息：
+
+```text
+max pending entries limit exceeded. discarding entry. total_pushed_entries:
12289 total_processed_entries: 4096 max_pending_entries: 8192
discarded_entries: 3172
+```
+
+该上限限制的是条目数量，因此实际占用多少内存取决于单条条目有多大——尤其取决于是否开启 `include_req_body` 和
`include_resp_body`，以及 body 的大小。下表是日志服务接受连接但始终不响应时 worker 常驻内存的峰值增长，测试使用
`http-logger`、同时收集请求和响应 body，批处理器其余配置保持默认：
+
+| 每请求收集的 body | 默认上限下 worker 内存峰值 |
+|-------------------|----------------------------|
+| 不收集 body | ~40 MB |
+| 1 KB 请求 + 1 KB 响应 | ~100 MB |
+| 4 KB 请求 + 4 KB 响应 | ~250 MB |
+| 16 KB 请求 + 16 KB 响应 | ~840 MB |
+
+内存开销与 body 大小大致成正比，因此如果收集的 body 超过几 KB，应调低
`max_pending_entries`。表中数值高于条目本身的体积，是因为已经交给发送方的批次同时持有条目和由其序列化出的载荷。
+
+只有在发送跟不上时该上限才会起作用。日志服务正常的情况下，积压量接近 `batch_max_size`——在相同环境下 3000 QPS 时不足 1000
条，因此默认值为正常运行留出了约 8 倍余量。调大 `batch_max_size` 会同步抬高正常状态下的积压量，此时也应相应调大
`max_pending_entries`。
+
以下代码显示了如何在你的插件中使用批处理器：
```lua
@@ -43,12 +67,17 @@ local bp_manager_mod =
require("apisix.utils.batch-processor-manager")
...
local plugin_name = "xxx-logger"
-local batch_processor_manager = bp_manager_mod.new(plugin_name)
+-- 第二个参数指定 max_pending_entries 所在的插件元数据名称，
+-- 仅当批处理器自身的名称与插件名不同时才需要传入
+local batch_processor_manager = bp_manager_mod.new("xxx logger", plugin_name)
local schema = {...}
+local metadata_schema = {...}
local _M = {
...
name = plugin_name,
schema = batch_processor_manager:wrap_schema(schema),
+ -- 向插件的元数据 schema 中加入 max_pending_entries
+ metadata_schema =
batch_processor_manager:wrap_metadata_schema(metadata_schema),
}
...
@@ -57,6 +86,8 @@ local _M = {
function _M.log(conf, ctx)
local entry = {...} -- data to log
+ -- 返回 true 表示该条目已无需调用方再处理：要么已推入现有处理器，
+ -- 要么因积压超过 max_pending_entries 而被丢弃
if batch_processor_manager:add_entry(conf, entry) then
return
end
diff --git a/docs/zh/latest/plugins/clickhouse-logger.md
b/docs/zh/latest/plugins/clickhouse-logger.md
index 2dbc8628de..0806519e58 100644
--- a/docs/zh/latest/plugins/clickhouse-logger.md
+++ b/docs/zh/latest/plugins/clickhouse-logger.md
@@ -71,7 +71,7 @@ description: clickhouse-logger 插件将请求和响应日志批量推送到 Cli
| 名称 | 类型 | 必选项 | 默认值 | 有效值 | 描述
|
|--------------------|---------|--------|--------|--------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| log_format | object | False | | | 以 JSON
键值对形式声明的自定义日志格式。值可通过 `$` 前缀引用 [APISIX 变量](../apisix-variable.md) 或 [NGINX
变量](https://nginx.org/en/docs/http/ngx_http_core_module.html)。该配置全局生效，对所有绑定
`clickhouse-logger` 的路由和服务生效。 |
-| max_pending_entries | integer | False | | >= 1 |
批处理器中允许的最大未处理条目数。达到此限制后，新条目将被丢弃，直到积压减少。
|
+| max_pending_entries | integer | False | 8192 | >= 1 |
待处理条目数的上限。积压超过该值后新条目会被丢弃，避免日志服务变慢或不可达时 worker 内存无限增长。该上限对应的内存开销参见
[批处理器](../batch-processor.md#限制积压条目数)。 |
## 示例
diff --git a/docs/zh/latest/plugins/datadog.md
b/docs/zh/latest/plugins/datadog.md
index 37d21dfd80..516d6cd2c2 100644
--- a/docs/zh/latest/plugins/datadog.md
+++ b/docs/zh/latest/plugins/datadog.md
@@ -63,6 +63,7 @@ description: datadog 插件与 Datadog 集成，将指标批量发送到 DogStat
| port | integer | 否 | 8125 | DogStatsD 服务器端口。
|
| namespace | string | 否 | "apisix" | APISIX agent
发送的所有自定义指标的前缀。有助于在指标图中定位实体，例如 `apisix.request.counter`。
|
| constant_tags | array | 否 | [ "source:apisix" ] |
嵌入到生成指标中的静态标签，便于按信号对指标进行分组。详见[标签定义](https://docs.datadoghq.com/getting_started/tagging/#defining-tags)。
|
+| max_pending_entries | integer | 否 | 8192 |
待处理条目数的上限。积压超过该值后新条目会被丢弃，避免日志服务变慢或不可达时 worker 内存无限增长。该上限对应的内存开销参见
[批处理器](../batch-processor.md#限制积压条目数)。 |
## 指标
diff --git a/docs/zh/latest/plugins/elasticsearch-logger.md
b/docs/zh/latest/plugins/elasticsearch-logger.md
index e7912512bf..45ba4d2c12 100644
--- a/docs/zh/latest/plugins/elasticsearch-logger.md
+++ b/docs/zh/latest/plugins/elasticsearch-logger.md
@@ -66,7 +66,7 @@ description: elasticsearch-logger Plugin 将请求和响应日志批量推送到
| 名称 | 类型 | 必选项 | 默认值 | 描述 |
|------|------|--------|--------|------|
| log_format | object | 否 | | 自定义日志格式以 JSON
的键值对声明。值支持字符串和嵌套对象（最多五层，超出部分将被截断）。字符串中可通过 `$` 前缀引用 [APISIX
变量](../apisix-variable.md) 和 [NGINX
变量](http://nginx.org/en/docs/varindex.html)。 |
-| max_pending_entries | integer | 否 | | 在批处理器开始丢弃条目之前，可缓冲在批处理器中的最大待处理条目数。 |
+| max_pending_entries | integer | 否 | 8192 |
待处理条目数的上限。积压超过该值后新条目会被丢弃，避免日志服务变慢或不可达时 worker 内存无限增长。该上限对应的内存开销参见
[批处理器](../batch-processor.md#限制积压条目数)。 |
## 示例
diff --git a/docs/zh/latest/plugins/google-cloud-logging.md
b/docs/zh/latest/plugins/google-cloud-logging.md
index 41af9f8c5f..2b67eb72a7 100644
--- a/docs/zh/latest/plugins/google-cloud-logging.md
+++ b/docs/zh/latest/plugins/google-cloud-logging.md
@@ -66,7 +66,7 @@ description: google-cloud-logging 插件将请求和响应日志批量推送到
| 名称 | 类型 | 必选项 | 默认值 | 有效值 | 描述
|
|--------------------|---------|--------|--------|--------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| log_format | object | False | | | 以 JSON
键值对形式声明的自定义日志格式。值可通过 `$` 前缀引用 [APISIX 变量](../apisix-variable.md) 或 [NGINX
变量](https://nginx.org/en/docs/http/ngx_http_core_module.html)。该配置全局生效，对所有绑定
`google-cloud-logging` 的路由和服务生效。 |
-| max_pending_entries | integer | False | | >= 1 |
批处理器中允许的最大未处理条目数。达到此限制后，新条目将被丢弃，直到积压减少。
|
+| max_pending_entries | integer | False | 8192 | >= 1 |
待处理条目数的上限。积压超过该值后新条目会被丢弃，避免日志服务变慢或不可达时 worker 内存无限增长。该上限对应的内存开销参见
[批处理器](../batch-processor.md#限制积压条目数)。 |
## 示例
diff --git a/docs/zh/latest/plugins/http-logger.md
b/docs/zh/latest/plugins/http-logger.md
index a73caaa991..81aefaae4c 100644
--- a/docs/zh/latest/plugins/http-logger.md
+++ b/docs/zh/latest/plugins/http-logger.md
@@ -63,11 +63,10 @@ description: http-logger 插件将请求和响应日志以 JSON 对象批量推
也可以通过配置插件元数据来设置日志格式，可用配置如下：
-| 名称 | 类型 | 必选项 | 描述
|
-|---------------------|---------|--------|----------------------------------------------------------------------------------------------------------------------------------------------------|
-| log_format | object | False | 以 JSON 键值对形式声明的自定义日志格式，值可以引用 [NGINX
变量](https://nginx.org/en/docs/http/ngx_http_core_module.html)。
|
-| max_pending_entries | integer | False |
批处理器中允许的最大未处理条目数。达到此限制后，新条目将被丢弃，直到积压减少。在 APISIX 3.15.0 版本中可用。
|
-
+| 名称 | 类型 | 必选项 | 默认值 | 描述
|
+|---------------------|---------|--------|---------|----------------------------------------------------------------------------------------------------------------------------------------------------|
+| log_format | object | False | | 以 JSON
键值对形式声明的自定义日志格式，值可以引用 [NGINX
变量](https://nginx.org/en/docs/http/ngx_http_core_module.html)。
|
+| max_pending_entries | integer | False | 8192 |
待处理条目数的上限。积压超过该值后新条目会被丢弃，避免日志服务变慢或不可达时 worker 内存无限增长。该上限对应的内存开销参见
[批处理器](../batch-processor.md#限制积压条目数)。 |
:::info IMPORTANT
插件元数据的配置为全局范围生效，将作用于所有使用 `http-logger` 插件的路由和服务。
diff --git a/docs/zh/latest/plugins/kafka-logger.md
b/docs/zh/latest/plugins/kafka-logger.md
index b67fa3710d..61fd64b859 100644
--- a/docs/zh/latest/plugins/kafka-logger.md
+++ b/docs/zh/latest/plugins/kafka-logger.md
@@ -144,8 +144,7 @@ description: kafka-logger 插件将请求和响应日志作为 JSON 对象批量
| 名称 | 类型 | 是否必需 | 默认值 | 描述
|
| ------------------- | ------- | -------- | ------ |
---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
|
| log_format | object | 否 | | 以 JSON
键值对声明的日志格式。值支持字符串和嵌套对象（最多五层，超出部分将被截断）。字符串中可通过在前面加上 `$` 来引用 [APISIX
变量](../apisix-variable.md) 或 [NGINX
内置变量](http://nginx.org/en/docs/varindex.html)。 |
-| max_pending_entries | integer | 否 | |
批处理器开始丢弃待处理条目之前可缓冲的最大待处理条目数。
|
-
+| max_pending_entries | integer | 否 | 8192 |
待处理条目数的上限。积压超过该值后新条目会被丢弃，避免日志服务变慢或不可达时 worker 内存无限增长。该上限对应的内存开销参见
[批处理器](../batch-processor.md#限制积压条目数)。 |
:::info 重要
插件元数据配置为全局生效。这意味着它将对所有使用 `kafka-logger` 插件的路由和服务生效。
diff --git a/docs/zh/latest/plugins/loggly.md b/docs/zh/latest/plugins/loggly.md
index 09e599f774..9b7feaea4f 100644
--- a/docs/zh/latest/plugins/loggly.md
+++ b/docs/zh/latest/plugins/loggly.md
@@ -68,6 +68,7 @@ description: API 网关 Apache APISIX loggly 插件可用于将日志转发到 S
| timeout | integer | 否 | 5000 |
| 发送数据请求超时时间（以毫秒为单位）。 |
| protocol | string | 否 | "syslog" | [ "syslog", "http",
"https" ] | 将日志发送到 Loggly 的协议。 |
| log_format | object | 否 | nil |
| 日志格式以 JSON 的键值对声明。值支持字符串和嵌套对象（最多五层，超出部分将被截断）。字符串中可通过在前面加上 `$` 来引用
[APISIX 变量](../../../en/latest/apisix-variable.md) 或 [NGINX
内置变量](http://nginx.org/en/docs/varindex.html)。 |
+| max_pending_entries | integer | 否 | 8192 | >= 1 |
待处理条目数的上限。积压超过该值后新条目会被丢弃，避免日志服务变慢或不可达时 worker 内存无限增长。该上限对应的内存开销参见
[批处理器](../batch-processor.md#限制积压条目数)。 |
APISIX 支持
[Syslog](https://documentation.solarwinds.com/en/success_center/loggly/content/admin/streaming-syslog-without-using-files.htm)、[HTTP/S](https://documentation.solarwinds.com/en/success_center/loggly/content/admin/http-bulk-endpoint.htm)（批量端点）协议将日志事件发送到
Loggly。**默认情况下 `protocol` 的值为 `syslog`**。该协议允许你通过一些细粒度的控制（基于上游 HTTP
响应代码的日志严重性映射）发送符合 RFC5424 的系统日志事件。但是 HTTP/S 批量端点非常适合以更快的传输速度发送更大量的日志事件。
diff --git a/docs/zh/latest/plugins/loki-logger.md
b/docs/zh/latest/plugins/loki-logger.md
index a0ff70592f..e90657e62c 100644
--- a/docs/zh/latest/plugins/loki-logger.md
+++ b/docs/zh/latest/plugins/loki-logger.md
@@ -70,7 +70,7 @@ description: loki-logger 插件通过 Loki HTTP API /loki/api/v1/push
将请求
| 名称 | 类型 | 必选项 | 默认值 | 有效值 | 描述 |
|------|------|--------|--------|--------|------|
| log_format | object | 否 | | | 日志格式以 JSON
的键值对声明。值支持字符串和嵌套对象（最多五层，超出部分将被截断）。字符串中可通过在前面加上 `$` 来引用 [APISIX
变量](../apisix-variable.md) 和 [NGINX
变量](http://nginx.org/en/docs/varindex.html)。 |
-| max_pending_entries | integer | 否 | | | 在批处理器开始丢弃条目之前，可缓冲在批处理器中的最大待处理条目数。 |
+| max_pending_entries | integer | 否 | 8192 | >= 1 |
待处理条目数的上限。积压超过该值后新条目会被丢弃，避免日志服务变慢或不可达时 worker 内存无限增长。该上限对应的内存开销参见
[批处理器](../batch-processor.md#限制积压条目数)。 |
## 示例
diff --git a/docs/zh/latest/plugins/rocketmq-logger.md
b/docs/zh/latest/plugins/rocketmq-logger.md
index fe48e50fd9..0149a56405 100644
--- a/docs/zh/latest/plugins/rocketmq-logger.md
+++ b/docs/zh/latest/plugins/rocketmq-logger.md
@@ -68,11 +68,10 @@ description: rocketmq-logger 插件将请求和响应日志以 JSON 对象批量
也可以通过配置插件元数据来设置日志格式，可用配置如下：
-| 名称 | 类型 | 必选项 | 描述
|
-|---------------------|---------|--------|----------------------------------------------------------------------------------------------------------------------------------------------------|
-| log_format | object | False | 以 JSON 键值对形式声明的自定义日志格式，值可以引用 [NGINX
变量](https://nginx.org/en/docs/http/ngx_http_core_module.html)。
|
-| max_pending_entries | integer | False |
批处理器中允许的最大未处理条目数。达到此限制后，新条目将被丢弃，直到积压减少。在 APISIX 3.15.0 版本中可用。
|
-
+| 名称 | 类型 | 必选项 | 默认值 | 描述
|
+|---------------------|---------|--------|---------|----------------------------------------------------------------------------------------------------------------------------------------------------|
+| log_format | object | False | | 以 JSON
键值对形式声明的自定义日志格式，值可以引用 [NGINX
变量](https://nginx.org/en/docs/http/ngx_http_core_module.html)。
|
+| max_pending_entries | integer | False | 8192 |
待处理条目数的上限。积压超过该值后新条目会被丢弃，避免日志服务变慢或不可达时 worker 内存无限增长。该上限对应的内存开销参见
[批处理器](../batch-processor.md#限制积压条目数)。 |
:::info IMPORTANT
插件元数据的配置为全局范围生效，将作用于所有使用 `rocketmq-logger` 插件的路由和服务。
diff --git a/docs/zh/latest/plugins/skywalking-logger.md
b/docs/zh/latest/plugins/skywalking-logger.md
index 1d61cb4b87..24bf2609fe 100644
--- a/docs/zh/latest/plugins/skywalking-logger.md
+++ b/docs/zh/latest/plugins/skywalking-logger.md
@@ -63,7 +63,7 @@ description: skywalking-logger 将请求和响应日志作为 JSON 对象批量
| 名称 | 类型 | 必选项 | 默认值 | 有效值 | 描述
|
| ---------------------- | ------- | ------ | -------------------- |
------------- |
---------------------------------------------------------------- |
| log_format | object | 否 | | 日志格式以 JSON
的键值对声明。值支持字符串和嵌套对象（最多五层，超出部分将被截断）。字符串中可通过在前面加上 `$` 来引用 [APISIX
变量](../apisix-variable.md) 或 [NGINX
内置变量](http://nginx.org/en/docs/varindex.html)。 |
-| max_pending_entries | integer | 否 | | | 在批处理器中开始删除待处理条目之前可以购买的最大待处理条目数。|
+| max_pending_entries | integer | 否 | 8192 | >= 1 |
待处理条目数的上限。积压超过该值后新条目会被丢弃，避免日志服务变慢或不可达时 worker 内存无限增长。该上限对应的内存开销参见
[批处理器](../batch-processor.md#限制积压条目数)。 |
## 示例
diff --git a/docs/zh/latest/plugins/sls-logger.md
b/docs/zh/latest/plugins/sls-logger.md
index f434a8b97d..1ab15264f8 100644
--- a/docs/zh/latest/plugins/sls-logger.md
+++ b/docs/zh/latest/plugins/sls-logger.md
@@ -85,6 +85,7 @@ title: sls-logger
| 名称 | 类型 | 必选项 | 默认值 | 有效值 | 描述
|
| ---------------- | ------- | ------ | ------------- | ------- |
------------------------------------------------ |
| log_format | object | 可选 | | | 日志格式以 JSON
的键值对声明。值支持字符串和嵌套对象（最多五层，超出部分将被截断）。字符串中可通过在前面加上 `$` 来引用 [APISIX
变量](../../../en/latest/apisix-variable.md) 或 [Nginx
内置变量](http://nginx.org/en/docs/varindex.html)。特别的，**该设置是全局生效的**，意味着指定
log_format 后，将对所有绑定 sls-logger 的 Route 或 Service 生效。 |
+| max_pending_entries | integer | 可选 | 8192 | >= 1 |
待处理条目数的上限。积压超过该值后新条目会被丢弃，避免日志服务变慢或不可达时 worker 内存无限增长。该上限对应的内存开销参见
[批处理器](../batch-processor.md#限制积压条目数)。 |
### 设置日志格式示例
diff --git a/docs/zh/latest/plugins/splunk-hec-logging.md
b/docs/zh/latest/plugins/splunk-hec-logging.md
index 3d948803ee..a68cca99ef 100644
--- a/docs/zh/latest/plugins/splunk-hec-logging.md
+++ b/docs/zh/latest/plugins/splunk-hec-logging.md
@@ -62,7 +62,7 @@ description: splunk-hec-logging 插件将请求和响应上下文信息序列化
| 名称 | 类型 | 必选项 | 默认值 | 有效值 | 描述
|
|--------------------|---------|--------|--------|--------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| log_format | object | False | | | 以 JSON
键值对形式声明的自定义日志格式。值可通过 `$` 前缀引用 [APISIX 变量](../apisix-variable.md) 或 [NGINX
变量](https://nginx.org/en/docs/http/ngx_http_core_module.html)。该配置全局生效，对所有绑定
`splunk-hec-logging` 的路由和服务生效。 |
-| max_pending_entries | integer | False | | >= 1 |
批处理器中允许的最大未处理条目数。达到此限制后，新条目将被丢弃，直到积压减少。
|
+| max_pending_entries | integer | False | 8192 | >= 1 |
待处理条目数的上限。积压超过该值后新条目会被丢弃，避免日志服务变慢或不可达时 worker 内存无限增长。该上限对应的内存开销参见
[批处理器](../batch-processor.md#限制积压条目数)。 |
## 示例
diff --git a/docs/zh/latest/plugins/syslog.md b/docs/zh/latest/plugins/syslog.md
index d550517d98..eb3eb03f0d 100644
--- a/docs/zh/latest/plugins/syslog.md
+++ b/docs/zh/latest/plugins/syslog.md
@@ -71,9 +71,10 @@ description: syslog 插件将请求和响应日志以 JSON 对象批量推送到
也可以通过配置插件元数据来设置日志格式，可用配置如下：
-| 名称 | 类型 | 必选项 | 描述
|
-|------------|--------|--------|----------------------------------------------------------------------------------------------------------------------------------------------------|
-| log_format | object | False | 以 JSON 键值对形式声明的自定义日志格式，值可以引用 [NGINX
变量](https://nginx.org/en/docs/http/ngx_http_core_module.html)。
|
+| 名称 | 类型 | 必选项 | 默认值 | 描述
|
+|------------|--------|--------|---------|----------------------------------------------------------------------------------------------------------------------------------------------------|
+| log_format | object | False | | 以 JSON 键值对形式声明的自定义日志格式，值可以引用 [NGINX
变量](https://nginx.org/en/docs/http/ngx_http_core_module.html)。
|
+| max_pending_entries | integer | False | 8192 |
待处理条目数的上限。积压超过该值后新条目会被丢弃，避免日志服务变慢或不可达时 worker 内存无限增长。该上限对应的内存开销参见
[批处理器](../batch-processor.md#限制积压条目数)。 |
:::info IMPORTANT
diff --git a/docs/zh/latest/plugins/tcp-logger.md
b/docs/zh/latest/plugins/tcp-logger.md
index 097e55bc20..9669c7a194 100644
--- a/docs/zh/latest/plugins/tcp-logger.md
+++ b/docs/zh/latest/plugins/tcp-logger.md
@@ -95,8 +95,7 @@ description: 本文介绍了 API 网关 Apache APISIX 如何使用 tcp-logger
| 名称 | 类型 | 必选项 | 默认值 | 有效值 | 描述
|
| ---------------- | ------- | ------ | ------------- | ------- |
------------------------------------------------ |
| log_format | object | 否 | | | 日志格式以 JSON
的键值对声明。值支持字符串和嵌套对象（最多五层，超出部分将被截断）。字符串中可通过在前面加上 `$` 来引用 [APISIX
变量](../apisix-variable.md) 或 [NGINX
内置变量](http://nginx.org/en/docs/varindex.html)。 |
-| max_pending_entries | integer | 否 | | | 在批处理器中开始删除待处理条目之前可以购买的最大待处理条目数。|
-
+| max_pending_entries | integer | 否 | 8192 | >= 1 |
待处理条目数的上限。积压超过该值后新条目会被丢弃，避免日志服务变慢或不可达时 worker 内存无限增长。该上限对应的内存开销参见
[批处理器](../batch-processor.md#限制积压条目数)。 |
:::info 注意
该设置全局生效。如果指定了 `log_format`，则所有绑定 `tcp-logger` 的路由或服务都将使用该日志格式。
diff --git a/docs/zh/latest/plugins/tencent-cloud-cls.md
b/docs/zh/latest/plugins/tencent-cloud-cls.md
index b35b764c76..1c8486b4a7 100644
--- a/docs/zh/latest/plugins/tencent-cloud-cls.md
+++ b/docs/zh/latest/plugins/tencent-cloud-cls.md
@@ -98,8 +98,7 @@ description: API 网关 Apache APISIX tencent-cloud-cls 插件可用于将日志
| 名称 | 类型 | 必选项 | 默认值 | 有效值 | 描述
|
| ---------------- | ------- | ------ | ------------- | ------- |
------------------------------------------------ |
| log_format | object | 否 | | | 日志格式以 JSON
的键值对声明。值支持字符串和嵌套对象（最多五层，超出部分将被截断）。字符串中可通过在前面加上 `$` 来引用 [APISIX
变量](../../../en/latest/apisix-variable.md) 或 [NGINX
内置变量](http://nginx.org/en/docs/varindex.html)。 |
-| max_pending_entries | integer | 否 | | | 在批处理器中开始删除待处理条目之前可以购买的最大待处理条目数。|
-
+| max_pending_entries | integer | 否 | 8192 | >= 1 |
待处理条目数的上限。积压超过该值后新条目会被丢弃，避免日志服务变慢或不可达时 worker 内存无限增长。该上限对应的内存开销参见
[批处理器](../batch-processor.md#限制积压条目数)。 |
:::info 重要
该设置全局生效。如果指定了 `log_format`，则所有绑定 `tencent-cloud-cls` 的路由或服务都将使用该日志格式。
diff --git a/docs/zh/latest/plugins/udp-logger.md
b/docs/zh/latest/plugins/udp-logger.md
index 19017b909a..3f4629f340 100644
--- a/docs/zh/latest/plugins/udp-logger.md
+++ b/docs/zh/latest/plugins/udp-logger.md
@@ -94,8 +94,7 @@ description: 本文介绍了 API 网关 Apache APISIX 如何使用 udp-logger
| 名称 | 类型 | 必选项 | 默认值 | 有效值 | 描述
|
| ---------------- | ------- | ------ | ------------- | ------- |
------------------------------------------------ |
| log_format | object | 否 | | | 日志格式以 JSON
的键值对声明。值支持字符串和嵌套对象（最多五层，超出部分将被截断）。字符串中可通过在前面加上 `$` 来引用 [APISIX
变量](../apisix-variable.md) 或 [NGINX
内置变量](http://nginx.org/en/docs/varindex.html)。 |
-| max_pending_entries | integer | 否 | | | 在批处理器中开始删除待处理条目之前可以购买的最大待处理条目数。|
-
+| max_pending_entries | integer | 否 | 8192 | >= 1 |
待处理条目数的上限。积压超过该值后新条目会被丢弃，避免日志服务变慢或不可达时 worker 内存无限增长。该上限对应的内存开销参见
[批处理器](../batch-processor.md#限制积压条目数)。 |
:::info 注意
该设置全局生效。如果指定了 `log_format`，则所有绑定 `udp-logger` 的路由或服务都将使用该日志格式。
diff --git a/t/plugin/ai-proxy-kafka-log.t b/t/plugin/ai-proxy-kafka-log.t
index a6fddc0525..b21bd84202 100644
--- a/t/plugin/ai-proxy-kafka-log.t
+++ b/t/plugin/ai-proxy-kafka-log.t
@@ -44,16 +44,16 @@ add_block_preprocessor(sub {
core.log.info("send data to kafka: ", data)
end
local old_add = bp_manager.add_entry
- bp_manager.add_entry = function(self, conf, entry, max_pending_entries)
- local ok = old_add(self, conf, entry, max_pending_entries)
+ bp_manager.add_entry = function(self, conf, entry)
+ local ok = old_add(self, conf, entry)
if ok then
log_send_data(entry)
end
return ok
end
local old_new = bp_manager.add_entry_to_new_processor
- bp_manager.add_entry_to_new_processor = function(self, conf, entry, ctx,
func, max_pending_entries)
- local ok = old_new(self, conf, entry, ctx, func, max_pending_entries)
+ bp_manager.add_entry_to_new_processor = function(self, conf, entry, ctx,
func)
+ local ok = old_new(self, conf, entry, ctx, func)
if ok then
log_send_data(entry)
end
diff --git a/t/plugin/kafka-logger-large-body.t
b/t/plugin/kafka-logger-large-body.t
index 29c207e8ae..d566bb8562 100644
--- a/t/plugin/kafka-logger-large-body.t
+++ b/t/plugin/kafka-logger-large-body.t
@@ -65,16 +65,16 @@ add_block_preprocessor(sub {
core.log.info("send data to kafka: ", data)
end
local old_add = bp_manager.add_entry
- bp_manager.add_entry = function(self, conf, entry, max_pending_entries)
- local ok = old_add(self, conf, entry, max_pending_entries)
+ bp_manager.add_entry = function(self, conf, entry)
+ local ok = old_add(self, conf, entry)
if ok then
log_send_data(entry)
end
return ok
end
local old_new = bp_manager.add_entry_to_new_processor
- bp_manager.add_entry_to_new_processor = function(self, conf, entry, ctx,
func, max_pending_entries)
- local ok = old_new(self, conf, entry, ctx, func, max_pending_entries)
+ bp_manager.add_entry_to_new_processor = function(self, conf, entry, ctx,
func)
+ local ok = old_new(self, conf, entry, ctx, func)
if ok then
log_send_data(entry)
end
diff --git a/t/plugin/kafka-logger-log-format.t
b/t/plugin/kafka-logger-log-format.t
index a4ec304e57..db4c895529 100644
--- a/t/plugin/kafka-logger-log-format.t
+++ b/t/plugin/kafka-logger-log-format.t
@@ -34,16 +34,16 @@ add_block_preprocessor(sub {
core.log.info("send data to kafka: ", data)
end
local old_add = bp_manager.add_entry
- bp_manager.add_entry = function(self, conf, entry, max_pending_entries)
- local ok = old_add(self, conf, entry, max_pending_entries)
+ bp_manager.add_entry = function(self, conf, entry)
+ local ok = old_add(self, conf, entry)
if ok then
log_send_data(entry)
end
return ok
end
local old_new = bp_manager.add_entry_to_new_processor
- bp_manager.add_entry_to_new_processor = function(self, conf, entry, ctx,
func, max_pending_entries)
- local ok = old_new(self, conf, entry, ctx, func, max_pending_entries)
+ bp_manager.add_entry_to_new_processor = function(self, conf, entry, ctx,
func)
+ local ok = old_new(self, conf, entry, ctx, func)
if ok then
log_send_data(entry)
end
diff --git a/t/plugin/kafka-logger.t b/t/plugin/kafka-logger.t
index 6e355ba816..61f17422e8 100644
--- a/t/plugin/kafka-logger.t
+++ b/t/plugin/kafka-logger.t
@@ -41,16 +41,16 @@ add_block_preprocessor(sub {
core.log.info("send data to kafka: ", data)
end
local old_add = bp_manager.add_entry
- bp_manager.add_entry = function(self, conf, entry, max_pending_entries)
- local ok = old_add(self, conf, entry, max_pending_entries)
+ bp_manager.add_entry = function(self, conf, entry)
+ local ok = old_add(self, conf, entry)
if ok then
log_send_data(entry)
end
return ok
end
local old_new = bp_manager.add_entry_to_new_processor
- bp_manager.add_entry_to_new_processor = function(self, conf, entry, ctx,
func, max_pending_entries)
- local ok = old_new(self, conf, entry, ctx, func, max_pending_entries)
+ bp_manager.add_entry_to_new_processor = function(self, conf, entry, ctx,
func)
+ local ok = old_new(self, conf, entry, ctx, func)
if ok then
log_send_data(entry)
end
diff --git a/t/plugin/kafka-logger2.t b/t/plugin/kafka-logger2.t
index 8ba42999b3..572d68218b 100644
--- a/t/plugin/kafka-logger2.t
+++ b/t/plugin/kafka-logger2.t
@@ -41,16 +41,16 @@ add_block_preprocessor(sub {
core.log.info("send data to kafka: ", data)
end
local old_add = bp_manager.add_entry
- bp_manager.add_entry = function(self, conf, entry, max_pending_entries)
- local ok = old_add(self, conf, entry, max_pending_entries)
+ bp_manager.add_entry = function(self, conf, entry)
+ local ok = old_add(self, conf, entry)
if ok then
log_send_data(entry)
end
return ok
end
local old_new = bp_manager.add_entry_to_new_processor
- bp_manager.add_entry_to_new_processor = function(self, conf, entry, ctx,
func, max_pending_entries)
- local ok = old_new(self, conf, entry, ctx, func, max_pending_entries)
+ bp_manager.add_entry_to_new_processor = function(self, conf, entry, ctx,
func)
+ local ok = old_new(self, conf, entry, ctx, func)
if ok then
log_send_data(entry)
end
diff --git a/t/plugin/kafka-logger4.t b/t/plugin/kafka-logger4.t
index 95969705d5..176b5a5ec3 100644
--- a/t/plugin/kafka-logger4.t
+++ b/t/plugin/kafka-logger4.t
@@ -41,16 +41,16 @@ add_block_preprocessor(sub {
core.log.info("send data to kafka: ", data)
end
local old_add = bp_manager.add_entry
- bp_manager.add_entry = function(self, conf, entry, max_pending_entries)
- local ok = old_add(self, conf, entry, max_pending_entries)
+ bp_manager.add_entry = function(self, conf, entry)
+ local ok = old_add(self, conf, entry)
if ok then
log_send_data(entry)
end
return ok
end
local old_new = bp_manager.add_entry_to_new_processor
- bp_manager.add_entry_to_new_processor = function(self, conf, entry, ctx,
func, max_pending_entries)
- local ok = old_new(self, conf, entry, ctx, func, max_pending_entries)
+ bp_manager.add_entry_to_new_processor = function(self, conf, entry, ctx,
func)
+ local ok = old_new(self, conf, entry, ctx, func)
if ok then
log_send_data(entry)
end
diff --git a/t/plugin/rocketmq-logger-log-format.t
b/t/plugin/rocketmq-logger-log-format.t
index fc9d0753e0..d8fda2b5a6 100644
--- a/t/plugin/rocketmq-logger-log-format.t
+++ b/t/plugin/rocketmq-logger-log-format.t
@@ -34,16 +34,16 @@ add_block_preprocessor(sub {
core.log.info("send data to rocketmq: ", data)
end
local old_add = bp_manager.add_entry
- bp_manager.add_entry = function(self, conf, entry, max_pending_entries)
- local ok = old_add(self, conf, entry, max_pending_entries)
+ bp_manager.add_entry = function(self, conf, entry)
+ local ok = old_add(self, conf, entry)
if ok then
log_send_data(entry)
end
return ok
end
local old_new = bp_manager.add_entry_to_new_processor
- bp_manager.add_entry_to_new_processor = function(self, conf, entry, ctx,
func, max_pending_entries)
- local ok = old_new(self, conf, entry, ctx, func, max_pending_entries)
+ bp_manager.add_entry_to_new_processor = function(self, conf, entry, ctx,
func)
+ local ok = old_new(self, conf, entry, ctx, func)
if ok then
log_send_data(entry)
end
diff --git a/t/plugin/rocketmq-logger.t b/t/plugin/rocketmq-logger.t
index cfa7ca6696..63c542196b 100644
--- a/t/plugin/rocketmq-logger.t
+++ b/t/plugin/rocketmq-logger.t
@@ -41,16 +41,16 @@ add_block_preprocessor(sub {
core.log.info("send data to rocketmq: ", data)
end
local old_add = bp_manager.add_entry
- bp_manager.add_entry = function(self, conf, entry, max_pending_entries)
- local ok = old_add(self, conf, entry, max_pending_entries)
+ bp_manager.add_entry = function(self, conf, entry)
+ local ok = old_add(self, conf, entry)
if ok then
log_send_data(entry)
end
return ok
end
local old_new = bp_manager.add_entry_to_new_processor
- bp_manager.add_entry_to_new_processor = function(self, conf, entry, ctx,
func, max_pending_entries)
- local ok = old_new(self, conf, entry, ctx, func, max_pending_entries)
+ bp_manager.add_entry_to_new_processor = function(self, conf, entry, ctx,
func)
+ local ok = old_new(self, conf, entry, ctx, func)
if ok then
log_send_data(entry)
end
diff --git a/t/plugin/rocketmq-logger2.t b/t/plugin/rocketmq-logger2.t
index 7da7a92367..80c9bf7f8d 100644
--- a/t/plugin/rocketmq-logger2.t
+++ b/t/plugin/rocketmq-logger2.t
@@ -41,16 +41,16 @@ add_block_preprocessor(sub {
core.log.info("send data to rocketmq: ", data)
end
local old_add = bp_manager.add_entry
- bp_manager.add_entry = function(self, conf, entry, max_pending_entries)
- local ok = old_add(self, conf, entry, max_pending_entries)
+ bp_manager.add_entry = function(self, conf, entry)
+ local ok = old_add(self, conf, entry)
if ok then
log_send_data(entry)
end
return ok
end
local old_new = bp_manager.add_entry_to_new_processor
- bp_manager.add_entry_to_new_processor = function(self, conf, entry, ctx,
func, max_pending_entries)
- local ok = old_new(self, conf, entry, ctx, func, max_pending_entries)
+ bp_manager.add_entry_to_new_processor = function(self, conf, entry, ctx,
func)
+ local ok = old_new(self, conf, entry, ctx, func)
if ok then
log_send_data(entry)
end
diff --git a/t/plugin/sls-logger.t b/t/plugin/sls-logger.t
index f5d5d027a8..6bc901e425 100644
--- a/t/plugin/sls-logger.t
+++ b/t/plugin/sls-logger.t
@@ -31,9 +31,9 @@ add_block_preprocessor(sub {
local bp_manager = require("apisix.utils.batch-processor-manager")
local core = require("apisix.core")
local old_f = bp_manager.add_entry_to_new_processor
- bp_manager.add_entry_to_new_processor = function(self, conf, entry, ctx,
func, max_pending_entries)
+ bp_manager.add_entry_to_new_processor = function(self, conf, entry, ctx,
func)
ngx.log(ngx.INFO, "batch_entry: ", entry.data or
core.json.encode(entry))
- return old_f(self, conf, entry, ctx, func, max_pending_entries)
+ return old_f(self, conf, entry, ctx, func)
end
_EOC_
diff --git a/t/plugin/syslog.t b/t/plugin/syslog.t
index 7ce5c1ed88..e8268fe8b8 100644
--- a/t/plugin/syslog.t
+++ b/t/plugin/syslog.t
@@ -35,16 +35,16 @@ add_block_preprocessor(sub {
core.log.info("collect_data:", data)
end
local _orig_add = bp_manager.add_entry
- bp_manager.add_entry = function(self, conf, entry, max_pending_entries)
- local ok = _orig_add(self, conf, entry, max_pending_entries)
+ bp_manager.add_entry = function(self, conf, entry)
+ local ok = _orig_add(self, conf, entry)
if ok then
log_collect_data(entry)
end
return ok
end
local _orig_new = bp_manager.add_entry_to_new_processor
- bp_manager.add_entry_to_new_processor = function(self, conf, entry,
ctx, func, max_pending_entries)
- local ok = _orig_new(self, conf, entry, ctx, func,
max_pending_entries)
+ bp_manager.add_entry_to_new_processor = function(self, conf, entry,
ctx, func)
+ local ok = _orig_new(self, conf, entry, ctx, func)
if ok then
log_collect_data(entry)
end
diff --git a/t/plugin/tcp-logger.t b/t/plugin/tcp-logger.t
index 76aca2112c..ef676ed3bb 100644
--- a/t/plugin/tcp-logger.t
+++ b/t/plugin/tcp-logger.t
@@ -31,9 +31,9 @@ add_block_preprocessor(sub {
local bp_manager = require("apisix.utils.batch-processor-manager")
local core = require("apisix.core")
local old_f = bp_manager.add_entry_to_new_processor
- bp_manager.add_entry_to_new_processor = function(self, conf, entry, ctx,
func, max_pending_entries)
+ bp_manager.add_entry_to_new_processor = function(self, conf, entry, ctx,
func)
ngx.log(ngx.INFO, "batch_entry: ", core.json.encode(entry))
- return old_f(self, conf, entry, ctx, func, max_pending_entries)
+ return old_f(self, conf, entry, ctx, func)
end
_EOC_
diff --git a/t/plugin/udp-logger.t b/t/plugin/udp-logger.t
index da81fcb49c..42531953dc 100644
--- a/t/plugin/udp-logger.t
+++ b/t/plugin/udp-logger.t
@@ -31,9 +31,9 @@ add_block_preprocessor(sub {
local bp_manager = require("apisix.utils.batch-processor-manager")
local core = require("apisix.core")
local old_f = bp_manager.add_entry_to_new_processor
- bp_manager.add_entry_to_new_processor = function(self, conf, entry, ctx,
func, max_pending_entries)
+ bp_manager.add_entry_to_new_processor = function(self, conf, entry, ctx,
func)
ngx.log(ngx.INFO, "batch_entry: ", core.json.encode(entry))
- return old_f(self, conf, entry, ctx, func, max_pending_entries)
+ return old_f(self, conf, entry, ctx, func)
end
_EOC_
diff --git a/t/utils/batch-processor-manager.t
b/t/utils/batch-processor-manager.t
new file mode 100644
index 0000000000..5e078ca06d
--- /dev/null
+++ b/t/utils/batch-processor-manager.t
@@ -0,0 +1,374 @@
+#
+# 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';
+
+log_level('info');
+repeat_each(1);
+no_long_string();
+no_root_location();
+run_tests;
+
+__DATA__
+
+=== TEST 1: the backlog is capped even with no plugin metadata configured
+--- config
+ location /t {
+ content_by_lua_block {
+ local bp_manager_mod =
require("apisix.utils.batch-processor-manager")
+ local bp_manager = bp_manager_mod.new("test logger", "test-logger")
+ -- a batch that is never flushed, so every accepted entry stays
pending
+ local conf = {
+ name = "test logger",
+ batch_max_size = 10000000,
+ inactive_timeout = 300,
+ buffer_duration = 300,
+ max_retry_count = 0,
+ retry_delay = 1,
+ }
+ local ctx = {var = {route_id = "1", server_addr = "127.0.0.1"}}
+ local func = function() return true end
+
+ -- a discard is reported as handled, so the caller only falls
through
+ -- while there is still no processor to push to
+ local fell_through = 0
+ for i = 1, 20000 do
+ if not bp_manager:add_entry(conf, i) then
+ fell_through = fell_through + 1
+ bp_manager:add_entry_to_new_processor(conf, i, ctx, func)
+ end
+ end
+ ngx.say("pushed: ", bp_manager.total_pushed_entries)
+ ngx.say("fell through: ", fell_through)
+ }
+ }
+--- request
+GET /t
+--- response_body
+pushed: 8192
+fell through: 1
+--- error_log
+max pending entries limit exceeded. discarding entry
+
+
+
+=== TEST 2: max_pending_entries in the plugin's metadata overrides the default
+--- config
+ location /t {
+ content_by_lua_block {
+ local plugin = require("apisix.plugin")
+ plugin.plugin_metadatas = {
+ get = function(self, name)
+ if name == "test-logger" then
+ return {value = {max_pending_entries = 3}}
+ end
+ end
+ }
+
+ local bp_manager_mod =
require("apisix.utils.batch-processor-manager")
+ local bp_manager = bp_manager_mod.new("test logger", "test-logger")
+ local conf = {
+ name = "test logger",
+ batch_max_size = 10000000,
+ inactive_timeout = 300,
+ buffer_duration = 300,
+ max_retry_count = 0,
+ retry_delay = 1,
+ }
+ local ctx = {var = {route_id = "1", server_addr = "127.0.0.1"}}
+ local func = function() return true end
+
+ local fell_through = 0
+ for i = 1, 10 do
+ if not bp_manager:add_entry(conf, i) then
+ fell_through = fell_through + 1
+ bp_manager:add_entry_to_new_processor(conf, i, ctx, func)
+ end
+ end
+ ngx.say("pushed: ", bp_manager.total_pushed_entries)
+ ngx.say("fell through: ", fell_through)
+ }
+ }
+--- request
+GET /t
+--- response_body
+pushed: 3
+fell through: 1
+--- error_log
+max_pending_entries: 3
+
+
+
+=== TEST 3: discarding is reported at most once per second, with a running
count
+--- config
+ location /t {
+ content_by_lua_block {
+ local plugin = require("apisix.plugin")
+ plugin.plugin_metadatas = {
+ get = function(self, name)
+ return {value = {max_pending_entries = 1}}
+ end
+ }
+
+ local bp_manager_mod =
require("apisix.utils.batch-processor-manager")
+ local bp_manager = bp_manager_mod.new("test logger", "test-logger")
+ local conf = {
+ name = "test logger",
+ batch_max_size = 10000000,
+ inactive_timeout = 300,
+ buffer_duration = 300,
+ max_retry_count = 0,
+ retry_delay = 1,
+ }
+ local ctx = {var = {route_id = "1", server_addr = "127.0.0.1"}}
+ local func = function() return true end
+
+ local push = function(n)
+ for i = 1, n do
+ local ok = bp_manager:add_entry(conf, i)
+ if not ok then
+ bp_manager:add_entry_to_new_processor(conf, i, ctx,
func)
+ end
+ end
+ end
+
+ -- one entry fits, the other 99 are discarded but reported only
once
+ push(100)
+ -- past the interval the next discard reports, and its count covers
+ -- every entry discarded since the previous report
+ ngx.sleep(1.1)
+ push(5)
+ ngx.say("done")
+ }
+ }
+--- request
+GET /t
+--- response_body
+done
+--- grep_error_log eval
+qr/discarded_entries: \d+/
+--- grep_error_log_out
+discarded_entries: 1
+discarded_entries: 99
+
+
+
+=== TEST 4: every batch-processor based logger exposes max_pending_entries
+--- config
+ location /t {
+ content_by_lua_block {
+ local loggers = {
+ "clickhouse-logger", "datadog", "elasticsearch-logger",
+ "google-cloud-logging", "http-logger", "kafka-logger", "lago",
+ "loggly", "loki-logger", "rocketmq-logger",
"skywalking-logger",
+ "sls-logger", "splunk-hec-logging", "syslog", "tcp-logger",
+ "tencent-cloud-cls", "udp-logger",
+ }
+ table.insert(loggers, "stream syslog")
+ for _, name in ipairs(loggers) do
+ local mod = require(name == "stream syslog"
+ and "apisix.stream.plugins.syslog"
+ or "apisix.plugins." .. name)
+ local schema = mod.metadata_schema
+ local prop = schema and schema.properties
+ and schema.properties.max_pending_entries
+ if not prop or not prop.default then
+ ngx.say(name, ": missing max_pending_entries default")
+ end
+ end
+ ngx.say("done")
+ }
+ }
+--- request
+GET /t
+--- response_body
+done
+
+
+
+=== TEST 5: every logger's batch processor reads its own plugin's metadata
+--- config
+ location /t {
+ content_by_lua_block {
+ -- a manager built without its plugin name looks metadata up under
the
+ -- batch processor's display name, silently ignoring any override
+ local loggers = {
+ "clickhouse-logger", "datadog", "elasticsearch-logger",
+ "google-cloud-logging", "http-logger", "kafka-logger", "lago",
+ "loggly", "loki-logger", "rocketmq-logger",
"skywalking-logger",
+ "sls-logger", "splunk-hec-logging", "syslog", "tcp-logger",
+ "tencent-cloud-cls", "udp-logger",
+ }
+ -- the stream plugin and the shared syslog module both log as
"syslog"
+ local modules = {["apisix.stream.plugins.syslog"] = "syslog"}
+ local order = {}
+ for _, name in ipairs(loggers) do
+ modules["apisix.plugins." .. name] = name
+ table.insert(order, "apisix.plugins." .. name)
+ end
+ table.insert(order, "apisix.stream.plugins.syslog")
+
+ local bp_manager_mod =
require("apisix.utils.batch-processor-manager")
+ local orig_new = bp_manager_mod.new
+ local created
+ local checked = 0
+
+ for _, path in ipairs(order) do
+ -- syslog keeps its batch processor in a module shared with the
+ -- stream plugin; reload it alongside so whichever plugin is
under
+ -- test is the one that builds it
+ package.loaded["apisix.plugins.syslog.init"] = nil
+ package.loaded[path] = nil
+
+ created = {}
+ bp_manager_mod.new = function(name, plugin_name)
+ local manager = orig_new(name, plugin_name)
+ table.insert(created, manager)
+ return manager
+ end
+ require(path)
+ bp_manager_mod.new = orig_new
+
+ if #created == 0 then
+ ngx.say(path, " built no batch processor")
+ end
+ for _, manager in ipairs(created) do
+ checked = checked + 1
+ if manager.plugin_name ~= modules[path] then
+ ngx.say(path, ": batch processor '", manager.name,
+ "' reads the metadata of '",
manager.plugin_name,
+ "', expected '", modules[path], "'")
+ end
+ end
+ end
+ ngx.say("checked ", checked, " batch processors")
+ }
+ }
+--- request
+GET /t
+--- response_body
+checked 20 batch processors
+
+
+
+=== TEST 6: max_pending_entries is validated through each plugin's metadata
schema
+--- config
+ location /t {
+ content_by_lua_block {
+ local core = require("apisix.core")
+ local loggers = {
+ "clickhouse-logger", "datadog", "elasticsearch-logger",
+ "google-cloud-logging", "http-logger", "kafka-logger", "lago",
+ "loggly", "loki-logger", "rocketmq-logger",
"skywalking-logger",
+ "sls-logger", "splunk-hec-logging", "syslog", "tcp-logger",
+ "tencent-cloud-cls", "udp-logger",
+ }
+ table.insert(loggers, "stream syslog")
+ for _, name in ipairs(loggers) do
+ local mod = require(name == "stream syslog"
+ and "apisix.stream.plugins.syslog"
+ or "apisix.plugins." .. name)
+ local ok, err = mod.check_schema({max_pending_entries = 100},
+ core.schema.TYPE_METADATA)
+ if not ok then
+ ngx.say(name, " rejected a valid override: ", err)
+ end
+ if mod.check_schema({max_pending_entries = 0},
+ core.schema.TYPE_METADATA) then
+ ngx.say(name, " accepted max_pending_entries = 0")
+ end
+ end
+ ngx.say("done")
+ }
+ }
+--- request
+GET /t
+--- response_body
+done
+
+
+
+=== TEST 7: the limit reaches a plugin that never carried it before
+--- extra_yaml_config
+plugins:
+ - sls-logger
+--- config
+location /t {
+ content_by_lua_block {
+ local http = require "resty.http"
+ local httpc = http.new()
+ local t = require("lib.test_admin").test
+
+ local code, body = t('/apisix/admin/plugin_metadata/sls-logger',
+ ngx.HTTP_PUT, {max_pending_entries = 1})
+ if code >= 300 then
+ ngx.status = code
+ ngx.say(body)
+ return
+ end
+
+ -- nothing listens on the log port, and retries keep the first entry
+ -- pending, so everything after it is over the limit
+ code, body = t('/apisix/admin/routes/1', ngx.HTTP_PUT, {
+ plugins = {
+ ["sls-logger"] = {
+ host = "127.0.0.1",
+ port = 1234,
+ project = "test-project",
+ logstore = "test-logstore",
+ access_key_id = "test-key-id",
+ access_key_secret = "test-key-secret",
+ batch_max_size = 1,
+ max_retry_count = 10,
+ retry_delay = 1,
+ timeout = 1,
+ },
+ },
+ upstream = {
+ nodes = {["127.0.0.1:1980"] = 1},
+ type = "roundrobin",
+ },
+ uri = "/hello",
+ })
+ if code >= 300 then
+ ngx.status = code
+ ngx.say(body)
+ return
+ end
+
+ local uri = "http://127.0.0.1:"; .. ngx.var.server_port .. "/hello"
+ for i = 1, 3 do
+ local res, err = httpc:request_uri(uri, {method = "GET"})
+ if not res then
+ ngx.say("request ", i, " failed: ", err)
+ return
+ end
+ if res.status ~= 200 then
+ ngx.say("request ", i, " returned ", res.status)
+ return
+ end
+ end
+ ngx.sleep(1)
+ ngx.say("passed")
+ }
+}
+--- request
+GET /t
+--- response_body
+passed
+--- error_log
+max pending entries limit exceeded. discarding entry
+--- timeout: 5
