[Minor] Replace rspamd_logger.debugm with proper lua_util.debugm

1 files changed, 91 insertions, 81 deletions

diff --git a/src/plugins/lua/elastic.lua b/src/plugins/lua/elastic.lua
index f3eb3cc4f..8bed9fcf4 100644
--- a/src/plugins/lua/elastic.lua
+++ b/src/plugins/lua/elastic.lua
@@ -72,7 +72,7 @@ local settings = {
   enabled = true,
   version = {
     autodetect_enabled = true,
-    autodetect_max_fail  = 30,
+    autodetect_max_fail = 30,
     -- override works only if autodetect is disabled
     override = {
       name = 'opensearch',
@@ -164,7 +164,7 @@ local Queue = {}
 Queue.__index = Queue
 
 function Queue:new()
-  local obj = {first = 1, last = 0, data = {}}
+  local obj = { first = 1, last = 0, data = {} }
   setmetatable(obj, self)
   return obj
 end
@@ -234,7 +234,7 @@ local buffer = {
 }
 
 local function contains(tbl, val)
-  for i=1,#tbl do
+  for i = 1, #tbl do
     if tbl[i]:lower() == val:lower() then
       return true
     end
@@ -244,7 +244,7 @@ end
 
 local function safe_get(table, ...)
   local value = table
-  for _, key in ipairs({...}) do
+  for _, key in ipairs({ ... }) do
     if value[key] == nil then
       return nil
     end
@@ -284,10 +284,10 @@ local function compare_versions(v1, v2)
   return 0 -- versions are equal
 end
 
-local function handle_error(action,component,limit)
+local function handle_error(action, component, limit)
   if states[component]['errors'] >= limit then
     rspamd_logger.errx(rspamd_config, 'cannot %s elastic %s, failed attempts: %s/%s, stop trying',
-      action, component:gsub('_', ' '), states[component]['errors'], limit)
+        action, component:gsub('_', ' '), states[component]['errors'], limit)
     states[component]['configured'] = true
   else
     states[component]['errors'] = states[component]['errors'] + 1
@@ -318,25 +318,25 @@ end
 local function is_empty(str)
   -- define a pattern that includes invisible unicode characters
   local str_cleared = str:gsub('[' ..
-    '\xC2\xA0'     .. -- U+00A0 non-breaking space
-    '\xE2\x80\x8B' .. -- U+200B zero width space
-    '\xEF\xBB\xBF' .. -- U+FEFF byte order mark (zero width no-break space)
-    '\xE2\x80\x8C' .. -- U+200C zero width non-joiner
-    '\xE2\x80\x8D' .. -- U+200D zero width joiner
-    '\xE2\x80\x8E' .. -- U+200E left-to-right mark
-    '\xE2\x80\x8F' .. -- U+200F right-to-left mark
-    '\xE2\x81\xA0' .. -- U+2060 word joiner
-    '\xE2\x80\xAA' .. -- U+202A left-to-right embedding
-    '\xE2\x80\xAB' .. -- U+202B right-to-left embedding
-    '\xE2\x80\xAC' .. -- U+202C pop directional formatting
-    '\xE2\x80\xAD' .. -- U+202D left-to-right override
-    '\xE2\x80\xAE' .. -- U+202E right-to-left override
-    '\xE2\x81\x9F' .. -- U+2061 function application
-    '\xE2\x81\xA1' .. -- U+2061 invisible separator
-    '\xE2\x81\xA2' .. -- U+2062 invisible times
-    '\xE2\x81\xA3' .. -- U+2063 invisible separator
-    '\xE2\x81\xA4' .. -- U+2064 invisible plus
-    ']', '') -- gsub replaces all matched characters with an empty string
+      '\xC2\xA0' .. -- U+00A0 non-breaking space
+      '\xE2\x80\x8B' .. -- U+200B zero width space
+      '\xEF\xBB\xBF' .. -- U+FEFF byte order mark (zero width no-break space)
+      '\xE2\x80\x8C' .. -- U+200C zero width non-joiner
+      '\xE2\x80\x8D' .. -- U+200D zero width joiner
+      '\xE2\x80\x8E' .. -- U+200E left-to-right mark
+      '\xE2\x80\x8F' .. -- U+200F right-to-left mark
+      '\xE2\x81\xA0' .. -- U+2060 word joiner
+      '\xE2\x80\xAA' .. -- U+202A left-to-right embedding
+      '\xE2\x80\xAB' .. -- U+202B right-to-left embedding
+      '\xE2\x80\xAC' .. -- U+202C pop directional formatting
+      '\xE2\x80\xAD' .. -- U+202D left-to-right override
+      '\xE2\x80\xAE' .. -- U+202E right-to-left override
+      '\xE2\x81\x9F' .. -- U+2061 function application
+      '\xE2\x81\xA1' .. -- U+2061 invisible separator
+      '\xE2\x81\xA2' .. -- U+2062 invisible times
+      '\xE2\x81\xA3' .. -- U+2063 invisible separator
+      '\xE2\x81\xA4' .. -- U+2064 invisible plus
+      ']', '') -- gsub replaces all matched characters with an empty string
   if str_cleared:match('[%S]') then
     return false
   else
@@ -350,7 +350,7 @@ local function fill_empty_strings(tbl, empty_value)
     if value and type(value) == 'table' then
       local nested_filtered = fill_empty_strings(value, empty_value)
       if next(nested_filtered) ~= nil then
-      filled_tbl[key] = nested_filtered
+        filled_tbl[key] = nested_filtered
       end
     elseif type(value) == 'boolean' then
       filled_tbl[key] = value
@@ -367,8 +367,8 @@ local function create_bulk_json(es_index, logs_to_send)
   local tbl = {}
   for _, row in pairs(logs_to_send) do
     local pipeline = ''
-    if settings['geoip']['enabled']then
-      pipeline = ',"pipeline":"'.. settings['geoip']['pipeline_name'] .. '"'
+    if settings['geoip']['enabled'] then
+      pipeline = ',"pipeline":"' .. settings['geoip']['pipeline_name'] .. '"'
     end
     table.insert(tbl, '{"index":{"_index":"' .. es_index .. '"' .. pipeline .. '}}')
     table.insert(tbl, ucl.to_format(row, 'json-compact'))
@@ -401,21 +401,21 @@ local function elastic_send_data(flush_all, task, cfg, ev_base)
     push_url = connect_prefix .. ip_addr .. '/' .. es_index .. '/_bulk'
 
     bulk_json = create_bulk_json(es_index, logs_to_send)
-    rspamd_logger.debugm(N, log_object, 'successfully composed payload with %s log lines', nlogs_to_send)
+    lua_util.debugm(N, log_object, 'successfully composed payload with %s log lines', nlogs_to_send)
   end
 
   local function http_callback(err, code, body, _)
     local push_done = false
     if err then
       rspamd_logger.errx(log_object, 'cannot send logs to elastic (%s): %s; failed attempts: %s/%s',
-        push_url, err, buffer['errors'], settings['limits']['max_fail'])
+          push_url, err, buffer['errors'], settings['limits']['max_fail'])
     elseif code == 200 then
       local parser = ucl.parser()
       local res, ucl_err = parser:parse_string(body)
       if not ucl_err and res then
         local obj = parser:get_object()
         push_done = true
