fix(batch-processor): bound the pending entries backlog by default

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
 
 

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),
 }
 
 

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,21 +334,17 @@ 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
 
     local process = function(entries)
         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

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,16 +263,15 @@ 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
 
     local process = function(entries)
         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
 
 

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
 
 

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
 
 

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)

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