diff options
Diffstat (limited to 'src/plugins/lua/elastic.lua')
| -rw-r--r-- | src/plugins/lua/elastic.lua | 172 |
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) |