-        rspamd_logger.debugm(N, log_object, 'successfully sent payload with %s logs', nlogs_to_send)
+        lua_util.debugm(N, log_object, 'successfully sent payload with %s logs', nlogs_to_send)
         if obj['errors'] then
           for _, value in pairs(obj['items']) do
             if value['index'] and value['index']['status'] >= 400 then
@@ -424,20 +424,20 @@ local function elastic_send_data(flush_all, task, cfg, ev_base)
               local error_type = safe_get(value, 'index', 'error', 'type') or ''
               local error_reason = safe_get(value, 'index', 'error', 'reason') or ''
               rspamd_logger.warnx(log_object,
-                'error while pushing logs to elastic, status: %s, index: %s, type: %s, reason: %s',
-                status, index, error_type, error_reason)
+                  'error while pushing logs to elastic, status: %s, index: %s, type: %s, reason: %s',
+                  status, index, error_type, error_reason)
             end
           end
         end
       else
         rspamd_logger.errx(log_object,
-          'cannot parse response from elastic (%s): %s; failed attempts: %s/%s',
-          push_url, ucl_err, buffer['errors'], settings['limits']['max_fail'])
+            'cannot parse response from elastic (%s): %s; failed attempts: %s/%s',
+            push_url, ucl_err, buffer['errors'], settings['limits']['max_fail'])
       end
     else
       rspamd_logger.errx(log_object,
-        'cannot send logs to elastic (%s) due to bad http status code: %s, response: %s; failed attempts: %s/%s',
-        push_url, code, body, buffer['errors'], settings['limits']['max_fail'])
+          'cannot send logs to elastic (%s) due to bad http status code: %s, response: %s; failed attempts: %s/%s',
+          push_url, code, body, buffer['errors'], settings['limits']['max_fail'])
     end
     -- proccess results
     if push_done then
