[Fix] Add safety valve to discard data on 10x limit overflow

Prevents unbounded memory growth by discarding buffered data when
limits are exceeded by 10x, logging an error to alert operators.

diff --git a/src/plugins/lua/clickhouse.lua b/src/plugins/lua/clickhouse.lua
index 04803e1be0c4fdb44def759f8843dc751471089a..260ce13fd50260aaa7e841dc3178fbcff942303f 100644 (file)
--- a/src/plugins/lua/clickhouse.lua
+++ b/src/plugins/lua/clickhouse.lua
@@ -1032,13 +1032,31 @@ local function clickhouse_maybe_send_data_periodic(cfg, ev_base, now)
     return 0
   end
 
+  local dominated = false
+
   if settings.limits.max_rows > 0 then
-    if nrows > settings.limits.max_rows then
+    if nrows > settings.limits.max_rows * 10 then
+      dominated = true
+      rspamd_logger.errx(cfg, 'row count limit exceeded 10x: %d rows (limit %d), discarding data',
+          nrows, settings.limits.max_rows)
+    elseif nrows > settings.limits.max_rows then
       need_collect = true
       reason = string.format('limit of rows has been reached: %d', nrows)
     end
   end
 
+  if settings.limits.max_memory > 0 then
+    if used_memory >= settings.limits.max_memory * 10 then
+      dominated = true
+      rspamd_logger.errx(cfg, 'memory limit exceeded 10x: %d bytes (limit %d), discarding data',
+          used_memory, settings.limits.max_memory)
+    elseif used_memory >= settings.limits.max_memory then
+      need_collect = true
+      reason = string.format('limit of memory has been reached: %d bytes used',
+          used_memory)
+    end
+  end
+
   if last_collection > 0 and settings.limits.max_interval > 0 then
     if now - last_collection > settings.limits.max_interval then
       need_collect = true
@@ -1048,20 +1066,18 @@ local function clickhouse_maybe_send_data_periodic(cfg, ev_base, now)
     end
   end
 
-  if settings.limits.max_memory > 0 then
-    if used_memory >= settings.limits.max_memory then
-      need_collect = true
-      reason = string.format('limit of memory has been reached: %d bytes used',
-          used_memory)
-    end
-  end
-
   if last_collection == 0 then
     last_collection = now
   end
 
-  if need_collect then
-    -- Do it atomic
+  if dominated then
+    nrows = 0
+    last_collection = now
+    used_memory = 0
+    data_rows = {}
+    custom_rows = {}
+    collectgarbage()
+  elseif need_collect then
     local saved_rows = data_rows
     local saved_custom = custom_rows
     nrows = 0

diff --git a/src/plugins/lua/elastic.lua b/src/plugins/lua/elastic.lua
index 6c59be685ab1f42a8be69860a46ae0bac9103800..31d8d966160d2864beb967091bc74c7eea6c77d2 100644 (file)
--- a/src/plugins/lua/elastic.lua
+++ b/src/plugins/lua/elastic.lua
@@ -683,7 +683,14 @@ local function periodic_send_data(cfg, ev_base)
   local flush_needed = false
 
   local nlogs_total = buffer['logs']:length()
-  if nlogs_total >= settings['limits']['max_rows'] then
+  if nlogs_total >= settings['limits']['max_rows'] * 10 then
+    rspamd_logger.errx(rspamd_config,
+        'row count limit exceeded 10x: %s rows (limit %s), discarding data',
+        nlogs_total, settings['limits']['max_rows'])
+    buffer['logs'] = lua_util.newdeque()
+    collectgarbage()
+    return
+  elseif nlogs_total >= settings['limits']['max_rows'] then
     rspamd_logger.infox(rspamd_config, 'flushing buffer by reaching max rows: %s/%s', nlogs_total,
         settings['limits']['max_rows'])
     flush_needed = true
@@ -1518,7 +1525,6 @@ end
 
 local opts = rspamd_config:get_all_opt('elastic')
 
--- Merge nested tables to preserve defaults when user provides partial config
 local function merge_settings(src, dst)
   for k, v in pairs(src) do
     if type(v) == 'table' and type(dst[k]) == 'table' then