@@ -447,8 +447,9 @@ local function elastic_send_data(flush_all, task, cfg, ev_base)
     else
       upstream:fail()
       if buffer['errors'] >= settings['limits']['max_fail'] then
-        rspamd_logger.errx(log_object, 'failed to send %s log lines, failed attempts: %s/%s, removing failed logs from bugger',
-          nlogs_to_send, buffer['errors'], settings['limits']['max_fail'])
+        rspamd_logger.errx(log_object,
+            'failed to send %s log lines, failed attempts: %s/%s, removing failed logs from bugger',
+            nlogs_to_send, buffer['errors'], settings['limits']['max_fail'])
         buffer['logs']:pop_first(nlogs_to_send)
         buffer['errors'] = 0
       else
@@ -466,7 +467,7 @@ local function elastic_send_data(flush_all, task, cfg, ev_base)
       },
       body = bulk_json,
       method = 'post',
-      callback=http_callback,
+      callback = http_callback,
       gzip = settings.use_gzip,
       keepalive = settings.use_keepalive,
       no_ssl_verify = settings.no_ssl_verify,
@@ -564,8 +565,8 @@ local function get_general_metadata(task)
   if task:has_from('smtp') then
     local from = task:get_from({ 'smtp', 'orig' })[1]
     if from and
-      from['user'] and #from['user'] > 0 and
-      from['domain'] and #from['domain'] > 0
+        from['user'] and #from['user'] > 0 and
+        from['domain'] and #from['domain'] > 0
     then
       r.from_user = from['user']
       r.from_domain = from['domain']:lower()
@@ -577,8 +578,8 @@ local function get_general_metadata(task)
   if task:has_from('mime') then
     local mime_from = task:get_from({ 'mime', 'orig' })[1]
     if mime_from and
-      mime_from['user'] and #mime_from['user'] > 0 and
-      mime_from['domain'] and #mime_from['domain'] > 0
+        mime_from['user'] and #mime_from['user'] > 0 and
+        mime_from['domain'] and #mime_from['domain'] > 0
     then
       r.mime_from_user = mime_from['user']
       r.mime_from_domain = mime_from['domain']:lower()
@@ -612,14 +613,14 @@ local function get_general_metadata(task)
       local l = {}
       for _, h in ipairs(hdr) do
         if settings['index_template']['headers_count_ignore_above'] ~= 0 and
-          #l >= settings['index_template']['headers_count_ignore_above']
+            #l >= settings['index_template']['headers_count_ignore_above']
         then
           table.insert(l, 'ignored above...')
           break
         end
         local header
         if settings['index_template']['headers_text_ignore_above'] ~= 0 and
-          h.decoded and #h.decoded >= headers_text_ignore_above
+            h.decoded and #h.decoded >= headers_text_ignore_above
         then
           header = h.decoded:sub(1, headers_text_ignore_above) .. '...'
         elseif h.decoded and #h.decoded > 0 then
@@ -663,10 +664,10 @@ local function get_general_metadata(task)
   local lang_t = {}
   if parts then
     for _, part in ipairs(parts) do
-        local l = part:get_language()
-        if l and not contains(lang_t, l) then
-          table.insert(lang_t, l)
-        end
+      local l = part:get_language()
+      if l and not contains(lang_t, l) then
+        table.insert(lang_t, l)
+      end
     end
     if #lang_t > 0 then
       r.language = lang_t
@@ -701,15 +702,15 @@ local function elastic_collect(task)
     if buffer['logs']:length() >= settings['limits']['max_rows'] then
       buffer['logs']:pop_first(settings['limits']['max_rows'])
       rspamd_logger.errx(task,
-        'elastic distro not supported, deleting %s logs from buffer due to reaching max rows limit',
-        settings['limits']['max_rows'])
+          'elastic distro not supported, deleting %s logs from buffer due to reaching max rows limit',
+          settings['limits']['max_rows'])
     end
   end
 
   local now = tostring(rspamd_util.get_time() * 1000)
   local row = { ['rspamd_meta'] = get_general_metadata(task), ['@timestamp'] = now }
   buffer['logs']:push(row)
-  rspamd_logger.debugm(N, task, 'saved log to buffer')
+  lua_util.debugm(N, task, 'saved log to buffer')
 end
 
 local function periodic_send_data(cfg, ev_base)
@@ -718,7 +719,8 @@ local function periodic_send_data(cfg, ev_base)
 
   local nlogs_total = buffer['logs']:length()
   if 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'])
+    rspamd_logger.infox(rspamd_config, 'flushing buffer by reaching max rows: %s/%s', nlogs_total,
+        settings['limits']['max_rows'])
     flush_needed = true
   else
     local first_row = buffer['logs']:get(1)
@@ -726,8 +728,9 @@ local function periodic_send_data(cfg, ev_base)
       local time_diff = now - first_row['@timestamp']
       local time_diff_sec = lua_util.round((time_diff / 1000), 1)
       if time_diff_sec > settings.limits.max_interval then
-        rspamd_logger.infox(rspamd_config, 'flushing buffer for %s by reaching max interval, oldest log in buffer written %s sec ago',
-          time_diff_sec, first_row['@timestamp'])
+        rspamd_logger.infox(rspamd_config,
+            'flushing buffer for %s by reaching max interval, oldest log in buffer written %s sec ago',
+            time_diff_sec, first_row['@timestamp'])
         flush_needed = true
       end
     end
@@ -770,8 +773,8 @@ local function configure_geoip_pipeline(cfg, ev_base)
       upstream:ok()
     else
       rspamd_logger.errx(rspamd_config,
-        'cannot configure elastic geoip pipeline (%s), status code: %s, response: %s',
-        geoip_url, code, body)
+          'cannot configure elastic geoip pipeline (%s), status code: %s, response: %s',
+          geoip_url, code, body)
       upstream:fail()
       handle_error('configure', 'geoip_pipeline', settings['limits']['max_fail'])
     end
@@ -807,7 +810,8 @@ local function put_index_policy(cfg, ev_base, upstream, host, policy_url, index_
       states['index_policy']['configured'] = true
       upstream:ok()
     else
-      rspamd_logger.errx(rspamd_config, 'cannot configure elastic index policy (%s), status code: %s, response: %s', policy_url, code, body)
+      rspamd_logger.errx(rspamd_config, 'cannot configure elastic index policy (%s), status code: %s, response: %s',
+          policy_url, code, body)
       upstream:fail()
       handle_error('configure', 'index_policy', settings['limits']['max_fail'])
     end
@@ -862,7 +866,8 @@ local function get_index_policy(cfg, ev_base, upstream, host, policy_url, index_
             local current_states = safe_get(remote_policy, 'policy', 'states')
             if not lua_util.table_cmp(our_policy['policy']['default_state'], current_default_state) then
               update_needed = true
-            elseif not lua_util.table_cmp(our_policy['policy']['ism_template'][1]['index_patterns'], current_ism_index_patterns) then
+            elseif not lua_util.table_cmp(our_policy['policy']['ism_template'][1]['index_patterns'],
+                current_ism_index_patterns) then
               update_needed = true
             elseif not lua_util.table_cmp(our_policy['policy']['states'], current_states) then
               update_needed = true
@@ -885,8 +890,8 @@ local function get_index_policy(cfg, ev_base, upstream, host, policy_url, index_
                 put_index_policy(cfg, ev_base, upstream, host, policy_url, index_policy_json)
               else
                 rspamd_logger.errx(rspamd_config,
-                  'current elastic index policy (%s) not returned correct seq_no/primary_term, policy will not be updated, response: %s',
-                  policy_url, body)
+                    'current elastic index policy (%s) not returned correct seq_no/primary_term, policy will not be updated, response: %s',
+                    policy_url, body)
                 upstream:fail()
                 handle_error('validate current', 'index_policy', settings['limits']['max_fail'])
               end
@@ -904,8 +909,8 @@ local function get_index_policy(cfg, ev_base, upstream, host, policy_url, index_
       end
     else
       rspamd_logger.errx(rspamd_config,
-        'cannot get current elastic index policy (%s), status code: %s, response: %s',
-        policy_url, code, body)
+          'cannot get current elastic index policy (%s), status code: %s, response: %s',
+          policy_url, code, body)
       handle_error('get current', 'index_policy', settings['limits']['max_fail'])
       upstream:fail()
     end
@@ -1032,7 +1037,7 @@ local function configure_index_policy(cfg, ev_base)
       }
       index_policy['policy']['phases']['delete'] = delete_obj
     end
-  -- opensearch state policy with hot state
+    -- opensearch state policy with hot state
   elseif detected_distro['name'] == 'opensearch' then
     local retry = {
       count = 3,
@@ -1376,7 +1381,7 @@ local function configure_index_template(cfg, ev_base)
       upstream:ok()
     else
       rspamd_logger.errx(rspamd_config, 'cannot configure elastic index template (%s), status code: %s, response: %s',
-        template_url, code, body)
+          template_url, code, body)
       upstream:fail()
       handle_error('configure', 'index_template', settings['limits']['max_fail'])
     end
@@ -1419,7 +1424,8 @@ local function verify_distro(manual)
     local supported_distro_info = supported_distro[detected_distro_name]
     -- check that detected_distro_version is valid
     if not detected_distro_version or type(detected_distro_version) ~= 'string' then
-      rspamd_logger.errx(rspamd_config, 'elastic version should be a string, but we received: %s', type(detected_distro_version))
+      rspamd_logger.errx(rspamd_config, 'elastic version should be a string, but we received: %s',
+          type(detected_distro_version))
       valid = false
     elseif detected_distro_version == '' then
       rspamd_logger.errx(rspamd_config, 'unsupported elastic version: empty string')
@@ -1429,18 +1435,20 @@ local function verify_distro(manual)
       local cmp_from = compare_versions(detected_distro_version, supported_distro_info['from'])
       if cmp_from == -1 then
         rspamd_logger.errx(rspamd_config, 'unsupported elastic version: %s, minimal supported version of %s is %s',
-          detected_distro_version, detected_distro_name, supported_distro_info['from'])
+            detected_distro_version, detected_distro_name, supported_distro_info['from'])
         valid = false
       else
         local cmp_till = compare_versions(detected_distro_version, supported_distro_info['till'])
         if (cmp_till >= 0) and not supported_distro_info['till_unknown'] then
-          rspamd_logger.errx(rspamd_config, 'unsupported elastic version: %s, maximum supported version of %s is less than %s',
-            detected_distro_version, detected_distro_name, supported_distro_info['till'])
+          rspamd_logger.errx(rspamd_config,
+              'unsupported elastic version: %s, maximum supported version of %s is less than %s',
+              detected_distro_version, detected_distro_name, supported_distro_info['till'])
           valid = false
         elseif (cmp_till >= 0) and supported_distro_info['till_unknown'] then
           rspamd_logger.warnx(rspamd_config,
-            'compatibility of elastic version: %s is unknown, maximum known supported version of %s is less than %s, use at your own risk',
-            detected_distro_version, detected_distro_name, supported_distro_info['till'])
+              'compatibility of elastic version: %s is unknown, maximum known supported version of %s is less than %s,' ..
+                  'use at your own risk',
+              detected_distro_version, detected_distro_name, supported_distro_info['till'])
           valid_unknown = true
         end
       end
@@ -1452,14 +1460,14 @@ local function verify_distro(manual)
   else
     if valid and manual then
       rspamd_logger.infox(
-        rspamd_config, 'assuming elastic distro: %s, version: %s', detected_distro_name, detected_distro_version)
+          rspamd_config, 'assuming elastic distro: %s, version: %s', detected_distro_name, detected_distro_version)
       detected_distro['supported'] = true
     elseif valid and not manual then
       rspamd_logger.infox(rspamd_config, 'successfully connected to elastic distro: %s, version: %s',
-        detected_distro_name, detected_distro_version)
+          detected_distro_name, detected_distro_version)
       detected_distro['supported'] = true
     else
-      handle_error('configure','distro',settings['version']['autodetect_max_fail'])
+      handle_error('configure', 'distro', settings['version']['autodetect_max_fail'])
     end
   end
 end
@@ -1468,7 +1476,8 @@ local function configure_distro(cfg, ev_base)
   if not settings['version']['autodetect_enabled'] then
     detected_distro['name'] = settings['version']['override']['name']
     detected_distro['version'] = settings['version']['override']['version']
-    rspamd_logger.infox(rspamd_config, 'automatic detection of elastic distro and version is disabled, taking configuration from settings')
+    rspamd_logger.infox(rspamd_config,
+        'automatic detection of elastic distro and version is disabled, taking configuration from settings')
     verify_distro(true)
   end
 
@@ -1481,7 +1490,8 @@ local function configure_distro(cfg, ev_base)
       rspamd_logger.errx(rspamd_config, 'cannot connect to elastic (%s): %s', root_url, err)
       upstream:fail()
     elseif code ~= 200 then
-      rspamd_logger.errx(rspamd_config, 'cannot connect to elastic (%s), status code: %s, response: %s', root_url, code, body)
+      rspamd_logger.errx(rspamd_config, 'cannot connect to elastic (%s), status code: %s, response: %s', root_url, code,
+          body)
       upstream:fail()
     else
       local parser = ucl.parser()
@@ -1492,10 +1502,10 @@ local function configure_distro(cfg, ev_base)
       else
         local obj = parser:get_object()
         if obj['tagline'] == "The OpenSearch Project: https://opensearch.org/" then
-            detected_distro['name'] = 'opensearch'
+          detected_distro['name'] = 'opensearch'
         end
         if obj['tagline'] == "You Know, for Search" then
-            detected_distro['name'] = 'elastic'
+          detected_distro['name'] = 'elastic'
         end
         if obj['version'] then
           if obj['version']['number'] then
@@ -1537,7 +1547,7 @@ end
 local opts = rspamd_config:get_all_opt('elastic')
 
 if opts then
-  for k,v in pairs(opts) do
+  for k, v in pairs(opts) do
     settings[k] = v
   end
 
@@ -1574,7 +1584,7 @@ if opts then
     rspamd_config:register_finish_script(function(task)
       local nlogs_total = buffer['logs']:length()
       if nlogs_total > 0 then
-        rspamd_logger.debugm(N, task, 'flushing buffer on shutdown, buffer size: %s', nlogs_total)
+        lua_util.debugm(N, task, 'flushing buffer on shutdown, buffer size: %s', nlogs_total)
         elastic_send_data(true, task)
       end
     end)